feat(config-repl): implement wait_all_with_timeout() for push fan-out
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 <stdatomic.h> 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
This commit is contained in:
parent
2ed7e34e25
commit
f3babec5bc
4 changed files with 224 additions and 16 deletions
|
|
@ -7,6 +7,7 @@
|
|||
#include <limits.h>
|
||||
#include <openssl/sha.h>
|
||||
#include <pthread.h>
|
||||
#include <stdatomic.h>
|
||||
#include <stdio.h>
|
||||
#include <stdlib.h>
|
||||
#include <string.h>
|
||||
|
|
@ -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);
|
||||
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue