lei.chen/myxce
1
0
Fork 1
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

feat(scheduler-core): extract xce_sched_core_t from monolithic scheduler struct (task 4.1)

Browse source 
Splits xce_scheduler_t along the controller-I/O seam: xce_sched_core_t
(new core.h/core.c) now holds all scheduling state (jobs/queues/workers/
users/projects/resource managers/plugins/three-thread-cbuf pipeline),
while xce_scheduler_t shrinks to {core; apictx; cluster_cfg}. Proven via
an nm-based link-time isolation test (zero netio/xcecli/idgen/delaymgr
symbols reachable from core.c) plus a reproducible wall-clock-call guard.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_019dw5WnN3DXStaEaPYfACU1
This commit is contained in:
Rongsong Shen 2026-08-03 10:55:58 +00:00
parent 604a564904
commit 7c229284e2
Signed by: rshen
SSH key fingerprint: SHA256:8Vf2pZ3h6rJa7EhGK+WaAQ9Mta/apZvteW6g9U/WKKk
26 changed files with 1233 additions and 661 deletions
Download patch file Download diff file Expand all files Collapse all files

6
.kiro/specs/scheduler-core/tasks.md
View file

@ -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

117
src/scheduler/include/core.h Normal file
View file

@ -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 <pthread.h>
#include <rtti.h>
#include <stdatomic.h>
#include <stdint.h>
#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

78
src/scheduler/include/scheduler.h
View file

@ -9,6 +9,7 @@
#include <time.h>
#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*);

16
src/scheduler/meson.build
View file

@ -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')

345
src/scheduler/src/core.c Normal file
View file

@ -0,0 +1,345 @@
#include <errno.h>
#include <rtti.h>
#include <stdlib.h>
#include <string.h>
#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; }

8
src/scheduler/src/handler.c
View file

@ -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;
}

8
src/scheduler/src/main.c
View file

@ -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;
}

36
src/scheduler/src/plugin.c
View file

@ -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;
}

8
src/scheduler/src/plugins/fcfs.c
View file

@ -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;
}

18
src/scheduler/src/plugins/limit.c
View file

@ -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;

2
src/scheduler/src/plugins/parjob_allocator.c
View file

@ -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;
}

16
src/scheduler/src/plugins/seqjob_allocator.c
View file

@ -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;

35
src/scheduler/src/resource.c
View file

@ -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;
}

95
src/scheduler/src/schedobj.c
View file

@ -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;
}

431
src/scheduler/src/scheduler.c
View file

@ -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, &params);
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;

31
src/scheduler/testsuite/meson.build
View file

@ -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'])

52
src/scheduler/testsuite/test.c
View file

@ -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);
}

315
src/scheduler/testsuite/test_core_isolation.c Normal file
View file

@ -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 <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unity/unity.h>
#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 <client.h>", "#include <netio.h>",
"#include <xceapi.h>", "#include <delay.h>",
"#include <idgen.h>", "#include <xce/cluster.h>",
};
#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: "<name> 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();
}

16
src/scheduler/testsuite/test_handlers.c
View file

@ -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) {

36
src/scheduler/testsuite/test_job_array_expand.c
View file

@ -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;

36
src/scheduler/testsuite/test_job_array_wire_e2e.c
View file

@ -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);

54
src/scheduler/testsuite/test_job_update.c
View file

@ -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 */
}

67
src/scheduler/testsuite/test_picker_config_fields.c
View file

@ -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);

2
src/scheduler/testsuite/test_picker_pendreason_factory_audit.c
View file

@ -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",

60
src/scheduler/testsuite/test_picker_wiring.c
View file

@ -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, &params);
TEST_ASSERT_NOT_NULL(sched->plugins);
sched->core.plugins = makeInstance(&xkvmgr_type, &params);
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);

6
src/scheduler/testsuite/test_plugin_picker_registration.c
View file

@ -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, &params);
TEST_ASSERT_NOT_NULL(sched->plugins);
sched->core.plugins = makeInstance(&xkvmgr_type, &params);
TEST_ASSERT_NOT_NULL(sched->core.plugins);
}
/* ---------------------------------------------------------------------- */