/* * test_scheduler_backpressure.c — worst-case decision/allocation * backpressure test (component-failover-adoption, task 3.3, boundary: * SchedulerControllerClient, Req 3 AC2). * * Scope note: this file proves Req 3 AC2 only -- the bounded, occupancy- * dependent buffering upstream of the publish call (picker/allocator * continue briefly, then genuinely block, then resume with nothing lost). * It does NOT exercise Req 7's retry-until-success-or-shutdown wrapper * around a `-E_XCE_NOT_READY` return from `xcecli_jobdispatch()`/ * `xcecli_set_pend_reason()` (that wrapper, and its own test coverage, is * task 3.2's `sched_publish_dispatch()`/`sched_publish_pend_reason()` work, * see test_scheduler_decision_publish_retry.c). The wrapped * `xcecli_set_pend_reason()` double below BLOCKS on a condvar rather than * ever returning `-E_XCE_NOT_READY`, so it never enters that retry loop at * all -- it stands in for the call still being in-flight/stuck mid-attempt, * one layer below where Req 7 applies. * * Links the REAL production scheduler_srcs (same technique * test_scheduler_connect.c / test_scheduler_decision_publish_retry.c * already established for this boundary -- see meson.build) and calls the * REAL, public sched_init(), which -- unlike task 3.2's rejoin test, which * deliberately made its own setup join fail -- is made to SUCCEED here * (xcecli_start()/xcecli_join() wrapped to return success with no I/O), * so the real sched_job_picker() and sched_allocator() threads (scheduler.c) * actually start. xcecli_set_pend_reason() is wrapped to BLOCK on a * condition variable under test control, standing in for an ongoing * controller outage (an XCECLI_FAILOVER_CALL retry budget stuck retrying) * without any real network I/O or busy-spinning. * * ── Why the picker thread is never driven, and jobs are hand-built ────── * * sched_job_picker()/sched_allocator() are `static` (scheduler.c) and can * only be started via the real sched_init() -- both threads start together, * there is no way to run just one. sched_job_picker() is kept PERMANENTLY * DORMANT here by never calling sched_go(): it blocks forever on * cbuf_get(&core.wbuf) (scheduler.c), so it never touches core.jbuf itself * and never contends with this file's own direct cbuf_put()/cbuf_tryput() * calls on core.jbuf, which stand in for the picker's role of feeding one * jobid at a time -- exactly the "without also running the full * sched_job_picker thread" narrowing this task's own text pre-authorizes. * * sched_allocator() IS the real, unmodified production function under * test here (not a stand-in) -- proving its genuine cbuf_put()-blocks * behavior against a full `decisions` cbuf is this test's actual point. * Driving it through sched_make_allocation() with a fully realistic * job/queue/worker/resource model (parse_resreq(), the seqjob_allocator * plugin's "slots" resource lookup, sched_collect_workers() against a real * resmgr) would need substantial scaffolding unrelated to what this test * verifies. Instead, each test job's xce_sched_obj_t is built directly * with tag=XCE_RESREQ_COMPOUND and a COMPOUND resreq with an EMPTY * `primaries` list: sched_make_compound_allocation()'s for-loop * (scheduler.c) then has zero iterations, so it returns rc=0 ("pend") * without ever touching a worker, a resource, or the seqjob_allocator * plugin -- alloc_rc==0 is exactly the same "job is pend" branch * sched_allocator() takes for a real no-matching-worker pend outcome, so * this fixture exercises the identical decision-construction and * cbuf_put(&core.decisions, ...) code path a real pend decision would. * The pend reason itself is pre-set directly on the job's top-level * `schedobj->reason` field before sched_allocator() ever sees it; * sched_make_compound_allocation() only overwrites that field when its own * ctx.reason is non-NULL, which never happens with zero primaries, so the * pre-set reason survives untouched into make_pend_decision(). * * ── Worst-case derivation (do not copy a number from design.md prose * without re-deriving it against live cbuf state -- this task's own * instruction) ────────────────────────────────────────────────────── * * Both cbuf capacities are read LIVE from the real, freshly-initialized * core (sched.core.decisions.ncap / sched.core.jbuf.ncap -- xce_sched_core_t * struct fields, mylock.h's cbuf_t) rather than hardcoded, and asserted * against the documented values (ndecisions=5, jbuf capacity=1, core.c) as * a sanity check only. * * The worst case (least slack) is: `decisions` is ALREADY FULL at the * instant the stuck decision is popped off it (design.md's "decisions cbuf * already at or near its capacity-5 occupancy at the moment the stall * begins"). This test constructs exactly that precondition (prefills * `decisions` to full capacity BEFORE starting the stall), then proves, * against the REAL sched_allocator() thread and REAL cbuf blocking * semantics (mylock.c, not mocked): * * 1. `decisions` has room for exactly ONE more decision once the stuck * one is popped (occupancy ncap-1) -- job A's PEND decision lands. * 2. The NEXT decision (job B's) is fully built (not dropped, not * errored -- Req 3 AC2's "no job picked or decided ... is ever lost") * but then genuinely BLOCKS trying to enqueue, because `decisions` is * full again. * 3. `jbuf` (capacity 1) absorbs exactly ONE more job (C) beyond that -- * the allocator had already drained job B out of jbuf before getting * stuck on `decisions` -- and then a further job (D) is refused for a * sustained bounded window, proving the picker itself would now * block too (Req 3 AC2's "and then sched_job_picker genuinely * block"), not silently drop work. * 4. Once the stall clears, every decision produced before AND during * the outage (the prefilled backlog plus jobs A/B/C) is eventually * drained and published -- none lost (Req 3 AC2's "no job picked or * decided during that bounded window is ever lost"). * * This matches requirements.md's "as few as 1-2" floor (job A enqueued + * job B built-but-blocked = 2 additional decisions) at the `decisions` * layer, plus the analogous one-slot-of-slack-then-genuine-block pattern * one layer down at `jbuf` -- derived here from the live cbuf mechanics, * not copied from prose. * * All "prove it's genuinely blocked" checks use cbuf_tryput() (non-blocking) * in a short bounded-retry loop, never a real blocking call -- this test * can never hang even if the backpressure guarantee were broken. */ #include #include #include #include #include #include #include #include "client.h" #include "controller.capnp.h" #include "core.h" #include "decision.h" #include "mylock.h" #include "preason.h" #include "resreq.h" #include "sched/types.h" #include "scheduler.h" #include "xce/cluster.h" #include "xlist.h" extern int scheduler_req_handler(xce_request_t*, void*); #define N_DUMMY_PREFILL 5 #define N_REAL_JOBS_FED 3 /* A, B, C */ #define N_TOTAL_DECISIONS (N_DUMMY_PREFILL + N_REAL_JOBS_FED) #define POLL_STEP_NS (5 * 1000 * 1000) /* 5ms */ #define POLL_TIMEOUT_MS 2000 /* ════════════════════════════════════════════════════════════════════════ * Test double: __wrap_xcecli_set_pend_reason -- blocks on a condvar until * the test releases it, standing in for a stuck outage retry (see file * header comment). __wrap_xcecli_start / __wrap_xcecli_join are trivial * always-succeed doubles (same rationale as * test_scheduler_decision_publish_retry.c's own __wrap_xcecli_start) so * this file's one real sched_init() call actually starts the picker/ * allocator threads instead of returning early. * ════════════════════════════════════════════════════════════════════════ */ static pthread_mutex_t g_pend_mtx = PTHREAD_MUTEX_INITIALIZER; static pthread_cond_t g_pend_release_cond = PTHREAD_COND_INITIALIZER; static pthread_cond_t g_pend_entered_cond = PTHREAD_COND_INITIALIZER; static int g_pend_release = 0; static int g_pend_entered = 0; static void reset_pend_wrap(void) { g_pend_release = 0; g_pend_entered = 0; } int __wrap_xcecli_start(xcecli_ctx_t c) { (void)c; return 0; } int __wrap_xcecli_join(xcecli_ctx_t c, int role, hostattrs_t* attrs) { (void)c; (void)role; (void)attrs; return 0; } int __wrap_xcecli_set_pend_reason(xcecli_ctx_t c, xce_preason_set_t* reasons) { (void)c; (void)reasons; pthread_mutex_lock(&g_pend_mtx); g_pend_entered = 1; pthread_cond_broadcast(&g_pend_entered_cond); while (!g_pend_release) { pthread_cond_wait(&g_pend_release_cond, &g_pend_mtx); } pthread_mutex_unlock(&g_pend_mtx); return 0; } static void wait_for_pend_wrap_entered(void) { pthread_mutex_lock(&g_pend_mtx); while (!g_pend_entered) { pthread_cond_wait(&g_pend_entered_cond, &g_pend_mtx); } pthread_mutex_unlock(&g_pend_mtx); } static void release_pend_wrap(void) { pthread_mutex_lock(&g_pend_mtx); g_pend_release = 1; pthread_cond_broadcast(&g_pend_release_cond); pthread_mutex_unlock(&g_pend_mtx); } void setUp(void) { reset_pend_wrap(); } void tearDown(void) { return; } /* ════════════════════════════════════════════════════════════════════════ * Fixtures * ════════════════════════════════════════════════════════════════════════ */ /* build_pend_only_job - see file header comment for why COMPOUND + zero * primaries is used: it drives sched_allocator()'s real "job is pend" * decision-construction path without needing any worker/resource/queue * scaffolding. */ static xce_sched_job_t* build_pend_only_job(uint64_t jobid) { xce_sched_job_t* job = (xce_sched_job_t*)calloc(1, sizeof(*job)); xce_sched_obj_t* schedobj = (xce_sched_obj_t*)calloc(1, sizeof(*schedobj)); xce_compound_schedobj_t* compound = (xce_compound_schedobj_t*)calloc(1, sizeof(*compound)); xce_compound_resreq_t* resreq = (xce_compound_resreq_t*)calloc(1, sizeof(*resreq)); xlist_init(&(resreq->primaries), 1); compound->owner = schedobj; compound->resreq = resreq; compound->reason = NULL; schedobj->tag = XCE_RESREQ_COMPOUND; schedobj->u.compound = compound; /* Top-level field, distinct from compound->reason above -- see file * header comment for why this survives sched_make_compound_allocation() * untouched. */ schedobj->reason = make_pendreason(jobid); job->jobid = jobid; job->state = JOB_STAT_PEND; job->schedobj = schedobj; return job; } static xce_decision_t* build_dummy_pend_decision(uint64_t jobid) { return make_pend_decision(make_pendreason(jobid)); } /* ════════════════════════════════════════════════════════════════════════ * Drain thread -- stands in for driver.c's decision-drain consumer * (xce_sched_driver_run_normal_mode_round()'s cbuf_get(&core.decisions) * loop). Pops exactly N_TOTAL_DECISIONS decisions, publishing each via the * REAL sched_publish_pend_reason() (scheduler.c). Its first iteration is * the one that blocks in the wrapped call (see file header comment). * ════════════════════════════════════════════════════════════════════════ */ typedef struct { xce_scheduler_t* sched; uint64_t seen_jobids[N_TOTAL_DECISIONS]; int seen_rcs[N_TOTAL_DECISIONS]; int n_seen; } drain_ctx_t; static void* drain_thread_fn(void* arg) { drain_ctx_t* dctx = (drain_ctx_t*)arg; int i; for (i = 0; i < N_TOTAL_DECISIONS; i++) { xce_decision_t* decision = NULL; xce_pendreason_t* reasons[1]; xce_preason_set_t reason_set; int rc; cbuf_get(&(dctx->sched->core.decisions), &decision); reasons[0] = decision->u.reason; reason_set.nreasons = 1; reason_set.reasons = reasons; rc = sched_publish_pend_reason(dctx->sched, &reason_set); dctx->seen_jobids[dctx->n_seen] = decision->u.reason->jobid; dctx->seen_rcs[dctx->n_seen] = rc; dctx->n_seen++; } return NULL; } /* ════════════════════════════════════════════════════════════════════════ * The test * ════════════════════════════════════════════════════════════════════════ */ static void test_worst_case_backpressure_blocks_then_recovers_no_loss(void) { xce_client_config_t config; xce_cluster_config_t cluster_cfg; xce_scheduler_t sched; pthread_t drain_tid; drain_ctx_t dctx; int rc; int i; int ncap_decisions; int ncap_jbuf; memset(&config, 0, sizeof(config)); config.nmasters = 1; config.masters[0] = "127.0.0.1"; config.master_ports[0] = 19997; config.from = "sched_backpressure_test"; config.key = "keepsecret$"; memset(&cluster_cfg, 0, sizeof(cluster_cfg)); cluster_cfg.candidates_count = 1; strncpy(cluster_cfg.candidates[0].hostname, "127.0.0.1", XCE_MAXHOSTNAME_LEN - 1); cluster_cfg.candidates[0].port = 19997; strncpy(cluster_cfg.identity, "sched_backpressure_test", sizeof(cluster_cfg.identity) - 1); memset(&sched, 0, sizeof(sched)); rc = sched_init(&sched, &config, &cluster_cfg, scheduler_req_handler); TEST_ASSERT_TRUE_MESSAGE( rc >= 0, "fixture: sched_init() must succeed (xcecli_start()/" "xcecli_join() wrapped to succeed with no I/O) so the " "real picker/allocator threads actually start"); ncap_decisions = sched.core.decisions.ncap; ncap_jbuf = sched.core.jbuf.ncap; TEST_ASSERT_TRUE_MESSAGE(ncap_decisions >= 1, "sanity: decisions cbuf must have positive " "capacity"); TEST_ASSERT_TRUE_MESSAGE(ncap_jbuf >= 1, "sanity: jbuf cbuf must have positive " "capacity"); TEST_ASSERT_EQUAL_INT_MESSAGE( N_DUMMY_PREFILL, ncap_decisions, "documented invariant (core.c: ndecisions=5) -- if this fails, " "N_DUMMY_PREFILL/N_TOTAL_DECISIONS above must be re-derived " "against the new live capacity, not just this assertion changed"); TEST_ASSERT_EQUAL_INT_MESSAGE( 1, ncap_jbuf, "documented invariant (core.c: jbuf capacity fixed at 1, a " "scheduler-core determinism invariant this task does not " "adjust)"); /* --- Step 1: prefill `decisions` to FULL capacity -- the worst-case * (least-slack) precondition. --- */ for (i = 0; i < ncap_decisions; i++) { xce_decision_t* d = build_dummy_pend_decision(9000 + (uint64_t)i); TEST_ASSERT_NOT_NULL(d); rc = cbuf_tryput(&(sched.core.decisions), &d); TEST_ASSERT_TRUE_MESSAGE( rc >= 0, "fixture: prefilling decisions to capacity must not " "itself block"); } /* --- Step 2: start the drain thread. Its first cbuf_get() pops the * first prefilled decision and calls the REAL * sched_publish_pend_reason(), which blocks inside the wrapped * xcecli_set_pend_reason() -- simulating an ongoing outage. --- */ dctx.sched = &sched; dctx.n_seen = 0; rc = pthread_create(&drain_tid, NULL, drain_thread_fn, &dctx); TEST_ASSERT_EQUAL_INT(0, rc); wait_for_pend_wrap_entered(); TEST_ASSERT_EQUAL_INT_MESSAGE( ncap_decisions - 1, sched.core.decisions.len, "the drain thread's single cbuf_get() must have removed exactly " "one decision, leaving decisions at capacity-1"); /* --- Step 3: feed job A through the REAL sched_allocator() thread * (via jbuf, standing in for the picker). With one free slot in * `decisions`, job A's PEND decision must land successfully -- this is * the "as few as 1" floor. --- */ { xce_sched_job_t* jobA = build_pend_only_job(9101); uint64_t jobidA = jobA->jobid; TEST_ASSERT_TRUE(sched_add_job(&(sched.core), jobA) >= 0); rc = cbuf_tryput(&(sched.core.jbuf), &jobidA); TEST_ASSERT_TRUE_MESSAGE( rc >= 0, "jbuf must have room for job A -- it starts empty " "since the dormant picker thread never touches it " "(sched_go() is never called in this test)"); } { int waited_ms = 0; while ((sched.core.decisions.len < ncap_decisions) && (waited_ms < POLL_TIMEOUT_MS)) { struct timespec ts = { 0, POLL_STEP_NS }; nanosleep(&ts, NULL); waited_ms += 5; } TEST_ASSERT_EQUAL_INT_MESSAGE( ncap_decisions, sched.core.decisions.len, "job A's decision must be the ONE additional decision the " "worst-case (already-full) occupancy has room for"); } /* --- Step 4: feed job B. `decisions` is full again, so * sched_allocator() must build job B's decision (Req 3 AC2: deciding is * not lost) but then genuinely BLOCK trying to enqueue it -- the * second half of the "1-2" floor. --- */ { xce_sched_job_t* jobB = build_pend_only_job(9102); uint64_t jobidB = jobB->jobid; TEST_ASSERT_TRUE(sched_add_job(&(sched.core), jobB) >= 0); rc = cbuf_tryput(&(sched.core.jbuf), &jobidB); TEST_ASSERT_TRUE_MESSAGE( rc >= 0, "jbuf must have room for job B -- the allocator " "already drained job A's jobid out of it"); } { /* Bounded window for the allocator to reach its blocking * cbuf_put() -- decisions occupancy must not exceed capacity once * it does. */ struct timespec ts = { 0, 100 * 1000 * 1000 }; nanosleep(&ts, NULL); } TEST_ASSERT_EQUAL_INT_MESSAGE( ncap_decisions, sched.core.decisions.len, "job B's decision must be stuck trying to enqueue -- decisions " "occupancy must never exceed its own capacity"); /* --- Step 5: prove the genuine block reaches jbuf too (Req 3 AC2: * "...and then sched_job_picker genuinely block"). jbuf absorbs * exactly one more job (C) -- the allocator already drained jbuf * before getting stuck on `decisions` -- and a further push (D) must * then fail for a sustained bounded window. --- */ { xce_sched_job_t* jobC = build_pend_only_job(9103); uint64_t jobidC = jobC->jobid; TEST_ASSERT_TRUE(sched_add_job(&(sched.core), jobC) >= 0); rc = cbuf_tryput(&(sched.core.jbuf), &jobidC); TEST_ASSERT_TRUE_MESSAGE( rc >= 0, "jbuf's one slot of slack must still accept job C"); } { uint64_t jobidD = 9104; int any_succeeded = 0; int probe; for (probe = 0; probe < 10; probe++) { struct timespec ts = { 0, 20 * 1000 * 1000 }; rc = cbuf_tryput(&(sched.core.jbuf), &jobidD); if (rc >= 0) { any_succeeded = 1; break; } nanosleep(&ts, NULL); } TEST_ASSERT_FALSE_MESSAGE( any_succeeded, "genuine block, Req 3 AC2: with sched_allocator stuck on a " "full decisions cbuf, jbuf's single slot is already occupied " "by job C -- a further job (D) must never be accepted over a " "sustained bounded window, proving the picker itself would " "now block too, not silently drop work"); } /* --- Step 6: the outage clears. Req 3 AC2: the scheduler must resume * automatically and publish every decision produced during the bounded * window, with none lost. --- */ release_pend_wrap(); rc = pthread_join(drain_tid, NULL); TEST_ASSERT_EQUAL_INT(0, rc); TEST_ASSERT_EQUAL_INT_MESSAGE( N_TOTAL_DECISIONS, dctx.n_seen, "every decision produced before AND during the outage -- the " "prefilled backlog plus jobs A/B/C -- must eventually be drained " "and published, none silently lost (Req 3 AC2)"); for (i = 0; i < dctx.n_seen; i++) { TEST_ASSERT_TRUE_MESSAGE( dctx.seen_rcs[i] >= 0, "every drained decision's publish call must eventually " "report success"); } { int saw_a = 0, saw_b = 0, saw_c = 0; for (i = 0; i < dctx.n_seen; i++) { if (dctx.seen_jobids[i] == 9101) { saw_a = 1; } if (dctx.seen_jobids[i] == 9102) { saw_b = 1; } if (dctx.seen_jobids[i] == 9103) { saw_c = 1; } } TEST_ASSERT_TRUE_MESSAGE( saw_a && saw_b && saw_c, "jobs A, B, and C -- decided/picked during the bounded outage " "-- must all eventually be published, not just the " "pre-outage backlog"); } printf("test_worst_case_backpressure_blocks_then_recovers_no_loss: " "PASS (decisions.ncap=%d jbuf.ncap=%d, worst-case additional " "slack before genuine block = 1 decision fully enqueued + 1 " "decision built-but-blocked + 1 job buffered in jbuf)\n", ncap_decisions, ncap_jbuf); /* --- Teardown: cancel + join the real threads before tearing down * core state, matching sched_fini()'s own ordering (scheduler.c) minus * the apictx cleanup calls (xcecli_wakeup/xcecli_stop/xcecli_ctx_fini) * this fixture's wrapped xcecli_start() never fully set up (it does no * real connect, so ctx->failover.lock was never mylock_init'd) -- * deliberately skipped rather than also wrapped, consistent with this * test suite's existing detect_leaks=0 convention (meson.build). --- */ pthread_cancel(sched.core.jobpicker); pthread_cancel(sched.core.allocator); { uint64_t wake = 0; cbuf_tryput(&(sched.core.jbuf), &wake); } pthread_join(sched.core.jobpicker, NULL); pthread_join(sched.core.allocator, NULL); xce_sched_core_fini(&(sched.core)); } int main(void) { UNITY_BEGIN(); RUN_TEST(test_worst_case_backpressure_blocks_then_recovers_no_loss); return UNITY_END(); }