Rongsong Shen
7aac018c00
f88e4ef4 moved ha_demotion_convergence_fn's WAL truncate + self-restart
and ha_catchup_continuation_fn's WAL streaming off the delaymgr thread
(which drives heartbeats and election timeouts cluster-wide) onto the
slow lane, but left both flags OFF: flipping them ON hung pre-existing
unit fixtures that demote a primary or trigger catch-up without ever
starting c->slowlane -- the work was queued in a lane with no workers and
never ran.
- slowlane_is_started(): true while the lane has live workers.
- Both dispatchers run inline when the flag is OFF *or* the lane was never
started. Production always starts the lane, so it always offloads; the
lightweight fixtures keep their old inline behaviour, and neither
one-shot action can be silently dropped.
- g_drive_async_demotion_offload / g_drive_async_catchup_offload now
default to 1. The offload tests save and restore the previous value
instead of forcing 0.
New test_catchup_offload_without_started_lane_runs_inline (RED before: the
catch-up body never ran, the stream gate was never reached). Controller
suite 1239/0 with both offloads ON by default.
Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
101 lines
4.2 KiB
C
101 lines
4.2 KiB
C
#pragma once
|
|
/*
|
|
* slowlane.h — SlowLane executor: a bounded pool of dedicated worker threads
|
|
* that runs long blocking I/O OFF the Raft drive threads (iomgr epoll loop,
|
|
* delaymgr timer), with admission control and per-key in-flight dedup.
|
|
*
|
|
* Service interface is verbatim from
|
|
* .kiro/specs/raft-drive-async/design.md (Components -> SlowLane).
|
|
*
|
|
* The lane owns scheduling/isolation, NOT the work content: the submitted
|
|
* `fn` (peer connect, snapshot transfer, config-push body) stays in its own
|
|
* module and is responsible for clearing any caller-side in-flight state on
|
|
* completion. `slowlane_submit` is non-blocking: a drive-thread caller
|
|
* enqueues and returns immediately, never waiting for `fn` to run.
|
|
*
|
|
* raft-drive-async task 1.2 — Requirements 3.1, 3.2, 3.3, 3.4, 1.5.
|
|
*/
|
|
#include <pthread.h>
|
|
#include <stdint.h>
|
|
|
|
typedef void (*slowlane_fn_t)(void *arg);
|
|
|
|
/*
|
|
* One queued unit of work. Allocated by the lane on submit, freed by the
|
|
* worker after `fn` returns. Held on the FIFO pending queue until a worker
|
|
* picks it up.
|
|
*/
|
|
typedef struct slowlane_task {
|
|
uint64_t key; /* dedup key; 0 disables dedup */
|
|
slowlane_fn_t fn;
|
|
void *arg;
|
|
struct slowlane_task *next; /* intrusive FIFO link */
|
|
} slowlane_task_t;
|
|
|
|
/*
|
|
* SlowLane executor state. All fields are lane-private and guarded by `lock`;
|
|
* callers touch the lane only through the API below. Bounded by design:
|
|
* - `n_workers` dedicated pthreads, isolated from the vtask request pool;
|
|
* - at most `max_inflight` tasks RUNNING concurrently (admission cap);
|
|
* - at most one in-flight task per non-zero key (dedup).
|
|
*/
|
|
typedef struct slowlane {
|
|
pthread_mutex_t lock;
|
|
pthread_cond_t cond; /* signalled when work or a slot frees up */
|
|
|
|
slowlane_task_t *q_head; /* FIFO pending queue (admission-deferred) */
|
|
slowlane_task_t *q_tail;
|
|
|
|
uint64_t *inflight_keys; /* running OR queued non-zero keys (dedup set) */
|
|
int n_inflight_keys;
|
|
int cap_inflight_keys;
|
|
|
|
int n_workers;
|
|
int max_inflight; /* admission cap on concurrently RUNNING tasks */
|
|
int n_running; /* tasks currently executing on a worker */
|
|
|
|
pthread_t *workers;
|
|
int started; /* worker threads created */
|
|
int stopping; /* fini requested: drain + join */
|
|
} slowlane_t;
|
|
|
|
/*
|
|
* Initialise the lane and start `n_workers` dedicated worker threads.
|
|
* `max_inflight` bounds the number of tasks RUNNING concurrently (excess is
|
|
* queued, never dropped, never run on a drive thread).
|
|
* Returns 0 on success, -EINVAL on a bad argument, -errno on a thread/alloc
|
|
* failure (lane left un-started).
|
|
*/
|
|
int slowlane_init(slowlane_t *sl, int n_workers, int max_inflight);
|
|
|
|
/*
|
|
* Submit `fn(arg)` to run exactly once on a lane worker. NON-BLOCKING: the
|
|
* caller (which may be a drive thread) returns immediately.
|
|
*
|
|
* Returns (per design SlowLane Service Interface):
|
|
* 0 enqueued AND a worker slot was free (running < max_inflight) —
|
|
* the task can start immediately
|
|
* -EEXIST `key` (non-zero) already in flight — deduped no-op, NOT enqueued
|
|
* -EAGAIN admission cap full at submit time (running >= max_inflight) — the
|
|
* task IS still queued/deferred (never dropped, never run on a drive
|
|
* thread); the caller is told it was deferred, not that it failed
|
|
* -errno allocation failure / lane stopping
|
|
*
|
|
* In all of {0, -EAGAIN} the work is enqueued and will run exactly once. Only
|
|
* -EEXIST and -errno mean the task was not enqueued.
|
|
*
|
|
* `key == 0` disables dedup (every key-0 submit is independent work).
|
|
*/
|
|
int slowlane_submit(slowlane_t *sl, uint64_t key, slowlane_fn_t fn, void *arg);
|
|
|
|
/* Non-zero once slowlane_init() has started worker threads (and before
|
|
* slowlane_fini()). A submit to a lane that was never started is queued but
|
|
* never runs; callers that must not drop work fall back to running it inline
|
|
* (bug-1450/1451). */
|
|
int slowlane_is_started(slowlane_t *sl);
|
|
|
|
/*
|
|
* Drain the pending queue, stop and join all workers. Shutdown only.
|
|
* After return the lane is dormant; re-init to reuse.
|
|
*/
|
|
void slowlane_fini(slowlane_t *sl);
|