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

feat(scheduler-core): implement core event batch push interface (task 4.3)

Browse source 
Defines xce_core_event_type_t/xce_core_event_t (9 event kinds: config,
job, worker, resource, resval full/delta, job state change, job update,
queue update, job rusage) in core.h. Implements xce_core_push_events()
in a new core_events.c (split out of core.c to stay under the 800-line
cap): rejects -EBUSY mid-session, otherwise applies each event's core-
state mutation and fires the corresponding picker hook synchronously,
exactly once per event, before returning. CONFIG_PUSH decomposes the
pushed config's embedded queues/resources/resvals via shared helpers
(also reused by QUEUE_UPDATE's upsert path), mirroring the existing
sched_init_config()/sched_init_resources()/sched_init_sharedInst()/
sched_init_queues(). Narrows sched_resolve_job_picker() to take
xce_sched_core_t* (pure type change, zero behavior change). Purely
additive — not wired into the live scheduling path yet (task 5.x).

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 16:05:33 +00:00
parent e9b5c3f589
commit e6c0a2448a
Signed by: rshen
SSH key fingerprint: SHA256:8Vf2pZ3h6rJa7EhGK+WaAQ9Mta/apZvteW6g9U/WKKk
10 changed files with 1613 additions and 43 deletions
Download patch file Download diff file Expand all files Collapse all files

119
src/scheduler/include/core.h
View file

@ -32,6 +32,7 @@
#include "dataq.h"
#include "mylock.h"
#include "plugin.h"
#include "sched/types.h"
#include "xce/types.h"
#include "zarray.h"
@ -114,4 +115,122 @@ 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*);
/*
* sched_resolve_job_picker - Req 1 AC 6 / Req 2 AC 5 (scheduler-core spec,
* task 2.4). Resolves core->job_pick_plugin against the plugin registry
* (core->plugins, the same plugin_register()/key_find() mechanism every
* other plugin category uses) and validates the result actually implements
* the picker MUST hooks (xce_job_picker_iface_t: next_job, begin_session,
* end_session, on_dispatch, on_pend). Returns 0 and sets *out on success; a
* negative errno on failure:
* -EINVAL bad arguments, or core->job_pick_plugin is unset (NULL), or
* the name IS registered but is not a usable picker
* -ENOENT the configured name is not registered at all
* Implemented in schedobj.c. Declared here (task 4.3, scheduler-core spec)
* -- moved from scheduler.h's xce_scheduler_t* signature to this
* xce_sched_core_t* signature so xce_core_push_events() (core.c, below) can
* call it directly without core.c depending on scheduler.h (forbidden --
* see this header's own boundary comment). Every existing call site already
* held an xce_scheduler_t* and now passes &sched->core instead; no
* resolution logic is duplicated.
*/
extern int sched_resolve_job_picker(xce_sched_core_t*, xce_sched_plugin_t**);
/*
* xce_core_event_type_t / xce_core_event_t - the core's typed input-event
* representation (scheduler-core spec, task 4.3, Req 5 AC 4). Exactly the 9
* event kinds task 4.3 scopes in: BEGIN_ROUND and END_SESSION belong to
* task 4.4's session-start signal, deliberately not added here.
*
* Each event carries a timestamp (`ts`, nanoseconds since epoch) and a
* type-specific payload (Req 5 AC 4's field tables). Six of the nine kinds
* additionally drive a picker state-change hook (xce_core_push_events()'s
* own doc comment, core.c, maps each type to its hook one-for-one);
* WORKERS_PUSH/RESOURCES_PUSH/RESVALS_FULL/RESVALS_DELTA are state-update
* only -- "picking is about job ordering, not resource availability"
* (plugin.h's own Req 1 AC 2 comment).
*/
typedef enum {
XCE_CORE_EVENT_CONFIG_PUSH = 1,
XCE_CORE_EVENT_JOBS_PUSH,
XCE_CORE_EVENT_WORKERS_PUSH,
XCE_CORE_EVENT_RESOURCES_PUSH,
XCE_CORE_EVENT_RESVALS_FULL,
XCE_CORE_EVENT_RESVALS_DELTA,
XCE_CORE_EVENT_JOB_STATE_CHANGE,
XCE_CORE_EVENT_JOB_UPDATE,
XCE_CORE_EVENT_QUEUE_UPDATE,
XCE_CORE_EVENT_JOB_RUSAGE,
} xce_core_event_type_t;
typedef struct {
xce_core_event_type_t type;
uint64_t ts; /* nanoseconds since epoch */
union {
struct {
xce_conf4sched_t* config;
} config_push;
struct {
int njobs;
xce_job_t** jobs;
} jobs_push;
struct {
int nworkers;
xce_worker_t** workers;
} workers_push;
struct {
int n;
xce_resource_t** resources;
} resources_push;
struct {
int n;
xce_resval_t** resvals;
} resvals_full;
struct {
xce_resval_list_t* deltas;
} resvals_delta;
struct {
uint64_t jobid;
uint32_t old_state;
uint32_t new_state;
int32_t exit_code;
} job_state_change;
struct {
uint64_t jobid;
uint32_t changed_fields;
xce_job_t* job;
} job_update;
struct {
xce_queue_t* queue;
} queue_update;
struct {
uint64_t jobid;
xce_job_rusages_t* rusages;
} job_rusage;
} u;
} xce_core_event_t;
/*
* xce_core_push_events - Req 5 AC 4 / AC 8 (scheduler-core spec, task 4.3;
* design.md "Driver -> Core Protocol", "Core Interface Contract"). Batch-
* pushes `nevents` events, applying each one's core-state mutation in
* order and -- for the six event kinds that have one -- synchronously
* calling the resolved picker's corresponding state-change hook exactly
* once per event, in the calling (driver) thread, before this call
* returns. See core.c for the full per-event-type contract and the
* type-to-hook mapping.
*
* Only valid between scheduling sessions (core->session_active == 0):
* this is what makes calling picker hooks synchronously, from the driver
* thread, safe -- the picker thread is blocked on cbuf_get(&wbuf) and the
* allocator thread on cbuf_get(&jbuf), so no session is concurrently
* reading the state this call mutates. Returns -EBUSY (no event applied)
* if a session is active; -EINVAL for bad arguments (core NULL, or events
* NULL with nevents > 0). Per-event failures (e.g. JOB_UPDATE naming an
* unknown jobid) are logged and do not abort the rest of the batch --
* same discipline scheduler.c's existing sched_update() dataq loop
* already follows.
*/
extern int xce_core_push_events(xce_sched_core_t*, xce_core_event_t*, int);
#endif

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

@ -112,21 +112,11 @@ extern int sched_init_config(xce_scheduler_t* sched, xce_conf4sched_t*);
* default to sched->job_pick_plugin when unset. See the field comment on
* xce_scheduler_t above -- this does NOT resolve/register the name. */
extern void sched_set_default_job_pick_plugin(xce_scheduler_t*);
/* Req 1 AC 6 / Req 2 AC 5 (scheduler-core spec, task 2.4): resolves
* sched->job_pick_plugin against the plugin registry (the same
* plugin_register()/key_find() mechanism every other plugin category
* uses) and validates the result actually implements the picker MUST
* hooks (xce_job_picker_iface_t: next_job, begin_session, end_session,
* on_dispatch, on_pend). Returns 0 and sets *out on success; a negative
* errno on failure:
* -EINVAL bad arguments, or sched->job_pick_plugin is unset (NULL),
* or the name IS registered but is not a usable picker
* -ENOENT the configured name is not registered at all
* Called from sched_init() (before the picker thread is created, so an
* unregistered/non-picker name fails scheduler startup synchronously) and
* from sched_job_picker() (the picker thread, at thread start) -- one
* implementation, two call sites, no duplicated lookup logic. */
extern int sched_resolve_job_picker(xce_scheduler_t*, xce_sched_plugin_t**);
/* sched_resolve_job_picker relocated to core.h (scheduler-core spec, task
* 4.3) -- it now operates on xce_sched_core_t*, not xce_scheduler_t*, so
* xce_core_push_events() (core.c) can call it without core.c depending on
* this header. Call sites in this file (sched_job_picker(), sched_allocator(),
* sched_init()) now pass &sched->core. See core.h for the full contract. */
extern int sched_update(xce_scheduler_t*);
extern void sched_go(xce_scheduler_t*);
extern void sched_begin_session(xce_scheduler_t*);

1
src/scheduler/meson.build
View file

@ -3,6 +3,7 @@ scheduler_srcs = [
'src/handler.c',
'src/scheduler.c',
'src/core.c',
'src/core_events.c',
'src/decision.c',
'src/schedobj.c',
'src/plugin.c',

716
src/scheduler/src/core_events.c Normal file
View file

@ -0,0 +1,716 @@
#include <errno.h>
#include <stdlib.h>
#include <string.h>
#include "controller.capnp.h"
#include "core.h"
#include "misc.h"
#include "resobj.h"
#include "sched/types.h"
#include "utils.h"
#include "xcealloc.h"
#include "xstring.h"
/*
* core_events.c - the core event interface and batch event push
* (scheduler-core spec, task 4.3, Req 5 AC 4 / AC 8). Split out of core.c
* (rather than appended there, as design.md's File Structure Plan
* originally sketched) purely to respect this project's 800-line-per-file
* hard cap -- core.c was already 407 lines before this task, and this
* feature's full state-mutation + hook-dispatch contract does not fit
* under that cap in the same file. Same boundary as core.c/core.h
* (design.md "Allowed Dependencies", Req 5 AC 3): zero controller I/O,
* network, filesystem, or wall-clock dependency -- this file must never
* include client.h/netio.h/xceapi.h/delay.h/idgen.h/xce/cluster.h, and
* must never call a wall-clock function directly. (Not yet covered by
* test_core_isolation.c's automated nm(1) audit, which today only inspects
* core.c's own compiled object -- extending that audit to this file is a
* reasonable follow-up, out of this task's scope.)
*
* See core.h for xce_core_event_type_t / xce_core_event_t /
* xce_core_push_events()'s own doc comments (the public contract).
*/
/*
* core_build_job/core_build_queue/core_build_worker - construct core-owned
* scheduling objects from the wire (xce_job_t/xce_queue_t/xce_worker_t)
* payloads carried by JOBS_PUSH/QUEUE_UPDATE/WORKERS_PUSH events.
*
* schedobj.c's make_sched_job()/make_sched_queue()/make_sched_worker()
* already implement this exact field-copy algorithm, but their `void* arg`
* is cast internally to xce_scheduler_t* -- even though their bodies only
* ever touch arg->core.* (never arg->apictx/arg->cluster_cfg), calling them
* with a bare xce_sched_core_t* that is not actually embedded inside a real
* xce_scheduler_t (as this file's own tests build) would rely on treating
* a foreign object as if it had xce_scheduler_t's layout -- undefined
* behavior this file will not introduce, on top of which this file must
* never depend on scheduler.h (this file's own boundary comment) to even
* name xce_scheduler_t in the first place. The three helpers below are the
* same algorithm, parameterized on xce_sched_core_t* directly -- reusing
* sched_get_user()/sched_get_queue()/sched_get_project()/parse_resreq()
* verbatim, not reinventing them.
*
* core_build_job() intentionally does not construct job->schedobj (the
* resreq-tag-dispatch allocation-context object make_sched_job() builds via
* its own make_sched_obj() helper): no picker hook reads job->schedobj
* (verified against every builtin_fcfs.c hook body), and schedobj/
* allocation-context construction belongs to the allocation path
* (scheduler-allocation spec), out of this task's event-interface scope.
*/
static xce_sched_job_t* core_build_job(xce_sched_core_t* core, xce_job_t* j)
{
xce_sched_job_t* job;
char* errmsg = NULL;
int rc;
if ((core == NULL) || (j == NULL) || (j->submit == NULL)) {
return NULL;
}
job = (xce_sched_job_t*)calloc(1, sizeof(*job));
if (job == NULL) {
return NULL;
}
job->jobid = j->jobid;
job->sub_time = j->sub_time;
job->queue_time = j->queue_time;
job->start_time = j->start_time;
job->state = j->state;
job->user = sched_get_user(core, j->submit->user);
job->queue = sched_get_queue(core, j->submit->queue);
job->project = sched_get_project(core, j->submit->project_name);
job->resreq = (xce_resreq_t*)calloc(1, sizeof(*(job->resreq)));
if (job->resreq == NULL) {
FREEUP(job);
return NULL;
}
{
char* resreq_str = j->submit->resreq;
if (NULLSTR(resreq_str)) {
resreq_str = "type == any";
}
rc = parse_resreq(job->resreq, resreq_str, core->resmgr, &errmsg);
}
if (rc < 0) {
ERROR("%s parse_resreq failed for jobid %llu: %s", __func__,
(unsigned long long)j->jobid, errmsg ? errmsg : "(null)");
FREEUP(errmsg);
FREEUP(job->resreq);
FREEUP(job);
return NULL;
}
job->mintasks = j->submit->mintasks;
job->maxtasks = j->submit->maxtasks;
job->priority = 0; /* FIXME: support job level priority (mirrors
* make_sched_job's own FIXME, schedobj.c) */
if (j->allocs != NULL) {
MOVE_PTR(job->allocs, j->allocs);
job->accounts = make_job_account(job->allocs);
}
return job;
}
static xce_sched_queue_t* core_build_queue(xce_queue_t* q)
{
xce_sched_queue_t* queue;
if (q == NULL) {
return NULL;
}
queue = (xce_sched_queue_t*)calloc(1, sizeof(*queue));
if (queue == NULL) {
return NULL;
}
queue->name = makeString(q->queue);
queue->attrs = q->attrs;
queue->status = q->status;
queue->priority = q->priority;
queue->accept_interval = q->accept_interval;
queue->sched_delay = q->sched_delay;
/* Same scope note as make_sched_queue's own (schedobj.c): project_name/
* job_pick_plugin have no wire-config source yet, and (matching
* existing behavior) resreq is not populated here either -- wiring
* these is a later task, not a regression introduced by this one. */
queue->project_name = NULL;
queue->job_pick_plugin = NULL;
return queue;
}
/*
* core_upsert_queue - shared add-or-update logic for one xce_queue_t wire
* payload, used by both the QUEUE_UPDATE event and CONFIG_PUSH's embedded
* config->queues[] array (Req 5 AC 4: CONFIG_PUSH's queues are documented
* to behave the same as individual queue updates, just batched). Looks up
* `updated->queue` in core->queues; if found, updates the existing
* xce_sched_queue_t in place (mirrors schedobj.c's
* sched_update_existing_queue()); otherwise builds+adds a new one via
* core_build_queue()/sched_add_queue() (mirrors scheduler.c's
* sched_init_queues()). Fires on_queue_update(picker, q) once if a picker
* is resolved, matching the QUEUE_UPDATE row's hook contract exactly.
* No-ops (WARN-logged) on a malformed payload or a failed build.
*/
static void core_upsert_queue(xce_sched_core_t* core, xce_sched_plugin_t* picker,
xce_queue_t* updated)
{
xce_sched_queue_t* q;
if ((updated == NULL) || (updated->queue == NULL)) {
WARN("%s NULL queue payload", __func__);
return;
}
q = sched_get_queue(core, updated->queue);
if (q == NULL) {
q = core_build_queue(updated);
if (q == NULL) {
WARN("%s failed to build new queue %s", __func__,
updated->queue);
return;
}
sched_add_queue(core, q);
} else {
q->status = updated->status;
q->priority = updated->priority;
q->accept_interval = updated->accept_interval;
q->sched_delay = updated->sched_delay;
if (updated->resreq != NULL) {
FREEUP(q->resreq);
q->resreq = makeString(updated->resreq);
}
}
if (picker != NULL) {
picker->picker_iface->on_queue_update(picker, q);
}
}
/*
* core_apply_config_resources/core_apply_config_resvals/
* core_apply_config_queues - CONFIG_PUSH's decomposition of a pushed
* xce_conf4sched_t's embedded queues[]/resources[]/resvals[] arrays into
* core state (Req 5 AC 4's CONFIG_PUSH row: "All queues, resource
* thresholds, ..." is part of the payload, and decomposing it into
* core->resmgr/core->sharedInst/core->queues is the core's job, not the
* driver's, since this IS the core-side entry point). These three mirror
* scheduler.c's sched_init_resources()/sched_init_sharedInst()/
* sched_init_queues() exactly (same algorithm, parameterized on
* xce_sched_core_t* directly -- same boundary rationale as
* core_build_job/core_build_queue/core_build_worker above), which is what
* sched_init_config() calls today for this exact same xce_conf4sched_t
* payload on the still-live scheduler.c path. core_apply_config_queues()
* reuses core_upsert_queue() rather than duplicating queue add-or-update
* logic a third time.
*/
static void core_apply_config_resources(xce_sched_core_t* core,
xce_conf4sched_t* config)
{
int i;
for (i = 0; i < config->n_resources; i++) {
xce_resource_t* r = config->resources[i];
xce_resource_t* r1;
if (r == NULL) {
continue;
}
r1 = makeResource(core->resmgr, r);
if (r1 == NULL) {
ERROR("%s add resource %s failed", __func__,
r->name ? r->name : "(null)");
}
}
}
static void core_apply_config_resvals(xce_sched_core_t* core,
xce_conf4sched_t* config)
{
int i;
for (i = 0; i < config->nresvals; i++) {
xce_name_val_t* nv = config->resvals[i];
xce_resource_t* r = NULL;
xce_val_t* val = NULL;
if ((nv == NULL) || (nv->name == NULL)) {
continue;
}
if ((key_find(core->resmgr, (void*)nv->name, (void**)&r) < 0)
|| (r == NULL)) {
WARN("%s unknown resource '%s' in resvals, skipping", __func__,
nv->name);
continue;
}
MOVE_PTR(val, nv->val);
zarray_put(&(core->sharedInst), r->idx, val);
}
}
static void core_apply_config_queues(xce_sched_core_t* core,
xce_sched_plugin_t* picker,
xce_conf4sched_t* config)
{
int i;
for (i = 0; i < config->n_queues; i++) {
if (config->queues[i] == NULL) {
continue;
}
core_upsert_queue(core, picker, config->queues[i]);
}
}
static xce_sched_worker_t* core_build_worker(xce_sched_core_t* core,
xce_worker_t* w)
{
xce_sched_worker_t* worker;
int i;
if ((core == NULL) || (w == NULL)) {
return NULL;
}
worker = (xce_sched_worker_t*)calloc(1, sizeof(*worker));
if (worker == NULL) {
return NULL;
}
worker->name = makeString(w->name);
worker->status = w->status;
worker->idx = -1;
zarray_init(&(worker->resvals), 128,
(zarray_item_killer_t)kill_sched_resval);
zarray_init(&(worker->resInUse), 128,
(zarray_item_killer_t)kill_sched_resval);
for (i = 0; i < w->nresvals_to_sched; i++) {
xce_name_val_t* nv = w->resvals_to_sched[i];
xce_resource_t* r = NULL;
xce_val_t* val = NULL;
if ((nv == NULL) || (nv->name == NULL)) {
continue;
}
if ((key_find(core->resmgr, (void*)nv->name, (void**)&r) < 0)
|| (r == NULL)) {
WARN("%s unknown resource '%s' in worker resvals, skipping",
__func__, nv->name);
continue;
}
MOVE_PTR(val, nv->val);
zarray_put(&(worker->resvals), r->idx, val);
}
return worker;
}
/*
* xce_core_push_events - see core.h's doc comment for the full contract
* (Req 5 AC 4 / AC 8). Per event type:
*
* CONFIG_PUSH core->config = event's config; configver++; then
* decomposes the pushed config's embedded queues[]/
* resources[]/resvals[] arrays into core->queues/
* core->resmgr/core->sharedInst via
* core_apply_config_resources()/
* core_apply_config_resvals()/
* core_apply_config_queues() (mirrors scheduler.c's
* sched_init_resources()/sched_init_sharedInst()/
* sched_init_queues(), the same decomposition
* sched_init_config() already performs for this exact
* payload) -- firing on_queue_update(picker, queue)
* once per embedded queue (core_upsert_queue(), shared
* with QUEUE_UPDATE below), then
* on_config(picker, config) if the hook is non-NULL.
* JOBS_PUSH core_build_job() + sched_add_job() per job, then
* on_job_push(picker, job) once PER JOB in the batch.
* WORKERS_PUSH core_build_worker() + sched_add_worker() per worker.
* No picker hook (Req 1 AC 2: worker updates are not
* forwarded to the picker).
* RESOURCES_PUSH makeResource()/key_find()-update-in-place per
* resource (mirrors scheduler.c's sched_update()
* XCE_SCHED_OP_UPDATE_RESOURCES case exactly). No
* picker hook.
* RESVALS_FULL key_find() + zarray_put(&core->sharedInst, ...) per
* value (mirrors XCE_SCHED_OP_UPDATE_RESVALS_FULL). No
* picker hook.
* RESVALS_DELTA sched_get_worker() + key_find() +
* zarray_put(&worker->resvals, ...) per delta (mirrors
* XCE_SCHED_OP_UPDATE_RESVALS, the existing
* per-worker resval-update case). No picker hook.
* JOB_STATE_CHANGE sched_get_job() + free stale accounts/allocs if the
* new state left RUN + state assignment (mirrors
* scheduler.c's sched_job_state_change()), then
* on_job_state_change(picker, job, new_state).
* JOB_UPDATE sched_get_job() + queue/resreq field-copy (mirrors
* schedobj.c's sched_update_existing_job(), minus its
* plugin_job_update() fan-out call -- that is the OLD
* per-plugin hook category, unrelated to the picker
* interface this event drives), then
* on_job_update(picker, job).
* QUEUE_UPDATE core_upsert_queue(): sched_get_queue(); update in
* place if found, else core_build_queue() +
* sched_add_queue() (mirrors schedobj.c's
* sched_update_existing_queue() for the "modified"
* half, add-or-update per Req 5 AC 4's "added or
* modified"), then on_queue_update(picker, queue).
* JOB_RUSAGE sched_get_job() + rusages ownership transfer (mirrors
* schedobj.c's sched_apply_job_rusage()), then
* on_job_rusage(picker, job, rusages) IF the hook is
* non-NULL -- Req 5 AC 4: this hook is OPTIONAL, a
* NULL hook is skipped, never treated as an error.
*
* A single picker resolution (sched_resolve_job_picker()) is shared across
* the whole batch. If resolution fails, state mutation still proceeds for
* every event (adding a job to core->jobs, say, does not depend on a
* picker existing) but every hook call is skipped for this call, with a
* WARN -- sched_init() already gates on this exact resolution succeeding
* before either scheduling thread is created, so this should not occur in
* practice; handled defensively, same "should never occur" posture
* scheduler.c's sched_allocator() already documents for its own call to
* this same resolver.
*
* Per-event failures (unknown jobid, malformed payload, ...) are logged
* (WARN/ERROR) and do not abort the rest of the batch -- same discipline
* scheduler.c's existing sched_update() dataq loop already follows.
*/
int xce_core_push_events(xce_sched_core_t* core, xce_core_event_t* events,
int nevents)
{
xce_sched_plugin_t* picker = NULL;
int rc;
int i;
if ((core == NULL) || (nevents < 0)
|| ((events == NULL) && (nevents > 0))) {
return -EINVAL;
}
if (atomic_load(&(core->session_active)) != 0) {
return -EBUSY;
}
rc = sched_resolve_job_picker(core, &picker);
if (rc < 0) {
WARN("%s failed to resolve job picker plugin (rc=%d); "
"state-change hooks will be skipped for this batch",
__func__, rc);
picker = NULL;
}
for (i = 0; i < nevents; i++) {
xce_core_event_t* ev = &(events[i]);
switch (ev->type) {
case XCE_CORE_EVENT_CONFIG_PUSH: {
xce_conf4sched_t* config = ev->u.config_push.config;
core->config = config;
core->configver++;
if (config != NULL) {
core_apply_config_resources(core, config);
core_apply_config_resvals(core, config);
core_apply_config_queues(core, picker, config);
}
if ((picker != NULL) && (picker->picker_iface->on_config != NULL)) {
picker->picker_iface->on_config(picker, config);
}
break;
}
case XCE_CORE_EVENT_JOBS_PUSH: {
int j;
for (j = 0; j < ev->u.jobs_push.njobs; j++) {
xce_sched_job_t* job
= core_build_job(core, ev->u.jobs_push.jobs[j]);
if (job == NULL) {
WARN("%s JOBS_PUSH: failed to build job (batch index %d)",
__func__, j);
continue;
}
sched_add_job(core, job);
if (picker != NULL) {
picker->picker_iface->on_job_push(picker, job);
}
}
break;
}
case XCE_CORE_EVENT_WORKERS_PUSH: {
int w;
for (w = 0; w < ev->u.workers_push.nworkers; w++) {
xce_sched_worker_t* worker
= core_build_worker(core, ev->u.workers_push.workers[w]);
if (worker == NULL) {
WARN("%s WORKERS_PUSH: failed to build worker (batch "
"index %d)",
__func__, w);
continue;
}
sched_add_worker(core, worker);
}
break;
}
case XCE_CORE_EVENT_RESOURCES_PUSH: {
int r;
for (r = 0; r < ev->u.resources_push.n; r++) {
xce_resource_t* res = ev->u.resources_push.resources[r];
xce_resource_t* existing;
if (res == NULL) {
continue;
}
existing = makeResource(core->resmgr, res);
if (existing == NULL) {
/* EEXIST: update-in-place on the existing resource
* (add-or-update), mirrors sched_update()'s own
* XCE_SCHED_OP_UPDATE_RESOURCES case. */
xce_resource_t* found = NULL;
if ((key_find(core->resmgr, res->name, (void**)&found)
== 0)
&& (found != NULL)) {
found->valtype = res->valtype;
found->ordtype = res->ordtype;
found->flags = res->flags;
found->interval = res->interval;
} else {
WARN("%s RESOURCES_PUSH: makeResource failed for "
"%s and lookup not found",
__func__, res->name);
}
}
}
break;
}
case XCE_CORE_EVENT_RESVALS_FULL: {
int r;
for (r = 0; r < ev->u.resvals_full.n; r++) {
xce_resval_t* rv = ev->u.resvals_full.resvals[r];
xce_resource_t* res = NULL;
if (rv == NULL) {
continue;
}
if ((key_find(core->resmgr, rv->name, (void**)&res) < 0)
|| (res == NULL)) {
WARN("%s RESVALS_FULL: unknown resource %s, skipping",
__func__, rv->name);
continue;
}
zarray_put(&(core->sharedInst), res->idx, rv->val);
}
break;
}
case XCE_CORE_EVENT_RESVALS_DELTA: {
xce_resval_list_t* deltas = ev->u.resvals_delta.deltas;
int r;
if (deltas == NULL) {
break;
}
for (r = 0; r < deltas->count; r++) {
xce_resval_t* rv = deltas->vals[r];
xce_sched_worker_t* w = NULL;
xce_resource_t* res = NULL;
if (rv == NULL) {
continue;
}
w = sched_get_worker(core, rv->host);
if (w == NULL) {
WARN("%s RESVALS_DELTA: unknown worker %s, skipping",
__func__, rv->host);
continue;
}
if ((key_find(core->resmgr, rv->name, (void**)&res) < 0)
|| (res == NULL)) {
WARN("%s RESVALS_DELTA: unknown resource %s, skipping",
__func__, rv->name);
continue;
}
zarray_put(&(w->resvals), res->idx, rv->val);
}
break;
}
case XCE_CORE_EVENT_JOB_STATE_CHANGE: {
uint64_t jobid = ev->u.job_state_change.jobid;
uint32_t new_state = ev->u.job_state_change.new_state;
xce_sched_job_t* job = sched_get_job(core, jobid);
if (job == NULL) {
WARN("%s JOB_STATE_CHANGE: unknown jobid %llu", __func__,
(unsigned long long)jobid);
break;
}
if (!JOB_IS_RUN(new_state)) {
if (job->accounts != NULL) {
free_job_account(job->accounts);
job->accounts = NULL;
}
if (job->allocs != NULL) {
free_JobAlloc_ptr(&(job->allocs));
}
}
job->state = new_state;
if (picker != NULL) {
picker->picker_iface->on_job_state_change(picker, job,
new_state);
}
break;
}
case XCE_CORE_EVENT_JOB_UPDATE: {
xce_job_t* updated = ev->u.job_update.job;
xce_sched_job_t* job;
if (updated == NULL) {
WARN("%s JOB_UPDATE: NULL job payload", __func__);
break;
}
job = sched_get_job(core, updated->jobid);
if (job == NULL) {
WARN("%s JOB_UPDATE: unknown jobid %llu", __func__,
(unsigned long long)updated->jobid);
break;
}
job->state = updated->state;
if (updated->submit != NULL) {
if (updated->submit->queue != NULL) {
xce_sched_queue_t* q
= sched_get_queue(core, updated->submit->queue);
if (q != NULL) {
job->queue = q;
job->priority = q->priority;
}
}
if (updated->submit->resreq != NULL) {
xce_resreq_t* newreq
= (xce_resreq_t*)calloc(1, sizeof(*newreq));
char* errmsg = NULL;
int prc;
if (newreq == NULL) {
break;
}
prc = parse_resreq(newreq, updated->submit->resreq,
core->resmgr, &errmsg);
if (prc < 0) {
ERROR("%s JOB_UPDATE parse_resreq failed: %s",
__func__, errmsg ? errmsg : "(null)");
FREEUP(errmsg);
FREEUP(newreq);
break;
}
FREEUP(job->resreq);
job->resreq = newreq;
}
}
if (picker != NULL) {
picker->picker_iface->on_job_update(picker, job);
}
break;
}
case XCE_CORE_EVENT_QUEUE_UPDATE: {
xce_queue_t* updated = ev->u.queue_update.queue;
if ((updated == NULL) || (updated->queue == NULL)) {
WARN("%s QUEUE_UPDATE: NULL queue payload", __func__);
break;
}
core_upsert_queue(core, picker, updated);
break;
}
case XCE_CORE_EVENT_JOB_RUSAGE: {
xce_job_rusages_t* rusages = ev->u.job_rusage.rusages;
xce_sched_job_t* job;
if (rusages == NULL) {
WARN("%s JOB_RUSAGE: NULL rusages payload", __func__);
break;
}
job = sched_get_job(core, rusages->jobid);
if (job == NULL) {
WARN("%s JOB_RUSAGE: unknown jobid %llu", __func__,
(unsigned long long)rusages->jobid);
break;
}
/* job takes ownership of rusages, same documented contract as
* sched_apply_job_rusage() (schedobj.c). */
if (job->rusages != rusages) {
free_JobRusages_ptr(&(job->rusages));
}
job->rusages = rusages;
if ((picker != NULL)
&& (picker->picker_iface->on_job_rusage != NULL)) {
picker->picker_iface->on_job_rusage(picker, job, rusages);
}
break;
}
default:
WARN("%s unknown event type %d at batch index %d, skipping",
__func__, (int)ev->type, i);
break;
}
}
return 0;
}

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

@ -367,37 +367,46 @@ void sched_set_default_job_pick_plugin(xce_scheduler_t* sched)
/*
* sched_resolve_job_picker - Req 1 AC 6 / Req 2 AC 5 (scheduler-core spec,
* task 2.4). See the declaration comment in scheduler.h for the full
* contract.
* task 2.4). See the declaration comment in core.h for the full contract.
*
* Kept here (rather than in scheduler.c, where both its callers --
* Kept here (rather than in scheduler.c, where its scheduler.c call sites --
* sched_init() and sched_job_picker() -- live) for the same testability
* reason sched_set_default_job_pick_plugin() was: this function only needs
* 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.
* core->job_pick_plugin and 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.
*
* Task 4.3 (scheduler-core spec): signature changed from xce_scheduler_t*
* to xce_sched_core_t* -- the function's own body only ever touched
* sched->core.job_pick_plugin/sched->core.plugins, never any other
* xce_scheduler_t field, so this is a pure narrowing, not a behavior
* change. Every call site already held an xce_scheduler_t* and now passes
* &sched->core instead (scheduler.c's sched_job_picker()/sched_allocator()/
* sched_init(), and this file's own callers in the white-box test suite).
* Declared in core.h now (moved from scheduler.h) so xce_core_push_events()
* (core.c) -- which must never depend on scheduler.h -- can call it
* directly.
*/
int sched_resolve_job_picker(xce_scheduler_t* sched, xce_sched_plugin_t** out)
int sched_resolve_job_picker(xce_sched_core_t* core, xce_sched_plugin_t** out)
{
xce_sched_plugin_t* plugin = NULL;
int rc = 0;
if ((sched == NULL) || (out == NULL)) {
if ((core == NULL) || (out == NULL)) {
return -EINVAL;
}
*out = NULL;
if (sched->core.job_pick_plugin == NULL) {
if (core->job_pick_plugin == NULL) {
return -EINVAL;
}
lock(sched->core.plugins, RTTI_LOCK_READ);
rc = key_find(sched->core.plugins, sched->core.job_pick_plugin,
(void**)&plugin);
unlock(sched->core.plugins);
lock(core->plugins, RTTI_LOCK_READ);
rc = key_find(core->plugins, core->job_pick_plugin, (void**)&plugin);
unlock(core->plugins);
if (rc <= 0) {
return (rc < 0) ? rc : -ENOENT;

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

@ -79,7 +79,7 @@ static void* sched_job_picker(void* arg)
xce_sched_plugin_t* job_picker = NULL;
int rc = 0;
rc = sched_resolve_job_picker(sched, &job_picker);
rc = sched_resolve_job_picker(&(sched->core), &job_picker);
if (rc < 0) {
ERROR("%s failed to resolve job picker plugin '%s' (rc=%d)", __func__,
sched->core.job_pick_plugin ? sched->core.job_pick_plugin
@ -309,7 +309,7 @@ static void* sched_allocator(void* arg)
* exact resolution succeeding (before either thread is started), so
* failure here should not occur in practice; handled the same
* fail-return-NULL way sched_job_picker() handles it, for symmetry. */
rc = sched_resolve_job_picker(sched, &job_picker);
rc = sched_resolve_job_picker(&(sched->core), &job_picker);
if (rc < 0) {
ERROR("%s failed to resolve job picker plugin '%s' (rc=%d)", __func__,
sched->core.job_pick_plugin ? sched->core.job_pick_plugin
@ -532,7 +532,7 @@ int sched_init(xce_scheduler_t* sched, xce_client_config_t* config,
* fast, without first paying for a controller connection attempt. */
sched_set_default_job_pick_plugin(sched);
rc = sched_resolve_job_picker(sched, &resolved_picker);
rc = sched_resolve_job_picker(&(sched->core), &resolved_picker);
if (rc < 0) {
ERROR("%s configured picker plugin '%s' failed to resolve as a "
"valid registered picker (rc=%d)",

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

@ -225,6 +225,22 @@ sched_core_jbuf_capacity_pin_test = executable(
+ '"'],
install: false)
# Contract test for the core event interface and batch event push
# (scheduler-core spec, task 4.3, Req 5 AC 4 / AC 8): xce_core_push_events()
# fires the resolved picker's corresponding state-change hook exactly once
# per event, synchronously, before it returns, and rejects with -EBUSY (no
# event applied) when a session is active. Same white-box #include pattern
# as test_core_jbuf_capacity_pin.c/test_core_isolation.c (pulls the real
# ../src/core.c into this TU) -- see the test's own header comment for why
# it provides its own real sched_resolve_job_picker() rather than
# #include-ing schedobj.c.
sched_core_events_test = executable(
'sched_core_events_test',
['test_core_events.c'],
include_directories:[incdir],
dependencies: [scheduler_test_deps],
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',
@ -261,3 +277,6 @@ test('scheduler core isolation test',
test('scheduler core jbuf capacity pin test',
sched_core_jbuf_capacity_pin_test, suite: 'scheduler',
env: ['ASAN_OPTIONS=detect_leaks=0'])
test('scheduler core events test',
sched_core_events_test, suite: 'scheduler',
env: ['ASAN_OPTIONS=detect_leaks=0'])

714
src/scheduler/testsuite/test_core_events.c Normal file
View file

@ -0,0 +1,714 @@
/*
* test_core_events.c - contract test for the core event interface and
* batch event push (scheduler-core spec, task 4.3, Req 5 AC 4 / AC 8).
*
* Observable under test: pushing a batch of events with the picker idle
* (core->session_active == 0) fires the corresponding state-change hook
* exactly once per event, synchronously, before xce_core_push_events()
* returns -- and the session-active rejection path (-EBUSY, no event
* applied) when a session IS active.
*
* White-box #include of both ../src/core.c and ../src/core_events.c (same
* convention as test_core_isolation.c / test_core_jbuf_capacity_pin.c) so
* this test exercises the REAL xce_sched_core_init()/xce_core_push_events(),
* not a re-implementation -- xce_core_push_events() lives in core_events.c
* (split out of core.c purely to respect the 800-line-per-file cap; see
* that file's own header comment), which itself calls straight back into
* core.c's real sched_get_job()/sched_add_job()/etc. accessors, all within
* this one translation unit. Stubs below are only for scheduler-internal
* externs these two files reference (as kvmgr datakiller callback
* addresses, or -- for sched_resolve_job_picker -- because the real
* implementation lives in schedobj.c, a separate translation unit this
* file does not pull in):
*
* - kill_sched_job/kill_sched_queue/kill_sched_worker/kill_sched_user/
* kill_sched_project/kill_sched_resval/plugin_unregister: pure no-op
* stubs, identical to test_core_jbuf_capacity_pin.c's own -- this
* file's fixtures are never torn down (never call
* xce_sched_core_fini()), same "fixtures aren't torn down" convention
* established across this directory, so these killers are never
* actually invoked; present only to satisfy the linker for
* xce_sched_core_init()'s kvmgr_params_t.datakiller fields.
* - sched_resolve_job_picker: a real, working reimplementation of
* schedobj.c's function of the same name (identical algorithm --
* key_find core->job_pick_plugin in core->plugins, validate the
* picker MUST hooks) -- needed because the true implementation lives
* in schedobj.c, and pulling that whole file in would drag in its own
* large stub surface (plugin_queue_update, plugin_job_update, etc.)
* for functions this file's tests never exercise. This IS real logic,
* not a null stub, because the hook-firing contract under test
* depends on picker resolution actually working.
*
* make_job_account/free_job_account/free_JobAlloc_ptr/free_JobRusages_ptr/
* makeResource/parse_resreq/makeString are all real, linked from the
* shared libxce library (src/lib), not stubbed -- same as every other
* scheduler white-box test in this directory.
*/
#include <errno.h>
#include <stdint.h>
#include <stdlib.h>
#include <string.h>
#include <unity/unity.h>
#include "kvmgr.h"
#include "scheduler.h"
#include "xstring.h"
/* Pull in the real xce_sched_core_init()/xce_core_push_events() under test
* into this TU (white-box #include, same convention as
* test_core_jbuf_capacity_pin.c's `#include "../src/core.c"`). */
#include "../src/core.c"
#include "../src/core_events.c"
void kill_sched_job(void* arg) { (void)arg; }
void kill_sched_queue(void* arg) { (void)arg; }
void kill_sched_worker(void* arg) { (void)arg; }
void kill_sched_user(void* arg) { (void)arg; }
void kill_sched_project(void* arg) { (void)arg; }
void kill_sched_resval(xce_val_t* val) { (void)val; }
void plugin_unregister(void* arg) { (void)arg; }
/* Real reimplementation of schedobj.c's sched_resolve_job_picker -- see
* file header comment for why this file provides its own copy instead of
* #include-ing schedobj.c. */
int sched_resolve_job_picker(xce_sched_core_t* core, xce_sched_plugin_t** out)
{
xce_sched_plugin_t* plugin = NULL;
int rc = 0;
if ((core == NULL) || (out == NULL)) {
return -EINVAL;
}
*out = NULL;
if (core->job_pick_plugin == NULL) {
return -EINVAL;
}
lock(core->plugins, RTTI_LOCK_READ);
rc = key_find(core->plugins, core->job_pick_plugin, (void**)&plugin);
unlock(core->plugins);
if (rc <= 0) {
return (rc < 0) ? rc : -ENOENT;
}
if ((plugin == NULL) || (plugin->picker_iface == NULL)
|| (plugin->picker_iface->next_job == NULL)
|| (plugin->picker_iface->begin_session == NULL)
|| (plugin->picker_iface->end_session == NULL)
|| (plugin->picker_iface->on_dispatch == NULL)
|| (plugin->picker_iface->on_pend == NULL)) {
return -EINVAL;
}
*out = plugin;
return 0;
}
void setUp(void) { return; }
void tearDown(void) { return; }
/* ---------------------------------------------------------------------- */
/* Test picker: records call counts + last-seen arguments per hook, so */
/* tests can assert "exactly once, with the right argument" per event. */
/* ---------------------------------------------------------------------- */
typedef struct {
int config_calls;
int job_push_calls;
int job_state_change_calls;
int job_update_calls;
int queue_update_calls;
int job_rusage_calls;
xce_conf4sched_t* last_config;
xce_sched_job_t* last_job_push;
xce_sched_job_t* last_job_state_change_job;
uint32_t last_job_state_change_state;
xce_sched_job_t* last_job_update;
xce_sched_queue_t* last_queue_update;
xce_sched_job_t* last_job_rusage_job;
xce_job_rusages_t* last_job_rusage_rusages;
} test_picker_counts_t;
static test_picker_counts_t g_counts;
static int test_begin_session(void* picker)
{
(void)picker;
return 0;
}
static xce_sched_job_t* test_next_job(void* picker)
{
(void)picker;
return NULL;
}
static int test_end_session(void* picker)
{
(void)picker;
return 0;
}
static int test_on_dispatch(void* picker, uint64_t jobid)
{
(void)picker;
(void)jobid;
return 0;
}
static int test_on_pend(void* picker, uint64_t jobid)
{
(void)picker;
(void)jobid;
return 0;
}
static int test_on_config(void* picker, xce_conf4sched_t* config)
{
(void)picker;
g_counts.config_calls++;
g_counts.last_config = config;
return 0;
}
static int test_on_job_push(void* picker, xce_sched_job_t* job)
{
(void)picker;
g_counts.job_push_calls++;
g_counts.last_job_push = job;
return 0;
}
static int test_on_job_state_change(void* picker, xce_sched_job_t* job,
uint32_t state)
{
(void)picker;
g_counts.job_state_change_calls++;
g_counts.last_job_state_change_job = job;
g_counts.last_job_state_change_state = state;
return 0;
}
static int test_on_job_update(void* picker, xce_sched_job_t* job)
{
(void)picker;
g_counts.job_update_calls++;
g_counts.last_job_update = job;
return 0;
}
static int test_on_queue_update(void* picker, xce_sched_queue_t* queue)
{
(void)picker;
g_counts.queue_update_calls++;
g_counts.last_queue_update = queue;
return 0;
}
static int test_on_job_rusage(void* picker, xce_sched_job_t* job,
xce_job_rusages_t* rusages)
{
(void)picker;
g_counts.job_rusage_calls++;
g_counts.last_job_rusage_job = job;
g_counts.last_job_rusage_rusages = rusages;
return 0;
}
/* Full picker: every hook wired, including the two optional ones
* (on_config/on_job_rusage). */
static xce_job_picker_iface_t g_full_picker_iface = {
.init = NULL,
.fini = NULL,
.begin_session = test_begin_session,
.next_job = test_next_job,
.end_session = test_end_session,
.on_dispatch = test_on_dispatch,
.on_pend = test_on_pend,
.on_queue_update = test_on_queue_update,
.on_job_push = test_on_job_push,
.on_job_state_change = test_on_job_state_change,
.on_job_update = test_on_job_update,
.on_config = test_on_config,
.on_job_rusage = test_on_job_rusage,
};
/* Picker with on_job_rusage left NULL (Req 5 AC 4: "optional -- NULL if
* not registered") -- proves JOB_RUSAGE applies state without erroring
* when the hook is absent. */
static xce_job_picker_iface_t g_no_rusage_picker_iface = {
.init = NULL,
.fini = NULL,
.begin_session = test_begin_session,
.next_job = test_next_job,
.end_session = test_end_session,
.on_dispatch = test_on_dispatch,
.on_pend = test_on_pend,
.on_queue_update = test_on_queue_update,
.on_job_push = test_on_job_push,
.on_job_state_change = test_on_job_state_change,
.on_job_update = test_on_job_update,
.on_config = test_on_config,
.on_job_rusage = NULL,
};
static xce_sched_plugin_t g_full_plugin
= { .name = "test_full_picker", .picker_iface = &g_full_picker_iface };
static xce_sched_plugin_t g_no_rusage_plugin
= { .name = "test_no_rusage_picker",
.picker_iface = &g_no_rusage_picker_iface };
/* ---------------------------------------------------------------------- */
/* Fixture: real xce_sched_core_init(), a test picker registered directly */
/* into core.plugins (key_add -- the same mechanism plugin_register() */
/* itself uses, schedobj.c not included so its wrapper is not available), */
/* and core.job_pick_plugin pointed at it. Never torn down (this */
/* directory's established convention -- see test_core_jbuf_capacity_pin.c */
/* header comment). */
/* ---------------------------------------------------------------------- */
static void fixture_init(xce_sched_core_t* core, xce_sched_plugin_t* plugin)
{
int rc;
memset(&g_counts, 0, sizeof(g_counts));
rc = xce_sched_core_init(core);
TEST_ASSERT_EQUAL_INT(0, rc);
lock(core->plugins, RTTI_LOCK_WRITE);
rc = key_add(core->plugins, (void*)plugin->name, (void*)plugin);
unlock(core->plugins);
TEST_ASSERT_GREATER_OR_EQUAL(0, rc);
core->job_pick_plugin = plugin->name;
}
static xce_job_t* make_job(uint64_t jobid, uint32_t state, char* queue,
char* user, char* project)
{
xce_job_t* j = (xce_job_t*)calloc(1, sizeof(*j));
j->jobid = jobid;
j->state = state;
j->sub_time = 1000;
j->queue_time = 1000;
j->start_time = 0;
j->submit = (job_submit_data_t*)calloc(1, sizeof(*(j->submit)));
j->submit->queue = queue;
j->submit->user = user;
j->submit->project_name = project;
j->submit->resreq = "type == any";
j->submit->mintasks = 1;
j->submit->maxtasks = 1;
return j;
}
/* ---------------------------------------------------------------------- */
/* Tests */
/* ---------------------------------------------------------------------- */
/* One event of each of the 6 hook-having kinds in a single batch --
* asserts each corresponding hook fired EXACTLY once, synchronously
* (already true by the time push_events returns, in this single-threaded
* test), with the right argument. */
static void
test_push_events_fires_each_hook_once_per_event_with_picker_idle(void)
{
xce_sched_core_t core;
xce_core_event_t events[6];
xce_conf4sched_t config;
xce_job_t* job1;
xce_queue_t wire_queue;
xce_job_t job_update_payload;
xce_job_rusages_t rusages;
int rc;
fixture_init(&core, &g_full_plugin);
memset(&config, 0, sizeof(config));
job1 = make_job(42, JOB_STAT_PEND, "normal", "alice", "proj1");
memset(&wire_queue, 0, sizeof(wire_queue));
wire_queue.queue = "normal";
wire_queue.priority = 5;
wire_queue.status = QUEUE_STAT_OPEN;
memset(events, 0, sizeof(events));
events[0].type = XCE_CORE_EVENT_CONFIG_PUSH;
events[0].u.config_push.config = &config;
events[1].type = XCE_CORE_EVENT_QUEUE_UPDATE;
events[1].u.queue_update.queue = &wire_queue;
events[2].type = XCE_CORE_EVENT_JOBS_PUSH;
events[2].u.jobs_push.njobs = 1;
events[2].u.jobs_push.jobs = &job1;
events[3].type = XCE_CORE_EVENT_JOB_STATE_CHANGE;
events[3].u.job_state_change.jobid = 42;
events[3].u.job_state_change.old_state = JOB_STAT_PEND;
events[3].u.job_state_change.new_state = JOB_STAT_RUN;
events[3].u.job_state_change.exit_code = 0;
memset(&job_update_payload, 0, sizeof(job_update_payload));
job_update_payload.jobid = 42;
job_update_payload.state = JOB_STAT_RUN;
events[4].type = XCE_CORE_EVENT_JOB_UPDATE;
events[4].u.job_update.jobid = 42;
events[4].u.job_update.changed_fields = 0;
events[4].u.job_update.job = &job_update_payload;
memset(&rusages, 0, sizeof(rusages));
rusages.jobid = 42;
events[5].type = XCE_CORE_EVENT_JOB_RUSAGE;
events[5].u.job_rusage.jobid = 42;
events[5].u.job_rusage.rusages = &rusages;
rc = xce_core_push_events(&core, events, 6);
TEST_ASSERT_EQUAL_INT(0, rc);
TEST_ASSERT_EQUAL_INT(1, g_counts.config_calls);
TEST_ASSERT_EQUAL_PTR(&config, g_counts.last_config);
TEST_ASSERT_EQUAL_INT(1, g_counts.queue_update_calls);
TEST_ASSERT_NOT_NULL(g_counts.last_queue_update);
TEST_ASSERT_EQUAL_STRING("normal", g_counts.last_queue_update->name);
TEST_ASSERT_EQUAL_INT(1, g_counts.job_push_calls);
TEST_ASSERT_NOT_NULL(g_counts.last_job_push);
TEST_ASSERT_EQUAL_UINT64(42, g_counts.last_job_push->jobid);
TEST_ASSERT_EQUAL_INT(1, g_counts.job_state_change_calls);
TEST_ASSERT_NOT_NULL(g_counts.last_job_state_change_job);
TEST_ASSERT_EQUAL_UINT64(42, g_counts.last_job_state_change_job->jobid);
TEST_ASSERT_EQUAL_UINT32(JOB_STAT_RUN, g_counts.last_job_state_change_state);
/* state mutation actually applied on the core-owned job object */
TEST_ASSERT_EQUAL_UINT32(JOB_STAT_RUN, sched_get_job(&core, 42)->state);
TEST_ASSERT_EQUAL_INT(1, g_counts.job_update_calls);
TEST_ASSERT_NOT_NULL(g_counts.last_job_update);
TEST_ASSERT_EQUAL_UINT64(42, g_counts.last_job_update->jobid);
TEST_ASSERT_EQUAL_INT(1, g_counts.job_rusage_calls);
TEST_ASSERT_NOT_NULL(g_counts.last_job_rusage_job);
TEST_ASSERT_EQUAL_PTR(&rusages, g_counts.last_job_rusage_rusages);
TEST_ASSERT_EQUAL_PTR(&rusages, sched_get_job(&core, 42)->rusages);
}
/* JOBS_PUSH with njobs > 1: on_job_push must fire once PER JOB, not once
* per event -- Req 5 AC 4's "on_job_push(picker, job) for each job in the
* batch". */
static void test_jobs_push_fires_on_job_push_once_per_job(void)
{
xce_sched_core_t core;
xce_core_event_t event;
xce_job_t* jobs[3];
int rc;
fixture_init(&core, &g_full_plugin);
jobs[0] = make_job(1, JOB_STAT_PEND, "normal", "alice", "proj1");
jobs[1] = make_job(2, JOB_STAT_PEND, "normal", "bob", "proj1");
jobs[2] = make_job(3, JOB_STAT_PEND, "normal", "carol", "proj1");
memset(&event, 0, sizeof(event));
event.type = XCE_CORE_EVENT_JOBS_PUSH;
event.u.jobs_push.njobs = 3;
event.u.jobs_push.jobs = jobs;
rc = xce_core_push_events(&core, &event, 1);
TEST_ASSERT_EQUAL_INT(0, rc);
TEST_ASSERT_EQUAL_INT(3, g_counts.job_push_calls);
TEST_ASSERT_NOT_NULL(sched_get_job(&core, 1));
TEST_ASSERT_NOT_NULL(sched_get_job(&core, 2));
TEST_ASSERT_NOT_NULL(sched_get_job(&core, 3));
}
/* WORKERS_PUSH/RESOURCES_PUSH/RESVALS_FULL/RESVALS_DELTA are state-update
* only -- Req 1 AC 2: "Hooks not defined on the picker interface (worker
* updates, resource updates, resval updates) are not forwarded to the
* picker." Every hook counter must stay at 0, while the underlying state
* IS genuinely mutated (not a no-op placeholder). */
static void test_non_picker_events_mutate_state_but_never_call_hooks(void)
{
xce_sched_core_t core;
xce_core_event_t events[4];
xce_worker_t wire_worker;
xce_worker_t* worker_ptrs[1];
xce_resource_t wire_resource;
xce_resource_t* resource_ptrs[1];
xce_resval_t full_resval;
xce_resval_t* full_vals[1];
xce_resval_t delta_resval;
xce_resval_t* delta_vals[1];
xce_resval_list_t deltas;
xce_val_t full_val;
xce_val_t delta_val;
int rc;
fixture_init(&core, &g_full_plugin);
memset(&wire_worker, 0, sizeof(wire_worker));
wire_worker.name = "hostA";
wire_worker.status = 0;
wire_worker.nresvals_to_sched = 0;
worker_ptrs[0] = &wire_worker;
memset(&wire_resource, 0, sizeof(wire_resource));
wire_resource.name = "cpu";
wire_resource.idx = 0;
resource_ptrs[0] = &wire_resource;
memset(&full_val, 0, sizeof(full_val));
memset(&full_resval, 0, sizeof(full_resval));
full_resval.name = "cpu";
full_resval.val = &full_val;
full_vals[0] = &full_resval;
memset(&delta_val, 0, sizeof(delta_val));
memset(&delta_resval, 0, sizeof(delta_resval));
delta_resval.name = "cpu";
delta_resval.host = "hostA";
delta_resval.val = &delta_val;
delta_vals[0] = &delta_resval;
deltas.count = 1;
deltas.vals = delta_vals;
memset(events, 0, sizeof(events));
events[0].type = XCE_CORE_EVENT_WORKERS_PUSH;
events[0].u.workers_push.nworkers = 1;
events[0].u.workers_push.workers = worker_ptrs;
events[1].type = XCE_CORE_EVENT_RESOURCES_PUSH;
events[1].u.resources_push.n = 1;
events[1].u.resources_push.resources = resource_ptrs;
events[2].type = XCE_CORE_EVENT_RESVALS_FULL;
events[2].u.resvals_full.n = 1;
events[2].u.resvals_full.resvals = full_vals;
events[3].type = XCE_CORE_EVENT_RESVALS_DELTA;
events[3].u.resvals_delta.deltas = &deltas;
/* worker must exist before RESVALS_DELTA can attribute a value to it */
rc = xce_core_push_events(&core, events, 1); /* WORKERS_PUSH only, first */
TEST_ASSERT_EQUAL_INT(0, rc);
rc = xce_core_push_events(&core, &events[1], 3); /* RESOURCES/RESVALS_FULL/RESVALS_DELTA */
TEST_ASSERT_EQUAL_INT(0, rc);
TEST_ASSERT_EQUAL_INT(0, g_counts.config_calls);
TEST_ASSERT_EQUAL_INT(0, g_counts.job_push_calls);
TEST_ASSERT_EQUAL_INT(0, g_counts.job_state_change_calls);
TEST_ASSERT_EQUAL_INT(0, g_counts.job_update_calls);
TEST_ASSERT_EQUAL_INT(0, g_counts.queue_update_calls);
TEST_ASSERT_EQUAL_INT(0, g_counts.job_rusage_calls);
TEST_ASSERT_NOT_NULL(sched_get_worker(&core, "hostA"));
}
/* JOB_RUSAGE's on_job_rusage hook is OPTIONAL (Req 5 AC 4) -- a picker
* without it must still apply the state mutation and must not error or
* crash. */
static void test_job_rusage_applies_state_when_hook_is_null(void)
{
xce_sched_core_t core;
xce_core_event_t events[2];
xce_job_t* job1;
xce_job_rusages_t rusages;
int rc;
fixture_init(&core, &g_no_rusage_plugin);
job1 = make_job(7, JOB_STAT_RUN, "normal", "alice", "proj1");
memset(&rusages, 0, sizeof(rusages));
rusages.jobid = 7;
memset(events, 0, sizeof(events));
events[0].type = XCE_CORE_EVENT_JOBS_PUSH;
events[0].u.jobs_push.njobs = 1;
events[0].u.jobs_push.jobs = &job1;
events[1].type = XCE_CORE_EVENT_JOB_RUSAGE;
events[1].u.job_rusage.jobid = 7;
events[1].u.job_rusage.rusages = &rusages;
rc = xce_core_push_events(&core, events, 2);
TEST_ASSERT_EQUAL_INT(0, rc);
TEST_ASSERT_EQUAL_INT(0, g_counts.job_rusage_calls);
TEST_ASSERT_EQUAL_PTR(&rusages, sched_get_job(&core, 7)->rusages);
}
/* CONFIG_PUSH must decompose the pushed xce_conf4sched_t's embedded
* queues[]/resources[]/resvals[] arrays into core state -- not just store
* the config pointer and bump configver (Req 5 AC 4's CONFIG_PUSH row:
* "All queues, resource thresholds, ... " is part of the payload, and
* decomposing it is the core's job, mirroring scheduler.c's
* sched_init_resources()/sched_init_sharedInst()/sched_init_queues(), all
* three of which sched_init_config() calls today for this exact same
* xce_conf4sched_t payload). Also asserts on_queue_update fires once per
* embedded queue (design intent: a config push's queues behave the same as
* individual queue updates, just batched). */
static void test_config_push_decomposes_queues_resources_and_resvals(void)
{
xce_sched_core_t core;
xce_core_event_t event;
xce_conf4sched_t config;
xce_resource_t wire_resource;
xce_resource_t* resources[1];
xce_val_t resval_value;
xce_name_val_t wire_resval;
xce_name_val_t* resvals[1];
xce_queue_t wire_queue;
xce_queue_t* queues[1];
xce_resource_t* found_resource = NULL;
xce_sched_queue_t* found_queue;
xce_val_t* shared_val;
int rc;
fixture_init(&core, &g_full_plugin);
memset(&wire_resource, 0, sizeof(wire_resource));
wire_resource.name = "cpu";
resources[0] = &wire_resource;
memset(&resval_value, 0, sizeof(resval_value));
resval_value.u.ivalue32 = 8;
memset(&wire_resval, 0, sizeof(wire_resval));
wire_resval.name = "cpu";
wire_resval.val = &resval_value;
resvals[0] = &wire_resval;
memset(&wire_queue, 0, sizeof(wire_queue));
wire_queue.queue = "batch";
wire_queue.priority = 7;
wire_queue.status = QUEUE_STAT_OPEN;
queues[0] = &wire_queue;
memset(&config, 0, sizeof(config));
config.n_resources = 1;
config.resources = resources;
config.nresvals = 1;
config.resvals = resvals;
config.n_queues = 1;
config.queues = queues;
memset(&event, 0, sizeof(event));
event.type = XCE_CORE_EVENT_CONFIG_PUSH;
event.u.config_push.config = &config;
rc = xce_core_push_events(&core, &event, 1);
TEST_ASSERT_EQUAL_INT(0, rc);
TEST_ASSERT_EQUAL_INT(1, g_counts.config_calls);
/* resource decomposed into core->resmgr */
rc = key_find(core.resmgr, (void*)"cpu", (void**)&found_resource);
TEST_ASSERT_GREATER_THAN_INT(0, rc);
TEST_ASSERT_NOT_NULL(found_resource);
/* resval decomposed into core->sharedInst at the resource's idx */
shared_val = (xce_val_t*)zarray_get(&(core.sharedInst), found_resource->idx);
TEST_ASSERT_NOT_NULL(shared_val);
TEST_ASSERT_EQUAL_INT(8, shared_val->u.ivalue32);
/* queue decomposed into core->queues, and on_queue_update fired once */
found_queue = sched_get_queue(&core, "batch");
TEST_ASSERT_NOT_NULL(found_queue);
TEST_ASSERT_EQUAL_INT(7, found_queue->priority);
TEST_ASSERT_EQUAL_INT(1, g_counts.queue_update_calls);
TEST_ASSERT_EQUAL_PTR(found_queue, g_counts.last_queue_update);
}
/* Session-active rejection (Req 5 AC 8 / design.md precondition):
* core_push_events() is only valid between sessions. */
static void
test_push_events_rejects_with_ebusy_when_session_active_no_mutation(void)
{
xce_sched_core_t core;
xce_core_event_t event;
xce_job_t* job1;
int rc;
fixture_init(&core, &g_full_plugin);
atomic_store(&(core.session_active), 1);
job1 = make_job(99, JOB_STAT_PEND, "normal", "alice", "proj1");
memset(&event, 0, sizeof(event));
event.type = XCE_CORE_EVENT_JOBS_PUSH;
event.u.jobs_push.njobs = 1;
event.u.jobs_push.jobs = &job1;
rc = xce_core_push_events(&core, &event, 1);
TEST_ASSERT_EQUAL_INT(-EBUSY, rc);
TEST_ASSERT_EQUAL_INT(0, g_counts.job_push_calls);
TEST_ASSERT_NULL(sched_get_job(&core, 99));
}
/* Bad-argument guard clauses. */
static void test_push_events_null_core_returns_einval(void)
{
xce_core_event_t event;
int rc;
memset(&event, 0, sizeof(event));
event.type = XCE_CORE_EVENT_CONFIG_PUSH;
rc = xce_core_push_events(NULL, &event, 1);
TEST_ASSERT_EQUAL_INT(-EINVAL, rc);
}
static void test_push_events_null_events_with_positive_nevents_returns_einval(void)
{
xce_sched_core_t core;
int rc;
fixture_init(&core, &g_full_plugin);
rc = xce_core_push_events(&core, NULL, 1);
TEST_ASSERT_EQUAL_INT(-EINVAL, rc);
}
/* Zero-length batch is a valid no-op (not an error). */
static void test_push_events_zero_nevents_is_a_valid_noop(void)
{
xce_sched_core_t core;
int rc;
fixture_init(&core, &g_full_plugin);
rc = xce_core_push_events(&core, NULL, 0);
TEST_ASSERT_EQUAL_INT(0, rc);
}
int main(void)
{
UNITY_BEGIN();
RUN_TEST(test_push_events_fires_each_hook_once_per_event_with_picker_idle);
RUN_TEST(test_jobs_push_fires_on_job_push_once_per_job);
RUN_TEST(test_non_picker_events_mutate_state_but_never_call_hooks);
RUN_TEST(test_job_rusage_applies_state_when_hook_is_null);
RUN_TEST(test_config_push_decomposes_queues_resources_and_resvals);
RUN_TEST(test_push_events_rejects_with_ebusy_when_session_active_no_mutation);
RUN_TEST(test_push_events_null_core_returns_einval);
RUN_TEST(test_push_events_null_events_with_positive_nevents_returns_einval);
RUN_TEST(test_push_events_zero_nevents_is_a_valid_noop);
return UNITY_END();
}

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

@ -297,7 +297,7 @@ static void test_unset_job_pick_plugin_resolves_to_builtin_fcfs_by_default(void)
sched_set_default_job_pick_plugin(&sched);
TEST_ASSERT_EQUAL_STRING("builtin_fcfs", sched.core.job_pick_plugin);
rc = sched_resolve_job_picker(&sched, &resolved);
rc = sched_resolve_job_picker(&sched.core, &resolved);
TEST_ASSERT_EQUAL_INT(0, rc);
TEST_ASSERT_NOT_NULL(resolved);
@ -311,7 +311,7 @@ static void test_unset_job_pick_plugin_resolves_to_builtin_fcfs_by_default(void)
/* tasks.md Observable, half 2: "starting it with an unregistered picker
* name fails startup with a clear error instead of silently falling
* back." -ENOENT specifically (distinct from -EINVAL below) --
* sched_resolve_job_picker's documented contract (scheduler.h). */
* sched_resolve_job_picker's documented contract (core.h). */
static void
test_unregistered_configured_name_fails_with_enoent_not_silent_fallback(void)
{
@ -324,7 +324,7 @@ test_unregistered_configured_name_fails_with_enoent_not_silent_fallback(void)
sched.core.job_pick_plugin = "totally_unregistered_picker_name";
rc = sched_resolve_job_picker(&sched, &resolved);
rc = sched_resolve_job_picker(&sched.core, &resolved);
TEST_ASSERT_EQUAL_INT(-ENOENT, rc);
/* Not a silent fallback to builtin_fcfs (still registered) or to
@ -358,7 +358,7 @@ static void test_registered_but_not_a_picker_fails_with_einval_not_enoent(void)
sched.core.job_pick_plugin = "not_a_picker";
rc = sched_resolve_job_picker(&sched, &resolved);
rc = sched_resolve_job_picker(&sched.core, &resolved);
TEST_ASSERT_EQUAL_INT(-EINVAL, rc);
TEST_ASSERT_NULL(resolved);
@ -377,7 +377,7 @@ test_unset_job_pick_plugin_without_defaulting_fails_with_einval(void)
fixture_init(&sched);
TEST_ASSERT_NULL(sched.core.job_pick_plugin);
rc = sched_resolve_job_picker(&sched, &resolved);
rc = sched_resolve_job_picker(&sched.core, &resolved);
TEST_ASSERT_EQUAL_INT(-EINVAL, rc);
TEST_ASSERT_NULL(resolved);
@ -411,7 +411,7 @@ test_resolved_legacy_plugin_has_job_elect_fn_set_selecting_legacy_branch(void)
sched.core.job_pick_plugin = "test_legacy_picker_for_wiring";
rc = sched_resolve_job_picker(&sched, &resolved);
rc = sched_resolve_job_picker(&sched.core, &resolved);
TEST_ASSERT_EQUAL_INT(0, rc);
TEST_ASSERT_NOT_NULL(resolved);
@ -437,7 +437,7 @@ test_resolved_builtin_fcfs_has_job_elect_fn_null_selecting_native_branch(void)
sched.core.job_pick_plugin = "builtin_fcfs"; /* explicit, not defaulted */
rc = sched_resolve_job_picker(&sched, &resolved);
rc = sched_resolve_job_picker(&sched.core, &resolved);
TEST_ASSERT_EQUAL_INT(0, rc);
TEST_ASSERT_NOT_NULL(resolved);
@ -467,12 +467,12 @@ test_legacy_and_native_pickers_coexist_and_resolve_independently(void)
TEST_ASSERT_GREATER_OR_EQUAL(0, register_builtin_fcfs(&sched));
sched.core.job_pick_plugin = "xce_sched_fcfs_stub";
rc = sched_resolve_job_picker(&sched, &resolved);
rc = sched_resolve_job_picker(&sched.core, &resolved);
TEST_ASSERT_EQUAL_INT(0, rc);
TEST_ASSERT_NOT_NULL(resolved->job_elect_fn);
sched.core.job_pick_plugin = "builtin_fcfs";
rc = sched_resolve_job_picker(&sched, &resolved);
rc = sched_resolve_job_picker(&sched.core, &resolved);
TEST_ASSERT_EQUAL_INT(0, rc);
TEST_ASSERT_NULL(resolved->job_elect_fn);
}