#include #include #include #include #include #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 mark_job_start(cluster_t*, job_t*, job_pidinfo_t*); static int mark_job_finish(cluster_t*, job_t*, xce_jobfinish_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; } /* 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". */ 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 */ 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) { 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); } /* raft-apply-decoupling task 6 hand-off target (task 6 itself is NOT built * by this task): invoked by the future ordered apply pump 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". Until task 6 exists, nothing ever calls this * function; see jobsubmit_commit_continuation's COMMITTED branch below, * which registers it as the hand-off target instead of applying or * replying itself. * * NOTE for whoever builds task 6.2: 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 * — so by the time this function is invoked, the job is very likely already * visible and already notified. This function most likely only needs to * send the batch's success reply (echoing the jobid) once every entry has * been applied, NOT reconstruct or re-notify 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); /* Task 6.2's success-reply belongs here (RequestReply_data_jobid, * gated on actx->replied exactly like jobsubmit_reply_failure_once, so * a partially-failed batch's earlier failure reply is never followed * by a contradictory success reply) — not built by this task, which * must not reply from the committed-outcome path at all (see the * struct doc comment above). */ if (done) 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; } /* 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); 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); 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->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). 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; } static int mark_job_start(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 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_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); apply_job_start_transition(job, info); xce_run_hooks(&(c->hooksys), XCE_CONTROLLER_HOOK_JOB, HOOK_AFTER_EVENT, HOOK_DETAIL_JOB_START, c, info, job); 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); mark_job_start(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). 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; } static int mark_job_finish(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 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); apply_job_finish_transition(job, finish); xce_run_hooks(&(c->hooksys), XCE_CONTROLLER_HOOK_JOB, HOOK_AFTER_EVENT, HOOK_DETAIL_JOB_EXIT, c, finish, job); 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); mark_job_finish(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; } 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; } /* 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; } 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; } 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); mark_job_fail(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; int state; 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; } lock(job, RTTI_LOCK_WRITE); state = job->job.state; if ((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); state |= JOB_STAT_EXIT; /* ensure that job finish time has been set */ job->job.finish_time = time(NULL); } state |= JOB_STAT_PDONE; mark_job_update(c, job, state); 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. */