From e819a5060029b3e167ef2e89b7d7eb0a78b00e3c Mon Sep 17 00:00:00 2001 From: Rongsong Shen Date: Wed, 5 Aug 2026 14:11:43 +0000 Subject: [PATCH] feat(scheduler-core): implement scheduler-side admin request handler (task 9.4) Implements the scheduler's first request-reply opcode (OPCode_schedAdmin). sched_handle_admin_request() (handler.c) decodes SchedAdminRequest and dispatches by action: recordStatus answers synchronously (pure read); recordStart/recordStop are deferred onto the existing pend_updates queue (XCE_SCHED_OP_ADMIN_RECORD) since sched_record_start()'s snapshot call requires the session_active==0 precondition only guaranteed on the driver thread, not the async request-handling thread. driver.c's xce_sched_driver_translate_pending_events() performs the deferred action and sends the reply via a new xcecli_reply_schedadmin() (client.c/client.h), correlated by seq copied before the original request is freed. Adversarial review found a genuine cross-thread race: sched_record_status() read driver-owned recording state (including a pointer the driver thread frees) from the vtask thread with no lock -- the codebase's first such race, missed by ASan (doesn't detect races). Fixed with a dedicated record_lock scoped strictly to in-memory field access, never spanning file I/O; sched_record_status() now strdup()s under the lock instead of returning a borrowed live pointer. Verified with a real two-thread lock-contention test (independently confirmed non-vacuous by removing the lock and observing the test fail) plus a deterministic use-after-free reproduction. 26/26 scheduler suite, clang-format clean. --- .kiro/specs/scheduler-core/tasks.md | 3 +- src/include/client.h | 20 + src/lib/client.c | 34 + src/scheduler/include/recording.h | 64 ++ src/scheduler/include/scheduler.h | 88 +- src/scheduler/src/driver.c | 173 ++++ src/scheduler/src/handler.c | 182 +++- src/scheduler/src/recording.c | 111 ++- src/scheduler/src/recording_events.c | 16 +- src/scheduler/src/scheduler.c | 85 +- src/scheduler/testsuite/meson.build | 27 + src/scheduler/testsuite/test_admin_request.c | 903 ++++++++++++++++++ .../testsuite/test_driver_decision_cb.c | 18 + .../testsuite/test_driver_session_ordering.c | 18 + .../testsuite/test_driver_translate.c | 18 + src/scheduler/testsuite/test_handlers.c | 40 +- .../testsuite/test_job_array_expand.c | 56 +- .../testsuite/test_job_array_wire_e2e.c | 11 + .../testsuite/test_record_lifecycle.c | 9 + 19 files changed, 1761 insertions(+), 115 deletions(-) create mode 100644 src/scheduler/testsuite/test_admin_request.c diff --git a/.kiro/specs/scheduler-core/tasks.md b/.kiro/specs/scheduler-core/tasks.md index 3c1d8cb2..4b3e88bd 100644 --- a/.kiro/specs/scheduler-core/tasks.md +++ b/.kiro/specs/scheduler-core/tasks.md @@ -242,11 +242,12 @@ - _Depends: 9.2_ - _Verified: 2026-08-05 — the write-failure error-handling half was already built in task 9.2 (`sched_record_abort_on_error()`, matching AC 10's clause verbatim: log, fflush+fclose, free+NULL path, recording=false); this task added the missing end-to-end test (`test_write_failure_disables_recording_and_driver_keeps_running`) using a genuine closed-fd write failure (fflush + setvbuf(_IONBF) + close(fd), confirmed by review running the binary directly and observing a real `rc=-5` abort log line, then a real subsequent session completing) rather than a mocked return code. Periodic checkpointing (the new work) hooks into `sched_record_write_events()`: once per call, `record_snapshot_interval > 0 && record_event_count % record_snapshot_interval == 0` triggers a fresh `SnapshotEntry` via the same `write_snapshot_entry()` helper `sched_record_start()` already uses (made non-static, zero duplication) — safe from task 8.2's `session_active` precondition because this call site is only ever reached from `xce_sched_driver_translate_pending_events()`, confirmed the sole production caller and confirmed to always run strictly before `xce_core_begin_session()`. `record_snapshot_interval == 0` genuinely produces zero checkpoints. The "once per call, not once per crossed boundary" semantics matches requirements.md's own embedded pseudocode verbatim, not an implementer shortcut. 13/13 new test binary, 25/25 scheduler suite, clang-format clean — independently re-verified by me via a fresh `meson compile`/`meson test --suite scheduler` run before marking complete. Two follow-ups flagged by review, logged below rather than folded in: `--record-snapshot-interval` has no CLI wiring anywhere (AC 11 cannot be configured non-zero in production as currently scoped — this task's own Observable is satisfied via the struct field directly, which is sufficient for this task, but the operator-facing gap remains); `test_record_lifecycle.c` grew to 1232 lines (826→1232), past this suite's prior precedent, with a natural AC10/AC11 split point identified but not acted on._ -- [ ] 9.4 Implement the scheduler-side admin request handler +- [x] 9.4 Implement the scheduler-side admin request handler - Decode a forwarded `SchedAdminRequest` arriving on the scheduler's controller-facing endpoint, dispatch by action to the record-start/stop/status functions built in 9.2/9.3, and encode+send the `SchedAdminReply` back to the controller on the same endpoint — this is the scheduler-side counterpart to the controller's forwarding dispatch (task 10.1), without which 10.1 has nothing to forward to - Observable: sending each of the three admin actions to the scheduler's endpoint produces the matching reply (recording state, file, event count, started-at populated correctly for each action) - _Requirements: 5.10_ - _Depends: 9.3_ + - _Verified: 2026-08-05 — this is the scheduler's first-ever request-reply opcode (`OPCode_schedAdmin`, no prior precedent). `sched_handle_admin_request()` (new, handler.c) decodes `SchedAdminRequest` and dispatches by action: `recordStatus` is answered synchronously (new `sched_record_status()`, recording.c, pure read); `recordStart`/`recordStop` are deferred — enqueued as a new `XCE_SCHED_OP_ADMIN_RECORD` op onto the existing `pend_updates` queue (same mechanism every controller-sourced update already uses) rather than called directly from the request-handling thread, because `sched_record_start()`'s `xce_sched_core_save_snapshot()` call requires task 8.2's `session_active==0` precondition, only guaranteed on the driver thread — confirmed by review as a genuine (not theoretical) concern: `vtaskmgr.c` spawns real pthreads for request handling, distinct from `main.c`'s single driver-loop thread. `driver.c`'s `xce_sched_driver_translate_pending_events()` intercepts the deferred op, performs the real start/stop, and sends the reply via a new `xcecli_reply_schedadmin()` (src/lib/client.c/client.h) correlated by a plain `seq` copied by value before the original request is freed — confirmed safe since `xcecli_ctx_t` is a single unmultiplexed connection. Every reply-field branch (ok/err for each action) independently verified against requirements.md's semantics table, including the subtlety that `recordStop`'s `file` value must be captured before `sched_record_stop()` frees `record_file_path`. Round 1 review found one real bug beyond either implementer session's own testing: `sched_record_status()`, reachable from the vtask thread, read `recording`/`record_file_path`/`record_event_count`/`record_start_ts` with no lock while the driver thread's `sched_record_stop()` mutates and frees those same fields unsynchronized — a genuine cross-thread use-after-free, the codebase's first such race, missed by the ASan-only test config (ASan doesn't catch races). Fixed with a new dedicated `mylock_t record_lock` (deliberately separate from `core.lock`), scoped strictly to the in-memory field reads/writes only (never across `fopen()`/`fwrite()`/capnp I/O — confirmed no vtask-blocks-on-disk-I/O hazard reintroduced), with `sched_record_status()` now `strdup()`ing `record_file_path` while holding the lock rather than returning a borrowed live pointer. Round 2 review independently confirmed the fix: exhaustive grep found every access site now correctly locked (with `recording_session.c`'s driver-thread-only functions and `handle_deferred_admin_record()`'s post-start/stop reads correctly reasoned as safe without the lock, both same-thread-only by construction); confirmed non-recursive `mylock_t` and no nested acquisition (no new deadlock); independently reproduced the fix's necessity by temporarily stripping the lock and confirming the new lock-contention test fails exactly as predicted, then restoring and reconfirming green. 26/26 scheduler suite (14/14 in the new `sched_admin_request_test`), clang-format clean — independently re-verified by me via a fresh `meson compile`/`meson test --suite scheduler` run before marking complete. New `xcecli_reply_schedadmin()` addition to shared `src/lib/client.c`/`src/include/client.h` scoped purely to this one function, confirmed not touching `src/controller/` (task 10.1's exclusive boundary)._ - [ ] 9.5 Recording lifecycle integration tests - Test the full start/stop/status round trip against a real filesystem (through the request handler from 9.4, not by calling the internal functions directly), including the path-validation rejection cases and a simulated disk-full condition diff --git a/src/include/client.h b/src/include/client.h index 1de3ac56..a7bb3f33 100644 --- a/src/include/client.h +++ b/src/include/client.h @@ -133,4 +133,24 @@ extern int xcecli_jobinfo(xcecli_ctx_t, uint64_t, xce_job_t**); extern int xcecli_reply_fetchfile(xcecli_ctx_t, xce_request_t*, xce_fetchfile_reply_t*); extern int xcecli_fetchfile(xcecli_ctx_t, xce_fetchfile_t*, xce_fetchfile_reply_t**); +/* + * xcecli_reply_schedadmin: send a SchedAdminReply back over `c`'s + * connection in response to a previously-received OPCode_schedAdmin + * request (scheduler-core spec, task 9.4, Req 5 AC 10). Mirrors + * xcecli_reply_fetchfile() above exactly, with one difference: it takes + * the original request's `seq` directly rather than an `xce_request_t*`, + * since `recordStart`/`recordStop` defer this reply to the driver thread + * (see driver.c's xce_sched_driver_translate_pending_events()), by which + * point the original xce_request_t has already been freed by the + * request-handling thread (xcectx.c's vtask_entry(), immediately after the + * registered handler returns) -- seq is the only field of it the reply + * needs (opcode is always OPCode_schedAdmin). `recordStatus` (answered + * synchronously from the handler thread, real request still in scope) + * uses this same function, simply passing `request->seq` directly. + * Returns xce_reply()'s own return value; -EINVAL if `c` or `areply` is + * NULL. + */ +extern int xcecli_reply_schedadmin(xcecli_ctx_t, uint64_t seq, + sched_admin_reply_t* areply); + #endif diff --git a/src/lib/client.c b/src/lib/client.c index c0b82236..01aec83d 100644 --- a/src/lib/client.c +++ b/src/lib/client.c @@ -969,6 +969,40 @@ int xcecli_reply_fetchfile(xcecli_ctx_t c, xce_request_t* request, return xce_reply(&(ctx->ep), &(ctx->ctx.myinfo), &reply); } +/* + * xcecli_reply_schedadmin - see client.h's declaration comment for the full + * contract (scheduler-core spec, task 9.4, Req 5 AC 10). Builds the + * RequestReply envelope directly (no init_reply_by_req(): unlike + * xcecli_reply_fetchfile() above, the caller may not have a live + * xce_request_t any more -- recordStart/recordStop replies are sent from + * the driver thread, after the original request has already been freed). + * `reply.status` is always 0: a successfully-built SchedAdminReply is + * always a valid RPC response, whether its own `.ok` field is true or + * false (business-level success/failure lives inside the SchedAdminReply + * payload itself, per its own `ok`/`error` fields -- same convention + * handle_config_request() (controller/config_repl.c) uses for + * ReplConfigReply's "up to date" case). + */ +int xcecli_reply_schedadmin(xcecli_ctx_t c, uint64_t seq, + sched_admin_reply_t* areply) +{ + api_ctx_t* ctx = (api_ctx_t*)c; + xce_reply_t reply; + + if ((ctx == NULL) || (areply == NULL)) { + return -EINVAL; + } + + memset(&reply, 0, sizeof(reply)); + reply.seq = seq; + reply.opcode = OPCode_schedAdmin; + reply.status = 0; + reply.data_which = RequestReply_data_schedadminreplymsg; + reply.data.schedadminreplymsg = areply; + + return xce_reply(&(ctx->ep), &(ctx->ctx.myinfo), &reply); +} + int xcecli_fetchfile(xcecli_ctx_t c, xce_fetchfile_t* msg, xce_fetchfile_reply_t** reply) { diff --git a/src/scheduler/include/recording.h b/src/scheduler/include/recording.h index 68abe15d..6f71c016 100644 --- a/src/scheduler/include/recording.h +++ b/src/scheduler/include/recording.h @@ -170,6 +170,12 @@ extern int sched_validate_record_filename(const char* record_dir, * * `record_file` may be NULL or empty to auto-generate a timestamped name * (`scheduler-replay-.replay`) beneath `sched->record_dir`. + * + * Concurrency (task 9.4 fix round): step 4's field commit takes + * `sched->record_lock` (scheduler.h's field comments) -- see + * sched_record_status()'s doc comment below for the cross-thread reader + * this protects against. The lock is held only around the five assignments + * themselves, never across steps 1-3's file I/O. */ extern int sched_record_start(xce_scheduler_t* sched, const char* record_file); @@ -192,6 +198,13 @@ extern int sched_record_start(xce_scheduler_t* sched, const char* record_file); * marker failed to write (state is still fully cleared either way, per the * error-handling clause above -- the negative return is diagnostic only, * not a signal that recording is still active). + * + * Concurrency (task 9.4 fix round): the fflush()+fclose()+state-clear + * sequence above takes `sched->record_lock` (scheduler.h's field + * comments) around the state-clear portion only -- see + * sched_record_status()'s doc comment below for the cross-thread reader + * this protects against. In particular, freeing `record_file_path` and + * setting it NULL happens atomically from that reader's point of view. */ extern int sched_record_stop(xce_scheduler_t* sched); @@ -225,6 +238,13 @@ extern int sched_record_stop(xce_scheduler_t* sched); * rationale. A checkpoint write failure disables recording via * sched_record_abort_on_error(), the exact same path as any other * recording write failure -- never special-cased. + * + * Concurrency (task 9.4 fix round): the per-event increment of + * `sched->record_event_count` takes `sched->record_lock` (scheduler.h's + * field comments), scoped to just the increment -- see + * sched_record_status()'s doc comment below for the cross-thread reader + * this protects against. `record_snapshot_interval` is driver-thread-only + * and not protected by this lock (scheduler.h's field comment). */ extern int sched_record_write_events(xce_scheduler_t* sched, xce_core_event_t* events, int n); @@ -283,4 +303,48 @@ extern int sched_record_session_end(xce_scheduler_t* sched, sched_record_session_ctx_t* ctx, uint64_t end_ts); +/* + * sched_record_status - Req 5 AC 10 (scheduler-core spec, task 9.4, + * "Reply field semantics by action" -- the `recordStatus` column). A pure + * read of `sched`'s CURRENT recording-state fields into a + * `sched_admin_reply_t` the caller (handler.c's scheduler_req_handler()) + * sends back over the wire immediately -- no file I/O, no state mutation, + * and (unlike sched_record_start()/sched_record_stop()) no session_active + * dependency, so it is safe to answer directly from the request-handling + * thread rather than deferring through the driver-thread pend_updates + * queue the way recordStart/recordStop must. + * + * Concurrency (task 9.4 fix round): this function is called from a vtask + * worker pthread (handler.c's scheduler_req_handler(), invoked with no + * lock held -- lib/xcectx.c), which runs concurrently with the + * single-threaded driver loop's sched_record_start()/sched_record_stop()/ + * sched_record_write_events()/sched_record_abort_on_error() -- all of + * which mutate the same fields this function reads, with no other + * synchronization. This function therefore takes `sched->record_lock` + * (scheduler.h's field comments) around every read of those fields, + * INCLUDING the copy of `record_file_path` described next. + * + * `reply->file` is a fresh, caller-owned heap copy of + * `sched->record_file_path` (via strdup(), taken while still holding + * `sched->record_lock` -- copying the bytes before the lock is released is + * what makes this safe), NOT a borrowed alias of the live pointer. Unlike + * every other capnp encode call in this codebase, this reply crosses a + * function-return boundary before it is ever encoded, so a borrowed + * pointer into `sched`'s live, mutable state cannot be handed back safely + * -- the driver thread could free() it the instant `record_lock` is + * released, well before the caller's eventual xcecli_reply_schedadmin() + * call. The caller MUST free(reply->file) once it is done (NULL when not + * recording -- nothing to free). + * + * Returns 0 on success, populating every field of `*reply` + * (`reply->ok = true` unconditionally). Returns -EINVAL if `sched` or + * `reply` is NULL; `*reply` is left untouched in that case. Returns + * -ENOMEM if the internal strdup() of `record_file_path` fails; `*reply` + * is left zeroed (via the initial memset()) in that case -- this is the + * one way reading status can now itself fail, a direct consequence of no + * longer handing back a borrowed pointer. + */ +extern int sched_record_status(xce_scheduler_t* sched, + sched_admin_reply_t* reply); + #endif diff --git a/src/scheduler/include/scheduler.h b/src/scheduler/include/scheduler.h index 0488c2a1..d435ec67 100644 --- a/src/scheduler/include/scheduler.h +++ b/src/scheduler/include/scheduler.h @@ -39,7 +39,22 @@ typedef enum { /* Notify-only single-job in-place field update. Distinct from * XCE_SCHED_OP_UPDATE_JOB (the dataset bulk-create path) so a notify * update to an existing job mutates it in place and never creates a job. */ - XCE_SCHED_OP_UPDATE_JOB_FIELDS + XCE_SCHED_OP_UPDATE_JOB_FIELDS, + /* Scheduler admin recording control (scheduler-core spec, task 9.4, Req + * 5 AC 10): a `recordStart`/`recordStop` SchedAdminRequest, deferred + * from the request-handling thread (handler.c's scheduler_req_handler()) + * to the driver thread. sched_record_start()/sched_record_stop() do + * real file I/O and sched_record_start() calls + * xce_sched_core_save_snapshot(), which requires the core's + * session_active precondition (task 8.2) -- a guarantee only held on + * the single-threaded driver loop (main.c), not the async request- + * handling thread. See xce_sched_update_request_t's `adminrecord` + * member below and driver.c's xce_sched_driver_translate_pending_events() + * for the full deferred-dispatch + deferred-reply design. + * `recordStatus` is a pure read of already-current struct fields (no + * file I/O, no session_active dependency) and is answered synchronously + * from the handler thread instead -- it never reaches this queue. */ + XCE_SCHED_OP_ADMIN_RECORD } xce_sched_op_t; typedef struct { @@ -79,6 +94,27 @@ typedef struct { struct { xce_job_rusages_t* rusages; } job_rusage; + struct { + /* SchedAdminAction_recordStart or SchedAdminAction_recordStop + * (controller.capnp.h) -- recordStatus never reaches this + * queue (answered synchronously, see xce_sched_op_t's own + * comment on XCE_SCHED_OP_ADMIN_RECORD above). */ + int action; + /* strdup'd copy of SchedAdminRequest.recordFile for + * recordStart (NULL for auto-generate, and always NULL for + * recordStop, which ignores it) -- the original xce_request_t + * is freed by the handler-thread caller (xcectx.c's + * vtask_entry()) immediately after scheduler_req_handler() + * returns, so this must be an independent, owned copy for the + * driver thread to read later. Freed by the deferred + * dispatcher (driver.c) once the action completes. */ + char* record_file; + /* The original request's seq -- the only field of the + * now-freed xce_request_t the deferred reply needs + * (init_reply_by_req()'s other field, opcode, is always + * OPCode_schedAdmin here, so it is not worth storing). */ + uint64_t seq; + } adminrecord; } u; } xce_sched_update_request_t; @@ -132,27 +168,63 @@ typedef struct { * writes -- see xce_sched_driver_translate_pending_events()'s and * xce_sched_driver_run_normal_mode_round()'s own comments. * - * Invariant: recording == true iff record_file != NULL. */ + * Invariant: recording == true iff record_file != NULL. + * + * Concurrency (scheduler-core spec, task 9.4 fix round): `recording`, + * `record_file_path`, `record_file`, `record_event_count`, and + * `record_start_ts` are written on the single-threaded driver loop + * (main.c) but ALSO read by sched_record_status() (recording.c), which + * runs on whichever vtask worker pthread handled the synchronous + * `recordStatus` SchedAdminRequest (handler.c's + * scheduler_req_handler(), invoked with no lock held -- + * lib/xcectx.c). That makes this a genuine cross-thread reader/writer + * pair with no other synchronization, so every read or write of these + * five fields MUST hold `record_lock` below -- see its own comment. */ bool recording; /* strdup'd copy of the resolved record file path; owned by the record * file lifecycle (strdup on start, free+NULL on stop or start/write - * failure) -- lets a future status query (task 9.4, out of this task's - * scope) answer with the current path even after record_file closes. */ + * failure) -- lets a status query (task 9.4) answer with the current + * path even after record_file closes. Protected by `record_lock` (see + * `recording`'s comment above and `record_lock`'s own comment below) -- + * sched_record_status() must copy this string's bytes (e.g. strdup()) + * while still holding the lock rather than handing back the live + * pointer, since the driver thread may free() it the instant the lock + * is released. */ char* record_file_path; - /* replay file handle; NULL when not recording. */ + /* replay file handle; NULL when not recording. Protected by + * `record_lock` for the same reason as `record_file_path` above. */ FILE* record_file; /* count of ReplayEvent messages written since recording started (used * by task 9.3's periodic-checkpoint trigger; incremented by - * recording.c's event-batch writer, task 9.2). */ + * recording.c's event-batch writer, task 9.2). Protected by + * `record_lock`. */ uint64_t record_event_count; /* N events between periodic snapshot checkpoints (0 = disabled). Read * by recording_events.c's sched_record_write_events() (task 9.3): * every time cumulative `record_event_count` lands exactly on a * multiple of this value, a fresh SnapshotEntry is appended to the - * record file (Req 5 AC 11). */ + * record file (Req 5 AC 11). Driver-thread-only (never read by + * sched_record_status()) -- not protected by `record_lock`. */ uint64_t record_snapshot_interval; - /* logical time (xce_core_now()) recording started; 0 = not recording. */ + /* logical time (xce_core_now()) recording started; 0 = not recording. + * Protected by `record_lock`. */ uint64_t record_start_ts; + /* Dedicated mutex (scheduler-core spec, task 9.4 fix round) guarding + * `recording`/`record_file_path`/`record_file`/`record_event_count`/ + * `record_start_ts` -- see `recording`'s comment above for the + * cross-thread reader/writer hazard this closes. Deliberately a NEW, + * separate lock rather than reusing `core.lock`: `core` + * (xce_sched_core_t) is the conceptually distinct core-vs-driver piece + * of state this struct's own header comment describes, and these + * recording fields are driver-level (like `record_dir`/`apictx`/ + * `cluster_cfg` above), so widening core.lock's scope to also cover + * them would blur that boundary for no benefit. Held only around the + * struct-field reads/writes themselves -- never across the actual file + * I/O (fopen()/fwrite()/capn_write_mem() etc.) -- matching this + * codebase's low-contention-lock-scope convention. Initialized by + * sched_init(), destroyed by sched_fini() (scheduler.c) -- same + * lifecycle pairing as `core.lock` (core.c). */ + mylock_t record_lock; } xce_scheduler_t; extern int sched_init(xce_scheduler_t*, xce_client_config_t*, diff --git a/src/scheduler/src/driver.c b/src/scheduler/src/driver.c index 46a850b4..8cd07e9f 100644 --- a/src/scheduler/src/driver.c +++ b/src/scheduler/src/driver.c @@ -1,4 +1,6 @@ #include +#include +#include #include #include @@ -281,6 +283,163 @@ static int translate_one_pending_update(xce_sched_core_t* core, } } +/* + * validation_error_text - classifies a sched_record_start() failure rc into + * the AC 10 error-codes-table bucket it belongs to (scheduler-core spec, + * task 9.4). sched_record_start() calls sched_validate_record_filename() + * (recording.c, task 9.1) FIRST, before ever touching the filesystem for + * real -- that function's own doc comment documents its complete return + * set as exactly these six codes, so a match here means the failure was a + * REJECTED record file NAME, never an I/O failure. Returns NULL for any + * other rc (the caller then falls back to the disk/write-failure message). + * `-EALREADY` (already recording) is handled by the caller before this is + * ever consulted -- not one of this function's cases. + */ +static char* validation_error_text(int rc) +{ + switch (rc) { + case -EINVAL: + case -ENAMETOOLONG: + case -ENOENT: + case -ENOTDIR: + case -EACCES: + case -EEXIST: + return "invalid record file name"; + default: + return NULL; + } +} + +/* + * handle_deferred_admin_record - performs the ACTUAL sched_record_start()/ + * sched_record_stop() call the request-handling thread (handler.c's + * sched_handle_admin_request()) deferred onto this queue, then sends the + * SchedAdminReply the original caller is still waiting on (scheduler-core + * spec, task 9.4, Req 5 AC 10's "Reply field semantics by action" table). + * + * `record_ctx` is `xce_sched_driver_run_normal_mode_round()`'s own `sched` + * parameter, threaded through unchanged (the SAME opaque void* task 9.2's + * per-event-batch recording-write hook already uses -- see this file's + * translate-loop comment). A NULL `record_ctx` (a caller that does not + * want recording, e.g. some unit tests) means there is no `xcecli_ctx_t` + * to reply on either -- this is logged loudly and the (still fully + * owned/allocated) request is simply dropped: the enqueuing caller was + * always guaranteed a NULL-record_ctx round would never legitimately + * enqueue this op in the first place (only handler.c's real production + * wiring does, and main.c always passes the real `sched`). + * + * Field-population strategy: every reply field except `ok`/`error` is + * populated by reading `sched`'s CURRENT state (recording/record_file_path/ + * record_event_count/record_start_ts) strictly AFTER the + * sched_record_start()/sched_record_stop() call returns. This single + * "always snapshot post-call" rule is deliberately uniform across every + * outcome (success, precondition failure, validation failure, I/O + * failure): sched_record_start()/sched_record_stop() never partially + * mutate state on any failure path (recording.c's own doc comments on + * each), so reading post-call state naturally yields the table's "current" + * values on success and its "(unchanged)" values on failure -- no + * per-outcome branching is needed for those four fields, only for + * `ok`/`error`. + * + * One exception each side must still capture BEFORE the call, since the + * function itself clears it as part of a successful call: + * - recordStop(ok): `file` must report the JUST-ENDED recording's path + * ("last path (informational)") -- but sched_record_stop() frees + + * NULLs sched->record_file_path unconditionally as part of its own + * state-clearing (recording.c), so it is snapshotted into a local + * buffer BEFORE the call. + * + * recordStop's only documented failure in the AC 10 error-codes table is + * "not recording" (rc == -EALREADY). A negative rc from a genuine + * end-of-recording-marker WRITE failure is treated as a successful stop + * for reply purposes: recording.h's own doc comment on sched_record_stop() + * says state "is still fully cleared either way ... the negative return is + * diagnostic only, not a signal that recording is still active" -- the + * admin caller asked for a stop, and a stop genuinely happened. + */ +static void handle_deferred_admin_record(void* record_ctx, + xce_sched_update_request_t* req) +{ + xce_scheduler_t* sched = (xce_scheduler_t*)record_ctx; + sched_admin_reply_t reply; + char iobuf[256]; + int rc; + + memset(&reply, 0, sizeof(reply)); + + if (sched == NULL) { + ERROR("%s admin record request (action=%d) enqueued with no " + "record_ctx -- cannot execute or reply", + __func__, req->u.adminrecord.action); + free(req->u.adminrecord.record_file); + return; + } + + if (req->u.adminrecord.action == SchedAdminAction_recordStart) { + rc = sched_record_start(sched, req->u.adminrecord.record_file); + + if (rc == 0) { + reply.ok = 1; + } else if (rc == -EALREADY) { + reply.ok = 0; + reply.error = "already recording"; + } else { + char* verr = validation_error_text(rc); + + reply.ok = 0; + if (verr != NULL) { + reply.error = verr; + } else { + snprintf(iobuf, sizeof(iobuf), "cannot write to %s: %s", + (req->u.adminrecord.record_file != NULL) + ? req->u.adminrecord.record_file + : "(auto-generated)", + strerror(-rc)); + reply.error = iobuf; + } + } + + reply.recording = sched->recording ? 1 : 0; + reply.file = sched->record_file_path; + reply.eventCount = sched->record_event_count; + reply.startedAt = sched->record_start_ts; + + xcecli_reply_schedadmin(sched->apictx, req->u.adminrecord.seq, &reply); + } else if (req->u.adminrecord.action == SchedAdminAction_recordStop) { + char last_path[PATH_MAX]; + + last_path[0] = '\0'; + if (sched->record_file_path != NULL) { + strncpy(last_path, sched->record_file_path, sizeof(last_path) - 1); + last_path[sizeof(last_path) - 1] = '\0'; + } + + rc = sched_record_stop(sched); + + if (rc == -EALREADY) { + reply.ok = 0; + reply.error = "not recording"; + reply.recording = 0; + reply.file = NULL; + reply.eventCount = 0; + reply.startedAt = 0; + } else { + reply.ok = 1; + reply.recording = sched->recording ? 1 : 0; + reply.file = (last_path[0] != '\0') ? last_path : NULL; + reply.eventCount = sched->record_event_count; + reply.startedAt = sched->record_start_ts; + } + + xcecli_reply_schedadmin(sched->apictx, req->u.adminrecord.seq, &reply); + } else { + ERROR("%s unrecognized deferred admin action %d", __func__, + req->u.adminrecord.action); + } + + free(req->u.adminrecord.record_file); +} + int xce_sched_driver_translate_pending_events(xce_sched_core_t* core, void* record_ctx) { @@ -321,6 +480,20 @@ int xce_sched_driver_translate_pending_events(xce_sched_core_t* core, xce_sched_update_request_t* req = ITPTR_TO_UPDATE_REQ(it); xce_core_event_t ev; + /* Req 5 AC 10 (scheduler-core spec, task 9.4): recording admin + * control is a driver-LEVEL concern, not a core scheduling event + * -- "the core has zero knowledge of whether recording is active" + * (requirements.md). Intercepted here, before + * translate_one_pending_update()/xce_core_push_events() ever see + * it, so it is never mistaken for (or batched alongside) a real + * xce_core_event_t. See handle_deferred_admin_record()'s own + * comment for the full deferred-dispatch-and-reply contract. */ + if (req->op == XCE_SCHED_OP_ADMIN_RECORD) { + handle_deferred_admin_record(record_ctx, req); + free(req); + continue; + } + if (!translate_one_pending_update(core, req, now, &ev)) { release_translated_payload(core, req, 0); free(req); diff --git a/src/scheduler/src/handler.c b/src/scheduler/src/handler.c index ae7e9bb2..927cbfd3 100644 --- a/src/scheduler/src/handler.c +++ b/src/scheduler/src/handler.c @@ -10,6 +10,7 @@ #include "helper.h" #include "xce/types.h" +#include "recording.h" #include "scheduler.h" #include "utils.h" @@ -68,14 +69,14 @@ static job_startinfo_t* clone_job_startinfo(const job_startinfo_t* src) return NULL; } -#define CLONE_STR_FIELD(f) \ - do { \ - if (src->f != NULL) { \ - d->f = strdup(src->f); \ - if (d->f == NULL) { \ - goto fail; \ - } \ - } \ +#define CLONE_STR_FIELD(f) \ + do { \ + if (src->f != NULL) { \ + d->f = strdup(src->f); \ + if (d->f == NULL) { \ + goto fail; \ + } \ + } \ } while (0) CLONE_STR_FIELD(exechome); @@ -114,14 +115,14 @@ static job_submit_data_t* clone_job_submit_data(const job_submit_data_t* src) d->mintasks = src->mintasks; d->maxtasks = src->maxtasks; -#define CLONE_STR_FIELD(f) \ - do { \ - if (src->f != NULL) { \ - d->f = strdup(src->f); \ - if (d->f == NULL) { \ - goto fail; \ - } \ - } \ +#define CLONE_STR_FIELD(f) \ + do { \ + if (src->f != NULL) { \ + d->f = strdup(src->f); \ + if (d->f == NULL) { \ + goto fail; \ + } \ + } \ } while (0) CLONE_STR_FIELD(queue); @@ -175,7 +176,7 @@ static xce_job_t* make_array_element(const xce_job_submit_t* job_array, return j; } -static int sched_on_job_array(xce_scheduler_t* sched, +static int sched_on_job_array(xce_scheduler_t* sched, xce_job_submit_t* job_array) { xce_job_t** jobs; @@ -236,8 +237,8 @@ static int sched_on_job_array(xce_scheduler_t* sched, fmt_jobid(jobidString, sizeof(jobidString) - 1, base_jobid); INFO("job array push received and expanded: base_jobid=%s " - "array_size=%llu new_tasks=%d", - jobidString, (unsigned long long)array_size, n_new); + "array_size=%llu new_tasks=%d", + jobidString, (unsigned long long)array_size, n_new); rc = sched_on_jobs(sched, n_new, jobs); if (rc < 0) { @@ -286,9 +287,9 @@ static int sched_handle_resources(xce_scheduler_t* sched, int count, return -ENOMEM; } - req->op = XCE_SCHED_OP_UPDATE_RESOURCES; - req->u.resourceupdate.nresources = count; - req->u.resourceupdate.resources = resources; + req->op = XCE_SCHED_OP_UPDATE_RESOURCES; + req->u.resourceupdate.nresources = count; + req->u.resourceupdate.resources = resources; enq_pend_update(sched, req); @@ -309,9 +310,9 @@ static int sched_handle_resvals(xce_scheduler_t* sched, int count, return -ENOMEM; } - req->op = XCE_SCHED_OP_UPDATE_RESVALS_FULL; - req->u.resvalfullupdate.nvals = count; - req->u.resvalfullupdate.vals = vals; + req->op = XCE_SCHED_OP_UPDATE_RESVALS_FULL; + req->u.resvalfullupdate.nvals = count; + req->u.resvalfullupdate.vals = vals; enq_pend_update(sched, req); @@ -400,8 +401,7 @@ static int sched_job_state_change(xce_scheduler_t* sched, uint64_t jobid, return 0; } -static int sched_handle_sched_update_job(xce_scheduler_t* sched, - xce_job_t* job) +static int sched_handle_sched_update_job(xce_scheduler_t* sched, xce_job_t* job) { xce_sched_update_request_t* req; char jobidString[128]; @@ -422,8 +422,8 @@ static int sched_handle_sched_update_job(xce_scheduler_t* sched, * (XCE_SCHED_OP_UPDATE_JOB_FIELDS), NOT the dataset bulk-create path * (XCE_SCHED_OP_UPDATE_JOB). An unknown jobid must yield -ENOENT, never a * new job. The dataset bulk-create path (sched_on_jobs) is unchanged. */ - req->op = XCE_SCHED_OP_UPDATE_JOB_FIELDS; - req->u.jobfieldsupdate.job = job; + req->op = XCE_SCHED_OP_UPDATE_JOB_FIELDS; + req->u.jobfieldsupdate.job = job; enq_pend_update(sched, req); @@ -431,7 +431,7 @@ static int sched_handle_sched_update_job(xce_scheduler_t* sched, } static int sched_handle_sched_update_queue(xce_scheduler_t* sched, - xce_queue_t* queue) + xce_queue_t* queue) { xce_sched_update_request_t* req; @@ -444,8 +444,8 @@ static int sched_handle_sched_update_queue(xce_scheduler_t* sched, return -ENOMEM; } - req->op = XCE_SCHED_OP_UPDATE_QUEUE; - req->u.queueupdate.queue = queue; + req->op = XCE_SCHED_OP_UPDATE_QUEUE; + req->u.queueupdate.queue = queue; enq_pend_update(sched, req); @@ -466,8 +466,8 @@ static int sched_handle_sched_update_resvals(xce_scheduler_t* sched, return -ENOMEM; } - req->op = XCE_SCHED_OP_UPDATE_RESVALS; - req->u.resvalupdate.resvals = resvals; + req->op = XCE_SCHED_OP_UPDATE_RESVALS; + req->u.resvalupdate.resvals = resvals; enq_pend_update(sched, req); @@ -475,7 +475,7 @@ static int sched_handle_sched_update_resvals(xce_scheduler_t* sched, } static int sched_handle_sched_job_rusage(xce_scheduler_t* sched, - xce_job_rusages_t* rusages) + xce_job_rusages_t* rusages) { xce_sched_update_request_t* req; @@ -488,8 +488,8 @@ static int sched_handle_sched_job_rusage(xce_scheduler_t* sched, return -ENOMEM; } - req->op = XCE_SCHED_OP_JOB_RUSAGE; - req->u.job_rusage.rusages = rusages; + req->op = XCE_SCHED_OP_JOB_RUSAGE; + req->u.job_rusage.rusages = rusages; enq_pend_update(sched, req); @@ -547,6 +547,111 @@ int sched_handle_notify(xce_scheduler_t* sched, xce_notify_data_t* notify) return rc; } +/* + * sched_handle_admin_request - decodes and dispatches an OPCode_schedAdmin + * request (scheduler-core spec, task 9.4, Req 5 AC 10). This is the + * scheduler-side counterpart to task 10.1's controller-side forwarding + * dispatch: the controller relays a SchedAdminRequest it received from + * xceadmin, verbatim, to this endpoint; this function is where it is + * finally interpreted. + * + * `recordStatus` is a pure read (recording.h's sched_record_status()) with + * no file I/O and no session_active dependency -- it is safe to answer + * directly, synchronously, on this (request-handling) thread, and does so + * via xcecli_reply_schedadmin() while `request` is still fully valid. + * + * `recordStart`/`recordStop` do real file I/O and (recordStart) call + * xce_sched_core_save_snapshot(), which requires the core's + * session_active == 0 precondition (task 8.2) -- a guarantee only held on + * the single-threaded driver loop (main.c), never on this async + * request-handling thread (tasks.md's Implementation Notes, "Cross-task + * interaction" entry preceding this task). So neither the action nor its + * reply can happen here: this function enqueues an + * XCE_SCHED_OP_ADMIN_RECORD request onto the SAME pend_updates queue every + * other controller-sourced update already flows through (enq_pend_update() + * above), captured with just enough context (action, a strdup'd copy of + * recordFile, and the request's seq) to survive past this function + * returning and `request` being freed by the caller (xcectx.c's + * vtask_entry(), immediately after this handler returns). Actually + * performing the deferred action and sending its reply is driver.c's + * xce_sched_driver_translate_pending_events()'s job -- see that file's own + * comment on its XCE_SCHED_OP_ADMIN_RECORD interception. + * + * Returns 0 on successful dispatch (synchronous reply sent, or the + * deferred request successfully enqueued); a negative errno otherwise + * (-EINVAL for a NULL argument/missing body/unrecognized action, -ENOMEM + * on an allocation failure). Mirrors every other case in this function: + * the return value is for internal logging only (vtask_entry(), xcectx.c, + * discards it) -- it is NOT itself what causes a reply to be sent. + */ +static int sched_handle_admin_request(xce_scheduler_t* sched, + xce_request_t* request) +{ + sched_admin_req_t* areq; + + if ((sched == NULL) || (request == NULL) + || (request->body.schedadminmsg == NULL)) { + return -EINVAL; + } + + areq = request->body.schedadminmsg; + + switch (areq->action) { + case SchedAdminAction_recordStatus: { + sched_admin_reply_t reply; + int rc = sched_record_status(sched, &reply); + int send_rc; + + if (rc < 0) { + return rc; + } + + /* recording.h's sched_record_status() doc comment (task 9.4 fix + * round): reply.file is now a fresh strdup()'d copy (or NULL), not + * a borrowed alias of sched->record_file_path -- caller-owned, so + * it must be freed here once the encode call above is done with + * it. */ + send_rc = xcecli_reply_schedadmin(sched->apictx, request->seq, &reply); + free(reply.file); + + return send_rc; + } + + case SchedAdminAction_recordStart: + case SchedAdminAction_recordStop: { + xce_sched_update_request_t* req; + + req = (xce_sched_update_request_t*)calloc(1, sizeof(*req)); + if (req == NULL) { + return -ENOMEM; + } + + req->op = XCE_SCHED_OP_ADMIN_RECORD; + req->u.adminrecord.action = areq->action; + req->u.adminrecord.seq = request->seq; + + if ((areq->action == SchedAdminAction_recordStart) + && (areq->recordFile != NULL) && (areq->recordFile[0] != '\0')) { + req->u.adminrecord.record_file = strdup(areq->recordFile); + if (req->u.adminrecord.record_file == NULL) { + free(req); + return -ENOMEM; + } + } else { + req->u.adminrecord.record_file = NULL; + } + + enq_pend_update(sched, req); + + return 0; + } + + default: + WARN("%s unrecognized SchedAdminAction %d", __func__, areq->action); + return -EINVAL; + } +} + int scheduler_req_handler(xce_request_t* request, void* args) { int rc = 0; @@ -579,6 +684,9 @@ int scheduler_req_handler(xce_request_t* request, void* args) case OPCode_notify: rc = sched_handle_notify(sched, request->body.notifymsg); break; + case OPCode_schedAdmin: + rc = sched_handle_admin_request(sched, request); + break; default: rc = -EINVAL; break; diff --git a/src/scheduler/src/recording.c b/src/scheduler/src/recording.c index d5ac8c67..25d6a409 100644 --- a/src/scheduler/src/recording.c +++ b/src/scheduler/src/recording.c @@ -484,11 +484,19 @@ int sched_record_start(xce_scheduler_t* sched, const char* record_file) return rc; } + /* record_lock (scheduler.h field comments, task 9.4 fix round): guards + * this commit against sched_record_status(), which can run + * concurrently on a vtask worker thread. Scope is deliberately just + * these five assignments -- every fopen()/write_replay_header()/ + * write_snapshot_entry() I/O call above has already completed by this + * point, so the lock never spans file I/O. */ + mylock_lock(&(sched->record_lock)); sched->record_file = f; sched->record_file_path = strdup(resolved_path); sched->recording = true; sched->record_event_count = 0; sched->record_start_ts = start_ts; + mylock_unlock(&(sched->record_lock)); return 0; } @@ -529,13 +537,28 @@ int sched_record_stop(xce_scheduler_t* sched) /* Clear recording state unconditionally -- Req 5 AC 10's "Error * handling" clause: a failed write disables recording rather than - * leaving it half-active. */ + * leaving it half-active. fflush()/fclose() run BEFORE record_lock is + * taken (task 9.4 fix round) -- they are file I/O on `record_file`, + * never contended by sched_record_status() (which never touches + * `record_file` itself, only reads the OTHER four fields under the + * lock), so there is nothing to protect here and no reason to hold the + * lock across them. */ fflush(sched->record_file); fclose(sched->record_file); + + /* record_lock (scheduler.h field comments): guards the four-field + * commit below against sched_record_status(), which can run + * concurrently on a vtask worker thread. In particular, `free()`ing + * record_file_path and then setting it NULL must happen atomically + * from that reader's point of view -- otherwise a concurrent + * sched_record_status() could read the freed pointer between the two + * statements (the exact use-after-free this fix closes). */ + mylock_lock(&(sched->record_lock)); sched->record_file = NULL; sched->recording = false; free(sched->record_file_path); sched->record_file_path = NULL; + mylock_unlock(&(sched->record_lock)); return rc; } @@ -550,20 +573,98 @@ int sched_record_stop(xce_scheduler_t* sched) * already broken -- disk full, permission revoked mid-write, etc. -- so * attempting one more write is not productive). */ +/* + * sched_record_status - see recording.h's doc comment for the full + * contract (scheduler-core spec, task 9.4, Req 5 AC 10). A pure read of + * `sched`'s current recording-state fields -- no file I/O, no mutation. + * `reply->file` deliberately aliases `sched->record_file_path` rather than + * copying it (recording.h documents the borrowed-pointer lifetime the + * caller must respect). + */ +int sched_record_status(xce_scheduler_t* sched, sched_admin_reply_t* reply) +{ + char* file_copy; + int had_path; + + if ((sched == NULL) || (reply == NULL)) { + return -EINVAL; + } + + memset(reply, 0, sizeof(*reply)); + + /* record_lock (scheduler.h field comments, task 9.4 fix round): this + * function runs on a vtask worker pthread (handler.c's + * scheduler_req_handler(), invoked with no lock held -- lib/xcectx.c), + * concurrently with the driver thread's sched_record_start()/ + * sched_record_stop()/sched_record_abort_on_error(), which mutate + * these same five fields with no other synchronization. The lock must + * stay held through the strdup() below, not just the four field reads + * -- copying record_file_path's BYTES while the lock is held is what + * makes the copy safe; handing back sched->record_file_path itself + * (the pre-fix behavior) would let the driver thread free() it out + * from under the caller the instant this function returns. */ + mylock_lock(&(sched->record_lock)); + + reply->recording = sched->recording ? 1 : 0; + reply->eventCount = sched->record_event_count; + reply->startedAt = sched->record_start_ts; + + had_path = (sched->record_file_path != NULL); + file_copy = had_path ? strdup(sched->record_file_path) : NULL; + + mylock_unlock(&(sched->record_lock)); + + /* strdup() failure (ENOMEM) is the one way this now-copying function + * can fail where the old borrowed-pointer version could not -- surface + * it rather than silently returning a status reply with a missing + * `file` field for a recording that is genuinely active. */ + if (had_path && (file_copy == NULL)) { + return -ENOMEM; + } + + reply->ok = 1; + reply->error = NULL; + /* Ownership: `reply->file` is now a caller-owned heap allocation (NULL + * when not recording) -- unlike every other capnp-encode call in this + * codebase, this reply crosses a function-return boundary (back to + * handler.c's sched_handle_admin_request()) before it is ever encoded, + * so a borrowed pointer into `sched`'s live state cannot be made safe + * here the way handle_deferred_admin_record()'s recordStop path (a + * same-stack-frame local buffer) manages it. The caller MUST + * free(reply->file). */ + reply->file = file_copy; + + return 0; +} + void sched_record_abort_on_error(xce_scheduler_t* sched, int rc) { + FILE* f; + ERROR("%s scheduler recording write to %s failed (rc=%d) -- disabling " "recording", __func__, (sched->record_file_path != NULL) ? sched->record_file_path : "?", rc); - if (sched->record_file != NULL) { - fflush(sched->record_file); - fclose(sched->record_file); - sched->record_file = NULL; + /* fflush()/fclose() happen BEFORE record_lock is taken, same rationale + * as sched_record_stop()'s identical ordering above -- `record_file` + * itself is never read by sched_record_status(), so there is nothing + * to protect, and the lock must not span file I/O. */ + f = sched->record_file; + if (f != NULL) { + fflush(f); + fclose(f); } + + /* record_lock (scheduler.h field comments): same atomicity requirement + * as sched_record_stop()'s identical commit -- free()ing + * record_file_path and setting it NULL must be indivisible from + * sched_record_status()'s point of view. */ + mylock_lock(&(sched->record_lock)); + sched->record_file = NULL; free(sched->record_file_path); sched->record_file_path = NULL; sched->recording = false; + mylock_unlock(&(sched->record_lock)); } diff --git a/src/scheduler/src/recording_events.c b/src/scheduler/src/recording_events.c index 22a1e6dd..88c39d54 100644 --- a/src/scheduler/src/recording_events.c +++ b/src/scheduler/src/recording_events.c @@ -39,9 +39,16 @@ extern int write_snapshot_entry(FILE* f, xce_sched_core_t* core, uint64_t ts); * unconditionally (cheap stack space) so its address stays valid across * the single encode+write call after the switch, regardless of which case * populated it. + * + * Takes `sched` (rather than a bare `uint64_t*` counter, task 9.4 fix + * round) so the post-write increment of `sched->record_event_count` can be + * taken under `sched->record_lock` -- that field is read concurrently by + * sched_record_status() on a vtask worker thread (scheduler.h's field + * comment). The lock is held ONLY around the increment itself, after the + * encode+write I/O above has already completed. */ static int write_one_replay_event(FILE* f, xce_core_event_t* ev, - uint64_t* event_count) + xce_scheduler_t* sched) { struct capn c; struct capn_segment* cs; @@ -151,7 +158,9 @@ static int write_one_replay_event(FILE* f, xce_core_event_t* ev, capn_free(&c); if (rc == 0) { - (*event_count)++; + mylock_lock(&(sched->record_lock)); + sched->record_event_count++; + mylock_unlock(&(sched->record_lock)); } return rc; @@ -202,8 +211,7 @@ int sched_record_write_events(xce_scheduler_t* sched, xce_core_event_t* events, } for (i = 0; i < n; i++) { - int rc = write_one_replay_event(sched->record_file, &events[i], - &(sched->record_event_count)); + int rc = write_one_replay_event(sched->record_file, &events[i], sched); if (rc < 0) { sched_record_abort_on_error(sched, rc); return rc; diff --git a/src/scheduler/src/scheduler.c b/src/scheduler/src/scheduler.c index 464e7668..e2662b09 100644 --- a/src/scheduler/src/scheduler.c +++ b/src/scheduler/src/scheduler.c @@ -477,7 +477,7 @@ static int sched_register_plugins(xce_scheduler_t* sched) int sched_init(xce_scheduler_t* sched, xce_client_config_t* config, const xce_cluster_config_t* cluster_cfg, - xce_request_handler_t handler) + xce_request_handler_t handler) { int rc = 0; /* Req 1 AC 6 / Req 2 AC 5 (scheduler-core spec, task 2.4): resolved @@ -493,6 +493,17 @@ int sched_init(xce_scheduler_t* sched, xce_client_config_t* config, memset(sched, 0x0, sizeof(*sched)); + /* scheduler-core spec, task 9.4 fix round: record_lock guards the + * recording-state fields below record_dir (scheduler.h's field + * comments) against the vtask-thread sched_record_status() reader / + * driver-thread writer race. Initialized here (paired with + * mylock_fini() in sched_fini()) the same way core.lock is initialized + * by xce_sched_core_init() just below -- must happen before any thread + * that could call sched_record_status() (the vtask workers, created far + * later by lib/xcectx.c once this scheduler registers its handler) can + * possibly run. */ + mylock_init(&(sched->record_lock)); + if (cluster_cfg != NULL) { sched->cluster_cfg = *cluster_cfg; } @@ -651,6 +662,10 @@ void sched_fini(xce_scheduler_t* sched) * same order as before this task, just relocated. */ xce_sched_core_fini(&(sched->core)); + /* scheduler-core spec, task 9.4 fix round: pairs with sched_init()'s + * mylock_init(&(sched->record_lock)) above. */ + mylock_fini(&(sched->record_lock)); + return; } @@ -692,8 +707,8 @@ static int sched_init_sharedInst(xce_scheduler_t* sched) if (key_find(sched->core.resmgr, (void*)nv->name, (void**)&r) < 0 || r == NULL) { - WARN("%s unknown resource '%s' in resvals, skipping", - __func__, nv->name); + WARN("%s unknown resource '%s' in resvals, skipping", __func__, + nv->name); continue; } @@ -793,18 +808,30 @@ static char* sched_op_name(xce_sched_op_t op) static char buf[32]; switch (op) { - case XCE_SCHED_OP_NONE: return "NONE"; - case XCE_SCHED_OP_UPDATE_CONFIG: return "UPDATE_CONFIG"; - case XCE_SCHED_OP_UPDATE_JOB: return "UPDATE_JOB"; - case XCE_SCHED_OP_JSTATE_CHANGE: return "JSTATE_CHANGE"; - case XCE_SCHED_OP_UPDATE_WORKER: return "UPDATE_WORKER"; - case XCE_SCHED_OP_UPDATE_QUEUE: return "UPDATE_QUEUE"; - case XCE_SCHED_OP_UPDATE_RESOURCES: return "UPDATE_RESOURCES"; - case XCE_SCHED_OP_UPDATE_RESVALS_FULL: return "UPDATE_RESVALS_FULL"; - case XCE_SCHED_OP_UPDATE_RESVALS: return "UPDATE_RESVALS"; - case XCE_SCHED_OP_JOB_RUSAGE: return "JOB_RUSAGE"; - case XCE_SCHED_OP_UPDATE_JOB_FIELDS: return "UPDATE_JOB_FIELDS"; - default: break; + case XCE_SCHED_OP_NONE: + return "NONE"; + case XCE_SCHED_OP_UPDATE_CONFIG: + return "UPDATE_CONFIG"; + case XCE_SCHED_OP_UPDATE_JOB: + return "UPDATE_JOB"; + case XCE_SCHED_OP_JSTATE_CHANGE: + return "JSTATE_CHANGE"; + case XCE_SCHED_OP_UPDATE_WORKER: + return "UPDATE_WORKER"; + case XCE_SCHED_OP_UPDATE_QUEUE: + return "UPDATE_QUEUE"; + case XCE_SCHED_OP_UPDATE_RESOURCES: + return "UPDATE_RESOURCES"; + case XCE_SCHED_OP_UPDATE_RESVALS_FULL: + return "UPDATE_RESVALS_FULL"; + case XCE_SCHED_OP_UPDATE_RESVALS: + return "UPDATE_RESVALS"; + case XCE_SCHED_OP_JOB_RUSAGE: + return "JOB_RUSAGE"; + case XCE_SCHED_OP_UPDATE_JOB_FIELDS: + return "UPDATE_JOB_FIELDS"; + default: + break; } snprintf(buf, sizeof(buf), "UNKNOWN(%d)", (int)op); @@ -858,13 +885,15 @@ int sched_update(xce_scheduler_t* sched) rc = sched_update_existing_queue(sched, req->u.queueupdate.queue); break; case XCE_SCHED_OP_UPDATE_RESOURCES: { - int i; + int i; for (i = 0; i < req->u.resourceupdate.nresources; i++) { xce_resource_t* r = req->u.resourceupdate.resources[i]; - if (r == NULL) continue; + if (r == NULL) + continue; xce_resource_t* existing = makeResource(sched->core.resmgr, r); if (existing == NULL) { - /* EEXIST: update-in-place on existing resource (add-or-update) */ + /* EEXIST: update-in-place on existing resource + * (add-or-update) */ xce_resource_t* found = NULL; if (key_find(sched->core.resmgr, r->name, (void**)&found) == 0 @@ -875,9 +904,11 @@ int sched_update(xce_scheduler_t* sched) found->ordtype = r->ordtype; found->flags = r->flags; found->interval = r->interval; - WARN("%s updated existing resource %s", __func__, r->name); + WARN("%s updated existing resource %s", __func__, + r->name); } else { - WARN("%s makeResource failed for %s and lookup not found", + WARN("%s makeResource failed for %s and lookup not " + "found", __func__, r->name); } } @@ -886,7 +917,7 @@ int sched_update(xce_scheduler_t* sched) break; } case XCE_SCHED_OP_UPDATE_RESVALS_FULL: { - int i; + int i; for (i = 0; i < req->u.resvalfullupdate.nvals; i++) { xce_resval_t* rv = req->u.resvalfullupdate.vals[i]; xce_resource_t* r = NULL; @@ -905,15 +936,17 @@ int sched_update(xce_scheduler_t* sched) break; } case XCE_SCHED_OP_UPDATE_RESVALS: { - int i; + int i; xce_resval_list_t* resvals = req->u.resvalupdate.resvals; for (i = 0; i < resvals->count; i++) { xce_resval_t* rv = resvals->vals[i]; xce_sched_worker_t* w = NULL; xce_resource_t* r = NULL; - if (rv == NULL) continue; + if (rv == NULL) + continue; w = sched_get_worker(&(sched->core), rv->host); - if (w == NULL) continue; + if (w == NULL) + continue; if (key_find(sched->core.resmgr, rv->name, (void**)&r) < 0 || r == NULL) { WARN("%s unknown resource %s in per-worker resvals update", @@ -943,8 +976,8 @@ int sched_update(xce_scheduler_t* sched) sched_pend_update_killer(it); if (rc < 0) { - WARN("%s: op %s failed: %s (%d)", __func__, - sched_op_name(req->op), strerror(-rc), rc); + WARN("%s: op %s failed: %s (%d)", __func__, sched_op_name(req->op), + strerror(-rc), rc); } } diff --git a/src/scheduler/testsuite/meson.build b/src/scheduler/testsuite/meson.build index 9833ec60..617fa2ce 100644 --- a/src/scheduler/testsuite/meson.build +++ b/src/scheduler/testsuite/meson.build @@ -434,6 +434,30 @@ sched_record_lifecycle_test = executable( dependencies: [scheduler_test_deps], install: false) +# Contract test for the scheduler-side OPCode_schedAdmin request handler +# (scheduler-core spec, task 9.4, Req 5 AC 10): decodes a forwarded +# SchedAdminRequest, dispatches by action to sched_record_start()/ +# sched_record_stop()/sched_record_status(), and sends back a +# SchedAdminReply matching requirements.md's field-semantics table. +# recordStatus is proven synchronous (answered directly on the handler +# thread, both idle and while genuinely recording); recordStart/recordStop +# are proven deferred (enqueued, not executed, until a subsequent +# xce_sched_driver_translate_pending_events() drain actually performs the +# real sched_record_start()/sched_record_stop() call and sends the reply). +# Same white-box #include pattern as sched_driver_translate_test +# (core.c/core_clock.c/core_events.c/driver.c) PLUS ../src/recording.c/ +# recording_events.c/snapshot.c/snapshot_wire.c (real record start/stop/ +# status/write, task 9.1/9.2/9.3) PLUS ../src/handler.c (the code under +# test). xcecli_reply_schedadmin() is a SPY test double, same pattern +# sched_driver_decision_cb_test already establishes for +# sched_publish_dispatch()/sched_publish_pend_reason(). +sched_admin_request_test = executable( + 'sched_admin_request_test', + ['test_admin_request.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', @@ -500,3 +524,6 @@ test('scheduler recording path validation test', test('scheduler record lifecycle test', sched_record_lifecycle_test, suite: 'scheduler', env: ['ASAN_OPTIONS=detect_leaks=0']) +test('scheduler admin request test', + sched_admin_request_test, suite: 'scheduler', + env: ['ASAN_OPTIONS=detect_leaks=0']) diff --git a/src/scheduler/testsuite/test_admin_request.c b/src/scheduler/testsuite/test_admin_request.c new file mode 100644 index 00000000..e258b7ff --- /dev/null +++ b/src/scheduler/testsuite/test_admin_request.c @@ -0,0 +1,903 @@ +/* + * test_admin_request.c - contract test for the scheduler-side + * OPCode_schedAdmin request handler (scheduler-core spec, task 9.4, Req 5 + * AC 10). This is the scheduler-side counterpart to task 10.1's + * controller-side forwarding dispatch (out of THIS spec's boundary): the + * controller relays a SchedAdminRequest verbatim to this endpoint, and this + * file proves the scheduler decodes it, dispatches by action to + * sched_record_start()/sched_record_stop()/sched_record_status() (tasks + * 9.2/9.3/9.4), and sends back a SchedAdminReply whose fields match + * requirements.md's "Reply field semantics by action" table. + * + * Observable under test: + * - recordStatus is answered SYNCHRONOUSLY, on the calling (handler) + * thread, directly from handler.c's sched_handle_admin_request() -- + * never enqueued -- both when not recording (all-zero/empty reply) and + * while genuinely recording (live event count/file/started-at). + * - recordStart/recordStop are DEFERRED: enqueued onto the same + * pend_updates dataq every other controller-sourced update flows + * through (XCE_SCHED_OP_ADMIN_RECORD), NOT executed synchronously -- + * `sched->recording` is provably unchanged immediately after + * scheduler_req_handler() returns -- and only actually performed (real + * sched_record_start()/sched_record_stop() call, real reply send) by a + * subsequent xce_sched_driver_translate_pending_events() drain + * (driver.c's handle_deferred_admin_record()). + * - recordStart while already recording / recordStop while not recording + * reply {ok=false, error="already recording"/"not recording"} with the + * table's documented (unchanged) field values. + * - An invalid recordFile (path separator) surfaces all the way through + * the deferred path as {ok=false, error="invalid record file name"}, + * never just an internal log line, and never actually opens a file. + * - An unrecognized SchedAdminAction returns -EINVAL, logs, and neither + * enqueues anything nor sends any reply -- no crash. + * + * White-box #include of ../src/core.c, ../src/core_clock.c, + * ../src/core_events.c, ../src/driver.c (same set test_driver_translate.c + * uses to exercise xce_sched_driver_translate_pending_events() for real) + * PLUS ../src/recording.c/recording_events.c (task 9.1/9.2/9.3's REAL + * sched_record_start()/sched_record_stop()/sched_record_status()/ + * sched_record_write_events() -- this file's whole point is proving the + * deferred path performs REAL file I/O, so these must be genuine, not + * stubbed) PLUS ../src/snapshot.c/snapshot_wire.c (task 8.1's real + * xce_sched_core_save_snapshot(), needed by sched_record_start()) PLUS + * ../src/handler.c (task 9.4's sched_handle_admin_request()/ + * scheduler_req_handler(), the code under test). + * + * recording_session.c is NOT linked: its sched_record_session_begin()/ + * sched_record_session_add_decision()/sched_record_session_end() trio is + * only reachable from xce_sched_driver_run_normal_mode_round(), which this + * file never calls (it drives xce_sched_driver_translate_pending_events() + * directly, same as test_driver_translate.c) -- stubbed below purely to + * satisfy the linker, same rationale as that file's identical stubs for the + * same reason. Likewise core_session.c/plugin.c are not linked (their + * exported functions are stubbed below); no real scheduling session is ever + * run by any test in this file. + * + * xcecli_reply_schedadmin() is a SPY test double (client.c's real + * implementation needs a live capnp connection this file never sets up) -- + * same pattern test_driver_decision_cb.c already establishes for + * sched_publish_dispatch()/sched_publish_pend_reason(). It deep-copies + * every field of the passed sched_admin_reply_t (including string fields) + * into fixed-size static storage rather than retaining the passed pointers: + * both handle_deferred_admin_record() (recordStop's `file`) and + * sched_record_status() (`file`, a BORROWED alias of + * sched->record_file_path) document that their reply pointers are only + * valid for the duration of the synchronous call -- recordStop's in + * particular aliases a stack-local buffer that is gone the instant + * handle_deferred_admin_record() returns, so retaining the raw pointer + * would be a stack-use-after-return read the very next line. + */ +#include +#include +#include +#include +#include +#include +#include +#include + +#include + +#include "controller.capnp.h" +#include "dataq.h" +#include "kvmgr.h" +#include "mylock.h" +#include "scheduler.h" +#include "xstring.h" + +/* Pull in the real functions under test into this TU (white-box #include, + * same convention as test_driver_translate.c/test_record_lifecycle.c). */ +#include "../src/core.c" +#include "../src/core_clock.c" +#include "../src/core_events.c" +#include "../src/driver.c" +#include "../src/recording.c" +#include "../src/recording_events.c" +#include "../src/snapshot.c" +#include "../src/snapshot_wire.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; } + +/* Link-satisfying stubs for scheduler.c's sched_publish_dispatch()/ + * sched_publish_pend_reason() -- driver.c's (real, #included above) + * xce_sched_driver_on_decision() references both; never actually invoked by + * any test in this file (no session is ever run, so no decision is ever + * drained). Same rationale as test_driver_translate.c's identical stubs. */ +int sched_publish_dispatch(xce_scheduler_t* sched, xce_job_dispatch_t* d) +{ + (void)sched; + (void)d; + return 0; +} + +int sched_publish_pend_reason(xce_scheduler_t* sched, xce_preason_set_t* r) +{ + (void)sched; + (void)r; + return 0; +} + +/* Link-satisfying stubs: xce_sched_driver_run_normal_mode_round() (driver.c, + * real, #included above, but never CALLED by any test here -- this file + * only drives xce_sched_driver_translate_pending_events() directly, same as + * test_driver_translate.c) references xce_core_begin_session()/ + * xce_core_signal_go()/xce_core_end_session() (core_session.c) and + * plugin_begin_session()/plugin_end_session() (plugin.c) / + * sched_account_jobs() (scheduler.c). Never actually invoked at runtime. */ +int xce_core_begin_session(xce_sched_core_t* core) +{ + (void)core; + return 0; +} + +int xce_core_signal_go(xce_sched_core_t* core) +{ + (void)core; + return 0; +} + +int xce_core_end_session(xce_sched_core_t* core) +{ + (void)core; + return 0; +} + +int plugin_begin_session(void* arg) +{ + (void)arg; + return 0; +} + +int plugin_end_session(void* arg) +{ + (void)arg; + return 0; +} + +int sched_account_jobs(xce_scheduler_t* sched) +{ + (void)sched; + return 0; +} + +/* Link-satisfying stubs for recording_session.c's real functions -- see + * this file's header comment for why recording_session.c itself is not + * linked. Never actually invoked (no session is ever run). */ +sched_record_session_ctx_t* sched_record_session_begin(xce_scheduler_t* sched, + uint64_t begin_round_ts) +{ + (void)sched; + (void)begin_round_ts; + return NULL; +} + +void sched_record_session_add_decision(sched_record_session_ctx_t* ctx, + uint64_t ts, xce_decision_t* decision) +{ + (void)ctx; + (void)ts; + (void)decision; +} + +int sched_record_session_end(xce_scheduler_t* sched, + sched_record_session_ctx_t* ctx, uint64_t end_ts) +{ + (void)sched; + (void)ctx; + (void)end_ts; + return 0; +} + +/* Real reimplementation of schedobj.c's sched_resolve_job_picker -- same + * rationale as test_driver_translate.c's own copy (schedobj.c is not + * linked). Never actually exercised (no test here pushes core events, so + * xce_core_push_events() -- the only real caller -- is never reached: every + * enqueued request in this file is XCE_SCHED_OP_ADMIN_RECORD, intercepted + * by driver.c BEFORE the translated batch/push path). Present only to + * satisfy the linker for core_events.c. */ +int sched_resolve_job_picker(xce_sched_core_t* core, xce_sched_plugin_t** out) +{ + (void)core; + (void)out; + return -EINVAL; +} + +/* Link-satisfying stub for scheduler.c's sched_init_config() -- handler.c's + * (real, #included below) OPCode_push/configpushmsg case references it; + * this file never exercises that path (out of scope -- this file's whole + * surface is OPCode_schedAdmin). Never actually invoked. */ +int sched_init_config(xce_scheduler_t* s, xce_conf4sched_t* c) +{ + (void)s; + (void)c; + return 0; +} + +/* Pull in the real sched_handle_admin_request()/scheduler_req_handler() + * under test into this TU. */ +#include "../src/handler.c" + +/* ---------------------------------------------------------------------- */ +/* Spy test double for client.c's xcecli_reply_schedadmin() -- see file */ +/* header comment for why every field is deep-copied rather than retaining */ +/* the passed sched_admin_reply_t's reply->file/reply->error pointers. */ +/* ---------------------------------------------------------------------- */ + +static int g_reply_calls; +static uint64_t g_last_seq; +static xcecli_ctx_t g_last_ctx; +static int g_last_ok; +static int g_last_error_null; +static char g_last_error[256]; +static int g_last_recording; +static int g_last_file_null; +static char g_last_file[PATH_MAX]; +static uint64_t g_last_event_count; +static uint64_t g_last_started_at; + +static void reset_reply_spy(void) +{ + g_reply_calls = 0; + g_last_seq = 0; + g_last_ctx = NULL; + g_last_ok = 0; + g_last_error_null = 1; + g_last_error[0] = '\0'; + g_last_recording = 0; + g_last_file_null = 1; + g_last_file[0] = '\0'; + g_last_event_count = 0; + g_last_started_at = 0; +} + +int xcecli_reply_schedadmin(xcecli_ctx_t c, uint64_t seq, + sched_admin_reply_t* areply) +{ + g_reply_calls++; + g_last_ctx = c; + g_last_seq = seq; + + if (areply == NULL) { + return 0; + } + + g_last_ok = areply->ok; + g_last_recording = areply->recording; + g_last_event_count = areply->eventCount; + g_last_started_at = areply->startedAt; + + g_last_error_null = (areply->error == NULL); + g_last_error[0] = '\0'; + if (areply->error != NULL) { + strncpy(g_last_error, areply->error, sizeof(g_last_error) - 1); + } + + g_last_file_null = (areply->file == NULL); + g_last_file[0] = '\0'; + if (areply->file != NULL) { + strncpy(g_last_file, areply->file, sizeof(g_last_file) - 1); + } + + return 0; +} + +void setUp(void) { reset_reply_spy(); } +void tearDown(void) { return; } + +/* ---------------------------------------------------------------------- */ +/* Fixture helpers -- same pattern as test_record_lifecycle.c's own. */ +/* ---------------------------------------------------------------------- */ + +static void make_tmp_dir(char* out, size_t out_len, const char* suffix) +{ + char tmpl[PATH_MAX]; + + snprintf(tmpl, sizeof(tmpl), "/tmp/sched_adminreq%s_XXXXXX", suffix); + TEST_ASSERT_NOT_NULL(mkdtemp(tmpl)); + TEST_ASSERT_TRUE(strlen(tmpl) < out_len); + strcpy(out, tmpl); +} + +/* `record_dir` must outlive `sched` (caller-owned storage), same convention + * as test_record_lifecycle.c's fixture_init(). No picker plugin is + * registered -- unlike test_record_lifecycle.c's fixture, no test in this + * file ever runs a scheduling session, so core->job_pick_plugin staying + * NULL is fine (sched_resolve_job_picker() above is never actually + * reached). */ +static void fixture_init(xce_scheduler_t* sched, char* record_dir) +{ + int rc; + + memset(sched, 0, sizeof(*sched)); + + /* This fixture hand-builds xce_scheduler_t rather than calling the full + * sched_init() (which starts real threads/network I/O this white-box + * suite never wants) -- record_lock (scheduler.h field comments, task + * 9.4 fix round) is normally initialized by sched_init(), so it must be + * initialized here too, the same way core.lock is initialized by the + * xce_sched_core_init() call just below. Every test in this file + * exercises sched_record_status()/sched_record_start()/ + * sched_record_stop()/sched_record_write_events() for real, all of + * which now take this lock. */ + mylock_init(&(sched->record_lock)); + + rc = xce_sched_core_init(&(sched->core)); + TEST_ASSERT_EQUAL_INT(0, rc); + + sched->core.pend_updates[0] = (dataq_t*)calloc(1, sizeof(dataq_t)); + dataq_init(sched->core.pend_updates[0], NULL); + sched->core.pend_updates[1] = (dataq_t*)calloc(1, sizeof(dataq_t)); + dataq_init(sched->core.pend_updates[1], NULL); + + sched->record_dir = record_dir; + /* Dummy non-NULL xcecli_ctx_t (an opaque void*) -- xcecli_reply_schedadmin + * is spied above, so nothing ever dereferences this; a real production + * ctx would come from xcecli_ctx_init(). */ + sched->apictx = (xcecli_ctx_t)sched; +} + +/* Builds a request/reply pair on the stack and drives it through the REAL + * production entry point, scheduler_req_handler() (handler.c) -- exactly + * the function main.c registers as the scheduler's request handler and the + * function that dispatches OPCode_schedAdmin to sched_handle_admin_request() + * (handler.c's own new `case` this task added). */ +static int send_admin_request(xce_scheduler_t* sched, int action, + const char* record_file, uint64_t seq) +{ + sched_admin_req_t areq; + xce_request_t request; + + memset(&areq, 0, sizeof(areq)); + areq.action = action; + areq.recordFile = (char*)record_file; + + memset(&request, 0, sizeof(request)); + request.opcode = OPCode_schedAdmin; + request.seq = seq; + request.body_which = Request_body_schedadminmsg; + request.body.schedadminmsg = &areq; + + return scheduler_req_handler(&request, sched); +} + +/* ---------------------------------------------------------------------- */ +/* recordStatus -- synchronous, answered directly from the handler thread. */ +/* ---------------------------------------------------------------------- */ + +static void test_recordstatus_not_recording_returns_empty_state(void) +{ + char record_dir[PATH_MAX]; + xce_scheduler_t sched; + int rc; + + make_tmp_dir(record_dir, sizeof(record_dir), "_status_idle"); + fixture_init(&sched, record_dir); + + rc = send_admin_request(&sched, SchedAdminAction_recordStatus, NULL, 111); + + TEST_ASSERT_EQUAL_INT(0, rc); + TEST_ASSERT_EQUAL_INT(1, g_reply_calls); + TEST_ASSERT_EQUAL_UINT64(111, g_last_seq); + TEST_ASSERT_EQUAL_PTR(sched.apictx, g_last_ctx); + TEST_ASSERT_TRUE(g_last_ok); + TEST_ASSERT_FALSE(g_last_recording); + TEST_ASSERT_TRUE(g_last_file_null); + TEST_ASSERT_EQUAL_UINT64(0, g_last_event_count); + TEST_ASSERT_EQUAL_UINT64(0, g_last_started_at); + + /* recordStatus never touches the deferred queue. */ + TEST_ASSERT_NULL(dataq_deq(sched.core.pend_updates[0], NULL)); +} + +static void test_recordstatus_while_recording_reflects_live_state(void) +{ + char record_dir[PATH_MAX]; + xce_scheduler_t sched; + xce_core_event_t ev; + int rc; + + make_tmp_dir(record_dir, sizeof(record_dir), "_status_live"); + fixture_init(&sched, record_dir); + + rc = sched_record_start(&sched, "live-status.replay"); + TEST_ASSERT_EQUAL_INT(0, rc); + TEST_ASSERT_TRUE(sched.recording); + + /* Real event writes via recording_events.c's genuine + * sched_record_write_events() -- record_event_count genuinely advances, + * not a fabricated field. */ + memset(&ev, 0, sizeof(ev)); + ev.type = XCE_CORE_EVENT_JOB_STATE_CHANGE; + ev.u.job_state_change.jobid = 7ULL; + TEST_ASSERT_EQUAL_INT(0, sched_record_write_events(&sched, &ev, 1)); + ev.u.job_state_change.jobid = 8ULL; + TEST_ASSERT_EQUAL_INT(0, sched_record_write_events(&sched, &ev, 1)); + TEST_ASSERT_EQUAL_UINT64(2ULL, sched.record_event_count); + + rc = send_admin_request(&sched, SchedAdminAction_recordStatus, NULL, 222); + + TEST_ASSERT_EQUAL_INT(0, rc); + TEST_ASSERT_EQUAL_INT(1, g_reply_calls); + TEST_ASSERT_TRUE(g_last_ok); + TEST_ASSERT_TRUE(g_last_recording); + TEST_ASSERT_FALSE(g_last_file_null); + TEST_ASSERT_TRUE(strstr(g_last_file, "live-status.replay") != NULL); + TEST_ASSERT_EQUAL_UINT64(2ULL, g_last_event_count); + TEST_ASSERT_TRUE(g_last_started_at > 0); + TEST_ASSERT_EQUAL_UINT64(sched.record_start_ts, g_last_started_at); +} + +/* ---------------------------------------------------------------------- */ +/* recordStatus concurrency -- record_lock (scheduler.h field comments, */ +/* task 9.4 fix round) genuinely protects the recording-state fields */ +/* against the vtask-thread sched_record_status() reader / driver-thread */ +/* writer race the code review flagged. Two DETERMINISTIC proofs rather */ +/* than a single flaky, purely timing-dependent race reproduction -- */ +/* task 9.2's TOCTOU test already established this codebase's precedent */ +/* of preferring a deterministic proof over a genuinely racy test: */ +/* */ +/* 1. record_lock is a REAL mutex a second, genuine pthread actually */ +/* blocks on -- not just "the code calls mylock_lock()" verified by */ +/* inspection. Proven by holding the lock on the main thread */ +/* (simulating "driver thread is mid-mutation"), launching a real */ +/* thread that calls sched_record_status(), and asserting it has NOT */ +/* completed after a generous wait -- then asserting it DOES */ +/* complete once the lock is released. The only timing dependency is */ +/* in the safe direction: a slow CI runner can only make the "still */ +/* blocked" assertion harder to observe, never produce a false */ +/* positive (record_lock is genuinely held for the whole wait). */ +/* 2. reply->file is a genuinely independent allocation, not an alias of */ +/* the live sched->record_file_path -- proven by taking a status */ +/* snapshot, then calling the real sched_record_stop() (which frees */ +/* sched->record_file_path for real) and confirming the snapshot's */ +/* reply->file is untouched. This reproduces the exact */ +/* use-after-free the finding identified, deterministically, with no */ +/* reliance on OS thread-scheduling timing at all. */ +/* ---------------------------------------------------------------------- */ + +typedef struct { + xce_scheduler_t* sched; + sched_admin_reply_t reply; + int rc; + atomic_int done; +} status_thread_arg_t; + +static void* status_thread_fn(void* arg) +{ + status_thread_arg_t* a = (status_thread_arg_t*)arg; + + a->rc = sched_record_status(a->sched, &(a->reply)); + atomic_store(&(a->done), 1); + + return NULL; +} + +static void test_recordstatus_genuinely_blocks_on_record_lock(void) +{ + char record_dir[PATH_MAX]; + xce_scheduler_t sched; + status_thread_arg_t arg; + pthread_t th; + + make_tmp_dir(record_dir, sizeof(record_dir), "_status_lockblock"); + fixture_init(&sched, record_dir); + + TEST_ASSERT_EQUAL_INT(0, sched_record_start(&sched, "lockblock.replay")); + + memset(&arg, 0, sizeof(arg)); + arg.sched = &sched; + atomic_store(&(arg.done), 0); + + /* Simulate "driver thread is mid-mutation": hold record_lock on THIS + * (main test) thread, exactly the window sched_record_start()/ + * sched_record_stop()/sched_record_abort_on_error() hold it for in + * production. */ + mylock_lock(&(sched.record_lock)); + + TEST_ASSERT_EQUAL_INT(0, pthread_create(&th, NULL, status_thread_fn, &arg)); + + /* Generous wait -- if record_lock were NOT actually taken by + * sched_record_status() (i.e. the fix regressed to a no-op), the + * spawned thread would complete almost instantly and `done` would + * already be 1 here. */ + usleep(150000); /* 150ms */ + + TEST_ASSERT_EQUAL_INT(0, atomic_load(&(arg.done))); + + mylock_unlock(&(sched.record_lock)); + + TEST_ASSERT_EQUAL_INT(0, pthread_join(th, NULL)); + + TEST_ASSERT_EQUAL_INT(1, atomic_load(&(arg.done))); + TEST_ASSERT_EQUAL_INT(0, arg.rc); + TEST_ASSERT_TRUE(arg.reply.recording); + TEST_ASSERT_NOT_NULL(arg.reply.file); + TEST_ASSERT_TRUE(strstr(arg.reply.file, "lockblock.replay") != NULL); + + free(arg.reply.file); + TEST_ASSERT_EQUAL_INT(0, sched_record_stop(&sched)); +} + +static void test_recordstatus_file_survives_concurrent_record_stop_free(void) +{ + char record_dir[PATH_MAX]; + xce_scheduler_t sched; + sched_admin_reply_t snapshot; + int rc; + + make_tmp_dir(record_dir, sizeof(record_dir), "_status_uaf"); + fixture_init(&sched, record_dir); + + rc = sched_record_start(&sched, "uaftest.replay"); + TEST_ASSERT_EQUAL_INT(0, rc); + + rc = sched_record_status(&sched, &snapshot); + TEST_ASSERT_EQUAL_INT(0, rc); + TEST_ASSERT_NOT_NULL(snapshot.file); + /* Genuinely a DIFFERENT allocation from the live pointer -- not an + * alias (the exact bug this fix closes: the pre-fix version set + * reply->file = sched->record_file_path directly). */ + TEST_ASSERT_TRUE(sched.record_file_path != snapshot.file); + TEST_ASSERT_EQUAL_STRING(sched.record_file_path, snapshot.file); + + /* sched_record_stop() frees sched->record_file_path for real -- if + * snapshot.file aliased it, this would be a genuine use-after-free + * (ASan would abort) the moment the next assertion reads it. */ + rc = sched_record_stop(&sched); + TEST_ASSERT_EQUAL_INT(0, rc); + TEST_ASSERT_NULL(sched.record_file_path); + + TEST_ASSERT_TRUE(strstr(snapshot.file, "uaftest.replay") != NULL); + + free(snapshot.file); +} + +/* ---------------------------------------------------------------------- */ +/* recordStart -- deferred, not synchronous. */ +/* ---------------------------------------------------------------------- */ + +static void test_recordstart_is_deferred_then_performed_by_driver_round(void) +{ + char record_dir[PATH_MAX]; + xce_scheduler_t sched; + char expect_path[PATH_MAX]; + int rc; + struct stat st; + + make_tmp_dir(record_dir, sizeof(record_dir), "_start_deferred"); + fixture_init(&sched, record_dir); + + rc = send_admin_request(&sched, SchedAdminAction_recordStart, + "admin-start.replay", 333); + TEST_ASSERT_EQUAL_INT(0, rc); + + /* Deferred, not synchronous: no start yet, no reply yet. */ + TEST_ASSERT_FALSE(sched.recording); + TEST_ASSERT_NULL(sched.record_file_path); + TEST_ASSERT_EQUAL_INT(0, g_reply_calls); + + /* A subsequent driver round actually performs the deferred start. */ + rc = xce_sched_driver_translate_pending_events(&(sched.core), &sched); + TEST_ASSERT_EQUAL_INT(0, rc); /* n==0: the admin op never enters the + * translated event batch. */ + + TEST_ASSERT_TRUE(sched.recording); + TEST_ASSERT_NOT_NULL(sched.record_file_path); + TEST_ASSERT_TRUE(strstr(sched.record_file_path, "admin-start.replay") + != NULL); + + /* The file is genuinely created on disk. */ + snprintf(expect_path, sizeof(expect_path), "%s/admin-start.replay", + record_dir); + TEST_ASSERT_EQUAL_INT(0, stat(expect_path, &st)); + TEST_ASSERT_TRUE(S_ISREG(st.st_mode)); + + /* And the deferred reply was genuinely sent, matching the "recordStart + * (ok)" column of requirements.md's field-semantics table. */ + TEST_ASSERT_EQUAL_INT(1, g_reply_calls); + TEST_ASSERT_EQUAL_UINT64(333, g_last_seq); + TEST_ASSERT_TRUE(g_last_ok); + TEST_ASSERT_TRUE(g_last_error_null); + TEST_ASSERT_TRUE(g_last_recording); + TEST_ASSERT_FALSE(g_last_file_null); + TEST_ASSERT_TRUE(strstr(g_last_file, "admin-start.replay") != NULL); + TEST_ASSERT_EQUAL_UINT64(0, g_last_event_count); + TEST_ASSERT_TRUE(g_last_started_at > 0); +} + +static void test_recordstart_empty_file_auto_generates_name(void) +{ + char record_dir[PATH_MAX]; + xce_scheduler_t sched; + int rc; + + make_tmp_dir(record_dir, sizeof(record_dir), "_start_auto"); + fixture_init(&sched, record_dir); + + /* NULL recordFile -- "empty = auto-generate" (requirements.md's + * SchedAdminRequest.recordFile field comment). */ + rc = send_admin_request(&sched, SchedAdminAction_recordStart, NULL, 444); + TEST_ASSERT_EQUAL_INT(0, rc); + + rc = xce_sched_driver_translate_pending_events(&(sched.core), &sched); + TEST_ASSERT_EQUAL_INT(0, rc); + + TEST_ASSERT_TRUE(sched.recording); + TEST_ASSERT_NOT_NULL(sched.record_file_path); + TEST_ASSERT_TRUE(strstr(sched.record_file_path, record_dir) != NULL); + TEST_ASSERT_TRUE(strstr(sched.record_file_path, ".replay") != NULL); + + TEST_ASSERT_EQUAL_INT(1, g_reply_calls); + TEST_ASSERT_TRUE(g_last_ok); + TEST_ASSERT_FALSE(g_last_file_null); +} + +static void test_recordstart_while_already_recording_replies_error(void) +{ + char record_dir[PATH_MAX]; + xce_scheduler_t sched; + char path_before[PATH_MAX]; + uint64_t started_before; + int rc; + + make_tmp_dir(record_dir, sizeof(record_dir), "_start_already"); + fixture_init(&sched, record_dir); + + rc = sched_record_start(&sched, "first.replay"); + TEST_ASSERT_EQUAL_INT(0, rc); + strcpy(path_before, sched.record_file_path); + started_before = sched.record_start_ts; + + rc = send_admin_request(&sched, SchedAdminAction_recordStart, + "second.replay", 555); + TEST_ASSERT_EQUAL_INT(0, rc); + + rc = xce_sched_driver_translate_pending_events(&(sched.core), &sched); + TEST_ASSERT_EQUAL_INT(0, rc); + + /* Original recording untouched by the rejected second start. */ + TEST_ASSERT_TRUE(sched.recording); + TEST_ASSERT_EQUAL_STRING(path_before, sched.record_file_path); + TEST_ASSERT_EQUAL_UINT64(0ULL, sched.record_event_count); + + TEST_ASSERT_EQUAL_INT(1, g_reply_calls); + TEST_ASSERT_EQUAL_UINT64(555, g_last_seq); + TEST_ASSERT_FALSE(g_last_ok); + TEST_ASSERT_FALSE(g_last_error_null); + TEST_ASSERT_EQUAL_STRING("already recording", g_last_error); + /* "(unchanged)" -- requirements.md's recordStart(err) column. */ + TEST_ASSERT_TRUE(g_last_recording); + TEST_ASSERT_FALSE(g_last_file_null); + TEST_ASSERT_EQUAL_STRING(path_before, g_last_file); + TEST_ASSERT_EQUAL_UINT64(0, g_last_event_count); + TEST_ASSERT_EQUAL_UINT64(started_before, g_last_started_at); + + TEST_ASSERT_EQUAL_INT(0, sched_record_stop(&sched)); +} + +static void test_recordstart_invalid_filename_replies_error(void) +{ + char record_dir[PATH_MAX]; + xce_scheduler_t sched; + char escape_path[PATH_MAX]; + int rc; + + make_tmp_dir(record_dir, sizeof(record_dir), "_start_invalid"); + fixture_init(&sched, record_dir); + + /* Contains a path separator -- rejected before any filesystem access + * (recording.c's sched_validate_record_filename(), task 9.1). */ + rc = send_admin_request(&sched, SchedAdminAction_recordStart, + "sub/dir.replay", 666); + TEST_ASSERT_EQUAL_INT(0, rc); + + rc = xce_sched_driver_translate_pending_events(&(sched.core), &sched); + TEST_ASSERT_EQUAL_INT(0, rc); + + /* No file opened, no directory created, recording never started. */ + TEST_ASSERT_FALSE(sched.recording); + TEST_ASSERT_NULL(sched.record_file_path); + TEST_ASSERT_NULL(sched.record_file); + snprintf(escape_path, sizeof(escape_path), "%s/sub", record_dir); + TEST_ASSERT_FALSE( + (access(escape_path, F_OK) == 0)); /* "sub" was never created */ + + TEST_ASSERT_EQUAL_INT(1, g_reply_calls); + TEST_ASSERT_FALSE(g_last_ok); + TEST_ASSERT_FALSE(g_last_error_null); + TEST_ASSERT_EQUAL_STRING("invalid record file name", g_last_error); + TEST_ASSERT_FALSE(g_last_recording); + TEST_ASSERT_TRUE(g_last_file_null); + TEST_ASSERT_EQUAL_UINT64(0, g_last_event_count); + TEST_ASSERT_EQUAL_UINT64(0, g_last_started_at); +} + +/* ---------------------------------------------------------------------- */ +/* recordStop -- deferred, not synchronous. */ +/* ---------------------------------------------------------------------- */ + +static void test_recordstop_is_deferred_then_performed_by_driver_round(void) +{ + char record_dir[PATH_MAX]; + xce_scheduler_t sched; + char path_before[PATH_MAX]; + uint64_t started_before; + xce_core_event_t ev; + int rc; + + make_tmp_dir(record_dir, sizeof(record_dir), "_stop_deferred"); + fixture_init(&sched, record_dir); + + rc = sched_record_start(&sched, "stopme.replay"); + TEST_ASSERT_EQUAL_INT(0, rc); + strcpy(path_before, sched.record_file_path); + started_before = sched.record_start_ts; + + memset(&ev, 0, sizeof(ev)); + ev.type = XCE_CORE_EVENT_JOB_STATE_CHANGE; + ev.u.job_state_change.jobid = 9ULL; + TEST_ASSERT_EQUAL_INT(0, sched_record_write_events(&sched, &ev, 1)); + TEST_ASSERT_EQUAL_UINT64(1ULL, sched.record_event_count); + + rc = send_admin_request(&sched, SchedAdminAction_recordStop, NULL, 777); + TEST_ASSERT_EQUAL_INT(0, rc); + + /* Deferred, not synchronous: still recording, no reply yet. */ + TEST_ASSERT_TRUE(sched.recording); + TEST_ASSERT_EQUAL_INT(0, g_reply_calls); + + rc = xce_sched_driver_translate_pending_events(&(sched.core), &sched); + TEST_ASSERT_EQUAL_INT(0, rc); + + TEST_ASSERT_FALSE(sched.recording); + TEST_ASSERT_NULL(sched.record_file_path); + TEST_ASSERT_NULL(sched.record_file); + + TEST_ASSERT_EQUAL_INT(1, g_reply_calls); + TEST_ASSERT_EQUAL_UINT64(777, g_last_seq); + TEST_ASSERT_TRUE(g_last_ok); + TEST_ASSERT_TRUE(g_last_error_null); + TEST_ASSERT_FALSE(g_last_recording); + /* "last path (informational)". */ + TEST_ASSERT_FALSE(g_last_file_null); + TEST_ASSERT_EQUAL_STRING(path_before, g_last_file); + /* "final count". */ + TEST_ASSERT_EQUAL_UINT64(1ULL, g_last_event_count); + /* "original start (historical)". */ + TEST_ASSERT_EQUAL_UINT64(started_before, g_last_started_at); +} + +static void test_recordstop_while_not_recording_replies_error(void) +{ + char record_dir[PATH_MAX]; + xce_scheduler_t sched; + int rc; + + make_tmp_dir(record_dir, sizeof(record_dir), "_stop_notrecording"); + fixture_init(&sched, record_dir); + + rc = send_admin_request(&sched, SchedAdminAction_recordStop, NULL, 888); + TEST_ASSERT_EQUAL_INT(0, rc); + TEST_ASSERT_EQUAL_INT(0, g_reply_calls); /* deferred */ + + rc = xce_sched_driver_translate_pending_events(&(sched.core), &sched); + TEST_ASSERT_EQUAL_INT(0, rc); + + TEST_ASSERT_FALSE(sched.recording); + + TEST_ASSERT_EQUAL_INT(1, g_reply_calls); + TEST_ASSERT_EQUAL_UINT64(888, g_last_seq); + TEST_ASSERT_FALSE(g_last_ok); + TEST_ASSERT_FALSE(g_last_error_null); + TEST_ASSERT_EQUAL_STRING("not recording", g_last_error); + TEST_ASSERT_FALSE(g_last_recording); + TEST_ASSERT_TRUE(g_last_file_null); + TEST_ASSERT_EQUAL_UINT64(0, g_last_event_count); + TEST_ASSERT_EQUAL_UINT64(0, g_last_started_at); +} + +/* ---------------------------------------------------------------------- */ +/* Unrecognized action / NULL argument guards. */ +/* ---------------------------------------------------------------------- */ + +static void test_unrecognized_action_returns_einval_no_crash_no_reply(void) +{ + char record_dir[PATH_MAX]; + xce_scheduler_t sched; + int rc; + + make_tmp_dir(record_dir, sizeof(record_dir), "_bad_action"); + fixture_init(&sched, record_dir); + + rc = send_admin_request(&sched, 999 /* not a valid SchedAdminAction */, + NULL, 1010); + + TEST_ASSERT_EQUAL_INT(-EINVAL, rc); + TEST_ASSERT_EQUAL_INT(0, g_reply_calls); /* no synchronous reply */ + TEST_ASSERT_NULL( + dataq_deq(sched.core.pend_updates[0], NULL)); /* nothing enqueued */ + + /* A subsequent driver round is a genuine no-op -- nothing to drain. */ + rc = xce_sched_driver_translate_pending_events(&(sched.core), &sched); + TEST_ASSERT_EQUAL_INT(0, rc); + TEST_ASSERT_EQUAL_INT(0, g_reply_calls); +} + +static void test_null_sched_returns_einval(void) +{ + sched_admin_req_t areq; + xce_request_t request; + + memset(&areq, 0, sizeof(areq)); + areq.action = SchedAdminAction_recordStatus; + + memset(&request, 0, sizeof(request)); + request.opcode = OPCode_schedAdmin; + request.body_which = Request_body_schedadminmsg; + request.body.schedadminmsg = &areq; + + TEST_ASSERT_EQUAL_INT(-EINVAL, sched_handle_admin_request(NULL, &request)); + TEST_ASSERT_EQUAL_INT(0, g_reply_calls); +} + +static void test_null_request_returns_einval(void) +{ + char record_dir[PATH_MAX]; + xce_scheduler_t sched; + + make_tmp_dir(record_dir, sizeof(record_dir), "_null_req"); + fixture_init(&sched, record_dir); + + TEST_ASSERT_EQUAL_INT(-EINVAL, sched_handle_admin_request(&sched, NULL)); + TEST_ASSERT_EQUAL_INT(0, g_reply_calls); +} + +static void test_null_schedadminmsg_body_returns_einval(void) +{ + char record_dir[PATH_MAX]; + xce_scheduler_t sched; + xce_request_t request; + + make_tmp_dir(record_dir, sizeof(record_dir), "_null_body"); + fixture_init(&sched, record_dir); + + memset(&request, 0, sizeof(request)); + request.opcode = OPCode_schedAdmin; + request.body_which = Request_body_schedadminmsg; + request.body.schedadminmsg = NULL; + + TEST_ASSERT_EQUAL_INT(-EINVAL, + sched_handle_admin_request(&sched, &request)); + TEST_ASSERT_EQUAL_INT(0, g_reply_calls); +} + +int main(void) +{ + UNITY_BEGIN(); + + RUN_TEST(test_recordstatus_not_recording_returns_empty_state); + RUN_TEST(test_recordstatus_while_recording_reflects_live_state); + RUN_TEST(test_recordstatus_genuinely_blocks_on_record_lock); + RUN_TEST(test_recordstatus_file_survives_concurrent_record_stop_free); + + RUN_TEST(test_recordstart_is_deferred_then_performed_by_driver_round); + RUN_TEST(test_recordstart_empty_file_auto_generates_name); + RUN_TEST(test_recordstart_while_already_recording_replies_error); + RUN_TEST(test_recordstart_invalid_filename_replies_error); + + RUN_TEST(test_recordstop_is_deferred_then_performed_by_driver_round); + RUN_TEST(test_recordstop_while_not_recording_replies_error); + + RUN_TEST(test_unrecognized_action_returns_einval_no_crash_no_reply); + RUN_TEST(test_null_sched_returns_einval); + RUN_TEST(test_null_request_returns_einval); + RUN_TEST(test_null_schedadminmsg_body_returns_einval); + + return UNITY_END(); +} diff --git a/src/scheduler/testsuite/test_driver_decision_cb.c b/src/scheduler/testsuite/test_driver_decision_cb.c index bf1d8caf..741711b6 100644 --- a/src/scheduler/testsuite/test_driver_decision_cb.c +++ b/src/scheduler/testsuite/test_driver_decision_cb.c @@ -134,6 +134,24 @@ int sched_record_write_events(xce_scheduler_t* sched, xce_core_event_t* events, return 0; } +/* sched_record_start/sched_record_stop (scheduler-core spec, task 9.4) -- + * driver.c's handle_deferred_admin_record() (new this task) calls both; + * this file never enqueues an XCE_SCHED_OP_ADMIN_RECORD request, so + * neither is actually invoked at runtime, same rationale as + * sched_record_write_events above. */ +int sched_record_start(xce_scheduler_t* sched, const char* record_file) +{ + (void)sched; + (void)record_file; + return 0; +} + +int sched_record_stop(xce_scheduler_t* sched) +{ + (void)sched; + return 0; +} + sched_record_session_ctx_t* sched_record_session_begin(xce_scheduler_t* sched, uint64_t begin_round_ts) { diff --git a/src/scheduler/testsuite/test_driver_session_ordering.c b/src/scheduler/testsuite/test_driver_session_ordering.c index 18e55d5a..d4dc169c 100644 --- a/src/scheduler/testsuite/test_driver_session_ordering.c +++ b/src/scheduler/testsuite/test_driver_session_ordering.c @@ -229,6 +229,24 @@ int sched_record_write_events(xce_scheduler_t* sched, xce_core_event_t* events, return 0; } +/* sched_record_start/sched_record_stop (scheduler-core spec, task 9.4) -- + * driver.c's handle_deferred_admin_record() (new this task) calls both; + * this file never enqueues an XCE_SCHED_OP_ADMIN_RECORD request, so + * neither is actually invoked at runtime, same rationale as + * sched_record_write_events above. */ +int sched_record_start(xce_scheduler_t* sched, const char* record_file) +{ + (void)sched; + (void)record_file; + return 0; +} + +int sched_record_stop(xce_scheduler_t* sched) +{ + (void)sched; + return 0; +} + sched_record_session_ctx_t* sched_record_session_begin(xce_scheduler_t* sched, uint64_t begin_round_ts) { diff --git a/src/scheduler/testsuite/test_driver_translate.c b/src/scheduler/testsuite/test_driver_translate.c index 2ea78fb9..cad98570 100644 --- a/src/scheduler/testsuite/test_driver_translate.c +++ b/src/scheduler/testsuite/test_driver_translate.c @@ -112,6 +112,24 @@ int sched_record_write_events(xce_scheduler_t* sched, xce_core_event_t* events, return 0; } +/* sched_record_start/sched_record_stop (scheduler-core spec, task 9.4) -- + * driver.c's handle_deferred_admin_record() (new this task) calls both; + * this file never enqueues an XCE_SCHED_OP_ADMIN_RECORD request, so + * neither is actually invoked at runtime, same rationale as + * sched_record_write_events above. */ +int sched_record_start(xce_scheduler_t* sched, const char* record_file) +{ + (void)sched; + (void)record_file; + return 0; +} + +int sched_record_stop(xce_scheduler_t* sched) +{ + (void)sched; + return 0; +} + sched_record_session_ctx_t* sched_record_session_begin(xce_scheduler_t* sched, uint64_t begin_round_ts) { diff --git a/src/scheduler/testsuite/test_handlers.c b/src/scheduler/testsuite/test_handlers.c index 5134ffe5..e8c98426 100644 --- a/src/scheduler/testsuite/test_handlers.c +++ b/src/scheduler/testsuite/test_handlers.c @@ -27,14 +27,30 @@ * test_job_array_expand.c for the real kvmgr-backed sched_get_job). */ int sched_init_config(xce_scheduler_t* s, xce_conf4sched_t* c) { - (void)s; (void)c; return 0; + (void)s; + (void)c; + return 0; } /* sched_get_job relocated to core.h (scheduler-core spec, task 4.1) -- now * operates on xce_sched_core_t*, not xce_scheduler_t*. */ xce_sched_job_t* sched_get_job(xce_sched_core_t* sched, uint64_t jobid) { - (void)sched; (void)jobid; return NULL; + (void)sched; + (void)jobid; + return NULL; +} + +/* sched_record_status (recording.c, scheduler-core spec, task 9.4) -- + * handler.c's OPCode_schedAdmin/recordStatus path references it; this file + * never exercises OPCode_schedAdmin (its own scope is notify/dataset + * enqueue), so a link-satisfying stub is enough, same rationale as + * sched_init_config above. */ +int sched_record_status(xce_scheduler_t* sched, sched_admin_reply_t* reply) +{ + (void)sched; + (void)reply; + return -EINVAL; } #include "../src/handler.c" @@ -118,8 +134,7 @@ static void test_handle_job_rusage_enqueues_rusage(void) xce_sched_op_t op = (xce_sched_op_t)-1; fixture_init(&sched); - TEST_ASSERT_EQUAL_INT(-EINVAL, - sched_handle_sched_job_rusage(&sched, NULL)); + TEST_ASSERT_EQUAL_INT(-EINVAL, sched_handle_sched_job_rusage(&sched, NULL)); xce_job_rusages_t* ru = (xce_job_rusages_t*)calloc(1, sizeof(*ru)); ru->jobid = 7; @@ -132,20 +147,18 @@ static void test_handle_job_rusage_enqueues_rusage(void) static void test_handle_update_resvals_skips_empty_enqueues_nonempty(void) { xce_scheduler_t sched; - xce_sched_op_t op = (xce_sched_op_t)-1; + xce_sched_op_t op = (xce_sched_op_t)-1; xce_resval_list_t empty = { .count = 0, .vals = NULL }; xce_resval_list_t full = { .count = 1, .vals = NULL }; fixture_init(&sched); /* NULL or count==0 -> skip (return 0, nothing enqueued) */ TEST_ASSERT_EQUAL_INT(0, sched_handle_sched_update_resvals(&sched, NULL)); - TEST_ASSERT_EQUAL_INT(0, - sched_handle_sched_update_resvals(&sched, &empty)); + TEST_ASSERT_EQUAL_INT(0, sched_handle_sched_update_resvals(&sched, &empty)); TEST_ASSERT_EQUAL_INT(0, pop_op(&sched, NULL)); /* count>0 -> enqueue UPDATE_RESVALS */ - TEST_ASSERT_EQUAL_INT(0, - sched_handle_sched_update_resvals(&sched, &full)); + TEST_ASSERT_EQUAL_INT(0, sched_handle_sched_update_resvals(&sched, &full)); TEST_ASSERT_EQUAL_INT(1, pop_op(&sched, &op)); TEST_ASSERT_EQUAL_INT(XCE_SCHED_OP_UPDATE_RESVALS, op); } @@ -154,9 +167,9 @@ static void test_handle_update_resvals_skips_empty_enqueues_nonempty(void) static void test_handle_resources_skips_empty_enqueues_nonempty(void) { - xce_scheduler_t sched; - xce_sched_op_t op = (xce_sched_op_t)-1; - xce_resource_t* one[1]; + xce_scheduler_t sched; + xce_sched_op_t op = (xce_sched_op_t)-1; + xce_resource_t* one[1]; fixture_init(&sched); TEST_ASSERT_EQUAL_INT(0, sched_handle_resources(&sched, 0, NULL)); @@ -188,7 +201,8 @@ static void test_handle_resvals_full_skips_empty_enqueues_nonempty(void) int main(int argc, char* argv[]) { - (void)argc; (void)argv; + (void)argc; + (void)argv; UNITY_BEGIN(); RUN_TEST(test_handle_update_job_enqueues_job_fields, 0); RUN_TEST(test_handle_update_queue_enqueues_queue, 0); diff --git a/src/scheduler/testsuite/test_job_array_expand.c b/src/scheduler/testsuite/test_job_array_expand.c index cb3e026c..a574239c 100644 --- a/src/scheduler/testsuite/test_job_array_expand.c +++ b/src/scheduler/testsuite/test_job_array_expand.c @@ -29,7 +29,20 @@ int sched_init_config(xce_scheduler_t* s, xce_conf4sched_t* c) { - (void)s; (void)c; return 0; + (void)s; + (void)c; + return 0; +} + +/* sched_record_status (recording.c, scheduler-core spec, task 9.4) -- + * handler.c's OPCode_schedAdmin/recordStatus path references it; this file + * never exercises OPCode_schedAdmin, so a link-satisfying stub is enough, + * same rationale as sched_init_config above. */ +int sched_record_status(xce_scheduler_t* sched, sched_admin_reply_t* reply) +{ + (void)sched; + (void)reply; + return -EINVAL; } /* REAL kvmgr-backed job add/get -- same rtti path production uses (mirrors @@ -84,7 +97,7 @@ static void fixture_init(xce_scheduler_t* sched) static void seed_existing_job(xce_scheduler_t* sched, uint64_t jobid) { xce_sched_job_t* job = (xce_sched_job_t*)calloc(1, sizeof(*job)); - job->jobid = jobid; + job->jobid = jobid; sched_add_job(&(sched->core), job); } @@ -99,9 +112,8 @@ static void seed_existing_job(xce_scheduler_t* sched, uint64_t jobid) static xce_job_submit_t* init_job_array(uint64_t base_jobid, uint64_t array_size) { - job_submit_data_t* submit - = (job_submit_data_t*)calloc(1, sizeof(*submit)); - xce_job_submit_t* ja = (xce_job_submit_t*)calloc(1, sizeof(*ja)); + job_submit_data_t* submit = (job_submit_data_t*)calloc(1, sizeof(*submit)); + xce_job_submit_t* ja = (xce_job_submit_t*)calloc(1, sizeof(*ja)); submit->queue = strdup("normal"); submit->user = strdup("tester"); @@ -145,12 +157,12 @@ static int pop_job_update(xce_scheduler_t* sched, xce_job_t*** jobs_out) static void test_job_array_push_expands_into_n_tasks(void) { - xce_scheduler_t sched; - xce_job_submit_t* ja; - dataset_t dataset; - xce_job_t** jobs = NULL; - int njobs; - int i; + xce_scheduler_t sched; + xce_job_submit_t* ja; + dataset_t dataset; + xce_job_t** jobs = NULL; + int njobs; + int i; fixture_init(&sched); ja = init_job_array(5000, 3); @@ -188,11 +200,11 @@ static void test_job_array_push_expands_into_n_tasks(void) static void test_job_array_redelivery_does_not_duplicate(void) { - xce_scheduler_t sched; - dataset_t dataset; - xce_job_t** jobs = NULL; - int njobs; - int i; + xce_scheduler_t sched; + dataset_t dataset; + xce_job_t** jobs = NULL; + int njobs; + int i; fixture_init(&sched); @@ -289,9 +301,9 @@ static void test_wire_dispatch_chain_delivers_n_tasks(void) xce_data_t data; xce_request_t request; xce_request_t decoded; - io_packet_t *pkt; + io_packet_t* pkt; dataset_t dataset; - xce_job_t **jobs = NULL; + xce_job_t** jobs = NULL; int njobs; int i; @@ -319,12 +331,13 @@ static void test_wire_dispatch_chain_delivers_n_tasks(void) request.body_which = Request_body_pushmsg; request.body.pushmsg = &data; - pkt = (io_packet_t *)calloc(1, sizeof(*pkt)); + pkt = (io_packet_t*)calloc(1, sizeof(*pkt)); TEST_ASSERT_NOT_NULL(pkt); TEST_ASSERT_EQUAL_INT(1, xce_encode(&request, pkt)); memset(&decoded, 0, sizeof(decoded)); - TEST_ASSERT_EQUAL_INT(0, xce_decode(pkt, &decoded)); /* consumes/frees pkt */ + TEST_ASSERT_EQUAL_INT(0, + xce_decode(pkt, &decoded)); /* consumes/frees pkt */ /* Act 1: the REAL transport dispatcher (task 5.3b's fix) -- exactly what * vtask_entry() (xcectx.c) calls on every inbound push. */ @@ -364,7 +377,8 @@ static void test_wire_dispatch_chain_delivers_n_tasks(void) int main(int argc, char* argv[]) { - (void)argc; (void)argv; + (void)argc; + (void)argv; UNITY_BEGIN(); RUN_TEST(test_job_array_push_expands_into_n_tasks, 0); RUN_TEST(test_job_array_redelivery_does_not_duplicate, 0); diff --git a/src/scheduler/testsuite/test_job_array_wire_e2e.c b/src/scheduler/testsuite/test_job_array_wire_e2e.c index 52074978..691b971b 100644 --- a/src/scheduler/testsuite/test_job_array_wire_e2e.c +++ b/src/scheduler/testsuite/test_job_array_wire_e2e.c @@ -79,6 +79,17 @@ int sched_init_config(xce_scheduler_t* s, xce_conf4sched_t* c) return 0; } +/* sched_record_status (recording.c, scheduler-core spec, task 9.4) -- + * handler.c's OPCode_schedAdmin/recordStatus path references it; this file + * never exercises OPCode_schedAdmin, so a link-satisfying stub is enough, + * same rationale as sched_init_config above. */ +int sched_record_status(xce_scheduler_t* sched, sched_admin_reply_t* reply) +{ + (void)sched; + (void)reply; + return -EINVAL; +} + /* REAL kvmgr-backed job add/get -- same rtti path production uses (mirrors * test_job_array_expand.c's precedent), needed so sched_on_job_array()'s * per-element dedup check (sched_get_job) is exercised for real. Relocated diff --git a/src/scheduler/testsuite/test_record_lifecycle.c b/src/scheduler/testsuite/test_record_lifecycle.c index 6b9bb472..c1a32ddd 100644 --- a/src/scheduler/testsuite/test_record_lifecycle.c +++ b/src/scheduler/testsuite/test_record_lifecycle.c @@ -342,6 +342,15 @@ static void fixture_init(xce_scheduler_t* sched, char* record_dir) memset(sched, 0, sizeof(*sched)); + /* record_lock (scheduler.h field comments, task 9.4 fix round) is + * normally initialized by sched_init(), which this white-box fixture + * deliberately bypasses (same rationale as this function's own header + * comment) -- initialize it here too, same pairing as the + * xce_sched_core_init() call just below initializing core.lock. Every + * sched_record_start()/sched_record_stop() call in this file now takes + * this lock. */ + mylock_init(&(sched->record_lock)); + rc = xce_sched_core_init(&(sched->core)); TEST_ASSERT_EQUAL_INT(0, rc);