1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
|
#pragma once
#include <library/cpp/monlib/dynamic_counters/counters.h>
#include <library/cpp/actors/core/actorsystem.h>
#include <library/cpp/actors/core/event_load.h>
#include <library/cpp/actors/util/rope.h>
#include <util/generic/deque.h>
#include <util/generic/vector.h>
#include <util/generic/map.h>
#include <util/stream/walk.h>
#include <library/cpp/actors/wilson/wilson_span.h>
#include "interconnect_common.h"
#include "interconnect_counters.h"
#include "packet.h"
#include "event_holder_pool.h"
namespace NActors {
#pragma pack(push, 1)
struct TChannelPart {
ui16 ChannelFlags;
ui16 Size;
static constexpr ui16 LastPartFlag = 0x8000;
static constexpr ui16 XdcFlag = 0x4000;
static constexpr ui16 ChannelMask = (1 << IEventHandle::ChannelBits) - 1;
static_assert((LastPartFlag & ChannelMask) == 0);
static_assert((XdcFlag & ChannelMask) == 0);
ui16 GetChannel() const { return ChannelFlags & ChannelMask; }
bool IsLastPart() const { return ChannelFlags & LastPartFlag; }
bool IsXdc() const { return ChannelFlags & XdcFlag; }
TString ToString() const {
return TStringBuilder() << "{Channel# " << GetChannel()
<< " IsLastPart# " << IsLastPart()
<< " IsXdc# " << IsXdc()
<< " Size# " << Size << "}";
}
};
#pragma pack(pop)
enum class EXdcCommand : ui8 {
DECLARE_SECTION = 1,
PUSH_DATA,
DECLARE_SECTION_INLINE,
};
struct TExSerializedEventTooLarge : std::exception {
const ui32 Type;
TExSerializedEventTooLarge(ui32 type)
: Type(type)
{}
};
class TEventOutputChannel : public TInterconnectLoggingBase {
public:
TEventOutputChannel(TEventHolderPool& pool, ui16 id, ui32 peerNodeId, ui32 maxSerializedEventSize,
std::shared_ptr<IInterconnectMetrics> metrics, TSessionParams params)
: TInterconnectLoggingBase(Sprintf("OutputChannel %" PRIu16 " [node %" PRIu32 "]", id, peerNodeId))
, Pool(pool)
, PeerNodeId(peerNodeId)
, ChannelId(id)
, Metrics(std::move(metrics))
, Params(std::move(params))
, MaxSerializedEventSize(maxSerializedEventSize)
{}
~TEventOutputChannel() {
}
std::pair<ui32, TEventHolder*> Push(IEventHandle& ev) {
TEventHolder& event = Pool.Allocate(Queue);
const ui32 bytes = event.Fill(ev) + sizeof(TEventDescr2);
OutputQueueSize += bytes;
if (event.Span = NWilson::TSpan(15 /*max verbosity*/, NWilson::TTraceId(ev.TraceId), "Interconnect.Queue")) {
event.Span
.Attribute("OutputQueueItems", static_cast<i64>(Queue.size()))
.Attribute("OutputQueueSize", static_cast<i64>(OutputQueueSize));
}
return std::make_pair(bytes, &event);
}
void DropConfirmed(ui64 confirm);
bool FeedBuf(TTcpPacketOutTask& task, ui64 serial, ui64 *weightConsumed);
bool IsEmpty() const {
return Queue.empty();
}
bool IsWorking() const {
return !IsEmpty();
}
ui32 GetQueueSize() const {
return (ui32)Queue.size();
}
ui64 GetBufferedAmountOfData() const {
return OutputQueueSize;
}
void NotifyUndelivered();
TEventHolderPool& Pool;
const ui32 PeerNodeId;
const ui16 ChannelId;
std::shared_ptr<IInterconnectMetrics> Metrics;
const TSessionParams Params;
const ui32 MaxSerializedEventSize;
ui64 UnaccountedTraffic = 0;
ui64 EqualizeCounterOnPause = 0;
ui64 WeightConsumedOnPause = 0;
enum class EState {
INITIAL,
BODY,
DESCRIPTOR,
SECTIONS,
};
EState State = EState::INITIAL;
protected:
ui64 OutputQueueSize = 0;
std::list<TEventHolder> Queue;
std::list<TEventHolder> NotYetConfirmed;
TRope::TConstIterator Iter;
TCoroutineChunkSerializer Chunker;
TEventSerializationInfo SerializationInfoContainer;
const TEventSerializationInfo *SerializationInfo = nullptr;
bool IsPartInline = false;
size_t PartLenRemain = 0;
size_t SectionIndex = 0;
std::vector<char> XdcData;
template<bool External>
bool SerializeEvent(TTcpPacketOutTask& task, TEventHolder& event, size_t *bytesSerialized);
bool FeedPayload(TTcpPacketOutTask& task, TEventHolder& event, ui64 *weightConsumed);
std::optional<bool> FeedInlinePayload(TTcpPacketOutTask& task, TEventHolder& event, ui64 *weightConsumed);
std::optional<bool> FeedExternalPayload(TTcpPacketOutTask& task, TEventHolder& event, ui64 *weightConsumed);
bool FeedDescriptor(TTcpPacketOutTask& task, TEventHolder& event, ui64 *weightConsumed);
void AccountTraffic() {
if (const ui64 amount = std::exchange(UnaccountedTraffic, 0)) {
Metrics->UpdateOutputChannelTraffic(ChannelId, amount);
}
}
friend class TInterconnectSessionTCP;
};
}
|
