feat(scheduler-core): implement record start/stop lifecycle (task 9.2)
Adds recording-state fields directly to xce_scheduler_t (no new
xce_sched_driver_t type, per task 9.1's precedent). sched_record_start()/
sched_record_stop() implement AC 10: validate via task 9.1's path
validator, open with a TOCTOU-hardened sequence (O_NOFOLLOW, no O_TRUNC,
fstat()+st_nlink==1 check before ftruncate()), snapshot the core, write
ReplayHeader+SnapshotEntry on start, ReplayEvent{endSession} on stop.
driver.c gains boolean-gated recording hooks in
xce_sched_driver_translate_pending_events() (per-event-batch ReplayEvent
writes, new recording_events.c) and xce_sched_driver_run_normal_mode_round()
(per-session ReplaySession writes, new recording_session.c) -- zero I/O/
allocation when inactive.
Two real bugs found and fixed during implementation: bug-1498 (O_TRUNC
would have truncated a hard-linked victim file before the TOCTOU check
could reject it) and bug-1499 (a transient edit dropped
xce_core_signal_go(), causing a 3-way thread deadlock).
test_record_lifecycle.c decodes the full start->events->session->stop
cycle with the real capnp decoders and asserts the exact message
sequence; TOCTOU regression tests use real symlink(2)/link(2) syscalls.
25/25 scheduler suite, clang-format clean.
This commit is contained in:
parent
4e5547ee55
commit
ff8d4f416e
SSH key fingerprint:
SHA256:8Vf2pZ3h6rJa7EhGK+WaAQ9Mta/apZvteW6g9U/WKKk
15 changed files with 2110 additions and 36 deletions
|
|
@ -49,8 +49,26 @@
|
|||
* translation batch (rare, defensive path); or `xce_core_push_events()`'s
|
||||
* own negative return (e.g. -EBUSY if a session was unexpectedly active)
|
||||
* passed straight through.
|
||||
*
|
||||
* `record_ctx` (scheduler-core spec, task 9.2, Req 5 AC 10): an opaque
|
||||
* `void*`, cast back to `xce_scheduler_t*` inside driver.c, for the per-
|
||||
* event-batch recording-write hook -- same `void*`-instead-of-
|
||||
* `xce_scheduler_t*` idiom xce_sched_driver_on_decision()/
|
||||
* xce_sched_driver_run_normal_mode_round() below already use, and for the
|
||||
* same reason (this file's boundary comment: depends on core.h only).
|
||||
* Pass NULL to disable recording for this call (zero I/O, zero extra
|
||||
* allocation, Req 5 AC 10 "Performance") -- every existing caller that
|
||||
* does not care about recording does exactly this.
|
||||
*
|
||||
* Deliberately a SEPARATE parameter from `core->driver_ctx`
|
||||
* (xce_core_set_decision_cb()'s decision-callback context): that field is
|
||||
* only an `xce_scheduler_t*` in production wiring (main.c) -- some tests
|
||||
* register it as a different, smaller struct for their own decision-
|
||||
* callback spy, so this function must not assume it is safe to
|
||||
* reinterpret as `xce_scheduler_t*`.
|
||||
*/
|
||||
extern int xce_sched_driver_translate_pending_events(xce_sched_core_t* core);
|
||||
extern int xce_sched_driver_translate_pending_events(xce_sched_core_t* core,
|
||||
void* record_ctx);
|
||||
|
||||
/*
|
||||
* xce_sched_driver_on_decision - the normal-mode driver's decision-publish
|
||||
|
|
|
|||
|
|
@ -3,16 +3,21 @@
|
|||
#define _RECORDING_H_ 1
|
||||
|
||||
#include <stddef.h>
|
||||
#include <stdint.h>
|
||||
|
||||
#include "scheduler.h"
|
||||
|
||||
/*
|
||||
* recording.h - scheduler-core spec, task 9.1 (Req 5 AC 10, "Record file
|
||||
* path validation"). First piece of design.md's File Structure Plan
|
||||
* `recording.c` ("NEW (Phase 3): record start/stop/status, ReplayEvent/
|
||||
* ReplaySession writes") -- this task adds ONLY the validation function
|
||||
* every future write must pass through first; the actual record start/
|
||||
* stop/status lifecycle (design.md's NormalDriver service interface --
|
||||
* sched_driver_record_start()/_stop()/_status(), `xce_sched_driver_t`'s
|
||||
* recording-state fields) is task 9.2's job and is NOT implemented here.
|
||||
* recording.h - scheduler-core spec, tasks 9.1/9.2 (Req 5 AC 10, "Record
|
||||
* file path validation" and "Runtime recording control -- detailed
|
||||
* mechanism"). Task 9.1 added ONLY sched_validate_record_filename() below,
|
||||
* the validation function every write must pass through first. Task 9.2
|
||||
* (this revision) adds the actual record start/stop lifecycle
|
||||
* (sched_record_start()/sched_record_stop()) plus the recording-write hooks
|
||||
* driver.c's normal-mode loop calls into -- design.md's NormalDriver
|
||||
* service interface, adapted to this codebase's single combined
|
||||
* `xce_scheduler_t` driver-level struct (there is no separate
|
||||
* `xce_sched_driver_t` -- see scheduler.h's field comment on `recording`).
|
||||
*
|
||||
* sched_validate_record_filename - validates a caller-supplied, admin-
|
||||
* authenticated but UNTRUSTED record file name (controller.capnp's
|
||||
|
|
@ -113,19 +118,155 @@
|
|||
* rejection in step 1 returns -EINVAL without a single filesystem call,
|
||||
* even if `record_dir` itself does not exist.
|
||||
*
|
||||
* NOTE for task 9.2 (not implemented here -- this task is validation-only):
|
||||
* this function's containment/aliasing checks are only valid up to the
|
||||
* moment it returns -- a TOCTOU race could still swap a symlink or hard
|
||||
* link into `out_path` between this call returning and task 9.2's eventual
|
||||
* fopen(). When task 9.2 opens `out_path` for writing, it should (a) pass
|
||||
* O_NOFOLLOW so a symlink swapped in after validation is refused at
|
||||
* open() rather than followed, and (b) fstat() the resulting fd and
|
||||
* reject unless st_nlink == 1, since O_NOFOLLOW alone does not defend
|
||||
* against a hard link swapped in after validation (no symlink is involved
|
||||
* in that race, so O_NOFOLLOW never triggers).
|
||||
* TOCTOU note (closed by sched_record_start() below, task 9.2): this
|
||||
* function's containment/aliasing checks are only valid up to the moment it
|
||||
* returns -- a TOCTOU race could still swap a symlink or hard link into
|
||||
* `out_path` between this call returning and the eventual fopen()/open().
|
||||
* sched_record_start() closes that window by (a) passing O_NOFOLLOW so a
|
||||
* symlink swapped in after validation is refused at open() rather than
|
||||
* followed, and (b) fstat()-ing the resulting fd and rejecting unless
|
||||
* st_nlink == 1, since O_NOFOLLOW alone does not defend against a hard link
|
||||
* swapped in after validation (no symlink is involved in that race, so
|
||||
* O_NOFOLLOW never triggers).
|
||||
*/
|
||||
extern int sched_validate_record_filename(const char* record_dir,
|
||||
const char* filename, char* out_path,
|
||||
size_t out_path_len);
|
||||
|
||||
/*
|
||||
* sched_record_start - Req 5 AC 10 (scheduler-core spec, task 9.2,
|
||||
* "Recording lifecycle -- On record start"). Begins a new recording:
|
||||
*
|
||||
* 1. Resolves/validates `record_file` (a bare filename, or NULL/"" to
|
||||
* auto-generate one) via sched_validate_record_filename() above,
|
||||
* against `sched->record_dir`.
|
||||
* 2. Opens the resolved path with the TOCTOU-hardened sequence this
|
||||
* header's own doc comment describes (O_NOFOLLOW + fstat()+
|
||||
* st_nlink==1 check) -- closes the race window
|
||||
* sched_validate_record_filename() itself flags as still-open after
|
||||
* its own return.
|
||||
* 3. Takes a full core snapshot (xce_sched_core_save_snapshot(),
|
||||
* core.h) and writes `ReplayHeader` followed by the initial
|
||||
* `SnapshotEntry` as the first two messages of the file (Req 5 AC 20:
|
||||
* "the first data message after ReplayHeader MUST be a
|
||||
* SnapshotEntry").
|
||||
* 4. On success, sets `sched->recording = true`, stores the resolved
|
||||
* path in `sched->record_file_path` (strdup'd), resets
|
||||
* `sched->record_event_count = 0`, and stamps
|
||||
* `sched->record_start_ts = xce_core_now(&sched->core)`.
|
||||
*
|
||||
* Preconditions: fails with -EALREADY if `sched->recording` is already
|
||||
* true (design.md's NormalDriver contract) -- no file is touched. Fails
|
||||
* with -EINVAL if `sched` is NULL.
|
||||
*
|
||||
* Postconditions: on success, `ReplayHeader` + the initial `SnapshotEntry`
|
||||
* are durably on disk (fwrite completed) before this function returns.
|
||||
* On any failure after opening the file (write error, snapshot failure),
|
||||
* the partially-written file is closed and this function returns a
|
||||
* negative errno WITHOUT setting `sched->recording` -- the caller may
|
||||
* retry with a fresh sched_record_start() call. Never leaves `recording`
|
||||
* true with a NULL `record_file`, or vice versa (scheduler.h's own
|
||||
* invariant comment on `xce_scheduler_t.recording`).
|
||||
*
|
||||
* `record_file` may be NULL or empty to auto-generate a timestamped name
|
||||
* (`scheduler-replay-<UTC timestamp>.replay`) beneath `sched->record_dir`.
|
||||
*/
|
||||
extern int sched_record_start(xce_scheduler_t* sched, const char* record_file);
|
||||
|
||||
/*
|
||||
* sched_record_stop - Req 5 AC 10 (scheduler-core spec, task 9.2,
|
||||
* "Recording lifecycle -- On record stop"). Ends the current recording:
|
||||
* writes a `ReplayEvent{body=endSession}` message (the end-of-recording
|
||||
* marker -- there is no separate top-level footer message type, see
|
||||
* requirements.md Req 5 AC 20), then fflush()+fclose()s `sched->record_file`
|
||||
* and clears `sched->recording`/`sched->record_file`/`sched->record_file_path`
|
||||
* (freeing the latter) regardless of whether the marker write succeeded
|
||||
* (matches Req 5 AC 10's "Error handling" clause: a failed record write
|
||||
* disables recording rather than leaving it in a half-broken active state).
|
||||
*
|
||||
* Preconditions: fails with -EALREADY, no state touched, if
|
||||
* `sched->recording` is already false. Fails with -EINVAL if `sched` is
|
||||
* NULL.
|
||||
*
|
||||
* Returns 0 on a clean stop; a negative errno if the end-of-recording
|
||||
* 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).
|
||||
*/
|
||||
extern int sched_record_stop(xce_scheduler_t* sched);
|
||||
|
||||
/*
|
||||
* sched_record_write_events - the per-event-batch recording-write hook
|
||||
* (Req 5 AC 10, "On each core_push_events() call ... For each event in the
|
||||
* batch, serialize and write the corresponding ReplayEvent message").
|
||||
* Called from xce_sched_driver_translate_pending_events() (driver.c) once
|
||||
* per drained batch, immediately after xce_core_push_events() and BEFORE
|
||||
* that function's own cleanup loop frees any translated payload (so every
|
||||
* pointer inside `events` is still guaranteed live).
|
||||
*
|
||||
* Boolean-gated: a NULL `sched`, a NULL/empty `events`, or
|
||||
* `sched->recording == false` is a zero-cost no-op (Req 5 AC 10,
|
||||
* "Performance"). On a write failure mid-batch, recording is disabled
|
||||
* (mirrors sched_record_stop()'s cleanup, minus the end-of-recording
|
||||
* marker -- the file is presumably broken already) and a negative errno is
|
||||
* returned; the caller (driver.c) does not need to react further -- normal
|
||||
* scheduling continues either way (Req 5 AC 10, "Error handling").
|
||||
*/
|
||||
extern int sched_record_write_events(xce_scheduler_t* sched,
|
||||
xce_core_event_t* events, int n);
|
||||
|
||||
/*
|
||||
* sched_record_session_ctx_t - opaque per-round accumulator for the
|
||||
* decisions a recording session collects between
|
||||
* sched_record_session_begin()/sched_record_session_end() (Req 5 AC 10,
|
||||
* "On core_go(): begin building a ReplaySession message" /
|
||||
* "In the decision drain loop: append a ReplayDecision" /
|
||||
* "On core_end_session(): finalize ... write the complete message").
|
||||
* Defined privately in recording.c; driver.c only ever holds a pointer.
|
||||
*/
|
||||
typedef struct sched_record_session_ctx_s sched_record_session_ctx_t;
|
||||
|
||||
/*
|
||||
* sched_record_session_begin - starts accumulating a ReplaySession for one
|
||||
* normal-mode round. Returns NULL (zero allocation) when `sched` is NULL,
|
||||
* `sched->recording` is false, or `sched->record_file` is NULL -- the
|
||||
* caller (xce_sched_driver_run_normal_mode_round(), driver.c) then simply
|
||||
* never calls sched_record_session_add_decision()/sched_record_session_end(),
|
||||
* satisfying Req 5 AC 10's "zero extra allocation when recording == false".
|
||||
* On success, returns a heap-allocated context stamped with
|
||||
* `begin_round_ts` (the ReplaySession.beginRound field).
|
||||
*/
|
||||
extern sched_record_session_ctx_t*
|
||||
sched_record_session_begin(xce_scheduler_t* sched, uint64_t begin_round_ts);
|
||||
|
||||
/*
|
||||
* sched_record_session_add_decision - appends one drained decision to
|
||||
* `ctx`'s in-progress ReplaySession.decisions list, tagged with `ts`. A
|
||||
* NULL `ctx` (recording inactive -- see sched_record_session_begin() above)
|
||||
* or NULL `decision` is a no-op. XCE_DECISION_PREEMPT is not yet supported
|
||||
* (mirrors driver.c's xce_sched_driver_on_decision() -- no live allocator
|
||||
* path produces one, and this is therefore currently unreachable, but is
|
||||
* handled loudly (ERROR log) rather than silently dropped or crashing) --
|
||||
* see driver.c for the shared rationale. XCE_DECISION_NONE/
|
||||
* XCE_DECISION_NEW_SESSION are likewise skipped (nothing to record).
|
||||
*/
|
||||
extern void sched_record_session_add_decision(sched_record_session_ctx_t* ctx,
|
||||
uint64_t ts,
|
||||
xce_decision_t* decision);
|
||||
|
||||
/*
|
||||
* sched_record_session_end - finalizes and writes the accumulated
|
||||
* ReplaySession message (begin_round + decisions + a SessionSummary
|
||||
* computed from `end_ts` and the tallies sched_record_session_add_decision()
|
||||
* collected) to `sched->record_file`, then frees `ctx` and everything it
|
||||
* owns. A NULL `ctx` is a no-op (0 is returned; nothing to free). On a
|
||||
* write failure, recording is disabled the same way
|
||||
* sched_record_write_events() does. Always frees `ctx` before returning,
|
||||
* even on failure or when `sched`'s recording state changed out from under
|
||||
* the caller between sched_record_session_begin() and this call.
|
||||
*/
|
||||
extern int sched_record_session_end(xce_scheduler_t* sched,
|
||||
sched_record_session_ctx_t* ctx,
|
||||
uint64_t end_ts);
|
||||
|
||||
#endif
|
||||
|
|
|
|||
|
|
@ -5,7 +5,9 @@
|
|||
#include <pthread.h>
|
||||
#include <rtti.h>
|
||||
#include <stdatomic.h>
|
||||
#include <stdbool.h>
|
||||
#include <stdint.h>
|
||||
#include <stdio.h>
|
||||
#include <time.h>
|
||||
|
||||
#include "client.h"
|
||||
|
|
@ -117,6 +119,37 @@ typedef struct {
|
|||
* xce_sched_driver_t (record_file_path/record_file/recording/
|
||||
* record_event_count/...) is task 9.2's job, not this field's. */
|
||||
char* record_dir;
|
||||
/* Req 5 AC 10 (scheduler-core spec, task 9.2): the recording lifecycle
|
||||
* state design.md sketches on a future xce_sched_driver_t -- per
|
||||
* record_dir's own comment above, this codebase keeps a single combined
|
||||
* driver-level struct (xce_scheduler_t), so these fields land here
|
||||
* instead of a new type. Owned exclusively by recording.c's
|
||||
* sched_record_start()/sched_record_stop() (recording.h) -- see those
|
||||
* functions' doc comments for the full lifecycle contract (path
|
||||
* validation + TOCTOU-hardened open on start, end-of-recording marker +
|
||||
* flush/close on stop). driver.c's normal-mode loop (driver.h) reads
|
||||
* `recording` to boolean-gate its per-event-batch/per-session recording
|
||||
* 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. */
|
||||
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. */
|
||||
char* record_file_path;
|
||||
/* replay file handle; NULL when not recording. */
|
||||
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). */
|
||||
uint64_t record_event_count;
|
||||
/* N events between periodic snapshot checkpoints (0 = disabled).
|
||||
* Reserved for task 9.3 -- inert (never read) as of task 9.2. */
|
||||
uint64_t record_snapshot_interval;
|
||||
/* logical time (xce_core_now()) recording started; 0 = not recording. */
|
||||
uint64_t record_start_ts;
|
||||
} xce_scheduler_t;
|
||||
|
||||
extern int sched_init(xce_scheduler_t*, xce_client_config_t*,
|
||||
|
|
|
|||
|
|
@ -12,6 +12,8 @@ scheduler_srcs = [
|
|||
'src/snapshot_load.c',
|
||||
'src/decision.c',
|
||||
'src/recording.c',
|
||||
'src/recording_events.c',
|
||||
'src/recording_session.c',
|
||||
'src/schedobj.c',
|
||||
'src/plugin.c',
|
||||
'src/resource.c',
|
||||
|
|
|
|||
|
|
@ -5,6 +5,7 @@
|
|||
#include "controller.capnp.h"
|
||||
#include "core.h"
|
||||
#include "driver.h"
|
||||
#include "recording.h"
|
||||
#include "scheduler.h"
|
||||
#include "utils.h"
|
||||
|
||||
|
|
@ -280,7 +281,8 @@ static int translate_one_pending_update(xce_sched_core_t* core,
|
|||
}
|
||||
}
|
||||
|
||||
int xce_sched_driver_translate_pending_events(xce_sched_core_t* core)
|
||||
int xce_sched_driver_translate_pending_events(xce_sched_core_t* core,
|
||||
void* record_ctx)
|
||||
{
|
||||
dataq_t* q;
|
||||
data_item_t* it;
|
||||
|
|
@ -374,6 +376,34 @@ int xce_sched_driver_translate_pending_events(xce_sched_core_t* core)
|
|||
|
||||
push_rc = xce_core_push_events(core, events, n);
|
||||
|
||||
/* Req 5 AC 10 (scheduler-core spec, task 9.2): per-event-batch
|
||||
* recording write. `record_ctx` is an opaque `void*` -- always the
|
||||
* owning `xce_scheduler_t*` when the caller wants recording, NULL
|
||||
* otherwise -- for the SAME reason xce_sched_driver_on_decision()/
|
||||
* xce_sched_driver_run_normal_mode_round() already take an opaque
|
||||
* `void*` rather than `xce_scheduler_t*` (this file's boundary comment:
|
||||
* driver.h depends on core.h only). Deliberately NOT read from
|
||||
* `core->driver_ctx`: that field is the decision-callback context
|
||||
* (xce_core_set_decision_cb()'s own `driver_ctx` argument), which is
|
||||
* only an `xce_scheduler_t*` in PRODUCTION wiring (main.c) -- some
|
||||
* white-box tests in this directory register it as a different,
|
||||
* smaller struct for their own decision-callback spy, so reinterpreting
|
||||
* it here would be a type-confusion bug, not just an unlikely-NULL
|
||||
* case. xce_sched_driver_run_normal_mode_round() below passes its own
|
||||
* `sched` parameter (always genuinely `xce_scheduler_t*`) explicitly.
|
||||
* A NULL `record_ctx` is the single boolean check + zero I/O + zero
|
||||
* allocation non-recording path (Req 5 AC 10, "Performance"). MUST run
|
||||
* before the cleanup loop below frees any translated payload --
|
||||
* sched_record_write_events() needs every `events[i]` pointer still
|
||||
* live. */
|
||||
{
|
||||
xce_scheduler_t* rec_sched = (xce_scheduler_t*)record_ctx;
|
||||
|
||||
if (rec_sched != NULL) {
|
||||
sched_record_write_events(rec_sched, events, n);
|
||||
}
|
||||
}
|
||||
|
||||
for (i = 0; i < n; i++) {
|
||||
release_translated_payload(core, reqs[i], (push_rc >= 0));
|
||||
free(reqs[i]);
|
||||
|
|
@ -505,8 +535,9 @@ int xce_sched_driver_on_decision(void* driver_ctx, xce_decision_t* decision)
|
|||
*/
|
||||
void xce_sched_driver_run_normal_mode_round(void* driver_ctx)
|
||||
{
|
||||
xce_scheduler_t* sched = (xce_scheduler_t*)driver_ctx;
|
||||
int rc;
|
||||
xce_scheduler_t* sched = (xce_scheduler_t*)driver_ctx;
|
||||
sched_record_session_ctx_t* rec_ctx = NULL;
|
||||
int rc;
|
||||
|
||||
if (sched == NULL) {
|
||||
return;
|
||||
|
|
@ -516,8 +547,11 @@ void xce_sched_driver_run_normal_mode_round(void* driver_ctx)
|
|||
* accumulated since the last round. Replaces sched_update()'s inline
|
||||
* dataq-swap handling. Only valid between sessions (driver.h's own doc
|
||||
* comment) -- called here, before xce_core_begin_session(), exactly
|
||||
* where sched_update() ran relative to sched_begin_session() before. */
|
||||
xce_sched_driver_translate_pending_events(&(sched->core));
|
||||
* where sched_update() ran relative to sched_begin_session() before.
|
||||
* `sched` is passed as the recording context (task 9.2) -- see that
|
||||
* function's own comment for why this is NOT read from
|
||||
* core->driver_ctx. */
|
||||
xce_sched_driver_translate_pending_events(&(sched->core), sched);
|
||||
|
||||
rc = xce_core_begin_session(&(sched->core));
|
||||
if (rc < 0) {
|
||||
|
|
@ -531,6 +565,20 @@ void xce_sched_driver_run_normal_mode_round(void* driver_ctx)
|
|||
plugin_begin_session(sched);
|
||||
sched_account_jobs(sched);
|
||||
|
||||
/* Req 5 AC 10 (scheduler-core spec, task 9.2): "On core_go(): begin
|
||||
* building a ReplaySession message with beginRound = xce_core_now(core)"
|
||||
* -- requirements.md's own annotated pseudocode places this build step
|
||||
* immediately BEFORE sched_go() (xce_core_signal_go() below), not
|
||||
* after, so it is placed here. sched_record_session_begin() itself is
|
||||
* the single boolean check + zero-allocation gate (Req 5 AC 10,
|
||||
* "Performance") -- it returns NULL immediately when `sched->recording`
|
||||
* is false, and every hook below is a cheap `rec_ctx != NULL` pointer
|
||||
* check with no further cost on that path. */
|
||||
if (sched->recording) {
|
||||
rec_ctx
|
||||
= sched_record_session_begin(sched, xce_core_now(&(sched->core)));
|
||||
}
|
||||
|
||||
xce_core_signal_go(&(sched->core));
|
||||
|
||||
for (;;) {
|
||||
|
|
@ -545,6 +593,17 @@ void xce_sched_driver_run_normal_mode_round(void* driver_ctx)
|
|||
break;
|
||||
}
|
||||
|
||||
/* Req 5 AC 10: "In the decision drain loop ... append a
|
||||
* ReplayDecision to the current ReplaySession.decisions list" --
|
||||
* captured BEFORE the publish callback below, since the decision's
|
||||
* payload pointers must be read while still guaranteed live (see
|
||||
* recording_session.c's sched_record_session_add_decision() for the
|
||||
* full ownership argument). */
|
||||
if (rec_ctx != NULL) {
|
||||
sched_record_session_add_decision(
|
||||
rec_ctx, xce_core_now(&(sched->core)), decision);
|
||||
}
|
||||
|
||||
if (sched->core.on_decision != NULL) {
|
||||
rc = sched->core.on_decision(sched->core.driver_ctx, decision);
|
||||
TRACE("%s rc %d", __func__, rc);
|
||||
|
|
@ -557,4 +616,10 @@ void xce_sched_driver_run_normal_mode_round(void* driver_ctx)
|
|||
|
||||
plugin_end_session(sched);
|
||||
xce_core_end_session(&(sched->core));
|
||||
|
||||
/* Req 5 AC 10: "On core_end_session(): finalize the ReplaySession with
|
||||
* session summary, write the complete message to the .replay file". */
|
||||
if (rec_ctx != NULL) {
|
||||
sched_record_session_end(sched, rec_ctx, xce_core_now(&(sched->core)));
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -1,18 +1,33 @@
|
|||
#include <errno.h>
|
||||
#include <fcntl.h>
|
||||
#include <limits.h>
|
||||
#include <stdio.h>
|
||||
#include <stdlib.h>
|
||||
#include <string.h>
|
||||
#include <sys/stat.h>
|
||||
#include <time.h>
|
||||
#include <unistd.h>
|
||||
|
||||
#include "controller.capnp.h"
|
||||
#include "core.h"
|
||||
#include "recording.h"
|
||||
#include "utils.h"
|
||||
|
||||
/*
|
||||
* recording.c - scheduler-core spec, task 9.1. See recording.h for the
|
||||
* full contract/rationale. This is deliberately the ONLY function in this
|
||||
* file as of task 9.1 -- record start/stop/status (task 9.2, design.md's
|
||||
* NormalDriver) extend this file later; nothing here anticipates that
|
||||
* work's shape beyond what this task's own scope requires.
|
||||
* recording.c - scheduler-core spec, tasks 9.1/9.2. See recording.h for the
|
||||
* full per-function contract/rationale. Task 9.1 added ONLY
|
||||
* sched_validate_record_filename() below. Task 9.2 (this revision) adds
|
||||
* the record start/stop lifecycle (sched_record_start()/sched_record_stop())
|
||||
* plus the shared message-write/error-abort helpers
|
||||
* (write_capnp_message()/sched_record_abort_on_error()) two sibling files
|
||||
* -- recording_events.c (the per-event-batch recording-write hook) and
|
||||
* recording_session.c (the per-session ReplaySession accumulator) -- also
|
||||
* use. Split across three files (this project's 800-line-per-file
|
||||
* convention, CLAUDE.md/coding-style.md) the same way task 8.1's
|
||||
* snapshot.c/snapshot_wire.c/snapshot_load.c split: shared, narrow helpers
|
||||
* are non-`static` here and redeclared via a plain `extern` at each
|
||||
* consumer's own use site, matching snapshot_wire.c's own documented idiom
|
||||
* rather than a shared private header.
|
||||
*/
|
||||
|
||||
/*
|
||||
|
|
@ -169,3 +184,360 @@ int sched_validate_record_filename(const char* record_dir, const char* filename,
|
|||
|
||||
return 0;
|
||||
}
|
||||
|
||||
/* ========================================================================
|
||||
* Task 9.2: record start/stop lifecycle.
|
||||
* ======================================================================== */
|
||||
|
||||
/*
|
||||
* write_capnp_message - serializes `c`'s current root message and appends
|
||||
* it to `f` as one self-framed Cap'n Proto message (Req 5 AC 20: "the
|
||||
* .replay file is a single stream of consecutive capnp messages ... no
|
||||
* extra outer framing needed"). Shared by every message writer in this
|
||||
* file and in recording_events.c/recording_session.c (non-`static`; see
|
||||
* this file's own header comment for why). Returns 0 on success; -EIO if
|
||||
* the message is empty or the full byte count could not be written;
|
||||
* -ENOMEM if the staging buffer could not be allocated.
|
||||
*/
|
||||
int write_capnp_message(FILE* f, struct capn* c)
|
||||
{
|
||||
int64_t sz;
|
||||
uint8_t* buf;
|
||||
size_t written;
|
||||
|
||||
sz = capn_size(c);
|
||||
if (sz <= 0) {
|
||||
return -EIO;
|
||||
}
|
||||
|
||||
buf = (uint8_t*)malloc((size_t)sz);
|
||||
if (buf == NULL) {
|
||||
return -ENOMEM;
|
||||
}
|
||||
|
||||
capn_write_mem(c, buf, (size_t)sz, 0);
|
||||
|
||||
written = fwrite(buf, 1, (size_t)sz, f);
|
||||
free(buf);
|
||||
|
||||
if (written != (size_t)sz) {
|
||||
return -EIO;
|
||||
}
|
||||
|
||||
return 0;
|
||||
}
|
||||
|
||||
/*
|
||||
* write_replay_header - builds and writes the ReplayHeader message (Req 5
|
||||
* AC 20's msg 1). `configHash` is intentionally left unset (reads back as
|
||||
* empty on decode): no utility anywhere in this codebase computes a config
|
||||
* hash today (verified -- the only sha256 handling elsewhere, e.g.
|
||||
* request.c's raw_sha256, is for CALLER-supplied bytes on the config-push
|
||||
* wire path, not something computed FROM an xce_conf4sched_t); introducing
|
||||
* one is out of this task's boundary. This mirrors the same documented
|
||||
* "Data field codec is a no-op" limitation typemap.h already calls out for
|
||||
* this exact field.
|
||||
*/
|
||||
static int write_replay_header(FILE* f, xce_scheduler_t* sched,
|
||||
uint64_t start_ts)
|
||||
{
|
||||
struct capn c;
|
||||
struct capn_segment* cs;
|
||||
replay_header_t hdr;
|
||||
replay_cluster_t cluster;
|
||||
char** controllers = NULL;
|
||||
int ncontrollers = 0;
|
||||
char wall[32];
|
||||
time_t now_wall;
|
||||
struct tm tmv;
|
||||
ReplayHeader_ptr ptr;
|
||||
int rc;
|
||||
int i;
|
||||
|
||||
memset(&hdr, 0, sizeof(hdr));
|
||||
memset(&cluster, 0, sizeof(cluster));
|
||||
|
||||
now_wall = time(NULL);
|
||||
gmtime_r(&now_wall, &tmv);
|
||||
strftime(wall, sizeof(wall), "%Y-%m-%dT%H:%M:%SZ", &tmv);
|
||||
|
||||
hdr.recorder_version = "1.0";
|
||||
hdr.recording_start_ts = start_ts;
|
||||
hdr.recording_start_wall = wall;
|
||||
hdr.sched_policy = sched->core.job_pick_plugin; /* may be NULL */
|
||||
|
||||
ncontrollers = sched->cluster_cfg.candidates_count;
|
||||
if (ncontrollers > 0) {
|
||||
controllers = (char**)calloc((size_t)ncontrollers, sizeof(char*));
|
||||
if (controllers == NULL) {
|
||||
return -ENOMEM;
|
||||
}
|
||||
for (i = 0; i < ncontrollers; i++) {
|
||||
char hostbuf[XCE_MAXHOSTNAME_LEN + 16];
|
||||
|
||||
snprintf(hostbuf, sizeof(hostbuf), "%s:%d",
|
||||
sched->cluster_cfg.candidates[i].hostname,
|
||||
sched->cluster_cfg.candidates[i].port);
|
||||
controllers[i] = strdup(hostbuf);
|
||||
}
|
||||
cluster.name = sched->cluster_cfg.identity;
|
||||
cluster.controllers = controllers;
|
||||
cluster.n_controllers = ncontrollers;
|
||||
cluster.recording_host = sched->cluster_cfg.identity;
|
||||
hdr.cluster = &cluster;
|
||||
}
|
||||
/* else: single-host mode (cluster_cfg.candidates_count == 0,
|
||||
* scheduler.h's own field comment) -- hdr.cluster stays NULL;
|
||||
* encode_ReplayHeader_ptr's transitive encode_ReplayCluster_ptr(...,
|
||||
* NULL) writes a null pointer, which decodes back as an empty
|
||||
* ReplayCluster, never a crash (same pattern every other optional
|
||||
* nested-struct field in this schema already relies on). */
|
||||
|
||||
capn_init_malloc(&c);
|
||||
cs = capn_root(&c).seg;
|
||||
|
||||
encode_ReplayHeader_ptr(cs, &ptr, &hdr);
|
||||
capn_setp(capn_root(&c), 0, ptr.p);
|
||||
|
||||
rc = write_capnp_message(f, &c);
|
||||
capn_free(&c);
|
||||
|
||||
if (controllers != NULL) {
|
||||
for (i = 0; i < ncontrollers; i++) {
|
||||
free(controllers[i]);
|
||||
}
|
||||
free(controllers);
|
||||
}
|
||||
|
||||
return rc;
|
||||
}
|
||||
|
||||
/*
|
||||
* write_snapshot_entry - builds and writes the initial SnapshotEntry
|
||||
* message (Req 5 AC 20's msg 2, "MUST be the first data message after
|
||||
* ReplayHeader"). Takes a full core snapshot via
|
||||
* xce_sched_core_save_snapshot() (core.h, task 8.1) and writes it.
|
||||
*
|
||||
* `snap`'s contents are deliberately never freed here: snapshot.c's own
|
||||
* "Ownership notes" (file header comment) document that `config` and each
|
||||
* `resources[]` entry are BORROWED live-core references, not deep copies
|
||||
* -- freeing them would corrupt/use-after-free the live scheduler. The
|
||||
* remaining fields (jobs/queues/workers/users/projects) ARE independently
|
||||
* heap-allocated by save_snapshot(), but no free-snapshot function exists
|
||||
* anywhere in this codebase yet (task 8.1's own test suite leaves every
|
||||
* save_snapshot() result unfreed for the same reason, detect_leaks=0
|
||||
* across this whole suite) -- writing one is out of this task's boundary.
|
||||
*/
|
||||
static int write_snapshot_entry(FILE* f, xce_sched_core_t* core, uint64_t ts)
|
||||
{
|
||||
struct capn c;
|
||||
struct capn_segment* cs;
|
||||
sched_snapshot_data_t snap;
|
||||
replay_snapshot_entry_t entry;
|
||||
SnapshotEntry_ptr ptr;
|
||||
int rc;
|
||||
|
||||
memset(&snap, 0, sizeof(snap));
|
||||
rc = xce_sched_core_save_snapshot(core, &snap);
|
||||
if (rc < 0) {
|
||||
return rc;
|
||||
}
|
||||
|
||||
memset(&entry, 0, sizeof(entry));
|
||||
entry.ts = ts;
|
||||
entry.data = &snap;
|
||||
|
||||
capn_init_malloc(&c);
|
||||
cs = capn_root(&c).seg;
|
||||
|
||||
encode_SnapshotEntry_ptr(cs, &ptr, &entry);
|
||||
capn_setp(capn_root(&c), 0, ptr.p);
|
||||
|
||||
rc = write_capnp_message(f, &c);
|
||||
capn_free(&c);
|
||||
|
||||
return rc;
|
||||
}
|
||||
|
||||
int sched_record_start(xce_scheduler_t* sched, const char* record_file)
|
||||
{
|
||||
char resolved_path[PATH_MAX];
|
||||
char auto_name[64];
|
||||
const char* filename;
|
||||
int fd;
|
||||
struct stat st;
|
||||
FILE* f;
|
||||
uint64_t start_ts;
|
||||
int rc;
|
||||
|
||||
if (sched == NULL) {
|
||||
return -EINVAL;
|
||||
}
|
||||
if (sched->recording) {
|
||||
return -EALREADY;
|
||||
}
|
||||
|
||||
filename = record_file;
|
||||
if ((filename == NULL) || (filename[0] == '\0')) {
|
||||
char timebuf[32];
|
||||
time_t now_wall = time(NULL);
|
||||
struct tm tmv;
|
||||
|
||||
gmtime_r(&now_wall, &tmv);
|
||||
strftime(timebuf, sizeof(timebuf), "%Y%m%d-%H%M%S", &tmv);
|
||||
snprintf(auto_name, sizeof(auto_name), "scheduler-replay-%s.replay",
|
||||
timebuf);
|
||||
filename = auto_name;
|
||||
}
|
||||
|
||||
rc = sched_validate_record_filename(sched->record_dir, filename,
|
||||
resolved_path, sizeof(resolved_path));
|
||||
if (rc < 0) {
|
||||
return rc;
|
||||
}
|
||||
|
||||
/* TOCTOU hardening -- see recording.h's doc comment on
|
||||
* sched_validate_record_filename() and on sched_record_start() itself:
|
||||
* O_NOFOLLOW refuses a symlink swapped in after validation; the
|
||||
* fstat()+st_nlink==1 check below additionally refuses a hard link
|
||||
* swapped in (O_NOFOLLOW alone does not catch that -- no symlink is
|
||||
* involved in that race).
|
||||
*
|
||||
* Deliberately NO O_TRUNC here: O_TRUNC truncates atomically as PART
|
||||
* OF open() itself, before this function's own fstat()/st_nlink check
|
||||
* ever runs -- against a hard-linked victim file (same inode, no
|
||||
* symlink involved, so O_NOFOLLOW never triggers), that would corrupt
|
||||
* the victim the instant open() succeeds, regardless of what the
|
||||
* check below then decides. Instead: open non-destructively, verify
|
||||
* safety via fstat()+st_nlink, and only ftruncate() once verified --
|
||||
* so a rejected hard link is NEVER truncated. Matches
|
||||
* sched_validate_record_filename()'s own documented "legitimate
|
||||
* overwrite target" acceptance of a pre-existing single-link regular
|
||||
* file (which this ftruncate() then genuinely overwrites, same
|
||||
* end-visible behavior as O_TRUNC would have given for that safe
|
||||
* case). */
|
||||
fd = open(resolved_path, O_WRONLY | O_CREAT | O_NOFOLLOW, 0640);
|
||||
if (fd < 0) {
|
||||
return (errno > 0) ? -errno : -EIO;
|
||||
}
|
||||
|
||||
if (fstat(fd, &st) != 0) {
|
||||
rc = (errno > 0) ? -errno : -EIO;
|
||||
close(fd);
|
||||
return rc;
|
||||
}
|
||||
if (!S_ISREG(st.st_mode) || (st.st_nlink != 1)) {
|
||||
close(fd);
|
||||
return -EACCES;
|
||||
}
|
||||
|
||||
if (ftruncate(fd, 0) != 0) {
|
||||
rc = (errno > 0) ? -errno : -EIO;
|
||||
close(fd);
|
||||
return rc;
|
||||
}
|
||||
|
||||
f = fdopen(fd, "w");
|
||||
if (f == NULL) {
|
||||
rc = (errno > 0) ? -errno : -EIO;
|
||||
close(fd);
|
||||
return rc;
|
||||
}
|
||||
|
||||
start_ts = xce_core_now(&(sched->core));
|
||||
|
||||
rc = write_replay_header(f, sched, start_ts);
|
||||
if (rc == 0) {
|
||||
rc = write_snapshot_entry(f, &(sched->core), start_ts);
|
||||
}
|
||||
|
||||
if (rc < 0) {
|
||||
ERROR("%s failed to write replay header/snapshot to %s: rc=%d",
|
||||
__func__, resolved_path, rc);
|
||||
fclose(f);
|
||||
return rc;
|
||||
}
|
||||
|
||||
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;
|
||||
|
||||
return 0;
|
||||
}
|
||||
|
||||
int sched_record_stop(xce_scheduler_t* sched)
|
||||
{
|
||||
struct capn c;
|
||||
struct capn_segment* cs;
|
||||
replay_event_t ev;
|
||||
ReplayEvent_ptr ptr;
|
||||
int rc;
|
||||
|
||||
if (sched == NULL) {
|
||||
return -EINVAL;
|
||||
}
|
||||
if (!sched->recording || (sched->record_file == NULL)) {
|
||||
return -EALREADY;
|
||||
}
|
||||
|
||||
memset(&ev, 0, sizeof(ev));
|
||||
ev.ts = xce_core_now(&(sched->core));
|
||||
ev.body_which = ReplayEvent_body_endSession;
|
||||
|
||||
capn_init_malloc(&c);
|
||||
cs = capn_root(&c).seg;
|
||||
encode_ReplayEvent_ptr(cs, &ptr, &ev);
|
||||
capn_setp(capn_root(&c), 0, ptr.p);
|
||||
|
||||
rc = write_capnp_message(sched->record_file, &c);
|
||||
capn_free(&c);
|
||||
|
||||
if (rc < 0) {
|
||||
ERROR("%s failed to write end-of-recording marker to %s: rc=%d",
|
||||
__func__,
|
||||
(sched->record_file_path != NULL) ? sched->record_file_path : "?",
|
||||
rc);
|
||||
}
|
||||
|
||||
/* Clear recording state unconditionally -- Req 5 AC 10's "Error
|
||||
* handling" clause: a failed write disables recording rather than
|
||||
* leaving it half-active. */
|
||||
fflush(sched->record_file);
|
||||
fclose(sched->record_file);
|
||||
sched->record_file = NULL;
|
||||
sched->recording = false;
|
||||
free(sched->record_file_path);
|
||||
sched->record_file_path = NULL;
|
||||
|
||||
return rc;
|
||||
}
|
||||
|
||||
/*
|
||||
* sched_record_abort_on_error - shared "a recording write failed" cleanup
|
||||
* (Req 5 AC 10's "Error handling" clause), used by both
|
||||
* recording_events.c's sched_record_write_events() and
|
||||
* recording_session.c's sched_record_session_end() (non-`static`; see this
|
||||
* file's own header comment for why). Mirrors sched_record_stop()'s
|
||||
* cleanup, minus the end-of-recording marker write (the file is presumably
|
||||
* already broken -- disk full, permission revoked mid-write, etc. -- so
|
||||
* attempting one more write is not productive).
|
||||
*/
|
||||
void sched_record_abort_on_error(xce_scheduler_t* sched, int rc)
|
||||
{
|
||||
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;
|
||||
}
|
||||
free(sched->record_file_path);
|
||||
sched->record_file_path = NULL;
|
||||
sched->recording = false;
|
||||
}
|
||||
|
|
|
|||
177
src/scheduler/src/recording_events.c
Normal file
177
src/scheduler/src/recording_events.c
Normal file
|
|
@ -0,0 +1,177 @@
|
|||
#include <errno.h>
|
||||
#include <stdio.h>
|
||||
#include <string.h>
|
||||
|
||||
#include "controller.capnp.h"
|
||||
#include "core.h"
|
||||
#include "recording.h"
|
||||
#include "utils.h"
|
||||
|
||||
/*
|
||||
* recording_events.c - scheduler-core spec, task 9.2 (Req 5 AC 10, "On each
|
||||
* core_push_events() call ... For each event in the batch, serialize and
|
||||
* write the corresponding ReplayEvent message"). Split out of recording.c
|
||||
* (this project's 800-line-per-file convention, CLAUDE.md/coding-style.md)
|
||||
* -- see recording.c's own file header comment, "Split across three
|
||||
* files", for the full rationale. This file owns the per-event-batch
|
||||
* recording-write hook driver.c's xce_sched_driver_translate_pending_events()
|
||||
* calls into.
|
||||
*
|
||||
* write_capnp_message()/sched_record_abort_on_error() are defined
|
||||
* (non-`static`) in recording.c and redeclared here via a plain `extern`,
|
||||
* matching this codebase's existing cross-.c-file idiom for a narrow
|
||||
* shared dependency (snapshot_wire.c's own documented precedent, itself
|
||||
* citing core_session.c's `extern void sched_worker_release_host_ctx(...)`).
|
||||
*/
|
||||
extern int write_capnp_message(FILE* f, struct capn* c);
|
||||
extern void sched_record_abort_on_error(xce_scheduler_t* sched, int rc);
|
||||
|
||||
/*
|
||||
* write_one_replay_event - builds and writes one ReplayEvent message from
|
||||
* a single translated `xce_core_event_t` (the SAME op-type table
|
||||
* driver.c's translate_one_pending_update() already established -- see
|
||||
* this function's per-case mapping below). Every wrapper local is declared
|
||||
* unconditionally (cheap stack space) so its address stays valid across
|
||||
* the single encode+write call after the switch, regardless of which case
|
||||
* populated it.
|
||||
*/
|
||||
static int write_one_replay_event(FILE* f, xce_core_event_t* ev,
|
||||
uint64_t* event_count)
|
||||
{
|
||||
struct capn c;
|
||||
struct capn_segment* cs;
|
||||
replay_event_t wire;
|
||||
ReplayEvent_ptr ptr;
|
||||
int rc;
|
||||
|
||||
xce_jobs_data_t jobs_wrap;
|
||||
xce_workers_data_t workers_wrap;
|
||||
xce_resource_data_t resources_wrap;
|
||||
xce_resval_list_t resvals_wrap;
|
||||
job_state_change_event_t jsc_wrap;
|
||||
job_update_event_t ju_wrap;
|
||||
|
||||
memset(&wire, 0, sizeof(wire));
|
||||
memset(&jobs_wrap, 0, sizeof(jobs_wrap));
|
||||
memset(&workers_wrap, 0, sizeof(workers_wrap));
|
||||
memset(&resources_wrap, 0, sizeof(resources_wrap));
|
||||
memset(&resvals_wrap, 0, sizeof(resvals_wrap));
|
||||
memset(&jsc_wrap, 0, sizeof(jsc_wrap));
|
||||
memset(&ju_wrap, 0, sizeof(ju_wrap));
|
||||
|
||||
wire.ts = ev->ts;
|
||||
|
||||
switch (ev->type) {
|
||||
case XCE_CORE_EVENT_CONFIG_PUSH:
|
||||
wire.body_which = ReplayEvent_body_configPush;
|
||||
wire.body.config_push = ev->u.config_push.config;
|
||||
break;
|
||||
|
||||
case XCE_CORE_EVENT_JOBS_PUSH:
|
||||
jobs_wrap.n_total = ev->u.jobs_push.njobs;
|
||||
jobs_wrap.count = ev->u.jobs_push.njobs;
|
||||
jobs_wrap.jobs = ev->u.jobs_push.jobs;
|
||||
wire.body_which = ReplayEvent_body_jobsPush;
|
||||
wire.body.jobs_push = &jobs_wrap;
|
||||
break;
|
||||
|
||||
case XCE_CORE_EVENT_WORKERS_PUSH:
|
||||
workers_wrap.n_total = ev->u.workers_push.nworkers;
|
||||
workers_wrap.count = ev->u.workers_push.nworkers;
|
||||
workers_wrap.workers = ev->u.workers_push.workers;
|
||||
wire.body_which = ReplayEvent_body_workersPush;
|
||||
wire.body.workers_push = &workers_wrap;
|
||||
break;
|
||||
|
||||
case XCE_CORE_EVENT_RESOURCES_PUSH:
|
||||
resources_wrap.n_total = ev->u.resources_push.n;
|
||||
resources_wrap.count = ev->u.resources_push.n;
|
||||
resources_wrap.resources = ev->u.resources_push.resources;
|
||||
wire.body_which = ReplayEvent_body_resourcesPush;
|
||||
wire.body.resources_push = &resources_wrap;
|
||||
break;
|
||||
|
||||
case XCE_CORE_EVENT_RESVALS_FULL:
|
||||
resvals_wrap.count = ev->u.resvals_full.n;
|
||||
resvals_wrap.vals = ev->u.resvals_full.resvals;
|
||||
wire.body_which = ReplayEvent_body_resvalsFull;
|
||||
wire.body.resvals_full = &resvals_wrap;
|
||||
break;
|
||||
|
||||
case XCE_CORE_EVENT_RESVALS_DELTA:
|
||||
wire.body_which = ReplayEvent_body_resvalsDelta;
|
||||
wire.body.resvals_delta = ev->u.resvals_delta.deltas;
|
||||
break;
|
||||
|
||||
case XCE_CORE_EVENT_JOB_STATE_CHANGE:
|
||||
jsc_wrap.jobid = ev->u.job_state_change.jobid;
|
||||
jsc_wrap.old_state = ev->u.job_state_change.old_state;
|
||||
jsc_wrap.new_state = ev->u.job_state_change.new_state;
|
||||
jsc_wrap.exit_code = ev->u.job_state_change.exit_code;
|
||||
wire.body_which = ReplayEvent_body_jobStateChange;
|
||||
wire.body.job_state_change = &jsc_wrap;
|
||||
break;
|
||||
|
||||
case XCE_CORE_EVENT_JOB_UPDATE:
|
||||
ju_wrap.jobid = ev->u.job_update.jobid;
|
||||
ju_wrap.changed_fields = ev->u.job_update.changed_fields;
|
||||
ju_wrap.job = ev->u.job_update.job;
|
||||
wire.body_which = ReplayEvent_body_jobUpdate;
|
||||
wire.body.job_update = &ju_wrap;
|
||||
break;
|
||||
|
||||
case XCE_CORE_EVENT_QUEUE_UPDATE:
|
||||
wire.body_which = ReplayEvent_body_queueUpdate;
|
||||
wire.body.queue_update = ev->u.queue_update.queue;
|
||||
break;
|
||||
|
||||
case XCE_CORE_EVENT_JOB_RUSAGE:
|
||||
wire.body_which = ReplayEvent_body_jobRusage;
|
||||
wire.body.job_rusage = ev->u.job_rusage.rusages;
|
||||
break;
|
||||
|
||||
default:
|
||||
WARN("%s unrecognized core event type %d, skipping recording write",
|
||||
__func__, (int)ev->type);
|
||||
return 0;
|
||||
}
|
||||
|
||||
capn_init_malloc(&c);
|
||||
cs = capn_root(&c).seg;
|
||||
|
||||
encode_ReplayEvent_ptr(cs, &ptr, &wire);
|
||||
capn_setp(capn_root(&c), 0, ptr.p);
|
||||
|
||||
rc = write_capnp_message(f, &c);
|
||||
capn_free(&c);
|
||||
|
||||
if (rc == 0) {
|
||||
(*event_count)++;
|
||||
}
|
||||
|
||||
return rc;
|
||||
}
|
||||
|
||||
int sched_record_write_events(xce_scheduler_t* sched, xce_core_event_t* events,
|
||||
int n)
|
||||
{
|
||||
int i;
|
||||
|
||||
if ((sched == NULL) || (events == NULL) || (n <= 0)) {
|
||||
return 0;
|
||||
}
|
||||
if (!sched->recording || (sched->record_file == NULL)) {
|
||||
return 0;
|
||||
}
|
||||
|
||||
for (i = 0; i < n; i++) {
|
||||
int rc = write_one_replay_event(sched->record_file, &events[i],
|
||||
&(sched->record_event_count));
|
||||
if (rc < 0) {
|
||||
sched_record_abort_on_error(sched, rc);
|
||||
return rc;
|
||||
}
|
||||
}
|
||||
|
||||
return 0;
|
||||
}
|
||||
244
src/scheduler/src/recording_session.c
Normal file
244
src/scheduler/src/recording_session.c
Normal file
|
|
@ -0,0 +1,244 @@
|
|||
#include <stdint.h>
|
||||
#include <stdio.h>
|
||||
#include <stdlib.h>
|
||||
#include <string.h>
|
||||
|
||||
#include "controller.capnp.h"
|
||||
#include "core.h"
|
||||
#include "decision.h"
|
||||
#include "recording.h"
|
||||
#include "utils.h"
|
||||
|
||||
/*
|
||||
* recording_session.c - scheduler-core spec, task 9.2 (Req 5 AC 10, "On
|
||||
* core_go(): begin building a ReplaySession message" / "In the decision
|
||||
* drain loop: append a ReplayDecision" / "On core_end_session(): finalize
|
||||
* ... write the complete message"). Split out of recording.c (this
|
||||
* project's 800-line-per-file convention, CLAUDE.md/coding-style.md) --
|
||||
* see recording.c's own file header comment, "Split across three files",
|
||||
* for the full rationale. This file owns the per-session ReplaySession
|
||||
* accumulator (sched_record_session_begin()/_add_decision()/_end())
|
||||
* driver.c's xce_sched_driver_run_normal_mode_round() calls into.
|
||||
*
|
||||
* write_capnp_message()/sched_record_abort_on_error() are defined
|
||||
* (non-`static`) in recording.c and redeclared here via a plain `extern`,
|
||||
* matching this codebase's existing cross-.c-file idiom for a narrow
|
||||
* shared dependency (snapshot_wire.c's own documented precedent, itself
|
||||
* citing core_session.c's `extern void sched_worker_release_host_ctx(...)`).
|
||||
*/
|
||||
extern int write_capnp_message(FILE* f, struct capn* c);
|
||||
extern void sched_record_abort_on_error(xce_scheduler_t* sched, int rc);
|
||||
|
||||
/*
|
||||
* sched_record_session_ctx_s - see recording.h's forward-declared typedef
|
||||
* for the public (opaque) contract.
|
||||
*/
|
||||
struct sched_record_session_ctx_s {
|
||||
uint64_t begin_round;
|
||||
replay_decision_t** decisions;
|
||||
int n;
|
||||
int cap;
|
||||
uint32_t n_dispatched;
|
||||
uint32_t n_pended;
|
||||
};
|
||||
|
||||
sched_record_session_ctx_t* sched_record_session_begin(xce_scheduler_t* sched,
|
||||
uint64_t begin_round_ts)
|
||||
{
|
||||
sched_record_session_ctx_t* ctx;
|
||||
|
||||
if ((sched == NULL) || !sched->recording || (sched->record_file == NULL)) {
|
||||
return NULL;
|
||||
}
|
||||
|
||||
ctx = (sched_record_session_ctx_t*)calloc(1, sizeof(*ctx));
|
||||
if (ctx == NULL) {
|
||||
return NULL;
|
||||
}
|
||||
|
||||
ctx->begin_round = begin_round_ts;
|
||||
|
||||
return ctx;
|
||||
}
|
||||
|
||||
/*
|
||||
* free_record_decision - frees one heap-allocated replay_decision_t built
|
||||
* by sched_record_session_add_decision() below, including the small
|
||||
* wrapper body struct it owns (dispatch/pend only -- preempt is never
|
||||
* appended, see that function's own comment). Never frees the underlying
|
||||
* xce_job_alloc_t / xce_job_preason_t payloads themselves -- those are
|
||||
* borrowed references into the live decision the driver drained, owned
|
||||
* elsewhere.
|
||||
*/
|
||||
static void free_record_decision(replay_decision_t* rd)
|
||||
{
|
||||
if (rd == NULL) {
|
||||
return;
|
||||
}
|
||||
|
||||
switch (rd->body_which) {
|
||||
case ReplayDecision_body_dispatch:
|
||||
free(rd->body.dispatch);
|
||||
break;
|
||||
case ReplayDecision_body_pend:
|
||||
free(rd->body.pend);
|
||||
break;
|
||||
default:
|
||||
break;
|
||||
}
|
||||
|
||||
free(rd);
|
||||
}
|
||||
|
||||
void sched_record_session_add_decision(sched_record_session_ctx_t* ctx,
|
||||
uint64_t ts, xce_decision_t* decision)
|
||||
{
|
||||
replay_decision_t* rd;
|
||||
|
||||
if ((ctx == NULL) || (decision == NULL)) {
|
||||
return;
|
||||
}
|
||||
|
||||
rd = (replay_decision_t*)calloc(1, sizeof(*rd));
|
||||
if (rd == NULL) {
|
||||
return;
|
||||
}
|
||||
rd->ts = ts;
|
||||
|
||||
switch (decision->op) {
|
||||
case XCE_DECISION_DISPATCH: {
|
||||
sched_dispatch_decision_t* dd
|
||||
= (sched_dispatch_decision_t*)calloc(1, sizeof(*dd));
|
||||
|
||||
if (dd == NULL) {
|
||||
free(rd);
|
||||
return;
|
||||
}
|
||||
dd->alloc = decision->u.alloc;
|
||||
rd->body_which = ReplayDecision_body_dispatch;
|
||||
rd->body.dispatch = dd;
|
||||
ctx->n_dispatched++;
|
||||
break;
|
||||
}
|
||||
|
||||
case XCE_DECISION_PEND: {
|
||||
xce_pendreason_t* pr = decision->u.reason;
|
||||
sched_pend_decision_t* pd;
|
||||
|
||||
if (pr == NULL) {
|
||||
free(rd);
|
||||
return;
|
||||
}
|
||||
pd = (sched_pend_decision_t*)calloc(1, sizeof(*pd));
|
||||
if (pd == NULL) {
|
||||
free(rd);
|
||||
return;
|
||||
}
|
||||
pd->jobid = pr->jobid;
|
||||
pd->reason = pr->reasons;
|
||||
rd->body_which = ReplayDecision_body_pend;
|
||||
rd->body.pend = pd;
|
||||
ctx->n_pended++;
|
||||
break;
|
||||
}
|
||||
|
||||
case XCE_DECISION_PREEMPT:
|
||||
/* No live allocator path produces this decision type -- see
|
||||
* driver.c's xce_sched_driver_on_decision() header comment for the
|
||||
* full "documented unreachable, not silently worked around"
|
||||
* rationale, which this recording path mirrors: log loud, write
|
||||
* nothing to the recording for this decision, never fabricate wire
|
||||
* data for a shape this codebase cannot yet produce for real. */
|
||||
ERROR("%s PREEMPT decision recording not yet implemented "
|
||||
"(scheduler-preemption spec territory) -- jobid=%llu dropped "
|
||||
"from replay file",
|
||||
__func__,
|
||||
(unsigned long long)((decision->u.preempt != NULL)
|
||||
? decision->u.preempt->jobid
|
||||
: 0));
|
||||
free(rd);
|
||||
return;
|
||||
|
||||
default:
|
||||
/* XCE_DECISION_NONE / XCE_DECISION_NEW_SESSION: nothing to
|
||||
* record. */
|
||||
free(rd);
|
||||
return;
|
||||
}
|
||||
|
||||
if (ctx->n == ctx->cap) {
|
||||
int newcap = ctx->cap ? (ctx->cap * 2) : 8;
|
||||
replay_decision_t** narr = (replay_decision_t**)realloc(
|
||||
ctx->decisions, (size_t)newcap * sizeof(*narr));
|
||||
|
||||
if (narr == NULL) {
|
||||
ERROR("%s out of memory growing decision batch (n=%d)", __func__,
|
||||
ctx->n);
|
||||
free_record_decision(rd);
|
||||
return;
|
||||
}
|
||||
ctx->decisions = narr;
|
||||
ctx->cap = newcap;
|
||||
}
|
||||
|
||||
ctx->decisions[ctx->n++] = rd;
|
||||
}
|
||||
|
||||
int sched_record_session_end(xce_scheduler_t* sched,
|
||||
sched_record_session_ctx_t* ctx, uint64_t end_ts)
|
||||
{
|
||||
struct capn c;
|
||||
struct capn_segment* cs;
|
||||
replay_session_t wire;
|
||||
session_summary_t summary;
|
||||
ReplaySession_ptr ptr;
|
||||
int rc;
|
||||
int i;
|
||||
|
||||
if (ctx == NULL) {
|
||||
return 0;
|
||||
}
|
||||
|
||||
if ((sched == NULL) || !sched->recording || (sched->record_file == NULL)) {
|
||||
for (i = 0; i < ctx->n; i++) {
|
||||
free_record_decision(ctx->decisions[i]);
|
||||
}
|
||||
free(ctx->decisions);
|
||||
free(ctx);
|
||||
return 0;
|
||||
}
|
||||
|
||||
memset(&summary, 0, sizeof(summary));
|
||||
summary.duration_ns
|
||||
= (end_ts >= ctx->begin_round) ? (end_ts - ctx->begin_round) : 0;
|
||||
summary.n_decisions = (uint32_t)ctx->n;
|
||||
summary.n_dispatched = ctx->n_dispatched;
|
||||
summary.n_pended = ctx->n_pended;
|
||||
|
||||
memset(&wire, 0, sizeof(wire));
|
||||
wire.begin_round = ctx->begin_round;
|
||||
wire.decisions = ctx->decisions;
|
||||
wire.n_decisions = ctx->n;
|
||||
wire.summary = &summary;
|
||||
|
||||
capn_init_malloc(&c);
|
||||
cs = capn_root(&c).seg;
|
||||
|
||||
encode_ReplaySession_ptr(cs, &ptr, &wire);
|
||||
capn_setp(capn_root(&c), 0, ptr.p);
|
||||
|
||||
rc = write_capnp_message(sched->record_file, &c);
|
||||
capn_free(&c);
|
||||
|
||||
if (rc < 0) {
|
||||
sched_record_abort_on_error(sched, rc);
|
||||
}
|
||||
|
||||
for (i = 0; i < ctx->n; i++) {
|
||||
free_record_decision(ctx->decisions[i]);
|
||||
}
|
||||
free(ctx->decisions);
|
||||
free(ctx);
|
||||
|
||||
return rc;
|
||||
}
|
||||
|
|
@ -412,6 +412,28 @@ sched_recording_path_validation_test = executable(
|
|||
dependencies: [scheduler_test_deps],
|
||||
install: false)
|
||||
|
||||
# Contract test for the record start/stop lifecycle and the driver-loop
|
||||
# recording-write hooks (scheduler-core spec, task 9.2, Req 5 AC 10): start
|
||||
# writes a decodable ReplayHeader then SnapshotEntry; a full start -> submit
|
||||
# event batch -> complete one real session (real picker + stand-in
|
||||
# allocator threads, same rendezvous design as
|
||||
# sched_driver_session_ordering_test) -> stop cycle produces exactly the
|
||||
# expected message stream ending in ReplayEvent{endSession}; recording
|
||||
# inactive costs zero file I/O; double-start/stop-when-not-recording fail
|
||||
# cleanly; a TOCTOU-style symlink/hard-link swap between path validation
|
||||
# and the eventual open() is rejected. Same white-box #include pattern as
|
||||
# sched_driver_session_ordering_test (core.c/core_clock.c/core_events.c/
|
||||
# core_session.c/driver.c/plugin.c) PLUS ../src/recording.c/
|
||||
# recording_events.c/recording_session.c (the code under test) PLUS
|
||||
# ../src/snapshot.c/snapshot_wire.c (task 8.1's real
|
||||
# xce_sched_core_save_snapshot(), needed by sched_record_start()).
|
||||
sched_record_lifecycle_test = executable(
|
||||
'sched_record_lifecycle_test',
|
||||
['test_record_lifecycle.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',
|
||||
|
|
@ -475,3 +497,6 @@ test('scheduler snapshot save load test',
|
|||
test('scheduler recording path validation test',
|
||||
sched_recording_path_validation_test, suite: 'scheduler',
|
||||
env: ['ASAN_OPTIONS=detect_leaks=0'])
|
||||
test('scheduler record lifecycle test',
|
||||
sched_record_lifecycle_test, suite: 'scheduler',
|
||||
env: ['ASAN_OPTIONS=detect_leaks=0'])
|
||||
|
|
|
|||
|
|
@ -115,6 +115,50 @@ int sched_account_jobs(xce_scheduler_t* sched)
|
|||
return 0;
|
||||
}
|
||||
|
||||
/* Link-satisfying stubs for recording.c/recording_events.c/
|
||||
* recording_session.c's real functions -- as of scheduler-core spec task
|
||||
* 9.2, driver.c's (real, #included above)
|
||||
* xce_sched_driver_translate_pending_events()/
|
||||
* xce_sched_driver_run_normal_mode_round() reference these. Neither is
|
||||
* exercised by this file's tests (out of THIS file's scope -- see
|
||||
* test_driver_translate.c/test_driver_session_ordering.c), and this file
|
||||
* never links recording.c's real implementations (they need
|
||||
* xce_sched_core_save_snapshot(), snapshot.c). Never actually invoked at
|
||||
* runtime by any test in this file. */
|
||||
int sched_record_write_events(xce_scheduler_t* sched, xce_core_event_t* events,
|
||||
int n)
|
||||
{
|
||||
(void)sched;
|
||||
(void)events;
|
||||
(void)n;
|
||||
return 0;
|
||||
}
|
||||
|
||||
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 (this file never links
|
||||
* schedobj.c/scheduler.c). Never actually exercised (no test here pushes
|
||||
|
|
|
|||
|
|
@ -201,6 +201,59 @@ int sched_fill_limit_requests(xce_scheduler_t* sched, xvec_idval_t* vec,
|
|||
return 0;
|
||||
}
|
||||
|
||||
/* Link-satisfying stubs for recording.c/recording_events.c/
|
||||
* recording_session.c's real functions -- as of scheduler-core spec task
|
||||
* 9.2, driver.c's (real, #included above)
|
||||
* xce_sched_driver_translate_pending_events()/
|
||||
* xce_sched_driver_run_normal_mode_round() reference these
|
||||
* unconditionally (the `sched.recording`/`record_ctx != NULL` gates that
|
||||
* decide whether they do real work are runtime checks INSIDE these
|
||||
* functions, not something that removes the call site itself -- the
|
||||
* linker needs the symbol regardless). This file's fixture never sets
|
||||
* `sched.recording` (memset(&sched, 0, ...) below leaves it false), so at
|
||||
* runtime xce_sched_driver_translate_pending_events() calls this stub with
|
||||
* a non-NULL `sched` (record_ctx == &sched, always non-NULL here) but
|
||||
* xce_sched_driver_run_normal_mode_round() never calls
|
||||
* sched_record_session_begin() with `sched.recording == true` -- either
|
||||
* way, these stubs never need to do real work for this file's ordering
|
||||
* proof, which is not about recording. Never links recording.c's real
|
||||
* implementations (they need xce_sched_core_save_snapshot(), snapshot.c,
|
||||
* an even bigger dependency surface this file's own fixture doesn't
|
||||
* otherwise need). */
|
||||
int sched_record_write_events(xce_scheduler_t* sched, xce_core_event_t* events,
|
||||
int n)
|
||||
{
|
||||
(void)sched;
|
||||
(void)events;
|
||||
(void)n;
|
||||
return 0;
|
||||
}
|
||||
|
||||
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 -- see
|
||||
* file header comment (same rationale/algorithm as test_core_full_cycle.c's
|
||||
* own copy). */
|
||||
|
|
|
|||
|
|
@ -90,6 +90,53 @@ int sched_publish_pend_reason(xce_scheduler_t* sched, xce_preason_set_t* r)
|
|||
return 0;
|
||||
}
|
||||
|
||||
/* Link-satisfying stubs for recording.c/recording_events.c/
|
||||
* recording_session.c's real functions -- as of scheduler-core spec task
|
||||
* 9.2, driver.c's (real, #included above)
|
||||
* xce_sched_driver_translate_pending_events() calls sched_record_write_events()
|
||||
* unconditionally (behind a NULL `record_ctx` check, not a call this test ever
|
||||
* takes since every call site below passes NULL), and
|
||||
* xce_sched_driver_run_normal_mode_round() (out of THIS file's scope, same as
|
||||
* the core_session.c/plugin.c/ scheduler.c stubs above) calls the
|
||||
* sched_record_session_* trio. This file never links recording.c's real
|
||||
* implementations (they need xce_sched_core_save_snapshot(), snapshot.c, an
|
||||
* even bigger dependency surface this file's own header comment already avoids
|
||||
* pulling in for sched_publish_dispatch()/sched_publish_pend_reason()). Never
|
||||
* actually invoked at runtime by any test in this file. */
|
||||
int sched_record_write_events(xce_scheduler_t* sched, xce_core_event_t* events,
|
||||
int n)
|
||||
{
|
||||
(void)sched;
|
||||
(void)events;
|
||||
(void)n;
|
||||
return 0;
|
||||
}
|
||||
|
||||
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;
|
||||
}
|
||||
|
||||
/* Link-satisfying stubs: as of scheduler-core spec task 5.2,
|
||||
* xce_sched_driver_run_normal_mode_round() (driver.c, real, #included
|
||||
* above, alongside xce_sched_driver_translate_pending_events() this file
|
||||
|
|
@ -696,7 +743,7 @@ static void test_drain_translates_and_pushes_all_ten_op_types(void)
|
|||
r10->u.config = config;
|
||||
enqueue(&core, r10);
|
||||
|
||||
rc = xce_sched_driver_translate_pending_events(&core);
|
||||
rc = xce_sched_driver_translate_pending_events(&core, NULL);
|
||||
|
||||
TEST_ASSERT_EQUAL_INT(10, rc);
|
||||
|
||||
|
|
@ -746,7 +793,7 @@ static void test_accumulates_between_drains_and_empty_drain_is_noop(void)
|
|||
fixture_init(&core);
|
||||
|
||||
/* Nothing pending yet. */
|
||||
rc = xce_sched_driver_translate_pending_events(&core);
|
||||
rc = xce_sched_driver_translate_pending_events(&core, NULL);
|
||||
TEST_ASSERT_EQUAL_INT(0, rc);
|
||||
|
||||
/* One record enqueued -- drained on the next call. Heap-allocated (see
|
||||
|
|
@ -764,13 +811,13 @@ static void test_accumulates_between_drains_and_empty_drain_is_noop(void)
|
|||
req->u.workerupdate.workers = workers;
|
||||
enqueue(&core, req);
|
||||
|
||||
rc = xce_sched_driver_translate_pending_events(&core);
|
||||
rc = xce_sched_driver_translate_pending_events(&core, NULL);
|
||||
TEST_ASSERT_EQUAL_INT(1, rc);
|
||||
TEST_ASSERT_NOT_NULL(sched_get_worker(&core, "hostB"));
|
||||
}
|
||||
|
||||
/* Nothing new pending -- must not re-process anything. */
|
||||
rc = xce_sched_driver_translate_pending_events(&core);
|
||||
rc = xce_sched_driver_translate_pending_events(&core, NULL);
|
||||
TEST_ASSERT_EQUAL_INT(0, rc);
|
||||
}
|
||||
|
||||
|
|
@ -780,7 +827,7 @@ static void test_accumulates_between_drains_and_empty_drain_is_noop(void)
|
|||
|
||||
static void test_null_core_returns_einval(void)
|
||||
{
|
||||
int rc = xce_sched_driver_translate_pending_events(NULL);
|
||||
int rc = xce_sched_driver_translate_pending_events(NULL, NULL);
|
||||
TEST_ASSERT_EQUAL_INT(-EINVAL, rc);
|
||||
}
|
||||
|
||||
|
|
@ -790,7 +837,7 @@ static void test_uninitialized_pend_updates_returns_einval(void)
|
|||
int rc;
|
||||
|
||||
memset(&core, 0, sizeof(core));
|
||||
rc = xce_sched_driver_translate_pending_events(&core);
|
||||
rc = xce_sched_driver_translate_pending_events(&core, NULL);
|
||||
|
||||
TEST_ASSERT_EQUAL_INT(-EINVAL, rc);
|
||||
}
|
||||
|
|
|
|||
826
src/scheduler/testsuite/test_record_lifecycle.c
Normal file
826
src/scheduler/testsuite/test_record_lifecycle.c
Normal file
|
|
@ -0,0 +1,826 @@
|
|||
/*
|
||||
* test_record_lifecycle.c - contract test for the record start/stop
|
||||
* lifecycle and the driver-loop recording-write hooks (scheduler-core
|
||||
* spec, task 9.2, Req 5 AC 10).
|
||||
*
|
||||
* Observable under test (this task's own text):
|
||||
* - start -> the file begins with a decodable ReplayHeader then a
|
||||
* decodable SnapshotEntry.
|
||||
* - start -> submit an event batch -> complete one session -> stop
|
||||
* produces a file with events and exactly one ReplaySession in
|
||||
* between, ending in a ReplayEvent{endSession} marker.
|
||||
* - when `sched->recording` is false throughout, no file I/O occurs.
|
||||
* - double-start fails cleanly; stop-when-not-recording fails cleanly.
|
||||
* - a TOCTOU-style symlink/hardlink swap between path validation and the
|
||||
* eventual open() is rejected, not silently followed.
|
||||
*
|
||||
* White-box #include of ../src/core.c, ../src/core_clock.c,
|
||||
* ../src/core_events.c, ../src/core_session.c, ../src/driver.c,
|
||||
* ../src/plugin.c (same file set/rationale as
|
||||
* test_driver_session_ordering.c -- a real picker pthread + stand-in
|
||||
* allocator thread is needed to drive xce_sched_driver_run_normal_mode_round()
|
||||
* through one genuine session, jbuf's hardcoded capacity-1 rendezvous
|
||||
* requires it) PLUS ../src/recording.c, ../src/recording_events.c,
|
||||
* ../src/recording_session.c (the code under test) PLUS ../src/snapshot.c,
|
||||
* ../src/snapshot_wire.c (task 8.1's REAL xce_sched_core_save_snapshot(),
|
||||
* needed by sched_record_start() -- NOT snapshot_load.c, which nothing
|
||||
* here ever calls). No jobs are ever pushed (same "picker produces zero
|
||||
* decisions" fixture shape as test_driver_session_ordering.c), so
|
||||
* decision.c is not needed either.
|
||||
*
|
||||
* Decoding: this file reads its own produced `.replay` files back with the
|
||||
* SAME generated read_X()/decode_X() functions (controller.capnp.h) group
|
||||
* 7's own round-trip tests (test_e2e_capnp.c) use, via capn_init_fp()
|
||||
* (sequential per-message stream reads, since a `.replay` file is
|
||||
* consecutive self-framed capnp messages with no outer framing, Req 5
|
||||
* AC 20) -- never by hand-parsing bytes.
|
||||
*
|
||||
* Stubs below are the exact same set test_driver_session_ordering.c
|
||||
* already establishes for this core.c/core_clock.c/core_events.c/
|
||||
* core_session.c/driver.c/plugin.c combination (kill_sched_*,
|
||||
* sched_worker_release_host_ctx, sched_publish_dispatch/
|
||||
* sched_publish_pend_reason, sched_fill_limit_requests,
|
||||
* sched_resolve_job_picker's real reimplementation) -- see that file's own
|
||||
* header comment for the full rationale. plugin_unregister is NOT stubbed
|
||||
* (real, from the genuinely-linked plugin.c). No stub is needed for
|
||||
* xce_core_now()/xce_sched_core_save_snapshot() here (unlike
|
||||
* test_recording_path_validation.c's narrower TU): core_clock.c and
|
||||
* snapshot.c/snapshot_wire.c are genuinely linked into this file.
|
||||
*/
|
||||
#include <errno.h>
|
||||
#include <fcntl.h>
|
||||
#include <limits.h>
|
||||
#include <pthread.h>
|
||||
#include <stdint.h>
|
||||
#include <stdio.h>
|
||||
#include <stdlib.h>
|
||||
#include <string.h>
|
||||
#include <sys/stat.h>
|
||||
#include <unistd.h>
|
||||
|
||||
#include <unity/unity.h>
|
||||
|
||||
#include "controller.capnp.h"
|
||||
#include "kvmgr.h"
|
||||
#include "preason.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_session_ordering.c). */
|
||||
#include "../src/core.c"
|
||||
#include "../src/core_clock.c"
|
||||
#include "../src/core_events.c"
|
||||
#include "../src/core_session.c"
|
||||
#include "../src/driver.c"
|
||||
#include "../src/plugin.c"
|
||||
#include "../src/recording.c"
|
||||
#include "../src/recording_events.c"
|
||||
#include "../src/recording_session.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; }
|
||||
|
||||
/* core_reset_workers()'s (core_session.c) linker dependency -- never
|
||||
* invoked (no worker is ever added here). */
|
||||
void sched_worker_release_host_ctx(xce_sched_worker_t* w) { (void)w; }
|
||||
|
||||
/* Link-satisfying stubs for scheduler.c's sched_publish_dispatch()/
|
||||
* sched_publish_pend_reason() -- never actually invoked (this fixture
|
||||
* pushes no jobs, so the driver's decision-drain loop never observes a
|
||||
* non-NULL decision). */
|
||||
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 stub for resource.c's sched_fill_limit_requests() --
|
||||
* plugin.c's plugin_prepare_allocation() references it; never actually
|
||||
* invoked here. */
|
||||
int sched_fill_limit_requests(xce_scheduler_t* sched, xvec_idval_t* vec,
|
||||
xlist_t* limits)
|
||||
{
|
||||
(void)sched;
|
||||
(void)vec;
|
||||
(void)limits;
|
||||
return 0;
|
||||
}
|
||||
|
||||
int sched_account_jobs(xce_scheduler_t* sched)
|
||||
{
|
||||
(void)sched;
|
||||
return 0;
|
||||
}
|
||||
|
||||
/* Real reimplementation of schedobj.c's sched_resolve_job_picker -- same
|
||||
* rationale/algorithm as test_driver_session_ordering.c's own copy
|
||||
* (schedobj.c is not linked). */
|
||||
int sched_resolve_job_picker(xce_sched_core_t* core, xce_sched_plugin_t** out)
|
||||
{
|
||||
xce_sched_plugin_t* plugin = NULL;
|
||||
int rc = 0;
|
||||
|
||||
if ((core == NULL) || (out == NULL)) {
|
||||
return -EINVAL;
|
||||
}
|
||||
|
||||
*out = NULL;
|
||||
|
||||
if (core->job_pick_plugin == NULL) {
|
||||
return -EINVAL;
|
||||
}
|
||||
|
||||
lock(core->plugins, RTTI_LOCK_READ);
|
||||
rc = key_find(core->plugins, core->job_pick_plugin, (void**)&plugin);
|
||||
unlock(core->plugins);
|
||||
|
||||
if (rc <= 0) {
|
||||
return (rc < 0) ? rc : -ENOENT;
|
||||
}
|
||||
|
||||
if ((plugin == NULL) || (plugin->picker_iface == NULL)
|
||||
|| (plugin->picker_iface->next_job == NULL)
|
||||
|| (plugin->picker_iface->begin_session == NULL)
|
||||
|| (plugin->picker_iface->end_session == NULL)
|
||||
|| (plugin->picker_iface->on_dispatch == NULL)
|
||||
|| (plugin->picker_iface->on_pend == NULL)) {
|
||||
return -EINVAL;
|
||||
}
|
||||
|
||||
*out = plugin;
|
||||
|
||||
return 0;
|
||||
}
|
||||
|
||||
void setUp(void) { return; }
|
||||
void tearDown(void) { return; }
|
||||
|
||||
/* ---------------------------------------------------------------------- */
|
||||
/* Stub picker plugin -- identical shape to */
|
||||
/* test_driver_session_ordering.c's own: no jobs are ever pushed, so */
|
||||
/* next_job() always returns NULL immediately. Its SOLE purpose is proving */
|
||||
/* the wbuf/jbuf rendezvous timing lets a real session complete. */
|
||||
/* ---------------------------------------------------------------------- */
|
||||
|
||||
static int stub_picker_begin_session(void* picker)
|
||||
{
|
||||
(void)picker;
|
||||
return 0;
|
||||
}
|
||||
|
||||
static xce_sched_job_t* stub_picker_next_job(void* picker)
|
||||
{
|
||||
(void)picker;
|
||||
return NULL;
|
||||
}
|
||||
|
||||
static int stub_picker_end_session(void* picker)
|
||||
{
|
||||
(void)picker;
|
||||
return 0;
|
||||
}
|
||||
|
||||
static int stub_picker_jobid_noop(void* picker, uint64_t jobid)
|
||||
{
|
||||
(void)picker;
|
||||
(void)jobid;
|
||||
return 0;
|
||||
}
|
||||
|
||||
static int stub_picker_queue_update_noop(void* picker, xce_sched_queue_t* q)
|
||||
{
|
||||
(void)picker;
|
||||
(void)q;
|
||||
return 0;
|
||||
}
|
||||
|
||||
static int stub_picker_job_noop(void* picker, xce_sched_job_t* job)
|
||||
{
|
||||
(void)picker;
|
||||
(void)job;
|
||||
return 0;
|
||||
}
|
||||
|
||||
static int stub_picker_job_state_change_noop(void* picker, xce_sched_job_t* job,
|
||||
uint32_t state)
|
||||
{
|
||||
(void)picker;
|
||||
(void)job;
|
||||
(void)state;
|
||||
return 0;
|
||||
}
|
||||
|
||||
static xce_job_picker_iface_t g_stub_picker_iface
|
||||
= { .init = NULL,
|
||||
.fini = NULL,
|
||||
.begin_session = stub_picker_begin_session,
|
||||
.next_job = stub_picker_next_job,
|
||||
.end_session = stub_picker_end_session,
|
||||
.on_dispatch = stub_picker_jobid_noop,
|
||||
.on_pend = stub_picker_jobid_noop,
|
||||
.on_queue_update = stub_picker_queue_update_noop,
|
||||
.on_job_push = stub_picker_job_noop,
|
||||
.on_job_state_change = stub_picker_job_state_change_noop,
|
||||
.on_job_update = stub_picker_job_noop,
|
||||
.on_config = NULL,
|
||||
.on_job_rusage = NULL };
|
||||
|
||||
static xce_sched_plugin_t g_stub_picker_plugin
|
||||
= { .name = "test_record_lifecycle_picker",
|
||||
.picker_iface = &g_stub_picker_iface };
|
||||
|
||||
static void* picker_thread_fn(void* arg)
|
||||
{
|
||||
xce_sched_core_t* core = (xce_sched_core_t*)arg;
|
||||
xce_sched_plugin_t* job_picker = NULL;
|
||||
int rc;
|
||||
int go = 0;
|
||||
uint64_t jobid;
|
||||
|
||||
rc = sched_resolve_job_picker(core, &job_picker);
|
||||
if (rc < 0) {
|
||||
return NULL;
|
||||
}
|
||||
|
||||
cbuf_get(&(core->wbuf), &go);
|
||||
|
||||
job_picker->picker_iface->begin_session(job_picker);
|
||||
|
||||
for (;;) {
|
||||
xce_sched_job_t* job = job_picker->picker_iface->next_job(job_picker);
|
||||
|
||||
if (job == NULL) {
|
||||
break;
|
||||
}
|
||||
|
||||
jobid = job->jobid;
|
||||
cbuf_put(&(core->jbuf), &jobid);
|
||||
}
|
||||
|
||||
jobid = 0;
|
||||
cbuf_put(&(core->jbuf), &jobid);
|
||||
|
||||
job_picker->picker_iface->end_session(job_picker);
|
||||
|
||||
return NULL;
|
||||
}
|
||||
|
||||
static void* stand_in_allocator_thread_fn(void* arg)
|
||||
{
|
||||
xce_sched_core_t* core = (xce_sched_core_t*)arg;
|
||||
uint64_t jobid;
|
||||
xce_decision_t* decision = NULL;
|
||||
|
||||
cbuf_get(&(core->jbuf), &jobid);
|
||||
(void)jobid;
|
||||
|
||||
cbuf_put(&(core->decisions), &decision);
|
||||
|
||||
return NULL;
|
||||
}
|
||||
|
||||
/* ---------------------------------------------------------------------- */
|
||||
/* Fixture helpers */
|
||||
/* ---------------------------------------------------------------------- */
|
||||
|
||||
static void make_tmp_dir(char* out, size_t out_len, const char* suffix)
|
||||
{
|
||||
char tmpl[PATH_MAX];
|
||||
|
||||
snprintf(tmpl, sizeof(tmpl), "/tmp/sched_reclife%s_XXXXXX", suffix);
|
||||
TEST_ASSERT_NOT_NULL(mkdtemp(tmpl));
|
||||
TEST_ASSERT_TRUE(strlen(tmpl) < out_len);
|
||||
strcpy(out, tmpl);
|
||||
}
|
||||
|
||||
/* Real xce_sched_core_init() + manually dataq_init'd pend_updates + the
|
||||
* stub picker plugin registered -- same pattern as
|
||||
* test_driver_session_ordering.c's fixture. `record_dir` must outlive
|
||||
* `sched` (caller-owned storage). Non-const to match
|
||||
* xce_scheduler_t.record_dir's own `char*` type (scheduler.h) -- every
|
||||
* caller passes a mutable local buffer anyway. */
|
||||
static void fixture_init(xce_scheduler_t* sched, char* record_dir)
|
||||
{
|
||||
int rc;
|
||||
|
||||
memset(sched, 0, sizeof(*sched));
|
||||
|
||||
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);
|
||||
|
||||
lock(sched->core.plugins, RTTI_LOCK_WRITE);
|
||||
rc = key_add(sched->core.plugins, (void*)g_stub_picker_plugin.name,
|
||||
(void*)&g_stub_picker_plugin);
|
||||
TEST_ASSERT_GREATER_OR_EQUAL(0, rc);
|
||||
unlock(sched->core.plugins);
|
||||
|
||||
sched->core.job_pick_plugin = g_stub_picker_plugin.name;
|
||||
sched->record_dir = record_dir;
|
||||
}
|
||||
|
||||
/* Runs one complete normal-mode round (real picker thread + stand-in
|
||||
* allocator thread, same design as
|
||||
* test_driver_session_ordering.c's own single test) and blocks until it
|
||||
* completes. */
|
||||
static void run_one_real_session(xce_scheduler_t* sched)
|
||||
{
|
||||
pthread_t picker_tid;
|
||||
pthread_t allocator_tid;
|
||||
int rc;
|
||||
|
||||
rc = pthread_create(&picker_tid, NULL, picker_thread_fn, &(sched->core));
|
||||
TEST_ASSERT_EQUAL_INT(0, rc);
|
||||
|
||||
rc = pthread_create(&allocator_tid, NULL, stand_in_allocator_thread_fn,
|
||||
&(sched->core));
|
||||
TEST_ASSERT_EQUAL_INT(0, rc);
|
||||
|
||||
xce_sched_driver_run_normal_mode_round(sched);
|
||||
|
||||
rc = pthread_join(picker_tid, NULL);
|
||||
TEST_ASSERT_EQUAL_INT(0, rc);
|
||||
rc = pthread_join(allocator_tid, NULL);
|
||||
TEST_ASSERT_EQUAL_INT(0, rc);
|
||||
}
|
||||
|
||||
/* ---------------------------------------------------------------------- */
|
||||
/* Test 1: start() writes a decodable ReplayHeader then a decodable */
|
||||
/* SnapshotEntry -- this task's own Observable, "byte-decodable by the */
|
||||
/* round-trip decoder from group 7". */
|
||||
/* ---------------------------------------------------------------------- */
|
||||
|
||||
static void test_start_writes_decodable_header_then_snapshot_entry(void)
|
||||
{
|
||||
char record_dir[PATH_MAX];
|
||||
xce_scheduler_t sched;
|
||||
char path_copy[PATH_MAX];
|
||||
int rc;
|
||||
FILE* f;
|
||||
struct capn c1;
|
||||
struct capn c2;
|
||||
ReplayHeader_ptr hdr;
|
||||
SnapshotEntry_ptr snap;
|
||||
capn_text rv;
|
||||
|
||||
make_tmp_dir(record_dir, sizeof(record_dir), "_hdr");
|
||||
fixture_init(&sched, record_dir);
|
||||
|
||||
rc = sched_record_start(&sched, "session.replay");
|
||||
TEST_ASSERT_EQUAL_INT(0, rc);
|
||||
TEST_ASSERT_TRUE(sched.recording);
|
||||
TEST_ASSERT_NOT_NULL(sched.record_file_path);
|
||||
TEST_ASSERT_NOT_NULL(sched.record_file);
|
||||
|
||||
strcpy(path_copy, sched.record_file_path);
|
||||
|
||||
rc = sched_record_stop(&sched);
|
||||
TEST_ASSERT_EQUAL_INT(0, rc);
|
||||
TEST_ASSERT_FALSE(sched.recording);
|
||||
|
||||
f = fopen(path_copy, "r");
|
||||
TEST_ASSERT_NOT_NULL(f);
|
||||
|
||||
/* msg 1: ReplayHeader */
|
||||
TEST_ASSERT_EQUAL_INT(0, capn_init_fp(&c1, f, 0));
|
||||
hdr.p = capn_getp(capn_root(&c1), 0, 1);
|
||||
rv = ReplayHeader_get_recorderVersion(hdr);
|
||||
TEST_ASSERT_EQUAL_STRING("1.0", rv.str);
|
||||
capn_free(&c1);
|
||||
|
||||
/* msg 2: SnapshotEntry -- MUST be the first data message after
|
||||
* ReplayHeader (Req 5 AC 20). */
|
||||
TEST_ASSERT_EQUAL_INT(0, capn_init_fp(&c2, f, 0));
|
||||
snap.p = capn_getp(capn_root(&c2), 0, 1);
|
||||
{
|
||||
SnapshotData_ptr sdata = SnapshotEntry_get_data(snap);
|
||||
TEST_ASSERT_FALSE(sdata.p.type == CAPN_NULL);
|
||||
}
|
||||
capn_free(&c2);
|
||||
|
||||
fclose(f);
|
||||
printf("test_start_writes_decodable_header_then_snapshot_entry: PASS\n");
|
||||
}
|
||||
|
||||
/* ---------------------------------------------------------------------- */
|
||||
/* Test 2: start -> submit an event batch -> complete one session -> stop */
|
||||
/* produces events + exactly one ReplaySession + a trailing endSession */
|
||||
/* marker -- this task's own literal Observable. */
|
||||
/* ---------------------------------------------------------------------- */
|
||||
|
||||
static void
|
||||
test_full_cycle_start_event_session_stop_produces_expected_stream(void)
|
||||
{
|
||||
char record_dir[PATH_MAX];
|
||||
xce_scheduler_t sched;
|
||||
xce_sched_update_request_t* req;
|
||||
char path_copy[PATH_MAX];
|
||||
int rc;
|
||||
FILE* f;
|
||||
struct capn c;
|
||||
|
||||
make_tmp_dir(record_dir, sizeof(record_dir), "_cycle");
|
||||
fixture_init(&sched, record_dir);
|
||||
|
||||
rc = sched_record_start(&sched, "cycle.replay");
|
||||
TEST_ASSERT_EQUAL_INT(0, rc);
|
||||
strcpy(path_copy, sched.record_file_path);
|
||||
|
||||
/* "submit an event batch": one JSTATE_CHANGE update, drained by
|
||||
* xce_sched_driver_translate_pending_events() (called from
|
||||
* xce_sched_driver_run_normal_mode_round() below) -- the SAME
|
||||
* pend_updates dataq mechanism handler.c populates in production. */
|
||||
req = (xce_sched_update_request_t*)calloc(1, sizeof(*req));
|
||||
req->op = XCE_SCHED_OP_JSTATE_CHANGE;
|
||||
req->u.job_state_change.jobid = 42ULL;
|
||||
req->u.job_state_change.state = 7u;
|
||||
dataq_enq(sched.core.pend_updates[0], &(req->link));
|
||||
|
||||
/* "complete a session": one full normal-mode round through a real
|
||||
* picker thread + stand-in allocator thread (no jobs pushed, so the
|
||||
* session produces zero decisions -- still exactly one ReplaySession
|
||||
* message, matching this task's Observable which requires "one
|
||||
* session in between", not any particular decision count). */
|
||||
run_one_real_session(&sched);
|
||||
|
||||
TEST_ASSERT_EQUAL_UINT64(1ULL, sched.record_event_count);
|
||||
|
||||
rc = sched_record_stop(&sched);
|
||||
TEST_ASSERT_EQUAL_INT(0, rc);
|
||||
|
||||
f = fopen(path_copy, "r");
|
||||
TEST_ASSERT_NOT_NULL(f);
|
||||
|
||||
/* msg 1: ReplayHeader */
|
||||
TEST_ASSERT_EQUAL_INT(0, capn_init_fp(&c, f, 0));
|
||||
capn_free(&c);
|
||||
|
||||
/* msg 2: SnapshotEntry (initial) */
|
||||
TEST_ASSERT_EQUAL_INT(0, capn_init_fp(&c, f, 0));
|
||||
capn_free(&c);
|
||||
|
||||
/* msg 3: ReplayEvent{jobStateChange, jobid=42} -- the submitted batch. */
|
||||
{
|
||||
ReplayEvent_ptr p;
|
||||
struct ReplayEvent wire;
|
||||
replay_event_t dst;
|
||||
|
||||
TEST_ASSERT_EQUAL_INT(0, capn_init_fp(&c, f, 0));
|
||||
p.p = capn_getp(capn_root(&c), 0, 1);
|
||||
read_ReplayEvent(&wire, p);
|
||||
memset(&dst, 0, sizeof(dst));
|
||||
decode_ReplayEvent(&dst, &wire);
|
||||
TEST_ASSERT_EQUAL_INT((int)ReplayEvent_body_jobStateChange,
|
||||
(int)dst.body_which);
|
||||
TEST_ASSERT_NOT_NULL(dst.body.job_state_change);
|
||||
TEST_ASSERT_EQUAL_UINT64(42ULL, dst.body.job_state_change->jobid);
|
||||
free_ReplayEvent(&dst);
|
||||
capn_free(&c);
|
||||
}
|
||||
|
||||
/* msg 4: exactly one ReplaySession. */
|
||||
{
|
||||
ReplaySession_ptr p;
|
||||
struct ReplaySession wire;
|
||||
|
||||
TEST_ASSERT_EQUAL_INT(0, capn_init_fp(&c, f, 0));
|
||||
p.p = capn_getp(capn_root(&c), 0, 1);
|
||||
read_ReplaySession(&wire, p);
|
||||
(void)wire; /* begin_round/decisions checked structurally only --
|
||||
* this fixture pushes no jobs, so 0 decisions is
|
||||
* correct and expected. */
|
||||
capn_free(&c);
|
||||
}
|
||||
|
||||
/* msg 5: the end-of-recording marker, ReplayEvent{endSession}. */
|
||||
{
|
||||
ReplayEvent_ptr p;
|
||||
struct ReplayEvent wire;
|
||||
replay_event_t dst;
|
||||
|
||||
TEST_ASSERT_EQUAL_INT(0, capn_init_fp(&c, f, 0));
|
||||
p.p = capn_getp(capn_root(&c), 0, 1);
|
||||
read_ReplayEvent(&wire, p);
|
||||
memset(&dst, 0, sizeof(dst));
|
||||
decode_ReplayEvent(&dst, &wire);
|
||||
TEST_ASSERT_EQUAL_INT((int)ReplayEvent_body_endSession,
|
||||
(int)dst.body_which);
|
||||
free_ReplayEvent(&dst);
|
||||
capn_free(&c);
|
||||
}
|
||||
|
||||
/* Exactly 5 messages -- nothing more to read. */
|
||||
TEST_ASSERT_NOT_EQUAL(0, capn_init_fp(&c, f, 0));
|
||||
|
||||
fclose(f);
|
||||
printf("test_full_cycle_start_event_session_stop_produces_expected_stream:"
|
||||
" PASS\n");
|
||||
}
|
||||
|
||||
/* ---------------------------------------------------------------------- */
|
||||
/* Test 3: when recording is inactive, the driver-loop hooks touch no */
|
||||
/* file/record state -- zero file I/O (Req 5 AC 10, "Performance"). */
|
||||
/* ---------------------------------------------------------------------- */
|
||||
|
||||
static void test_recording_inactive_hooks_are_zero_cost_noops(void)
|
||||
{
|
||||
char record_dir[PATH_MAX];
|
||||
xce_scheduler_t sched;
|
||||
xce_core_event_t events[1];
|
||||
sched_record_session_ctx_t* ctx;
|
||||
int rc;
|
||||
|
||||
make_tmp_dir(record_dir, sizeof(record_dir), "_inactive");
|
||||
fixture_init(&sched, record_dir);
|
||||
/* sched.recording deliberately never set true. */
|
||||
|
||||
memset(events, 0, sizeof(events));
|
||||
events[0].type = XCE_CORE_EVENT_JOB_STATE_CHANGE;
|
||||
events[0].u.job_state_change.jobid = 99ULL;
|
||||
|
||||
rc = sched_record_write_events(&sched, events, 1);
|
||||
TEST_ASSERT_EQUAL_INT(0, rc);
|
||||
TEST_ASSERT_NULL(sched.record_file);
|
||||
TEST_ASSERT_NULL(sched.record_file_path);
|
||||
|
||||
ctx = sched_record_session_begin(&sched, 123ULL);
|
||||
TEST_ASSERT_NULL(ctx);
|
||||
TEST_ASSERT_NULL(sched.record_file);
|
||||
|
||||
/* Driving a full real round with recording inactive must not create
|
||||
* any file either. */
|
||||
run_one_real_session(&sched);
|
||||
TEST_ASSERT_FALSE(sched.recording);
|
||||
TEST_ASSERT_NULL(sched.record_file);
|
||||
TEST_ASSERT_NULL(sched.record_file_path);
|
||||
TEST_ASSERT_EQUAL_UINT64(0ULL, sched.record_event_count);
|
||||
}
|
||||
|
||||
/* ---------------------------------------------------------------------- */
|
||||
/* Test 4: double-start / stop-when-not-recording fail cleanly. */
|
||||
/* ---------------------------------------------------------------------- */
|
||||
|
||||
static void test_double_start_fails_cleanly(void)
|
||||
{
|
||||
char record_dir[PATH_MAX];
|
||||
xce_scheduler_t sched;
|
||||
int rc;
|
||||
|
||||
make_tmp_dir(record_dir, sizeof(record_dir), "_dblstart");
|
||||
fixture_init(&sched, record_dir);
|
||||
|
||||
rc = sched_record_start(&sched, "a.replay");
|
||||
TEST_ASSERT_EQUAL_INT(0, rc);
|
||||
|
||||
rc = sched_record_start(&sched, "b.replay");
|
||||
TEST_ASSERT_EQUAL_INT(-EALREADY, rc);
|
||||
/* Original recording untouched by the rejected second start. */
|
||||
TEST_ASSERT_TRUE(sched.recording);
|
||||
TEST_ASSERT_TRUE(strstr(sched.record_file_path, "a.replay") != NULL);
|
||||
|
||||
rc = sched_record_stop(&sched);
|
||||
TEST_ASSERT_EQUAL_INT(0, rc);
|
||||
}
|
||||
|
||||
static void test_stop_when_not_recording_fails_cleanly(void)
|
||||
{
|
||||
char record_dir[PATH_MAX];
|
||||
xce_scheduler_t sched;
|
||||
int rc;
|
||||
|
||||
make_tmp_dir(record_dir, sizeof(record_dir), "_nostart");
|
||||
fixture_init(&sched, record_dir);
|
||||
|
||||
rc = sched_record_stop(&sched);
|
||||
TEST_ASSERT_EQUAL_INT(-EALREADY, rc);
|
||||
|
||||
rc = sched_record_start(&sched, "c.replay");
|
||||
TEST_ASSERT_EQUAL_INT(0, rc);
|
||||
rc = sched_record_stop(&sched);
|
||||
TEST_ASSERT_EQUAL_INT(0, rc);
|
||||
|
||||
/* Stopping again after a clean stop also fails cleanly. */
|
||||
rc = sched_record_stop(&sched);
|
||||
TEST_ASSERT_EQUAL_INT(-EALREADY, rc);
|
||||
}
|
||||
|
||||
static void test_null_sched_returns_einval(void)
|
||||
{
|
||||
TEST_ASSERT_EQUAL_INT(-EINVAL, sched_record_start(NULL, "x.replay"));
|
||||
TEST_ASSERT_EQUAL_INT(-EINVAL, sched_record_stop(NULL));
|
||||
}
|
||||
|
||||
/* ---------------------------------------------------------------------- */
|
||||
/* Test 5: TOCTOU -- a symlink/hard link swapped in at the resolved path */
|
||||
/* between path validation and the eventual open() is rejected, not */
|
||||
/* silently followed (recording.h's own documented mandate for this task). */
|
||||
/* */
|
||||
/* Genuinely racing a background thread against the exact few-instruction */
|
||||
/* window inside sched_record_start() between its internal */
|
||||
/* sched_validate_record_filename() call and its open() call would be */
|
||||
/* inherently flaky (no test hook exists to pause mid-function, nor should */
|
||||
/* one be added to production code just for this). Instead, both tests */
|
||||
/* below reproduce the STATE a genuine race would produce -- a malicious */
|
||||
/* symlink/hard link occupying the target path at the moment */
|
||||
/* sched_record_start() is invoked, planted via real symlink(2)/link(2) */
|
||||
/* syscalls (no mocking) -- and prove the full end-to-end call rejects it, */
|
||||
/* AND (the part that specifically proves the open()-time hardening this */
|
||||
/* task added, not just sched_validate_record_filename()'s pre-existing */
|
||||
/* containment check) that the exact open()+fstat()+st_nlink==1 sequence */
|
||||
/* sched_record_start() runs internally -- reproduced here verbatim, same */
|
||||
/* flags -- independently rejects the swap. Either layer alone would be */
|
||||
/* enough in production; proving the second layer here is what closes the */
|
||||
/* TOCTOU gap this task's own docstring (recording.h) flags: a real race */
|
||||
/* would land AFTER sched_validate_record_filename() already accepted the */
|
||||
/* path, i.e. exactly the scenario where ONLY the open()-time check saves */
|
||||
/* it. */
|
||||
/* ---------------------------------------------------------------------- */
|
||||
|
||||
static int toctou_hardened_open(const char* path)
|
||||
{
|
||||
int fd;
|
||||
struct stat st;
|
||||
|
||||
/* Verbatim copy of sched_record_start()'s own hardening sequence
|
||||
* (recording.c) -- see this test's own header comment for why this is
|
||||
* reproduced rather than racing a real thread. Deliberately NO
|
||||
* O_TRUNC: it truncates atomically as part of open() itself, before
|
||||
* the fstat()/st_nlink check below ever runs -- against a hard-linked
|
||||
* victim (same inode, no symlink involved) that would corrupt the
|
||||
* victim regardless of what the check then decides. ftruncate() only
|
||||
* runs after the check passes. */
|
||||
fd = open(path, O_WRONLY | O_CREAT | O_NOFOLLOW, 0640);
|
||||
if (fd < 0) {
|
||||
return (errno > 0) ? -errno : -EIO;
|
||||
}
|
||||
|
||||
if (fstat(fd, &st) != 0) {
|
||||
int rc = (errno > 0) ? -errno : -EIO;
|
||||
close(fd);
|
||||
return rc;
|
||||
}
|
||||
if (!S_ISREG(st.st_mode) || (st.st_nlink != 1)) {
|
||||
close(fd);
|
||||
return -EACCES;
|
||||
}
|
||||
|
||||
if (ftruncate(fd, 0) != 0) {
|
||||
int rc = (errno > 0) ? -errno : -EIO;
|
||||
close(fd);
|
||||
return rc;
|
||||
}
|
||||
|
||||
close(fd);
|
||||
return 0;
|
||||
}
|
||||
|
||||
static void test_toctou_symlink_planted_after_validation_rejected(void)
|
||||
{
|
||||
char record_dir[PATH_MAX];
|
||||
char outside_dir[PATH_MAX];
|
||||
char outside_secret[PATH_MAX];
|
||||
char resolved[PATH_MAX];
|
||||
char target[PATH_MAX];
|
||||
const char* secret_content = "do-not-touch";
|
||||
int rc;
|
||||
FILE* f;
|
||||
char readback[64];
|
||||
|
||||
make_tmp_dir(record_dir, sizeof(record_dir), "_toctou_sym_src");
|
||||
make_tmp_dir(outside_dir, sizeof(outside_dir), "_toctou_sym_dst");
|
||||
|
||||
snprintf(outside_secret, sizeof(outside_secret), "%s/secret", outside_dir);
|
||||
f = fopen(outside_secret, "w");
|
||||
TEST_ASSERT_NOT_NULL(f);
|
||||
fputs(secret_content, f);
|
||||
fclose(f);
|
||||
|
||||
/* Simulates "validation just resolved this path and found nothing
|
||||
* there yet" (a fresh filename -- sched_validate_record_filename()
|
||||
* itself accepts this, since lstat() on a nonexistent entry short-
|
||||
* circuits its containment checks entirely). */
|
||||
rc = sched_validate_record_filename(record_dir, "race.replay", resolved,
|
||||
sizeof(resolved));
|
||||
TEST_ASSERT_EQUAL_INT(0, rc);
|
||||
|
||||
/* The race: a symlink appears at the just-validated path, pointing
|
||||
* OUTSIDE record_dir, in the window before open() runs. */
|
||||
snprintf(target, sizeof(target), "%s/race.replay", record_dir);
|
||||
TEST_ASSERT_EQUAL_INT(0, symlink(outside_secret, target));
|
||||
|
||||
/* Layer 1: the open()-time hardening sequence itself rejects it. */
|
||||
rc = toctou_hardened_open(resolved);
|
||||
TEST_ASSERT_TRUE(rc < 0);
|
||||
|
||||
/* Layer 2 (defense in depth): a real end-to-end sched_record_start()
|
||||
* call against the same record_dir/filename is ALSO rejected --
|
||||
* sched_validate_record_filename() re-runs and now sees the symlink,
|
||||
* closing the loop even though this particular ordering happens to be
|
||||
* caught one layer earlier than the pure race would be. */
|
||||
{
|
||||
xce_scheduler_t sched;
|
||||
|
||||
fixture_init(&sched, record_dir);
|
||||
rc = sched_record_start(&sched, "race.replay");
|
||||
TEST_ASSERT_TRUE(rc < 0);
|
||||
TEST_ASSERT_FALSE(sched.recording);
|
||||
TEST_ASSERT_NULL(sched.record_file);
|
||||
}
|
||||
|
||||
/* The outside secret file was never opened for writing/truncated. */
|
||||
f = fopen(outside_secret, "r");
|
||||
TEST_ASSERT_NOT_NULL(f);
|
||||
memset(readback, 0, sizeof(readback));
|
||||
TEST_ASSERT_NOT_NULL(fgets(readback, sizeof(readback), f));
|
||||
fclose(f);
|
||||
TEST_ASSERT_EQUAL_STRING(secret_content, readback);
|
||||
}
|
||||
|
||||
static void test_toctou_hardlink_planted_after_validation_rejected(void)
|
||||
{
|
||||
char record_dir[PATH_MAX];
|
||||
char outside_dir[PATH_MAX];
|
||||
char victim[PATH_MAX];
|
||||
char resolved[PATH_MAX];
|
||||
char aliased[PATH_MAX];
|
||||
const char* victim_content = "do-not-alias";
|
||||
int rc;
|
||||
FILE* f;
|
||||
char readback[64];
|
||||
|
||||
make_tmp_dir(record_dir, sizeof(record_dir), "_toctou_hl_src");
|
||||
make_tmp_dir(outside_dir, sizeof(outside_dir), "_toctou_hl_dst");
|
||||
|
||||
snprintf(victim, sizeof(victim), "%s/victim", outside_dir);
|
||||
f = fopen(victim, "w");
|
||||
TEST_ASSERT_NOT_NULL(f);
|
||||
fputs(victim_content, f);
|
||||
fclose(f);
|
||||
|
||||
rc = sched_validate_record_filename(record_dir, "alias.replay", resolved,
|
||||
sizeof(resolved));
|
||||
TEST_ASSERT_EQUAL_INT(0, rc);
|
||||
|
||||
/* The race: a hard link to the OUTSIDE victim file appears at the
|
||||
* just-validated path -- same inode, no symlink involved, so
|
||||
* O_NOFOLLOW alone would NOT catch this (recording.h's own doc
|
||||
* comment: "no symlink is involved in that race, so O_NOFOLLOW never
|
||||
* triggers") -- only the fstat()+st_nlink==1 check does. */
|
||||
snprintf(aliased, sizeof(aliased), "%s/alias.replay", record_dir);
|
||||
TEST_ASSERT_EQUAL_INT(0, link(victim, aliased));
|
||||
|
||||
rc = toctou_hardened_open(resolved);
|
||||
TEST_ASSERT_EQUAL_INT(-EACCES, rc);
|
||||
|
||||
{
|
||||
xce_scheduler_t sched;
|
||||
|
||||
fixture_init(&sched, record_dir);
|
||||
rc = sched_record_start(&sched, "alias.replay");
|
||||
TEST_ASSERT_TRUE(rc < 0);
|
||||
TEST_ASSERT_FALSE(sched.recording);
|
||||
TEST_ASSERT_NULL(sched.record_file);
|
||||
}
|
||||
|
||||
f = fopen(victim, "r");
|
||||
TEST_ASSERT_NOT_NULL(f);
|
||||
memset(readback, 0, sizeof(readback));
|
||||
TEST_ASSERT_NOT_NULL(fgets(readback, sizeof(readback), f));
|
||||
fclose(f);
|
||||
TEST_ASSERT_EQUAL_STRING(victim_content, readback);
|
||||
}
|
||||
|
||||
int main(void)
|
||||
{
|
||||
UNITY_BEGIN();
|
||||
|
||||
RUN_TEST(test_start_writes_decodable_header_then_snapshot_entry);
|
||||
RUN_TEST(test_full_cycle_start_event_session_stop_produces_expected_stream);
|
||||
RUN_TEST(test_recording_inactive_hooks_are_zero_cost_noops);
|
||||
RUN_TEST(test_double_start_fails_cleanly);
|
||||
RUN_TEST(test_stop_when_not_recording_fails_cleanly);
|
||||
RUN_TEST(test_null_sched_returns_einval);
|
||||
RUN_TEST(test_toctou_symlink_planted_after_validation_rejected);
|
||||
RUN_TEST(test_toctou_hardlink_planted_after_validation_rejected);
|
||||
|
||||
return UNITY_END();
|
||||
}
|
||||
|
|
@ -43,6 +43,30 @@
|
|||
/* Pull in the real sched_validate_record_filename() under test. */
|
||||
#include "../src/recording.c"
|
||||
|
||||
/* Link-satisfying stubs: recording.c also defines sched_record_start()/
|
||||
* sched_record_stop() (scheduler-core spec, task 9.2) as of this revision,
|
||||
* which reference xce_core_now() (core_clock.c) and
|
||||
* xce_sched_core_save_snapshot() (snapshot.c) -- neither pulled into this
|
||||
* TU, same rationale as every other narrow white-box test in this
|
||||
* directory (this file's own header comment: "the function under test has
|
||||
* no internal state or fixture dependencies", still true of
|
||||
* sched_validate_record_filename() itself; this task's own additions are
|
||||
* exercised by a separate, dedicated test file). Never actually invoked at
|
||||
* runtime by any test in this file. */
|
||||
uint64_t xce_core_now(xce_sched_core_t* core)
|
||||
{
|
||||
(void)core;
|
||||
return 0;
|
||||
}
|
||||
|
||||
int xce_sched_core_save_snapshot(xce_sched_core_t* core,
|
||||
sched_snapshot_data_t* out)
|
||||
{
|
||||
(void)core;
|
||||
(void)out;
|
||||
return -ENOSYS;
|
||||
}
|
||||
|
||||
void setUp(void) { return; }
|
||||
void tearDown(void) { return; }
|
||||
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue