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

Add real uTun stream bench node with payload and latency validation

master
evgeny 1 day ago
parent
856fbc61b1
commit
6332e75e61
2 changed files with 382 additions and 1 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. 10
      tests/Makefile.am
  2. 373
      tests/stream_netns/node.c

10
tests/Makefile.am
Unescape View File

@ -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-условий.

373
tests/stream_netns/node.c
Unescape View File

@ -0,0 +1,373 @@
/* Агент стенда: настоящие JOIN/Merkle/ETCP и автоматический выбор поставщиков.
* stdin/stdout — только JSONL RPC; диагностика ядра пишется в отдельный файл.
* Заголовок нагрузки и SHA256 проверяют opaque DATA после восстановления FEC. */
#include <errno.h>
#include <fcntl.h>
#include <inttypes.h>
#include <signal.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <time.h>
#include <unistd.h>
#include <openssl/sha.h>
#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;
}
Write Preview
Loading…
Cancel
Save