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
|
#include <library/cpp/actors/interconnect/outgoing_stream.h>
#include <library/cpp/testing/unittest/registar.h>
#include <util/random/entropy.h>
#include <util/stream/null.h>
#define Ctest Cnull
Y_UNIT_TEST_SUITE(OutgoingStream) {
Y_UNIT_TEST(Basic) {
std::vector<char> buffer;
buffer.resize(4 << 20);
TReallyFastRng32 rng(EntropyPool());
for (char *p = buffer.data(); p != buffer.data() + buffer.size(); p += sizeof(ui32)) {
*reinterpret_cast<ui32*>(p) = rng();
}
for (ui32 nIter = 0; nIter < 10; ++nIter) {
Cerr << "nIter# " << nIter << Endl;
size_t base = 0; // number of dropped bytes
size_t sendOffset = 0; // offset to base
size_t pending = 0; // number of bytes in queue
NInterconnect::TOutgoingStreamT<4096> stream;
size_t numRewindsRemain = 10;
while (base != buffer.size()) {
const size_t bytesToEnd = buffer.size() - (base + sendOffset);
Ctest << "base# " << base << " sendOffset# " << sendOffset << " pending# " << pending
<< " bytesToEnd# " << bytesToEnd;
UNIT_ASSERT_VALUES_EQUAL(stream.CalculateOutgoingSize(), pending + sendOffset);
UNIT_ASSERT_VALUES_EQUAL(stream.CalculateUnsentSize(), pending);
const size_t maxBuffers = 128;
std::vector<NActors::TConstIoVec> iov;
stream.ProduceIoVec(iov, maxBuffers, Max<size_t>());
size_t offset = base + sendOffset;
for (const auto& [ptr, len] : iov) {
UNIT_ASSERT(memcmp(buffer.data() + offset, ptr, len) == 0);
offset += len;
}
UNIT_ASSERT(iov.size() == maxBuffers || offset == base + sendOffset + pending);
const char *nextData = buffer.data() + base + sendOffset + pending;
const size_t nextDataMaxLen = bytesToEnd - pending;
const size_t nextDataLen = nextDataMaxLen ? rng() % Min<size_t>(16384, nextDataMaxLen) + 1 : 0;
if (size_t bytesToScan = sendOffset + pending) {
bytesToScan = rng() % bytesToScan + 1;
size_t offset = base + sendOffset + pending - bytesToScan;
stream.ScanLastBytes(bytesToScan, [&](TContiguousSpan span) {
UNIT_ASSERT(offset + span.size() <= base + sendOffset + pending);
UNIT_ASSERT(memcmp(buffer.data() + offset, span.data(), span.size()) == 0);
offset += span.size();
});
UNIT_ASSERT_VALUES_EQUAL(offset, base + sendOffset + pending);
}
enum class EAction {
COPY_APPEND,
WRITE,
REF_APPEND,
ADVANCE,
REWIND,
DROP,
BOOKMARK
};
std::vector<EAction> actions;
if (nextDataLen) {
actions.push_back(EAction::COPY_APPEND);
actions.push_back(EAction::WRITE);
actions.push_back(EAction::REF_APPEND);
actions.push_back(EAction::BOOKMARK);
}
if (numRewindsRemain && sendOffset > 65536) {
actions.push_back(EAction::REWIND);
}
actions.push_back(EAction::ADVANCE);
actions.push_back(EAction::DROP);
switch (actions[rng() % actions.size()]) {
case EAction::COPY_APPEND: {
Ctest << " COPY_APPEND nextDataLen# " << nextDataLen;
auto span = stream.AcquireSpanForWriting(nextDataLen);
UNIT_ASSERT(span.size() != 0);
memcpy(span.data(), nextData, span.size());
stream.Append(span);
pending += span.size();
break;
}
case EAction::WRITE:
Ctest << " WRITE nextDataLen# " << nextDataLen;
stream.Write({nextData, nextDataLen});
pending += nextDataLen;
break;
case EAction::REF_APPEND:
Ctest << " REF_APPEND nextDataLen# " << nextDataLen;
stream.Append({nextData, nextDataLen});
pending += nextDataLen;
break;
case EAction::ADVANCE: {
const size_t advance = rng() % Min<size_t>(4096, pending + 1);
Ctest << " ADVANCE advance# " << advance;
stream.Advance(advance);
sendOffset += advance;
pending -= advance;
break;
}
case EAction::REWIND:
Ctest << " REWIND";
stream.Rewind();
pending += sendOffset;
sendOffset = 0;
--numRewindsRemain;
break;
case EAction::DROP: {
const size_t drop = rng() % Min<size_t>(65536, sendOffset + 1);
Ctest << " DROP drop# " << drop;
stream.DropFront(drop);
base += drop;
sendOffset -= drop;
break;
}
case EAction::BOOKMARK:
Ctest << " BOOKMARK nextDataLen# " << nextDataLen;
auto bookmark = stream.Bookmark(nextDataLen);
stream.WriteBookmark(std::move(bookmark), {nextData, nextDataLen});
pending += nextDataLen;
break;
}
Ctest << Endl;
}
}
}
}
|
