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

Exercise automatic stream selection, INFO-only probes and verified overlap failover

master
evgeny 3 days ago
parent
99359f987c
commit
bc49b452f9
1 changed files with 125 additions and 0 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. 125
      tests/test_stream_service.c

125
tests/test_stream_service.c
Unescape View File

@ -46,6 +46,11 @@ static size_t stale_select_length;
static etcp_recv_fn pressure_handlers[4];
static int simulated_cpu;
static unsigned global_data[4];
static unsigned automatic_data[4], automatic_gaps[4], automatic_end[4], automatic_probes[4], automatic_accepts;
static uint64_t automatic_next[4];
static int automatic_drop_relay;
static etcp_recv_fn automatic_handlers[4];
static uint8_t automatic_request[STREAM_WIRE_HEADER + 1];
err_t __real_uasync_get_load(struct UASYNC *async, struct uasync_load *load);
err_t __wrap_uasync_get_load(struct UASYNC *async, struct uasync_load *load) {
err_t result = __real_uasync_get_load(async, load);
@ -60,6 +65,36 @@ static void on_global_pressure(struct UTUN_INSTANCE *inst, const struct stream_e
else CHECK(event->type == STREAM_EVENT_GAP || event->type == STREAM_EVENT_CLOSED || event->type == STREAM_EVENT_END);
}
static void automatic_raw(struct ETCP_CONN *conn, struct ll_entry *entry) {
unsigned node = 0; while (node < 4 && nodes[node] != conn->instance) node++; CHECK(node < 4);
if (entry->len >= STREAM_WIRE_HEADER) {
uint8_t command = entry->dgram[2];
if (command == STREAM_CMD_INFO_REQ) automatic_probes[node]++;
if (command == STREAM_CMD_ACCEPT) automatic_accepts++;
if (node == 3 && conn->peer_node_id == nodes[2]->node_id && command == STREAM_CMD_SUBSCRIBE) {
memcpy(automatic_request, entry->dgram, STREAM_WIRE_HEADER);
automatic_request[2] = STREAM_CMD_UNSUBSCRIBE; automatic_request[STREAM_WIRE_HEADER] = STREAM_REASON_TRANSPORT;
}
if (automatic_drop_relay && node == 2 && conn->peer_node_id == nodes[1]->node_id && command == STREAM_CMD_DATA) {
queue_dgram_free(entry); queue_entry_free(entry); return;
}
}
automatic_handlers[node](conn, entry);
}
static void on_automatic(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 && !memcmp(event->data, pressure_payload, 1180));
CHECK(event->first_seq == automatic_next[node]++); automatic_data[node]++;
} else if (event->type == STREAM_EVENT_GAP) {
CHECK(event->first_seq == automatic_next[node] && event->last_seq >= event->first_seq);
automatic_next[node] = event->last_seq + 1; automatic_gaps[node]++;
} else if (event->type == STREAM_EVENT_END) automatic_end[node]++;
else CHECK(event->type == STREAM_EVENT_CLOSED);
}
static void pressure_raw(struct ETCP_CONN *conn, struct ll_entry *entry) {
unsigned node = 0; while (node < 4 && nodes[node] != conn->instance) node++; CHECK(node < 4);
if (entry->len >= STREAM_WIRE_HEADER) {
@ -567,6 +602,95 @@ static void test_global_pressure(void) {
DEBUG_INFO(DEBUG_CATEGORY_STREAM, "[PASS] global overload: CPU alone preserves flows; confirmed multi-peer backlog releases a shared rate budget");
}
/* Default API сам выбирает выдачи: пустой источник, общий старт, цикл 30 с, overlap, stall и реальный CLOSED. */
static void test_automatic(void) {
test_clock_run_for(ua, 10000);
struct STREAM_HANDLE *publisher = NULL, *subscribers[3] = {0}, *second = NULL, *manual = NULL;
struct stream_parameters parameters = {.n = 2, .r = 1, .block_size = STREAM_RS_DEFAULT_SIZE};
CHECK(stream_publish_open(nodes[3], GROUP2, &parameters, &publisher) == STREAM_OK);
struct stream_info origin; CHECK(stream_get_info(publisher, &origin) == STREAM_OK);
for (unsigned i = 1; i < 3; i++) wait_state(i, 3, GROUP2, origin.descriptor.source_epoch, 0);
for (unsigned i = 1; i < 4; i++) {
automatic_handlers[i] = nodes[i]->api_bindings.callbacks[ETCP_ID_STREAM];
CHECK(etcp_bind(nodes[i], ETCP_ID_STREAM, automatic_raw) == 0);
}
CHECK(node_conn_direct_compatible(nodes[1], nodes[3]->node_id));
CHECK(stream_subscribe_manual_open(nodes[1], GROUP2, nodes[3]->node_id, &manual) == STREAM_OK);
struct STREAM_RECEIVE *probe = NULL;
CHECK(stream_probe_open(manual, nodes[3]->node_id, 0, &probe) == STREAM_OK);
struct stream_receive_info probe_info; uint64_t deadline = get_time_tb() + 100000;
do { test_clock_poll(ua, 100); CHECK(stream_receive_get_info(probe, &probe_info) == STREAM_OK); }
while (!probe_info.info_ready && get_time_tb() < deadline);
CHECK(probe_info.info_ready && probe_info.probe_only && !probe_info.handle && !probe_info.blocks && !automatic_accepts);
CHECK(probe_info.first_available == UINT64_MAX && probe_info.last_available == UINT64_MAX);
CHECK(probe_info.path.count == 1 && probe_info.path.nodes[0] == nodes[3]->node_id);
CHECK(probe_info.accepting && probe_info.free_slots && probe_info.path_rtt_valid && probe_info.rtt_valid);
CHECK(stream_probe_next(probe, 1) == STREAM_OK); deadline = get_time_tb() + 100000;
do { test_clock_poll(ua, 100); CHECK(stream_receive_get_info(probe, &probe_info) == STREAM_OK); }
while (!probe_info.info_ready && get_time_tb() < deadline);
CHECK(probe_info.info_ready && probe_info.subblock == 1 && !automatic_accepts);
stream_receive_close(probe); stream_close(manual); manual = NULL;
CHECK(stream_subscribe_open(nodes[1], GROUP2, nodes[3]->node_id, &subscribers[1]) == STREAM_OK);
CHECK(stream_subscribe_open(nodes[1], GROUP2, nodes[3]->node_id, &second) == STREAM_OK);
CHECK(stream_subscribe_manual_open(nodes[1], GROUP2, nodes[3]->node_id, &manual) == STREAM_ERR_STATE && !manual);
wait_state(2, 1, GROUP2, origin.descriptor.source_epoch, 1);
CHECK(stream_subscribe_open(nodes[2], GROUP2, nodes[3]->node_id, &subscribers[2]) == STREAM_OK);
struct stream_selection_settings settings = {200, 1000, 100};
for (unsigned i = 1; i < 3; i++) {
CHECK(stream_selection_configure(subscribers[i], &settings) == STREAM_OK);
stream_set_event_callback(nodes[i], on_automatic, NULL);
}
test_clock_run_for(ua, 3000);
CHECK(!automatic_data[1] && !automatic_data[2]);
memset(pressure_payload, 0xa9, 1180); 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);
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);
CHECK(state[i].automatic && state[i].subflows == 2 && state[i].cycles == 1);
if (frame == 80) initial_probes[i] = state[i].probes_started;
else CHECK(state[i].probes_started == initial_probes[i]);
}
}
for (unsigned i = 1; i < 3; i++) {
CHECK(stream_selection_get_info(subscribers[i], &state[i]) == STREAM_OK);
CHECK(state[i].cycles == 2 && state[i].probes_started > initial_probes[i] && state[i].subflows == 2);
CHECK(automatic_data[i] == 350 && !automatic_gaps[i]);
}
CHECK(state[2].migrations && state[2].providers == 2 && state[2].protected_by_fec);
CHECK(!automatic_probes[2]); /* Mobile relay не пробуется ни одним планировщиком. */
struct stream_selection_info same; CHECK(stream_selection_get_info(second, &same) == STREAM_OK);
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);
test_clock_run_for(ua, 1200);
}
CHECK(stream_selection_get_info(subscribers[2], &state[2]) == STREAM_OK);
CHECK(state[2].migrations > migrations && state[2].providers == 1 && state[2].subflows == 2);
CHECK(automatic_data[2] == 380 && !automatic_gaps[2]); automatic_drop_relay = 0;
CHECK(automatic_request[0]); uint64_t failures = state[2].failures;
struct ETCP_CONN *source_connection = instance_find_conn(nodes[3], nodes[2]->node_id); CHECK(source_connection && source_connection->links_up);
struct ll_entry *cancel = ll_alloc_lldgram(sizeof(automatic_request)); CHECK(cancel);
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);
test_clock_run_for(ua, 1200);
}
CHECK(stream_selection_get_info(subscribers[2], &state[2]) == STREAM_OK);
CHECK(state[2].failures > failures && state[2].providers == 2 && state[2].subflows == 2);
CHECK(automatic_data[2] == 400 && !automatic_gaps[2] && source_connection->links_up);
CHECK(stream_publish_finish(publisher) == STREAM_OK); run_until(&automatic_end[1], 1); run_until(&automatic_end[2], 1);
for (unsigned i = 1; i < 3; i++) stream_close(subscribers[i]); stream_close(publisher);
test_clock_run_for(ua, 1000);
for (unsigned i = 1; i < 4; i++) CHECK(etcp_bind(nodes[i], ETCP_ID_STREAM, automatic_handlers[i]) == 0);
stream_set_event_callback(nodes[1], on_stream, NULL); stream_set_event_callback(nodes[2], NULL, NULL);
DEBUG_INFO(DEBUG_CATEGORY_STREAM, "[PASS] automatic streams: INFO-only empty source, shared owners, 30s cycle, FEC spread, overlap, stall and CLOSED failover");
}
/* Ошибка размера/SQLite не меняет подпись, userinfo, версию и производное Merkle-дерево. */
static void test_profile_atomicity(void) {
char before[MS_USERINFO_MAX + 1], after[MS_USERINFO_MAX + 1], huge[MS_USERINFO_MAX + 1];
@ -713,6 +837,7 @@ int main(void) {
test_network();
test_pressure();
test_global_pressure();
test_automatic();
/* GROUP2 не соединена, но подписанный профиль можно читать из общей таблицы. */
struct stream_info second; CHECK(stream_get_info(publisher2, &second) == STREAM_OK);

Write Preview
Loading…
Cancel
Save