From d326ee1b2e01f5c707d9d3d706d2fc5d68e0a71b Mon Sep 17 00:00:00 2001 From: Rongsong Shen Date: Thu, 7 May 2026 17:23:50 +0000 Subject: [PATCH] feat: P2-3 repl_wait_ack() with pthread_cond_timedwait + ACK deadlock prevention MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Implements sync replication wait: primary blocks until at least one non-catching-up standby ACKs the target LSN, or times out. Uses pthread_cond_timedwait with CLOCK_REALTIME deadline. Inner loop rechecks all standbys after each wakeup to guard against spurious wakeups. ACK deadlock prevention: repl_on_ack() is called exclusively from the io_worker read callback (not vtaskmgr), so it only does atomic_store + pthread_cond_signal — safe to call while vcpu threads block in repl_wait_ack(). 4 new TDD tests (all green, 9 total in test_repl_unit): - test_wait_ack_no_standbys_returns_enodev - test_wait_ack_already_acked - test_wait_ack_timeout (verifies >=80ms elapsed) - test_wait_ack_woken_by_background_ack (pthread thread signals after 30ms) --- src/controller/repl.c | 59 +++++++++++++-- src/controller/testsuite/test_repl_unit.c | 90 +++++++++++++++++++++++ 2 files changed, 144 insertions(+), 5 deletions(-) diff --git a/src/controller/repl.c b/src/controller/repl.c index 8d4dd22..7b6b281 100644 --- a/src/controller/repl.c +++ b/src/controller/repl.c @@ -201,12 +201,61 @@ int repl_broadcast(replicator_t *r, const wal_entry_hdr_t *hdr, int repl_wait_ack(replicator_t *r, uint64_t target_lsn, uint32_t timeout_ms) { - (void)r; - (void)target_lsn; - (void)timeout_ms; + if (!r) return -EINVAL; - /* Full implementation in P2-3 */ - return -ENODEV; + /* Check whether any non-catching-up standby exists */ + mylock_lock(&r->lock); + int has_standby = 0; + for (int i = 0; i < MAX_STANDBYS; i++) { + if (r->standbys[i].ep != NULL && + !atomic_load_explicit(&r->standbys[i].in_catch_up, + memory_order_acquire)) { + has_standby = 1; + break; + } + } + mylock_unlock(&r->lock); + + if (!has_standby) return -ENODEV; + + /* Compute absolute deadline for pthread_cond_timedwait (CLOCK_REALTIME) */ + struct timespec deadline; + clock_gettime(CLOCK_REALTIME, &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; + } + + pthread_mutex_lock(&r->ack_lock); + int rc = -ETIMEDOUT; + while (1) { + /* Recheck every standby after each wakeup (spurious-wakeup guard) */ + mylock_lock(&r->lock); + for (int i = 0; i < MAX_STANDBYS; i++) { + standby_conn_t *s = &r->standbys[i]; + if (s->ep == NULL) continue; + if (atomic_load_explicit(&s->in_catch_up, memory_order_acquire)) + continue; + if (atomic_load_explicit(&s->acked_lsn, memory_order_acquire) + >= target_lsn) { + rc = 0; + break; + } + } + mylock_unlock(&r->lock); + if (rc == 0) break; + + int wait_rc = pthread_cond_timedwait(&r->ack_cond, &r->ack_lock, + &deadline); + if (wait_rc == ETIMEDOUT) { + rc = -ETIMEDOUT; + break; + } + } + pthread_mutex_unlock(&r->ack_lock); + return rc; } /* ============================================================ diff --git a/src/controller/testsuite/test_repl_unit.c b/src/controller/testsuite/test_repl_unit.c index a16258c..98de040 100644 --- a/src/controller/testsuite/test_repl_unit.c +++ b/src/controller/testsuite/test_repl_unit.c @@ -1,7 +1,10 @@ #include +#include #include #include #include +#include +#include /* Use the same Unity include pattern as other controller tests */ #include #include "repl.h" @@ -74,6 +77,89 @@ static void test_repl_trim_lsn_disconnected_slot_expiry(void) { printf("test_repl_trim_lsn_disconnected_slot_expiry: PASS\n"); } +/* ============================================================ + * P2-3: repl_wait_ack() tests + * ============================================================ */ + +/* Test 1: repl_wait_ack returns -ENODEV when no standbys connected */ +static void test_wait_ack_no_standbys_returns_enodev(void) { + replicator_t r; + repl_init(&r); + int rc = repl_wait_ack(&r, /*target_lsn=*/10, /*timeout_ms=*/50); + TEST_ASSERT_EQUAL_INT(-ENODEV, rc); + repl_fini(&r); + printf("test_wait_ack_no_standbys_returns_enodev: PASS\n"); +} + +/* Test 2: repl_wait_ack returns 0 when standby already ACKed target LSN */ +static void test_wait_ack_already_acked(void) { + replicator_t r; + int dummy_ep = 0; + repl_init(&r); + repl_add_standby(&r, (void*)&dummy_ep, "sb", 7001); + /* Pre-set acked_lsn to 20 (already past target 10) */ + repl_on_ack(&r, (void*)&dummy_ep, 20); + int rc = repl_wait_ack(&r, 10, 50); + TEST_ASSERT_EQUAL_INT(0, rc); + repl_fini(&r); + printf("test_wait_ack_already_acked: PASS\n"); +} + +/* Test 3: repl_wait_ack times out when standby never ACKs */ +static void test_wait_ack_timeout(void) { + replicator_t r; + int dummy_ep = 0; + repl_init(&r); + repl_add_standby(&r, (void*)&dummy_ep, "sb", 7001); + /* acked_lsn stays at 0 — never reaches target 100 */ + struct timespec start, end; + clock_gettime(CLOCK_MONOTONIC, &start); + int rc = repl_wait_ack(&r, 100, 100 /* 100ms timeout */); + clock_gettime(CLOCK_MONOTONIC, &end); + TEST_ASSERT_EQUAL_INT(-ETIMEDOUT, rc); + /* Verify we actually waited ~100ms (not returned immediately) */ + long elapsed_ms = (end.tv_sec - start.tv_sec) * 1000 + + (end.tv_nsec - start.tv_nsec) / 1000000; + TEST_ASSERT_TRUE(elapsed_ms >= 80); /* at least 80ms */ + repl_fini(&r); + printf("test_wait_ack_timeout: PASS\n"); +} + +/* Test 4: ACK from background thread wakes repl_wait_ack */ +struct ack_args { + replicator_t *r; + void *ep; + uint64_t lsn; + int delay_ms; +}; + +static void* ack_sender_thread(void* arg) { + struct ack_args* a = arg; + struct timespec ts = { 0, (long)a->delay_ms * 1000000L }; + nanosleep(&ts, NULL); + repl_on_ack(a->r, a->ep, a->lsn); + return NULL; +} + +static void test_wait_ack_woken_by_background_ack(void) { + replicator_t r; + int dummy_ep = 0; + repl_init(&r); + repl_add_standby(&r, (void*)&dummy_ep, "sb", 7001); + + struct ack_args args = { &r, (void*)&dummy_ep, 15, 30 /* 30ms delay */ }; + pthread_t th; + pthread_create(&th, NULL, ack_sender_thread, &args); + + /* Wait for LSN 15 with 200ms timeout — ACK arrives after 30ms */ + int rc = repl_wait_ack(&r, 15, 200); + pthread_join(th, NULL); + + TEST_ASSERT_EQUAL_INT(0, rc); + repl_fini(&r); + printf("test_wait_ack_woken_by_background_ack: PASS\n"); +} + void test_repl_unit(void) { UNITY_BEGIN(); RUN_TEST(test_repl_init_zeroes_nstandbys); @@ -81,5 +167,9 @@ void test_repl_unit(void) { RUN_TEST(test_repl_add_standby_max); RUN_TEST(test_repl_trim_lsn_no_standbys); RUN_TEST(test_repl_trim_lsn_disconnected_slot_expiry); + RUN_TEST(test_wait_ack_no_standbys_returns_enodev); + RUN_TEST(test_wait_ack_already_acked); + RUN_TEST(test_wait_ack_timeout); + RUN_TEST(test_wait_ack_woken_by_background_ack); UNITY_END(); }