diff --git a/.kiro/specs/scheduler-core/tasks.md b/.kiro/specs/scheduler-core/tasks.md index 753ed854..6f59f1d7 100644 --- a/.kiro/specs/scheduler-core/tasks.md +++ b/.kiro/specs/scheduler-core/tasks.md @@ -74,20 +74,22 @@ - _Requirements: 4.1, 4.2, 6.1, 6.2, 6.3, 6.4_ - _Depends: 2.5_ -- [ ] 3. Run Phase 1 HA soak regression +- [x] 3. Run Phase 1 HA soak regression - Run sequential and chaotic HA failover soak rounds against a cluster running the new picker architecture - Observable: zero new job loss, decision divergence, or liveness stall attributable to this phase's changes across the soak run - _Requirements: 4.3_ - _Depends: 2.6_ + - _Verified: 15-round sequential soak, 2026-08-03 — PASS: 15 rounds, 48 jobs submitted, zero job loss; all 3 nodes led (ctrl-1,ctrl-2,ctrl-3)_ - [ ] 4. Foundation: Scheduler Core Extraction -- [ ] 4.1 Extract the scheduling core from the monolithic scheduler struct +- [x] 4.1 Extract the scheduling core from the monolithic scheduler struct - Split scheduling state (jobs, queues, workers, resources, resvals, users, projects, plugin instances) into its own core structure holding zero controller I/O state (no API client context, no cluster config) - Confirm the core links only against the approved minimal data-structure subset and never links network/timer/API libraries — no wall-clock time call appears anywhere in the extracted core's source - Observable: the core structure compiles and links with zero symbols from the networking/API libraries; a build-time or link-time check confirms this - _Requirements: 5.1, 5.2_ - _Depends: 3_ + - _Verified: 2026-08-03 — new xce_sched_core_t (core.h/core.c) holds all scheduling state, xce_scheduler_t shrank to {core; apictx; cluster_cfg}; nm-based isolation test (test_core_isolation.c) proves zero netio/xcecli/idgen/delaymgr symbols; 14/14 scheduler suite passing; independent review APPROVED (own fresh build/test/nm run + reproduced RED)_ - [ ] 4.2 Pin the determinism invariants - Hardcode the picker-to-allocator handoff channel's capacity to 1 inside core initialization — never expose it as a parameter diff --git a/src/scheduler/include/core.h b/src/scheduler/include/core.h new file mode 100644 index 00000000..3369be67 --- /dev/null +++ b/src/scheduler/include/core.h @@ -0,0 +1,117 @@ +#if !defined(_CORE_H_) + +#define _CORE_H_ 1 + +/* + * core.h - xce_sched_core_t: the scheduler's pure scheduling-state + * structure (scheduler-core spec, task 4.1, Req 5.1/5.2). + * + * Boundary (design.md "SchedulerCore" / "Existing Architecture Analysis"): + * `xce_scheduler_t` (scheduler.h) splits along the controller-I/O seam into + * this structure (no `apictx`/`cluster_cfg` -- zero controller I/O, network, + * or wall-clock dependency) and the driver-level fields that remain on + * `xce_scheduler_t` itself. Every field below is the SAME field, same name, + * same type, that lived directly on `xce_scheduler_t` before this task -- + * this is a relocation, not a redesign; `xce_scheduler_t` now embeds this + * struct as its `core` member instead of declaring these fields itself. + * + * Allowed dependencies only (design.md "Allowed Dependencies", Req 5 AC 3): + * ordq, dataq, cbuf, xlist, zarray, kvmgr, rtti. This header and core.c + * must NEVER include client.h/netio.h/xceapi.h/delay.h/idgen.h/ + * xce/cluster.h, and must never call a wall-clock function (time(), + * clock_gettime(), gettimeofday()) -- enforced by + * testsuite/test_core_isolation.c (source-text scan + an nm(1) undefined- + * symbol audit of core.c's compiled object, the build-time proof this + * task's Observable requires). + */ + +#include +#include +#include +#include + +#include "dataq.h" +#include "mylock.h" +#include "sched/types.h" +#include "xce/types.h" +#include "zarray.h" + +/* xce_sched_core_t is also forward-declared in plugin.h (picker lifecycle + * hook parameter type); this typedef and the full struct definition below + * are the canonical declaration -- same idiom as cluster_t across + * myconf.h/persist.h/ha.h/cluster_def.h (an identical repeated typedef is + * valid C11/gnu17, so include order between core.h and plugin.h doesn't + * matter). */ +typedef struct xce_sched_core_s xce_sched_core_t; + +struct xce_sched_core_s { + mylock_t lock; + int stop; + uint64_t nextid; + xce_conf4sched_t* config; + uint64_t configver; + /* Req 2 AC 1 (scheduler-core spec, task 1.3): global picker plugin + * name. NULL until sched_set_default_job_pick_plugin() (schedobj.c) + * applies the "builtin_fcfs" default, or a future config-driven + * assignment sets it explicitly. */ + char* job_pick_plugin; + var plugins; + void* seqjob_allocator; + void* parjob_allocator; + void* affinity_allocator; + void* gpu_allocator; + dataq_t* pend_updates[2]; + var resmgr; + var ugmgr; + var qgmgr; + var pgmgr; + var hgmgr; + zarray_t sharedInst; + zarray_t sharedInUse; + var jobs; + var queues; + var workers; + var users; + var projects; + pthread_t jobpicker; + pthread_t allocator; + cbuf_t wbuf; /* action to trigger scheduling */ + cbuf_t jbuf; /* job picked for allocation */ + cbuf_t decisions; /* scheduler decisions */ + _Atomic int session_active; /* 1 during scheduling session, 0 between */ + _Atomic int eval_in_progress; /* 1 while eval_select loop runs */ + _Atomic int staging_merge_complete; /* 1 after staging merge completes */ + select_eval_ctx_t* sel_eval_ctx; /* session-level memoisation context; NULL + between sessions */ +}; + +/* + * xce_sched_core_init/xce_sched_core_fini - allocate/initialize, and tear + * down, exactly the state above. Deliberately narrower than the design's + * eventual Service Interface (`xce_sched_core_init(core, decision_cb, + * driver_ctx, prng_seed)`, design.md "SchedulerCore" component) -- the + * decision callback, driver context, and PRNG seed belong to the event/ + * decision protocol tasks 4.3-4.5 build; this task only extracts state + * init/fini, mirroring the non-I/O portion of today's sched_init()/ + * sched_fini() (scheduler.c). + */ +extern int xce_sched_core_init(xce_sched_core_t*); +extern void xce_sched_core_fini(xce_sched_core_t*); + +/* Relocated state-only accessors/mutators (task 4.1): moved out of + * scheduler.c/schedobj.c, now operating on xce_sched_core_t* -- zero + * controller I/O, exactly like the rest of this header. */ +extern xce_sched_job_t* sched_get_job(xce_sched_core_t*, uint64_t); +extern int sched_add_job(xce_sched_core_t*, xce_sched_job_t*); +extern xce_sched_queue_t* sched_get_queue(xce_sched_core_t*, char*); +extern int sched_add_queue(xce_sched_core_t*, xce_sched_queue_t*); +extern xce_sched_worker_t* sched_get_worker(xce_sched_core_t*, char*); +extern int sched_add_worker(xce_sched_core_t*, xce_sched_worker_t*); +extern xce_sched_user_t* sched_get_user(xce_sched_core_t*, char*); +extern int sched_add_user(xce_sched_core_t*, xce_sched_user_t*); +extern xce_sched_project_t* sched_get_project(xce_sched_core_t*, char*); +extern int sched_add_project(xce_sched_core_t*, xce_sched_project_t*); +extern int sched_job_is_ready(xce_sched_job_t*); +extern int sched_queue_is_ready(xce_sched_queue_t*); + +#endif diff --git a/src/scheduler/include/scheduler.h b/src/scheduler/include/scheduler.h index a9e5682d..0842f019 100644 --- a/src/scheduler/include/scheduler.h +++ b/src/scheduler/include/scheduler.h @@ -9,6 +9,7 @@ #include #include "client.h" +#include "core.h" #include "dataq.h" #include "decision.h" #include "group.h" @@ -84,49 +85,23 @@ typedef struct { - 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 { - mylock_t lock; - int stop; - xcecli_ctx_t apictx; - uint64_t nextid; - xce_conf4sched_t* config; - uint64_t configver; - /* Req 2 AC 1 (scheduler-core spec, task 1.3): global picker plugin - * name. NULL until sched_set_default_job_pick_plugin() (schedobj.c) - * applies the "builtin_fcfs" default, or a future config-driven - * assignment sets it explicitly. This field only carries the *name* -- - * resolving it against the plugin registry (plugin_register/key_find) - * and failing startup loudly on an unregistered name (Req 2 AC 5) is - * task 2.4's job. */ - char* job_pick_plugin; - var plugins; - void* seqjob_allocator; - void* parjob_allocator; - void* affinity_allocator; - void* gpu_allocator; - dataq_t* pend_updates[2]; - var resmgr; - var ugmgr; - var qgmgr; - var pgmgr; - var hgmgr; - zarray_t sharedInst; - zarray_t sharedInUse; - var jobs; - var queues; - var workers; - var users; - var projects; - pthread_t jobpicker; - pthread_t allocator; - cbuf_t wbuf; /* action to trigger scheduling */ - cbuf_t jbuf; /* job picked for allocation */ - cbuf_t decisions; /* scheduler decisions */ - _Atomic int session_active; /* 1 during scheduling session, 0 between */ - _Atomic int eval_in_progress; /* 1 while eval_select loop runs */ - _Atomic int staging_merge_complete; /* 1 after staging merge completes */ - select_eval_ctx_t* sel_eval_ctx; /* session-level memoisation context; NULL between sessions */ - xce_cluster_config_t cluster_cfg; /* HA candidate list — empty means single-host mode */ + xce_sched_core_t core; + xcecli_ctx_t apictx; + xce_cluster_config_t + cluster_cfg; /* HA candidate list — empty means single-host mode */ } xce_scheduler_t; extern int sched_init(xce_scheduler_t*, xce_client_config_t*, @@ -156,21 +131,14 @@ 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*); -extern xce_sched_job_t* sched_get_job(xce_scheduler_t*, uint64_t); -extern xce_sched_queue_t* sched_get_queue(xce_scheduler_t*, char*); -extern xce_sched_worker_t* sched_get_worker(xce_scheduler_t*, char*); -extern xce_sched_user_t* sched_get_user(xce_scheduler_t*, char*); -extern xce_sched_project_t* sched_get_project(xce_scheduler_t*, char*); -extern int sched_add_job(xce_scheduler_t*, xce_sched_job_t*); -extern int sched_add_queue(xce_scheduler_t*, xce_sched_queue_t*); +/* 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*); -extern int sched_add_user(xce_scheduler_t*, xce_sched_user_t*); -extern int sched_add_worker(xce_scheduler_t*, xce_sched_worker_t*); -extern int sched_add_project(xce_scheduler_t*, xce_sched_project_t*); -extern int sched_job_is_ready(xce_sched_job_t*); -extern int sched_queue_is_ready(xce_sched_queue_t*); +extern int sched_apply_job_rusage(xce_scheduler_t*, xce_job_rusages_t*); extern int sched_commit_decision(xce_scheduler_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*); diff --git a/src/scheduler/meson.build b/src/scheduler/meson.build index dc27350d..a0633fe6 100644 --- a/src/scheduler/meson.build +++ b/src/scheduler/meson.build @@ -2,6 +2,7 @@ scheduler_srcs = [ 'src/main.c', 'src/handler.c', 'src/scheduler.c', + 'src/core.c', 'src/decision.c', 'src/schedobj.c', 'src/plugin.c', @@ -33,4 +34,19 @@ scheduler = executable('xce_scheduler', link_args: wrap_link_args, install: true) +# Task 4.1 (scheduler-core spec, Req 5.1/5.2): isolated compile target for +# core.c alone, consumed by testsuite/test_core_isolation.c to prove at +# build time that core.c references zero symbols from the networking/API +# libraries (netio/xceapi/client/delay/idgen). A static_library() here need +# not LINK against anything for this purpose -- compiling src/core.c to an +# object is enough; the object's own undefined-symbol table (inspected via +# nm(1) in the test) reflects exactly what core.c's code calls, independent +# of which libraries are nominally "dependencies" of this build target. +sched_core_isolation_lib = static_library( + 'sched_core_isolation', + ['src/core.c'], + include_directories: incdir, + dependencies: scheduler_deps, + install: false) + subdir('testsuite') diff --git a/src/scheduler/src/core.c b/src/scheduler/src/core.c new file mode 100644 index 00000000..16dd34da --- /dev/null +++ b/src/scheduler/src/core.c @@ -0,0 +1,345 @@ +#include +#include +#include +#include + +#include "core.h" +#include "decision.h" +#include "group.h" +#include "plugin.h" +#include "sched/types.h" +#include "xstring.h" + +/* + * core.c - xce_sched_core_t state init/fini and the relocated jobs/queues/ + * workers/users/projects CRUD accessors (scheduler-core spec, task 4.1, + * Req 5.1/5.2). See core.h for the boundary this file must respect: zero + * controller I/O, network, filesystem, or wall-clock dependency. + * + * xce_sched_core_init()/xce_sched_core_fini() cover exactly the core-state + * portion of today's sched_init()/sched_fini() (scheduler.c) -- plugin + * REGISTRATION (sched_register_plugins(), which still needs the full + * xce_scheduler_t* for legacy plugin->sched storage), the controller-join + * sequence (the client API family in client.h), and picker/allocator + * thread creation/join remain in + * scheduler.c's sched_init()/sched_fini(), which call these two functions + * for the state-only portion and otherwise keep their existing shape and + * execution order unchanged (no behavior change, Req 4.1/4.2 regression + * bar). `pend_updates`'s dataq_init also stays in scheduler.c's sched_init() + * -- its killer callback (sched_pend_update_killer) is intimately coupled + * to xce_sched_update_request_t/xce_sched_op_t and sched_update()'s dispatch + * switch, both still scheduler.c-owned; only the storage (the dataq_t* + * pointers themselves) lives in xce_sched_core_t. + */ + +int xce_sched_core_init(xce_sched_core_t* core) +{ + int ndecisions = 5; /* FIXME: make this configurable */ + int njobs = 1; /* FIXME: make this configurable */ + + kvmgr_params_t plugins_params = { .key_type = KVMGR_KEY_STRING, + .nslots = 16, + .delta = 8, + .datakiller = plugin_unregister }; + kvmgr_params_t jobs_params = { .key_type = KVMGR_KEY_INT, + .nslots = 5381, /* FIXME */ + .delta = 1024, + .datakiller = kill_sched_job }; + kvmgr_params_t queues_params + = { .key_type = KVMGR_KEY_STRING, + .nslots = 59, + .delta = 16, + .datakiller = (destroy_fn_t)kill_sched_queue }; + kvmgr_params_t workers_params + = { .key_type = KVMGR_KEY_STRING, + .nslots = 5381, + .delta = 64, + .datakiller = (destroy_fn_t)kill_sched_worker }; + kvmgr_params_t users_params + = { .key_type = KVMGR_KEY_STRING, + .nslots = 5381, + .delta = 64, + .datakiller = (destroy_fn_t)kill_sched_user }; + kvmgr_params_t projects_params + = { .key_type = KVMGR_KEY_STRING, + .nslots = 31, + .delta = 8, + .datakiller = (destroy_fn_t)kill_sched_project }; + kvmgr_params_t resmgr_params = { .key_type = KVMGR_KEY_STRING, + .nslots = 64, + .delta = 8, + .datakiller = NULL }; + + if (core == NULL) { + return -EINVAL; + } + + memset(core, 0x0, sizeof(*core)); + + resreq_init(1); /* enable CSE intern table for scheduler process */ + + core->nextid = 1; + core->configver = 0; + + xstring_table_init(1024 * 1024); + + mylock_init(&(core->lock)); + cbuf_init(&(core->wbuf), 1, sizeof(int)); + cbuf_init(&(core->jbuf), njobs, sizeof(uint64_t)); + cbuf_init(&(core->decisions), ndecisions, sizeof(xce_decision_t*)); + + core->resmgr = makeInstance(&xkvmgr_type, &resmgr_params); + + groupmgr_init(&(core->ugmgr), 64, 64); + groupmgr_init(&(core->qgmgr), 16, 16); + groupmgr_init(&(core->pgmgr), 16, 16); + groupmgr_init(&(core->hgmgr), 64, 64); + + core->plugins = makeInstance(&xkvmgr_type, &plugins_params); + if (core->plugins == NULL) { + return -errno; + } + + core->jobs = makeInstance(&xkvmgr_type, &jobs_params); + core->queues = makeInstance(&xkvmgr_type, &queues_params); + core->workers = makeInstance(&xkvmgr_type, &workers_params); + core->users = makeInstance(&xkvmgr_type, &users_params); + core->projects = makeInstance(&xkvmgr_type, &projects_params); + + zarray_init(&(core->sharedInst), 128, + (zarray_item_killer_t)kill_sched_resval); + zarray_init(&(core->sharedInUse), 128, NULL); + + return 0; +} + +void xce_sched_core_fini(xce_sched_core_t* core) +{ + if (core == NULL) { + return; + } + + cbuf_fini(&(core->decisions)); + cbuf_fini(&(core->jbuf)); + cbuf_fini(&(core->wbuf)); + + dataq_fini(core->pend_updates[0]); + dataq_fini(core->pend_updates[1]); + + destroy(core->resmgr); + + destroy(core->ugmgr); + destroy(core->qgmgr); + destroy(core->pgmgr); + destroy(core->hgmgr); + + resreq_fini(); + + mylock_fini(&(core->lock)); + + return; +} + +int sched_add_job(xce_sched_core_t* core, xce_sched_job_t* j) +{ + int rc; + + lock(core->jobs, RTTI_LOCK_WRITE); + rc = key_add(core->jobs, (void*)j->jobid, j); + unlock(core->jobs); + + return rc; +} + +xce_sched_job_t* sched_get_job(xce_sched_core_t* core, uint64_t jobid) +{ + int rc = 0; + xce_sched_job_t* job = NULL; + + lock(core->jobs, RTTI_LOCK_READ); + rc = key_find(core->jobs, (void*)jobid, (void**)&job); + unlock(core->jobs); + + if (rc <= 0) { + return NULL; + } + + return job; +} + +int sched_add_queue(xce_sched_core_t* core, xce_sched_queue_t* q) +{ + int rc; + + lock(core->queues, RTTI_LOCK_WRITE); + rc = key_add(core->queues, (void*)q->name, q); + unlock(core->queues); + + return rc; +} + +xce_sched_queue_t* sched_get_queue(xce_sched_core_t* core, char* name) +{ + xce_sched_queue_t* q = NULL; + int rc = 0; + + lock(core->queues, RTTI_LOCK_READ); + rc = key_find(core->queues, (void*)name, (void**)&q); + unlock(core->queues); + + if (rc <= 0) { + return NULL; + } + + return q; +} + +int sched_add_worker(xce_sched_core_t* core, xce_sched_worker_t* w) +{ + int rc; + + lock(core->workers, RTTI_LOCK_WRITE); + rc = key_add(core->workers, (void*)w->name, (void*)w); + unlock(core->workers); + if (rc >= 0) { + w->idx = rc; + } + + return rc; +} + +xce_sched_worker_t* sched_get_worker(xce_sched_core_t* core, char* name) +{ + xce_sched_worker_t* w = NULL; + int rc = 0; + + lock(core->workers, RTTI_LOCK_READ); + rc = key_find(core->workers, (void*)name, (void**)&w); + unlock(core->workers); + + if (rc <= 0) { + return NULL; + } + + return w; +} + +int sched_add_user(xce_sched_core_t* core, xce_sched_user_t* u) +{ + int rc = 0; + + lock(core->users, RTTI_LOCK_WRITE); + rc = key_add(core->users, (void*)u->name, (void*)u); + unlock(core->users); + + return rc; +} + +/* make_sched_user's `arg` is, by construction, always an xce_sched_core_t* + * (the only two call sites are sched_get_user below, and the test suite's + * fixtures which pass `&sched.core` -- see sched/types.h's `void*` + * declaration, which lets this file and its callers agree on the real + * pointer type without widening that header's public signature). */ +xce_sched_user_t* make_sched_user(void* arg, char* name) +{ + xce_sched_core_t* core = (xce_sched_core_t*)arg; + xce_sched_user_t* user = NULL; + + user = (xce_sched_user_t*)calloc(1, sizeof(*user)); + if (user == NULL) { + return NULL; + } + + user->name = makeString(name); + + sched_add_user(core, user); + + return user; +} + +xce_sched_user_t* sched_get_user(xce_sched_core_t* core, char* name) +{ + xce_sched_user_t* user = NULL; + int rc = 0; + + lock(core->users, RTTI_LOCK_READ); + rc = key_find(core->users, (void*)name, (void**)&user); + unlock(core->users); + if (rc < 0) { + return NULL; + } + if (user != NULL) { + return user; + } + + return make_sched_user(core, name); +} + +int sched_add_project(xce_sched_core_t* core, xce_sched_project_t* p) +{ + int rc; + + lock(core->projects, RTTI_LOCK_WRITE); + rc = key_add(core->projects, (void*)p->name, (void*)p); + unlock(core->projects); + + return rc; +} + +/* Same `arg`-is-always-xce_sched_core_t* convention as make_sched_user + * above; the test suite's direct call site passes `&sched.core`. */ +xce_sched_project_t* make_sched_project(void* arg, char* name) +{ + xce_sched_core_t* core = (xce_sched_core_t*)arg; + xce_sched_project_t* project = NULL; + + project = (xce_sched_project_t*)calloc(1, sizeof(*project)); + if (project == NULL) { + return NULL; + } + + if ((name == NULL) || (strlen(name) == 0)) { + project->name = makeString("default"); + } else { + project->name = makeString(name); + } + + /* Req 2 AC (scheduler-core spec, task 1.3): scaffolding only -- no + * quota/hierarchy inputs exist yet, so priority defaults to 0 + * (calloc() above already zeroes it; explicit here for + * documentation/testability). */ + project->priority = 0; + + sched_add_project(core, project); + + return project; +} + +xce_sched_project_t* sched_get_project(xce_sched_core_t* core, char* name) +{ + xce_sched_project_t* project = NULL; + int rc = 0; + + if ((name == NULL) || (strlen(name) == 0)) { + name = "default"; + } + + lock(core->projects, RTTI_LOCK_READ); + rc = key_find(core->projects, (void*)name, (void**)&project); + unlock(core->projects); + if (rc < 0) { + return NULL; + } + + if (rc > 0) { + return project; + } + + return make_sched_project(core, name); +} + +/* FIXME this is just stub, we need to add more */ +int sched_job_is_ready(xce_sched_job_t* j) { return 1; } + +/* FIXME this is just stub, we need to add more */ +int sched_queue_is_ready(xce_sched_queue_t* q) { return 1; } diff --git a/src/scheduler/src/handler.c b/src/scheduler/src/handler.c index 7a598da8..ae7e9bb2 100644 --- a/src/scheduler/src/handler.c +++ b/src/scheduler/src/handler.c @@ -16,9 +16,9 @@ static void enq_pend_update(xce_scheduler_t* sched, xce_sched_update_request_t* req) { - mylock_lock(&(sched->lock)); - dataq_enq(sched->pend_updates[0], &(req->link)); - mylock_unlock(&(sched->lock)); + mylock_lock(&(sched->core.lock)); + dataq_enq(sched->core.pend_updates[0], &(req->link)); + mylock_unlock(&(sched->core.lock)); } static int sched_on_jobs(xce_scheduler_t* sched, int njobs, xce_job_t** jobs) @@ -208,7 +208,7 @@ static int sched_on_job_array(xce_scheduler_t* sched, * (sched_add_job -> key_add -> -EEXIST), applied here per-element so * an already-materialized element is never re-queued. Not a new * dedup mechanism. */ - if (sched_get_job(sched, jobid) != NULL) { + if (sched_get_job(&(sched->core), jobid) != NULL) { continue; } diff --git a/src/scheduler/src/main.c b/src/scheduler/src/main.c index dcb51322..b348f7c4 100644 --- a/src/scheduler/src/main.c +++ b/src/scheduler/src/main.c @@ -79,8 +79,8 @@ static void stop_scheduler(xce_scheduler_t* sched) INFO("stop to scheduling jobs..."); - sched->stop = 1; - cbuf_tryput(&(sched->decisions), &decision); + sched->core.stop = 1; + cbuf_tryput(&(sched->core.decisions), &decision); } static void sig_handler(int signo) @@ -169,7 +169,7 @@ int main(int argc, char* argv[]) xce_job_alloc_t* allocs[1]; xce_job_dispatch_t jdispatch = { .njobs = 1, .allocs = allocs }; - cbuf_get(&(sched.decisions), &decision); + cbuf_get(&(sched.core.decisions), &decision); TRACE("%s decision = %p", __func__, decision); @@ -200,7 +200,7 @@ int main(int argc, char* argv[]) sched_end_session(&sched); - if (sched.stop) { + if (sched.core.stop) { break; } diff --git a/src/scheduler/src/plugin.c b/src/scheduler/src/plugin.c index 8e86d3cf..a5bc97ea 100644 --- a/src/scheduler/src/plugin.c +++ b/src/scheduler/src/plugin.c @@ -52,7 +52,7 @@ static xce_sched_job_t* legacy_picker_next_job(void* picker) * init_fn/fini_fn directly (below, and above in plugin_unregister()), and * begin_session_fn/end_session_fn are already driven by * plugin_begin_session()/plugin_end_session() (existing iteration over - * sched->plugins). Re-invoking them through the picker_iface hooks too + * sched->core.plugins). Re-invoking them through the picker_iface hooks too * would double-call plugin lifecycle callbacks. */ static int legacy_picker_init(void* picker, xce_sched_core_t* core) { @@ -156,9 +156,9 @@ int plugin_register(void* arg1, void* arg2) return rc; } - lock(sched->plugins, RTTI_LOCK_WRITE); - rc = key_add(sched->plugins, (void*)plugin->name, (void*)plugin); - unlock(sched->plugins); + lock(sched->core.plugins, RTTI_LOCK_WRITE); + rc = key_add(sched->core.plugins, (void*)plugin->name, (void*)plugin); + unlock(sched->core.plugins); if (rc >= 0) { plugin->plugin_id = rc; @@ -181,9 +181,9 @@ void plugin_unregister(void* arg) return; } - lock(sched->plugins, RTTI_LOCK_WRITE); - key_del(sched->plugins, plugin->name); - unlock(sched->plugins); + lock(sched->core.plugins, RTTI_LOCK_WRITE); + key_del(sched->core.plugins, plugin->name); + unlock(sched->core.plugins); if (plugin->fini_fn != NULL) { plugin->fini_fn(plugin); @@ -200,7 +200,7 @@ int plugin_config_update(void* arg, xce_conf4sched_t* config) return -EINVAL; } - rc = iter_make(&it, sched->plugins); + rc = iter_make(&it, sched->core.plugins); if (rc < 0) { return rc; } @@ -237,7 +237,7 @@ int plugin_job_update(void* arg, xce_sched_job_t* job) return -EINVAL; } - rc = iter_make(&it, sched->plugins); + rc = iter_make(&it, sched->core.plugins); if (rc < 0) { return rc; } @@ -274,7 +274,7 @@ int plugin_queue_update(void* arg, xce_sched_queue_t* queue) return -EINVAL; } - rc = iter_make(&it, sched->plugins); + rc = iter_make(&it, sched->core.plugins); if (rc < 0) { return rc; } @@ -311,7 +311,7 @@ int plugin_worker_update(void* arg, xce_sched_worker_t* worker) return -EINVAL; } - rc = iter_make(&it, sched->plugins); + rc = iter_make(&it, sched->core.plugins); if (rc < 0) { return rc; } @@ -348,7 +348,7 @@ int plugin_begin_session(void* arg) return -EINVAL; } - rc = iter_make(&it, sched->plugins); + rc = iter_make(&it, sched->core.plugins); if (rc < 0) { return rc; } @@ -385,7 +385,7 @@ int plugin_end_session(void* arg) return -EINVAL; } - rc = iter_make(&it, sched->plugins); + rc = iter_make(&it, sched->core.plugins); if (rc < 0) { return rc; } @@ -422,7 +422,7 @@ int plugin_commit_decision(void* arg1, xce_decision_t* decision) return -EINVAL; } - rc = iter_make(&it, sched->plugins); + rc = iter_make(&it, sched->core.plugins); if (rc < 0) { return rc; } @@ -459,7 +459,7 @@ int plugin_job_state_change(void* arg, xce_sched_job_t* job, uint32_t state) return -EINVAL; } - rc = iter_make(&it, sched->plugins); + rc = iter_make(&it, sched->core.plugins); if (rc < 0) { return rc; } @@ -497,7 +497,7 @@ int plugin_job_rusage_update(void* arg, xce_sched_job_t* job, return -EINVAL; } - rc = iter_make(&it, sched->plugins); + rc = iter_make(&it, sched->core.plugins); if (rc < 0) { return rc; } @@ -534,7 +534,7 @@ int plugin_job_election(void* arg, xce_sched_job_t* job) return -EINVAL; } - rc = iter_make(&it, sched->plugins); + rc = iter_make(&it, sched->core.plugins); if (rc < 0) { return rc; } @@ -585,7 +585,7 @@ int plugin_prepare_allocation(void* arg, xce_sched_ctx_t* ctx) xvec_idval_init(&(ctx->alloc_req.host), 8); } - rc = iter_make(&it, sched->plugins); + rc = iter_make(&it, sched->core.plugins); if (rc < 0) { return rc; } diff --git a/src/scheduler/src/plugins/fcfs.c b/src/scheduler/src/plugins/fcfs.c index 50dbc60a..535ba835 100644 --- a/src/scheduler/src/plugins/fcfs.c +++ b/src/scheduler/src/plugins/fcfs.c @@ -211,7 +211,7 @@ static void* fcfs_elect_job(xce_sched_plugin_t* plugin) pthread_testcancel(); - cbuf_get(&(sched->wbuf), &go); + cbuf_get(&(sched->core.wbuf), &go); privdata->new_session = 0; for (it = queues->head; it != NULL; it = it->next) { @@ -236,7 +236,7 @@ static void* fcfs_elect_job(xce_sched_plugin_t* plugin) fmt_jobid(jobidString, sizeof(jobidString), jobid); TRACE("%s jobid %s", __func__, jobidString); - cbuf_put(&(sched->jbuf), &jobid); + cbuf_put(&(sched->core.jbuf), &jobid); } if (privdata->new_session) { @@ -251,7 +251,7 @@ static void* fcfs_elect_job(xce_sched_plugin_t* plugin) /* end of scheduling session */ jobid = 0; - cbuf_put(&(sched->jbuf), &jobid); + cbuf_put(&(sched->core.jbuf), &jobid); sleep(1); } @@ -310,7 +310,7 @@ static int fcfs_del_job(xce_sched_plugin_t* plugin, uint64_t jobid) return 0; } - job = sched_get_job(plugin->sched, jobid); + job = sched_get_job(&(((xce_scheduler_t*)plugin->sched)->core), jobid); if (job == NULL) { return 0; } diff --git a/src/scheduler/src/plugins/limit.c b/src/scheduler/src/plugins/limit.c index 326b3177..5bcffd31 100644 --- a/src/scheduler/src/plugins/limit.c +++ b/src/scheduler/src/plugins/limit.c @@ -62,8 +62,8 @@ static int limit_update_config(xce_sched_plugin_t* plugin, int rc; xce_limit_t* limit; - rc = build_limit(&limit, (sched->ugmgr), (sched->qgmgr), (sched->pgmgr), - (sched->hgmgr), limits[i]); + rc = build_limit(&limit, (sched->core.ugmgr), (sched->core.qgmgr), + (sched->core.pgmgr), (sched->core.hgmgr), limits[i]); if (rc < 0) { return rc; } @@ -168,7 +168,7 @@ static int match_job_global_limit_(xce_sched_plugin_t* plugin, return -EINVAL; } - if (is_host_based_limit(&(sched->resmgr), limit) > 0) { + if (is_host_based_limit(&(sched->core.resmgr), limit) > 0) { return 0; } @@ -195,7 +195,7 @@ static int match_job_global_limit_(xce_sched_plugin_t* plugin, FREEUP(vresname); if (vres == NULL) { - rc = instantiate_limit(&vres, &(sched->resmgr), limit, obj->user, + rc = instantiate_limit(&vres, &(sched->core.resmgr), limit, obj->user, obj->queue, obj->project); if (rc < 0) { return rc; @@ -210,7 +210,7 @@ static int match_job_global_limit_(xce_sched_plugin_t* plugin, } } - rc = update_limit_res_value(&(sched->sharedInst), vres); + rc = update_limit_res_value(&(sched->core.sharedInst), vres); if (rc >= 0) { rc = xlist_store(&(obj->glimits), vres); } @@ -232,7 +232,7 @@ static int limit_job_election(void* arg, xce_sched_job_t* job) fmt_jobid(jobidString, sizeof(jobidString), job->jobid); TRACE("%s job %s plugin %s", __func__, jobidString, plugin->name); - if (obj->configver >= sched->configver) { + if (obj->configver >= sched->core.configver) { return 0; } @@ -267,7 +267,7 @@ static int limit_job_election(void* arg, xce_sched_job_t* job) iter_destroy(&it); - obj->configver = sched->configver; + obj->configver = sched->core.configver; return rc; } @@ -305,7 +305,7 @@ static int match_job_host_limit_(xce_sched_plugin_t* plugin, return 0; } - if (is_host_based_limit(&(sched->resmgr), limit) <= 0) { + if (is_host_based_limit(&(sched->core.resmgr), limit) <= 0) { return 0; } @@ -344,7 +344,7 @@ static int match_job_host_limit_(xce_sched_plugin_t* plugin, FREEUP(vresname); if (vres == NULL) { - rc = instantiate_limit(&vres, &(sched->resmgr), limit, obj->user, + rc = instantiate_limit(&vres, &(sched->core.resmgr), limit, obj->user, obj->queue, obj->project); if (rc < 0) { return rc; diff --git a/src/scheduler/src/plugins/parjob_allocator.c b/src/scheduler/src/plugins/parjob_allocator.c index 9e731668..93be6fe5 100644 --- a/src/scheduler/src/plugins/parjob_allocator.c +++ b/src/scheduler/src/plugins/parjob_allocator.c @@ -19,7 +19,7 @@ static int parjob_allocator_init(xce_sched_plugin_t* plugin) { xce_scheduler_t* sched = (xce_scheduler_t*)plugin->sched; - sched->parjob_allocator = plugin; + sched->core.parjob_allocator = plugin; return 1; } diff --git a/src/scheduler/src/plugins/seqjob_allocator.c b/src/scheduler/src/plugins/seqjob_allocator.c index 4d3e1018..112181aa 100644 --- a/src/scheduler/src/plugins/seqjob_allocator.c +++ b/src/scheduler/src/plugins/seqjob_allocator.c @@ -20,7 +20,7 @@ static int seqjob_allocator_init(xce_sched_plugin_t* plugin) { xce_scheduler_t* sched = (xce_scheduler_t*)plugin->sched; - sched->seqjob_allocator = plugin; + sched->core.seqjob_allocator = plugin; return 1; } @@ -74,18 +74,18 @@ static int seqjob_prepare_allocation(void* arg, xce_sched_ctx_t* ctx) if (ctx->worker == NULL) { ctx->nhosts = 1; ctx->nslots = 1; - calc_wanted_global_resvals(&(ctx->alloc_req.shared), sched->resmgr, + calc_wanted_global_resvals(&(ctx->alloc_req.shared), sched->core.resmgr, &(primary->rusages), PER_TASK, 1, 1); } else { - calc_wanted_host_resvals(&(ctx->alloc_req.host), sched->resmgr, + calc_wanted_host_resvals(&(ctx->alloc_req.host), sched->core.resmgr, &(primary->rusages), PER_TASK, 1, 1); } /* add slots into requested resources */ - lock(sched->resmgr, RTTI_LOCK_READ); - rc = key_find(sched->resmgr, (void*)"slots", (void**)&r); - unlock(sched->resmgr); - + lock(sched->core.resmgr, RTTI_LOCK_READ); + rc = key_find(sched->core.resmgr, (void*)"slots", (void**)&r); + unlock(sched->core.resmgr); + if (rc < 0) { return rc; } @@ -139,7 +139,7 @@ static int my_alloc(xce_sched_plugin_t* plugin, xce_sched_ctx_t* ctx) for (i = 0; i < ctx->alloc_req.shared.count; i++) { xce_resource_t* r = NULL; - rc = vget(sched->resmgr, ((int*)ctx->unmeet.idxs.data)[i], + rc = vget(sched->core.resmgr, ((int*)ctx->unmeet.idxs.data)[i], (void**)&r); if (rc <= 0) { continue; diff --git a/src/scheduler/src/resource.c b/src/scheduler/src/resource.c index cd00066b..46a7804c 100644 --- a/src/scheduler/src/resource.c +++ b/src/scheduler/src/resource.c @@ -121,7 +121,7 @@ static int unsatisfy_(xce_scheduler_t* sched, int nres, int* idxs, float* total, xce_resource_t* r = NULL; float need = inuse[i] + uncomitted[i] + wanted[i] - total[i]; - vget(sched->resmgr, idxs[i], (void**)&r); + vget(sched->core.resmgr, idxs[i], (void**)&r); TRACE("%s unmet resource %d (%s) shortfall %f", __func__, idxs[i], r ? r->name : "?", need); @@ -214,12 +214,12 @@ int sched_meet_host_selection(xce_scheduler_t* sched, xce_sched_worker_t* w, } /* Fast path: use session-level memoisation context when available. */ - if (sched->sel_eval_ctx != NULL) { + if (sched->core.sel_eval_ctx != NULL) { /* Reset memoisation bitmap only when moving to a different host. */ - if (sched->sel_eval_ctx->host != w->host_eval_ctx) - select_eval_ctx_reset(sched->sel_eval_ctx, w->host_eval_ctx); + if (sched->core.sel_eval_ctx->host != w->host_eval_ctx) + select_eval_ctx_reset(sched->core.sel_eval_ctx, w->host_eval_ctx); - int rc = eval_select(s->root, sched->sel_eval_ctx); + int rc = eval_select(s->root, sched->core.sel_eval_ctx); return (rc > 0) ? 1 : 0; } @@ -252,13 +252,13 @@ static int sched_check_shared_satisfy_(xce_scheduler_t* sched, xvec_t* total, return rc; } - rc = get_resvals(&(sched->sharedInst), nres, (int*)wanted->idxs.data, + rc = get_resvals(&(sched->core.sharedInst), nres, (int*)wanted->idxs.data, (float*)total->data); if (rc < 0) { return rc; } - rc = get_resvals(&(sched->sharedInUse), nres, (int*)wanted->idxs.data, + rc = get_resvals(&(sched->core.sharedInUse), nres, (int*)wanted->idxs.data, (float*)inuse->data); if (rc < 0) { return rc; @@ -401,9 +401,9 @@ static int sched_update_res_accounts(xce_scheduler_t* sched, zarray_t* array, xce_resource_t* r = NULL; int rc = 0; - lock(sched->resmgr, RTTI_LOCK_READ); - rc = key_find(sched->resmgr, (void*)acct->resname, (void**)&r); - unlock(sched->resmgr); + lock(sched->core.resmgr, RTTI_LOCK_READ); + rc = key_find(sched->core.resmgr, (void*)acct->resname, (void**)&r); + unlock(sched->core.resmgr); if (rc < 0) { return rc; @@ -413,14 +413,15 @@ static int sched_update_res_accounts(xce_scheduler_t* sched, zarray_t* array, if (r == NULL) { int rc; - rc = add_vres(&r, sched->resmgr, acct->resname, global); + rc = add_vres(&r, sched->core.resmgr, acct->resname, global); if (rc < 0) { return rc; } - lock(sched->resmgr, RTTI_LOCK_READ); - rc = key_find(sched->resmgr, (void*)acct->resname, (void**)&r); - unlock(sched->resmgr); + lock(sched->core.resmgr, RTTI_LOCK_READ); + rc = key_find(sched->core.resmgr, (void*)acct->resname, + (void**)&r); + unlock(sched->core.resmgr); if (rc < 0) { return rc; } @@ -507,7 +508,7 @@ static int sched_try_alloc_on_host_(xce_sched_worker_t* w, va_list ap) for (i = 0; i < ctx->unmeet.count; i++) { xce_resource_t* r = NULL; - rc = vget(sched->resmgr, ((int*)ctx->unmeet.idxs.data)[i], + rc = vget(sched->core.resmgr, ((int*)ctx->unmeet.idxs.data)[i], (void**)&r); if (rc < 0) { continue; @@ -540,7 +541,7 @@ static int sched_update_account(xce_scheduler_t* sched, { int i; - sched_update_res_accounts(sched, &(sched->sharedInUse), 1, + sched_update_res_accounts(sched, &(sched->core.sharedInUse), 1, account->nres_shared, account->accounts_shared, charge); for (i = 0; i < account->nhosts; i++) { @@ -551,7 +552,7 @@ static int sched_update_account(xce_scheduler_t* sched, continue; } - w = sched_get_worker(sched, haccount->host); + w = sched_get_worker(&(sched->core), haccount->host); if (w == NULL) { continue; } diff --git a/src/scheduler/src/schedobj.c b/src/scheduler/src/schedobj.c index 62cf82a7..3a459760 100644 --- a/src/scheduler/src/schedobj.c +++ b/src/scheduler/src/schedobj.c @@ -166,9 +166,9 @@ xce_sched_job_t* make_sched_job(void* arg, xce_job_t* j) job->queue_time = j->queue_time; job->start_time = j->start_time; job->state = j->state; - job->user = sched_get_user(sched, j->submit->user); - job->queue = sched_get_queue(sched, j->submit->queue); - job->project = sched_get_project(sched, j->submit->project_name); + job->user = sched_get_user(&(sched->core), j->submit->user); + job->queue = sched_get_queue(&(sched->core), j->submit->queue); + job->project = sched_get_project(&(sched->core), j->submit->project_name); job->resreq = (xce_resreq_t*)calloc(1, sizeof(*(job->resreq))); if (job->resreq == NULL) { return NULL; @@ -179,7 +179,7 @@ xce_sched_job_t* make_sched_job(void* arg, xce_job_t* j) if (NULLSTR(resreq_str)) { resreq_str = "type == any"; } - rc = parse_resreq(job->resreq, resreq_str, sched->resmgr, &errmsg); + rc = parse_resreq(job->resreq, resreq_str, sched->core.resmgr, &errmsg); } if (rc < 0) { ERROR("%s parse_resreq failed, %s", __func__, errmsg); @@ -300,7 +300,7 @@ xce_sched_worker_t* make_sched_worker(void* arg, xce_worker_t* w) continue; } - if (key_find(sched->resmgr, (void*)nv->name, (void**)&r) < 0 + if (key_find(sched->core.resmgr, (void*)nv->name, (void**)&r) < 0 || r == NULL) { WARN("%s unknown resource '%s' in worker resvals, skipping", __func__, nv->name); @@ -326,23 +326,6 @@ void kill_sched_user(void* arg) FREEUP(user); } -xce_sched_user_t* make_sched_user(void* arg, char* name) -{ - xce_scheduler_t* sched = (xce_scheduler_t*)arg; - xce_sched_user_t* user = NULL; - - user = (xce_sched_user_t*)calloc(1, sizeof(*user)); - if (user == NULL) { - return NULL; - } - - user->name = makeString(name); - - sched_add_user(sched, user); - - return user; -} - void kill_sched_project(void* arg) { xce_sched_project_t* project = (xce_sched_project_t*)arg; @@ -355,42 +338,15 @@ void kill_sched_project(void* arg) FREEUP(project); } -xce_sched_project_t* make_sched_project(void* arg, char* name) -{ - xce_scheduler_t* sched = (xce_scheduler_t*)arg; - xce_sched_project_t* project = NULL; - - project = (xce_sched_project_t*)calloc(1, sizeof(*project)); - if (project == NULL) { - return NULL; - } - - if ((name == NULL) || (strlen(name) == 0)) { - project->name = makeString("default"); - } else { - project->name = makeString(name); - } - - /* Req 2 AC (scheduler-core spec, task 1.3): scaffolding only -- no - * quota/hierarchy inputs exist yet, so priority defaults to 0 - * (calloc() above already zeroes it; explicit here for - * documentation/testability). */ - project->priority = 0; - - sched_add_project(sched, project); - - return project; -} - /* * sched_set_default_job_pick_plugin - Req 2 AC 1 (scheduler-core spec, - * task 1.3). Applies the "builtin_fcfs" default to sched->job_pick_plugin + * task 1.3). Applies the "builtin_fcfs" default to sched->core.job_pick_plugin * when it is unset (NULL). Idempotent: a name already present (whether * previously defaulted or set by a future config-driven path) is left * untouched, so this is safe to call defensively. * * This function only assigns the *name* -- it never touches the plugin - * registry (sched->plugins) and never calls plugin_register()/key_find(). + * registry (sched->core.plugins) and never calls plugin_register()/key_find(). * Resolving the configured name against a registered picker plugin, and * failing scheduler startup loudly when the name is not registered * (Req 2 AC 5), is task 2.4's job. Kept as a separate, standalone function @@ -404,8 +360,8 @@ void sched_set_default_job_pick_plugin(xce_scheduler_t* sched) return; } - if (sched->job_pick_plugin == NULL) { - sched->job_pick_plugin = makeString("builtin_fcfs"); + if (sched->core.job_pick_plugin == NULL) { + sched->core.job_pick_plugin = makeString("builtin_fcfs"); } } @@ -417,11 +373,11 @@ void sched_set_default_job_pick_plugin(xce_scheduler_t* sched) * Kept here (rather than in scheduler.c, where both its callers -- * sched_init() and sched_job_picker() -- live) for the same testability * reason sched_set_default_job_pick_plugin() was: this function only needs - * sched->job_pick_plugin and sched->plugins (the plugin registry kvmgr), - * not the network/xcecli dependencies sched_init() otherwise pulls in, so - * it can be exercised directly by a white-box test of this translation - * unit (mirroring test_picker_config_fields.c's own #include "schedobj.c" - * convention) without linking scheduler.c. + * sched->core.job_pick_plugin and sched->core.plugins (the plugin registry + * kvmgr), not the network/xcecli dependencies sched_init() otherwise pulls + * in, so it can be exercised directly by a white-box test of this + * translation unit (mirroring test_picker_config_fields.c's own #include + * "schedobj.c" convention) without linking scheduler.c. */ int sched_resolve_job_picker(xce_scheduler_t* sched, xce_sched_plugin_t** out) { @@ -434,13 +390,14 @@ int sched_resolve_job_picker(xce_scheduler_t* sched, xce_sched_plugin_t** out) *out = NULL; - if (sched->job_pick_plugin == NULL) { + if (sched->core.job_pick_plugin == NULL) { return -EINVAL; } - lock(sched->plugins, RTTI_LOCK_READ); - rc = key_find(sched->plugins, sched->job_pick_plugin, (void**)&plugin); - unlock(sched->plugins); + lock(sched->core.plugins, RTTI_LOCK_READ); + rc = key_find(sched->core.plugins, sched->core.job_pick_plugin, + (void**)&plugin); + unlock(sched->core.plugins); if (rc <= 0) { return (rc < 0) ? rc : -ENOENT; @@ -594,7 +551,7 @@ int sched_update_existing_queue(xce_scheduler_t* sched, xce_queue_t* updated) return -EINVAL; } - q = sched_get_queue(sched, updated->queue); + q = sched_get_queue(&(sched->core), updated->queue); if (q == NULL) { return -ENOENT; } @@ -636,7 +593,7 @@ int sched_update_existing_job(xce_scheduler_t* sched, xce_job_t* updated) return -EINVAL; } - job = sched_get_job(sched, updated->jobid); + job = sched_get_job(&(sched->core), updated->jobid); if (job == NULL) { return -ENOENT; } @@ -650,8 +607,8 @@ int sched_update_existing_job(xce_scheduler_t* sched, xce_job_t* updated) * per-job priority, so job priority follows the (possibly new) queue, * consistent with the queue-derived priority model. */ if (updated->submit->queue != NULL) { - xce_sched_queue_t* q = - sched_get_queue(sched, updated->submit->queue); + xce_sched_queue_t* q + = sched_get_queue(&(sched->core), updated->submit->queue); if (q != NULL) { job->queue = q; job->priority = q->priority; @@ -669,8 +626,8 @@ int sched_update_existing_job(xce_scheduler_t* sched, xce_job_t* updated) } char* errmsg = NULL; - int rc = parse_resreq(newreq, updated->submit->resreq, - sched->resmgr, &errmsg); + int rc = parse_resreq(newreq, updated->submit->resreq, + sched->core.resmgr, &errmsg); if (rc < 0) { ERROR("%s parse_resreq failed: %s", __func__, errmsg ? errmsg : "(null)"); @@ -707,7 +664,7 @@ int sched_apply_job_rusage(xce_scheduler_t* sched, xce_job_rusages_t* rusages) return -EINVAL; } - job = sched_get_job(sched, rusages->jobid); + job = sched_get_job(&(sched->core), rusages->jobid); if (job == NULL) { return -ENOENT; } diff --git a/src/scheduler/src/scheduler.c b/src/scheduler/src/scheduler.c index b3cf96ae..85fdcf91 100644 --- a/src/scheduler/src/scheduler.c +++ b/src/scheduler/src/scheduler.c @@ -22,17 +22,19 @@ static int sched_session_active(xce_scheduler_t* sched) { - return atomic_load_explicit(&sched->session_active, memory_order_acquire); + return atomic_load_explicit(&sched->core.session_active, + memory_order_acquire); } static int sched_eval_in_progress(xce_scheduler_t* sched) { - return atomic_load_explicit(&sched->eval_in_progress, memory_order_acquire); + return atomic_load_explicit(&sched->core.eval_in_progress, + memory_order_acquire); } static int sched_staging_merge_complete(xce_scheduler_t* sched) { - return atomic_load_explicit(&sched->staging_merge_complete, + return atomic_load_explicit(&sched->core.staging_merge_complete, memory_order_acquire); } @@ -42,11 +44,11 @@ static int sched_staging_merge_complete(xce_scheduler_t* sched) * * Replaces the hardcoded "xce_sched_fcfs" literal (Req 1 AC 6) with * sched_resolve_job_picker() (schedobj.c), which looks up - * sched->job_pick_plugin through the same plugin_register()/key_find() - * mechanism every plugin category uses. sched_init() already performed - * this same resolution synchronously before this thread was even created - * (so an unregistered/non-picker name fails scheduler startup, not this - * thread quietly) -- re-resolving here is the redundant "one + * sched->core.job_pick_plugin through the same plugin_register()/ + * key_find() mechanism every plugin category uses. sched_init() already + * performed this same resolution synchronously before this thread was even + * created (so an unregistered/non-picker name fails scheduler startup, not + * this thread quietly) -- re-resolving here is the redundant "one * implementation, two call sites" the scheduler.h doc comment describes, * not duplicated logic. * @@ -80,7 +82,9 @@ static void* sched_job_picker(void* arg) rc = sched_resolve_job_picker(sched, &job_picker); if (rc < 0) { ERROR("%s failed to resolve job picker plugin '%s' (rc=%d)", __func__, - sched->job_pick_plugin ? sched->job_pick_plugin : "(null)", rc); + sched->core.job_pick_plugin ? sched->core.job_pick_plugin + : "(null)", + rc); errno = -rc; return NULL; @@ -102,7 +106,7 @@ static void* sched_job_picker(void* arg) * cancel point for the picker thread between sessions, the same * property plugins/fcfs.c's own fcfs_elect_job() already relies on * for its identical cbuf_get(&sched->wbuf) call. */ - cbuf_get(&(sched->wbuf), &go); + cbuf_get(&(sched->core.wbuf), &go); job_picker->picker_iface->begin_session(job_picker); @@ -117,12 +121,12 @@ static void* sched_job_picker(void* arg) } jobid = job->jobid; - cbuf_put(&(sched->jbuf), &jobid); + cbuf_put(&(sched->core.jbuf), &jobid); } /* end of scheduling session */ jobid = 0; - cbuf_put(&(sched->jbuf), &jobid); + cbuf_put(&(sched->core.jbuf), &jobid); job_picker->picker_iface->end_session(job_picker); } @@ -135,7 +139,7 @@ static int sched_make_seqjob_allocation(xce_scheduler_t* sched, { xce_sched_plugin_t* plugin = NULL; - plugin = sched->seqjob_allocator; + plugin = sched->core.seqjob_allocator; if (plugin->job_alloc_fn == NULL) { ERROR("%s plugin %s has no allocation function", __func__, plugin->name); @@ -150,7 +154,7 @@ static int sched_make_parjob_allocation(xce_scheduler_t* sched, { xce_sched_plugin_t* plugin = NULL; - plugin = sched->parjob_allocator; + plugin = sched->core.parjob_allocator; if (plugin->job_alloc_fn == NULL) { ERROR("%s plugin %s has no allocation function", __func__, plugin->name); @@ -308,7 +312,9 @@ static void* sched_allocator(void* arg) rc = sched_resolve_job_picker(sched, &job_picker); if (rc < 0) { ERROR("%s failed to resolve job picker plugin '%s' (rc=%d)", __func__, - sched->job_pick_plugin ? sched->job_pick_plugin : "(null)", rc); + sched->core.job_pick_plugin ? sched->core.job_pick_plugin + : "(null)", + rc); errno = -rc; return NULL; @@ -323,15 +329,15 @@ static void* sched_allocator(void* arg) pthread_testcancel(); - cbuf_get(&(sched->jbuf), &jobid); + cbuf_get(&(sched->core.jbuf), &jobid); if (jobid == 0) { /* end of scheduling session */ decision = NULL; - cbuf_put(&(sched->decisions), &decision); + cbuf_put(&(sched->core.decisions), &decision); continue; } - job = sched_get_job(sched, jobid); + job = sched_get_job(&sched->core, jobid); if (job == NULL) { fmt_jobid(jobidString, sizeof(jobidString), jobid); TRACE("%s job %s can not be found", __func__, jobidString); @@ -378,7 +384,7 @@ static void* sched_allocator(void* arg) if (decision != NULL) { commit_decision(sched, decision); - cbuf_put(&(sched->decisions), &decision); + cbuf_put(&(sched->core.decisions), &decision); } } @@ -469,62 +475,17 @@ static int sched_register_plugins(xce_scheduler_t* sched) return 0; } -static int sched_plugin_init(xce_scheduler_t* sched) -{ - kvmgr_params_t params = { .key_type = KVMGR_KEY_STRING, - .nslots = 16, - .delta = 8, - .datakiller = plugin_unregister }; - - sched->plugins = makeInstance(&xkvmgr_type, ¶ms); - if (sched->plugins == NULL) { - return -errno; - } - - return 0; -} - int sched_init(xce_scheduler_t* sched, xce_client_config_t* config, const xce_cluster_config_t* cluster_cfg, xce_request_handler_t handler) { - int rc = 0; - int ndecisions = 5; /* FIXME: make this configurable */ - int njobs = 1; /* FIXME: make this configurable */ + int rc = 0; /* Req 1 AC 6 / Req 2 AC 5 (scheduler-core spec, task 2.4): resolved * picker plugin, validated synchronously below before the picker * thread is created. Only used for the validation side-effect here -- * the picker thread (sched_job_picker) re-resolves it independently at * thread start (scheduler.h's sched_resolve_job_picker doc comment). */ xce_sched_plugin_t* resolved_picker = NULL; - kvmgr_params_t jobs_params = { .key_type = KVMGR_KEY_INT, - .nslots = 5381, /* FIXME */ - .delta = 1024, - .datakiller = kill_sched_job }; - kvmgr_params_t queues_params - = { .key_type = KVMGR_KEY_STRING, - .nslots = 59, - .delta = 16, - .datakiller = (destroy_fn_t)kill_sched_queue }; - kvmgr_params_t workers_params - = { .key_type = KVMGR_KEY_STRING, - .nslots = 5381, - .delta = 64, - .datakiller = (destroy_fn_t)kill_sched_worker }; - kvmgr_params_t users_params - = { .key_type = KVMGR_KEY_STRING, - .nslots = 5381, - .delta = 64, - .datakiller = (destroy_fn_t)kill_sched_user }; - kvmgr_params_t projects_params - = { .key_type = KVMGR_KEY_STRING, - .nslots = 31, - .delta = 8, - .datakiller = (destroy_fn_t)kill_sched_project }; - kvmgr_params_t resmgr_params = { .key_type = KVMGR_KEY_STRING, - .nslots = 64, - .delta = 8, - .datakiller = NULL }; if ((sched == NULL) || (config == NULL) || (handler == NULL)) { return -EINVAL; @@ -536,47 +497,30 @@ int sched_init(xce_scheduler_t* sched, xce_client_config_t* config, sched->cluster_cfg = *cluster_cfg; } - resreq_init(1); /* enable CSE intern table for scheduler process */ + /* Req 5.1/5.2 (scheduler-core spec, task 4.1): all scheduling-state + * init (jobs/queues/workers/users/projects/resources kvmgrs, the + * plugin registry, the three cbufs, nextid/configver, the CSE/xstring + * intern tables) is now xce_sched_core_init()'s job (core.c) -- zero + * controller I/O, exactly the property this task's Observable checks. + * pend_updates' storage lives on xce_sched_core_t too, but its + * dataq_init killer callback (sched_pend_update_killer, above) stays + * driver-owned (see core.c's own file header comment for why), so it + * is wired here, immediately after. */ + rc = xce_sched_core_init(&(sched->core)); + if (rc < 0) { + return rc; + } - sched->nextid = 1; - sched->configver = 0; + sched->core.pend_updates[0] = (dataq_t*)calloc(1, sizeof(dataq_t)); + dataq_init(sched->core.pend_updates[0], sched_pend_update_killer); - xstring_table_init(1024 * 1024); - - mylock_init(&(sched->lock)); - cbuf_init(&(sched->wbuf), 1, sizeof(int)); - cbuf_init(&(sched->jbuf), njobs, sizeof(uint64_t)); - cbuf_init(&(sched->decisions), ndecisions, sizeof(xce_decision_t*)); - - sched->pend_updates[0] = (dataq_t*)calloc(1, sizeof(dataq_t)); - dataq_init(sched->pend_updates[0], sched_pend_update_killer); - - sched->pend_updates[1] = (dataq_t*)calloc(1, sizeof(dataq_t)); - dataq_init(sched->pend_updates[1], sched_pend_update_killer); - - sched->resmgr = makeInstance(&xkvmgr_type, &resmgr_params); - - groupmgr_init(&(sched->ugmgr), 64, 64); - groupmgr_init(&(sched->qgmgr), 16, 16); - groupmgr_init(&(sched->pgmgr), 16, 16); - groupmgr_init(&(sched->hgmgr), 64, 64); - - sched_plugin_init(sched); - - sched->jobs = makeInstance(&xkvmgr_type, &jobs_params); - sched->queues = makeInstance(&xkvmgr_type, &queues_params); - sched->workers = makeInstance(&xkvmgr_type, &workers_params); - sched->users = makeInstance(&xkvmgr_type, &users_params); - sched->projects = makeInstance(&xkvmgr_type, &projects_params); - - zarray_init(&(sched->sharedInst), 128, - (zarray_item_killer_t)kill_sched_resval); - zarray_init(&(sched->sharedInUse), 128, NULL); + sched->core.pend_updates[1] = (dataq_t*)calloc(1, sizeof(dataq_t)); + dataq_init(sched->core.pend_updates[1], sched_pend_update_killer); sched_register_plugins(sched); /* Req 1 AC 6 / Req 2 AC 1, AC 5, Req 3 AC 1 (scheduler-core spec, task - * 2.4): default sched->job_pick_plugin to "builtin_fcfs" when the + * 2.4): default sched->core.job_pick_plugin to "builtin_fcfs" when the * operator left it unset (task 1.3's sched_set_default_job_pick_plugin(), * deliberately not wired into sched_init() before this task -- see * tasks.md's Implementation Notes), then resolve and validate the @@ -593,7 +537,9 @@ int sched_init(xce_scheduler_t* sched, xce_client_config_t* config, ERROR("%s configured picker plugin '%s' failed to resolve as a " "valid registered picker (rc=%d)", __func__, - sched->job_pick_plugin ? sched->job_pick_plugin : "(null)", rc); + sched->core.job_pick_plugin ? sched->core.job_pick_plugin + : "(null)", + rc); return rc; } @@ -666,12 +612,14 @@ int sched_init(xce_scheduler_t* sched, xce_client_config_t* config, } } - rc = pthread_create(&(sched->jobpicker), NULL, &sched_job_picker, sched); + rc = pthread_create(&(sched->core.jobpicker), NULL, &sched_job_picker, + sched); if (rc < 0) { return rc; } - rc = pthread_create(&(sched->allocator), NULL, &sched_allocator, sched); + rc = pthread_create(&(sched->core.allocator), NULL, &sched_allocator, + sched); return rc; } @@ -685,35 +633,23 @@ void sched_fini(xce_scheduler_t* sched) return; } - pthread_cancel(sched->jobpicker); - pthread_join(sched->jobpicker, &res); + pthread_cancel(sched->core.jobpicker); + pthread_join(sched->core.jobpicker, &res); - pthread_cancel(sched->allocator); + pthread_cancel(sched->core.allocator); /* try to wake up allocator */ - cbuf_tryput(&(sched->jbuf), &jobid); - pthread_join(sched->allocator, &res); + cbuf_tryput(&(sched->core.jbuf), &jobid); + pthread_join(sched->core.allocator, &res); xcecli_wakeup(sched->apictx, (int)-EINTR); xcecli_stop(sched->apictx); xcecli_ctx_fini(sched->apictx); - cbuf_fini(&(sched->decisions)); - cbuf_fini(&(sched->jbuf)); - cbuf_fini(&(sched->wbuf)); - - dataq_fini(sched->pend_updates[0]); - dataq_fini(sched->pend_updates[1]); - - destroy(sched->resmgr); - - destroy(sched->ugmgr); - destroy(sched->qgmgr); - destroy(sched->pgmgr); - destroy(sched->hgmgr); - - resreq_fini(); - - mylock_fini(&(sched->lock)); + /* Req 5.1/5.2 (scheduler-core spec, task 4.1): the remaining teardown + * (cbufs, pend_updates dataqs, resmgr/group managers, the CSE intern + * table, the core lock) is xce_sched_core_fini()'s job (core.c) -- + * same order as before this task, just relocated. */ + xce_sched_core_fini(&(sched->core)); return; } @@ -721,7 +657,7 @@ void sched_fini(xce_scheduler_t* sched) static int sched_init_resources(xce_scheduler_t* sched) { int i; - xce_conf4sched_t* config = sched->config; + xce_conf4sched_t* config = sched->core.config; for (i = 0; i < config->n_resources; i++) { xce_resource_t* r = config->resources[i]; @@ -731,7 +667,7 @@ static int sched_init_resources(xce_scheduler_t* sched) continue; } - r1 = makeResource(sched->resmgr, r); + r1 = makeResource(sched->core.resmgr, r); if (r1 == NULL) { ERROR("%s add resource %s failed", __func__, @@ -745,8 +681,8 @@ static int sched_init_sharedInst(xce_scheduler_t* sched) { int i; - for (i = 0; i < sched->config->nresvals; i++) { - xce_name_val_t* nv = sched->config->resvals[i]; + for (i = 0; i < sched->core.config->nresvals; i++) { + xce_name_val_t* nv = sched->core.config->resvals[i]; xce_resource_t* r = NULL; xce_val_t* val = NULL; @@ -754,7 +690,7 @@ static int sched_init_sharedInst(xce_scheduler_t* sched) continue; } - if (key_find(sched->resmgr, (void*)nv->name, (void**)&r) < 0 + if (key_find(sched->core.resmgr, (void*)nv->name, (void**)&r) < 0 || r == NULL) { WARN("%s unknown resource '%s' in resvals, skipping", __func__, nv->name); @@ -762,7 +698,7 @@ static int sched_init_sharedInst(xce_scheduler_t* sched) } MOVE_PTR(val, nv->val); - zarray_put(&(sched->sharedInst), r->idx, val); + zarray_put(&(sched->core.sharedInst), r->idx, val); } return 0; @@ -771,13 +707,13 @@ static int sched_init_sharedInst(xce_scheduler_t* sched) static int sched_init_queues(xce_scheduler_t* sched) { int i; - xce_conf4sched_t* config = sched->config; + xce_conf4sched_t* config = sched->core.config; for (i = 0; i < config->n_queues; i++) { xce_sched_queue_t* q = make_sched_queue(sched, config->queues[i]); plugin_queue_update(sched, q); - sched_add_queue(sched, q); + sched_add_queue(&(sched->core), q); } return 0; @@ -785,13 +721,13 @@ static int sched_init_queues(xce_scheduler_t* sched) int sched_init_config(xce_scheduler_t* sched, xce_conf4sched_t* config) { plugin_config_update(sched, config); - sched->config = config; + sched->core.config = config; sched_init_resources(sched); sched_init_sharedInst(sched); sched_init_queues(sched); - sched->configver++; /* increase the version of configuration */ + sched->core.configver++; /* increase the version of configuration */ return 0; } @@ -805,7 +741,7 @@ static int sched_update_jobs(xce_scheduler_t* sched, int njobs, xce_sched_job_t* j = make_sched_job(sched, jobs[i]); plugin_job_update(sched, j); - sched_add_job(sched, j); + sched_add_job(&(sched->core), j); } return 0; @@ -814,7 +750,7 @@ static int sched_update_jobs(xce_scheduler_t* sched, int njobs, static int sched_job_state_change(xce_scheduler_t* sched, uint64_t jobid, uint32_t state) { - xce_sched_job_t* job = sched_get_job(sched, jobid); + xce_sched_job_t* job = sched_get_job(&(sched->core), jobid); int rc = 0; if (job == NULL) { return -ENOENT; @@ -846,7 +782,7 @@ static int sched_update_workers(xce_scheduler_t* sched, int nworkers, xce_sched_worker_t* w = make_sched_worker(sched, workers[i]); plugin_worker_update(sched, w); - sched_add_worker(sched, w); + sched_add_worker(&(sched->core), w); } return 0; @@ -880,11 +816,11 @@ int sched_update(xce_scheduler_t* sched) dataq_t* q = NULL; /* swap dataq */ - mylock_lock(&(sched->lock)); - q = sched->pend_updates[0]; - sched->pend_updates[0] = sched->pend_updates[1]; - sched->pend_updates[1] = q; - mylock_unlock(&(sched->lock)); + mylock_lock(&(sched->core.lock)); + q = sched->core.pend_updates[0]; + sched->core.pend_updates[0] = sched->core.pend_updates[1]; + sched->core.pend_updates[1] = q; + mylock_unlock(&(sched->core.lock)); while (1) { data_item_t* it; @@ -926,11 +862,12 @@ int sched_update(xce_scheduler_t* sched) for (i = 0; i < req->u.resourceupdate.nresources; i++) { xce_resource_t* r = req->u.resourceupdate.resources[i]; if (r == NULL) continue; - xce_resource_t* existing = makeResource(sched->resmgr, r); + xce_resource_t* existing = makeResource(sched->core.resmgr, r); if (existing == NULL) { /* EEXIST: update-in-place on existing resource (add-or-update) */ xce_resource_t* found = NULL; - if (key_find(sched->resmgr, r->name, (void**)&found) == 0 + if (key_find(sched->core.resmgr, r->name, (void**)&found) + == 0 && found != NULL) { /* Copy mutable integer fields; desc/name are identity, * distributions/hbased_defined are deep-structs. */ @@ -956,13 +893,13 @@ int sched_update(xce_scheduler_t* sched) if (rv == NULL) { continue; } - if (key_find(sched->resmgr, rv->name, (void**)&r) < 0 + if (key_find(sched->core.resmgr, rv->name, (void**)&r) < 0 || r == NULL) { WARN("%s unknown resource %s in resvals update", __func__, rv->name); continue; } - zarray_put(&sched->sharedInst, r->idx, rv->val); + zarray_put(&sched->core.sharedInst, r->idx, rv->val); } rc = 0; break; @@ -975,9 +912,10 @@ int sched_update(xce_scheduler_t* sched) xce_sched_worker_t* w = NULL; xce_resource_t* r = NULL; if (rv == NULL) continue; - w = sched_get_worker(sched, rv->host); + w = sched_get_worker(&(sched->core), rv->host); if (w == NULL) continue; - if (key_find(sched->resmgr, rv->name, (void**)&r) < 0 || r == NULL) { + if (key_find(sched->core.resmgr, rv->name, (void**)&r) < 0 + || r == NULL) { WARN("%s unknown resource %s in per-worker resvals update", __func__, rv->name); continue; @@ -1017,7 +955,7 @@ void sched_go(xce_scheduler_t* sched) { int go = 1; - cbuf_put(&(sched->wbuf), &go); + cbuf_put(&(sched->core.wbuf), &go); return; } @@ -1029,7 +967,7 @@ static int my_reset_resval_(xce_scheduler_t* sched, xce_val_t* val) static int sched_reset_shared_resval(xce_scheduler_t* sched) { - return zarray_iter(&(sched->sharedInUse), + return zarray_iter(&(sched->core.sharedInUse), (zarray_iter_fn_t)my_reset_resval_, sched); } @@ -1055,7 +993,7 @@ static int sched_reset_workers(xce_scheduler_t* sched) iterator_t it; int rc = 0; - rc = iter_make(&it, sched->workers); + rc = iter_make(&it, sched->core.workers); if (rc < 0) { return rc; } @@ -1085,7 +1023,7 @@ static int sched_account_jobs(xce_scheduler_t* sched) iterator_t it; int rc = 0; - rc = iter_make(&it, sched->jobs); + rc = iter_make(&it, sched->core.jobs); if (rc < 0) { return rc; } @@ -1118,15 +1056,15 @@ static int sched_account_jobs(xce_scheduler_t* sched) void sched_begin_session(xce_scheduler_t* sched) { - atomic_store_explicit(&sched->session_active, 1, memory_order_release); - atomic_store_explicit(&sched->staging_merge_complete, 0, + atomic_store_explicit(&sched->core.session_active, 1, memory_order_release); + atomic_store_explicit(&sched->core.staging_merge_complete, 0, memory_order_release); /* Allocate a fresh memoisation context sized to the current interned-node * population. sched_reset_workers (called below) frees each worker's * stale host_eval_ctx so they are rebuilt lazily during this session. */ - select_eval_ctx_fini(sched->sel_eval_ctx); - sched->sel_eval_ctx = select_eval_ctx_init(sel_get_node_count()); + select_eval_ctx_fini(sched->core.sel_eval_ctx); + sched->core.sel_eval_ctx = select_eval_ctx_init(sel_get_node_count()); plugin_begin_session(sched); @@ -1136,17 +1074,17 @@ void sched_begin_session(xce_scheduler_t* sched) sched_account_jobs(sched); /* Staging merge done — sched_end_session's GC guard reads this field. */ - atomic_store_explicit(&sched->staging_merge_complete, 1, + atomic_store_explicit(&sched->core.staging_merge_complete, 1, memory_order_release); return; } void sched_end_session(xce_scheduler_t* sched) { - atomic_store_explicit(&sched->session_active, 0, memory_order_release); + atomic_store_explicit(&sched->core.session_active, 0, memory_order_release); - select_eval_ctx_fini(sched->sel_eval_ctx); - sched->sel_eval_ctx = NULL; + select_eval_ctx_fini(sched->core.sel_eval_ctx); + sched->core.sel_eval_ctx = NULL; plugin_end_session(sched); @@ -1166,7 +1104,7 @@ void sched_end_session(xce_scheduler_t* sched) xce_sched_job_t** live_jobs = calloc(cap, sizeof(*live_jobs)); if (live_jobs != NULL) { - int rc = iter_make(&it, sched->jobs); + int rc = iter_make(&it, sched->core.jobs); if (rc >= 0) { for (;;) { rc = iter_next(&it, (void**)&job); @@ -1196,7 +1134,7 @@ void sched_end_session(xce_scheduler_t* sched) if (live_jobs == NULL) { /* OOM: cannot safely GC — skip to avoid sweeping nodes still in use */ - atomic_store_explicit(&sched->staging_merge_complete, 1, + atomic_store_explicit(&sched->core.staging_merge_complete, 1, memory_order_release); return; } @@ -1249,163 +1187,10 @@ void sched_end_session(xce_scheduler_t* sched) recompact_node_ids(); free(live_jobs); - atomic_store_explicit(&sched->staging_merge_complete, 1, + atomic_store_explicit(&sched->core.staging_merge_complete, 1, memory_order_release); } -int sched_add_job(xce_scheduler_t* sched, xce_sched_job_t* j) -{ - int rc; - - lock(sched->jobs, RTTI_LOCK_WRITE); - rc = key_add(sched->jobs, (void*)j->jobid, j); - unlock(sched->jobs); - - return rc; -} - -xce_sched_job_t* sched_get_job(xce_scheduler_t* sched, uint64_t jobid) -{ - int rc = 0; - xce_sched_job_t* job = NULL; - - lock(sched->jobs, RTTI_LOCK_READ); - rc = key_find(sched->jobs, (void*)jobid, (void**)&job); - unlock(sched->jobs); - - if (rc <= 0) { - return NULL; - } - - return job; -} - -int sched_add_queue(xce_scheduler_t* sched, xce_sched_queue_t* q) -{ - int rc; - - lock(sched->queues, RTTI_LOCK_WRITE); - rc = key_add(sched->queues, (void*)q->name, q); - unlock(sched->queues); - - return rc; -} - -xce_sched_queue_t* sched_get_queue(xce_scheduler_t* sched, char* name) -{ - xce_sched_queue_t* q = NULL; - int rc = 0; - - lock(sched->queues, RTTI_LOCK_READ); - rc = key_find(sched->queues, (void*)name, (void**)&q); - unlock(sched->queues); - - if (rc <= 0) { - return NULL; - } - - return q; -} - -int sched_add_worker(xce_scheduler_t* sched, xce_sched_worker_t* w) -{ - int rc; - - lock(sched->workers, RTTI_LOCK_WRITE); - rc = key_add(sched->workers, (void*)w->name, (void*)w); - unlock(sched->workers); - if (rc >= 0) { - w->idx = rc; - } - - return rc; -} - -xce_sched_worker_t* sched_get_worker(xce_scheduler_t* sched, char* name) -{ - xce_sched_worker_t* w = NULL; - int rc = 0; - - lock(sched->workers, RTTI_LOCK_READ); - rc = key_find(sched->workers, (void*)name, (void**)&w); - unlock(sched->workers); - - if (rc <= 0) { - return NULL; - } - - return w; -} - -int sched_add_user(xce_scheduler_t* sched, xce_sched_user_t* u) -{ - int rc = 0; - - lock(sched->users, RTTI_LOCK_WRITE); - rc = key_add(sched->users, (void*)u->name, (void*)u); - unlock(sched->users); - - return rc; -} - -xce_sched_user_t* sched_get_user(xce_scheduler_t* sched, char* name) -{ - xce_sched_user_t* user = NULL; - int rc = 0; - - lock(sched->users, RTTI_LOCK_READ); - rc = key_find(sched->users, (void*)name, (void**)&user); - unlock(sched->users); - if (rc < 0) { - return NULL; - } - if (user != NULL) { - return user; - } - - return make_sched_user(sched, name); -} - -int sched_add_project(xce_scheduler_t* sched, xce_sched_project_t* p) -{ - int rc; - - lock(sched->projects, RTTI_LOCK_WRITE); - rc = key_add(sched->projects, (void*)p->name, (void*)p); - unlock(sched->projects); - - return rc; -} - -xce_sched_project_t* sched_get_project(xce_scheduler_t* sched, char* name) -{ - xce_sched_project_t* project = NULL; - int rc = 0; - - if ((name == NULL) || (strlen(name) == 0)) { - name = "default"; - } - - lock(sched->projects, RTTI_LOCK_READ); - rc = key_find(sched->projects, (void*)name, (void**)&project); - unlock(sched->projects); - if (rc < 0) { - return NULL; - } - - if (rc > 0) { - return project; - } - - return make_sched_project(sched, name); -} - -/* FIXME this is just stub, we need to add more */ -int sched_job_is_ready(xce_sched_job_t* j) { return 1; } - -/* FIXME this is just stub, we need to add more */ -int sched_queue_is_ready(xce_sched_queue_t* q) { return 1; } - int sched_commit_decision(xce_scheduler_t* sched, xce_decision_t* decision) { int rc = 0; @@ -1417,7 +1202,7 @@ int sched_commit_decision(xce_scheduler_t* sched, xce_decision_t* decision) } if (decision->op == XCE_DECISION_DISPATCH) { - job = sched_get_job(sched, decision->u.alloc->jobid); + job = sched_get_job(&(sched->core), decision->u.alloc->jobid); job->allocs = decision->u.alloc; job->accounts = make_job_account(job->allocs); @@ -1476,7 +1261,7 @@ int sched_collect_workers(xce_scheduler_t* sched, xlist_t* lst, groupfn = default_worker_groupfn_; } - rc = iter_make(&it, sched->workers); + rc = iter_make(&it, sched->core.workers); if (rc < 0) { return rc; } @@ -1517,9 +1302,9 @@ uint64_t sched_get_id(xce_scheduler_t* sched) { uint64_t id; - mylock_lock(&(sched->lock)); - id = sched->nextid++; - mylock_unlock(&(sched->lock)); + mylock_lock(&(sched->core.lock)); + id = sched->core.nextid++; + mylock_unlock(&(sched->core.lock)); return id; } @@ -1545,7 +1330,7 @@ xce_alloc_t** sched_make_allocs(xce_scheduler_t* sched, char* reserve, int nres, int rc = 0; int d; - rc = vget(sched->resmgr, idxs[i], (void**)&r); + rc = vget(sched->core.resmgr, idxs[i], (void**)&r); d = (duration != NULL) ? duration[i] : -1; allocs[i] = NULL; diff --git a/src/scheduler/testsuite/meson.build b/src/scheduler/testsuite/meson.build index 5971854d..849c307d 100644 --- a/src/scheduler/testsuite/meson.build +++ b/src/scheduler/testsuite/meson.build @@ -175,6 +175,34 @@ sched_pendreason_factory_audit_test = executable( + '"'], install: false) +# Regression guard for task 4.1's Observable (scheduler-core spec, Req +# 5.1/5.2): "the core structure compiles and links with zero symbols from +# the networking/API libraries ... a build-time or link-time check +# confirms this." Same source-text-scan genre as +# sched_picker_core_loop_boundary_test / sched_pendreason_factory_audit_test +# above, PLUS a genuine link-time proof: it inspects the undefined-symbol +# table (via nm(1)) of sched_core_isolation_lib (defined in the parent +# meson.build -- core.c compiled alone, isolated from the rest of the +# scheduler). link_with establishes the build-order dependency so that +# archive exists before this test runs; XCE_CORE_LIB_PATH bakes in its +# absolute path the same way XCE_SCHEDULER_C_PATH does above. +sched_core_isolation_test = executable( + 'sched_core_isolation_test', + ['test_core_isolation.c'], + include_directories:[incdir], + dependencies: [scheduler_test_deps], + link_with: [sched_core_isolation_lib], + c_args: ['-DXCE_CORE_H_PATH="' + + (meson.current_source_dir() / '..' / 'include' / 'core.h') + + '"', + '-DXCE_CORE_C_PATH="' + + (meson.current_source_dir() / '..' / 'src' / 'core.c') + + '"', + '-DXCE_CORE_LIB_PATH="' + + sched_core_isolation_lib.full_path() + + '"'], + install: false) + # Register as meson tests so `meson test --suite scheduler` runs them all. # detect_leaks=0 matches the suite convention (fixtures are not torn down). test('scheduler test', scheduler_test, suite: 'scheduler', @@ -205,3 +233,6 @@ test('scheduler picker core loop boundary test', test('scheduler pendreason factory audit test', sched_pendreason_factory_audit_test, suite: 'scheduler', env: ['ASAN_OPTIONS=detect_leaks=0']) +test('scheduler core isolation test', + sched_core_isolation_test, suite: 'scheduler', + env: ['ASAN_OPTIONS=detect_leaks=0']) diff --git a/src/scheduler/testsuite/test.c b/src/scheduler/testsuite/test.c index 1761fc9c..9e414bcd 100644 --- a/src/scheduler/testsuite/test.c +++ b/src/scheduler/testsuite/test.c @@ -12,12 +12,25 @@ * This also brings in scheduler.h (with the real xce_scheduler_t) and all deps. */ #include "../src/resource.c" -/* Stubs for symbols used by resource.c functions not under test */ -xce_sched_worker_t* sched_get_worker(xce_scheduler_t* s, char* n) { (void)s; (void)n; return NULL; } +/* Stubs for symbols used by resource.c functions not under test. + * sched_get_worker/sched_get_queue relocated to core.h (scheduler-core + * spec, task 4.1) -- now operate on xce_sched_core_t*, not + * xce_scheduler_t*. */ +xce_sched_worker_t* sched_get_worker(xce_sched_core_t* s, char* n) +{ + (void)s; + (void)n; + return NULL; +} int plugin_prepare_allocation(void* s, xce_sched_ctx_t* c) { (void)s; (void)c; return 0; } int sched_ctx_get_uncomit_host(xvec_t** v, xce_sched_ctx_t* c, int i) { (void)v; (void)c; (void)i; return 0; } int plugin_queue_update(void* s, xce_sched_queue_t* q) { (void)s; (void)q; return 0; } -xce_sched_queue_t* sched_get_queue(xce_scheduler_t* s, char* n) { (void)s; (void)n; return NULL; } +xce_sched_queue_t* sched_get_queue(xce_sched_core_t* s, char* n) +{ + (void)s; + (void)n; + return NULL; +} /* Stub: sched_update_existing_queue returns -EINVAL when updated is NULL, * -ENOENT when queue not found (sched_get_queue stub returns NULL). @@ -27,7 +40,7 @@ int sched_update_existing_queue(xce_scheduler_t* sched, xce_queue_t* updated) if (updated == NULL) { return -EINVAL; } - if (sched_get_queue(sched, updated->queue) == NULL) { + if (sched_get_queue(&(sched->core), updated->queue) == NULL) { return -ENOENT; } return 0; @@ -127,9 +140,9 @@ test_scheduler_t_has_atomic_gc_fields(void) xce_scheduler_t sched; memset(&sched, 0, sizeof(sched)); - TEST_ASSERT_EQUAL_INT(0, atomic_load(&sched.session_active)); - TEST_ASSERT_EQUAL_INT(0, atomic_load(&sched.eval_in_progress)); - TEST_ASSERT_EQUAL_INT(0, atomic_load(&sched.staging_merge_complete)); + TEST_ASSERT_EQUAL_INT(0, atomic_load(&sched.core.session_active)); + TEST_ASSERT_EQUAL_INT(0, atomic_load(&sched.core.eval_in_progress)); + TEST_ASSERT_EQUAL_INT(0, atomic_load(&sched.core.staging_merge_complete)); } static void @@ -222,15 +235,15 @@ test_gc_predicates_idle_invariant(void) xce_scheduler_t sched; memset(&sched, 0, sizeof(sched)); /* staging_merge_complete=1 means GC is safe to run */ - atomic_store(&sched.staging_merge_complete, 1); + atomic_store(&sched.core.staging_merge_complete, 1); sel_optimize_set(1); sel_node_table_init(); /* Verify the atomic fields are readable — testing the predicate contract */ - TEST_ASSERT_EQUAL_INT(0, atomic_load(&sched.session_active)); - TEST_ASSERT_EQUAL_INT(0, atomic_load(&sched.eval_in_progress)); - TEST_ASSERT_EQUAL_INT(1, atomic_load(&sched.staging_merge_complete)); + TEST_ASSERT_EQUAL_INT(0, atomic_load(&sched.core.session_active)); + TEST_ASSERT_EQUAL_INT(0, atomic_load(&sched.core.eval_in_progress)); + TEST_ASSERT_EQUAL_INT(1, atomic_load(&sched.core.staging_merge_complete)); /* GC operations on an empty node table must not crash */ gc_clear_marks(); @@ -256,19 +269,20 @@ test_staging_merge_complete_gc_guard(void) memset(&sched, 0, sizeof(sched)); /* Simulate the post-begin_session state the fix produces */ - atomic_store(&sched.session_active, 0); /* session ended */ - atomic_store(&sched.eval_in_progress, 0); - atomic_store(&sched.staging_merge_complete, 1); /* fix: begin_session sets this */ + atomic_store(&sched.core.session_active, 0); /* session ended */ + atomic_store(&sched.core.eval_in_progress, 0); + atomic_store(&sched.core.staging_merge_complete, + 1); /* fix: begin_session sets this */ /* The GC guard in sched_end_session: must NOT fire */ - int guard_fires = atomic_load(&sched.eval_in_progress) - || !atomic_load(&sched.staging_merge_complete); + int guard_fires = atomic_load(&sched.core.eval_in_progress) + || !atomic_load(&sched.core.staging_merge_complete); TEST_ASSERT_EQUAL_INT(0, guard_fires); /* Confirm the unfixed state (staging=0) would block GC */ - atomic_store(&sched.staging_merge_complete, 0); - int guard_fires_unfixed = atomic_load(&sched.eval_in_progress) - || !atomic_load(&sched.staging_merge_complete); + atomic_store(&sched.core.staging_merge_complete, 0); + int guard_fires_unfixed = atomic_load(&sched.core.eval_in_progress) + || !atomic_load(&sched.core.staging_merge_complete); TEST_ASSERT_EQUAL_INT(1, guard_fires_unfixed); } diff --git a/src/scheduler/testsuite/test_core_isolation.c b/src/scheduler/testsuite/test_core_isolation.c new file mode 100644 index 00000000..d22f3e74 --- /dev/null +++ b/src/scheduler/testsuite/test_core_isolation.c @@ -0,0 +1,315 @@ +/* + * test_core_isolation.c - regression guard for task 4.1's Observable + * (scheduler-core spec, Req 5.1/5.2): "the core structure compiles and + * links with zero symbols from the networking/API libraries; a build-time + * or link-time check confirms this" -- and the accompanying constraint + * "no wall-clock time call appears anywhere in the extracted core's + * source." + * + * Two layers, same genre as test_picker_core_loop_boundary.c / + * test_picker_pendreason_factory_audit.c (source-text scans of the real, + * current on-disk files, not a frozen snapshot): + * + * 1. Source-text scan of core.h/core.c: no #include of client.h, + * netio.h, xceapi.h, delay.h, idgen.h, or xce/cluster.h (the + * "Allowed Dependencies" list in design.md names exactly these as + * forbidden); no call-site substring matching a symbol exported only + * by those libraries; no wall-clock call (time()/clock_gettime()/ + * gettimeofday()) anywhere in core.c. + * + * 2. The literal build-time/link-time proof the Observable asks for, + * not achievable by a text scan alone: core.c is compiled in + * isolation into its own static library (meson.build's + * sched_core_isolation_lib, this test's XCE_CORE_LIB_PATH), and this + * test inspects that archive's UNDEFINED SYMBOL TABLE via nm(1) -- + * proving what core.c's *compiled code actually calls*, independent + * of what any header happens to transitively declare (xce/types.h, + * used by literally every file in this directory, itself includes + * netio.h for unrelated reasons -- so a "no forbidden #include" + * check alone would either false-positive on that pre-existing, + * harmless transitive include, or have to special-case it; the nm + * check sidesteps that entirely by checking *usage*, not + * *visibility*). + * + * Why nm on a static_library() rather than the full xce_scheduler + * executable: the executable links the whole monolithic libxce_dep, so + * every symbol from netio/xceapi/client/delay/idgen would appear in ITS + * undefined-symbol table regardless of whether core.c itself ever calls + * one -- that would prove nothing. Compiling core.c alone into its own + * archive (sched_core_isolation_lib, meson.build) and inspecting THAT + * object's undefined symbols isolates exactly what this one translation + * unit references. + */ +#include +#include +#include + +#include + +#ifndef XCE_CORE_H_PATH +#error \ + "XCE_CORE_H_PATH must be defined by meson.build to core.h's absolute path" +#endif +#ifndef XCE_CORE_C_PATH +#error \ + "XCE_CORE_C_PATH must be defined by meson.build to core.c's absolute path" +#endif +#ifndef XCE_CORE_LIB_PATH +#error \ + "XCE_CORE_LIB_PATH must be defined by meson.build to the absolute path of core.c's isolated static_library() archive" +#endif + +/* Forbidden headers (design.md "Allowed Dependencies", Req 5 AC 3): the + * core must never directly include any of these. */ +static const char* const FORBIDDEN_INCLUDES[] = { + "#include \"client.h\"", "#include \"netio.h\"", + "#include \"xceapi.h\"", "#include \"delay.h\"", + "#include \"idgen.h\"", "#include \"xce/cluster.h\"", + "#include ", "#include ", + "#include ", "#include ", + "#include ", "#include ", +}; +#define NUM_FORBIDDEN_INCLUDES \ + (sizeof(FORBIDDEN_INCLUDES) / sizeof(FORBIDDEN_INCLUDES[0])) + +/* Forbidden call-site symbols: exported only by netio.c/xceapi.c/client.c/ + * delay.c/idgen.c (verified against those headers' own extern lists) -- + * none of these are legitimately callable from a core with zero controller + * I/O, network, or timer dependency. */ +static const char* const FORBIDDEN_SYMBOLS[] = { + "xcecli_", + "endpoint_", + "iomgr_", + "idgen_", + "delaymgr_", + "delay_task", + "xce_connect_cluster", + "xce_reconnect_", + "xce_parse_peers", + "xce_cluster_config_validate", + "socket_connect", + "create_ssl_context", + "initialize_openssl", + "cleanup_openssl", + "io_fsm_start", +}; +#define NUM_FORBIDDEN_SYMBOLS \ + (sizeof(FORBIDDEN_SYMBOLS) / sizeof(FORBIDDEN_SYMBOLS[0])) + +/* Wall-clock calls (Req 5 AC "no wall-clock time call appears anywhere in + * the extracted core's source"). */ +static const char* const FORBIDDEN_WALLCLOCK[] = { + "time(", + "clock_gettime(", + "gettimeofday(", +}; +#define NUM_FORBIDDEN_WALLCLOCK \ + (sizeof(FORBIDDEN_WALLCLOCK) / sizeof(FORBIDDEN_WALLCLOCK[0])) + +static char* read_whole_file(const char* path) +{ + FILE* f = fopen(path, "rb"); + if (f == NULL) { + return NULL; + } + + if (fseek(f, 0, SEEK_END) != 0) { + fclose(f); + return NULL; + } + + long len = ftell(f); + if (len < 0) { + fclose(f); + return NULL; + } + rewind(f); + + char* buf = (char*)malloc((size_t)len + 1); + if (buf == NULL) { + fclose(f); + return NULL; + } + + size_t nread = fread(buf, 1, (size_t)len, f); + fclose(f); + buf[nread] = '\0'; + + return buf; +} + +void setUp(void) { return; } +void tearDown(void) { return; } + +/* Sanity check: both source paths baked in by meson.build must actually be + * readable -- without this, a stale path would make every scan below + * silently pass on zero bytes of text (the exact false-confidence failure + * mode this whole test exists to avoid). */ +static void test_core_source_paths_are_readable(void) +{ + char* h = read_whole_file(XCE_CORE_H_PATH); + char* c = read_whole_file(XCE_CORE_C_PATH); + + TEST_ASSERT_NOT_NULL_MESSAGE(h, XCE_CORE_H_PATH); + TEST_ASSERT_NOT_NULL_MESSAGE(c, XCE_CORE_C_PATH); + + free(h); + free(c); +} + +static void test_core_never_includes_forbidden_headers(void) +{ + char* h = read_whole_file(XCE_CORE_H_PATH); + char* c = read_whole_file(XCE_CORE_C_PATH); + size_t i; + + TEST_ASSERT_NOT_NULL(h); + TEST_ASSERT_NOT_NULL(c); + + for (i = 0; i < NUM_FORBIDDEN_INCLUDES; i++) { + char msg[256]; + + snprintf(msg, sizeof(msg), + "core.h directly includes forbidden header '%s'", + FORBIDDEN_INCLUDES[i]); + TEST_ASSERT_NULL_MESSAGE(strstr(h, FORBIDDEN_INCLUDES[i]), msg); + + snprintf(msg, sizeof(msg), + "core.c directly includes forbidden header '%s'", + FORBIDDEN_INCLUDES[i]); + TEST_ASSERT_NULL_MESSAGE(strstr(c, FORBIDDEN_INCLUDES[i]), msg); + } + + free(h); + free(c); +} + +static void test_core_never_calls_forbidden_symbols(void) +{ + char* h = read_whole_file(XCE_CORE_H_PATH); + char* c = read_whole_file(XCE_CORE_C_PATH); + size_t i; + + TEST_ASSERT_NOT_NULL(h); + TEST_ASSERT_NOT_NULL(c); + + for (i = 0; i < NUM_FORBIDDEN_SYMBOLS; i++) { + char msg[256]; + + snprintf(msg, sizeof(msg), + "core.h references forbidden networking/API symbol '%s'", + FORBIDDEN_SYMBOLS[i]); + TEST_ASSERT_NULL_MESSAGE(strstr(h, FORBIDDEN_SYMBOLS[i]), msg); + + snprintf(msg, sizeof(msg), + "core.c references forbidden networking/API symbol '%s'", + FORBIDDEN_SYMBOLS[i]); + TEST_ASSERT_NULL_MESSAGE(strstr(c, FORBIDDEN_SYMBOLS[i]), msg); + } + + free(h); + free(c); +} + +static void test_core_never_calls_wallclock_functions(void) +{ + char* c = read_whole_file(XCE_CORE_C_PATH); + size_t i; + + TEST_ASSERT_NOT_NULL(c); + + for (i = 0; i < NUM_FORBIDDEN_WALLCLOCK; i++) { + char msg[256]; + + snprintf(msg, sizeof(msg), + "core.c calls wall-clock function '%s' -- the core must " + "access time exclusively through a future xce_core_now() " + "(scheduler-core spec, task 4.5)", + FORBIDDEN_WALLCLOCK[i]); + TEST_ASSERT_NULL_MESSAGE(strstr(c, FORBIDDEN_WALLCLOCK[i]), msg); + } + + free(c); +} + +/* + * test_core_object_has_no_forbidden_undefined_symbols - the literal + * build-time/link-time proof: core.c, compiled in isolation into + * sched_core_isolation_lib (meson.build), must not reference (as an + * undefined external symbol in the compiled object) anything matching the + * forbidden prefixes above. This is independent of what any header + * transitively declares -- it reflects only what core.c's compiled code + * actually calls. + */ +static void test_core_object_has_no_forbidden_undefined_symbols(void) +{ + char cmd[4096]; + char line[1024]; + FILE* p; + int any_undefined_seen = 0; + + snprintf(cmd, sizeof(cmd), "nm --format=posix -u \"%s\" 2>&1", + XCE_CORE_LIB_PATH); + + p = popen(cmd, "r"); + TEST_ASSERT_NOT_NULL_MESSAGE(p, "failed to popen() nm(1)"); + + while (fgets(line, sizeof(line), p) != NULL) { + char symbol[512]; + char type; + size_t i; + + /* posix format for an undefined symbol: " U" (no + * value/size columns). Skip anything that doesn't parse as + * exactly that -- e.g. a blank line or an archive member header + * some nm versions print. */ + if (sscanf(line, "%511s %c", symbol, &type) != 2) { + continue; + } + if (type != 'U') { + continue; + } + + any_undefined_seen = 1; + + for (i = 0; i < NUM_FORBIDDEN_SYMBOLS; i++) { + if (strncmp(symbol, FORBIDDEN_SYMBOLS[i], + strlen(FORBIDDEN_SYMBOLS[i])) + == 0) { + char msg[768]; + + snprintf(msg, sizeof(msg), + "core.c's compiled object has an undefined " + "reference to '%s', matching forbidden " + "networking/API prefix '%s' -- core.c must never " + "link networking/API library symbols (Req 5.1/" + "5.2)", + symbol, FORBIDDEN_SYMBOLS[i]); + pclose(p); + TEST_FAIL_MESSAGE(msg); + } + } + } + + pclose(p); + + /* If nm produced no undefined symbols at all, either the archive path + * is wrong or nm itself failed silently -- either way this test would + * be vacuously passing without having checked anything real. */ + TEST_ASSERT_TRUE_MESSAGE( + any_undefined_seen, + "nm reported zero undefined symbols for XCE_CORE_LIB_PATH -- " + "the archive path is likely wrong (this test would otherwise pass " + "vacuously)"); +} + +int main(void) +{ + UNITY_BEGIN(); + RUN_TEST(test_core_source_paths_are_readable); + RUN_TEST(test_core_never_includes_forbidden_headers); + RUN_TEST(test_core_never_calls_forbidden_symbols); + RUN_TEST(test_core_never_calls_wallclock_functions); + RUN_TEST(test_core_object_has_no_forbidden_undefined_symbols); + return UNITY_END(); +} diff --git a/src/scheduler/testsuite/test_handlers.c b/src/scheduler/testsuite/test_handlers.c index 35c7ae8b..5134ffe5 100644 --- a/src/scheduler/testsuite/test_handlers.c +++ b/src/scheduler/testsuite/test_handlers.c @@ -30,7 +30,9 @@ int sched_init_config(xce_scheduler_t* s, xce_conf4sched_t* c) (void)s; (void)c; return 0; } -xce_sched_job_t* sched_get_job(xce_scheduler_t* sched, uint64_t jobid) +/* sched_get_job relocated to core.h (scheduler-core spec, task 4.1) -- now + * operates on xce_sched_core_t*, not xce_scheduler_t*. */ +xce_sched_job_t* sched_get_job(xce_sched_core_t* sched, uint64_t jobid) { (void)sched; (void)jobid; return NULL; } @@ -47,11 +49,11 @@ void tearDown(void) static void fixture_init(xce_scheduler_t* sched) { memset(sched, 0, sizeof(*sched)); - mylock_init(&(sched->lock)); - sched->pend_updates[0] = (dataq_t*)calloc(1, sizeof(dataq_t)); - dataq_init(sched->pend_updates[0], NULL); - sched->pend_updates[1] = (dataq_t*)calloc(1, sizeof(dataq_t)); - dataq_init(sched->pend_updates[1], NULL); + mylock_init(&(sched->core.lock)); + sched->core.pend_updates[0] = (dataq_t*)calloc(1, sizeof(dataq_t)); + dataq_init(sched->core.pend_updates[0], NULL); + sched->core.pend_updates[1] = (dataq_t*)calloc(1, sizeof(dataq_t)); + dataq_init(sched->core.pend_updates[1], NULL); } /* Dequeue one pending request (NULL if the queue is empty) and return its op @@ -59,7 +61,7 @@ static void fixture_init(xce_scheduler_t* sched) * the suite's leaks-off convention). Returns 1 if an item was present. */ static int pop_op(xce_scheduler_t* sched, xce_sched_op_t* op_out) { - data_item_t* it = dataq_deq(sched->pend_updates[0], NULL); + data_item_t* it = dataq_deq(sched->core.pend_updates[0], NULL); xce_sched_update_request_t* req; if (it == NULL) { diff --git a/src/scheduler/testsuite/test_job_array_expand.c b/src/scheduler/testsuite/test_job_array_expand.c index 7ca66c51..cb3e026c 100644 --- a/src/scheduler/testsuite/test_job_array_expand.c +++ b/src/scheduler/testsuite/test_job_array_expand.c @@ -33,23 +33,25 @@ int sched_init_config(xce_scheduler_t* s, xce_conf4sched_t* c) } /* REAL kvmgr-backed job add/get -- same rtti path production uses (mirrors - * test_job_update.c's precedent for exercising sched_get_job "for real"). */ -int sched_add_job(xce_scheduler_t* sched, xce_sched_job_t* j) + * test_job_update.c's precedent for exercising sched_get_job "for real"). + * Relocated to core.h (scheduler-core spec, task 4.1) -- now operate on + * xce_sched_core_t*, not xce_scheduler_t*. */ +int sched_add_job(xce_sched_core_t* core, xce_sched_job_t* j) { int rc; - lock(sched->jobs, RTTI_LOCK_WRITE); - rc = key_add(sched->jobs, (void*)j->jobid, j); - unlock(sched->jobs); + lock(core->jobs, RTTI_LOCK_WRITE); + rc = key_add(core->jobs, (void*)j->jobid, j); + unlock(core->jobs); return rc; } -xce_sched_job_t* sched_get_job(xce_scheduler_t* sched, uint64_t jobid) +xce_sched_job_t* sched_get_job(xce_sched_core_t* core, uint64_t jobid) { int rc; xce_sched_job_t* job = NULL; - lock(sched->jobs, RTTI_LOCK_READ); - rc = key_find(sched->jobs, (void*)jobid, (void**)&job); - unlock(sched->jobs); + lock(core->jobs, RTTI_LOCK_READ); + rc = key_find(core->jobs, (void*)jobid, (void**)&job); + unlock(core->jobs); if (rc <= 0) { return NULL; } @@ -71,19 +73,19 @@ static void fixture_init(xce_scheduler_t* sched) .datakiller = NULL }; memset(sched, 0, sizeof(*sched)); - mylock_init(&(sched->lock)); - sched->pend_updates[0] = (dataq_t*)calloc(1, sizeof(dataq_t)); - dataq_init(sched->pend_updates[0], NULL); - sched->pend_updates[1] = (dataq_t*)calloc(1, sizeof(dataq_t)); - dataq_init(sched->pend_updates[1], NULL); - sched->jobs = makeInstance(&xkvmgr_type, &jobs_params); + mylock_init(&(sched->core.lock)); + sched->core.pend_updates[0] = (dataq_t*)calloc(1, sizeof(dataq_t)); + dataq_init(sched->core.pend_updates[0], NULL); + sched->core.pend_updates[1] = (dataq_t*)calloc(1, sizeof(dataq_t)); + dataq_init(sched->core.pend_updates[1], NULL); + sched->core.jobs = makeInstance(&xkvmgr_type, &jobs_params); } static void seed_existing_job(xce_scheduler_t* sched, uint64_t jobid) { xce_sched_job_t* job = (xce_sched_job_t*)calloc(1, sizeof(*job)); job->jobid = jobid; - sched_add_job(sched, job); + sched_add_job(&(sched->core), job); } /* Builds a heap-allocated, heap-owned-strings (jobArray, submit) pair -- the @@ -119,7 +121,7 @@ static xce_job_submit_t* init_job_array(uint64_t base_jobid, * mirror sched_pend_update_killer). Returns -1 if the queue is empty. */ static int pop_job_update(xce_scheduler_t* sched, xce_job_t*** jobs_out) { - data_item_t* it = dataq_deq(sched->pend_updates[0], NULL); + data_item_t* it = dataq_deq(sched->core.pend_updates[0], NULL); xce_sched_update_request_t* req; int njobs; diff --git a/src/scheduler/testsuite/test_job_array_wire_e2e.c b/src/scheduler/testsuite/test_job_array_wire_e2e.c index 1867f2cb..52074978 100644 --- a/src/scheduler/testsuite/test_job_array_wire_e2e.c +++ b/src/scheduler/testsuite/test_job_array_wire_e2e.c @@ -81,24 +81,26 @@ int sched_init_config(xce_scheduler_t* s, xce_conf4sched_t* c) /* REAL kvmgr-backed job add/get -- same rtti path production uses (mirrors * test_job_array_expand.c's precedent), needed so sched_on_job_array()'s - * per-element dedup check (sched_get_job) is exercised for real. */ -int sched_add_job(xce_scheduler_t* sched, xce_sched_job_t* j) + * per-element dedup check (sched_get_job) is exercised for real. Relocated + * to core.h (scheduler-core spec, task 4.1) -- now operate on + * xce_sched_core_t*, not xce_scheduler_t*. */ +int sched_add_job(xce_sched_core_t* core, xce_sched_job_t* j) { int rc; - lock(sched->jobs, RTTI_LOCK_WRITE); - rc = key_add(sched->jobs, (void*)j->jobid, j); - unlock(sched->jobs); + lock(core->jobs, RTTI_LOCK_WRITE); + rc = key_add(core->jobs, (void*)j->jobid, j); + unlock(core->jobs); return rc; } -xce_sched_job_t* sched_get_job(xce_scheduler_t* sched, uint64_t jobid) +xce_sched_job_t* sched_get_job(xce_sched_core_t* core, uint64_t jobid) { int rc; xce_sched_job_t* job = NULL; - lock(sched->jobs, RTTI_LOCK_READ); - rc = key_find(sched->jobs, (void*)jobid, (void**)&job); - unlock(sched->jobs); + lock(core->jobs, RTTI_LOCK_READ); + rc = key_find(core->jobs, (void*)jobid, (void**)&job); + unlock(core->jobs); if (rc <= 0) { return NULL; @@ -118,12 +120,12 @@ static void fixture_init(xce_scheduler_t* sched) .datakiller = NULL }; memset(sched, 0, sizeof(*sched)); - mylock_init(&(sched->lock)); - sched->pend_updates[0] = (dataq_t*)calloc(1, sizeof(dataq_t)); - dataq_init(sched->pend_updates[0], NULL); - sched->pend_updates[1] = (dataq_t*)calloc(1, sizeof(dataq_t)); - dataq_init(sched->pend_updates[1], NULL); - sched->jobs = makeInstance(&xkvmgr_type, &jobs_params); + mylock_init(&(sched->core.lock)); + sched->core.pend_updates[0] = (dataq_t*)calloc(1, sizeof(dataq_t)); + dataq_init(sched->core.pend_updates[0], NULL); + sched->core.pend_updates[1] = (dataq_t*)calloc(1, sizeof(dataq_t)); + dataq_init(sched->core.pend_updates[1], NULL); + sched->core.jobs = makeInstance(&xkvmgr_type, &jobs_params); } /* Pop one pending update request, asserting it is an UPDATE_JOB op; hands @@ -132,7 +134,7 @@ static void fixture_init(xce_scheduler_t* sched) * test_job_array_expand.c's identical helper. */ static int pop_job_update(xce_scheduler_t* sched, xce_job_t*** jobs_out) { - data_item_t* it = dataq_deq(sched->pend_updates[0], NULL); + data_item_t* it = dataq_deq(sched->core.pend_updates[0], NULL); xce_sched_update_request_t* req; int njobs; @@ -168,7 +170,7 @@ static void materialize_popped_jobs(xce_scheduler_t* sched, xce_job_t** jobs, for (i = 0; i < njobs; i++) { xce_sched_job_t* job = (xce_sched_job_t*)calloc(1, sizeof(*job)); job->jobid = jobs[i]->jobid; - sched_add_job(sched, job); + sched_add_job(&(sched->core), job); free_Job_ptr(&jobs[i]); } free(jobs); diff --git a/src/scheduler/testsuite/test_job_update.c b/src/scheduler/testsuite/test_job_update.c index 1006e86c..59fe6755 100644 --- a/src/scheduler/testsuite/test_job_update.c +++ b/src/scheduler/testsuite/test_job_update.c @@ -31,12 +31,16 @@ /* Stubs for scheduler-internal externs schedobj.c references but which are * defined in other .c files not linked into this test. (make_sched_obj and - * kill_sched_job are static in schedobj.c and come in via the #include.) */ -xce_sched_user_t* sched_get_user(xce_scheduler_t* s, char* n) + * kill_sched_job are static in schedobj.c and come in via the #include.) + * sched_get_user/sched_get_project/sched_get_queue/sched_add_user/ + * sched_add_project/sched_add_job/sched_get_job relocated to core.h + * (scheduler-core spec, task 4.1) -- now operate on xce_sched_core_t*, not + * xce_scheduler_t*. */ +xce_sched_user_t* sched_get_user(xce_sched_core_t* s, char* n) { (void)s; (void)n; return NULL; } -xce_sched_project_t* sched_get_project(xce_scheduler_t* s, char* n) +xce_sched_project_t* sched_get_project(xce_sched_core_t* s, char* n) { (void)s; (void)n; return NULL; } @@ -48,8 +52,18 @@ int plugin_job_rusage_update(void* s, xce_sched_job_t* j, xce_job_rusages_t* r) } xce_job_account_t* make_job_account(xce_job_alloc_t* a) { (void)a; return NULL; } void free_job_account(xce_job_account_t* a) { (void)a; } -int sched_add_user(xce_scheduler_t* s, xce_sched_user_t* u) { (void)s; (void)u; return 0; } -int sched_add_project(xce_scheduler_t* s, xce_sched_project_t* p) { (void)s; (void)p; return 0; } +int sched_add_user(xce_sched_core_t* s, xce_sched_user_t* u) +{ + (void)s; + (void)u; + return 0; +} +int sched_add_project(xce_sched_core_t* s, xce_sched_project_t* p) +{ + (void)s; + (void)p; + return 0; +} void sched_worker_release_host_ctx(xce_sched_worker_t* w) { (void)w; } int plugin_prepare_allocation(void* s, xce_sched_ctx_t* c) { (void)s; (void)c; return 0; } @@ -57,7 +71,7 @@ int plugin_prepare_allocation(void* s, xce_sched_ctx_t* c) { (void)s; (void)c; r static xce_sched_queue_t* g_test_queues[8]; static int g_test_nqueues; -xce_sched_queue_t* sched_get_queue(xce_scheduler_t* s, char* name) +xce_sched_queue_t* sched_get_queue(xce_sched_core_t* s, char* name) { int i; (void)s; @@ -74,22 +88,22 @@ xce_sched_queue_t* sched_get_queue(xce_scheduler_t* s, char* name) } /* REAL kvmgr-backed job add/get - same rtti path the production code uses. */ -int sched_add_job(xce_scheduler_t* sched, xce_sched_job_t* j) +int sched_add_job(xce_sched_core_t* core, xce_sched_job_t* j) { int rc; - lock(sched->jobs, RTTI_LOCK_WRITE); - rc = key_add(sched->jobs, (void*)j->jobid, j); - unlock(sched->jobs); + lock(core->jobs, RTTI_LOCK_WRITE); + rc = key_add(core->jobs, (void*)j->jobid, j); + unlock(core->jobs); return rc; } -xce_sched_job_t* sched_get_job(xce_scheduler_t* sched, uint64_t jobid) +xce_sched_job_t* sched_get_job(xce_sched_core_t* core, uint64_t jobid) { int rc; xce_sched_job_t* job = NULL; - lock(sched->jobs, RTTI_LOCK_READ); - rc = key_find(sched->jobs, (void*)jobid, (void**)&job); - unlock(sched->jobs); + lock(core->jobs, RTTI_LOCK_READ); + rc = key_find(core->jobs, (void*)jobid, (void**)&job); + unlock(core->jobs); if (rc <= 0) { return NULL; } @@ -120,8 +134,8 @@ static void fixture_init(xce_scheduler_t* sched) resreq_init(1); xstring_table_init(1024 * 16); - sched->jobs = makeInstance(&xkvmgr_type, &jobs_params); - sched->resmgr = makeInstance(&xkvmgr_type, &res_params); + sched->core.jobs = makeInstance(&xkvmgr_type, &jobs_params); + sched->core.resmgr = makeInstance(&xkvmgr_type, &res_params); g_test_nqueues = 0; memset(g_test_queues, 0, sizeof(g_test_queues)); @@ -146,7 +160,7 @@ static xce_sched_job_t* seed_job(xce_scheduler_t* sched, uint64_t jobid) job->resreq = (xce_resreq_t*)calloc(1, sizeof(*job->resreq)); /* schedobj is live runtime state that MUST survive an in-place update */ job->schedobj = (xce_sched_obj_t*)calloc(1, sizeof(*job->schedobj)); - sched_add_job(sched, job); + sched_add_job(&(sched->core), job); return job; } @@ -182,7 +196,7 @@ test_update_existing_job_mutates_in_place(void) TEST_ASSERT_EQUAL_INT(0, rc); /* SAME object mutated in place, not replaced */ - TEST_ASSERT_EQUAL_PTR(job, sched_get_job(&sched, 42)); + TEST_ASSERT_EQUAL_PTR(job, sched_get_job(&sched.core, 42)); /* mutable fields updated */ TEST_ASSERT_EQUAL_UINT(JOB_STAT_RUN, job->state); TEST_ASSERT_EQUAL_PTR(g_test_queues[0], job->queue); @@ -209,7 +223,7 @@ test_update_unknown_job_returns_enoent_no_create(void) TEST_ASSERT_EQUAL_INT(-ENOENT, rc); /* And NO job was created via the update path */ - TEST_ASSERT_NULL(sched_get_job(&sched, 999)); + TEST_ASSERT_NULL(sched_get_job(&sched.core, 999)); free(upd->submit); free(upd); @@ -250,7 +264,7 @@ test_rusage_stored_on_existing_job(void) /* stored ON the job (ownership transferred to the job) */ TEST_ASSERT_EQUAL_PTR(ru, job->rusages); /* the job object itself is unchanged identity */ - TEST_ASSERT_EQUAL_PTR(job, sched_get_job(&sched, 42)); + TEST_ASSERT_EQUAL_PTR(job, sched_get_job(&sched.core, 42)); /* job owns ru now; kill_sched_job/free_job_rusages will release it */ } diff --git a/src/scheduler/testsuite/test_picker_config_fields.c b/src/scheduler/testsuite/test_picker_config_fields.c index bb9ecb04..c3a30d9c 100644 --- a/src/scheduler/testsuite/test_picker_config_fields.c +++ b/src/scheduler/testsuite/test_picker_config_fields.c @@ -48,29 +48,22 @@ #include "scheduler.h" #include "xstring.h" -/* Pull the real implementation under test into this TU. */ +/* Pull the real implementations under test into this TU. sched_get_user/ + * sched_get_project/sched_get_queue/sched_get_job/sched_add_user/ + * sched_add_project/make_sched_user/make_sched_project relocated to core.c + * (scheduler-core spec, task 4.1) -- pulling core.c in alongside schedobj.c + * (rather than re-stubbing them) is what lets this file keep exercising the + * REAL make_sched_project, per this file's own header comment. */ +#include "../src/core.c" #include "../src/schedobj.c" -/* Stubs for scheduler-internal externs schedobj.c references but which are - * defined in other .c files not linked into this test (same set as - * test_job_update.c). Not exercised by the tests below -- present only to - * satisfy the linker for this whole-file #include. */ -xce_sched_user_t* sched_get_user(xce_scheduler_t* s, char* n) -{ - (void)s; (void)n; return NULL; -} -xce_sched_project_t* sched_get_project(xce_scheduler_t* s, char* n) -{ - (void)s; (void)n; return NULL; -} -xce_sched_queue_t* sched_get_queue(xce_scheduler_t* s, char* n) -{ - (void)s; (void)n; return NULL; -} -xce_sched_job_t* sched_get_job(xce_scheduler_t* s, uint64_t jobid) -{ - (void)s; (void)jobid; return NULL; -} +/* Stubs for scheduler-internal externs schedobj.c/core.c reference but + * which are defined in other .c files not linked into this test. Not + * exercised by the tests below -- present only to satisfy the linker for + * this whole-file #include. plugin_unregister is core.c's kvmgr datakiller + * for the (never-called-in-this-test) plugins kvmgr inside + * xce_sched_core_init() -- still needs a resolvable address even though + * that code path never runs here. */ int plugin_job_update(void* s, xce_sched_job_t* j) { (void)s; (void)j; return 0; } int plugin_queue_update(void* s, xce_sched_queue_t* q) { (void)s; (void)q; return 0; } int plugin_job_rusage_update(void* s, xce_sched_job_t* j, xce_job_rusages_t* r) @@ -78,11 +71,10 @@ int plugin_job_rusage_update(void* s, xce_sched_job_t* j, xce_job_rusages_t* r) (void)s; (void)j; (void)r; return 0; } xce_job_account_t* make_job_account(xce_job_alloc_t* a) { (void)a; return NULL; } -void free_job_account(xce_job_account_t* a) { (void)a; } -int sched_add_user(xce_scheduler_t* s, xce_sched_user_t* u) { (void)s; (void)u; return 0; } -int sched_add_project(xce_scheduler_t* s, xce_sched_project_t* p) { (void)s; (void)p; return 0; } +void free_job_account(xce_job_account_t* a) { (void)a; } void sched_worker_release_host_ctx(xce_sched_worker_t* w) { (void)w; } int plugin_prepare_allocation(void* s, xce_sched_ctx_t* c) { (void)s; (void)c; return 0; } +void plugin_unregister(void* arg) { (void)arg; } void setUp(void) { return; } void tearDown(void) { return; } @@ -117,12 +109,12 @@ static void test_scheduler_job_pick_plugin_field_is_char_star_and_unset_by_defau xce_scheduler_t sched; fixture_init(&sched); - TEST_ASSERT_NULL(sched.job_pick_plugin); + TEST_ASSERT_NULL(sched.core.job_pick_plugin); /* Type check: assigning a char* literal must compile without a cast -- * this is a compile-time proof of the declared field type. */ - sched.job_pick_plugin = "some_name"; - TEST_ASSERT_EQUAL_STRING("some_name", sched.job_pick_plugin); + sched.core.job_pick_plugin = "some_name"; + TEST_ASSERT_EQUAL_STRING("some_name", sched.core.job_pick_plugin); } /* Req 2 AC 1's literal wording: "defaulting to builtin_fcfs when unset". */ @@ -131,12 +123,12 @@ static void test_sched_set_default_job_pick_plugin_applies_builtin_fcfs_when_uns xce_scheduler_t sched; fixture_init(&sched); - TEST_ASSERT_NULL(sched.job_pick_plugin); + TEST_ASSERT_NULL(sched.core.job_pick_plugin); sched_set_default_job_pick_plugin(&sched); - TEST_ASSERT_NOT_NULL(sched.job_pick_plugin); - TEST_ASSERT_EQUAL_STRING("builtin_fcfs", sched.job_pick_plugin); + TEST_ASSERT_NOT_NULL(sched.core.job_pick_plugin); + TEST_ASSERT_EQUAL_STRING("builtin_fcfs", sched.core.job_pick_plugin); } /* "...when unset" implies idempotence: an already-configured name (however @@ -147,11 +139,11 @@ test_sched_set_default_job_pick_plugin_does_not_clobber_configured_value(void) xce_scheduler_t sched; fixture_init(&sched); - sched.job_pick_plugin = makeString("priority_picker"); + sched.core.job_pick_plugin = makeString("priority_picker"); sched_set_default_job_pick_plugin(&sched); - TEST_ASSERT_EQUAL_STRING("priority_picker", sched.job_pick_plugin); + TEST_ASSERT_EQUAL_STRING("priority_picker", sched.core.job_pick_plugin); } static void test_sched_set_default_job_pick_plugin_is_null_safe(void) @@ -201,7 +193,7 @@ static void test_queue_with_no_override_inherits_global_picker_name(void) TEST_ASSERT_NOT_NULL(queue); TEST_ASSERT_NULL(queue->job_pick_plugin); - TEST_ASSERT_EQUAL_STRING("builtin_fcfs", sched.job_pick_plugin); + TEST_ASSERT_EQUAL_STRING("builtin_fcfs", sched.core.job_pick_plugin); } /* tasks.md Observable, half 2: "a queue config with an override reports @@ -228,8 +220,9 @@ test_queue_with_override_reports_its_own_override_not_global_default(void) TEST_ASSERT_NOT_NULL(queue->job_pick_plugin); TEST_ASSERT_EQUAL_STRING("priority_picker", queue->job_pick_plugin); /* The override is reported, NOT the global default. */ - TEST_ASSERT_TRUE(strcmp(queue->job_pick_plugin, sched.job_pick_plugin) != 0); - TEST_ASSERT_EQUAL_STRING("builtin_fcfs", sched.job_pick_plugin); + TEST_ASSERT_TRUE(strcmp(queue->job_pick_plugin, sched.core.job_pick_plugin) + != 0); + TEST_ASSERT_EQUAL_STRING("builtin_fcfs", sched.core.job_pick_plugin); } /* project_name is independently settable/readable (structural field check @@ -263,7 +256,7 @@ static void test_make_sched_project_priority_field_defaults_to_zero(void) fixture_init(&sched); - project = make_sched_project(&sched, "acct-42"); + project = make_sched_project(&sched.core, "acct-42"); TEST_ASSERT_NOT_NULL(project); TEST_ASSERT_EQUAL_STRING("acct-42", project->name); @@ -281,7 +274,7 @@ static void test_project_priority_field_is_int32_and_independently_settable(void fixture_init(&sched); - project = make_sched_project(&sched, "acct-7"); + project = make_sched_project(&sched.core, "acct-7"); project->priority = 7; TEST_ASSERT_EQUAL_INT32(7, project->priority); diff --git a/src/scheduler/testsuite/test_picker_pendreason_factory_audit.c b/src/scheduler/testsuite/test_picker_pendreason_factory_audit.c index 20d991c6..a09defb5 100644 --- a/src/scheduler/testsuite/test_picker_pendreason_factory_audit.c +++ b/src/scheduler/testsuite/test_picker_pendreason_factory_audit.c @@ -56,10 +56,12 @@ #endif static const char* const SCANNED_RELATIVE_PATHS[] = { + "include/core.h", "include/decision.h", "include/plugin.h", "include/scheduler.h", "include/sched/types.h", + "src/core.c", "src/decision.c", "src/handler.c", "src/main.c", diff --git a/src/scheduler/testsuite/test_picker_wiring.c b/src/scheduler/testsuite/test_picker_wiring.c index 8e107652..d1d10f25 100644 --- a/src/scheduler/testsuite/test_picker_wiring.c +++ b/src/scheduler/testsuite/test_picker_wiring.c @@ -67,25 +67,31 @@ * which the REAL plugin.c now provides in this TU) as * test_picker_config_fields.c uses for its own schedobj.c-only include. * ---------------------------------------------------------------------- */ -xce_sched_user_t* sched_get_user(xce_scheduler_t* s, char* n) +/* sched_get_user/sched_get_project/sched_get_queue/sched_get_job/ + * sched_add_user/sched_add_project relocated to core.h (scheduler-core + * spec, task 4.1) -- now operate on xce_sched_core_t*, not + * xce_scheduler_t*. Kept as stubs here (not pulling in core.c's real + * implementations): this file never exercises make_sched_job/queue + * lookup, only picker registration/resolution. */ +xce_sched_user_t* sched_get_user(xce_sched_core_t* s, char* n) { (void)s; (void)n; return NULL; } -xce_sched_project_t* sched_get_project(xce_scheduler_t* s, char* n) +xce_sched_project_t* sched_get_project(xce_sched_core_t* s, char* n) { (void)s; (void)n; return NULL; } -xce_sched_queue_t* sched_get_queue(xce_scheduler_t* s, char* n) +xce_sched_queue_t* sched_get_queue(xce_sched_core_t* s, char* n) { (void)s; (void)n; return NULL; } -xce_sched_job_t* sched_get_job(xce_scheduler_t* s, uint64_t jobid) +xce_sched_job_t* sched_get_job(xce_sched_core_t* s, uint64_t jobid) { (void)s; (void)jobid; @@ -97,13 +103,13 @@ xce_job_account_t* make_job_account(xce_job_alloc_t* a) return NULL; } void free_job_account(xce_job_account_t* a) { (void)a; } -int sched_add_user(xce_scheduler_t* s, xce_sched_user_t* u) +int sched_add_user(xce_sched_core_t* s, xce_sched_user_t* u) { (void)s; (void)u; return 0; } -int sched_add_project(xce_scheduler_t* s, xce_sched_project_t* p) +int sched_add_project(xce_sched_core_t* s, xce_sched_project_t* p) { (void)s; (void)p; @@ -146,7 +152,7 @@ void setUp(void) { return; } void tearDown(void) { return; } /* ---------------------------------------------------------------------- */ -/* Fixture: a real sched->plugins kvmgr (same setup scheduler.c's +/* Fixture: a real sched->core.plugins kvmgr (same setup scheduler.c's * sched_plugin_init() uses) plus the xstring/CSE intern tables * sched_set_default_job_pick_plugin()'s makeString() call needs. Nothing is * torn down between tests (suite convention, ASAN_OPTIONS=detect_leaks=0). */ @@ -161,8 +167,8 @@ static void fixture_init(xce_scheduler_t* sched) memset(sched, 0, sizeof(*sched)); resreq_init(1); xstring_table_init(1024 * 16); - sched->plugins = makeInstance(&xkvmgr_type, ¶ms); - TEST_ASSERT_NOT_NULL(sched->plugins); + sched->core.plugins = makeInstance(&xkvmgr_type, ¶ms); + TEST_ASSERT_NOT_NULL(sched->core.plugins); } /* Minimal legacy (job_elect_fn-only) plugin, mirroring @@ -202,9 +208,9 @@ static void test_register_builtin_fcfs_registers_under_name_builtin_fcfs(void) rc = register_builtin_fcfs(&sched); TEST_ASSERT_GREATER_OR_EQUAL(0, rc); - lock(sched.plugins, RTTI_LOCK_READ); - rc = key_find(sched.plugins, "builtin_fcfs", (void**)&found); - unlock(sched.plugins); + lock(sched.core.plugins, RTTI_LOCK_READ); + rc = key_find(sched.core.plugins, "builtin_fcfs", (void**)&found); + unlock(sched.core.plugins); TEST_ASSERT_GREATER_THAN(0, rc); TEST_ASSERT_NOT_NULL(found); @@ -224,9 +230,9 @@ static void test_register_builtin_fcfs_is_native_not_legacy(void) fixture_init(&sched); TEST_ASSERT_GREATER_OR_EQUAL(0, register_builtin_fcfs(&sched)); - lock(sched.plugins, RTTI_LOCK_READ); - key_find(sched.plugins, "builtin_fcfs", (void**)&found); - unlock(sched.plugins); + lock(sched.core.plugins, RTTI_LOCK_READ); + key_find(sched.core.plugins, "builtin_fcfs", (void**)&found); + unlock(sched.core.plugins); TEST_ASSERT_NOT_NULL(found); TEST_ASSERT_NULL(found->job_elect_fn); @@ -264,9 +270,9 @@ static void test_register_builtin_fcfs_runs_init_via_init_fn_adapter(void) fixture_init(&sched); TEST_ASSERT_GREATER_OR_EQUAL(0, register_builtin_fcfs(&sched)); - lock(sched.plugins, RTTI_LOCK_READ); - key_find(sched.plugins, "builtin_fcfs", (void**)&found); - unlock(sched.plugins); + lock(sched.core.plugins, RTTI_LOCK_READ); + key_find(sched.core.plugins, "builtin_fcfs", (void**)&found); + unlock(sched.core.plugins); TEST_ASSERT_NOT_NULL(found); TEST_ASSERT_NOT_NULL(found->data); @@ -286,10 +292,10 @@ static void test_unset_job_pick_plugin_resolves_to_builtin_fcfs_by_default(void) fixture_init(&sched); TEST_ASSERT_GREATER_OR_EQUAL(0, register_builtin_fcfs(&sched)); - TEST_ASSERT_NULL(sched.job_pick_plugin); + TEST_ASSERT_NULL(sched.core.job_pick_plugin); sched_set_default_job_pick_plugin(&sched); - TEST_ASSERT_EQUAL_STRING("builtin_fcfs", sched.job_pick_plugin); + TEST_ASSERT_EQUAL_STRING("builtin_fcfs", sched.core.job_pick_plugin); rc = sched_resolve_job_picker(&sched, &resolved); @@ -316,7 +322,7 @@ test_unregistered_configured_name_fails_with_enoent_not_silent_fallback(void) fixture_init(&sched); TEST_ASSERT_GREATER_OR_EQUAL(0, register_builtin_fcfs(&sched)); - sched.job_pick_plugin = "totally_unregistered_picker_name"; + sched.core.job_pick_plugin = "totally_unregistered_picker_name"; rc = sched_resolve_job_picker(&sched, &resolved); @@ -350,7 +356,7 @@ static void test_registered_but_not_a_picker_fails_with_einval_not_enoent(void) * categories), which plugin_register() leaves untouched (task 1.2). */ TEST_ASSERT_GREATER_OR_EQUAL(0, plugin_register(&sched, non_picker)); - sched.job_pick_plugin = "not_a_picker"; + sched.core.job_pick_plugin = "not_a_picker"; rc = sched_resolve_job_picker(&sched, &resolved); @@ -369,7 +375,7 @@ test_unset_job_pick_plugin_without_defaulting_fails_with_einval(void) int rc; fixture_init(&sched); - TEST_ASSERT_NULL(sched.job_pick_plugin); + TEST_ASSERT_NULL(sched.core.job_pick_plugin); rc = sched_resolve_job_picker(&sched, &resolved); @@ -403,7 +409,7 @@ test_resolved_legacy_plugin_has_job_elect_fn_set_selecting_legacy_branch(void) legacy = make_legacy_plugin("test_legacy_picker_for_wiring"); TEST_ASSERT_GREATER_OR_EQUAL(0, plugin_register(&sched, legacy)); - sched.job_pick_plugin = "test_legacy_picker_for_wiring"; + sched.core.job_pick_plugin = "test_legacy_picker_for_wiring"; rc = sched_resolve_job_picker(&sched, &resolved); @@ -429,7 +435,7 @@ test_resolved_builtin_fcfs_has_job_elect_fn_null_selecting_native_branch(void) fixture_init(&sched); TEST_ASSERT_GREATER_OR_EQUAL(0, register_builtin_fcfs(&sched)); - sched.job_pick_plugin = "builtin_fcfs"; /* explicit, not defaulted */ + sched.core.job_pick_plugin = "builtin_fcfs"; /* explicit, not defaulted */ rc = sched_resolve_job_picker(&sched, &resolved); @@ -460,12 +466,12 @@ test_legacy_and_native_pickers_coexist_and_resolve_independently(void) TEST_ASSERT_GREATER_OR_EQUAL(0, plugin_register(&sched, legacy)); TEST_ASSERT_GREATER_OR_EQUAL(0, register_builtin_fcfs(&sched)); - sched.job_pick_plugin = "xce_sched_fcfs_stub"; + sched.core.job_pick_plugin = "xce_sched_fcfs_stub"; rc = sched_resolve_job_picker(&sched, &resolved); TEST_ASSERT_EQUAL_INT(0, rc); TEST_ASSERT_NOT_NULL(resolved->job_elect_fn); - sched.job_pick_plugin = "builtin_fcfs"; + sched.core.job_pick_plugin = "builtin_fcfs"; rc = sched_resolve_job_picker(&sched, &resolved); TEST_ASSERT_EQUAL_INT(0, rc); TEST_ASSERT_NULL(resolved->job_elect_fn); diff --git a/src/scheduler/testsuite/test_plugin_picker_registration.c b/src/scheduler/testsuite/test_plugin_picker_registration.c index 09bfe1e0..9057912b 100644 --- a/src/scheduler/testsuite/test_plugin_picker_registration.c +++ b/src/scheduler/testsuite/test_plugin_picker_registration.c @@ -92,7 +92,7 @@ static void start_log_capture(void) static void stop_log_capture(void) { tlog_reg_early_printf_callback(NULL); } /* ---------------------------------------------------------------------- */ -/* Fixture: a real sched->plugins kvmgr -- same setup as scheduler.c's +/* Fixture: a real sched->core.plugins kvmgr -- same setup as scheduler.c's * (static, so not reusable) sched_plugin_init(). Each test gets its own * fresh instance; nothing is torn down (suite convention, see meson.build: * ASAN_OPTIONS=detect_leaks=0). */ @@ -106,8 +106,8 @@ static void fixture_init(xce_scheduler_t* sched) .datakiller = plugin_unregister }; memset(sched, 0, sizeof(*sched)); - sched->plugins = makeInstance(&xkvmgr_type, ¶ms); - TEST_ASSERT_NOT_NULL(sched->plugins); + sched->core.plugins = makeInstance(&xkvmgr_type, ¶ms); + TEST_ASSERT_NOT_NULL(sched->core.plugins); } /* ---------------------------------------------------------------------- */