jeka
/
utun2
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity
Browse Source

Test immediate relay forwarding and log batch authentication

master
evgeny 7 days ago
parent
01d7fd5bef
commit
43a600c5a5
3 changed files with 28 additions and 9 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Split View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 2
      src/stream/stream.c
  2. 5
      src/stream/stream_buffer.c
  3. 30
      tests/test_stream_service.c

2
src/stream/stream.c
Unescape View File

@ -298,7 +298,7 @@ static void session_deliver(void *arg) {
if (result == STREAM_OK) {
if (!s->decoded) {
s->decoded = 1;
DEBUG_INFO(DEBUG_CATEGORY_STREAM, "stream first batch reconstructed group=%llx source=%llx epoch=%llx seq=%llu k=%u",
DEBUG_INFO(DEBUG_CATEGORY_STREAM, "stream first batch reconstructed group=%llx source=%llx epoch=%llx seq=%llu k=%u signature=verified",
(unsigned long long)s->descriptor.group_id, (unsigned long long)s->descriptor.source_node_id,
(unsigned long long)s->descriptor.source_epoch, (unsigned long long)s->next_seq,
(unsigned)(s->descriptor.n - s->descriptor.r));

5
src/stream/stream_buffer.c
Unescape View File

@ -182,9 +182,10 @@ int stream_buffer_create(struct stream_buffer **out, const struct stream_descrip
/* Прекращение владельцем всех обращений предшествует destroy; кольцо ничего не знает о чужих NCD handles. */
void stream_buffer_destroy(struct stream_buffer *b) {
if (!b) return;
DEBUG_INFO(DEBUG_CATEGORY_STREAM, "buffer destroy source=%llx blocks=%u batches=%u bytes=%zu evicted=%llu duplicates=%llu",
DEBUG_INFO(DEBUG_CATEGORY_STREAM, "buffer destroy source=%llx blocks=%u batches=%u bytes=%zu evicted=%llu duplicates=%llu rejected=%llu",
(unsigned long long)b->descriptor.source_node_id, b->stats.blocks, b->stats.batches, b->stats.stored_bytes,
(unsigned long long)b->stats.evicted_batches, (unsigned long long)b->stats.duplicates);
(unsigned long long)b->stats.evicted_batches, (unsigned long long)b->stats.duplicates,
(unsigned long long)b->stats.rejected_batches);
for (uint32_t i = 0; i < b->stats.batches; i++) u_free(b->heap[i]);
if (b->slots) for (uint32_t i = 0; i < b->capacity; i++) u_free(b->slots[i].path);
rs_free(b->rs);

30
tests/test_stream_service.c
Unescape View File

@ -24,6 +24,7 @@
#define GROUP1 UINT64_C(710001)
#define GROUP2 UINT64_C(710002)
#define SINGLE_INPUT_CAPACITY (STREAM_RS_DEFAULT_SIZE - RS_HEADER_SIZE - STREAM_BATCH_OVERHEAD)
#define CHECK(expr) do { if (!(expr)) { \
DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "test_stream_service failed line=%d: %s", __LINE__, #expr); exit(1); \
} } while (0)
@ -84,7 +85,7 @@ err_t __wrap_uasync_get_load(struct UASYNC *async, struct uasync_load *load) {
static void on_global_pressure(struct UTUN_INSTANCE *inst, const struct stream_event *event, void *arg) {
(void)arg; unsigned node = inst == nodes[1] ? 1 : 2;
CHECK(event->descriptor.source_node_id == nodes[3]->node_id && event->descriptor.group_id == GROUP2);
if (event->type == STREAM_EVENT_DATA) { CHECK(event->length == 1180); global_data[node]++; }
if (event->type == STREAM_EVENT_DATA) { CHECK(event->length == SINGLE_INPUT_CAPACITY); global_data[node]++; }
else CHECK(event->type == STREAM_EVENT_GAP || event->type == STREAM_EVENT_CLOSED || event->type == STREAM_EVENT_END);
}
@ -622,6 +623,21 @@ static void test_network(void) {
/* Снять два лишних проверочных субпотока: шесть оставшихся восстанавливают новые пачки. */
for (unsigned i = 6; i < 8; i++) { stream_receive_close(up[i]); up[i] = NULL; stream_receive_close(down[i]); down[i] = NULL; }
network_write(publisher, 3); run_until(&network_data[1], 2); run_until(&network_data[2], 2); run_until(&network_gaps[2], 1);
/* Relay пересылает первую часть, пока у него и downstream ещё нет k частей для проверки общей подписи. */
for (unsigned sub = 1; sub < 6; sub++) { stream_receive_close(up[sub]); up[sub] = NULL; }
CHECK(stream_receive_get_info(down[0], &info) == STREAM_OK && info.blocks == 2);
network_write(publisher, 4); deadline = get_time_tb() + 100000;
do { test_clock_poll(ua, 100); CHECK(stream_receive_get_info(down[0], &info) == STREAM_OK); }
while (info.blocks < 3 && get_time_tb() < deadline);
CHECK(info.blocks == 3 && network_data[1] == 2 && network_data[2] == 2);
uint8_t pending_output[sizeof(expected)]; size_t pending_length = 0;
for (unsigned node = 1; node < 3; node++) {
struct stream_local pending; CHECK(stream_local_get(nodes[node], &origin.descriptor, &pending) == STREAM_OK);
CHECK(pending.next == 4 && stream_buffer_decode(pending.buffer, 4, pending_output, sizeof(pending_output), &pending_length) == STREAM_WAIT);
}
for (unsigned sub = 1; sub < 6; sub++) CHECK(stream_receive_open(subscribers[1], nodes[3]->node_id, sub, 4, &up[sub]) == STREAM_OK);
run_until(&network_data[1], 3); run_until(&network_data[2], 3);
DEBUG_INFO(DEBUG_CATEGORY_STREAM, "[PASS] live relay forwards before batch decode; consumers wait for k parts and source signature");
CHECK(stream_publish_finish(publisher) == STREAM_OK);
run_until(&network_end[1], 1); run_until(&network_end[2], 1);
/* RENEW каждые 2с удерживает даже окончившуюся выдачу; затем имитируем отсутствие продлений. */
@ -791,7 +807,7 @@ static void test_global_pressure(void) {
CHECK(stream_subscribe_manual_open(nodes[i], GROUP2, nodes[3]->node_id, &subscribers[i]) == STREAM_OK);
stream_set_event_callback(nodes[i], on_global_pressure, NULL);
}
uint8_t payload[1180]; memset(payload, 3, sizeof(payload)); size_t accepted;
uint8_t payload[SINGLE_INPUT_CAPACITY]; memset(payload, 3, sizeof(payload)); size_t accepted;
CHECK(stream_publish_write(publisher, payload, sizeof(payload), &accepted) == STREAM_OK && accepted == sizeof(payload));
for (unsigned i = 1; i < 3; i++) for (unsigned sub = 0; sub < 2; sub++)
CHECK(stream_receive_open(subscribers[i], nodes[3]->node_id, sub, UINT64_MAX, &receives[i][sub]) == STREAM_OK);
@ -902,10 +918,12 @@ static void test_automatic(void) {
}
test_clock_run_for(ua, 3000);
CHECK(!automatic_data[1] && !automatic_data[2]);
memset(pressure_payload, 0xa9, 1180); size_t accepted;
size_t capacity = stream_publish_capacity(publisher);
CHECK(capacity == SINGLE_INPUT_CAPACITY && capacity <= sizeof(pressure_payload));
memset(pressure_payload, 0xa9, capacity); size_t accepted;
uint64_t initial_probes[3] = {0}; struct stream_selection_info state[3];
for (unsigned frame = 0; frame < 350; frame++) {
CHECK(stream_publish_write(publisher, pressure_payload, 1180, &accepted) == STREAM_OK && accepted == 1180);
CHECK(stream_publish_write(publisher, pressure_payload, capacity, &accepted) == STREAM_OK && accepted == capacity);
test_clock_run_for(ua, 1200);
if (frame == 80 || frame == 180) for (unsigned i = 1; i < 3; i++) {
CHECK(stream_selection_get_info(subscribers[i], &state[i]) == STREAM_OK);
@ -925,7 +943,7 @@ static void test_automatic(void) {
CHECK(same.probes_started == state[1].probes_started); stream_close(second); second = NULL;
uint64_t migrations = state[2].migrations; automatic_drop_relay = 1;
for (unsigned frame = 0; frame < 30; frame++) {
CHECK(stream_publish_write(publisher, pressure_payload, 1180, &accepted) == STREAM_OK);
CHECK(stream_publish_write(publisher, pressure_payload, capacity, &accepted) == STREAM_OK);
test_clock_run_for(ua, 1200);
}
CHECK(stream_selection_get_info(subscribers[2], &state[2]) == STREAM_OK);
@ -937,7 +955,7 @@ static void test_automatic(void) {
memcpy(cancel->dgram, automatic_request, sizeof(automatic_request)); cancel->len = sizeof(automatic_request);
automatic_handlers[3](source_connection, cancel);
for (unsigned frame = 0; frame < 20; frame++) {
CHECK(stream_publish_write(publisher, pressure_payload, 1180, &accepted) == STREAM_OK);
CHECK(stream_publish_write(publisher, pressure_payload, capacity, &accepted) == STREAM_OK);
test_clock_run_for(ua, 1200);
}
CHECK(stream_selection_get_info(subscribers[2], &state[2]) == STREAM_OK);

Write Preview
Loading…
Cancel
Save