From 6332e75e6164fed54a78f5c1517b12e8a8149208 Mon Sep 17 00:00:00 2001 From: evgeny Date: Sun, 11 Oct 2026 16:43:16 +0200 Subject: [PATCH] Add real uTun stream bench node with payload and latency validation --- tests/Makefile.am | 10 +- tests/stream_netns/node.c | 373 ++++++++++++++++++++++++++++++++++++++ 2 files changed, 382 insertions(+), 1 deletion(-) create mode 100644 tests/stream_netns/node.c diff --git a/tests/Makefile.am b/tests/Makefile.am index df4e567a..de8a6eca 100644 --- a/tests/Makefile.am +++ b/tests/Makefile.am @@ -1,7 +1,15 @@ # Tests Makefile.am for utun - all tests with new ll_queue library EXTRA_DIST = fixtures/reality_openssl_hello.hex data/silero_speech.pcm data/silero_speech.txt test_net.h \ - run_tests.py test_run_tests.py test_build_script.py + run_tests.py test_run_tests.py test_build_script.py stream_netns/topology.py stream_netns/network.py \ + stream_netns/probe.py stream_netns/run.py + +if OS_LINUX +# Отдельный привилегированный стенд не запускается обычным make check. +EXTRA_PROGRAMS = stream_netns/stream_lab_node +stream_netns_stream_lab_node_SOURCES = stream_netns/node.c +stream_netns_stream_lab_node_LDADD = $(LIBUTUN) $(CRYPTO_LIBS) $(COMMON_LIBS) +endif # Порядок по замерам Linux: быстрые проверки раньше длительных; состав зависит от прежних configure-условий. diff --git a/tests/stream_netns/node.c b/tests/stream_netns/node.c new file mode 100644 index 00000000..d0706387 --- /dev/null +++ b/tests/stream_netns/node.c @@ -0,0 +1,373 @@ +/* Агент стенда: настоящие JOIN/Merkle/ETCP и автоматический выбор поставщиков. + * stdin/stdout — только JSONL RPC; диагностика ядра пишется в отдельный файл. + * Заголовок нагрузки и SHA256 проверяют opaque DATA после восстановления FEC. */ +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include "../../lib/debug_config.h" +#include "../../lib/json_flat.h" +#include "../../lib/mem.h" +#include "../../src/chat/chat_core.h" +#include "../../src/chat/chat_sync.h" +#include "../../src/chat/invite_build.h" +#include "../../src/chat/invite_link.h" +#include "../../src/stream/stream_priv.h" +#include "../../src/stream/stream_member.h" +#include "../../src/utun_instance.h" + +#define LAB_STREAMS 4 +#define LAB_HISTOGRAM 4096 + +struct lab_stream { + struct STREAM_HANDLE *handle; + uint64_t source, generated, received, bytes, gaps, gap_batches, errors, ends, closed, next, first, latency_max; + uint64_t latency_sum, histogram[LAB_HISTOGRAM], last_data_ns; + uint64_t seed, interval_ns, due_ns; + uint8_t *payload; + size_t size; + int publisher, started; +}; +struct lab { + struct UTUN_INSTANCE *inst; + struct lab_stream streams[LAB_STREAMS]; + struct chat_invite_build *invite; + uint64_t invite_id, group; + int errors; +}; +static volatile sig_atomic_t stopping; + +static uint64_t now_ns(void) { + struct timespec time; + clock_gettime(CLOCK_MONOTONIC, &time); + return (uint64_t)time.tv_sec * UINT64_C(1000000000) + time.tv_nsec; +} + +static void stop_signal(int signal_number) { (void)signal_number; stopping = 1; } + +static void json_string(const char *value) { + putchar('"'); + for (const unsigned char *p = (const unsigned char *)value; *p; p++) { + if (*p == '"' || *p == '\\') { putchar('\\'); putchar(*p); } + else if (*p < 32) printf("\\u%04x", *p); + else putchar(*p); + } + putchar('"'); +} + +static void reply(uint64_t id, int result) { + printf("{\"id\":%" PRIu64 ",\"rc\":%d}\n", id, result); +} + +static int number(const char *json, const char *key, uint64_t *value) { + char buffer[32], *end; + if (json_flat_get(json, key, buffer, sizeof(buffer)) || !buffer[0] || buffer[0] == '-') return -1; + errno = 0; + uint64_t parsed = strtoull(buffer, &end, 10); + if (errno || *end) return -1; + *value = parsed; + return 0; +} + +static struct lab_stream *slot(struct lab *lab, uint64_t source, int create) { + for (unsigned i = 0; i < LAB_STREAMS; i++) if (lab->streams[i].source == source) return &lab->streams[i]; + for (unsigned i = 0; create && i < LAB_STREAMS; i++) { + if (!lab->streams[i].source) { lab->streams[i].source = source; return &lab->streams[i]; } + } + return NULL; +} + +static void event(struct UTUN_INSTANCE *inst, const struct stream_event *event, void *arg) { + (void)inst; + struct lab *lab = arg; + struct lab_stream *stream = slot(lab, event->descriptor.source_node_id, 0); + if (!stream) { + DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "lab: event without owner source=%016" PRIx64, event->descriptor.source_node_id); + lab->errors++; return; + } + if (event->type == STREAM_EVENT_CLOSED) { stream->closed++; return; } + if (event->type == STREAM_EVENT_END) { stream->ends++; return; } + if (event->type == STREAM_EVENT_GAP) { + stream->gaps++; stream->gap_batches += event->last_seq - event->first_seq + 1; + if (stream->started && event->first_seq != stream->next) { + DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "lab: GAP cursor expected=%" PRIu64 " first=%" PRIu64, stream->next, event->first_seq); + stream->errors++; + } + stream->started = 1; stream->next = event->last_seq + 1; + DEBUG_WARN(DEBUG_CATEGORY_STREAM, "lab: GAP source=%016" PRIx64 " first=%" PRIu64 " last=%" PRIu64, + stream->source, event->first_seq, event->last_seq); + return; + } + const uint8_t *data = event->data; + uint8_t digest[32]; + int valid = event->length >= 80 && !memcmp(data, "SLABv001", 8); + if (valid) { + SHA256(data, event->length - 32, digest); + valid = !memcmp(digest, data + event->length - 32, 32) && stream_read_be(data + 8, 8) == event->first_seq && + stream_read_be(data + 32, 8) == stream->source && stream_read_be(data + 40, 4) == event->length; + } + if (!valid || (stream->started && event->first_seq != stream->next)) { + DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "lab: DATA validation source=%016" PRIx64 " seq=%" PRIu64 + " expected=%" PRIu64 " bytes=%zu checksum=%d", stream->source, event->first_seq, stream->next, event->length, valid); + stream->errors++; + } + if (!stream->received) stream->first = event->first_seq; + stream->started = 1; stream->next = event->last_seq + 1; + stream->received++; stream->bytes += event->length; stream->last_data_ns = now_ns(); + if (valid) { + uint64_t generated_ns = stream_read_be(data + 16, 8); + if (generated_ns > stream->last_data_ns) { + DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "lab: CLOCK_MONOTONIC moved backwards"); stream->errors++; return; + } + uint64_t delay = (stream->last_data_ns - generated_ns) / UINT64_C(1000000); + stream->latency_sum += delay; + if (delay > stream->latency_max) stream->latency_max = delay; + stream->histogram[delay < LAB_HISTOGRAM ? delay : LAB_HISTOGRAM - 1]++; + } +} + +static uint64_t percentile(const struct lab_stream *stream, unsigned percent) { + uint64_t target = (stream->received * percent + 99) / 100, count = 0; + if (!target) return 0; + for (unsigned i = 0; i < LAB_HISTOGRAM; i++) { count += stream->histogram[i]; if (count >= target) return i; } + return LAB_HISTOGRAM - 1; +} + +static void statistics(struct lab *lab, uint64_t id) { + struct TOPO_GROUP *group = topo_groups_find(lab->inst->topo_groups, lab->group); + printf("{\"id\":%" PRIu64 ",\"rc\":0,\"node\":\"%" PRIu64 "\",\"group\":\"%" PRIu64 + "\",\"errors\":%d,\"online\":[", id, lab->inst->node_id, lab->group, lab->errors); + int separator = 0; + for (struct ll_entry *e = group ? group->nodes->head : NULL; e; e = e->next) { + struct TOPO_GROUP_NODE *node = (struct TOPO_GROUP_NODE *)e; + if (stream_member_online(lab->inst, lab->group, node->node_id)) { + printf("%s\"%" PRIu64 "\"", separator++ ? "," : "", node->node_id); + } + } + printf("],\"streams\":["); separator = 0; + for (unsigned i = 0; i < LAB_STREAMS; i++) { + struct lab_stream *stream = &lab->streams[i]; + if (!stream->source) continue; + printf("%s{\"source\":\"%" PRIu64 "\",\"publisher\":%d,\"generated\":%" PRIu64 ",\"received\":%" PRIu64 + ",\"bytes\":%" PRIu64 ",\"gaps\":%" PRIu64 ",\"gap_batches\":%" PRIu64 ",\"errors\":%" PRIu64 + ",\"ends\":%" PRIu64 ",\"closed\":%" PRIu64 ",\"first_seq\":%" PRIu64 ",\"next_seq\":%" PRIu64 + ",\"age_ms\":%" PRIu64 ",\"latency_p50_ms\":%" PRIu64 ",\"latency_p95_ms\":%" PRIu64 + ",\"latency_p99_ms\":%" PRIu64 ",\"latency_max_ms\":%" PRIu64, + separator++ ? "," : "", stream->source, stream->publisher, stream->generated, stream->received, + stream->bytes, stream->gaps, stream->gap_batches, stream->errors, stream->ends, stream->closed, + stream->first, stream->next, stream->last_data_ns ? (now_ns() - stream->last_data_ns) / UINT64_C(1000000) : 0, + percentile(stream, 50), percentile(stream, 95), percentile(stream, 99), stream->latency_max); + struct stream_info info; + if (stream->handle && stream_get_info(stream->handle, &info) == STREAM_OK) { + struct stream_selection_info selection; + memset(&selection, 0, sizeof(selection)); + stream_selection_get_info(stream->handle, &selection); + printf(",\"active\":%d,\"epoch\":\"%" PRIu64 "\",\"stored_bytes\":%zu,\"capacity_bytes\":%zu" + ",\"evicted_batches\":%" PRIu64 ",\"subflows\":%u,\"providers\":%u,\"probes\":%" PRIu64 + ",\"cycles\":%" PRIu64 ",\"migrations\":%" PRIu64 ",\"failures\":%" PRIu64 ",\"restarts\":%" PRIu64 + ",\"protected\":%d,\"paths\":[", stream_handle_active(stream->handle), info.descriptor.source_epoch, + info.buffer.stored_bytes, info.buffer.capacity_bytes, info.buffer.evicted_batches, selection.subflows, + selection.providers, selection.probes_started, selection.cycles, selection.migrations, selection.failures, + selection.restarts, selection.protected_by_fec); + struct stream_local local; + int path_separator = 0; + if (stream_local_get(lab->inst, &info.descriptor, &local) == STREAM_OK) { + for (uint8_t sub = 0; sub < info.descriptor.n; sub++) { + uint64_t first, last; + struct stream_path path; + if (stream_buffer_range(local.buffer, sub, &first, &last) != STREAM_OK || + stream_buffer_path(local.buffer, last, sub, &path) != STREAM_OK) continue; + printf("%s{\"sub\":%u,\"seq\":%" PRIu64 ",\"nodes\":[", path_separator++ ? "," : "", sub, last); + for (unsigned p = 0; p < path.count; p++) printf("%s\"%" PRIu64 "\"", p ? "," : "", path.nodes[p]); + printf("]}"); + } + } + printf("]"); + } + printf("}"); + } + printf("],\"peer_load\":["); separator = 0; + for (struct ll_entry *e = group ? group->nodes->head : NULL; e; e = e->next) { + struct TOPO_GROUP_NODE *node = (struct TOPO_GROUP_NODE *)e; + struct stream_peer_load_info load; + if (stream_peer_get_load(lab->inst, node->node_id, &load) != STREAM_OK) continue; + printf("%s{\"peer\":\"%" PRIu64 "\",\"queued_bytes\":%" PRIu64 ",\"growth_rate\":%" PRIu64 + ",\"transport_bytes\":%" PRIu64 ",\"inflight_bytes\":%" PRIu64 ",\"pressure\":%d,\"confirmed\":%d}", + separator++ ? "," : "", node->node_id, load.load.queued.bytes, load.load.growth_rate, + load.transport_queued_bytes, load.inflight_bytes, load.load.pressure, load.load.confirmed); + } + printf("]}\n"); +} + +static void invite_ready(void *arg, int result, const char *link) { + struct lab *lab = arg; + lab->invite = NULL; + printf("{\"id\":%" PRIu64 ",\"rc\":%d,\"link\":", lab->invite_id, result); + json_string(link ? link : ""); printf("}\n"); +} + +static void command(struct lab *lab, const char *json) { + uint64_t id = 0, source = 0, value = 0; + char operation[32]; + if (number(json, "id", &id) || json_flat_get(json, "op", operation, sizeof(operation))) { + DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "lab: invalid RPC"); reply(id, -1); return; + } + DEBUG_INFO(DEBUG_CATEGORY_STREAM, "lab: RPC id=%" PRIu64 " op=%s", id, operation); + if (!strcmp(operation, "stats")) { statistics(lab, id); return; } + if (!strcmp(operation, "quit")) { stopping = 1; reply(id, 0); return; } + if (!strcmp(operation, "create")) { + if (lab->group) { reply(id, -1); return; } + chat_core_create_channel_auto(lab->inst, "stream-netns"); + for (struct ll_entry *e = lab->inst->topo_groups->group_list->head; e; e = e->next) { + struct TOPO_GROUP *group = (struct TOPO_GROUP *)e; + if (group->group_type == TOPO_GROUP_TYPE_CHAT) { lab->group = group->group_id; break; } + } + printf("{\"id\":%" PRIu64 ",\"rc\":%d,\"group\":\"%" PRIu64 "\"}\n", id, lab->group ? 0 : -1, lab->group); return; + } + if (!strcmp(operation, "invite")) { + if (lab->invite || !lab->group) { reply(id, -1); return; } + lab->invite_id = id; + lab->invite = chat_invite_build_link(lab->inst, lab->group, lab->inst->node_id, NULL, invite_ready, lab); + if (!lab->invite) reply(id, -1); + return; + } + if (!strcmp(operation, "join")) { + char link[2048], error[256]; + struct InviteData invite; + uint8_t addresses[2048]; + if (json_flat_get(json, "link", link, sizeof(link)) || invite_link_decode(link, strlen(link), &invite, error, sizeof(error))) { + DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "lab: invalid invite"); reply(id, -1); return; + } + int size = invite_serialize_addrs(&invite, addresses, sizeof(addresses)); + if (size <= 0) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "lab: invite has no endpoint"); reply(id, -1); return; } + lab->group = invite.channelId; + chat_sync_connect_from_invite(lab->inst, invite.channelId, invite.nodeId, invite.pubkey, addresses, + invite.addrCount, size, invite.password_len ? invite.password : NULL, invite.join_key); + reply(id, 0); return; + } + int publish = !strcmp(operation, "publish"); + if (publish) source = lab->inst->node_id; + else if (number(json, "source", &source)) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "lab: missing source"); reply(id, -1); return; } + struct lab_stream *stream = slot(lab, source, publish || !strcmp(operation, "subscribe")); + if (!stream) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "lab: no stream slot"); reply(id, -1); return; } + if (publish || !strcmp(operation, "subscribe")) { + if (stream->handle) { reply(id, 0); return; } + int result = publish ? stream_publish_open(lab->inst, lab->group, NULL, &stream->handle) : + stream_subscribe_open(lab->inst, lab->group, source, &stream->handle); + if (result == STREAM_OK && publish) { + struct stream_info info; + stream_get_info(stream->handle, &info); + stream->size = stream_input_size(&info.descriptor); + stream->payload = u_malloc(stream->size); + stream->publisher = 1; + if (!stream->payload) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "lab: payload allocation failed"); stopping = 1; lab->errors++; } + if (!number(json, "seed", &value)) stream->seed = value; + } + reply(id, result); return; + } + if (!strcmp(operation, "rate") && stream->publisher && !number(json, "bps", &value)) { + stream->interval_ns = value ? stream->size * UINT64_C(8000000000) / value : 0; + if (value && !stream->interval_ns) stream->interval_ns = 1; + stream->due_ns = now_ns(); reply(id, 0); return; + } + if (!strcmp(operation, "finish") && stream->publisher) { + stream->interval_ns = 0; reply(id, stream_publish_finish(stream->handle)); return; + } + if (!strcmp(operation, "close")) { + stream->interval_ns = 0; + if (stream->handle) stream_close(stream->handle); + stream->handle = NULL; reply(id, 0); return; + } + DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "lab: unsupported RPC op=%s", operation); reply(id, -1); +} + +static void generate(struct lab *lab) { + for (unsigned i = 0; i < LAB_STREAMS; i++) { + struct lab_stream *stream = &lab->streams[i]; + for (unsigned burst = 0; stream->interval_ns && now_ns() >= stream->due_ns && burst < 8; burst++) { + uint8_t *data = stream->payload; + memset(data, 0, 48); memcpy(data, "SLABv001", 8); + stream_write_be(data + 8, stream->generated, 8); stream_write_be(data + 16, now_ns(), 8); + stream_write_be(data + 24, stream->seed, 8); stream_write_be(data + 32, stream->source, 8); + stream_write_be(data + 40, stream->size, 4); + uint64_t random = stream->seed ^ stream->generated ^ stream->source; + for (size_t p = 48; p < stream->size - 32; p++) { + random ^= random << 13; random ^= random >> 7; random ^= random << 17; data[p] = (uint8_t)random; + } + SHA256(data, stream->size - 32, data + stream->size - 32); + size_t accepted = 0; + int result = stream_publish_write(stream->handle, data, stream->size, &accepted); + if (result != STREAM_OK || accepted != stream->size) { + DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "lab: write failed rc=%d accepted=%zu expected=%zu", result, accepted, stream->size); + lab->errors++; stream->interval_ns = 0; break; + } + stream->generated++; stream->due_ns += stream->interval_ns; + } + } +} + +int main(int argc, char **argv) { + if (argc == 3 && !strcmp(argv[1], "--identity")) { + uint8_t private_key[32], public_key[32]; + SHA256((const unsigned char *)argv[2], strlen(argv[2]), private_key); + if (sc_compute_public_key_from_private(private_key, public_key) != SC_OK) return 1; + printf("{\"private\":\""); + for (unsigned i = 0; i < 32; i++) printf("%02x", private_key[i]); + printf("\",\"public\":\""); + for (unsigned i = 0; i < 32; i++) printf("%02x", public_key[i]); + printf("\",\"node\":\"%" PRIu64 "\"}\n", sc_derive_node_id_from_pubkey(public_key)); return 0; + } + if (argc != 3) { fprintf(stderr, "Usage: %s CONFIG LOG | --identity SEED\n", argv[0]); return 1; } + setvbuf(stdout, NULL, _IOLBF, 0); + signal(SIGTERM, stop_signal); signal(SIGINT, stop_signal); signal(SIGPIPE, SIG_IGN); + debug_config_init(); debug_set_level(DEBUG_LEVEL_WARN); debug_enable_console(0); debug_enable_file_output(argv[2], 1); + utun_instance_set_tun_init_enabled(0); + struct UASYNC *async = uasync_create(); + struct lab *lab = u_calloc(1, sizeof(*lab)); + if (!async || !lab) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "lab: initialization allocation failed"); return 1; } + lab->inst = utun_instance_create(async, argv[1]); + if (!lab->inst || utun_core_start(lab->inst) || chat_service_start(lab->inst)) { + DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "lab: service startup failed"); return 1; + } + debug_enable_console(0); + stream_set_event_callback(lab->inst, event, lab); + if (fcntl(STDIN_FILENO, F_SETFL, fcntl(STDIN_FILENO, F_GETFL) | O_NONBLOCK) < 0) { + DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "lab: stdin nonblocking failed: %s", strerror(errno)); return 1; + } + printf("{\"ready\":true,\"node\":\"%" PRIu64 "\"}\n", lab->inst->node_id); + char input[8192]; size_t used = 0; + while (!stopping) { + uasync_poll(async, 100); generate(lab); + ssize_t size = read(STDIN_FILENO, input + used, sizeof(input) - used - 1); + if (size == 0) break; + if (size < 0) { + if (errno != EAGAIN && errno != EINTR) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "lab: stdin failed: %s", strerror(errno)); lab->errors++; break; } + continue; + } + used += (size_t)size; input[used] = 0; + char *line; + while ((line = memchr(input, '\n', used))) { + *line = 0; command(lab, input); + size_t consumed = (size_t)(line - input) + 1; + memmove(input, input + consumed, used - consumed); used -= consumed; input[used] = 0; + } + if (used == sizeof(input) - 1) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "lab: RPC exceeds input limit"); lab->errors++; break; } + } + if (lab->invite) chat_invite_build_cancel(lab->invite); + for (unsigned i = 0; i < LAB_STREAMS; i++) { + if (lab->streams[i].handle) stream_close(lab->streams[i].handle); + u_free(lab->streams[i].payload); + if (lab->streams[i].errors) lab->errors++; + } + int errors = lab->errors; + utun_instance_destroy(lab->inst); uasync_destroy(async, 1); u_free(lab); + DEBUG_INFO(DEBUG_CATEGORY_STREAM, "lab: stopped errors=%d", errors); + debug_disable_file_output(); return errors ? 1 : 0; +}