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

test_chat_sync_stress: simplify to 20 batches, add profiling (avg per-record time, sync time, total)

topo_upd
Evgeny 3 months ago
parent
bb425c6722
commit
03310db2cd
1 changed files with 61 additions and 72 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. 109
      tests/test_chat_sync_stress.c

109
tests/test_chat_sync_stress.c
Unescape View File

@ -1,4 +1,4 @@
// test_chat_sync_stress.c — 10 раундов: 100 батчей вставок + 1 синхронизация за раунд
// test_chat_sync_stress.c — 20 батчей случайных вставок на A и B, синхронизация, замер производительности
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
@ -20,11 +20,10 @@
#include "../lib/u_async.h"
#include "../lib/debug_config.h"
#define ROUNDS 10
#define BATCHES_PER_ROUND 100
#define TOTAL_BATCHES 20
#define MAX_PER_BATCH 100
#define SYNC_TIMEOUT_TB 300000 // 30s
#define TOTAL_TIMEOUT_TB (ROUNDS * SYNC_TIMEOUT_TB + 600000)
#define TOTAL_TIMEOUT_TB 600000 // 60s
#define POLL_INTERVAL_MS 5
#define NODE_ID_A 0xAAAAAAAAAAAAAAAAULL
@ -98,7 +97,7 @@ static void cleanup_temp_configs(void) {
static void test_timeout(void* arg) { (void)arg; test_phase = 2; }
static int wait_for(const char* desc, int timeout_tb) {
static int wait_for(const char* desc, int timeout_tb, uint64_t* elapsed_out) {
uint64_t start = get_time_tb();
uint32_t ca0 = si_a ? db_sync_count(si_a) : 0, cb0 = si_b ? db_sync_count(si_b) : 0;
DEBUG_WARN(DEBUG_CATEGORY_DEBUG, "SYNC_START %s A=%u B=%u expect=%u", desc, ca0, cb0, expected_total);
@ -115,28 +114,10 @@ static int wait_for(const char* desc, int timeout_tb) {
desc, ca, cb, expected_total, (unsigned long long)elapsed);
if (test_phase == 0) test_phase = 2;
}
if (elapsed_out) *elapsed_out = (ok ? elapsed : 0);
return ok;
}
static int insert_batch(struct DB_SYNC_INSTANCE* si, int seq, int count, uint64_t node_id) {
if (count == 0) return 0;
uint64_t t0 = get_time_tb();
char buf[384];
for (int i = 0; i < count && test_phase == 0; i++) {
snprintf(buf, sizeof(buf),
"{\"seq\":%d,\"n\":%llu,\"ch\":\"test\",\"ct\":\"text\",\"d\":\"msg_%d\","
"\"pad\":\"xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx\"}",
seq + i, (unsigned long long)node_id, seq + i);
if (db_sync_insert(si, buf) < 0) {
DEBUG_ERROR(DEBUG_CATEGORY_DEBUG, "INSERT_FAIL seq=%d", seq + i);
return -1;
}
}
uint64_t dt = (get_time_tb() - t0) / 10;
if (dt > 50) DEBUG_ERROR(DEBUG_CATEGORY_DEBUG, "INSERT_BATCH %d recs %llums", count, (unsigned long long)dt);
return 0;
}
int main(void) {
g_seed = (unsigned int)time(NULL);
srand(g_seed);
@ -164,7 +145,7 @@ int main(void) {
uint64_t t0 = get_time_tb();
// Phase 1: local
// Phase 1: local inserts on A
if (db_sync_insert(si_a, "{\"test\":1}") != 0
|| db_sync_insert(si_a, "{\"test\":2}") != 0
|| db_sync_count(si_a) != 2) {
@ -173,60 +154,68 @@ int main(void) {
}
DEBUG_WARN(DEBUG_CATEGORY_DEBUG, "PHASE1_OK A=%u", db_sync_count(si_a));
// Phase 2: initial sync
// Phase 2: initial sync A→B
expected_total = 2;
if (!wait_for("phase2", SYNC_TIMEOUT_TB)) { test_phase = 2; goto cleanup; }
{ uint64_t p2elapsed; if (!wait_for("phase2", SYNC_TIMEOUT_TB, &p2elapsed)) { test_phase = 2; goto cleanup; } }
DEBUG_WARN(DEBUG_CATEGORY_DEBUG, "PHASE2_OK A=%u B=%u", db_sync_count(si_a), db_sync_count(si_b));
// Stress rounds
// Phase 3: 20 random batches
int global_seq = 3;
int rounds_ok = 0;
uint64_t insert_total_tb = 0;
int insert_count = 0;
uint32_t records_a = 0, records_b = 0;
for (int r = 0; r < ROUNDS && test_phase == 0; r++) {
uint32_t added = 0, cnt_a = 0, cnt_b = 0;
DEBUG_WARN(DEBUG_CATEGORY_DEBUG, "ROUND%d_INSERT A=%u B=%u expect=%u",
r, db_sync_count(si_a), db_sync_count(si_b), expected_total);
for (int b = 0; b < BATCHES_PER_ROUND && test_phase == 0; b++) {
uint64_t t_insert = get_time_tb();
for (int b = 0; b < TOTAL_BATCHES && test_phase == 0; b++) {
int n = rand() % (MAX_PER_BATCH + 1);
if (n == 0) continue;
int side = rand() & 1;
struct DB_SYNC_INSTANCE* si = side ? si_b : si_a;
uint64_t node = side ? NODE_ID_B : NODE_ID_A;
if (insert_batch(si, global_seq, n, node) < 0) { test_phase = 2; break; }
uint64_t bt0 = get_time_tb();
for (int i = 0; i < n; i++) {
char buf[384];
snprintf(buf, sizeof(buf),
"{\"seq\":%d,\"n\":%llu,\"ch\":\"test\",\"ct\":\"text\",\"d\":\"msg_%d\","
"\"pad\":\"xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx\"}",
global_seq + i, (unsigned long long)node, global_seq + i);
if (db_sync_insert(si, buf) < 0) {
DEBUG_ERROR(DEBUG_CATEGORY_DEBUG, "INSERT_FAIL seq=%d", global_seq + i);
test_phase = 2; break;
}
}
uint64_t bt = get_time_tb() - bt0;
insert_total_tb += bt;
insert_count += n;
if (side) records_b += n; else records_a += n;
global_seq += n;
added += n;
expected_total += n;
if (side) cnt_b += n; else cnt_a += n;
if ((added & 127) == 127) uasync_poll(ua, POLL_INTERVAL_MS);
if (b % 4 == 3) uasync_poll(ua, POLL_INTERVAL_MS);
}
if (test_phase != 0) break;
if (test_phase != 0) goto cleanup;
DEBUG_WARN(DEBUG_CATEGORY_DEBUG, "ROUND%d_ADDED added=%u A=%u B=%u total=%u",
r, added, cnt_a, cnt_b, expected_total);
uint64_t insert_total_ms = insert_total_tb / 10;
DEBUG_WARN(DEBUG_CATEGORY_DEBUG, "INSERT_DONE records=%d A=%u B=%u avg_=%lluus/rec",
insert_count, records_a, records_b,
insert_count > 0 ? (unsigned long long)(insert_total_tb * 100 / insert_count) : 0);
uint64_t ts = get_time_tb();
int synced = wait_for("round_sync", SYNC_TIMEOUT_TB);
uint64_t elapsed = (get_time_tb() - ts) / 10;
// Phase 4: sync wait
uint64_t sync_elapsed = 0;
if (!wait_for("sync_final", SYNC_TIMEOUT_TB, &sync_elapsed)) { test_phase = 2; goto cleanup; }
uint64_t total_ms = (get_time_tb() - t0) / 10;
if (test_phase == 0) {
uint32_t ca = db_sync_count(si_a), cb = db_sync_count(si_b);
if (!synced || ca != expected_total || cb != expected_total) {
DEBUG_ERROR(DEBUG_CATEGORY_DEBUG, "ROUND%d_FAIL A=%u B=%u expect=%u sync=%llums",
r, ca, cb, expected_total, (unsigned long long)elapsed);
fprintf(stderr, "FAIL round %d: A=%u B=%u expected=%u\n", r, ca, cb, expected_total);
test_phase = 2; break;
DEBUG_WARN(DEBUG_CATEGORY_DEBUG, "ALL_OK records=%d insert=%llums sync=%llums total=%llums A=%u B=%u",
insert_count, (unsigned long long)insert_total_ms,
(unsigned long long)sync_elapsed, (unsigned long long)total_ms, ca, cb);
printf("PROFILE: records=%d insert_avg=%lluus sync=%llums total=%llums A=%u B=%u\n",
insert_count,
insert_count > 0 ? (unsigned long long)(insert_total_tb * 100 / insert_count) : 0,
(unsigned long long)sync_elapsed, (unsigned long long)total_ms, ca, cb);
}
rounds_ok++;
DEBUG_WARN(DEBUG_CATEGORY_DEBUG, "ROUND%d_OK sync=%llums A=%u B=%u",
r, (unsigned long long)elapsed, ca, cb);
}
uint64_t total_ms = (get_time_tb() - t0) / 10;
if (test_phase == 0)
DEBUG_WARN(DEBUG_CATEGORY_DEBUG, "ALL_OK rounds=%d A=%u B=%u total=%llums",
rounds_ok, db_sync_count(si_a), db_sync_count(si_b), (unsigned long long)total_ms);
cleanup:
DEBUG_WARN(DEBUG_CATEGORY_DEBUG, "CLEANUP A=%u B=%u phase=%d",

Write Preview
Loading…
Cancel
Save