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

tests: add standalone Merkle sync reconnect and triangle stress test

master
evgeny 5 days ago
parent
34a151e4cb
commit
3aa0cd4c06
2 changed files with 525 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. 10
      tests/Makefile.am
  2. 515
      tests/test_merkle_sync_stress.c

10
tests/Makefile.am
Unescape View File

@ -145,6 +145,16 @@ check_PROGRAMS += test_speex_aec
# Долгие тесты: запускаются только вручную
noinst_PROGRAMS =
EXTRA_PROGRAMS = test_merkle_sync_stress
CLEANFILES = test_merkle_sync_stress$(EXEEXT)
# Разрывы двух узлов и race трёх SQLite-баз; не входит в all/check/check.sh.
.PHONY: check-merkle-sync-stress
check-merkle-sync-stress: test_merkle_sync_stress$(EXEEXT)
./test_merkle_sync_stress$(EXEEXT)
test_merkle_sync_stress_SOURCES = test_merkle_sync_stress.c
test_merkle_sync_stress_LDADD = $(LIBUTUN) $(COMMON_LIBS)
# Basic includes
AM_CFLAGS = -g

515
tests/test_merkle_sync_stress.c
Unescape View File

@ -0,0 +1,515 @@
/* Долгий сетевой тест Merkle sync, запускается только вручную.
* Два узла: force-close -> 1..100 изменений -> reconnect -> полная сверка SQLite.
* Три узла: треугольник, записи во время обмена, периодические проверки сходимости.
* Модель: key BE64, version BE64, value[32]; побеждает большая версия.
* Эталон меняется только генератором локальных записей, никогда при сетевом приёме. */
#include <inttypes.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include "../lib/debug_config.h"
#include "../lib/mem.h"
#include "../lib/platform_compat.h"
#include "../lib/socket_compat.h"
#include "../src/utun_instance.h"
#include "../src/chat/merkle_sync.h"
#include "../src/routing_layer/topo_node.h"
#include "../src/transport_layer/etcp.h"
#include "../src/transport_layer/etcp_connections.h"
#include "../src/transport_layer/node_conn_direct.h"
#define RECORDS 10000
#define MAX_BATCH 100
#define RACE_BATCHES 1000
#define RACE_BURST 20
#define WIRE_SIZE 48
#define TEST_NS "1001"
#define TEST_SVC 0x72
#define SEED UINT64_C(0x4d65726b6c653130)
#define SECOND_TB UINT64_C(10000)
struct record { uint8_t wire[WIRE_SIZE]; };
struct node {
struct UTUN_INSTANCE* inst;
sqlite3_stmt* scan;
sqlite3_stmt* write;
unsigned port;
uint64_t pages, continued, received, applied, stale;
};
static struct {
struct UASYNC* ua;
struct node nodes[3];
struct NODE_CONN_DIRECT* links[3][3];
struct record expected[RECORDS];
uint64_t rng, version, started, disconnect_max;
int count, inserted, phase, batch, reconnects, callbacks;
int race_left, race_writes, race_overlap, page_interleaves;
uint64_t last_pages;
void* producer;
} test;
/* На ошибке сохраняем контекст, корни, объёмы и статистику каждого узла. */
static void fail(const char* expression, int line) {
DEBUG_ERROR(DEBUG_CATEGORY_DEBUG, "stress: FAIL seed=%016" PRIx64 " phase=%d batch=%d inserted=%d version=%" PRIu64
" line=%d: %s", SEED, test.phase, test.batch, test.inserted, test.version, line, expression);
for (int i = 0; i < test.count; i++) {
struct node* n = &test.nodes[i];
if (!n->inst) continue;
DEBUG_ERROR(DEBUG_CATEGORY_DEBUG, "stress: node=%d pages=%" PRIu64 " continued=%" PRIu64 " received=%" PRIu64
" applied=%" PRIu64 " stale=%" PRIu64 " SQL=%s", i, n->pages, n->continued, n->received, n->applied,
n->stale, sqlite3_errmsg(n->inst->topo_sqlite_db));
uint8_t root[MT_HASH_SIZE];
if (merkle_sync_read_hash(n->inst, MT_DATA_MEMBERS, TEST_NS, 0, 0, root) == 0)
log_dump(DEBUG_LEVEL_ERROR, DEBUG_CATEGORY_DEBUG, "root", root, sizeof(root));
}
exit(EXIT_FAILURE);
}
#define CHECK(expr) do { if (!(expr)) fail(#expr, __LINE__); } while (0)
static uint64_t read64(const uint8_t* p) {
uint64_t value = 0;
for (int i = 0; i < 8; i++) value = (value << 8) | p[i];
return value;
}
static void write64(uint8_t* p, uint64_t value) {
for (int i = 0; i < 8; i++) p[i] = (uint8_t)(value >> (56 - 8 * i));
}
/* SplitMix64 не зависит от платформенной реализации rand(). */
static uint64_t next_random(void) {
uint64_t z = (test.rng += UINT64_C(0x9e3779b97f4a7c15));
z = (z ^ (z >> 30)) * UINT64_C(0xbf58476d1ce4e5b9);
z = (z ^ (z >> 27)) * UINT64_C(0x94d049bb133111eb);
return z ^ (z >> 31);
}
static int compare_records(const void* a, const void* b) { return memcmp(a, b, 8); }
static void shuffled_order(int order[RECORDS]) {
for (int i = 0; i < RECORDS; i++) order[i] = i;
for (int i = RECORDS - 1; i > 0; i--) {
int j = (int)(next_random() % (uint64_t)(i + 1));
int tmp = order[i];
order[i] = order[j];
order[j] = tmp;
}
}
/* BLOB-ключи дают unsigned-порядок SQLite, включая старший бит node_id. */
static void scan_leaf(struct node* n, uint64_t prefix, int after_valid, const merkle_key_t* after) {
uint8_t first[8], last[8], zero[8] = {0};
write64(first, prefix);
write64(last, prefix | ((UINT64_C(1) << (64 - MT_MAX_LEVEL * 5)) - 1));
CHECK(sqlite3_bind_blob(n->scan, 1, first, 8, SQLITE_TRANSIENT) == SQLITE_OK);
CHECK(sqlite3_bind_blob(n->scan, 2, last, 8, SQLITE_TRANSIENT) == SQLITE_OK);
CHECK(sqlite3_bind_int(n->scan, 3, after_valid) == SQLITE_OK);
CHECK(sqlite3_bind_blob(n->scan, 4, after_valid ? after->bytes : zero, 8, SQLITE_TRANSIENT) == SQLITE_OK);
}
/* Каноническое представление записи без padding и платформенного byte order. */
static void read_record(sqlite3_stmt* st, uint8_t wire[WIRE_SIZE]) {
CHECK(sqlite3_column_bytes(st, 0) == 8 && sqlite3_column_bytes(st, 2) == 32);
CHECK(sqlite3_column_int64(st, 1) > 0);
memcpy(wire, sqlite3_column_blob(st, 0), 8);
write64(wire + 8, (uint64_t)sqlite3_column_int64(st, 1));
memcpy(wire + 16, sqlite3_column_blob(st, 2), 32);
}
/* Хеш модели вычисляется исключительно из локальной базы узла. */
static int leaf_hash(void* ctx, const char* ns, uint8_t level, uint64_t prefix, EVP_MD_CTX* sha) {
struct node* n = ctx;
CHECK(!strcmp(ns, TEST_NS) && level == MT_MAX_LEVEL);
scan_leaf(n, prefix, 0, NULL);
int rc, count = 0;
while ((rc = sqlite3_step(n->scan)) == SQLITE_ROW) {
uint8_t wire[WIRE_SIZE];
read_record(n->scan, wire);
CHECK(EVP_DigestUpdate(sha, wire, sizeof(wire)) == 1);
count++;
}
CHECK(rc == SQLITE_DONE);
CHECK(sqlite3_reset(n->scan) == SQLITE_OK);
return count;
}
/* Страница ограничена ёмкостью движка; дополнительная строка определяет more. */
static int get_page(void* ctx, const char* ns, uint64_t prefix, int after_valid, const merkle_key_t* after,
uint8_t* data, size_t* len, merkle_key_t* next, int* more) {
struct node* n = ctx;
CHECK(!strcmp(ns, TEST_NS) && *len >= WIRE_SIZE);
scan_leaf(n, prefix, after_valid, after);
size_t used = 0;
int rc;
memset(next, 0, sizeof(*next));
while ((rc = sqlite3_step(n->scan)) == SQLITE_ROW && used + WIRE_SIZE <= *len) {
read_record(n->scan, data + used);
memcpy(next->bytes, data + used, 8);
used += WIRE_SIZE;
}
CHECK(rc == SQLITE_ROW || rc == SQLITE_DONE);
*more = rc == SQLITE_ROW;
*len = used;
CHECK(sqlite3_reset(n->scan) == SQLITE_OK);
n->pages++;
n->continued += *more;
return 0;
}
/* Старая или повторно доставленная версия не должна перезаписывать новую. */
static int put_record(struct node* n, const uint8_t wire[WIRE_SIZE]) {
CHECK(read64(wire + 8) > 0 && read64(wire + 8) <= INT64_MAX);
CHECK(sqlite3_bind_blob(n->write, 1, wire, 8, SQLITE_STATIC) == SQLITE_OK);
CHECK(sqlite3_bind_int64(n->write, 2, (sqlite3_int64)read64(wire + 8)) == SQLITE_OK);
CHECK(sqlite3_bind_blob(n->write, 3, wire + 16, 32, SQLITE_STATIC) == SQLITE_OK);
CHECK(sqlite3_step(n->write) == SQLITE_DONE);
int changed = sqlite3_changes(n->inst->topo_sqlite_db);
CHECK(sqlite3_reset(n->write) == SQLITE_OK);
return changed;
}
/* Принятые записи и индекс публикуются одной транзакцией; changed будит другие сессии. */
static int apply_page(void* ctx, const char* ns, uint64_t peer, const uint8_t* data, size_t len) {
struct node* n = ctx;
CHECK(!strcmp(ns, TEST_NS) && peer != n->inst->node_id && len % WIRE_SIZE == 0);
if (!len) return 0;
uint64_t prefix = merkle_sync_level_prefix(read64(data), MT_MAX_LEVEL);
for (size_t off = 0; off < len; off += WIRE_SIZE) {
CHECK(merkle_sync_level_prefix(read64(data + off), MT_MAX_LEVEL) == prefix);
CHECK(off == 0 || memcmp(data + off - WIRE_SIZE, data + off, 8) < 0);
CHECK(read64(data + off + 8) > 0 && read64(data + off + 8) <= INT64_MAX);
}
sqlite3* db = n->inst->topo_sqlite_db;
CHECK(sqlite3_exec(db, "BEGIN", NULL, NULL, NULL) == SQLITE_OK);
int changed = 0;
for (size_t off = 0; off < len; off += WIRE_SIZE) changed += put_record(n, data + off);
if (changed) CHECK(merkle_sync_recompute_path(n->inst, MT_DATA_MEMBERS, ns, prefix) == 1);
CHECK(sqlite3_exec(db, "COMMIT", NULL, NULL, NULL) == SQLITE_OK);
if (changed) merkle_sync_changed(n->inst, MT_DATA_MEMBERS, ns);
n->received += len / WIRE_SIZE;
n->applied += changed;
n->stale += len / WIRE_SIZE - changed;
return 0;
}
static const struct merkle_sync_data_ops ops = {
.update_bucket_hash = leaf_hash, .get_page = get_page, .apply_items = apply_page
};
/* ОС выбирает порт; ядро создаётся сразу после освобождения пробного сокета. */
static unsigned free_port(void) {
socket_t fd = socket(AF_INET, SOCK_DGRAM, 0);
CHECK(fd != SOCKET_INVALID);
struct sockaddr_in addr = {0};
addr.sin_family = AF_INET;
addr.sin_addr.s_addr = htonl(INADDR_LOOPBACK);
CHECK(bind(fd, (struct sockaddr*)&addr, sizeof(addr)) == 0);
socklen_t len = sizeof(addr);
CHECK(getsockname(fd, (struct sockaddr*)&addr, &len) == 0);
unsigned port = ntohs(addr.sin_port);
socket_close_wrapper(fd);
return port;
}
/* Отдельное ядро и SQLite :memory:, без чата, TUN и фонового автоподключения групп. */
static void create_node(int index) {
struct node* n = &test.nodes[index];
struct SC_MYKEYS keys;
/* Фиксированные тестовые ключи сохраняют роли leader/follower между прогонами. */
for (int i = 0; i < 32; i++) keys.private_key[i] = (uint8_t)(49 + index * 47 + i * 13);
CHECK(sc_compute_public_key_from_private(keys.private_key, keys.public_key) == SC_OK);
char private_key[65], public_key[65], config[1024];
for (int i = 0; i < 32; i++) {
snprintf(private_key + 2 * i, 3, "%02x", keys.private_key[i]);
snprintf(public_key + 2 * i, 3, "%02x", keys.public_key[i]);
}
n->port = free_port();
snprintf(config, sizeof(config), "[global]\nmy_private_key=%s\nmy_public_key=%s\nmy_node_id=0x%016" PRIx64
"\nkeepalive_adaptive=0\n[server: udp]\naddr=127.0.0.1:%u\ntype=public\n[allowed_keys]\nallow_all=1\n",
private_key, public_key, sc_derive_node_id_from_pubkey(keys.public_key), n->port);
n->inst = utun_instance_create_from_str(test.ua, config);
CHECK(n->inst && utun_core_start(n->inst) == 0);
sqlite3* db = n->inst->topo_sqlite_db;
CHECK(sqlite3_exec(db, "CREATE TABLE stress_records(key BLOB PRIMARY KEY CHECK(length(key)=8),"
"version INTEGER NOT NULL,value BLOB NOT NULL CHECK(length(value)=32)) WITHOUT ROWID", NULL, NULL, NULL) == SQLITE_OK);
CHECK(sqlite3_prepare_v2(db, "SELECT key,version,value FROM stress_records WHERE key>=?1 AND key<=?2 "
"AND (?3=0 OR key>?4) ORDER BY key", -1, &n->scan, NULL) == SQLITE_OK);
CHECK(sqlite3_prepare_v2(db, "INSERT INTO stress_records VALUES(?1,?2,?3) ON CONFLICT(key) DO UPDATE SET "
"version=excluded.version,value=excluded.value WHERE excluded.version>stress_records.version",
-1, &n->write, NULL) == SQLITE_OK);
CHECK(merkle_sync_init(n->inst, TEST_SVC, &ops, n) == 0);
DEBUG_INFO(DEBUG_CATEGORY_DEBUG, "stress: node=%d ready id=%016" PRIx64 " port=%u", index, n->inst->node_id, n->port);
}
/* Обе стороны держат ровно по одному handle: force_close не оставляет чужих владельцев. */
static void open_link(int a, int b) {
CHECK(!test.links[a][b]);
struct node* dest = &test.nodes[b];
struct TOPO_ADDR4 address = { .addr = {127, 0, 0, 1}, .port = dest->port, .protocol = TOPO_PROTO_UDP };
struct TOPO_NODE node = { .node_id = dest->inst->node_id, .v4_addrs = &address };
memcpy(node.public_key, dest->inst->my_keys.public_key, 32);
CHECK(node_conn_direct_open_node(test.nodes[a].inst, node.node_id, NULL, NULL, &test.links[a][b], &node, NULL) != NCD_ERR);
}
static int transport_ready(int a, int b) {
struct ETCP_CONN* c = instance_find_conn(test.nodes[a].inst, test.nodes[b].inst->node_id);
return c && c->initialized && c->links_up && !c->close_requested && c->state != 2;
}
/* Инициируем одну сторону, а входящий транспорт берём во владение после UP. */
static void connect_pair(int a, int b) {
open_link(a, b);
uint64_t deadline = get_time_tb() + 10 * SECOND_TB;
while (!transport_ready(a, b) || !transport_ready(b, a)) {
CHECK(get_time_tb() < deadline);
uasync_poll(test.ua, 1);
}
open_link(b, a);
}
/* Только обработка отложенного close, без ожидания keepalive или fin_wait. */
static void disconnect_pair(int a, int b) {
uint64_t started = get_time_tb();
CHECK(test.links[a][b] && test.links[b][a]);
node_conn_direct_force_close(test.links[a][b]);
node_conn_direct_force_close(test.links[b][a]);
test.links[a][b] = test.links[b][a] = NULL;
while (instance_find_conn(test.nodes[a].inst, test.nodes[b].inst->node_id)
|| instance_find_conn(test.nodes[b].inst, test.nodes[a].inst->node_id)) {
CHECK(get_time_tb() - started < SECOND_TB);
uasync_poll(test.ua, 0);
}
uint64_t elapsed = get_time_tb() - started;
if (elapsed > test.disconnect_max) test.disconnect_max = elapsed;
test.reconnects++;
}
static void sync_done(uint64_t peer, const char* ns, int result, void* arg) {
(void)arg;
if (result != MT_OK) DEBUG_ERROR(DEBUG_CATEGORY_DEBUG, "stress: sync peer=%016" PRIx64 " ns=%s result=%d", peer, ns, result);
CHECK(result == MT_OK && !strcmp(ns, TEST_NS));
test.callbacks++;
}
/* Запрос контрольной точки только у leader; follower создаёт сессию по входящему BEGIN. */
static void start_pair(int a, int b) {
if (test.nodes[a].inst->node_id > test.nodes[b].inst->node_id) { int tmp = a; a = b; b = tmp; }
CHECK(merkle_sync_start(test.nodes[a].inst, MT_DATA_MEMBERS, test.nodes[b].inst->node_id, TEST_NS, sync_done, NULL) == 0);
}
static int roots_equal(void) {
uint8_t first[MT_HASH_SIZE], current[MT_HASH_SIZE];
CHECK(merkle_sync_read_hash(test.nodes[0].inst, MT_DATA_MEMBERS, TEST_NS, 0, 0, first) == 0);
for (int i = 1; i < test.count; i++) {
CHECK(merkle_sync_read_hash(test.nodes[i].inst, MT_DATA_MEMBERS, TEST_NS, 0, 0, current) == 0);
if (memcmp(first, current, sizeof(first))) return 0;
}
return 1;
}
/* Сравнение с независимым эталоном обнаружит одинаково неверные базы и лишние записи. */
static void verify_databases(void) {
for (int i = 0; i < test.count; i++) {
sqlite3_stmt* st = NULL;
CHECK(sqlite3_prepare_v2(test.nodes[i].inst->topo_sqlite_db,
"SELECT key,version,value FROM stress_records ORDER BY key", -1, &st, NULL) == SQLITE_OK);
int count = 0;
for (int r = 0; r < RECORDS; r++) {
const uint8_t* expected = test.expected[r].wire;
if (!read64(expected + 8)) continue;
CHECK(sqlite3_step(st) == SQLITE_ROW);
uint8_t actual[WIRE_SIZE];
read_record(st, actual);
if (memcmp(actual, expected, WIRE_SIZE)) {
DEBUG_ERROR(DEBUG_CATEGORY_DEBUG, "stress: data mismatch node=%d record=%d key=%016" PRIx64
" expected_version=%" PRIu64 " actual_key=%016" PRIx64 " actual_version=%" PRIu64,
i, r, read64(expected), read64(expected + 8), read64(actual), read64(actual + 8));
log_dump(DEBUG_LEVEL_ERROR, DEBUG_CATEGORY_DEBUG, "expected", expected, WIRE_SIZE);
log_dump(DEBUG_LEVEL_ERROR, DEBUG_CATEGORY_DEBUG, "actual", actual, WIRE_SIZE);
CHECK(0);
}
count++;
}
CHECK(count == test.inserted && sqlite3_step(st) == SQLITE_DONE);
CHECK(sqlite3_finalize(st) == SQLITE_OK);
}
}
/* Timeout — страховка зависания теста, разрыв связи никогда не ждёт сетевого timeout. */
static void wait_converged(int callbacks) {
uint64_t deadline = get_time_tb() + 120 * SECOND_TB;
while (test.callbacks < callbacks || !roots_equal()) {
CHECK(get_time_tb() < deadline);
uasync_poll(test.ua, 1);
}
CHECK(test.callbacks == callbacks);
verify_databases();
}
/* Меняем память эталона и одну SQLite; листья пачки пересчитываются по одному разу. */
static void mutate(int writer, const int* indices, int size, int inserting) {
struct node* n = &test.nodes[writer];
sqlite3* db = n->inst->topo_sqlite_db;
uint64_t prefixes[MAX_BATCH];
int leaves = 0;
CHECK(size >= 1 && size <= MAX_BATCH);
CHECK(sqlite3_exec(db, "BEGIN", NULL, NULL, NULL) == SQLITE_OK);
for (int i = 0; i < size; i++) {
uint8_t* wire = test.expected[indices[i]].wire;
CHECK((read64(wire + 8) == 0) == inserting);
if (inserting) test.inserted++;
write64(wire + 8, ++test.version);
for (int j = 16; j < WIRE_SIZE; j++) wire[j] = (uint8_t)next_random();
wire[16] = (uint8_t)writer;
CHECK(put_record(n, wire) == 1);
uint64_t prefix = merkle_sync_level_prefix(read64(wire), MT_MAX_LEVEL);
int p = 0;
while (p < leaves && prefixes[p] != prefix) p++;
if (p == leaves) prefixes[leaves++] = prefix;
}
for (int i = 0; i < leaves; i++)
CHECK(merkle_sync_recompute_path(n->inst, MT_DATA_MEMBERS, TEST_NS, prefixes[i]) == 1);
CHECK(sqlite3_exec(db, "COMMIT", NULL, NULL, NULL) == SQLITE_OK);
merkle_sync_changed(n->inst, MT_DATA_MEMBERS, TEST_NS);
DEBUG_DEBUG(DEBUG_CATEGORY_DEBUG, "stress: mutate phase=%d batch=%d writer=%d size=%d leaves=%d version=%" PRIu64,
test.phase, test.batch, writer, size, leaves, test.version);
}
/* После каждого разрыва меняем случайную пачку, затем ждём новый обмен и сверяем обе базы. */
static void reconnect_phase(int inserting) {
int order[RECORDS];
shuffled_order(order);
test.phase = inserting ? 1 : 2;
test.batch = 0;
for (int done = 0; done < RECORDS;) {
disconnect_pair(0, 1);
int size = 1 + (int)(next_random() % MAX_BATCH);
if (size > RECORDS - done) size = RECORDS - done;
test.batch++;
mutate(test.batch % 2, order + done, size, inserting);
connect_pair(0, 1);
int callbacks = test.callbacks + 1;
start_pair(0, 1);
wait_converged(callbacks);
done += size;
if (test.batch == 1 || test.batch % 20 == 0 || done == RECORDS)
DEBUG_INFO(DEBUG_CATEGORY_DEBUG, "stress: reconnect verified phase=%d batch=%d size=%d done=%d/%d records=%d"
" closes=%d close_max=%.1fms elapsed=%.1fs", test.phase, test.batch, size, done, RECORDS, test.inserted,
test.reconnects, test.disconnect_max / 10.0, (get_time_tb() - test.started) / (double)SECOND_TB);
}
}
/* Короткий таймер перемежает локальные записи с обработкой настоящих сетевых страниц. */
static void race_write(void* arg) {
(void)arg;
test.producer = NULL;
if (!roots_equal()) test.race_overlap++;
uint64_t pages = 0;
for (int i = 0; i < 3; i++) pages += test.nodes[i].pages;
if (test.race_left < RACE_BURST && pages > test.last_pages) test.page_interleaves++;
test.last_pages = pages;
int size = 1 + (int)(next_random() % MAX_BATCH), indices[MAX_BATCH];
for (int i = 0; i < size; i++) {
/* Общий небольшой набор горячих ключей создаёт повторные изменения и старые версии. */
indices[i] = (int)(next_random() % (i % 4 == 0 ? 32 : RECORDS));
}
test.batch++;
mutate(test.race_writes % 3, indices, size, 0);
test.race_writes++;
if (--test.race_left) {
test.producer = uasync_set_timeout(test.ua, 10 + next_random() % 40, NULL, race_write, "merkle stress writer");
CHECK(test.producer != NULL);
}
}
/* Треугольник остаётся подключённым; после подписки дальнейшие обмены запускает только changed. */
static void race_phase(void) {
test.phase = 3;
test.batch = 0;
test.count = 3;
create_node(2);
connect_pair(0, 2);
connect_pair(1, 2);
int callbacks = test.callbacks + 2;
start_pair(0, 2);
start_pair(1, 2);
wait_converged(callbacks);
DEBUG_INFO(DEBUG_CATEGORY_DEBUG, "stress: triangle ready, all three databases contain %d records", test.inserted);
for (int burst = 0; burst < RACE_BATCHES / RACE_BURST; burst++) {
test.race_left = RACE_BURST;
test.producer = uasync_set_timeout(test.ua, 1, NULL, race_write, "merkle stress writer");
CHECK(test.producer != NULL);
uint64_t deadline = get_time_tb() + 120 * SECOND_TB;
while (test.race_left) {
CHECK(get_time_tb() < deadline);
uasync_poll(test.ua, 1);
}
wait_converged(test.callbacks);
DEBUG_INFO(DEBUG_CATEGORY_DEBUG, "stress: triangle verified batches=%d/%d mutations=%" PRIu64
" overlapping=%d page_interleaves=%d elapsed=%.1fs", test.race_writes, RACE_BATCHES, test.version,
test.race_overlap, test.page_interleaves, (get_time_tb() - test.started) / (double)SECOND_TB);
}
CHECK(test.race_overlap > RACE_BATCHES / 2 && test.page_interleaves > RACE_BATCHES / 4);
uint64_t continued = 0, stale = 0;
for (int i = 0; i < 3; i++) { continued += test.nodes[i].continued; stale += test.nodes[i].stale; }
CHECK(continued > 0 && stale > 0);
DEBUG_INFO(DEBUG_CATEGORY_DEBUG, "stress: race coverage continued_pages=%" PRIu64 " stale_or_duplicate_records=%" PRIu64,
continued, stale);
}
int main(void) {
debug_config_init();
debug_set_level(DEBUG_LEVEL_WARN);
/* Полная сверка 10k строк между poll намеренно занимает >10ms; ошибки SYS сохраняем. */
debug_set_category_level(DEBUG_CATEGORY_SYS, DEBUG_LEVEL_ERROR);
debug_set_category_level(DEBUG_CATEGORY_DEBUG, DEBUG_LEVEL_INFO);
if (getenv("UTUN_TEST_DEBUG")) {
debug_set_category_level(DEBUG_CATEGORY_DEBUG, DEBUG_LEVEL_DEBUG);
debug_set_category_level(DEBUG_CATEGORY_MEMBER_SYNC, DEBUG_LEVEL_DEBUG);
}
test.rng = SEED;
test.started = get_time_tb();
DEBUG_INFO(DEBUG_CATEGORY_DEBUG, "stress: start seed=%016" PRIx64 " records=%d batch=1..100 race_batches=%d",
SEED, RECORDS, RACE_BATCHES);
/* 1024 записей одного листа требуют нескольких страниц; остальные распределены по дереву. */
for (uint32_t i = 0; i < RECORDS; i++) {
uint64_t key = i < 1024 ? i : ((uint64_t)(i * UINT32_C(2654435761)) << 32) | i;
write64(test.expected[i].wire, key);
}
qsort(test.expected, RECORDS, sizeof(test.expected[0]), compare_records);
for (int i = 1; i < RECORDS; i++) CHECK(memcmp(test.expected[i - 1].wire, test.expected[i].wire, 8) < 0);
test.ua = uasync_create();
CHECK(test.ua != NULL);
test.count = 2;
create_node(0);
create_node(1);
connect_pair(0, 1);
reconnect_phase(1);
reconnect_phase(0);
race_phase();
CHECK(test.producer == NULL);
DEBUG_INFO(DEBUG_CATEGORY_DEBUG, "stress: cleanup nodes=%d reconnects=%d callbacks=%d", test.count, test.reconnects, test.callbacks);
/* Освобождаем владельцев без нового poll: destroy ниже сразу закрывает весь транспорт. */
for (int a = 0; a < test.count; a++) for (int b = 0; b < test.count; b++) {
if (test.links[a][b]) node_conn_direct_close(test.links[a][b]);
test.links[a][b] = NULL;
}
for (int i = 0; i < test.count; i++) {
struct node* n = &test.nodes[i];
merkle_sync_destroy(n->inst);
CHECK(sqlite3_finalize(n->scan) == SQLITE_OK && sqlite3_finalize(n->write) == SQLITE_OK);
utun_instance_destroy(n->inst);
n->inst = NULL;
}
uasync_poll(test.ua, 0);
CHECK(test.ua->timer_alloc_count == test.ua->timer_free_count);
CHECK(test.ua->socket_alloc_count == test.ua->socket_free_count);
uasync_destroy(test.ua, 0);
DEBUG_INFO(DEBUG_CATEGORY_DEBUG, "stress: PASS 10000 inserts + 10000 updates with reconnect; 1000 triangle race batches, elapsed=%.1fs",
(get_time_tb() - test.started) / (double)SECOND_TB);
return EXIT_SUCCESS;
}
Write Preview
Loading…
Cancel
Save