forked from lei.chen/myxce
109 lines
3.9 KiB
C
109 lines
3.9 KiB
C
/*
|
|
* egress.c — commit-gated scheduler egress (ha-convergence-recovery, Req 3).
|
|
*
|
|
* The scheduler is the one consumer whose actions cannot be revoked, so it must
|
|
* never observe a job whose submit entry is uncommitted (orphan-dispatch
|
|
* prevention). This file holds the commit-watermark predicate and the
|
|
* scheduler-bound collect wrapper that applies it.
|
|
*
|
|
* job_submit_committed(c, j)
|
|
* submit_lsn == 0 || submit_lsn <= committed_lsn
|
|
* One atomic load + one compare; lock-free and infallible.
|
|
* submit_lsn == 0 (pre-feature / direct create) is always exportable so
|
|
* legacy and replayed jobs are never stranded (Req 3.1, 3.2).
|
|
*
|
|
* collect_jobs_committed(c, fn, l)
|
|
* Identical contract to collect_jobs (dataex.c) — collects matching jobs,
|
|
* references and READ-locks each, leaving them for unlock_collect_jobs —
|
|
* additionally excluding any job whose submit entry is uncommitted. For
|
|
* scheduler-bound callers ONLY (xcesched_push_jobs). Status/API read
|
|
* paths keep plain collect_jobs (Req 3.4 — read consistency out of scope).
|
|
*
|
|
* Lives in its own translation unit (not dataex.c) so the controller unit test
|
|
* binary can link the REAL predicate without pulling in dataex.c's notify() /
|
|
* collect_confvars() definitions, which the test suite already stubs. This is
|
|
* the same egress-filter component the design assigns to dataex.c, factored out
|
|
* for testability (D3 — refinement, not contradiction).
|
|
*/
|
|
|
|
/* rtti.h first — defines 'var'/'typeinfo_t' used by kvmgr.h (via cluster.h). */
|
|
#include <rtti.h>
|
|
|
|
#include <errno.h>
|
|
#include <stdatomic.h>
|
|
#include <stdbool.h>
|
|
#include <stdint.h>
|
|
|
|
#include "cluster.h"
|
|
#include "persist.h"
|
|
#include "utils.h" /* CONTAINER_OF */
|
|
#include "xce/cluster_def.h"
|
|
#include "xce/types.h"
|
|
#include "xlist.h"
|
|
|
|
/* Commit-watermark predicate (Req 3.1, 3.2). Lock-free: one atomic load of
|
|
* committed_lsn and one comparison. submit_lsn == 0 means unknown/pre-feature
|
|
* and is always exportable. */
|
|
bool job_submit_committed(cluster_t* c, job_t* j)
|
|
{
|
|
if (j == NULL)
|
|
return false;
|
|
if (j->submit_lsn == 0)
|
|
return true;
|
|
uint64_t committed
|
|
= atomic_load_explicit(&c->persist.committed_lsn, memory_order_acquire);
|
|
return j->submit_lsn <= committed;
|
|
}
|
|
|
|
/* Read-lock each collected job, mirroring collect_jobs' lock_collect_jobs
|
|
* (dataex.c) so the contract — caller must release via unlock_collect_jobs — is
|
|
* identical for scheduler-bound callers. */
|
|
static void egress_lock_collected(xlist_t* l, int op)
|
|
{
|
|
for (int i = 0; i < l->count; i++) {
|
|
xce_job_t* xj = (xce_job_t*)xlist_get(l, i);
|
|
if (xj == NULL)
|
|
continue;
|
|
job_t* j = CONTAINER_OF(xj, job_t, job);
|
|
lock(j, op);
|
|
}
|
|
}
|
|
|
|
/* Scheduler-bound collect: same iteration + reference + read-lock contract as
|
|
* collect_jobs (dataex.c:294), additionally gated on the commit watermark.
|
|
* `fn`, when non-NULL, is an additional caller predicate applied first (matches
|
|
* collect_jobs' check_job_ usage in xcesched_push_jobs). */
|
|
void collect_jobs_committed(cluster_t* c, collect_check_fn_t fn, xlist_t* l)
|
|
{
|
|
iterator_t it;
|
|
|
|
lock(c->jobmgr, RTTI_LOCK_READ);
|
|
if (iter_make(&it, c->jobmgr) < 0) {
|
|
unlock(c->jobmgr);
|
|
return;
|
|
}
|
|
|
|
for (;;) {
|
|
job_t* j = NULL;
|
|
int rc = iter_next(&it, (void**)&j);
|
|
if ((rc < 0) || (rc == RTTI_ITER_NO_DATA))
|
|
break;
|
|
if (j == NULL)
|
|
continue;
|
|
|
|
/* Caller predicate first (parity with collect_jobs), then the commit
|
|
* watermark gate — exclude jobs whose submit entry is uncommitted. */
|
|
if ((fn != NULL) && (!fn(j)))
|
|
continue;
|
|
if (!job_submit_committed(c, j))
|
|
continue;
|
|
|
|
reference(j);
|
|
xlist_store(l, &(j->job));
|
|
}
|
|
|
|
iter_destroy(&it);
|
|
unlock(c->jobmgr);
|
|
|
|
egress_lock_collected(l, RTTI_LOCK_READ);
|
|
}
|