wdgzyjx/myxce
1
0
Fork 0
forked from lei.chen/myxce
Code Pull requests Activity
myxce/src/controller/job.c
Rongsong Shen fe4352fb36 fix(raft-apply-decoupling): bug-1419 close resubmit-dedup gap left by apply-after-commit deferral 
jobid_range_exists() only checks c->jobmgr, but task 3.1 deliberately
defers job construction until after commit -- so a submission that is
durably WAL-appended and awaiting its commit outcome is invisible to
that check. A client-side retry (XCECLI_FAILOVER_CALL resending the
same jobid after a transport error or failover mid-flight) sails past
the guard and appends a second, non-identical WAL_JOB_SUBMIT entry for
the same jobid -- confirmed via waldump on a live chaotic-soak node
(duplicate submits, same jobid, differing subtime/CRC) -- which can
trip persist_recover()'s post-replay apply-correctness check into a
genuine APPLY_DIVERGENT verdict.

Adds a process-lifetime in-flight-jobid-range registry in job.c that
handle_jobsubmit atomically check-and-claims immediately after the
existing jobid_range_exists check, releasing from
jobsubmit_async_ctx_release (the single choke point already shared by
the commit-continuation, apply-complete, and OOM early-out paths). A
resubmit for a still-in-flight jobid now gets -EEXIST with no second
WAL append.

Adds a matching XCE_TESTING reset seam (declared in cluster.h) so a
test that deliberately leaves a submission's commit continuation
unresolved through teardown doesn't leak its jobid range into a later
test reusing the same range -- commit_engine_fini frees pending
continuations without invoking their callback.

Regression test added to test_commit_barrier_reply.c (RED->GREEN
confirmed). Full 78-group controller_unit_test sweep: 782 tests, 0
failures.

This fixes only the WAL duplicate-submit mechanism (layer 1 of
bug-1419). The deeper PreVote-freshness/apply-divergence liveness
deadlock (layer 2 -- no recovery path when the only fresh-enough node
is apply-divergent-fenced) remains open and needs its own design
discussion; buglog.json updated accordingly.
2026-07-07 14:28:53 +00:00

1915 lines
68 KiB
C
Raw Blame History

#include <errno.h>
#include <signal.h>
#include <stdbool.h>
#include <string.h>
#include <time.h>
#include "cluster.h"
#include "controller.capnp.h"
#include "jobj.h"
#include "persist.h"
#include "preason.h"
#include "xce/cluster_def.h"
#include "xce/errnox.h"
#include "xce/types.h"
#include "xstring.h"
static int record_job_start_fact(cluster_t*, job_t*, job_pidinfo_t*);
static int record_job_finish_fact(cluster_t*, job_t*, xce_jobfinish_t*);
static int record_job_fail_fact(cluster_t*, job_t*, job_failinfo_t*);
static int mark_job_fail(cluster_t*, job_t*, job_failinfo_t*);
static int mark_job_update(cluster_t*, job_t*, int);
/* A job is terminal once it carries a DONE or EXIT bit. Terminal jobs reject
* backward (terminal -> non-terminal) transitions and treat re-applied
* terminal transitions as duplicates; forward bit-OR merges (PDONE, PERR,
* suspend bits) on a terminal job remain legal. */
#define JOB_STATE_TERMINAL_BITS (JOB_STAT_DONE | JOB_STAT_EXIT)
static inline bool job_state_is_terminal(int state)
{
return (state & JOB_STATE_TERMINAL_BITS) != 0;
}
/* Shared backward predicate for START and UPDATE (Req 5.2): a transition is
* backward when the job is already terminal but the requested next state drops
* every terminal bit it currently holds (e.g. START overwriting DONE with a
* bare JOB_STAT_RUN, or an UPDATE clearing the terminal bits). A forward merge
* that preserves at least one terminal bit (postdone OR'ing PDONE) is NOT
* backward. */
static inline bool job_state_is_backward(int cur, int next)
{
return job_state_is_terminal(cur) && !job_state_is_terminal(next);
}
static job_t* create_job(cluster_t* c, xce_job_t* j)
{
job_t* job = NULL;
job = makeJob(c->jobmgr, j);
return job;
}
/* bug-1419: in-flight submission registry, closing the resubmit-dedup gap
* task 3.1's apply-after-commit deferral opened.
*
* Before task 3.1, jobid_range_exists()'s c->jobmgr check was airtight
* against a client's own retried submit (xcecli_jobsubmit wrapped in
* XCECLI_FAILOVER_CALL, client_failover.c) because job construction ran
* synchronously in this same request: by the time a retry could possibly
* reach the controller again, the first attempt's job already existed in
* c->jobmgr. client_failover.c's own doc comment names this "the
* controller's client-jobid replay guard" and relies on it to reclassify a
* retried write's EEXIST as success (xce_failover_classify's
* "Idempotent-replay rule").
*
* Task 3.1 deliberately deferred job construction to AFTER commit (Req 2.1,
* 2.3) — correct for its own goal (no compensating rollback needed on a
* failed commit) — but this reopened exactly the window the replay guard
* depended on: a submit that is durably appended and will commit, but has
* not yet run through apply_committed_entry() and landed in c->jobmgr, is
* invisible to jobid_range_exists(). A client retry (e.g. after a
* transport error or failover mid-flight, before it ever saw the first
* attempt's reply) landing in that window sails through the check and
* appends a SECOND WAL_JOB_SUBMIT for the same jobid with a fresh subtime —
* confirmed via waldump on a live chaotic-soak node (non-byte-identical
* duplicate submits, same jobid, differing subtime/CRC), which can
* ultimately trip persist_recover()'s post-replay apply-correctness check
* (task 5.1) into a genuine APPLY_DIVERGENT verdict on restart.
*
* Fix: track jobids that have been accepted for WAL append but have not
* yet had their outcome fully resolved (failed, or applied) in a small
* process-local registry, and treat a hit there identically to a
* c->jobmgr hit — both mean "this exact write already happened or is
* already in flight," which is precisely what a retried write must see as
* a duplicate. Populated in handle_jobsubmit() right after this check
* passes (before any WAL work); cleared in jobsubmit_async_ctx_release()
* (the single choke point already reached on every resolution path:
* commit failure, apply completion, and OOM/alloc-failure early-outs). */
typedef struct inflight_submit {
uint64_t base_jobid;
uint64_t array_size;
struct inflight_submit* next;
} inflight_submit_t;
static pthread_mutex_t s_inflight_submits_lock = PTHREAD_MUTEX_INITIALIZER;
static inflight_submit_t* s_inflight_submits = NULL;
static bool inflight_range_overlaps_locked(uint64_t base_jobid,
uint64_t array_size)
{
for (inflight_submit_t* n = s_inflight_submits; n != NULL; n = n->next) {
uint64_t lo = base_jobid, hi = base_jobid + array_size;
uint64_t nlo = n->base_jobid, nhi = n->base_jobid + n->array_size;
if (lo < nhi && nlo < hi)
return true;
}
return false;
}
/* Atomically check-then-register under one lock hold, closing the TOCTOU a
* separate check + separate add would leave between two concurrent
* handle_jobsubmit calls for the same jobid (a retry's new connection can
* race the still-in-flight original request under this codebase's
* vtask-dispatched, not single-threaded-per-jobid, request handling).
* Returns true (registered) if no overlap existed; false (already in
* flight, NOT registered) if one did — the caller must reject on false
* exactly as it already does for a c->jobmgr hit. */
static bool inflight_range_try_add(uint64_t base_jobid, uint64_t array_size)
{
bool ok = false;
pthread_mutex_lock(&s_inflight_submits_lock);
if (!inflight_range_overlaps_locked(base_jobid, array_size)) {
inflight_submit_t* n = calloc(1, sizeof(*n));
if (n != NULL) {
n->base_jobid = base_jobid;
n->array_size = array_size;
n->next = s_inflight_submits;
s_inflight_submits = n;
ok = true;
}
/* calloc failure: fall through with ok=false -- rejecting an
* accepted-looking submit on OOM is safe (the client's own retry
* logic already handles a rejected attempt); silently admitting an
* untracked in-flight submission on OOM is not. */
}
pthread_mutex_unlock(&s_inflight_submits_lock);
return ok;
}
static void inflight_range_remove(uint64_t base_jobid, uint64_t array_size)
{
pthread_mutex_lock(&s_inflight_submits_lock);
inflight_submit_t** pp = &s_inflight_submits;
while (*pp != NULL) {
if ((*pp)->base_jobid == base_jobid && (*pp)->array_size == array_size) {
inflight_submit_t* dead = *pp;
*pp = dead->next;
free(dead);
break;
}
pp = &(*pp)->next;
}
pthread_mutex_unlock(&s_inflight_submits_lock);
}
#ifdef XCE_TESTING
/* Test-only seam: s_inflight_submits is process-lifetime, not tied to any
* cluster_t fixture. In production that's fine -- a restart wipes it along
* with every other in-memory registry, and persist_recover() rebuilds
* c->jobmgr fresh from the WAL. But a unit-test binary reuses the same
* process across many fixtures, and a test that deliberately leaves a
* submission's commit continuation unresolved through teardown (bug-1419's
* own regression test) has no other event left to release its entry --
* commit_engine_fini() frees pending continuations without invoking their
* callback. Without this, that entry outlives the fixture and can spuriously
* reject an unrelated later test that reuses the same jobid range. */
void inflight_submits_reset_for_test(void)
{
pthread_mutex_lock(&s_inflight_submits_lock);
inflight_submit_t* n = s_inflight_submits;
while (n != NULL) {
inflight_submit_t* dead = n;
n = n->next;
free(dead);
}
s_inflight_submits = NULL;
pthread_mutex_unlock(&s_inflight_submits_lock);
}
#endif
/* True if any jobid in [base_jobid, base_jobid + array_size) already exists
* in c->jobmgr (raft-apply-decoupling task 3.1). Checked BEFORE any WAL work
* so a colliding submission is still rejected immediately with no side
* effect, matching the fast-fail behavior create_job()'s own key_find
* guard used to provide when job construction ran before the commit
* barrier — now that construction is deferred to after commit, a collision
* detected only post-commit could no longer be cleanly reported as "never
* happened".
*
* bug-1419: this check alone is no longer sufficient (see
* inflight_range_try_add's doc comment) — handle_jobsubmit additionally
* calls inflight_range_try_add() after this check passes, which atomically
* closes the remaining not-yet-applied window this function cannot see. */
static bool jobid_range_exists(cluster_t* c, uint64_t base_jobid,
uint64_t array_size)
{
for (uint64_t i = 0; i < array_size; i++) {
job_t* existing = NULL;
int found;
lock(c->jobmgr, RTTI_LOCK_READ);
found = key_find(c->jobmgr, (void*)(base_jobid + i),
(void**)&existing);
unlock(c->jobmgr);
if (found > 0 && existing != NULL)
return true;
}
return false;
}
/* Build one xce_job_t submission payload per array element, WITHOUT
* constructing or registering any job object (raft-apply-decoupling task
* 3.1 / Req 2.1, 2.3) — these are plain heap structs carrying only the data
* needed to build each entry's WAL payload; `free_job_submit_payloads` frees
* them. `submit->submit` is borrowed (owned by the caller's request), never
* duplicated or freed here. Returns 0 on success, -ENOMEM on allocation
* failure (payloads already stored are left for the caller to free). */
static int build_job_submit_payloads(uint64_t base_jobid,
xce_job_submit_t* submit, xlist_t* plist)
{
time_t now = time(NULL);
xlist_init(plist, submit->array_size);
for (uint64_t i = 0; i < submit->array_size; i++) {
xce_job_t* jp = calloc(1, sizeof(*jp));
if (jp == NULL)
return -ENOMEM;
if (xlist_store(plist, jp) < 0) {
free(jp);
return -ENOMEM;
}
jp->submit = submit->submit;
jp->jobid = base_jobid + i;
jp->sub_time = now;
jp->queue_time = now;
jp->state = JOB_STAT_PEND;
}
return 0;
}
static int xlist_free_job_payload_(xce_job_t* j, void* arg)
{
(void)arg;
free(j);
return 0;
}
static void free_job_submit_payloads(xlist_t* plist)
{
xlist_iter(plist, (xlist_iter_fn_t)xlist_free_job_payload_, NULL);
xlist_fini(plist);
}
int load_jobs(void* c)
{
return 0;
}
/* Reconstruct a job from a WAL_JOB_SUBMIT Cap'n Proto payload.
* Uses the jobid embedded in the serialized xce_job_t (not the caller's arg).
* entry_lsn is the WAL entry LSN of this submit record (ha-convergence-recovery
* 3.3): the restored job is tagged with it so its commit status is decidable at
* scheduler-export time. 0 = unknown (pre-feature / direct create) = exportable.
* Returns -EEXIST if the job is already present (idempotent replay guard). */
int restore_job(cluster_t* c, uint64_t jobid_unused,
const void* capnp_payload, size_t len, uint64_t entry_lsn)
{
(void)jobid_unused;
if (!capnp_payload || len == 0)
return -EINVAL;
struct capn c_ctx;
capn_init_mem(&c_ctx, (const uint8_t*)capnp_payload, len, 0);
Job_ptr jp;
jp.p = capn_getp(capn_root(&c_ctx), 0, 1);
struct Job cj;
xce_job_t j;
memset(&j, 0, sizeof(j));
read_Job(&cj, jp);
decode_Job(&j, &cj);
capn_free(&c_ctx);
/* Safety guard: skip if already present (WAL replay may see duplicates
* when recovery restarts from a segment boundary). */
job_t* existing = NULL;
lock(c->jobmgr, RTTI_LOCK_READ);
int found = key_find(c->jobmgr, (void*)j.jobid, (void**)&existing);
unlock(c->jobmgr);
if (found > 0 && existing != NULL) {
/* j was fully decoded above (submit/allocs/pidinfo/pend_reasons all
* populated by decode_Job) and create_job() is never reached on this
* path, so ownership never moved -- every decoded pointer is still
* ours to release here (ha-entity-snapshots task 5.2: fixes the
* pre-existing decode leak on this early-return path, called out in
* tasks.md's Implementation Notes). */
free_Job(&j);
return -EEXIST;
}
job_t* job = create_job(c, &j);
int saved_rc = job ? 0 : -errno;
/* create_job() (jobj.c job_init) takes ownership of j.allocs/j.pidinfo/
* j.pend_reasons via MOVE_PTR (nulling them here on success) and makes
* its OWN deduplicated copies of j.submit/j.ckptinfo (makeJobsub/
* makeString) -- so j.submit, the one field decode_Job always populates
* that job_init never moves, is never released by create_job and must
* always be freed here. free_Job() null-checks each of its four fields
* before freeing (submit/allocs/pidinfo/pend_reasons), so calling it
* unconditionally is safe whether create_job succeeded (three of the
* four already nulled by the MOVE_PTR above) or failed before job_init
* ran any MOVE_PTR (none nulled, all four correctly released) -- never a
* double free either way (ha-entity-snapshots task 5.2, pre-existing
* leak noted in tasks.md's Implementation Notes). */
free_Job(&j);
if (!job)
return saved_rc;
/* ha-convergence-recovery (3.3): tag the restored job with its submit
* entry's LSN so the egress watermark predicate can decide its commit
* status (job_submit_committed). On a promoted primary, replayed entries
* sit at or below the promotion truncation point, so restored jobs are
* exportable by construction. */
job->submit_lsn = entry_lsn;
return 0;
}
/* Apply a recorded job state transition during WAL recovery.
* Looks up the job (which must already exist from WAL_JOB_SUBMIT replay)
* and overwrites its state with the value stored in the WAL entry. */
int restore_job_state(cluster_t* c, uint64_t jobid, uint32_t state)
{
job_t* j = NULL;
lock(c->jobmgr, RTTI_LOCK_READ);
key_find(c->jobmgr, (void*)jobid, (void**)&j);
unlock(c->jobmgr);
if (!j)
return -ENOENT;
lock(j, RTTI_LOCK_WRITE);
j->job.state = (int)state;
unlock(j);
return 0;
}
static int get_nslots_in_alloc(xce_host_alloc_t* alloc)
{
int i;
for (i = 0; i < alloc->nallocs; i++) {
if (strcmp(alloc->allocs[i]->resname, "slots") == 0) {
return (int)alloc->allocs[i]->value;
}
}
return 1;
}
static void free_launchinfo(xce_launch_t* launch)
{
int i;
for (i = 0; i < launch->nhosts; i++) {
FREEUP(launch->hosts[i]);
}
FREEUP(launch->hosts);
}
static int mk_launchinfo(xce_launch_t* launch, xce_job_alloc_t* alloc)
{
int i;
int rc = 0;
memset(launch, 0x0, sizeof(*launch));
launch->hosts = (xce_host_launch_t**)calloc(alloc->nhosts,
sizeof(xce_host_launch_t*));
if (launch->hosts == NULL) {
return -ENOMEM;
}
launch->nhosts = alloc->nhosts;
for (i = 0; i < alloc->nhosts; i++) {
launch->hosts[i]
= (xce_host_launch_t*)calloc(1, sizeof(xce_host_launch_t));
if (launch->hosts[i] == NULL) {
rc = -ENOMEM;
break;
}
launch->hosts[i]->host = alloc->allocs[i]->host;
launch->hosts[i]->nranks = get_nslots_in_alloc(alloc->allocs[i]);
}
if (rc < 0) {
free_launchinfo(launch);
}
return rc;
}
static int dispatch_a_job_(cluster_t* c, endpoint_t* ep, xce_job_alloc_t* alloc)
{
int rc = 0;
job_t* j = NULL;
worker_t* w = NULL;
xce_host_alloc_t** allocs;
int nc;
xce_job_exec_t einfo;
xce_launch_t launchinfo;
TRACE("%s jobid %ld", __func__, alloc->jobid);
lock(c->jobmgr, RTTI_LOCK_READ);
rc = key_find(c->jobmgr, (void*)alloc->jobid, (void**)&j);
if (j != NULL) {
reference(j);
}
unlock(c->jobmgr);
if (rc < 0) {
return rc;
}
if (j == NULL) {
PERROR("jobmgr_lockjob");
return -errno;
}
nc = alloc->nhosts;
if (nc <= 0) {
dereference(j);
return -EINVAL;
}
/* FIXME: change case of nc > 1 just hard coded mpi as
* the real job type later
*/
mk_launchinfo(&launchinfo, alloc);
allocs = (xce_host_alloc_t**)alloc->allocs;
/* using head node of job */
lock(c->workermgr, RTTI_LOCK_READ);
rc = key_find(c->workermgr, (void*)allocs[0]->host, (void**)&w);
if (w != NULL) {
reference(w);
}
unlock(c->workermgr);
if (rc < 0) {
return rc;
}
if (w == NULL) {
free_launchinfo(&launchinfo);
return -errno;
}
lock(w, RTTI_LOCK_READ);
if ((w->ep == NULL) || (w->host.status != 0)) {
ERROR("worker %s is not in correct state, status %x ep %p",
w->host.name, w->host.status, w->ep);
free_launchinfo(&launchinfo);
SET_ERROR(E_XCE_BAD_WORKER);
unlock(w);
dereference(w);
return -errno;
}
/* FIXME: we may consider to optimize when a job has enter a queue? */
if (1) {
char* q_pre_exec;
char* q_post_exec;
rc = get_queue_exec(c, j->job.submit->queue, &q_pre_exec, &q_post_exec);
if (rc < 0) {
unlock(w);
dereference(w);
dereference(j);
free_launchinfo(&launchinfo);
return rc;
}
if (j->job.submit->start_info->q_pre_exec != NULL) { }
if (j->job.submit->start_info->q_post_exec != NULL) { }
j->job.submit->start_info->q_pre_exec = q_pre_exec;
j->job.submit->start_info->q_post_exec = q_post_exec;
}
einfo.jobid = alloc->jobid;
einfo.start_info = j->job.submit->start_info;
einfo.launchinfo = &launchinfo;
rc = xce_startjob_req(w->ep, &einfo);
unlock(w);
dereference(w);
free_launchinfo(&launchinfo);
if (j->job.allocs != NULL) {
free_JobAlloc_ptr(&(j->job.allocs));
}
j->job.allocs = (xce_job_alloc_t*)calloc(1, sizeof(xce_job_alloc_t));
j->job.allocs->jobid = alloc->jobid;
j->job.allocs->nhosts = nc;
MOVE_PTR(j->job.allocs->allocs, allocs);
alloc->allocs = NULL;
if (rc < 0) {
SET_ERROR(-rc);
PERROR("xce_startjob_req");
return rc;
}
return rc;
}
int job_ctrl(cluster_t* c, xce_jobctrl_t* ctrinfo)
{
uint64_t jobid = ctrinfo->jobid;
job_t* j = NULL;
int rc = 0;
TRACE("%s job_ctrl job %ld", __func__, jobid);
lock(c->jobmgr, RTTI_LOCK_READ);
rc = key_find(c->jobmgr, (void*)jobid, (void**)&j);
if (j != NULL) {
reference(j);
}
unlock(c->jobmgr);
if (rc < 0) {
return rc;
}
if (j == NULL) {
PERROR("jobmgr_lockjob");
return -ENOENT;
}
/* bug-447: key_find returns 1 on found (binary librtti, unnormalizable
* upstream). Normalize the found-rc to 0 so the not-started siginfo
* branch below (which appends via the mark_job_* hooks but never
* reassigns rc) returns 0 to the handler's commit-barrier pattern
* instead of leaking 1 as the reply status. Real errors are
* untouched: rc < 0 returned above, the j == NULL miss returned
* -ENOENT, and every other branch sets rc itself. */
rc = 0;
/* FIXME: eliminate held multiple locks */
lock(j, RTTI_LOCK_READ);
if (j->job.allocs != NULL) {
xce_host_alloc_t** allocs = j->job.allocs->allocs;
worker_t* w = NULL;
/* job control command will be sent to first node
* for parallel jobs
*/
lock(c->workermgr, RTTI_LOCK_READ);
rc = key_find(c->workermgr, (void*)allocs[0]->host, (void**)&w);
if (w != NULL) {
reference(w);
}
unlock(c->workermgr);
if (rc < 0) {
unlock(j);
dereference(j);
return rc;
}
if (w != NULL) {
lock(w, RTTI_LOCK_READ);
/* FIXME: better method to check host status */
if ((w->ep != NULL) && (w->host.status == 0)) {
rc = xce_jobctrl_req(w->ep, ctrinfo);
if (rc < 0) {
ERROR("job ctrl %ld on host %s returns %d", jobid,
w->host.name, rc);
}
} else {
ERROR("worker %s is not in correct state, status %x ep %p",
w->host.name, w->host.status, w->ep);
rc = -errno;
}
unlock(w);
dereference(w);
} else {
rc = -errno;
}
} else {
/* job has not been started yet */
job_failinfo_t failinfo = {
.jobid = jobid,
};
xce_use(&failinfo);
switch (ctrinfo->action) {
case JobControl_u_siginfo:
switch (ctrinfo->u.siginfo->signo) {
case SIGCONT:
if (j->job.state
& (JOB_STAT_PSUSP | JOB_STAT_USUSP | JOB_STAT_SSUSP)) {
int state = j->job.state;
/* resume suspend job? */
state &= ~(JOB_STAT_PSUSP | JOB_STAT_USUSP);
mark_job_update(c, j, state);
}
break;
case SIGSTOP:
case SIGTSTP:
if (!(j->job.state
& (JOB_STAT_PSUSP | JOB_STAT_USUSP | JOB_STAT_SSUSP))) {
int state = j->job.state;
state |= JOB_STAT_PSUSP;
mark_job_update(c, j, state);
}
break;
case SIGKILL:
case SIGTERM:
default:
if (j->job.state
& (JOB_STAT_DONE | JOB_STAT_EXIT | JOB_STAT_ERR)) {
break;
}
/* FIXME: */
failinfo.code = 128 + ctrinfo->u.siginfo->signo;
failinfo.detail = "Job killed by signal";
mark_job_fail(c, j, &failinfo);
break;
}
break;
case JobControl_u_ckptoption:
if (j->job.state & (JOB_STAT_DONE | JOB_STAT_EXIT | JOB_STAT_ERR)) {
ERROR("job <%ld> has been finished, no checkpoint can be done.",
jobid);
} else {
ERROR("job %ld has not been started, no checkpoint can be done",
jobid);
}
rc = -EINVAL;
break;
default:
rc = -EINVAL;
break;
}
}
unlock(j);
return rc;
}
static void notify_dispatch(cluster_t* c, endpoint_t* ep, uint64_t jobid,
int code)
{
/* clang-format off */
job_failinfo_t info = {
.jobid = jobid,
.code = code
};
xce_notify_data_t data = {
.event = NotifyData_data_jobDispatchFail,
.data = {
.dispatchfail = &info
}
};
/* clang-format on */
notify(c, ep, &data);
}
int handle_jobdispatch(cluster_t* c, endpoint_t* ep, xce_request_t* request)
{
int rc = 0;
int i;
xce_job_dispatch_t* dispatchmsg = request->body.dispatchmsg;
xce_reply_t reply;
/* reply job dispatch that we'll handle request */
init_reply_by_req(&reply, request);
reply.status = 0;
reply.data_which = RequestReply_data_nodata;
xce_reply(ep, c->myhost, &reply);
for (i = 0; i < dispatchmsg->njobs; i++) {
int rc1;
xce_job_alloc_t* alloc = NULL;
MOVE_PTR(alloc, dispatchmsg->allocs[i]);
rc1 = dispatch_a_job_(c, ep, alloc);
if (rc1 < 0) {
/* notify dispatch fail */
notify_dispatch(c, ep, alloc->jobid, rc1);
} else {
/* notify dispatch ok */
notify_dispatch(c, ep, alloc->jobid, 0);
}
free_JobAlloc_ptr(&(alloc));
}
return rc;
}
/* raft-apply-decoupling task 3.2: shared state for one handle_jobsubmit()
* call's asynchronously-resolving per-entry commit continuations. Exactly
* one instance per client request — an array submission registers one
* continuation per array element, all sharing this struct, so the client
* still gets exactly one reply for the whole batch (matching pre-3.2
* per-request reply semantics), never one reply per array element.
*
* Lifetime: allocated in handle_jobsubmit() before it returns; released by
* whichever per-entry continuation (jobsubmit_commit_continuation) or
* apply-complete callback (jobsubmit_apply_complete) observes `pending`
* reach 0 — the last one out closes the door.
*
* IMPORTANT: a COMMITTED outcome does NOT decrement `pending` by itself
* (see jobsubmit_commit_continuation below) — task 3.2's own contract is
* that a committed entry is handed off to the not-yet-built ordered apply
* pump (task 6.1/6.2), and only jobsubmit_apply_complete (invoked BY that
* future pump) decrements `pending` for it. A batch with any committed
* entry is therefore only ever freed once task 6 exists and actually
* applies it. This mirrors the design's already-accepted "durably
* committed, not yet applied" posture for retained facts during quorum
* loss (Req 6.1) — it is not a leak. */
typedef struct {
cluster_t* c;
endpoint_t* ep; /* ref'd via endpoint_ref(); released exactly
* once, in jobsubmit_async_ctx_release() */
uint64_t seq; /* xce_request_t.seq, copied for the reply */
uint32_t opcode; /* xce_request_t.opcode, copied for the reply */
uint64_t jobid; /* base jobid of the (possibly array) submission,
* echoed back on the success reply (task 6.2) --
* matches pre-3.1 semantics (git history commit
* 6f1fa97), which always reported the BASE id,
* never a per-element id */
uint64_t array_size; /* bug-1419: paired with jobid to remove the
* matching inflight_range_add() entry on
* release -- see jobid_range_exists' doc
* comment for the full contract */
pthread_mutex_t lock; /* guards pending/replied below */
int pending; /* entries not yet FULLY resolved (see above) */
int replied; /* 1 once a reply has been sent for this batch */
} jobsubmit_async_ctx_t;
static void jobsubmit_async_ctx_release(jobsubmit_async_ctx_t* actx)
{
/* bug-1419: this is the single choke point every resolution path
* (commit failure, apply completion, and the OOM/alloc-failure
* early-outs in handle_jobsubmit below) already reaches -- clearing
* the in-flight registration here means a retry is only ever
* dedup-blocked while this exact submission's outcome is genuinely
* still unresolved, never after. */
inflight_range_remove(actx->jobid, actx->array_size);
endpoint_unref(actx->ep);
pthread_mutex_destroy(&actx->lock);
free(actx);
}
/* Sends exactly one failure reply for the whole batch — the first failure
* observed wins; later failures/hand-offs never send a second, possibly
* contradictory reply for the same original request (Req 9.3, read at the
* per-REQUEST granularity a job-array submission actually has one of). */
static void jobsubmit_reply_failure_once(jobsubmit_async_ctx_t* actx, int rc)
{
xce_reply_t reply;
int send = 0;
pthread_mutex_lock(&actx->lock);
if (!actx->replied) {
actx->replied = 1;
send = 1;
}
pthread_mutex_unlock(&actx->lock);
if (!send)
return;
memset(&reply, 0, sizeof(reply));
reply.seq = actx->seq;
reply.opcode = actx->opcode;
reply.status = rc;
reply.data_which = RequestReply_data_nodata;
xce_reply(actx->ep, actx->c->myhost, &reply);
}
/* Sends exactly one success reply for the whole batch, carrying the base
* jobid — the sibling of jobsubmit_reply_failure_once, gated on the SAME
* actx->replied flag (Req 9.3) so a batch that already sent a failure reply
* for a sibling element never follows it with a contradictory success reply,
* and vice versa. Only ever reached from jobsubmit_apply_complete once every
* element's continuation has resolved (raft-apply-decoupling task 6.2). */
static void jobsubmit_reply_success_once(jobsubmit_async_ctx_t* actx)
{
xce_reply_t reply;
int send = 0;
pthread_mutex_lock(&actx->lock);
if (!actx->replied) {
actx->replied = 1;
send = 1;
}
pthread_mutex_unlock(&actx->lock);
if (!send)
return;
memset(&reply, 0, sizeof(reply));
reply.seq = actx->seq;
reply.opcode = actx->opcode;
reply.status = 0;
reply.data_which = RequestReply_data_jobid;
reply.data.jobid = actx->jobid;
xce_reply(actx->ep, actx->c->myhost, &reply);
}
/* raft-apply-decoupling task 6 hand-off target: invoked by the ordered apply
* pump (persist_apply_pump_run(), task 6.1/6.2) immediately after it applies
* this entry via apply_committed_entry() — task 6.2's own contract, "fire a
* command's registered continuation immediately after its entry applies".
* See jobsubmit_commit_continuation's COMMITTED branch below, which
* registers this as the hand-off target instead of applying or replying
* itself.
*
* apply_committed_entry()'s WAL_JOB_SUBMIT case (persist.c) already
* constructs the job (via restore_job()) AND fires the HOOK_DETAIL_JOB_NEW
* AFTER_EVENT notification, independently per entry, strictly BEFORE the
* apply pump reaches the persist_take_pending_apply()/invoke step that calls
* this function — so by the time this function runs, the job is already
* visible in c->jobmgr and already notified. This function therefore only
* sends the batch's success reply (echoing the base jobid) once every entry
* has been applied; it never reconstructs or re-notifies anything — see the
* removed-code note above handle_jobsubmit's old (pre-3.2) construct/notify
* block for the full reasoning (git history, commit 6f1fa97, task 3.1). */
static void jobsubmit_apply_complete(void* ctx, uint64_t lsn)
{
jobsubmit_async_ctx_t* actx = (jobsubmit_async_ctx_t*)ctx;
int done;
(void)lsn;
pthread_mutex_lock(&actx->lock);
actx->pending--;
done = (actx->pending <= 0);
pthread_mutex_unlock(&actx->lock);
if (done) {
jobsubmit_reply_success_once(actx);
jobsubmit_async_ctx_release(actx);
}
}
/* raft-apply-decoupling task 3.2: per-entry outcome of one array element's
* commit_submit_async() registration (persist_submit_job_payloads_async's
* `fn` callback).
*
* On FAILED/timed-out (rc != 0): reply failure now — no apply is needed on
* this path (Req 2.3). This entry has nothing further to wait on, so it
* decrements `pending` itself.
*
* On COMMITTED (rc == 0): task 3.2 must NOT apply the job or reply here
* (the task's own explicit instruction) — hand the entry off to the
* not-yet-built ordered apply pump (task 6.1/6.2) via
* persist_register_pending_apply(), the producer-side hand-off this task
* builds. `pending` is intentionally NOT decremented here for a committed
* entry — see jobsubmit_async_ctx_t's doc comment. */
static void jobsubmit_commit_continuation(void* ctx, uint64_t lsn, int rc)
{
jobsubmit_async_ctx_t* actx = (jobsubmit_async_ctx_t*)ctx;
if (rc != 0) {
int done;
WARN("handle_jobsubmit: lsn=%lu commit failed/timed out (rc=%d) — "
"rejecting; no job ever constructed for it",
(unsigned long)lsn, rc);
jobsubmit_reply_failure_once(actx, rc);
pthread_mutex_lock(&actx->lock);
actx->pending--;
done = (actx->pending <= 0);
pthread_mutex_unlock(&actx->lock);
if (done)
jobsubmit_async_ctx_release(actx);
return;
}
/* COMMITTED: hand off to task 6's ordered apply pump instead of
* applying/replying directly from this continuation — see
* persist_register_pending_apply's doc comment (persist.h) for the
* full hand-off contract. */
persist_register_pending_apply(&actx->c->persist, lsn,
jobsubmit_apply_complete, actx);
}
int handle_jobsubmit(cluster_t* c, endpoint_t* ep, xce_request_t* request)
{
uint64_t jobid;
xce_job_submit_t* submit;
xce_reply_t reply;
xlist_t plist; /* xce_job_t* WAL payloads, pre-commit */
uint64_t* lsns = NULL;
int rc;
submit = request->body.jobsubmsg;
init_reply_by_req(&reply, request);
if (NULLSTR(submit->submit->user) || NULLSTR(submit->submit->queue)
|| (submit->submit->start_info == NULL)
|| NULLSTR(submit->submit->start_info->cmdline)) {
errno = EINVAL;
reply.status = -errno;
reply.data_which = RequestReply_data_nodata;
xce_reply(ep, c->myhost, &reply);
return -errno;
}
xce_run_hooks(&(c->hooksys), XCE_CONTROLLER_HOOK_JOB, HOOK_BEFORE_EVENT,
HOOK_DETAIL_JOB_NEW, c, submit);
jobid = submit->jobid;
if (jobid_range_exists(c, jobid, submit->array_size)) {
errno = EEXIST;
reply.status = -errno;
reply.data_which = RequestReply_data_nodata;
xce_reply(ep, c->myhost, &reply);
return -errno;
}
/* bug-1419: atomically claim this jobid range as in-flight, closing the
* window jobid_range_exists' c->jobmgr check alone cannot see (see its
* doc comment). A false return means another submission (a genuine
* concurrent collision, or -- the actual motivating case -- this exact
* request's own retried resend racing its still-in-flight original
* attempt) already holds this range; reject exactly as the
* jobid_range_exists branch above does. Released in
* jobsubmit_async_ctx_release() once this submission's outcome is
* fully resolved; the two error returns below (which occur BEFORE
* actx exists to carry that release) remove it directly. */
if (!inflight_range_try_add(jobid, submit->array_size)) {
errno = EEXIST;
reply.status = -errno;
reply.data_which = RequestReply_data_nodata;
xce_reply(ep, c->myhost, &reply);
return -errno;
}
/* Build only the WAL payload data for each array element — no job
* object is constructed or inserted into c->jobmgr until the commit
* below succeeds (raft-apply-decoupling task 3.1 / Req 2.1, 2.3): a
* commit that fails or times out must leave the job table untouched,
* and the only way to guarantee that structurally is to never
* construct anything until commit has already returned success. */
rc = build_job_submit_payloads(jobid, submit, &plist);
if (rc == 0) {
lsns = calloc((size_t)plist.count, sizeof(*lsns));
if (lsns == NULL)
rc = -ENOMEM;
}
if (rc < 0) {
free_job_submit_payloads(&plist);
inflight_range_remove(jobid, submit->array_size);
reply.status = rc;
reply.data_which = RequestReply_data_nodata;
xce_reply(ep, c->myhost, &reply);
return rc;
}
/* Explicit raft step (raft-apply-decoupling item #1 / bug-1306, task
* 3.1; converted to the continuation mechanism by task 3.2): register
* every payload's commit outcome instead of blocking on the majority
* barrier. This handler returns as soon as every payload is durably
* appended locally and its outcome-resolution path is established —
* BEFORE any of them necessarily commit. From this point on, job
* construction and the client reply happen from
* jobsubmit_commit_continuation / jobsubmit_apply_complete (above),
* never again from this stack frame.
*
* REMOVED (raft-apply-decoupling task 3.2) relative to task 3.1's
* version of this function: the post-commit create_job()-loop +
* jlist-build + xce_run_hooks(AFTER_EVENT, HOOK_DETAIL_JOB_NEW) block
* (bug-1408's fix: unconditional per-successfully-constructed-element
* notification) is gone, not merely deferred. It is SUPERSEDED, not
* resurrected, by apply_committed_entry()'s existing WAL_JOB_SUBMIT
* case (persist.c, raft-apply-decoupling task 2.2, already built):
* that function already does restore_job() + fires HOOK_DETAIL_JOB_NEW
* per entry, independently per array element — which is structurally
* immune to the bug-1408 class (a sibling's apply failure can never
* suppress another sibling's notification, because each LSN applies
* through its own independent call). The full reasoning is preserved
* in this task's status report; the removed code itself remains in
* git history at commit 6f1fa97. Whoever builds task 6.1/6.2 should
* call apply_committed_entry() (not resurrect this loop) and should
* expect jobsubmit_apply_complete to need only the success reply. */
jobsubmit_async_ctx_t* actx = calloc(1, sizeof(*actx));
if (actx == NULL) {
free_job_submit_payloads(&plist);
free(lsns);
inflight_range_remove(jobid, submit->array_size);
reply.status = -ENOMEM;
reply.data_which = RequestReply_data_nodata;
xce_reply(ep, c->myhost, &reply);
return -ENOMEM;
}
actx->c = c;
actx->ep = ep;
actx->seq = request->seq;
actx->opcode = request->opcode;
actx->jobid = jobid;
actx->array_size = submit->array_size;
actx->pending = plist.count;
actx->replied = 0;
pthread_mutex_init(&actx->lock, NULL);
endpoint_ref(ep);
(void)persist_submit_job_payloads_async(&c->persist, &plist, lsns,
jobsubmit_commit_continuation,
actx);
/* Every payload's WAL data is no longer needed once appended: the
* future apply pump (task 6) re-reads each committed entry's payload
* from the WAL itself — the same way restart replay already does —
* not from this in-memory copy. persist_submit_job_payloads_async has
* ALREADY invoked jobsubmit_commit_continuation, synchronously or
* asynchronously, exactly once for every payload by (or after) the
* time it returns; its own return value carries no information this
* function still needs to act on. */
free_job_submit_payloads(&plist);
free(lsns);
/* The calling thread returns here, before any payload's commit outcome
* is necessarily known (raft-apply-decoupling task 3.2, Req 1.1, 1.4,
* 2.1, 2.3) — every reply from this point on is sent by
* jobsubmit_reply_failure_once or (once task 6 exists)
* jobsubmit_apply_complete, off this stack frame. */
return 0;
}
/* Guarded state-transition core for a worker-reported START, extracted from
* mark_job_start (raft-apply-decoupling task 2.1, Req 3.3/7.1/7.2; mark_job_start
* itself was renamed record_job_start_fact by task 4.1, which stopped calling
* this core from the receipt path — see that function's own doc comment).
* Contains ONLY the monotonic-rank guard and the state mutation — it fires NO
* hook itself, so hook-firing can be relocated by a later task (the shared
* apply seam) without touching this guard logic again. Safe to invoke
* directly, bypassing the worker-report receipt path (on_job_start) entirely,
* as long as the caller already holds `job` under RTTI_LOCK_WRITE. */
job_transition_result_t apply_job_start_transition(job_t* job,
job_pidinfo_t* info)
{
int cur = job->job.state;
/* Transition guard (Req 5.2): START sets the non-terminal JOB_STAT_RUN. A
* START on an already-terminal job (stale/duplicate worker report after
* the job finished) is a backward transition — reject it and preserve
* the terminal state. */
if (job_state_is_backward(cur, JOB_STAT_RUN)) {
WARN("job %ld: rejecting backward START on terminal state 0x%x — "
"preserving terminal state",
job->job.jobid, cur);
return JOB_TRANSITION_REJECTED_BACKWARD;
}
/* update job information */
job->job.state = JOB_STAT_RUN;
job->job.start_time = time(NULL);
if (job->job.pidinfo == NULL) {
job->job.pidinfo = (job_pidinfo_t*)calloc(1, sizeof(job_pidinfo_t));
}
job->job.pidinfo->npids = copy_int_list((int**)&(job->job.pidinfo->pids),
(int*)info->pids, info->npids);
job->job.pidinfo->npgids = copy_int_list((int**)&(job->job.pidinfo->pgids),
(int*)info->pgids, info->npgids);
if (info->exdata != NULL) {
job->job.pidinfo->exdata = makeString(info->exdata);
} else {
job->job.pidinfo->exdata = NULL;
}
return JOB_TRANSITION_APPLIED;
}
/* raft-apply-decoupling task 4.1 (Req 5.1): renamed from mark_job_start —
* this is now durable-append-only. The pre-mutation BEFORE_EVENT validation
* hook still fires at the identical point, unchanged; the state mutation
* (apply_job_start_transition) and the AFTER_EVENT notification hook both
* relocate to run from apply_committed_entry() (task 2.2) once this fact's
* WAL entry commits to a majority — driven by the future apply pump
* (task 6). Until then, a worker-reported START has no live effect on job
* state; only the durable append happens here. */
static int record_job_start_fact(cluster_t* c, job_t* job, job_pidinfo_t* info)
{
/* Peek the identical guard apply_job_start_transition() re-checks
* internally, so this extraction changes NO observable behavior: today
* no hook fires on a rejected transition, and BEFORE_EVENT strictly
* precedes the (now-deferred) mutation. The peek is a pure read of
* job->job.state, and every caller holds `job` under RTTI_LOCK_WRITE, so
* it cannot race with the core's own (redundant, defensive) re-check of
* the same guard. Calling apply_job_start_transition here on the real
* `job` is safe and mutates nothing — its own internal guard re-check
* hits the identical condition and returns before any assignment; the
* call exists only to produce its WARN log line (logs + no-op,
* unchanged from before this task). */
if (job_state_is_backward(job->job.state, JOB_STAT_RUN)) {
apply_job_start_transition(job, info); /* logs + no-op, unchanged */
return 0;
}
xce_run_hooks(&(c->hooksys), XCE_CONTROLLER_HOOK_JOB, HOOK_BEFORE_EVENT,
HOOK_DETAIL_JOB_START, c, info, job);
/* Durable append only — no mutation, no AFTER_EVENT firing here (Req
* 5.1, 5.3). JOB_STAT_RUN is the resultant state a START always
* produces once past the guard above; apply_committed_entry() will
* re-validate the same guard against whatever state is current AT
* APPLY TIME before setting it. */
(void)persist_submit_job_fact(&c->persist, WAL_JOB_START, job->job.jobid,
JOB_STAT_RUN);
return 0;
}
int on_job_start(cluster_t* c, job_pidinfo_t* info)
{
job_t* job = NULL;
int rc;
TRACE("%s jobid %ld", __func__, info->jobid);
lock(c->jobmgr, RTTI_LOCK_READ);
rc = key_find(c->jobmgr, (void*)info->jobid, (void**)&job);
if (job != NULL) {
reference(job);
}
unlock(c->jobmgr);
if ((rc <= 0) || (job == NULL)) {
ERROR("can not find job %ld", info->jobid);
return -errno;
}
lock(job, RTTI_LOCK_WRITE);
record_job_start_fact(c, job, info);
unlock(job);
dereference(job);
return 0;
}
/* Guarded state-transition core for a worker-reported FINISH, extracted from
* mark_job_finish (raft-apply-decoupling task 2.1, Req 3.3/7.1/7.2; mark_job_finish
* itself was renamed record_job_finish_fact by task 4.1, which stopped calling
* this core from the receipt path — see that function's own doc comment).
* Contains ONLY the monotonic-rank guard and the state mutation — fires NO
* hook, so hook-firing can be relocated later without touching this guard
* logic again. Safe to invoke directly, bypassing the worker-report receipt
* path (on_job_finish) entirely, as long as the caller already holds `job` under
* RTTI_LOCK_WRITE. */
job_transition_result_t apply_job_finish_transition(job_t* job,
xce_jobfinish_t* finish)
{
int cur = job->job.state;
/* Transition guard (Req 5.1): FINISH drives the job to a terminal state
* (DONE on exit_code 0, else EXIT). If the job is already terminal the
* FINISH has already been applied (retransmit / WAL replay) — treat it
* as a no-op duplicate so no second WAL entry is appended. */
if (job_state_is_terminal(cur)) {
INFO("job %ld: duplicate FINISH on terminal state 0x%x — no-op",
job->job.jobid, cur);
return JOB_TRANSITION_NOOP_DUPLICATE;
}
/* update job information */
if (finish->exit_code == 0) {
job->job.state = JOB_STAT_DONE;
} else {
job->job.state = JOB_STAT_EXIT;
}
job->job.finish_time = time(NULL);
/* fill job run time/rusage information */
if (job->job.exit_info == NULL) {
job->job.exit_info = (job_exitinfo_t*)calloc(1, sizeof(job_exitinfo_t));
}
if (job->job.exit_info != NULL) {
job->job.exit_info->exitcode = finish->exit_code;
}
job->job.run_time = job->job.finish_time - job->job.start_time;
return JOB_TRANSITION_APPLIED;
}
/* raft-apply-decoupling task 4.1 (Req 5.1): renamed from mark_job_finish —
* durable-append-only, mirroring record_job_start_fact above. */
static int record_job_finish_fact(cluster_t* c, job_t* job,
xce_jobfinish_t* finish)
{
/* Peek the identical guard apply_job_finish_transition() re-checks
* internally, so this extraction changes NO observable behavior: today
* no hook fires on a duplicate-terminal FINISH, and BEFORE_EVENT strictly
* precedes the (now-deferred) mutation. The peek is a pure read of
* job->job.state, and every caller holds `job` under RTTI_LOCK_WRITE, so
* it cannot race with the core's own (redundant, defensive) re-check of
* the same guard. */
if (job_state_is_terminal(job->job.state)) {
apply_job_finish_transition(job, finish); /* logs + no-op, unchanged */
return 0;
}
xce_run_hooks(&(c->hooksys), XCE_CONTROLLER_HOOK_JOB, HOOK_BEFORE_EVENT,
HOOK_DETAIL_JOB_EXIT, c, finish, job);
/* Durable append only — no mutation, no AFTER_EVENT firing here (Req
* 5.1, 5.3). The resultant state is deterministic from finish->exit_code
* alone, independent of current job state. */
(void)persist_submit_job_fact(&c->persist, WAL_JOB_FINISH, job->job.jobid,
(finish->exit_code == 0) ? JOB_STAT_DONE
: JOB_STAT_EXIT);
return 0;
}
int on_job_finish(cluster_t* c, xce_jobfinish_t* finish)
{
job_t* job = NULL;
TRACE("%s job finish jobid %ld exit code %d", __func__, finish->jobid,
finish->exit_code);
lock(c->jobmgr, RTTI_LOCK_READ);
key_find(c->jobmgr, (void*)finish->jobid, (void**)&job);
if (job != NULL) {
reference(job);
}
unlock(c->jobmgr);
if (job == NULL) {
ERROR("can not find job %ld", finish->jobid);
return -errno;
}
lock(job, RTTI_LOCK_WRITE);
record_job_finish_fact(c, job, finish);
unlock(job);
dereference(job);
return 0;
}
/* State-transition core for a worker-reported FAIL, extracted from
* mark_job_fail (raft-apply-decoupling task 2.1, Req 3.3/7.1/7.2). Contains
* ONLY the state mutation — fires NO hook, so hook-firing can be relocated
* later without touching this logic again. mark_job_fail has no monotonic-
* rank guard today (unlike START/FINISH/UPDATE): a FAIL always applies,
* merging JOB_STAT_PERR onto an existing DONE/EXIT terminal state rather
* than being rejected — that existing merge behavior is preserved unchanged
* by this extraction, not newly added. Safe to invoke directly, bypassing
* the worker-report receipt path (on_job_fail) entirely, as long as the
* caller already holds `job` under RTTI_LOCK_WRITE. */
job_transition_result_t apply_job_fail_transition(job_t* job,
job_failinfo_t* failinfo)
{
switch (failinfo->stage) {
case JOB_EXEC_STAGE_Q_POST_EXEC:
case JOB_EXEC_STAGE_POST_EXEC:
if ((job->job.state & (JOB_STAT_EXIT | JOB_STAT_DONE)) == 0) {
WARN("job %ld got post exec fail notification but job's exit "
"status has not been set",
job->job.jobid);
job->job.state |= JOB_STAT_EXIT;
}
job->job.state |= JOB_STAT_PERR;
break;
default:
job->job.state = JOB_STAT_EXIT;
break;
}
job->job.finish_time = time(NULL);
return JOB_TRANSITION_APPLIED;
}
/* Used only by job_ctrl's SIGKILL/SIGTERM not-started-job branch (a client
* command path, do_jobcontrol_op — out of raft-apply-decoupling task 4.1's
* boundary) — mutate-then-notify, unchanged. record_job_fail_fact below is
* the separate, durable-append-only sibling used by the worker-fact receipt
* path (on_job_fail); the two must stay independent since job_ctrl's
* synchronous command semantics are not this task's concern. */
static int mark_job_fail(cluster_t* c, job_t* job, job_failinfo_t* failinfo)
{
xce_run_hooks(&(c->hooksys), XCE_CONTROLLER_HOOK_JOB, HOOK_BEFORE_EVENT,
HOOK_DETAIL_JOB_FAIL, c, failinfo, job);
apply_job_fail_transition(job, failinfo);
xce_run_hooks(&(c->hooksys), XCE_CONTROLLER_HOOK_JOB, HOOK_AFTER_EVENT,
HOOK_DETAIL_JOB_FAIL, c, failinfo, job);
return 0;
}
/* raft-apply-decoupling task 4.1 (Req 5.1): the worker-fact receipt sibling
* of mark_job_fail above — durable-append-only, used only by on_job_fail.
* mark_job_fail itself is untouched (job_ctrl still needs its mutate-then-
* notify semantics; that command path is out of this task's boundary).
*
* mark_job_fail has no monotonic-rank guard (a FAIL always applies), so
* unlike record_job_start_fact/record_job_finish_fact there is no
* peek/rejected branch here — BEFORE_EVENT fires unconditionally, exactly as
* mark_job_fail's always does today.
*
* task 6.1 round 2 fix (merge-staleness corruption, Req 11.3): the resultant
* state depends on the CURRENT job state (PERR merges onto an existing
* DONE/EXIT terminal state rather than resetting it — apply_job_fail_
* transition's switch below). Earlier this was precomputed HERE, at receipt
* time, on a throwaway shallow copy of `job`, and the precomputed absolute
* result was baked into the WAL entry. That is unsafe under deferred apply:
* another fact for the same job (e.g. a FINISH) can still be pending apply
* when this FAIL is received, so `job->job.state` at receipt time may not
* be the state this FAIL's merge should actually be computed against —
* baking a stale snapshot silently clobbers whatever the other fact applies
* first (bug found in task 6.1 round 1 review). Fixed by carrying ONLY
* `failinfo->stage` (via _pad's WAL_JOB_FAIL_STAGE_FLAG — see wal_types.h's
* doc comment on wal_jobstate_t._pad) and deferring the actual merge
* computation to apply time, against the job's real current state then
* (apply_committed_entry's WAL_JOB_FAIL case, persist.c). */
static int record_job_fail_fact(cluster_t* c, job_t* job,
job_failinfo_t* failinfo)
{
xce_run_hooks(&(c->hooksys), XCE_CONTROLLER_HOOK_JOB, HOOK_BEFORE_EVENT,
HOOK_DETAIL_JOB_FAIL, c, failinfo, job);
/* Durable append only — no mutation of the real `job`, no AFTER_EVENT
* firing here (Req 5.1, 5.3). `state` is unused for this apply-time-
* recomputed shape (0 is written for forward-compatibility/diagnostics
* only); the WAL_JOB_FAIL_STAGE_FLAG bit in `_pad` tells
* apply_committed_entry() to route it to the real
* apply_job_fail_transition() core instead of an absolute overwrite. */
(void)persist_submit_job_fact_ex(&c->persist, WAL_JOB_FAIL,
job->job.jobid, 0,
WAL_JOB_FAIL_STAGE_FLAG |
failinfo->stage);
return 0;
}
/* Guarded state-transition core for an UPDATE, extracted from
* mark_job_update (raft-apply-decoupling task 2.1, Req 3.3/7.1/7.2). Contains
* ONLY the monotonic-rank guards and the state mutation — fires NO hook, so
* hook-firing can be relocated later without touching this guard logic
* again. Safe to invoke directly, bypassing the mark_job_update call sites
* entirely, as long as the caller already holds `job` under
* RTTI_LOCK_WRITE. */
job_transition_result_t apply_job_update_transition(job_t* job, int state)
{
int cur = job->job.state;
/* Transition guard (Req 5.1/5.2), sharing the backward predicate with
* START: an UPDATE that clears every terminal bit a terminal job holds is
* backward — reject it and preserve the terminal state. A forward merge
* that keeps a terminal bit (postdone OR'ing PDONE) is not backward and
* proceeds. A no-op update (requested state already current) is a
* duplicate — skip it so it appends no duplicate WAL entry. */
if (job_state_is_backward(cur, state)) {
WARN("job %ld: rejecting backward UPDATE 0x%x -> 0x%x — preserving "
"terminal state",
job->job.jobid, cur, state);
return JOB_TRANSITION_REJECTED_BACKWARD;
}
if (state == cur) {
INFO("job %ld: duplicate UPDATE to state 0x%x — no-op", job->job.jobid,
cur);
return JOB_TRANSITION_NOOP_DUPLICATE;
}
/* update job state */
job->job.state = state;
return JOB_TRANSITION_APPLIED;
}
/* Used only by job_ctrl's SIGCONT/SIGSTOP suspend-resume branches (a client
* command path, do_jobcontrol_op — out of raft-apply-decoupling task 4.1's
* boundary) — mutate-then-notify, unchanged. record_job_postdone_fact below
* is the separate, durable-append-only sibling used by the worker-fact
* receipt path (on_job_postdone); the two must stay independent since
* job_ctrl's synchronous command semantics are not this task's concern. */
static int mark_job_update(cluster_t* c, job_t* job, int state)
{
int cur = job->job.state;
/* Peek the identical guards apply_job_update_transition() re-checks
* internally, so this extraction changes NO observable behavior: today
* no hook fires on a rejected/duplicate transition, and BEFORE_EVENT
* strictly precedes the mutation. The peek is a pure read of
* job->job.state, and every caller holds `job` under RTTI_LOCK_WRITE, so
* it cannot race with the core's own (redundant, defensive) re-check of
* the same guards. */
if (job_state_is_backward(cur, state) || (state == cur)) {
apply_job_update_transition(job, state); /* logs + no-op, unchanged */
return 0;
}
xce_run_hooks(&(c->hooksys), XCE_CONTROLLER_HOOK_JOB, HOOK_BEFORE_EVENT,
HOOK_DETAIL_JOB_UPDATE, c, job);
apply_job_update_transition(job, state);
xce_run_hooks(&(c->hooksys), XCE_CONTROLLER_HOOK_JOB, HOOK_AFTER_EVENT,
HOOK_DETAIL_JOB_UPDATE, c, job);
return 0;
}
/* Guarded state-transition core for a postdone fact, extracted from
* on_job_postdone's old inline mutation (raft-apply-decoupling task 6.1
* round 2, Req 3.3/7.1-7.3). Unlike a generic UPDATE, postdone's formula
* (conditionally backstop EXIT, then OR in JOB_STAT_PDONE) is a pure
* function of the job's CURRENT state — so this core is meant to be invoked
* AT APPLY TIME, against whatever state actually holds then, never against
* a receipt-time snapshot (a receipt-time snapshot is exactly the merge-
* staleness hazard this round fixes: another fact for the same job, e.g. a
* FINISH, may still be in flight and apply AFTER receipt but BEFORE this
* fact's own apply, under deferred apply). No-op when PDONE is already set
* and no backstop is needed (idempotent re-apply, Req 3.4). Fires NO hook —
* same convention as the other apply_job_*_transition cores; hook-firing
* stays the caller's (apply_committed_entry's) responsibility. Caller must
* hold `job` under RTTI_LOCK_WRITE.
*
* Known, previously-accepted residual (task 4.1, unchanged by this round):
* finish_time is not part of the wal_jobstate_t wire format, so the old
* defensive `job->job.finish_time = time(NULL)` backstop write has no
* apply-time equivalent — a postdone arriving with no prior exit status
* recorded still gets no defensive finish_time set anywhere. */
job_transition_result_t apply_job_postdone_transition(job_t* job)
{
int next = job->job.state;
if ((next & (JOB_STAT_EXIT | JOB_STAT_DONE)) == 0) {
WARN("job %ld got post exec fail notification but job's exit "
"status has not been set",
job->job.jobid);
next |= JOB_STAT_EXIT;
}
next |= JOB_STAT_PDONE;
if (next == job->job.state) {
INFO("job %ld: duplicate postdone — no-op", job->job.jobid);
return JOB_TRANSITION_NOOP_DUPLICATE;
}
job->job.state = next;
return JOB_TRANSITION_APPLIED;
}
/* raft-apply-decoupling task 4.1/6.1 (Req 5.1, 3.3, 7.1-7.3): durable-
* append-only sibling of the old inline postdone mutation, used only by
* on_job_postdone. Replaces record_job_update_fact (removed in task 6.1
* round 2 — this was its only caller): postdone's mutation is a pure
* function of current state (see apply_job_postdone_transition above), so
* nothing needs to be computed or peeked at receipt time — the WAL entry
* carries no state value at all, just a marker (_pad ==
* WAL_JOB_UPDATE_POSTDONE_MARK) telling apply_committed_entry() to re-
* derive the merge fresh, against whatever the job's real current state is
* when this entry actually applies (fixes the same class of merge-
* staleness corruption record_job_fail_fact's fix addresses above). */
static int record_job_postdone_fact(cluster_t* c, job_t* job)
{
xce_run_hooks(&(c->hooksys), XCE_CONTROLLER_HOOK_JOB, HOOK_BEFORE_EVENT,
HOOK_DETAIL_JOB_UPDATE, c, job);
/* Durable append only — no mutation, no AFTER_EVENT firing here (Req
* 5.1, 5.3). */
(void)persist_submit_job_fact_ex(&c->persist, WAL_JOB_UPDATE,
job->job.jobid, 0,
WAL_JOB_UPDATE_POSTDONE_MARK);
return 0;
}
int on_job_fail(cluster_t* c, job_failinfo_t* failinfo)
{
job_t* job = NULL;
TRACE("%s job fail jobid %ld code %d", __func__, failinfo->jobid,
failinfo->code);
lock(c->jobmgr, RTTI_LOCK_READ);
key_find(c->jobmgr, (void*)failinfo->jobid, (void**)&job);
if (job != NULL) {
reference(job);
}
unlock(c->jobmgr);
if (job == NULL) {
ERROR("can not find job %ld", failinfo->jobid);
return -errno;
}
lock(job, RTTI_LOCK_WRITE);
record_job_fail_fact(c, job, failinfo);
unlock(job);
dereference(job);
return 0;
}
int on_job_stop(cluster_t* c, uint64_t jobid)
{
job_t* job = NULL;
int state;
TRACE("%s job stop jobid %ld", __func__, jobid);
lock(c->jobmgr, RTTI_LOCK_READ);
key_find(c->jobmgr, (void*)jobid, (void**)&job);
if (job != NULL) {
reference(job);
}
unlock(c->jobmgr);
if (job == NULL) {
ERROR("can not find job %ld", jobid);
return -errno;
}
lock(job, RTTI_LOCK_WRITE);
state = job->job.state;
xce_run_hooks(&(c->hooksys), XCE_CONTROLLER_HOOK_JOB, HOOK_BEFORE_EVENT,
HOOK_DETAIL_JOB_STOP, c, job);
state |= JOB_STAT_USUSP;
job->job.state = state;
xce_run_hooks(&(c->hooksys), XCE_CONTROLLER_HOOK_JOB, HOOK_AFTER_EVENT,
HOOK_DETAIL_JOB_STOP, c, job);
unlock(job);
dereference(job);
return 0;
}
int on_job_resume(cluster_t* c, uint64_t jobid)
{
job_t* job = NULL;
int state;
TRACE("%s job resume jobid %ld", __func__, jobid);
lock(c->jobmgr, RTTI_LOCK_READ);
key_find(c->jobmgr, (void*)jobid, (void**)&job);
if (job != NULL) {
reference(job);
}
unlock(c->jobmgr);
if (job == NULL) {
ERROR("can not find job %ld", jobid);
return -errno;
}
lock(job, RTTI_LOCK_WRITE);
state = job->job.state;
xce_run_hooks(&(c->hooksys), XCE_CONTROLLER_HOOK_JOB, HOOK_BEFORE_EVENT,
HOOK_DETAIL_JOB_RESUME, c, job);
state &= ~JOB_STAT_USUSP;
job->job.state = state;
xce_run_hooks(&(c->hooksys), XCE_CONTROLLER_HOOK_JOB, HOOK_AFTER_EVENT,
HOOK_DETAIL_JOB_RESUME, c, job);
unlock(job);
dereference(job);
return 0;
}
int on_job_postdone(cluster_t* c, endpoint_t* ep, uint64_t jobid)
{
job_t* job = NULL;
TRACE("%s job post done jobid %ld", __func__, jobid);
lock(c->jobmgr, RTTI_LOCK_READ);
key_find(c->jobmgr, (void*)jobid, (void**)&job);
if (job != NULL) {
reference(job);
}
unlock(c->jobmgr);
if (job == NULL) {
ERROR("can not find job %ld", jobid);
return -errno;
}
/* raft-apply-decoupling task 6.1 round 2 (Req 5.1, 11.3): the backstop-
* EXIT-if-unset + OR-in-PDONE mutation formerly computed HERE, at
* receipt time, is now deferred entirely to apply_job_postdone_
* transition() (called from apply_committed_entry() once this fact
* commits) — see that function's doc comment for why baking it here
* against a receipt-time snapshot was unsafe under deferred apply. */
lock(job, RTTI_LOCK_WRITE);
record_job_postdone_fact(c, job);
unlock(job);
dereference(job);
return 0;
}
int on_job_ckpt(cluster_t* c, endpoint_t* ep, xce_ckptinfo_t* ckptinfo)
{
uint64_t jobid;
char* info = NULL;
job_t* j = NULL;
if (ckptinfo == NULL) {
ERROR("%s ckptinfo should not be NULL", __func__);
return -EINVAL;
}
jobid = ckptinfo->jobid;
info = ckptinfo->ckptinfo;
TRACE("%s job %ld ckptinfo %s", __func__, jobid, info);
lock(c->jobmgr, RTTI_LOCK_READ);
key_find(c->jobmgr, (void*)jobid, (void**)&j);
if (j != NULL) {
reference(j);
}
unlock(c->jobmgr);
if (j == NULL) {
ERROR("%s can not find job %ld", __func__, jobid);
return -errno;
}
lock(j, RTTI_LOCK_WRITE);
xce_run_hooks(&(c->hooksys), XCE_CONTROLLER_HOOK_JOB, HOOK_BEFORE_EVENT,
HOOK_DETAIL_JOB_CKPT, c, j);
if (j->job.ckptinfo != NULL) {
dereference(j->job.ckptinfo);
}
j->job.ckptinfo = makeString(info);
xce_run_hooks(&(c->hooksys), XCE_CONTROLLER_HOOK_JOB, HOOK_AFTER_EVENT,
HOOK_DETAIL_JOB_CKPT, c, j);
unlock(j);
dereference(j);
return 0;
}
int do_jobcontrol_op(cluster_t* c, endpoint_t* ep, xce_request_t* request)
{
int rc = 0;
xce_reply_t reply;
init_reply_by_req(&reply, request);
rc = job_ctrl(c, request->body.jobctrlmsg);
/* Commit barrier (ha-commit-barrier-rollout Req 1.1-1.4): the success
* reply is sent only after the operation's WAL entry is majority-
* committed; barrier failures (-ETIMEDOUT, ...) become the error
* reply. Errors from job_ctrl reply immediately — no barrier wait.
* job_ctrl released all entity locks before returning, so blocking
* here holds no lock. timeout 0 = configured sync_timeout_ms. */
if (rc == 0) {
rc = persist_commit_barrier(&c->persist, 0);
if (rc != 0)
WARN("%s commit barrier failed, rc %d", __func__, rc);
}
reply.status = rc;
reply.data_which = RequestReply_data_nodata;
xce_reply(ep, c->myhost, &reply);
return rc;
}
int update_job_rusages(cluster_t* c, endpoint_t* ep, xce_request_t* request)
{
xce_job_rusages_t* rusages = request->body.updatemsg->data.jobrusages;
job_t* j = NULL;
worker_t* w = NULL;
char* host = NULL;
TRACE("%s trying to update job(%ld) rusage", __func__, rusages->jobid);
xce_run_hooks(&(c->hooksys), XCE_CONTROLLER_HOOK_JOB, HOOK_BEFORE_EVENT,
HOOK_DETAIL_JOB_RUSAGE_UPDATE, c, rusages);
lock(c->jobmgr, RTTI_LOCK_READ);
key_find(c->jobmgr, (void*)rusages->jobid, (void**)&j);
if (j != NULL) {
reference(j);
}
unlock(c->jobmgr);
if (j == NULL) {
PERROR("jobmgr_lockjob");
return -errno;
}
w = get_ep_worker(c, ep);
if (w == NULL) {
dereference(j);
return errno;
}
host = makeString(w->host.name);
if (host == NULL) {
dereference(j);
return -ENOMEM;
}
TRACE("%s update job (%ld) rusage for host %s", __func__, rusages->jobid,
host);
lock(j, RTTI_LOCK_WRITE);
xce_run_hooks(&(c->hooksys), XCE_CONTROLLER_HOOK_JOB, HOOK_BEFORE_EVENT,
HOOK_DETAIL_JOB_UPDATE, c, j);
update_host_runusage(&(j->job.run_usage), host, rusages->count,
rusages->usages);
update_pids(&(j->job.pidinfo), rusages->npids, rusages->pids);
update_pgids(&(j->job.pidinfo), rusages->npgids, rusages->pgids);
/* FIXME
* update other fields of job rusage later
*/
xce_run_hooks(&(c->hooksys), XCE_CONTROLLER_HOOK_JOB, HOOK_AFTER_EVENT,
HOOK_DETAIL_JOB_UPDATE, c, j);
xce_run_hooks(&(c->hooksys), XCE_CONTROLLER_HOOK_JOB, HOOK_AFTER_EVENT,
HOOK_DETAIL_JOB_RUSAGE_UPDATE, c, rusages);
unlock(j);
dereference(j);
return 0;
}
static int set_preason(cluster_t* c, xce_pendreason_t* reasons)
{
job_t* j = NULL;
xce_job_t* job = NULL;
TRACE("%s jobid %ld", __func__, reasons->jobid);
lock(c->jobmgr, RTTI_LOCK_READ);
key_find(c->jobmgr, (void*)reasons->jobid, (void**)&j);
if (j != NULL) {
reference(j);
}
unlock(c->jobmgr);
if (j == NULL) {
PERROR("jobmgr_lockjob");
return -ENOENT;
}
lock(j, RTTI_LOCK_WRITE);
xce_run_hooks(&(c->hooksys), XCE_CONTROLLER_HOOK_JOB, HOOK_BEFORE_EVENT,
HOOK_DETAIL_SETPREASON, c, j);
job = &(j->job);
if (job->pend_reasons != NULL) {
erase_job_preason(job->pend_reasons);
FREEUP(job->pend_reasons);
}
MOVE_PTR(job->pend_reasons, reasons->reasons);
xce_run_hooks(&(c->hooksys), XCE_CONTROLLER_HOOK_JOB, HOOK_AFTER_EVENT,
HOOK_DETAIL_SETPREASON, c, j);
unlock(j);
dereference(j);
return 0;
}
int do_set_preason(cluster_t* c, endpoint_t* ep, xce_request_t* request)
{
int i;
xce_preason_set_t* reasons = request->body.preasonmsg;
xce_reply_t reply;
init_reply_by_req(&reply, request);
reply.status = 0;
reply.data_which = RequestReply_data_nodata;
xce_reply(ep, c->myhost, &reply);
for (i = 0; i < reasons->nreasons; i++) {
int rc = 0;
if (reasons->reasons[i] == NULL) {
continue;
}
rc = set_preason(c, reasons->reasons[i]);
if (rc < 0) {
PERROR("set_preason");
}
}
return 0;
}
int do_clr_preason(cluster_t* c, endpoint_t* ep, xce_request_t* request)
{
int i;
xce_reply_t reply;
xce_clear_reason_t* clreason = request->body.clrpreasonmsg;
init_reply_by_req(&reply, request);
reply.status = 0;
reply.data_which = RequestReply_data_nodata;
xce_reply(ep, c->myhost, &reply);
for (i = 0; i < clreason->njobs; i++) {
uint64_t jobid = clreason->jobs[i];
job_t* j;
xce_job_t* job;
lock(c->jobmgr, RTTI_LOCK_READ);
key_find(c->jobmgr, (void*)jobid, (void**)&j);
if (j != NULL) {
reference(j);
}
unlock(c->jobmgr);
if (j == NULL) {
PERROR("jobmgr_lockjob");
continue;
}
lock(j, RTTI_LOCK_WRITE);
xce_run_hooks(&(c->hooksys), XCE_CONTROLLER_HOOK_JOB, HOOK_BEFORE_EVENT,
HOOK_DETAIL_CLRPREASON, c, j);
job = &(j->job);
if (job->pend_reasons != NULL) {
erase_job_preason(job->pend_reasons);
FREEUP(job->pend_reasons);
}
xce_run_hooks(&(c->hooksys), XCE_CONTROLLER_HOOK_JOB, HOOK_AFTER_EVENT,
HOOK_DETAIL_CLRPREASON, c, j);
unlock(j);
dereference(j);
}
return 0;
}
/* job_is_terminal and job_should_retain live in persist.c —
* they only depend on xce_job_t fields and persist_config_t,
* and must be accessible from the unit test binary which links persist.c. */
View git blame Copy permalink