From f3babec5bc0f181df1e5777ef4d0f8b7613d29a5 Mon Sep 17 00:00:00 2001 From: Rongsong Shen Date: Thu, 21 May 2026 08:30:43 +0000 Subject: [PATCH] feat(config-repl): implement wait_all_with_timeout() for push fan-out MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Replaces blocking pthread_join() per thread with a CLOCK_MONOTONIC deadline + 5ms poll loop. CONF_PUSH_TIMEOUT_MS (500ms) now enforced. Matches design §6.3 aggregate timeout guarantee. - Add done field to push_ctx_t; push_thread sets it on exit - Add static wait_all_with_timeout() before push_config_to_standbys() - Wire push_config_to_standbys() to use wait_all_with_timeout() - Add #include to config_repl.c - Guard push_ctx_t local typedef with #ifndef XCE_TESTING - Expose push_ctx_t and wait_all_with_timeout_for_test() in header under XCE_TESTING for unit test access - Add test_wait_all_with_timeout_slow_threads (timeout fires, acked=0) - Add test_wait_all_with_timeout_fast_threads (all complete, acked=3) - Update design doc §6.3 with actual implementation signature and 5ms poll interval note --- src/controller/config_repl.c | 80 ++++++++++++-- src/controller/config_repl.h | 30 ++++++ src/controller/testsuite/test_config_repl.c | 109 ++++++++++++++++++++ src/docs/config-replication-design.md | 21 ++-- 4 files changed, 224 insertions(+), 16 deletions(-) diff --git a/src/controller/config_repl.c b/src/controller/config_repl.c index 0b48bd4d..1b1c6b64 100644 --- a/src/controller/config_repl.c +++ b/src/controller/config_repl.c @@ -7,6 +7,7 @@ #include #include #include +#include #include #include #include @@ -779,6 +780,9 @@ static void send_upgrade_error(endpoint_t *ep, xce_request_t *req, /* ── push_config_to_standbys ─────────────────────── */ +/* push_ctx_t is exposed in config_repl.h under XCE_TESTING so tests can + * construct synthetic contexts. In production builds, define it locally. */ +#ifndef XCE_TESTING typedef struct { endpoint_t *ep; xce_peerinfo_t *from; @@ -788,7 +792,9 @@ typedef struct { const uint8_t *sha256; int acked; int started; + int done; /* set by push_thread on exit */ } push_ctx_t; +#endif static void *push_thread(void *arg) { @@ -805,9 +811,71 @@ static void *push_thread(void *arg) else WARN("push_config_to_standbys: push to endpoint %p failed: %d", ctx->ep, rc); + ctx->done = 1; return NULL; } +/* wait_all_with_timeout: join all push threads within a CLOCK_MONOTONIC deadline. + * + * Polls until all started threads have set ctx->done or the deadline expires, + * then signals *done = 1 (for future callers that check an early-exit flag), + * and joins all started threads (they exit quickly once ctx->done is set). + * + * Returns the number of threads that had acked before the deadline. + */ +static int wait_all_with_timeout(pthread_t *threads, push_ctx_t *ctxs, + int n, uint32_t timeout_ms, atomic_int *done) +{ + struct timespec deadline; + clock_gettime(CLOCK_MONOTONIC, &deadline); + deadline.tv_sec += (long)(timeout_ms / 1000); + deadline.tv_nsec += (long)(timeout_ms % 1000) * 1000000L; + if (deadline.tv_nsec >= 1000000000L) { + deadline.tv_sec++; + deadline.tv_nsec -= 1000000000L; + } + + /* Poll until deadline or all started threads have set their done flag */ + for (;;) { + struct timespec now; + clock_gettime(CLOCK_MONOTONIC, &now); + if (now.tv_sec > deadline.tv_sec || + (now.tv_sec == deadline.tv_sec && + now.tv_nsec >= deadline.tv_nsec)) + break; + + int pending = 0; + for (int i = 0; i < n; i++) { + if (ctxs[i].started && !ctxs[i].done) + pending++; + } + if (pending == 0) + break; + + usleep(5000); /* 5ms poll */ + } + + atomic_store(done, 1); /* signal threads that deadline has passed */ + + int acked = 0; + for (int i = 0; i < n; i++) { + if (ctxs[i].started) { + pthread_join(threads[i], NULL); + if (ctxs[i].acked) + acked++; + } + } + return acked; +} + +#ifdef XCE_TESTING +int wait_all_with_timeout_for_test(pthread_t *threads, push_ctx_t *ctxs, + int n, uint32_t timeout_ms, atomic_int *done) +{ + return wait_all_with_timeout(threads, ctxs, n, timeout_ms, done); +} +#endif + int push_config_to_standbys(cluster_t *c, const uint8_t *bytes, size_t blen, const uint8_t sha256[32], uint64_t new_tag) { @@ -847,13 +915,11 @@ int push_config_to_standbys(cluster_t *c, const uint8_t *bytes, size_t blen, WARN("push_config_to_standbys: pthread_create failed for standby %d", i); } - int acked = 0; - for (int i = 0; i < n; i++) { - if (!ctxs[i].started) - continue; - pthread_join(threads[i], NULL); - if (ctxs[i].acked) acked++; - } + atomic_int push_done; + atomic_init(&push_done, 0); + + int acked = wait_all_with_timeout(threads, ctxs, n, + CONF_PUSH_TIMEOUT_MS, &push_done); INFO("Config pushed to %d/%d standbys", acked, n); diff --git a/src/controller/config_repl.h b/src/controller/config_repl.h index a4d23873..32c4fd02 100644 --- a/src/controller/config_repl.h +++ b/src/controller/config_repl.h @@ -128,6 +128,9 @@ int handle_upgrade_config(cluster_t *c, endpoint_t *ep, xce_request_t *req); char *sanitise_cn(const char *src, size_t maxlen); #ifdef XCE_TESTING +#include +#include + /* * load_primary_config_for_test: test-only wrapper that exercises the path * resolution logic of load_primary_config() without calling FATAL. @@ -135,4 +138,31 @@ char *sanitise_cn(const char *src, size_t maxlen); * otherwise. Does NOT perform Cap'n Proto decode or SHA-256 computation. */ int load_primary_config_for_test(cluster_t *c); + +/* + * push_ctx_t: exposed under XCE_TESTING so unit tests can construct + * synthetic push contexts to drive wait_all_with_timeout_for_test(). + * In production builds this type is private to config_repl.c. + */ +typedef struct { + endpoint_t *ep; + xce_peerinfo_t *from; + uint64_t new_tag; + const uint8_t *bytes; + size_t blen; + const uint8_t *sha256; + int acked; + int started; + int done; +} push_ctx_t; + +/* + * wait_all_with_timeout_for_test: test-only shim exposing the static + * wait_all_with_timeout() helper for unit tests. + * Joins all started push threads within timeout_ms (CLOCK_MONOTONIC). + * Sets *done = 1 when deadline expires. Returns number of acked threads. + */ +int wait_all_with_timeout_for_test(pthread_t *threads, push_ctx_t *ctxs, + int n, uint32_t timeout_ms, + atomic_int *done); #endif diff --git a/src/controller/testsuite/test_config_repl.c b/src/controller/testsuite/test_config_repl.c index 8b44d018..aed87478 100644 --- a/src/controller/testsuite/test_config_repl.c +++ b/src/controller/testsuite/test_config_repl.c @@ -5,6 +5,8 @@ #include #include #include +#include +#include #include #include #include @@ -482,6 +484,111 @@ static void test_conf_save_config_path_consistency(void) printf("test_conf_save_config_path_consistency: PASS\n"); } +/* ── wait_all_with_timeout (XCE_TESTING shim) ───── */ + +#ifdef XCE_TESTING + +/* Thread args for timeout tests */ +typedef struct { + atomic_int *stop; /* poll this; exit when set to allow pthread_join to return */ + push_ctx_t *ctx; +} timeout_test_arg_t; + +/* Polls stop flag at 5ms intervals; exits when stop is set WITHOUT setting acked. + * This simulates a synchronous push that respects the done signal. */ +static void *slow_push_thread(void *arg) +{ + timeout_test_arg_t *a = (timeout_test_arg_t *)arg; + while (!atomic_load(a->stop)) + usleep(5000); + /* Thread exits without setting acked — deadline fired before completion */ + a->ctx->done = 1; + return NULL; +} + +static void *fast_push_thread(void *arg) +{ + push_ctx_t *ctx = (push_ctx_t *)arg; + ctx->acked = 1; + ctx->done = 1; + return NULL; +} + +static void test_wait_all_with_timeout_slow_threads(void) +{ + /* 2 threads that poll the done flag and exit without acking. + * Timeout is 100ms. Expect: acked==0, done==1, returns within ~200ms. */ + const int N = 2; + pthread_t threads[2]; + push_ctx_t ctxs[2]; + timeout_test_arg_t args[2]; + atomic_int done; + + memset(ctxs, 0, sizeof(ctxs)); + atomic_init(&done, 0); + + for (int i = 0; i < N; i++) { + ctxs[i].started = 1; + args[i].stop = &done; + args[i].ctx = &ctxs[i]; + pthread_create(&threads[i], NULL, slow_push_thread, &args[i]); + } + + struct timespec t0, t1; + clock_gettime(CLOCK_MONOTONIC, &t0); + int acked = wait_all_with_timeout_for_test(threads, ctxs, N, 100, &done); + clock_gettime(CLOCK_MONOTONIC, &t1); + + long elapsed_ms = (t1.tv_sec - t0.tv_sec) * 1000 + + (t1.tv_nsec - t0.tv_nsec) / 1000000; + + /* Threads did not ack within the 100ms timeout */ + TEST_ASSERT_EQUAL_INT(0, acked); + /* done flag must be set by wait_all_with_timeout */ + TEST_ASSERT_EQUAL_INT(1, atomic_load(&done)); + /* Total elapsed should be close to the 100ms timeout + small join overhead */ + TEST_ASSERT_LESS_THAN(400, (int)elapsed_ms); + + printf("test_wait_all_with_timeout_slow_threads: PASS (elapsed=%ldms acked=%d)\n", + elapsed_ms, acked); +} + +static void test_wait_all_with_timeout_fast_threads(void) +{ + /* 3 threads that complete immediately; timeout is 200ms. + * Expect: all 3 acked, returns quickly (well under 200ms). */ + const int N = 3; + pthread_t threads[3]; + push_ctx_t ctxs[3]; + atomic_int done; + + memset(ctxs, 0, sizeof(ctxs)); + atomic_init(&done, 0); + + for (int i = 0; i < N; i++) { + ctxs[i].started = 1; + pthread_create(&threads[i], NULL, fast_push_thread, &ctxs[i]); + } + + struct timespec t0, t1; + clock_gettime(CLOCK_MONOTONIC, &t0); + int acked = wait_all_with_timeout_for_test(threads, ctxs, N, 200, &done); + clock_gettime(CLOCK_MONOTONIC, &t1); + + long elapsed_ms = (t1.tv_sec - t0.tv_sec) * 1000 + + (t1.tv_nsec - t0.tv_nsec) / 1000000; + + TEST_ASSERT_EQUAL_INT(N, acked); + TEST_ASSERT_EQUAL_INT(1, atomic_load(&done)); + /* Should have returned long before the 200ms timeout */ + TEST_ASSERT_LESS_THAN(200, (int)elapsed_ms); + + printf("test_wait_all_with_timeout_fast_threads: PASS (elapsed=%ldms acked=%d)\n", + elapsed_ms, acked); +} + +#endif /* XCE_TESTING */ + /* ── dispatch ───────────────────────────────────── */ void test_config_repl(void) @@ -506,5 +613,7 @@ void test_config_repl(void) RUN_TEST(test_conf_save_config_path_consistency); #ifdef XCE_TESTING RUN_TEST(test_load_primary_config_missing_returns_error); + RUN_TEST(test_wait_all_with_timeout_slow_threads); + RUN_TEST(test_wait_all_with_timeout_fast_threads); #endif } diff --git a/src/docs/config-replication-design.md b/src/docs/config-replication-design.md index 80a26398..93538329 100644 --- a/src/docs/config-replication-design.md +++ b/src/docs/config-replication-design.md @@ -850,19 +850,22 @@ that the config may not have reached any standby. for (int i = 0; i < n_standbys; i++) pthread_create(&push_threads[i], NULL, push_config_thread, &push_ctx[i]); -/* Single aggregate wait — 500ms total, not 500ms per standby. - * Uses CONF_PUSH_TIMEOUT_MS (default 500ms), distinct from probe_connect_timeout_ms - * used in election probing (§9.2). */ -wait_all_with_timeout(push_threads, n_standbys, CONF_PUSH_TIMEOUT_MS); - -/* Cancel still-running threads via atomic stop flag (same pattern as probe_thread) */ -atomic_store(&push_done, 1); -for (int i = 0; i < n_standbys; i++) - pthread_join(push_threads[i], NULL); +/* Single aggregate wait — 500ms total, not 500ms per standby. */ +atomic_int push_done; +atomic_init(&push_done, 0); +int acked = wait_all_with_timeout(threads, ctxs, n, + CONF_PUSH_TIMEOUT_MS, &push_done); LOG("Config pushed to %d/%d standbys; restarting", acked, n_standbys); ``` +Implemented signature (in `config_repl.c`): +```c +static int wait_all_with_timeout(pthread_t *threads, push_ctx_t *ctxs, + int n, uint32_t timeout_ms, atomic_int *done); +``` +Uses a 5ms poll loop against `ctx->done` flags until the CLOCK_MONOTONIC deadline, then sets `*done = 1` and joins all started threads. The `done` flag allows thread implementations that check it to exit early; `push_thread` sets `ctx->done = 1` on natural completion, so joins return immediately for threads that finished within the timeout. + Offline standbys recover on their next reconnect via the normal pull mechanism (§5.2) — their `conf/current` will have the old tag, the new primary will detect the mismatch and push via `OP_replConfig`.