myxce/src/scheduler/include/scheduler.h

#if !defined(_SCHEDULER_H_)

#define _SCHEDULER_H_ 1

#include <pthread.h>
#include <rtti.h>
#include <stdatomic.h>
#include <stdbool.h>
#include <stdint.h>
#include <stdio.h>
#include <time.h>

#include "client.h"
#include "core.h"
#include "dataq.h"
#include "decision.h"
#include "group.h"
#include "kvmgr.h"
#include "mylock.h"
#include "plugin.h"
#include "xce/cluster.h"
#include "xce/types.h"
#include "xlist.h"
#include "xvec.h"

typedef void* (*sched_group_worker_fn_t)(void*, xce_sched_worker_t*);

typedef enum {
    XCE_SCHED_OP_NONE = 0,
    XCE_SCHED_OP_UPDATE_CONFIG,
    XCE_SCHED_OP_UPDATE_JOB,
    XCE_SCHED_OP_JSTATE_CHANGE,
    XCE_SCHED_OP_UPDATE_WORKER,
    XCE_SCHED_OP_UPDATE_QUEUE,
    XCE_SCHED_OP_UPDATE_RESOURCES,
    XCE_SCHED_OP_UPDATE_RESVALS_FULL,
    XCE_SCHED_OP_UPDATE_RESVALS,
    XCE_SCHED_OP_JOB_RUSAGE,
    /* Notify-only single-job in-place field update. Distinct from
     * XCE_SCHED_OP_UPDATE_JOB (the dataset bulk-create path) so a notify
     * update to an existing job mutates it in place and never creates a job. */
    XCE_SCHED_OP_UPDATE_JOB_FIELDS,
    /* Compact job-array push (job-array-single-wal-record's xcesched_
     * push_jobarray() wire shape: ONE JobSubmit{jobid, arraySize, submit}
     * carrying an array's base jobid, element count, and one shared job
     * spec) -- deferred from the request-handling thread (handler.c's
     * scheduler_req_handler()) to the driver thread, the same pattern
     * XCE_SCHED_OP_ADMIN_RECORD below already establishes. Per-element
     * expansion (N independent deep clones via clone_job_submit_data()/
     * clone_job_startinfo(), driver.c) used to run inline in
     * sched_on_job_array() on whatever vtaskmgr thread handled the incoming
     * push -- for a large concurrent burst (live-observed: 6x50-element
     * array pushes landing within ~2ms of each other) this occupies several
     * of the vtaskmgr pool's threads for the duration of the expansion, the
     * exact "long/unbounded work on a pool that must stay free for
     * liveness-critical messages" anti-pattern ha-async-io.md's Thread-role
     * contract warns about (the pool is shared with the scheduler's own
     * controller connection traffic). The vtask-thread handler now only
     * takes ownership of the compact descriptor and enqueues it -- O(1),
     * bounded -- translate_one_pending_update() (driver.c) does the actual
     * per-element expansion on the dedicated round-processing thread
     * instead, which carries no connection liveness responsibility. */
    XCE_SCHED_OP_UPDATE_JOB_ARRAY,
    /* Scheduler admin recording control (scheduler-core spec, task 9.4, Req
     * 5 AC 10): a `recordStart`/`recordStop` SchedAdminRequest, deferred
     * from the request-handling thread (handler.c's scheduler_req_handler())
     * to the driver thread. sched_record_start()/sched_record_stop() do
     * real file I/O and sched_record_start() calls
     * xce_sched_core_save_snapshot(), which requires the core's
     * session_active precondition (task 8.2) -- a guarantee only held on
     * the single-threaded driver loop (main.c), not the async request-
     * handling thread. See xce_sched_update_request_t's `adminrecord`
     * member below and driver.c's xce_sched_driver_translate_pending_events()
     * for the full deferred-dispatch + deferred-reply design.
     * `recordStatus` is a pure read of already-current struct fields (no
     * file I/O, no session_active dependency) and is answered synchronously
     * from the handler thread instead -- it never reaches this queue. */
    XCE_SCHED_OP_ADMIN_RECORD
} xce_sched_op_t;

typedef struct {
    data_item_t    link;
    xce_sched_op_t op;
    union {
        xce_conf4sched_t* config;
        struct {
            int         njobs;
            xce_job_t** jobs;
        } jobupdate;
        struct {
            xce_job_t* job;
        } jobfieldsupdate;
        struct {
            /* Input: raw compact array descriptor -- ownership transferred
             * here from sched_handle_dataset()'s XceData_data_jobArray
             * case. See XCE_SCHED_OP_UPDATE_JOB_ARRAY's own comment above. */
            xce_job_submit_t* job_array;
            /* Output: populated by translate_one_pending_update()'s
             * expansion of job_array above -- the N per-element clones,
             * the SAME pointers aliased into the translated
             * XCE_CORE_EVENT_JOBS_PUSH event's jobs_push.jobs, freed by
             * release_translated_payload() the same way
             * XCE_SCHED_OP_UPDATE_JOB's own jobupdate.jobs is freed. Zero/
             * NULL until expansion runs (or if it produced nothing to
             * push -- an all-duplicate redelivery). */
            int         expanded_njobs;
            xce_job_t** expanded_jobs;
        } jobarrayupdate;
        struct {
            int            nworkers;
            xce_worker_t** workers;
        } workerupdate;
        struct {
            uint64_t jobid;
            uint32_t state;
        } job_state_change;
        struct {
            xce_queue_t* queue;
        } queueupdate;
        struct {
            int              nresources;
            xce_resource_t** resources;
        } resourceupdate;
        struct {
            int            nvals;
            xce_resval_t** vals;
        } resvalfullupdate;
        struct {
            xce_resval_list_t* resvals;
        } resvalupdate;
        struct {
            xce_job_rusages_t* rusages;
        } job_rusage;
        struct {
            /* SchedAdminAction_recordStart or SchedAdminAction_recordStop
             * (controller.capnp.h) -- recordStatus never reaches this
             * queue (answered synchronously, see xce_sched_op_t's own
             * comment on XCE_SCHED_OP_ADMIN_RECORD above). */
            int action;
            /* strdup'd copy of SchedAdminRequest.recordFile for
             * recordStart (NULL for auto-generate, and always NULL for
             * recordStop, which ignores it) -- the original xce_request_t
             * is freed by the handler-thread caller (xcectx.c's
             * vtask_entry()) immediately after scheduler_req_handler()
             * returns, so this must be an independent, owned copy for the
             * driver thread to read later. Freed by the deferred
             * dispatcher (driver.c) once the action completes. */
            char* record_file;
            /* The original request's seq -- the only field of the
             * now-freed xce_request_t the deferred reply needs
             * (init_reply_by_req()'s other field, opcode, is always
             * OPCode_schedAdmin here, so it is not worth storing). */
            uint64_t seq;
        } adminrecord;
    } u;
} xce_sched_update_request_t;

#define ITPTR_TO_UPDATE_REQ(p)                                                 \
    ((xce_sched_update_request_t*)(((char*)p)                                  \
                                   - offsetof(xce_sched_update_request_t,      \
                                              link)))

/*
 * xce_scheduler_t - scheduler-core spec, task 4.1 (Req 5.1/5.2): splits
 * along the controller-I/O seam into `core` (xce_sched_core_t, core.h --
 * all scheduling state, zero controller I/O) and the two fields below
 * (controller client context + HA candidate list). Every field that used
 * to live directly on this struct is unchanged in name/type/semantics --
 * it now lives on `core` (see core.h's field-by-field comment). Existing
 * callers holding an `xce_scheduler_t*` reach relocated state through
 * `&sched->core` at the specific call sites that need the narrower
 * `xce_sched_core_t*` type (the accessors declared in core.h); anything
 * still operating on the full driver-level struct is unaffected.
 */
typedef struct {
    xce_sched_core_t core;
    xcecli_ctx_t     apictx;
    xce_cluster_config_t
        cluster_cfg; /* HA candidate list. Empty (candidates_count==0) only
                      * transiently before sched_init() runs, when no
                      * --peers was configured; sched_init() always
                      * synthesizes a single-candidate default in that case
                      * (component-failover-adoption, task 3.1, Req 1 AC5)
                      * before connecting, so it is populated with >=1
                      * candidate for the rest of the process's life --
                      * single-host is the candidates_count=1 case of the
                      * one HA mechanism, not a separate mode. */
    /* Req 5 AC 10 (scheduler-core spec, task 9.1): recording directory,
     * provisioned EXACTLY ONE WAY -- the --record-dir startup CLI flag
     * (main.c), default "/var/log/xce/". Never settable via any
     * request-path field (SchedAdminRequest included) -- see
     * recording.h's sched_validate_record_filename() and design.md's
     * NormalDriver component. Lives here (not xce_sched_core_t) because
     * recording is a driver-level concern, the same reason apictx/
     * cluster_cfg above live here rather than on core. Points into argv
     * or a static string literal (never heap-owned), same convention as
     * this struct's neighboring driver-level string fields elsewhere in
     * this codebase (e.g. main.c's `logfile`) -- no free() needed. The
     * rest of the recording STATE design.md sketches for a future
     * xce_sched_driver_t (record_file_path/record_file/recording/
     * record_event_count/...) is task 9.2's job, not this field's. */
    char* record_dir;
    /* Req 5 AC 10 (scheduler-core spec, task 9.2): the recording lifecycle
     * state design.md sketches on a future xce_sched_driver_t -- per
     * record_dir's own comment above, this codebase keeps a single combined
     * driver-level struct (xce_scheduler_t), so these fields land here
     * instead of a new type. Owned exclusively by recording.c's
     * sched_record_start()/sched_record_stop() (recording.h) -- see those
     * functions' doc comments for the full lifecycle contract (path
     * validation + TOCTOU-hardened open on start, end-of-recording marker +
     * flush/close on stop). driver.c's normal-mode loop (driver.h) reads
     * `recording` to boolean-gate its per-event-batch/per-session recording
     * writes -- see xce_sched_driver_translate_pending_events()'s and
     * xce_sched_driver_run_normal_mode_round()'s own comments.
     *
     * Invariant: recording == true iff record_file != NULL.
     *
     * Concurrency (scheduler-core spec, task 9.4 fix round): `recording`,
     * `record_file_path`, `record_file`, `record_event_count`, and
     * `record_start_ts` are written on the single-threaded driver loop
     * (main.c) but ALSO read by sched_record_status() (recording.c), which
     * runs on whichever vtask worker pthread handled the synchronous
     * `recordStatus` SchedAdminRequest (handler.c's
     * scheduler_req_handler(), invoked with no lock held --
     * lib/xcectx.c). That makes this a genuine cross-thread reader/writer
     * pair with no other synchronization, so every read or write of these
     * five fields MUST hold `record_lock` below -- see its own comment. */
    bool recording;
    /* strdup'd copy of the resolved record file path; owned by the record
     * file lifecycle (strdup on start, free+NULL on stop or start/write
     * failure) -- lets a status query (task 9.4) answer with the current
     * path even after record_file closes. Protected by `record_lock` (see
     * `recording`'s comment above and `record_lock`'s own comment below) --
     * sched_record_status() must copy this string's bytes (e.g. strdup())
     * while still holding the lock rather than handing back the live
     * pointer, since the driver thread may free() it the instant the lock
     * is released. */
    char* record_file_path;
    /* replay file handle; NULL when not recording. Protected by
     * `record_lock` for the same reason as `record_file_path` above. */
    FILE* record_file;
    /* count of ReplayEvent messages written since recording started (used
     * by task 9.3's periodic-checkpoint trigger; incremented by
     * recording.c's event-batch writer, task 9.2). Protected by
     * `record_lock`. */
    uint64_t record_event_count;
    /* N events between periodic snapshot checkpoints (0 = disabled). Read
     * by recording_events.c's sched_record_write_events() (task 9.3):
     * every time cumulative `record_event_count` lands exactly on a
     * multiple of this value, a fresh SnapshotEntry is appended to the
     * record file (Req 5 AC 11). Driver-thread-only (never read by
     * sched_record_status()) -- not protected by `record_lock`. */
    uint64_t record_snapshot_interval;
    /* logical time (xce_core_now()) recording started; 0 = not recording.
     * Protected by `record_lock`. */
    uint64_t record_start_ts;
    /* Dedicated mutex (scheduler-core spec, task 9.4 fix round) guarding
     * `recording`/`record_file_path`/`record_file`/`record_event_count`/
     * `record_start_ts` -- see `recording`'s comment above for the
     * cross-thread reader/writer hazard this closes. Deliberately a NEW,
     * separate lock rather than reusing `core.lock`: `core`
     * (xce_sched_core_t) is the conceptually distinct core-vs-driver piece
     * of state this struct's own header comment describes, and these
     * recording fields are driver-level (like `record_dir`/`apictx`/
     * `cluster_cfg` above), so widening core.lock's scope to also cover
     * them would blur that boundary for no benefit. Held only around the
     * struct-field reads/writes themselves -- never across the actual file
     * I/O (fopen()/fwrite()/capn_write_mem() etc.) -- matching this
     * codebase's low-contention-lock-scope convention. Initialized by
     * sched_init(), destroyed by sched_fini() (scheduler.c) -- same
     * lifecycle pairing as `core.lock` (core.c). */
    mylock_t record_lock;
    /* bug-1846: a config push received on the handler thread (handler.c's
     * scheduler_req_handler()), parked here until the driver loop applies
     * it between rounds via sched_apply_pending_config(). The controller
     * re-pushes config on every scheduler join, and the scheduler re-joins
     * from inside a round's decision publish (sched_check_pending_rejoin(),
     * via sched_publish_retry_loop()) -- applying it on arrival reset the
     * picker's cursor mid-session and ended the round early. Only the
     * newest push is kept (each is a complete Conf4Sched). Owned by this
     * struct until applied; freed by sched_fini() if never applied.
     * Protected by `pending_config_lock`. */
    xce_conf4sched_t* pending_config;
    mylock_t          pending_config_lock;
} xce_scheduler_t;

extern int  sched_init(xce_scheduler_t*, xce_client_config_t*,
                       const xce_cluster_config_t*, xce_request_handler_t);
extern void sched_fini(xce_scheduler_t*);
extern int  sched_init_config(xce_scheduler_t* sched, xce_conf4sched_t*);
/* bug-1846: see xce_scheduler_t.pending_config. sched_stash_config() is
 * the handler-thread half (takes ownership of `config`, replacing -- and
 * freeing -- any older unapplied push); sched_apply_pending_config() is the
 * driver-thread half, run between rounds, and returns 1 if a config was
 * applied, 0 if none was pending, or sched_init_config()'s negative rc. */
extern int  sched_stash_config(xce_scheduler_t* sched, xce_conf4sched_t*);
extern int  sched_apply_pending_config(xce_scheduler_t* sched);
/* Req 2 AC 1 (scheduler-core spec, task 1.3): applies the "builtin_fcfs"
 * default to sched->job_pick_plugin when unset. See the field comment on
 * xce_scheduler_t above -- this does NOT resolve/register the name. */
extern void sched_set_default_job_pick_plugin(xce_scheduler_t*);
/* sched_resolve_job_picker relocated to core.h (scheduler-core spec, task
 * 4.3) -- it now operates on xce_sched_core_t*, not xce_scheduler_t*, so
 * xce_core_push_events() (core.c) can call it without core.c depending on
 * this header. Call sites in sched_election.c (sched_job_picker(),
 * sched_allocator()) and this file (sched_init()) now pass &sched->core
 * (or, for the two thread bodies, `core` directly -- see sched_election.c's
 * own header comment, scheduler-sim-live-allocation spec task 3.1). See
 * core.h for the full contract. */
/* scheduler-sim-live-allocation spec, task 3.1 (Unified Picker/Allocator
 * Threads): sched_job_picker()/sched_allocator() relocated out of this
 * file into their own file, sched_election.c -- the single implementation
 * both normal mode (sched_init(), below, passing &(sched->core)) and a
 * future simulation driver (task 4.1, passing &(sim->core)) spawn. Retyped
 * from xce_scheduler_t* to xce_sched_core_t*; pthread_create()'s own
 * callback signature (void* (*)(void*)) is unchanged either way. Declared
 * here rather than a new header, mirroring this file's own established
 * convention for every other function this spec's migration
 * relocated/retyped (sched_meet_host_selection, sched_try_alloc_on_host_,
 * etc. below). */
extern void* sched_job_picker(void* arg);
extern void* sched_allocator(void* arg);
/* scheduler-sim-live-allocation spec, task 3.1: exported (was `static`) --
 * its sole caller, sched_allocator(), moved out of this file (scheduler.c)
 * into sched_election.c, so this cross-file call needs external linkage.
 * Signature already core-safe since task 2.2. */
extern int  sched_make_allocation(xce_sched_core_t*, xce_sched_obj_t*);
extern int  sched_update(xce_scheduler_t*);
extern void sched_go(xce_scheduler_t*);
extern void sched_begin_session(xce_scheduler_t*);
extern void sched_end_session(xce_scheduler_t*);
/* sched_account_jobs/sched_charge_account/sched_uncharge_account/
 * sched_commit_job_alloc relocated to core.h (scheduler-sim-decision-parity
 * spec, task 1.1) -- same rationale as sched_resolve_job_picker above: they
 * now operate on xce_sched_core_t*, not xce_scheduler_t*, so the simulation
 * driver (mode-agnostic, no xce_scheduler_t* of its own) gains the same
 * accounting capability the normal-mode driver already had. Call sites in
 * this file (sched_begin_session() below, sched_commit_decision()) now pass
 * &sched->core. See core.h for the full contract; body unchanged. */
/* sched_get_job/sched_add_job/sched_get_queue/sched_add_queue/
 * sched_get_worker/sched_add_worker/sched_get_user/sched_add_user/
 * sched_get_project/sched_add_project/sched_job_is_ready/
 * sched_queue_is_ready relocated to core.h (scheduler-core spec, task 4.1)
 * -- they now operate on xce_sched_core_t*, not xce_scheduler_t*. */
extern int  sched_update_existing_queue(xce_scheduler_t*, xce_queue_t*);
extern int  sched_update_existing_job(xce_scheduler_t*, xce_job_t*);
extern int  sched_apply_job_rusage(xce_scheduler_t*, xce_job_rusages_t*);
/* scheduler-sim-live-allocation spec, task 2.2 (Core-Safe Allocation
 * Chain): sched_commit_decision() is now core-typed (xce_sched_core_t*),
 * not xce_scheduler_t* -- it, plugin_commit_decision(), sched_get_job()
 * and sched_commit_job_alloc() (the latter two already relocated to
 * core.h) are all core-safe now, so the whole DISPATCH-commit path is
 * callable from either driver. Its sole production caller,
 * commit_decision() (scheduler.c, static), wraps its own still-
 * xce_scheduler_t*-typed argument as &(sched->core) at the call site. */
extern int  sched_commit_decision(xce_sched_core_t*, xce_decision_t*);
extern int  sched_publish_pend_reason(xce_scheduler_t*, xce_preason_set_t*);
extern int  sched_publish_dispatch(xce_scheduler_t*, xce_job_dispatch_t*);
/* bug-1760: exposed (was `static sched_publish_check_rejoin`) so
 * xce_sched_driver_run_normal_mode_round() (driver.c) can call it
 * unconditionally once per round, not only as a side effect of a
 * successful decision publish -- see scheduler.c's doc comment on this
 * function for the chicken-and-egg deadlock this closes. */
extern void sched_check_pending_rejoin(xce_scheduler_t*);
extern void sched_worker_release_host_ctx(xce_sched_worker_t*);
/* scheduler-sim-live-allocation spec, task 1.3: retyped from
 * xce_scheduler_t* to xce_sched_core_t* -- these now operate on shared core
 * state directly, callable identically from either driver (both
 * xce_scheduler_t and xce_sched_sim_driver_t embed xce_sched_core_t core as
 * their first member). Body unchanged besides sched->core.X -> core->X
 * field-access updates. sched_check_shared_satisfy()/sched_check_host_
 * satisfy() relocated to core.h (scheduler-driver-equivalence fix,
 * bug-1836) -- same rationale as sched_account_jobs' own relocation there:
 * sim_loop.c needs to call them without pulling in scheduler.h's
 * controller-facing includes (client.h, xce/cluster.h), which
 * test_sim_isolation.c forbids. */
extern int  sched_meet_host_selection(xce_sched_core_t*, xce_sched_worker_t*,
                                      xce_select_t*);
extern int  sched_collect_workers(xce_scheduler_t*, xlist_t*,
                                  zarray_iter_skip_fn_t,
                                  sched_group_worker_fn_t);
extern uint64_t       sched_get_id(xce_scheduler_t*);
extern xce_alloc_t**  sched_make_allocs(xce_scheduler_t*, char*, int, int*,
                                        float*, int*);
/* scheduler-allocation spec, task 3.1 remediation: exported (was `static`,
 * wrapped by a variadic sched_try_alloc_on_hosts() whose sole caller was
 * seqjob_allocator.c's pre-rewrap xlist_find() call site -- both now
 * deleted) so both seqjob_allocator.c and parjob_allocator.c can call this
 * shared Composition-Algorithm try_alloc leaf directly with real
 * parameters, instead of each carrying its own private copy (today its
 * one allocator caller is the framework routine in alloc_runner.c). */
/* scheduler-sim-live-allocation spec, task 2.1 (Core-Safe Allocation
 * Chain): retyped from xce_scheduler_t* to xce_sched_core_t* -- callers
 * still owning an xce_scheduler_t* pass &(sched->core) (the framework
 * routine sched_allocator_run(), alloc_runner.c, instead passes the
 * plugin's own `core` field straight through); both xce_scheduler_t and
 * xce_sched_sim_driver_t embed xce_sched_core_t core as their first-class
 * field, so either driver's own core state satisfies this signature
 * identically. */
extern int            sched_try_alloc_on_host_(xce_sched_worker_t*,
                                               xce_sched_core_t*,
                                               xce_sched_ctx_t*);
extern int            get_resvals(zarray_t*, int, int*, float*);
extern int            put_resvals(zarray_t*, int, int*, float*);
extern request_res_t* make_resource_request(char*, float);
extern int sched_fill_limit_requests(xce_sched_alloc_req_t*, xlist_t*,
                                     xlist_t*, int, int);

#endif