#if !defined(_CLUSTER_H_) #define _CLUSTER_H_ 1 #include "config.h" #include #include #include #include "hook.h" #include "htab.h" #include "kvmgr.h" #include "myconf.h" #include "mylock.h" #include "netio.h" #include "typemap.h" /* xce_job_state_entry_t -- worker-rejoin-resync task 2.2 */ #include "vtaskmgr.h" #include "worker.h" #include "xce/cluster_def.h" #include "xce/types.h" #include "xlist.h" #include "zarray.h" typedef int (*collect_check_fn_t)(void*); extern int check_controller(void*); extern int cluster_init_phase1(cluster_t*); extern int cluster_open_listener(cluster_t*); extern int cluster_register_listener(cluster_t*); extern int cluster_init_phase2(cluster_t*); extern int cluster_init(cluster_t*); extern int cluster_start(cluster_t*); extern int cluster_run(cluster_t*); extern void cluster_fini(cluster_t*); extern int register_types(cluster_t*); extern int worker_ctrl(cluster_t*, xce_hostctrl_t*); extern xce_resource_t* get_resource(cluster_t*, char*); extern int add_to_xlist(xlist_t*, void*); extern int add_worker_to_xlist(xlist_t*, void*); extern int add_queue_to_xlist(xlist_t*, void*); extern int add_job_to_xlist(xlist_t*, void*); extern void resource_killer(void*); extern int add_resource(void*, char*, char*, int, xce_order_t, int, int); extern int update_resvals(cluster_t*, worker_t*, int, xce_resval_t*); extern int load_jobs(void*); extern int restore_job(cluster_t*, uint64_t, const void*, size_t, uint64_t); /* job-array-single-wal-record task 3.1 (design.md Array Apply Expansion): * shared N-way expansion primitive for a committed WAL_JOB_ARRAY_SUBMIT * payload -- see the doc comment on the definition (job.c) for the full * atomicity/idempotency contract. out_new_jobs is optional; when non-NULL * and the call genuinely creates new elements, it is populated (ownership * transferred) with every newly created job_t* in index order. */ extern int restore_job_array(cluster_t*, const void*, size_t, uint64_t, xlist_t*); #ifdef XCE_TESTING /* job-array-single-wal-record task 3.1: forces restore_job_array()'s * scratch-build loop to fail construction at the given 0-based element * index (simulating resource exhaustion), so a test can prove the * all-or-nothing atomicity contract. Pass a negative index, or call the * clear helper, to disable. Compiled out of production. */ extern void job_array_apply_test_set_force_fail_at(int index); extern void job_array_apply_test_clear_force_fail(void); /* job-array-single-wal-record task 3.1 remediation: forces * restore_job_array()'s BULK-INSERT pass (the key_add(c->jobmgr, ...) loop, * distinct from the scratch-build loop above) to fail at the given 0-based * index WITHIN THAT PASS, simulating c->jobmgr's backing store hitting a * genuine realloc() OOM while growing under the write lock. Lets a test * prove this failure mode is reported as a negative errno, never a false * built.count success. Pass a negative index, or call the clear helper, to * disable. Compiled out of production. */ extern void job_array_apply_test_set_bulk_insert_force_fail_at(int index); extern void job_array_apply_test_clear_bulk_insert_force_fail(void); #endif extern int update_job(void*, uint64_t, xce_kv_list_t*); extern int do_job_started(void*, uint64_t, int, int*, int, int*); extern queue_t* create_queue(cluster_t*, xce_queue_t*); extern int build_queues(void*); extern void queue_fini(queue_t*); extern int queue_ctrl(cluster_t*, xce_queuectrl_t*); extern int xce_update_data(endpoint_t*, cluster_t*, int, int, int*, int, xce_kv_t*); extern int on_scheduler_join(cluster_t*); extern int on_apiserver_join(cluster_t*, endpoint_t*); extern int xce_push_reslist(endpoint_t*, cluster_t*, int, xce_resource_t**); extern int xce_push_confvars(endpoint_t*, cluster_t*, int, xce_config_t**); extern int xce_push_workerlist(endpoint_t*, cluster_t*, int, xce_worker_t**); extern int xce_push_joblist(endpoint_t*, cluster_t*, int, xce_job_t**); extern int xce_push_queuelist(endpoint_t*, cluster_t*, int, xce_queue_t**); extern int xcesched_push_job(cluster_t*, uint64_t); extern int on_worker_join(void*, endpoint_t*); extern int get_resval(cluster_t*, char*, char*, xce_val_t*); extern int handle_push(struct capn*, cluster_t*, endpoint_t*, struct Request*); extern int do_get_op(cluster_t*, endpoint_t*, xce_request_t*); extern int do_update_op(cluster_t*, endpoint_t*, xce_request_t*); extern int handle_join(cluster_t*, endpoint_t*, xce_request_t*); extern int handle_reply(endpoint_t*, struct Request*, cluster_t*); extern int handle_jobdispatch(cluster_t*, endpoint_t*, xce_request_t*); extern int handle_jobsubmit(cluster_t*, endpoint_t*, xce_request_t*); #ifdef XCE_TESTING /* bug-1419: clears job.c's process-lifetime in-flight-jobid registry. * Tests that deliberately leave a submission's commit continuation * unresolved through teardown (no ack, no timeout, no truncation) have no * other event to release that registry entry -- call this from fixture * teardown so it doesn't outlive the test and spuriously reject an * unrelated later test reusing the same jobid range. */ extern void inflight_submits_reset_for_test(void); #endif extern int handle_worker_leave(cluster_t*, int); extern int mark_worker_unavail(cluster_t*, worker_t*); extern int mark_worker_back(cluster_t*, worker_t*); extern int handle_scheduler_leave(cluster_t*); extern int handle_clusterconf(cluster_t*, endpoint_t*, xce_request_t*); /* controller-status spec (task 2.1): read-only status query. * ctrlstatus_fill snapshots local HA/config state into *st (no I/O, no state * mutation); ctrlstatus_free_fields releases what fill allocated (the peer * pointer array). handle_ctrlstatus is the dispatch entry point that fills, * replies via xce_reply, then frees. */ extern void ctrlstatus_fill(cluster_t*, xce_ctrlstatus_t*); extern void ctrlstatus_free_fields(xce_ctrlstatus_t*); extern int handle_ctrlstatus(cluster_t*, endpoint_t*, xce_request_t*); extern void collect_confvars(cluster_t*, collect_check_fn_t, xlist_t*); extern void collect_queues(cluster_t*, collect_check_fn_t, xlist_t*); extern void collect_resources(cluster_t*, collect_check_fn_t, xlist_t*); extern void collect_jobs(cluster_t*, collect_check_fn_t, xlist_t*); extern void collect_workers(cluster_t*, collect_check_fn_t, xlist_t*); /* ha-convergence-recovery (3.1-3.3): commit-watermark egress gate for * scheduler-bound job exports. Defined in dataex.c. * job_submit_committed — true iff the job's submit entry is committed * (submit_lsn == 0 || submit_lsn <= committed_lsn); * one lock-free atomic load + compare. * collect_jobs_committed — identical contract to collect_jobs, additionally * excluding jobs whose submit entry is uncommitted. * For scheduler-bound callers ONLY; status/API read * paths keep plain collect_jobs (Req 3.4). */ extern bool job_submit_committed(cluster_t*, job_t*); extern void collect_jobs_committed(cluster_t*, collect_check_fn_t, xlist_t*); extern int get_queue_exec(cluster_t*, char*, char**, char**); extern void unlock_collect_queues(xlist_t*); extern void unlock_collect_jobs(xlist_t*); extern void unlock_collect_workers(xlist_t*); /* Outcome of a guarded job state-transition core (raft-apply-decoupling * task 2.1, Req 3.3/7.1/7.2): APPLIED means the mutation happened; * REJECTED_BACKWARD/NOOP_DUPLICATE mean the monotonic-rank guard held the * transition back and no mutation occurred. These cores fire no hook * themselves — hook-firing is the caller's responsibility, so it can be * relocated (to the apply seam) without touching this guard logic again. */ typedef enum { JOB_TRANSITION_APPLIED = 0, JOB_TRANSITION_REJECTED_BACKWARD, JOB_TRANSITION_NOOP_DUPLICATE, } job_transition_result_t; /* Guarded state-transition cores extracted from mark_job_start/finish/fail/ * update (job.c). Each applies exactly one fact's mutation to an * already-locked job_t, subject to the same monotonic-rank guard the * receipt-time wrapper used to apply inline. Callable directly — e.g. by a * test, or later by the shared apply seam — without going through the * worker-report receipt path. */ extern job_transition_result_t apply_job_start_transition(job_t*, job_pidinfo_t*); extern job_transition_result_t apply_job_finish_transition(job_t*, xce_jobfinish_t*); extern job_transition_result_t apply_job_fail_transition(job_t*, job_failinfo_t*); extern job_transition_result_t apply_job_update_transition(job_t*, int); /* Guarded state-transition core for a postdone fact (raft-apply-decoupling * task 6.1 round 2, Req 3.3/7.1-7.3), extracted from on_job_postdone's old * inline mutation. Unlike a generic UPDATE, postdone's formula (conditional * EXIT backstop, then OR in JOB_STAT_PDONE) is a pure function of the job's * CURRENT state — so, like the other cores above, it is meant to be invoked * at apply time against whatever state actually holds then, never against a * receipt-time snapshot. No-op (JOB_TRANSITION_NOOP_DUPLICATE) when PDONE is * already set and no backstop is needed. Caller must hold `job` under * RTTI_LOCK_WRITE. */ extern job_transition_result_t apply_job_postdone_transition(job_t*); /* worker-rejoin-resync task 2.1 (Req 4.6, 5.4, 7.1-7.3): returns a job the * worker reports as unrecognized back to PEND, releasing any resource * allocation it was carrying (bug-1631) so a later redispatch never * inherits a stale allocation from the prior dispatch. Follows * dispatch_a_job_()'s own eager-mutation-plus-explicit-notify pattern, NOT * record_job_finish_fact's deferred one -- see design.md's "Controller: * job.c" Components section for the full rationale. * * Precondition: caller holds `job` under RTTI_LOCK_WRITE on entry and does * not release it until this call returns -- matching apply_job_update_ * transition's own calling convention (e.g. mark_job_update's callers). * This function does not lock/unlock `job` itself. * * Returns the job_transition_result_t of the underlying apply_job_update_ * transition(job, JOB_STAT_PEND) call, cast to int, so the caller can tell * JOB_TRANSITION_APPLIED apart from JOB_TRANSITION_REJECTED_BACKWARD/ * JOB_TRANSITION_NOOP_DUPLICATE exactly as that guard's other callers * already do. Only on JOB_TRANSITION_APPLIED does this function release * job->job.allocs (if non-NULL) and durably append the fact -- a rejected * or no-op outcome touches neither (Req 4.3/4.4). * * Unlike record_job_finish_fact/record_job_fail_fact, the caller (not this * function) must call persist_notify_job_updated_already_persisted(c, job) * itself, AFTER releasing the write lock, exactly once, whenever this * returns JOB_TRANSITION_APPLIED -- omitting it reproduces bug-1679 at this * call site. */ extern int record_job_reconcile_pending_fact(cluster_t*, job_t*); /* worker-rejoin-resync task 2.2 (Req 4.1, 4.3, 4.4, 4.7, 4.8, 7.1-7.3): * records a job the worker reports running that the controller has no * existing non-terminal record of ("unqueried extra", Req 3.6) as assigned * to the reporting worker `w`, matching `report`'s reported status -- * dispatch_a_job_()'s own eager-mutation-plus-explicit-notify pattern, NOT * record_job_finish_fact's deferred one -- see design.md's "Controller: * job.c" Components section, and job.c's own doc comment on this function * for the full rationale (including why job->job.allocs is deliberately * left NULL, and why there is no separate "host assignment" field for this * function to set beyond job->job.state). * * Precondition: caller holds `job` under RTTI_LOCK_WRITE on entry and does * not release it until this call returns -- this function does not * lock/unlock `job` itself. Unlike record_job_reconcile_pending_fact above, * this function performs no backward-transition guard of its own: the * caller must only invoke it when the controller has no existing * non-terminal record of this job (Req 4.8's own trigger condition), so * there is nothing more-final to clobber. * * Always returns (int)JOB_TRANSITION_APPLIED -- this outcome has no * reject/no-op case (unlike record_job_reconcile_pending_fact's * REJECTED_BACKWARD/NOOP_DUPLICATE). Matching that sibling's contract, the * caller (not this function) must call * persist_notify_job_updated_already_persisted(c, job) itself, AFTER * releasing the write lock, exactly once -- omitting it reproduces bug-1679 * at this call site. */ extern int record_job_reconcile_assign_fact(cluster_t*, job_t*, worker_t*, xce_job_state_entry_t*); /* worker-rejoin-resync task 3.2 (bug-1747 fix, Req 4.2): record_job_finish_ * fact()'s worker-reported-finish-time sibling, for the reconciliation * finished-success path only. Same body/guard/hook-fire/deferred-append * pattern as record_job_finish_fact() (job.c, unmodified) -- differs only * in passing `finish_time_ms` (converted ms -> s) to persist_submit_job_ * finish_fact() instead of time(NULL). See job.c's own doc comment on this * function for the full rationale, including the persist_submit_job_ * finish_fact() unit confirmation. * * Precondition: caller holds `job` under RTTI_LOCK_WRITE on entry and keeps * holding it until this call returns -- this function does not lock/unlock * `job` itself, matching record_job_finish_fact's own convention. * * DEFERRED pattern (unlike record_job_reconcile_pending_fact/assign_fact * above): this function only durably appends the fact; it does NOT mutate * job's in-memory state and the caller must NOT call persist_notify_job_ * updated_already_persisted() afterward -- exactly like record_job_finish_ * fact(), the mutation and notify happen later, via apply_committed_entry's * replay of the committed WAL_JOB_FINISH entry. */ extern int record_job_finish_fact_at(cluster_t*, job_t*, xce_jobfinish_t*, uint64_t); extern int on_job_start(cluster_t*, job_pidinfo_t*); extern int on_job_finish(cluster_t*, xce_jobfinish_t*); extern int on_job_fail(cluster_t*, job_failinfo_t*); extern int on_job_stop(cluster_t*, uint64_t); extern int on_job_resume(cluster_t*, uint64_t); extern int on_job_postdone(cluster_t*, endpoint_t*, uint64_t); extern int on_job_ckpt(cluster_t*, endpoint_t*, xce_ckptinfo_t*); extern int notify(cluster_t*, endpoint_t*, xce_notify_data_t*); extern int check_worker_(xce_worker_t*); extern int update_resval_list(cluster_t*, worker_t*, xce_resval_list_t*); extern int do_jobcontrol_op(cluster_t*, endpoint_t*, xce_request_t*); extern int job_ctrl(cluster_t*, xce_jobctrl_t*); extern int update_job_rusages(cluster_t*, endpoint_t*, xce_request_t*); extern int update_host_runusage(xce_job_runusage_t**, char*, int, xce_rusage_t**); extern void aggregate_job_runusage(xce_job_runusage_t*); extern int update_pids(job_pidinfo_t**, int, int*); extern int update_pgids(job_pidinfo_t**, int, int*); extern int get_ep_role(endpoint_t*); extern worker_t* get_ep_worker(cluster_t*, endpoint_t*); extern int check_worker_(xce_worker_t*); extern int update_resval_list(cluster_t*, worker_t*, xce_resval_list_t*); extern int do_jobcontrol_op(cluster_t*, endpoint_t*, xce_request_t*); extern int job_ctrl(cluster_t*, xce_jobctrl_t*); extern int do_queuecontrol_op(cluster_t*, endpoint_t*, xce_request_t*); extern int do_hostcontrol_op(cluster_t*, endpoint_t*, xce_request_t*); extern int do_runtasks(cluster_t*, endpoint_t*, xce_request_t*); extern int on_task_start(cluster_t*, xce_task_pidinfo_t*); extern int on_task_fail(cluster_t*, xce_task_failinfo_t*); extern int on_task_finish(cluster_t*, xce_task_finish_t*); extern int xce_controller_res_init(cluster_t*); extern int map_resource_to_host(char*, cluster_t*); extern int release_resource_from_worker(char*, cluster_t*); extern int collect_resval(var, zarray_t*, xlist_t*); extern int do_set_preason(cluster_t*, endpoint_t*, xce_request_t*); extern int do_clr_preason(cluster_t*, endpoint_t*, xce_request_t*); extern int on_fetchfile(cluster_t*, endpoint_t*, xce_request_t*); extern void fetchfile_signal(uint64_t jobid, int status, xce_fetchfile_reply_t*); /* bug-1693: wakes every fetchfile waiter (job.c) proxied to worker `w` * immediately on disconnect -- see fetchfile_signal_worker_disconnect()'s * own doc comment in job.c and schedadmin_signal_disconnect()'s sibling * comment in cluster.c for the shared Decision Log rationale. */ extern void fetchfile_signal_worker_disconnect(worker_t*); /* bug-1693: wakes the single pending OPCode_schedAdmin waiter (cluster.c) * immediately on scheduler disconnect, with a distinguishable -ECONNRESET * status -- see its doc comment in cluster.c. */ extern int schedadmin_signal_disconnect(void); extern int handle_apiserver_leave(cluster_t*, endpoint_t*); extern int add_apiserver(cluster_t*, char*, endpoint_t*); extern int apiserver_init(cluster_t*); #endif