add kiro-sdd documents

This commit is contained in:
Rongsong Shen 2026-05-13 10:58:51 +00:00
parent e847a32546
commit 033f34c07f
10 changed files with 1758 additions and 0 deletions

View file

@ -0,0 +1,469 @@
# 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:
1. **Enqueue phase** (handler.c, runs on IO thread) — lightweight: validate, allocate request struct, enqueue into `pend_updates[0]`
2. **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 initial `Conf4Sched` config 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 `resvals` zarray 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_pushmsg` case block (lines 185–198)
- The static functions `sched_handle_jobpush()` and `sched_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: helper.c / helper.h — Name-lookup functions (Req 8)
#### New function: `xce_notify_event_name`
```
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*`.
#### New function: `xce_sched_op_name`
```
Signature: char* xce_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.
```
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__,
xce_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()`, `xce_sched_op_name()` | 8 |
| `src/include/helper.h` | Modified — add function declarations | 8 |
| `src/scheduler/testsuite/test.c` | Modified — add unit tests for all new handlers | All |
## Build Order
1. `src/include/helper.h` — add declarations (no dependencies)
2. `src/lib/helper.c` — add implementations (depends on step 1)
3. `src/scheduler/include/scheduler.h` — extend enums and structs (depends on typemap headers)
4. `src/scheduler/include/plugin.h` — add rusage hook typedef (depends on xce/types.h)
5. `src/scheduler/src/plugin.c` — add `plugin_job_rusage_update()` (depends on step 4)
6. `src/scheduler/src/schedobj.c` — add `sched_update_existing_queue()` (depends on step 3)
7. `src/scheduler/src/scheduler.c` — add dequeue processing, error logging, update functions (depends on steps 3–6)
8. `src/scheduler/src/handler.c` — add handlers, remove dead code, update logging (depends on steps 1–7)
9. `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. |
| `xce_sched_op_name` must handle future enum additions | Low | Follow `xce_dataset_table` pattern — return `"UNKNOWN(%d)"` for unrecognized values. |