lei.chen/myxce
1
0
Fork 1
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

feat(ha-commit-durability): demand-driven group-commit flusher + durable_lsn watermark (task 2.1)

Browse source
This commit is contained in:
Rongsong Shen 2026-06-18 06:06:55 +00:00
parent 7e6f09f804
commit cc83caf5a0
4 changed files with 567 additions and 27 deletions
Download patch file Download diff file Expand all files Collapse all files

294
src/controller/persist.c
View file

@ -167,41 +167,196 @@ static int rotate_segment(persist_t* p)
}
/* ------------------------------------------------------------------ */
/* Dedicated fsync pthread (fires every wal_sync_ms milliseconds) */
/* Demand-driven group-commit flusher (ha-commit-durability task 2.1) */
/* */
/* Repurposes the old fixed-interval fsync timer into a flusher driven */
/* by the append-vs-synced gap. On each pass it batches one fdatasync */
/* over all entries appended since the last sync, advances durable_lsn */
/* to the synced tip, and wakes the commit evaluation. --wal-sync-ms */
/* now bounds the MAX batching latency (a no-further-append entry is */
/* still synced within the cadence), not a durability gap. */
/* */
/* LOST-WAKEUP / CORRECTNESS REASONING (gating item): */
/* */
/* The "work pending" condition is LEVEL-triggered on two persistent */
/* variables, NOT an edge: pending ⇔ durable_lsn < current_lsn - 1 */
/* (flusher_work_pending below, evaluated under p->lock). Because it */
/* is a comparison of stored state — not a transient flag consumed on */
/* read — a signal that races the flusher's wait can never be lost: */
/* */
/* - persist_wal_async advances current_lsn under p->lock, THEN */
/* signals fsync_cond. If the flusher is mid-flush when the append */
/* lands, current_lsn has already moved past the tip the flusher */
/* captured, so after that flush durable_lsn < current_lsn - 1 */
/* STILL holds; the flusher re-checks the level under p->lock on */
/* its next loop iteration and flushes again (it does NOT wait */
/* first). The missed signal is therefore harmless — the level */
/* itself carries the pending work forward. */
/* - If the append's signal arrives while the flusher sits in */
/* pthread_cond_wait, it wakes it. If it arrives in the tiny window */
/* after the flusher decided "no work" but before it re-enters the */
/* wait, the flusher re-evaluates the level under p->lock before */
/* waiting (we re-check inside the fsync_mutex-held region by */
/* reading current_lsn/durable_lsn under p->lock first), so it */
/* will not block while work is pending. */
/* */
/* "NEVER durable before its fdatasync": the flusher snapshots */
/* tip = current_lsn - 1 AND calls fdatasync WHILE HOLDING p->lock. */
/* persist_wal_async also pwrites under p->lock, so no append can */
/* extend the segment between the tip snapshot and the fdatasync — by */
/* the time fdatasync returns, every byte for LSNs <= tip is on stable */
/* storage. durable_lsn is only then advanced to tip (monotonic max). */
/* Releasing p->lock and broadcasting commit-eval happens after the */
/* advance, so a WAIT caller that observes durable_lsn >= its LSN is */
/* guaranteed the bytes are synced. */
/* */
/* Lock discipline: p->lock and fsync_mutex are NEVER held together */
/* (fsync_mutex guards only the cond wait; the flush section takes */
/* p->lock alone). This keeps fdatasync off any lock the iomgr / */
/* heartbeat path needs and preserves liveness (Req 1.6). */
/* ------------------------------------------------------------------ */
/* Test-only fault-injection probe (ha-commit-durability task 2.1 error path).
* Returns true when the test has forced fdatasync to be treated as failed.
* Compiled to a constant `false` in production (no -DXCE_TESTING), so the
* fail-loud gate below collapses to the pure rc != 0 check with zero overhead
* and no dependency on the test-only struct field. */
static inline bool flusher_fsync_forced_fail(persist_t* p)
{
#ifdef XCE_TESTING
return atomic_load_explicit(&p->test_force_fsync_fail,
memory_order_acquire) != 0;
#else
(void)p;
return false;
#endif
}
/* Returns true iff there is un-synced appended data (level predicate).
* MUST be called with p->lock held. current_lsn - 1 is the appended tip
* (current_lsn is the NEXT lsn to assign); LSN 0 reserved, so a fresh
* store has current_lsn == 1 → tip 0 → no pending work. */
static bool flusher_work_pending_locked(persist_t* p)
{
uint64_t tip = (p->current_lsn > 0) ? p->current_lsn - 1 : 0;
uint64_t dur = atomic_load_explicit(&p->durable_lsn, memory_order_acquire);
return dur < tip;
}
/* Perform one group-commit flush: snapshot the appended tip, fdatasync the
* live segment, and advance durable_lsn to the synced tip. Takes p->lock
* for the whole snapshot+sync+advance so no concurrent append can extend
* the segment between the tip read and the fdatasync. Returns the new
* durable_lsn (unchanged if there was nothing to sync). */
static uint64_t flusher_flush_once(persist_t* p)
{
pthread_mutex_lock(&p->lock);
uint64_t tip = (p->current_lsn > 0) ? p->current_lsn - 1 : 0;
uint64_t dur = atomic_load_explicit(&p->durable_lsn, memory_order_acquire);
if (dur >= tip || p->wal_fd < 0) {
/* Nothing new to sync (or no segment yet). Still flush the fd for a
* clean-shutdown / fallback pass, but do not advance the watermark.
* The watermark is already <= tip here, so a failed fallback sync
* cannot make it claim more than is durable — but we still surface
* the error loudly so a degrading disk is never silent (Req 1.1). */
if (p->wal_fd >= 0) {
int rc = fdatasync(p->wal_fd);
if (rc != 0 || flusher_fsync_forced_fail(p)) {
int e = (rc != 0) ? errno : EIO;
ERROR("persist: flusher fallback fdatasync FAILED (errno=%d %s)"
" — durable_lsn held at %" PRIu64 "; WAL not advanced",
e, strerror(e), dur);
}
}
pthread_mutex_unlock(&p->lock);
return dur;
}
/* fdatasync under p->lock: appends also hold p->lock to pwrite, so the
* segment cannot grow between this tip snapshot and the sync. Every byte
* for LSNs <= tip is therefore on stable storage when fdatasync returns
* SUCCESSFULLY. */
int rc = fdatasync(p->wal_fd);
/* FAIL-LOUD GATE (Req 1.1/1.2): advance durable_lsn ONLY on a successful
* sync. On EIO/ENOSPC the bytes are NOT on stable storage, so leaving the
* watermark unchanged keeps the non-durable LSNs from being counted toward
* the commit majority. We do NOT retry into an advanced watermark — the
* level predicate stays pending and the next pass retries the sync. */
if (rc != 0 || flusher_fsync_forced_fail(p)) {
int e = (rc != 0) ? errno : EIO;
ERROR("persist: flusher fdatasync FAILED (errno=%d %s) — durable_lsn "
"held at %" PRIu64 " (tip=%" PRIu64 " NOT made durable); commit "
"majority will not count these LSNs",
e, strerror(e), dur, tip);
pthread_mutex_unlock(&p->lock);
return dur;
}
/* Monotonic advance to the synced tip — never past it (the invariant
* durable_lsn <= current_lsn - 1 is preserved because tip == current_lsn-1
* was read under this same lock). One fdatasync advances durable_lsn past
* however many LSNs were appended since the last sync (batching). */
if (tip > dur)
atomic_store_explicit(&p->durable_lsn, tip, memory_order_release);
pthread_mutex_unlock(&p->lock);
return tip;
}
static void* fsync_worker(void* arg)
{
persist_t* p = arg;
/* Sleep the configured interval via a timed condvar wait rather than an
* uninterruptible nanosleep, so persist_stop wakes us immediately instead
* of blocking its pthread_join for up to a full wal_sync_ms (5s in tests
* that set a slow interval — was the controller-suite timeout cause). */
pthread_mutex_lock(&p->fsync_mutex);
while (!atomic_load_explicit(&p->fsync_stop, memory_order_acquire)) {
long ms = (long)p->config.wal_sync_ms;
struct timespec dl;
clock_gettime(CLOCK_REALTIME, &dl);
dl.tv_sec += ms / 1000;
dl.tv_nsec += (ms % 1000) * 1000000L;
if (dl.tv_nsec >= 1000000000L) {
dl.tv_sec += 1;
dl.tv_nsec -= 1000000000L;
}
pthread_cond_timedwait(&p->fsync_cond, &p->fsync_mutex, &dl);
/* Stop requested during the wait: exit before the next fdatasync. */
if (atomic_load_explicit(&p->fsync_stop, memory_order_acquire))
break;
/* fdatasync under p->lock — never held together with fsync_mutex. */
pthread_mutex_unlock(&p->fsync_mutex);
/* Re-evaluate the level predicate BEFORE waiting so a signal that
* raced in just before we entered the wait is not needed for
* correctness — if work is already pending we skip the wait entirely
* and flush. (Reads current_lsn/durable_lsn under p->lock.) */
pthread_mutex_lock(&p->lock);
if (p->wal_fd >= 0)
fdatasync(p->wal_fd);
bool pending = flusher_work_pending_locked(p);
pthread_mutex_unlock(&p->lock);
if (!pending) {
/* No pending work: wait for an append-signal OR the cadence
* deadline (bounded fallback flush so a lone no-further-append
* entry is still synced within wal_sync_ms — Req 3.1). */
long ms = (long)p->config.wal_sync_ms;
struct timespec dl;
clock_gettime(CLOCK_REALTIME, &dl);
dl.tv_sec += ms / 1000;
dl.tv_nsec += (ms % 1000) * 1000000L;
if (dl.tv_nsec >= 1000000000L) {
dl.tv_sec += 1;
dl.tv_nsec -= 1000000000L;
}
pthread_cond_timedwait(&p->fsync_cond, &p->fsync_mutex, &dl);
if (atomic_load_explicit(&p->fsync_stop, memory_order_acquire))
break;
}
/* Flush section: never hold fsync_mutex across the fdatasync (the
* flush takes p->lock alone). Release fsync_mutex, flush, re-acquire.
* After the flush we loop and re-check the level — if a new append
* landed during the flush, the level is still pending and we flush
* again without needing its signal (lost-wakeup-safe). */
pthread_mutex_unlock(&p->fsync_mutex);
flusher_flush_once(p);
/* COMMIT-EVAL WAKE (ordering fixed here; consumer wired by task 2.2):
* durable_lsn is advanced INSIDE flusher_flush_once (atomic store with
* release) BEFORE we return here. A WAIT caller's commit evaluation
* reads durable_lsn via an acquire load, so once it is signalled it
* observes the advanced watermark. Task 2.2 adds the explicit
* commit-eval condvar broadcast at THIS point (after the durable_lsn
* advance) — the "advance THEN wake" ordering this loop establishes is
* what makes that wake lost-wakeup-safe. Nothing consumes durable_lsn
* for commit yet, so no wake is needed in 2.1. */
pthread_mutex_lock(&p->fsync_mutex);
}
pthread_mutex_unlock(&p->fsync_mutex);
@ -255,6 +410,11 @@ int persist_init(persist_t* p, persist_config_t* cfg, cluster_t* c)
atomic_store(&p->wal_record_count, 0);
atomic_store(&p->checkpoint_in_progress, 0);
/* Durable watermark starts at 0 ("nothing fdatasync'd"): mirrors the
* LSN-0-reserved committed_lsn convention. It advances only on the
* flusher thread after a successful fdatasync (ha-commit-durability 2.1). */
atomic_store(&p->durable_lsn, 0);
/* Create WAL and snapshot directories */
if (mkdir_p(p->wal_path, 0700) != 0) return -errno;
if (mkdir_p(p->snap_path, 0700) != 0) return -errno;
@ -317,7 +477,30 @@ void persist_stop(persist_t* p)
pthread_mutex_lock(&p->lock);
if (p->wal_fd >= 0) {
fdatasync(p->wal_fd);
/* Final synchronous flush-or-exit on stop: fdatasync everything
* appended so far and advance durable_lsn to the synced tip so a
* post-stop read reflects the durable state (the flusher thread is
* already joined and will not advance it again).
*
* FAIL-LOUD GATE (Req 1.1/1.2): advance the watermark ONLY when the
* final sync SUCCEEDS — on EIO/ENOSPC the bytes are not on stable
* storage, so durable_lsn must stay where it is and the failure is
* surfaced loudly. */
int rc = fdatasync(p->wal_fd);
if (rc == 0 && !flusher_fsync_forced_fail(p)) {
uint64_t tip = (p->current_lsn > 0) ? p->current_lsn - 1 : 0;
uint64_t dur = atomic_load_explicit(&p->durable_lsn,
memory_order_acquire);
if (tip > dur)
atomic_store_explicit(&p->durable_lsn, tip,
memory_order_release);
} else {
int e = (rc != 0) ? errno : EIO;
ERROR("persist: final fdatasync on stop FAILED (errno=%d %s) — "
"durable_lsn left unchanged; appended-but-unsynced entries "
"are NOT durable",
e, strerror(e));
}
close(p->wal_fd);
p->wal_fd = -1;
}
@ -412,6 +595,17 @@ int persist_wal_async(persist_t* p, uint32_t type,
pthread_mutex_unlock(&p->lock);
/* Signal the group-commit flusher (ha-commit-durability 2.1): current_lsn
* advanced and the entry's bytes are pwritten, so durable_lsn < tip now
* holds (the level predicate). The signal is sent AFTER releasing p->lock
* and is level-triggered — if it races the flusher and is "lost", the
* flusher re-checks durable_lsn < current_lsn-1 under p->lock and flushes
* anyway (see fsync_worker). We do NOT fdatasync inline on this path; the
* flusher batches the sync across all pending appends. */
pthread_mutex_lock(&p->fsync_mutex);
pthread_cond_signal(&p->fsync_cond);
pthread_mutex_unlock(&p->fsync_mutex);
/* WAL record-count checkpoint trigger: schedule an async checkpoint once
* the record count reaches the configured threshold. The fetch_add is
* outside the lock — atomic_fetch_add is sufficient for the counter; the
@ -532,11 +726,43 @@ int persist_flush(persist_t* p)
{
pthread_mutex_lock(&p->lock);
int rc = (p->wal_fd >= 0) ? fdatasync(p->wal_fd) : 0;
/* persist_flush is a synchronous durability point (tests, clean shutdown):
* it fdatasync'd everything appended so far, so advance durable_lsn to the
* synced tip too. Without this, an explicit-flush-only path (slow flusher
* cadence) would leave durable_lsn lagging even though the bytes are on
* disk. Monotonic max; preserves durable_lsn <= current_lsn - 1 (tip read
* under the same p->lock). Gated on rc == 0 (fail-loud, Req 1.1/1.2): a
* failed flush must not advance the watermark. */
if (rc == 0 && !flusher_fsync_forced_fail(p)) {
uint64_t tip = (p->current_lsn > 0) ? p->current_lsn - 1 : 0;
uint64_t dur = atomic_load_explicit(&p->durable_lsn,
memory_order_acquire);
if (tip > dur)
atomic_store_explicit(&p->durable_lsn, tip, memory_order_release);
} else if (rc != 0 || flusher_fsync_forced_fail(p)) {
int e = (rc != 0) ? errno : EIO;
ERROR("persist: persist_flush fdatasync FAILED (errno=%d %s) — "
"durable_lsn not advanced; flushed bytes are NOT durable",
e, strerror(e));
}
pthread_mutex_unlock(&p->lock);
return rc;
}
uint64_t persist_durable_lsn(persist_t* p)
{
return atomic_load_explicit(&p->durable_lsn, memory_order_acquire);
}
#ifdef XCE_TESTING
void persist_test_set_force_fsync_fail(persist_t* p, int fail)
{
if (p)
atomic_store_explicit(&p->test_force_fsync_fail, fail ? 1 : 0,
memory_order_release);
}
/* ------------------------------------------------------------------ *
* Durability-verification test seam (ha-commit-durability Req 3.2/3.3)*
* *
@ -1823,6 +2049,20 @@ bool worker_should_retain(const xce_worker_t* w, const persist_config_t* cfg)
* Idempotent: calling twice with same cutoff_lsn returns 0 both times.
* ============================================================ */
/* Clamp durable_lsn down to `tip` when the log tip recedes below it (truncate
* / divergent-suffix discard). Without this, durable_lsn could exceed
* current_lsn - 1 after a truncate, violating its invariant and letting the
* commit majority count self at an LSN whose bytes were just removed. MUST be
* called with p->lock held (mirrors the committed_lsn clamp in the truncate
* paths). durable_lsn only ever decreases here — the sole exception to its
* monotonic-non-decreasing rule (ha-commit-durability 2.1 State Management). */
static void clamp_durable_lsn_locked(persist_t* p, uint64_t tip)
{
uint64_t dur = atomic_load_explicit(&p->durable_lsn, memory_order_acquire);
if (dur > tip)
atomic_store_explicit(&p->durable_lsn, tip, memory_order_release);
}
int persist_truncate_wal_above(persist_t* p, uint64_t cutoff_lsn)
{
pthread_mutex_lock(&p->lock);
@ -1902,6 +2142,7 @@ int persist_truncate_wal_above(persist_t* p, uint64_t cutoff_lsn)
p->last_crc = WAL_CRC_NONE;
atomic_store_explicit(&p->committed_lsn, cutoff_lsn,
memory_order_release);
clamp_durable_lsn_locked(p, cutoff_lsn);
pthread_mutex_unlock(&p->lock);
return 0;
}
@ -1980,6 +2221,7 @@ int persist_truncate_wal_above(persist_t* p, uint64_t cutoff_lsn)
p->current_lsn = cutoff_lsn + 1;
p->last_crc = new_last_crc;
atomic_store_explicit(&p->committed_lsn, cutoff_lsn, memory_order_release);
clamp_durable_lsn_locked(p, cutoff_lsn);
pthread_mutex_unlock(&p->lock);
return 0;