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

tests: keep Merkle stress records changing and verify final versions

master
evgeny 5 days ago
parent
17777c540b
commit
7506da41f0
1 changed files with 59 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. 68
      tests/test_merkle_sync.c

68
tests/test_merkle_sync.c
Unescape View File

@ -969,6 +969,9 @@ struct str_state {
struct UTUN_INSTANCE* inst[STRESS_N];
struct ms_data data[STRESS_N];
struct ms_test_ctx ctx[STRESS_N];
uint32_t versions[STRESS_N]; /* Последняя версия каждого владельца записи. */
unsigned int observer_starts[2];
uint64_t spam_started_us;
void* global_timeout; /* setup: links-up timeout */
void* safety_timer; /* spam+converge safety */
void* stop_timer; /* spam stop */
@ -996,9 +999,14 @@ static void _str_verify(void);
static void _str_fail_timeout(void* arg);
static void _str_sync_done(uint64_t peer, const char* ns, int result, void* arg) {
(void)peer; (void)ns; (void)result;
struct str_state* s = (struct str_state*)arg;
if (s) __sync_fetch_and_sub(&s->sync_pending, 1);
if (!s || s->phase != 0) return;
if (result != MT_OK) {
DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "stress initial sync: peer=%016llx ns=%s result=%d",
(unsigned long long)peer, ns, result);
s->phase = 2;
}
__sync_fetch_and_sub(&s->sync_pending, 1);
}
static void _str_spam_cb(void* arg);
@ -1016,7 +1024,7 @@ static void _str_spam_cb(void* arg) {
/* spammer: bump own member, sync to hub */
uint64_t key = ((uint64_t)idx) << 60;
_data_insert(&gs->data[idx], key, (uint32_t)(gs->data[idx].count + 1));
_data_insert(&gs->data[idx], key, ++gs->versions[idx]);
merkle_sync_recompute_path(gs->inst[idx], MT_DATA_MEMBERS, MS_NS_STRESS, key);
merkle_sync_start(gs->inst[idx], MT_DATA_MEMBERS, gs->inst[STRESS_HUB]->node_id, MS_NS_STRESS, NULL, NULL);
@ -1027,6 +1035,7 @@ static void _str_obs_spam_cb(void* arg) {
if (!gs || !gs->spam_active) return;
int idx = (int)(intptr_t)arg;
gs->observer_starts[idx - STRESS_OBS_A]++;
merkle_sync_start(gs->inst[idx], MT_DATA_MEMBERS, gs->inst[STRESS_HUB]->node_id, MS_NS_STRESS, NULL, NULL);
int delay_tb = ((rand() % 10) + 1) * 10; /* 1-10ms */
@ -1060,7 +1069,7 @@ static int _str_wait_for(const char* desc, int (*cond)(void), int timeout_tb) {
uint64_t start = get_time_tb();
while (!cond() && (get_time_tb() - start) < (uint64_t)timeout_tb && s->phase == 0)
uasync_poll(s->ua, 1);
if (cond()) return 1;
if (s->phase == 0 && cond()) return 1;
printf(" STRESS TIMEOUT: %s\n", desc); s->phase = 2;
return 0;
}
@ -1144,14 +1153,15 @@ static int _str_init(void) {
/* --- seed each instance with its own member; sync to hub --- */
for (int i = 1; i < STRESS_N; i++) {
uint64_t key = ((uint64_t)i) << 60;
_data_insert(&s->data[i], key, (uint32_t)i);
s->versions[i] = (uint32_t)i;
_data_insert(&s->data[i], key, s->versions[i]);
merkle_sync_recompute_path(s->inst[i], MT_DATA_MEMBERS, MS_NS_STRESS, key);
}
/* initial sync: each spoke → hub */
s->sync_pending = STRESS_N - 1;
for (int i = 1; i < STRESS_N; i++)
merkle_sync_start(s->inst[i], MT_DATA_MEMBERS, s->inst[STRESS_HUB]->node_id, MS_NS_STRESS, _str_sync_done, s);
_str_wait_for("initial sync", _str_cond_sync_done, STRESS_SYNC_TB);
if (!_str_wait_for("initial sync", _str_cond_sync_done, STRESS_SYNC_TB)) goto fail;
/* observers: sync with hub too */
@ -1192,7 +1202,9 @@ static void _str_stop_spam(void* arg) {
if (!gs) return;
gs->stop_timer = NULL;
gs->spam_active = 0;
printf(" spam stopped\n");
printf(" spam stopped after %lluus; writer updates:", (unsigned long long)(get_time_us() - gs->spam_started_us));
for (int i = STRESS_SPAM_BEGIN; i <= STRESS_SPAM_END; i++) printf(" %u", gs->versions[i] - (uint32_t)i);
printf("; observer starts: %u %u\n", gs->observer_starts[0], gs->observer_starts[1]);
/* короткий дренаж — даём in-flight спам-синкам утихнуть перед финальными раундами */
gs->drain_timer = uasync_set_timeout(gs->ua, STRESS_DRAIN_TB, NULL, _str_drain_cb, "str_drain");
}
@ -1215,8 +1227,14 @@ static void _str_start_final_round(void) {
}
static void _str_final_done(uint64_t peer, const char* ns, int result, void* arg) {
(void)peer; (void)ns; (void)result; (void)arg;
if (!gs) return;
(void)arg;
if (!gs || gs->phase != 0) return;
if (result != MT_OK) {
DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "stress final sync: round=%d peer=%016llx ns=%s result=%d",
gs->final_round, (unsigned long long)peer, ns, result);
gs->phase = 2;
return;
}
gs->final_pending--;
if (gs->final_pending <= 0) _str_check_roots();
}
@ -1225,6 +1243,37 @@ static void _str_verify(void) {
struct str_state* s = gs;
int ok = 1;
// Нагрузка должна менять записи; равные корни сами по себе не доказывают доставку последних версий.
for (int owner = STRESS_SPAM_BEGIN; owner <= STRESS_SPAM_END; owner++) {
if (s->versions[owner] - (uint32_t)owner >= 10) continue;
DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "stress writer=%d made fewer than ten updates", owner);
ok = 0;
}
if (s->observer_starts[0] < 10 || s->observer_starts[1] < 10) {
DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "stress observer starts too few: %u %u",
s->observer_starts[0], s->observer_starts[1]);
ok = 0;
}
for (int i = 0; i < STRESS_N && ok; i++) {
if (s->data[i].count != STRESS_N - 1) {
DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "stress inst=%d items=%d expected=%d", i, s->data[i].count, STRESS_N - 1);
ok = 0;
break;
}
for (int owner = 1; owner < STRESS_N && ok; owner++) {
uint64_t key = ((uint64_t)owner) << 60;
int found = 0;
for (int j = 0; j < s->data[i].count; j++) {
if (s->data[i].items[j].key == key && s->data[i].items[j].val == s->versions[owner]) { found = 1; break; }
}
if (!found) {
DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "stress inst=%d missing final version: owner=%d version=%u",
i, owner, s->versions[owner]);
ok = 0;
}
}
}
/* tree sizes match pairwise */
int na = _db_count_rows(s->inst[STRESS_HUB]->topo_sqlite_db, MS_NS_STRESS);
for (int i = 1; i < STRESS_N && ok; i++) {
@ -1314,6 +1363,7 @@ static void test_stress_spam(void) {
/* --- start spam --- */
s->spam_active = 1;
s->spam_started_us = get_time_us();
srand(12345);
for (int i = STRESS_SPAM_BEGIN; i <= STRESS_SPAM_END; i++) {
/* stagger initial delays */

Write Preview
Loading…
Cancel
Save