The spec/design placed the scheduler-op name lookup as xce_sched_op_name in lib/helper, but xce_sched_op_t is a scheduler-only type and src/lib must not depend on scheduler headers — so it is correctly a static sched_op_name() in scheduler.c. Code kept as-is (correct); docs amended to match: - Task 1 intro + 1.1/1.3 + 5.7 + 9.1: scheduler-layer placement + layering note. - design.md Component 8/9, file table, risks: same reconciliation. Only xce_notify_event_name (int-typed, lib-level) lives in lib/helper. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
20 KiB
Design: scheduler-controller-msg-gap
Feature: scheduler-controller-msg-gap Created: 2026-05-11 Status: Design Generated
Architecture Pattern & Boundary Map
Pattern: Extension of existing message handler infrastructure. The scheduler uses a two-phase processing model:
- Enqueue phase (handler.c, runs on IO thread) — lightweight: validate, allocate request struct, enqueue into
pend_updates[0] - Dequeue phase (scheduler.c
sched_update(), runs on main thread) — heavyweight: mutate scheduler state, call plugin hooks, trigger scheduling sessions
All new handlers follow this pattern. No architectural boundary changes needed.
Scope boundary: Only src/scheduler/ source files, src/lib/helper.c, and src/include/helper.h are modified. No controller-side changes. No Cap'n Proto schema changes.
Technology Stack & Alignment
| Aspect | Choice | Rationale |
|---|---|---|
| Language | C99 | Matches existing codebase convention |
| IPC protocol | Cap'n Proto (existing) | No schema changes — only handler gaps filled |
| Build system | Meson (existing) | No build changes needed |
| Threading model | Same as existing — handler enqueues, main thread dequeues | Preserves thread safety guarantees |
| Logging | Existing TRACE/INFO/WARN/ERROR macros |
Consistent with codebase |
| Plugin hooks | Follow existing plugin_*_update() pattern |
Consistency with established plugin API |
Components & Interface Contracts
Component 1: handler.c — Notify handlers (Req 3, 4, 5, 6)
Extend sched_handle_notify() to process the four currently-stubbed NotifyData events.
sched_handle_sched_update_job
Signature: static int sched_handle_sched_update_job(
xce_scheduler_t* sched,
xce_job_t* job)
Behavior:
- If job == NULL, return -EINVAL
- Enqueue XCE_SCHED_OP_UPDATE_JOB with job pointer
- On calloc failure, return -ENOMEM
- Return 0 on success
Requirement traceability: 3.1, 3.2, 3.3
sched_handle_sched_update_queue
Signature: static int sched_handle_sched_update_queue(
xce_scheduler_t* sched,
xce_queue_t* queue)
Behavior:
- If queue == NULL, return -EINVAL
- Enqueue XCE_SCHED_OP_UPDATE_QUEUE with queue pointer
- On calloc failure, return -ENOMEM
- Return 0 on success
Requirement traceability: 4.1, 4.2, 4.3
sched_handle_sched_update_resvals
Signature: static int sched_handle_sched_update_resvals(
xce_scheduler_t* sched,
xce_resval_list_t* resvals)
Behavior:
- If resvals == NULL or count == 0, return 0
- Enqueue XCE_SCHED_OP_UPDATE_RESVALS with resvals pointer
- On calloc failure, return -ENOMEM
- Return 0 on success
Requirement traceability: 5.1, 5.2, 5.3
sched_handle_sched_job_rusage
Signature: static int sched_handle_sched_job_rusage(
xce_scheduler_t* sched,
xce_job_rusages_t* rusages)
Behavior:
- If rusages == NULL, return -EINVAL
- Enqueue XCE_SCHED_OP_JOB_RUSAGE with rusages pointer
- On calloc failure, return -ENOMEM
- Return 0 on success
Requirement traceability: 6.1, 6.2, 6.3
sched_handle_notify switch changes
The existing switch in sched_handle_notify() gains four new cases:
| NotifyData event | Handler function | Previous behavior |
|---|---|---|
NotifyData_data_schedUpdateJob |
sched_handle_sched_update_job(sched, notify->data.sched_update_job) |
TRACE log only |
NotifyData_data_schedUpdateQueue |
sched_handle_sched_update_queue(sched, notify->data.sched_update_queue) |
Empty break |
NotifyData_data_schedUpdateResvals |
sched_handle_sched_update_resvals(sched, notify->data.sched_update_resvals) |
Empty break |
NotifyData_data_schedJobRusages |
sched_handle_sched_job_rusage(sched, notify->data.sched_job_rusages) |
Empty break |
Component 2: handler.c — Dataset handlers (Req 1)
sched_handle_resources
Signature: static int sched_handle_resources(
xce_scheduler_t* sched,
xce_resource_data_t* resdata)
Behavior:
- If resdata == NULL or resdata->count == 0, return 0
- Enqueue XCE_SCHED_OP_UPDATE_RESOURCES with resdata pointer
- On calloc failure, return -ENOMEM
- Return 0 on success
Requirement traceability: 1.1, 1.3, 1.4
sched_handle_resvals
Signature: static int sched_handle_resvals(
xce_scheduler_t* sched,
xce_resval_data_t* resvaldata)
Behavior:
- If resvaldata == NULL or resvaldata->count == 0, return 0
- Enqueue XCE_SCHED_OP_UPDATE_RESVALS_FULL with resvaldata pointer
- On calloc failure, return -ENOMEM
- Return 0 on success
Requirement traceability: 1.2, 1.3, 1.4
sched_handle_dataset switch changes
| Data table | Handler function | Previous behavior |
|---|---|---|
Data_data_resources |
sched_handle_resources(sched, dataset->data.resources) |
Empty break |
Data_data_resvals |
sched_handle_resvals(sched, dataset->data.resvals) |
Empty break |
Component 3: scheduler.h — New op types and update request fields
Extend xce_sched_op_t and xce_sched_update_request_t:
typedef enum {
XCE_SCHED_OP_NONE = 0,
XCE_SCHED_OP_UPDATE_CONFIG,
XCE_SCHED_OP_UPDATE_JOB,
XCE_SCHED_OP_JSTATE_CHANGE,
XCE_SCHED_OP_UPDATE_WORKER,
XCE_SCHED_OP_UPDATE_QUEUE, // NEW
XCE_SCHED_OP_UPDATE_RESOURCES, // NEW
XCE_SCHED_OP_UPDATE_RESVALS_FULL, // NEW — from dataset bulk resvals
XCE_SCHED_OP_UPDATE_RESVALS, // NEW — from notify incremental resvals
XCE_SCHED_OP_JOB_RUSAGE // NEW
} xce_sched_op_t;
Extend xce_sched_update_request_t union:
struct {
xce_queue_t* queue; // for UPDATE_QUEUE
} queueupdate;
struct {
xce_resource_data_t* resdata; // for UPDATE_RESOURCES
} resourceupdate;
struct {
xce_resval_data_t* resvaldata; // for UPDATE_RESVALS_FULL
} resvalfullupdate;
struct {
xce_resval_list_t* resvals; // for UPDATE_RESVALS (notify)
} resvalupdate;
struct {
xce_job_rusages_t* rusages; // for JOB_RUSAGE
} job_rusage;
Component 4: scheduler.c — Dequeue processing in sched_update()
Extend the switch (req->op) in sched_update() with new cases:
XCE_SCHED_OP_UPDATE_QUEUE
- Find existing queue by name with
sched_get_queue() - If found: update mutable fields (status, priority, accept_interval, sched_delay, resreq)
- If not found: return
-ENOENT— queues are created only via the initialConf4Schedconfig push, never via notify - Call
plugin_queue_update()
Requirement traceability: 4.1, 4.2
XCE_SCHED_OP_UPDATE_RESOURCES
- Iterate
resdata->resources[] - For each: call
makeResource(sched->resmgr, r)to add or update - NULL entries skipped (1.3)
Requirement traceability: 1.1
XCE_SCHED_OP_UPDATE_RESVALS_FULL
- Iterate
resvaldata->vals[] - For each: find resource by name in
sched->resmgr, update value - Unknown resources skipped with WARN log
Requirement traceability: 1.2
XCE_SCHED_OP_UPDATE_RESVALS
- Iterate
resvals->vals[] - For each: find worker by host name via
sched_get_worker() - If found: update worker's
resvalszarray at the resource index - Unknown hosts skipped (5.2)
Requirement traceability: 5.1, 5.2, 5.3
XCE_SCHED_OP_JOB_RUSAGE
- Find job by jobid via
sched_get_job() - If not found: return
-ENOENT(6.2) - Store rusage data on job object
- Call
plugin_job_rusage_update()
Requirement traceability: 6.1, 6.2
Component 5: scheduler.c — Job update-in-place (Req 3)
New function: sched_update_existing_job
Signature: static int sched_update_existing_job(
xce_scheduler_t* sched,
xce_job_t* updated)
Behavior:
- Find existing job by jobid via sched_get_job()
- If not found: return -ENOENT — jobs are created only via
the Data_data_schedjobs dataset push, never via notify
- If found: update mutable fields (state, priority, allocs)
- Call plugin_job_update()
- Return 0 on success, negative on error
Requirement traceability: 3.1, 3.2, 3.3
This is wired into the existing XCE_SCHED_OP_UPDATE_JOB case — no new op type needed since sched_on_jobs() already creates new jobs. The change is to make sched_update_jobs() also handle job updates for existing jobids by replacing rather than just adding.
Component 6: plugin.h / plugin.c — New rusage hook (Req 6)
Add to plugin_hooks_t:
typedef int (*plugin_job_rusage_hook_t)(void*, xce_sched_job_t*, xce_job_rusages_t*);
Add to plugin_hooks_t struct:
plugin_job_rusage_hook_t rusage_hook;
Add global function:
int plugin_job_rusage_update(void* sched, xce_sched_job_t* job, xce_job_rusages_t* rusages);
This follows the exact pattern of plugin_job_state_change() — iterate registered plugins, call their hook if non-NULL.
Component 7: handler.c — Dead code removal (Req 2)
Remove from scheduler_req_handler():
- The entire
Request_body_pushmsgcase block (lines 185–198) - The static functions
sched_handle_jobpush()andsched_handle_workerpush()
These are unreachable because vtask_entry() in src/lib/xcectx.c rewrites all pushmsg into datamsg.
Requirement traceability: 2.1, 2.2
Component 8: Name-lookup functions (Req 8)
Layering correction (bug-765 G-7, 2026-06-18): these two helpers live in DIFFERENT layers.
xce_notify_event_name(int)is lib-level and lives insrc/lib/helper.c/src/include/helper.h. The scheduler-op name lookup operates onxce_sched_op_t, a scheduler-only type (scheduler.h);src/libmust not include scheduler headers (no layering inversion), so it is a staticsched_op_name(xce_sched_op_t)insrc/scheduler/src/scheduler.c, NOTxce_sched_op_namein lib/helper. The original "both in helper.c" text below is superseded for the scheduler-op helper.
New function: xce_notify_event_name (lib — src/lib/helper.c)
Signature: char* xce_notify_event_name(int event)
Behavior: Switch onNotifyData_data_* enum values, return symbolic string.
Fallback: "UNKNOWN(%d)" with numeric substitution.
Matches pattern of xce_dataset_table() — takes int, returns char*.
Scheduler-op name lookup: static sched_op_name (scheduler — src/scheduler/src/scheduler.c)
Signature: static char* sched_op_name(xce_sched_op_t op)
Behavior: Switch on XCE_SCHED_OP_* enum values, return symbolic string.
Fallback: "UNKNOWN(%d)" with numeric substitution.
Kept in the scheduler layer (NOT lib/helper as xce_sched_op_name) because xce_sched_op_t is a scheduler-only type and src/lib must not depend on scheduler headers — see the layering correction above. Required for Requirement 7's error logging in sched_update().
Changes to handler.c logging
| Current | Changed to |
|---|---|
TRACE("%s table = %d", __func__, dataset->table) |
TRACE("%s table = %s", __func__, xce_dataset_table(dataset->table)) |
TRACE("%s...", __func__) (in stub handlers) |
INFO("%s event = %s jobid = %s", __func__, xce_notify_event_name(notify->event), jobidString) |
Requirement traceability: 8.1, 8.2, 8.3, 8.4
Component 9: scheduler.c — Error logging in sched_update() (Req 7)
In the sched_update() main loop, after the switch (req->op) block, when rc < 0:
WARN("%s: op %s failed: %s (%d)", __func__,
sched_op_name(req->op), strerror(-rc), rc);
Requirement traceability: 7.1
Component 10: schedobj.c — Queue update function (Req 4)
New function: sched_update_existing_queue
Signature: int sched_update_existing_queue(
xce_scheduler_t* sched,
xce_queue_t* updated)
Behavior:
- Find existing queue by name via sched_get_queue()
- If not found: return -ENOENT — queues are created only via
the initial Conf4Sched config push, never via notify
- If found: update mutable fields:
sched_queue->status = updated->status
sched_queue->priority = updated->priority
sched_queue->accept_interval = updated->accept_interval
sched_queue->sched_delay = updated->sched_delay
// resreq is a string field — reallocate and replace
- Call plugin_queue_update()
- Return 0 on success, negative on error
Requirement traceability: 4.1, 4.2
Component 11: sched_pend_update_killer updates (scheduler.c)
Extend the sched_pend_update_killer destructor to handle new op types:
case XCE_SCHED_OP_UPDATE_QUEUE:
// queue pointer belongs to scheduler, no free needed
break;
case XCE_SCHED_OP_UPDATE_RESOURCES:
// resource_data rows belong to scheduler, no free needed
break;
case XCE_SCHED_OP_UPDATE_RESVALS_FULL:
// resval_data rows owned by capnp decode, freed by free_Request
break;
case XCE_SCHED_OP_UPDATE_RESVALS:
// resval_list owned by capnp decode, freed by free_Request
break;
case XCE_SCHED_OP_JOB_RUSAGE:
// rusage struct owned by capnp decode, freed by free_Request
break;
Important: Memory ownership must be traced carefully. For XCE_SCHED_OP_UPDATE_JOB and XCE_SCHED_OP_UPDATE_WORKER, the existing pattern nullifies dataset->nrows and dataset->rows after enqueue to transfer ownership. New queue/rusage/resvals handlers should follow the same pattern — the request struct owns the data, and free_Request() in xcectx.c handles cleanup.
Data Flow Diagrams
Notify message processing flow
Controller (sched.c) Scheduler (handler.c) Scheduler (scheduler.c)
───────────────────── ───────────────────── ─────────────────────
xce_notify_req()
│
├── schedStartJob ──────────► sched_job_state_change() ──► enq_pend_update(JSTATE_CHANGE)
│ │
├── schedExitJob ───────────► sched_job_state_change() ──► │
│ │
├── schedFailJob ───────────► sched_job_state_change() ──► │
│ │
├── schedUpdateJob ────────► sched_handle_sched_update_job() ─► enq_pend_update(UPDATE_JOB) [NEW]
│ │
├── schedUpdateQueue ──────► sched_handle_sched_update_queue() ─► enq_pend_update(UPDATE_QUEUE) [NEW]
│ │
├── schedUpdateResvals ────► sched_handle_sched_update_resvals() ─► enq_pend_update(UPDATE_RESVALS) [NEW]
│ │
└── schedJobRusages ───────► sched_handle_sched_job_rusage() ──► enq_pend_update(JOB_RUSAGE) [NEW]
│
▼
sched_update() main loop
├── UPDATE_JOB ──► sched_update_existing_job()
├── UPDATE_QUEUE ──► sched_update_existing_queue()
├── UPDATE_RESVALS ──► update worker resvals
├── UPDATE_RESOURCES ──► update resmgr [NEW]
├── UPDATE_RESVALS_FULL ──► update sharedInst [NEW]
└── JOB_RUSAGE ──► store on job + plugin hook [NEW]
Dataset message processing flow
Controller (dataex.c) Scheduler (handler.c) Scheduler (scheduler.c)
──────────────────── ───────────────────── ─────────────────────
xce_push_data_req()
│
├── Data_data_schedjobs ──► sched_on_jobs() ──────────► enq_pend_update(UPDATE_JOB)
│ │
├── Data_data_schedworkers ► sched_on_workers() ───────► enq_pend_update(UPDATE_WORKER)
│ │
├── Data_data_resources ──► sched_handle_resources() ─► enq_pend_update(UPDATE_RESOURCES) [NEW]
│ │
└── Data_data_resvals ────► sched_handle_resvals() ───► enq_pend_update(UPDATE_RESVALS_FULL) [NEW]
│
▼
sched_update() main loop
File Change Summary
| File | Change Type | Requirements |
|---|---|---|
src/scheduler/src/handler.c |
Modified — add handlers, remove dead code, update logging | 1, 2, 3, 4, 5, 6, 8 |
src/scheduler/include/scheduler.h |
Modified — extend enums and structs | 1, 3, 4, 5, 6 |
src/scheduler/src/scheduler.c |
Modified — add dequeue cases, error logging, update functions | 1, 3, 4, 5, 6, 7 |
src/scheduler/include/plugin.h |
Modified — add rusage hook typedef and global function | 6 |
src/scheduler/src/plugin.c |
Modified — add plugin_job_rusage_update() implementation |
6 |
src/scheduler/src/schedobj.c |
Modified — add sched_update_existing_queue() |
4 |
src/lib/helper.c |
Modified — add xce_notify_event_name() (lib-level, int arg); the scheduler-op name lookup is a static sched_op_name() in scheduler.c instead — see Component 8 layering note |
8 |
src/include/helper.h |
Modified — declare xce_notify_event_name() |
8 |
src/scheduler/testsuite/test.c |
Modified — add unit tests for all new handlers | All |
Build Order
src/include/helper.h— add declarations (no dependencies)src/lib/helper.c— add implementations (depends on step 1)src/scheduler/include/scheduler.h— extend enums and structs (depends on typemap headers)src/scheduler/include/plugin.h— add rusage hook typedef (depends on xce/types.h)src/scheduler/src/plugin.c— addplugin_job_rusage_update()(depends on step 4)src/scheduler/src/schedobj.c— addsched_update_existing_queue()(depends on step 3)src/scheduler/src/scheduler.c— add dequeue processing, error logging, update functions (depends on steps 3–6)src/scheduler/src/handler.c— add handlers, remove dead code, update logging (depends on steps 1–7)src/scheduler/testsuite/test.c— add unit tests (depends on all above)
Risks & Mitigations
| Risk | Severity | Mitigation |
|---|---|---|
| Memory ownership of queued data — capnp-decoded structs freed before dequeue processes them | High | Follow existing pattern: for dataset paths, nullify dataset->nrows/dataset->rows after enqueue to transfer ownership. For notify paths, copy essential data into the request struct rather than holding pointers to capnp memory. |
sched_update_existing_job must handle concurrent modification — handler thread enqueues while main thread processes |
Medium | Already mitigated by existing double-buffer swap in sched_update(). The handler thread only touches pend_updates[0] while the main thread drains pend_updates[1]. |
New UPDATE_QUEUE op requires queue name lookup — potential lock contention |
Low | sched_get_queue() already uses read lock. Queue updates are infrequent. |
sched_op_name (static, scheduler.c) must handle future enum additions |
Low | Follow xce_dataset_table pattern — return "UNKNOWN(%d)" for unrecognized values. |