lei.chen/myxce
1
0
Fork 1
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions
myxce/src/scheduler/testsuite/test_repl_trace_emit.c
Rongsong Shen 4df29606f6
feat(scheduler-sim-decision-trace): emit traces inline and apply settings at session boundaries 
Task 4.6: session begin makes the next settings current, stamps the
session number and publishes them to the core; session end withdraws and
releases them. Emission marks the round traced, flushes the collapsed
range and reprints the suppressed echo in continue mode, renders to the
console and the trace file with identical bytes, and stores the record.
sim_print_decision is exported unchanged.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
2026-10-08 14:42:27 +00:00

708 lines
20 KiB
C
Raw Blame History

/*
* test_repl_trace_emit.c - inline trace emission and the trace session
* hooks (scheduler-sim-decision-trace spec, task 4.6; design.md
* "SimIntegration", "ReplTraceControl", "System Flows" -> "Session
* lifecycle and capture" and "Drain-loop emission order", "Error
* Handling"; Requirements 1.6, 7.1, 7.2, 7.3, 8.1, 8.5).
*
* Links every scheduler source except main.c, so the emitter runs against
* the real REPL and simulation code it calls: repl_round_collapse_flush()
* (repl_rollup.c) and the decision echo printer sim_print_decision()
* (sim_loop.c). The sim driver and REPL state are zeroed fixtures; nothing
* is started. Each test drives the hooks the way the drain loop will
* (task 5.3): session begin, then per decision the echo gate (step mode
* prints the echo itself, continue mode suppresses it) followed by the
* emitter, then session end.
*
* Covers:
* - session begin makes `next` current, stamps the session number,
* resets the position and publishes the control block; with tracing
* off it publishes nothing (1.6);
* - session end withdraws the control block, drops `cur` (closing a
* replaced file) and clears the in-progress flag (8.5);
* - step mode: a traced decision's trace follows its echo line, and a
* no-decision trace sits where the echo would be (7.1, 7.2);
* - continue mode: the round's first traced item flushes the collapsed
* range, then the suppressed echo line, then the trace; later items
* do not flush again (7.3);
* - the per-round flags reset at session begin, including after a round
* that ended in step mode;
* - the file copy is byte-equal to the console trace (8.1);
* - a failed write to the trace file is reported once per session.
*/
#define _GNU_SOURCE
#include <errno.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/stat.h>
#include <unistd.h>
#include <unity/unity.h>
#include "repl_internal.h"
#include "sched/trace.h"
#include "sim.h"
#include "utils.h"
#define JOB_A ((uint64_t)42 << 32)
#define JOB_B ((uint64_t)43 << 32)
static xce_sched_sim_driver_t g_sim;
static xce_sched_repl_state_t g_rs;
static char g_dir[64];
static char g_cwd[4096];
/* stdout capture */
static int g_cap_fd = -1;
static int g_cap_saved = -1;
static char g_cap_path[64];
void setUp(void)
{
memset(&g_sim, 0, sizeof(g_sim));
memset(&g_rs, 0, sizeof(g_rs));
g_rs.units_mode = XCE_REPL_UNITS_BOTH;
g_rs.mode = XCE_REPL_MODE_STEP;
g_sim.interactive = true;
g_sim.repl_state = &g_rs;
TEST_ASSERT_NOT_NULL(getcwd(g_cwd, sizeof(g_cwd)));
snprintf(g_dir, sizeof(g_dir), "/tmp/xce_trace_emit_XXXXXX");
TEST_ASSERT_NOT_NULL(mkdtemp(g_dir));
TEST_ASSERT_EQUAL_INT(0, chdir(g_dir));
}
void tearDown(void)
{
repl_trace_shutdown(&g_sim);
free(g_rs.collapse_pend_jobids);
g_rs.collapse_pend_jobids = NULL;
g_sim.repl_state = NULL;
unlink("a.trace");
unlink("b.trace");
TEST_ASSERT_EQUAL_INT(0, chdir(g_cwd));
rmdir(g_dir);
}
/* ---- helpers ------------------------------------------------------------- */
static void cap_begin(void)
{
snprintf(g_cap_path, sizeof(g_cap_path), "/tmp/xce_trace_emit_out_XXXXXX");
g_cap_fd = mkstemp(g_cap_path);
TEST_ASSERT_TRUE(g_cap_fd >= 0);
fflush(stdout);
g_cap_saved = dup(STDOUT_FILENO);
TEST_ASSERT_TRUE(g_cap_saved >= 0);
TEST_ASSERT_TRUE(dup2(g_cap_fd, STDOUT_FILENO) >= 0);
}
/* Ends the capture and returns what was printed (caller frees). */
static char* cap_end(void)
{
off_t len;
char* buf;
fflush(stdout);
dup2(g_cap_saved, STDOUT_FILENO);
close(g_cap_saved);
len = lseek(g_cap_fd, 0, SEEK_CUR);
TEST_ASSERT_TRUE(len >= 0);
buf = (char*)calloc(1, (size_t)len + 1);
TEST_ASSERT_NOT_NULL(buf);
TEST_ASSERT_TRUE(lseek(g_cap_fd, 0, SEEK_SET) == 0);
if (len > 0) {
TEST_ASSERT_TRUE(read(g_cap_fd, buf, (size_t)len) == len);
}
close(g_cap_fd);
unlink(g_cap_path);
return buf;
}
static void cmd(const char* line)
{
cap_begin();
(void)repl_trace_cmd(&g_sim, line);
free(cap_end());
}
static char* read_file(const char* path)
{
FILE* f = fopen(path, "r");
char* buf;
long len;
TEST_ASSERT_NOT_NULL(f);
TEST_ASSERT_EQUAL_INT(0, fseek(f, 0, SEEK_END));
len = ftell(f);
TEST_ASSERT_TRUE(len >= 0);
rewind(f);
buf = (char*)calloc(1, (size_t)len + 1);
TEST_ASSERT_NOT_NULL(buf);
if (len > 0) {
TEST_ASSERT_EQUAL_size_t((size_t)len, fread(buf, 1, (size_t)len, f));
}
fclose(f);
return buf;
}
/* A record for `jobid` under the session's published settings. */
static xce_sched_trace_rec_t* new_rec(uint64_t jobid,
xce_sched_trace_rec_kind_t kind)
{
xce_sched_trace_ctl_t ctl;
xce_sched_trace_rec_t* rec;
memset(&ctl, 0, sizeof(ctl));
ctl.all = true;
ctl.level = XCE_TRACE_LEVEL_SUMMARY;
ctl.line_bound = 1000;
ctl.session_no = 3;
ctl.position = 1;
rec = sched_trace_rec_new(&ctl, jobid, XCE_TRACE_REC_DECISION);
TEST_ASSERT_NOT_NULL(rec);
if (kind != XCE_TRACE_REC_DECISION) {
sched_trace_rec_finish(rec, kind, 0);
}
return rec;
}
/* The renderer's text for `rec` in the current units mode (caller frees). */
static char* render(const xce_sched_trace_rec_t* rec)
{
char* buf = NULL;
size_t len = 0;
FILE* f = open_memstream(&buf, &len);
TEST_ASSERT_NOT_NULL(f);
repl_trace_render(rec, g_rs.units_mode, f, NULL);
fclose(f);
TEST_ASSERT_TRUE(len > 0);
return buf;
}
typedef struct {
xce_decision_t d;
xce_job_alloc_t alloc;
xce_pendreason_t reason;
} fixture_decision_t;
static void dispatch_of(fixture_decision_t* f, uint64_t jobid)
{
memset(f, 0, sizeof(*f));
f->alloc.jobid = jobid;
f->d.op = XCE_DECISION_DISPATCH;
f->d.u.alloc = &(f->alloc);
f->d.trace = new_rec(jobid, XCE_TRACE_REC_DECISION);
}
static void pend_of(fixture_decision_t* f, uint64_t jobid)
{
memset(f, 0, sizeof(*f));
f->reason.jobid = jobid;
f->d.op = XCE_DECISION_PEND;
f->d.u.reason = &(f->reason);
f->d.trace = new_rec(jobid, XCE_TRACE_REC_DECISION);
}
static void trace_only_of(fixture_decision_t* f, uint64_t jobid)
{
memset(f, 0, sizeof(*f));
f->d.op = XCE_DECISION_TRACE;
f->d.trace = new_rec(jobid, XCE_TRACE_REC_NOT_FOUND);
}
static char* echo_line(const char* op, uint64_t jobid)
{
char id[64];
char* s = NULL;
TEST_ASSERT_TRUE(fmt_jobid(id, sizeof(id), jobid) >= 0);
TEST_ASSERT_TRUE(asprintf(&s, "[sim] %s jobid=%s\n", op, id) > 0);
return s;
}
/* What the drain loop does per traced decision (task 5.3): the echo gate
* (prints only in step mode), then the emitter. */
static void drain_one(xce_decision_t* d)
{
if (d->op == XCE_DECISION_TRACE) {
repl_trace_on_trace_only(&g_sim, d);
return;
}
if (g_rs.mode != XCE_REPL_MODE_CONTINUE) {
sim_print_decision(d);
}
repl_trace_on_decision(&g_sim, d);
}
/* An active collapsed range over idle rounds 3-4, which a flush prints. */
static void open_idle_range(void)
{
g_rs.collapse_active = true;
g_rs.collapse_start_round = 3;
g_rs.collapse_end_round = 4;
}
static char* flush_line(void)
{
char* out;
cap_begin();
repl_round_collapse_flush(&g_sim);
out = cap_end();
TEST_ASSERT_TRUE(strlen(out) > 0);
return out;
}
static char* concat3(const char* a, const char* b, const char* c)
{
char* s = NULL;
TEST_ASSERT_TRUE(asprintf(&s, "%s%s%s", a, b, c) >= 0);
return s;
}
/* ---- session hooks (1.6, 8.5) -------------------------------------------- */
void test_session_begin_publishes_next_with_stamp(void)
{
xce_sched_trace_ctl_t* ctl;
cmd("trace on all");
TEST_ASSERT_NOT_NULL(g_rs.trace.next);
ctl = repl_trace_settings_ctl(g_rs.trace.next);
ctl->position = 17;
g_rs.session_count = 9;
g_rs.trace.round_traced = true;
g_rs.trace.round_flushed = true;
repl_trace_session_begin(&g_sim);
TEST_ASSERT_TRUE(g_rs.trace.session_in_progress);
TEST_ASSERT_TRUE(g_rs.trace.cur == g_rs.trace.next);
TEST_ASSERT_TRUE(g_sim.core.trace_ctl == ctl);
TEST_ASSERT_EQUAL_UINT64(9, ctl->session_no);
TEST_ASSERT_EQUAL_INT(0, ctl->position);
TEST_ASSERT_FALSE(g_rs.trace.round_traced);
TEST_ASSERT_FALSE(g_rs.trace.round_flushed);
repl_trace_session_end(&g_sim);
TEST_ASSERT_NULL(g_sim.core.trace_ctl);
TEST_ASSERT_NULL(g_rs.trace.cur);
TEST_ASSERT_FALSE(g_rs.trace.session_in_progress);
TEST_ASSERT_NOT_NULL(g_rs.trace.next); /* still the next settings */
}
void test_session_begin_with_tracing_off_publishes_nothing(void)
{
g_sim.core.trace_ctl = NULL;
repl_trace_session_begin(&g_sim);
TEST_ASSERT_TRUE(g_rs.trace.session_in_progress);
TEST_ASSERT_NULL(g_rs.trace.cur);
TEST_ASSERT_NULL(g_sim.core.trace_ctl);
repl_trace_session_end(&g_sim);
TEST_ASSERT_FALSE(g_rs.trace.session_in_progress);
}
void test_command_during_session_applies_next_session(void)
{
xce_repl_trace_settings_t* first;
cmd("trace on all");
first = g_rs.trace.next;
repl_trace_session_begin(&g_sim);
cmd("trace off"); /* 1.6: the session keeps its own settings */
TEST_ASSERT_TRUE(g_rs.trace.cur == first);
TEST_ASSERT_TRUE(g_sim.core.trace_ctl == repl_trace_settings_ctl(first));
repl_trace_session_end(&g_sim);
TEST_ASSERT_NULL(g_sim.core.trace_ctl);
repl_trace_session_begin(&g_sim);
TEST_ASSERT_NULL(g_rs.trace.cur);
TEST_ASSERT_NULL(g_sim.core.trace_ctl);
repl_trace_session_end(&g_sim);
}
void test_replaced_file_kept_until_session_end(void)
{
fixture_decision_t f;
char* a;
char* b;
char* expect;
cmd("trace on all file a.trace");
repl_trace_session_begin(&g_sim);
cmd("trace on all file b.trace"); /* replaces next during the session */
TEST_ASSERT_TRUE(repl_trace_file_conflicts(&g_sim, "a.trace"));
trace_only_of(&f, JOB_A);
expect = render(f.d.trace);
cap_begin();
drain_one(&f.d);
free(cap_end());
TEST_ASSERT_NULL(f.d.trace);
repl_trace_session_end(&g_sim); /* 8.5: a.trace is closed now */
TEST_ASSERT_FALSE(repl_trace_file_conflicts(&g_sim, "a.trace"));
TEST_ASSERT_TRUE(repl_trace_file_conflicts(&g_sim, "b.trace"));
a = read_file("a.trace");
b = read_file("b.trace");
TEST_ASSERT_EQUAL_STRING(expect, a); /* captured under a's settings */
TEST_ASSERT_EQUAL_STRING("", b);
free(a);
free(b);
free(expect);
}
/* ---- step mode (7.1, 7.2) -------------------------------------------------
*/
void test_step_mode_trace_follows_echo(void)
{
fixture_decision_t f;
fixture_decision_t t;
char* out;
char* echo;
char* tr_d;
char* tr_t;
char* expect;
cmd("trace on all");
repl_trace_session_begin(&g_sim);
open_idle_range(); /* never flushed in step mode */
dispatch_of(&f, JOB_A);
trace_only_of(&t, JOB_B);
echo = echo_line("DISPATCH", JOB_A);
tr_d = render(f.d.trace);
tr_t = render(t.d.trace);
cap_begin();
drain_one(&f.d);
drain_one(&t.d);
out = cap_end();
expect = concat3(echo, tr_d, tr_t);
TEST_ASSERT_EQUAL_STRING(expect, out);
TEST_ASSERT_NULL(f.d.trace); /* handed to the store */
TEST_ASSERT_NULL(t.d.trace);
TEST_ASSERT_TRUE(repl_trace_store_bytes(&g_rs.trace) > 0);
TEST_ASSERT_TRUE(g_rs.trace.round_traced);
TEST_ASSERT_FALSE(g_rs.trace.round_flushed);
TEST_ASSERT_TRUE(g_rs.collapse_active);
repl_trace_session_end(&g_sim);
free(out);
free(echo);
free(tr_d);
free(tr_t);
free(expect);
}
/* ---- continue mode (7.3) --------------------------------------------------
*/
void test_continue_mode_flush_echo_then_trace(void)
{
fixture_decision_t f1;
fixture_decision_t f2;
fixture_decision_t t;
char* flush;
char* out;
char* e1;
char* e2;
char* tr1;
char* tr2;
char* trt;
char* part1;
char* part2;
char* expect;
/* the flush line this range prints, taken from the real flush */
open_idle_range();
flush = flush_line();
TEST_ASSERT_FALSE(g_rs.collapse_active);
cmd("trace on all");
g_rs.mode = XCE_REPL_MODE_CONTINUE;
repl_trace_session_begin(&g_sim);
open_idle_range();
dispatch_of(&f1, JOB_A);
pend_of(&f2, JOB_B);
trace_only_of(&t, JOB_B + 1);
e1 = echo_line("DISPATCH", JOB_A);
e2 = echo_line("PEND", JOB_B);
tr1 = render(f1.d.trace);
tr2 = render(f2.d.trace);
trt = render(t.d.trace);
cap_begin();
drain_one(&f1.d);
drain_one(&f2.d);
drain_one(&t.d);
out = cap_end();
part1 = concat3(flush, e1, tr1);
part2 = concat3(e2, tr2, trt);
expect = concat3(part1, part2, "");
TEST_ASSERT_EQUAL_STRING(expect, out);
TEST_ASSERT_TRUE(g_rs.trace.round_traced);
TEST_ASSERT_TRUE(g_rs.trace.round_flushed);
TEST_ASSERT_FALSE(g_rs.collapse_active);
repl_trace_session_end(&g_sim);
free(flush);
free(out);
free(e1);
free(e2);
free(tr1);
free(tr2);
free(trt);
free(part1);
free(part2);
free(expect);
}
void test_continue_mode_trace_only_first_flushes_without_echo(void)
{
fixture_decision_t t;
char* flush;
char* trt;
char* out;
char* expect;
open_idle_range();
flush = flush_line();
cmd("trace on all");
g_rs.mode = XCE_REPL_MODE_CONTINUE;
repl_trace_session_begin(&g_sim);
open_idle_range();
trace_only_of(&t, JOB_A);
trt = render(t.d.trace);
cap_begin();
drain_one(&t.d);
out = cap_end();
expect = concat3(flush, trt, "");
TEST_ASSERT_EQUAL_STRING(expect, out);
repl_trace_session_end(&g_sim);
free(flush);
free(trt);
free(out);
free(expect);
}
/* ---- per-round flags ------------------------------------------------------
*/
void test_flags_reset_after_round_ended_in_step_mode(void)
{
fixture_decision_t f;
fixture_decision_t g;
char* flush;
char* out;
cmd("trace on all");
/* round 1: traced in continue mode, then a breakpoint and `step` */
g_rs.mode = XCE_REPL_MODE_CONTINUE;
repl_trace_session_begin(&g_sim);
dispatch_of(&f, JOB_A);
cap_begin();
drain_one(&f.d);
free(cap_end());
TEST_ASSERT_TRUE(g_rs.trace.round_traced);
TEST_ASSERT_TRUE(g_rs.trace.round_flushed);
g_rs.mode = XCE_REPL_MODE_STEP;
repl_trace_session_end(&g_sim);
/* round 2 begins in step mode: neither flag leaks into it */
repl_trace_session_begin(&g_sim);
TEST_ASSERT_FALSE(g_rs.trace.round_traced);
TEST_ASSERT_FALSE(g_rs.trace.round_flushed);
repl_trace_session_end(&g_sim);
/* round 3, continue mode: the first traced item flushes again */
open_idle_range();
flush = flush_line();
g_rs.mode = XCE_REPL_MODE_CONTINUE;
repl_trace_session_begin(&g_sim);
open_idle_range();
dispatch_of(&g, JOB_B);
cap_begin();
drain_one(&g.d);
out = cap_end();
TEST_ASSERT_TRUE(strncmp(out, flush, strlen(flush)) == 0);
repl_trace_session_end(&g_sim);
free(flush);
free(out);
}
void test_step_mode_round_flags_reset_at_next_begin(void)
{
fixture_decision_t f;
cmd("trace on all");
repl_trace_session_begin(&g_sim);
dispatch_of(&f, JOB_A);
cap_begin();
drain_one(&f.d);
free(cap_end());
TEST_ASSERT_TRUE(g_rs.trace.round_traced);
repl_trace_session_end(&g_sim);
repl_trace_session_begin(&g_sim);
TEST_ASSERT_FALSE(g_rs.trace.round_traced);
TEST_ASSERT_FALSE(g_rs.trace.round_flushed);
repl_trace_session_end(&g_sim);
}
/* ---- file copy (8.1) ------------------------------------------------------
*/
void test_file_copy_byte_equal_to_console(void)
{
fixture_decision_t t1;
fixture_decision_t t2;
fixture_decision_t f;
char* console;
char* file;
char* echo;
char* tr_f;
char* expect;
g_rs.units_mode = XCE_REPL_UNITS_HUMAN;
cmd("trace on all level full file a.trace");
repl_trace_session_begin(&g_sim);
/* trace-only records: the console shows exactly the trace */
trace_only_of(&t1, JOB_A);
trace_only_of(&t2, JOB_B);
cap_begin();
drain_one(&t1.d);
drain_one(&t2.d);
console = cap_end();
file = read_file("a.trace"); /* flushed after every record */
TEST_ASSERT_TRUE(strlen(console) > 0);
TEST_ASSERT_EQUAL_STRING(console, file);
free(file);
/* a decision: the console adds only the echo line */
dispatch_of(&f, JOB_A + 1);
echo = echo_line("DISPATCH", JOB_A + 1);
tr_f = render(f.d.trace);
free(console);
cap_begin();
drain_one(&f.d);
console = cap_end();
expect = concat3(echo, tr_f, "");
TEST_ASSERT_EQUAL_STRING(expect, console);
file = read_file("a.trace");
TEST_ASSERT_TRUE(strlen(file) >= strlen(tr_f));
TEST_ASSERT_EQUAL_STRING(tr_f, file + strlen(file) - strlen(tr_f));
repl_trace_session_end(&g_sim);
free(console);
free(file);
free(echo);
free(tr_f);
free(expect);
}
/* ---- error handling -------------------------------------------------------
*/
void test_file_write_error_reported_once_per_session(void)
{
fixture_decision_t t1;
fixture_decision_t t2;
fixture_decision_t t3;
char* out;
const char* msg = "[sim] trace: write to /dev/full failed: ";
char* p;
if (access("/dev/full", W_OK) != 0) {
TEST_IGNORE_MESSAGE("/dev/full is not available");
}
cmd("trace on all file /dev/full");
TEST_ASSERT_NOT_NULL(repl_trace_settings_file(g_rs.trace.next));
repl_trace_session_begin(&g_sim);
trace_only_of(&t1, JOB_A);
trace_only_of(&t2, JOB_B);
cap_begin();
drain_one(&t1.d);
drain_one(&t2.d);
out = cap_end();
p = strstr(out, msg);
TEST_ASSERT_NOT_NULL(p);
TEST_ASSERT_NULL(strstr(p + 1, msg)); /* once per session */
TEST_ASSERT_NOT_NULL(strstr(out, strerror(ENOSPC)));
TEST_ASSERT_NULL(t2.d.trace); /* console and store carry on */
free(out);
repl_trace_session_end(&g_sim);
repl_trace_session_begin(&g_sim);
trace_only_of(&t3, JOB_A);
cap_begin();
drain_one(&t3.d);
out = cap_end();
TEST_ASSERT_NOT_NULL(strstr(out, msg)); /* reported again next session */
free(out);
repl_trace_session_end(&g_sim);
}
/* A record larger than the stdio buffer fails inside the renderer's own
* writes; the reported cause is still the real one (ENOSPC). */
static void test_file_write_error_inside_render_reports_cause(void)
{
fixture_decision_t t;
char* out;
int i;
if (access("/dev/full", W_OK) != 0) {
TEST_IGNORE_MESSAGE("/dev/full is not available");
}
cmd("trace on all file /dev/full");
repl_trace_session_begin(&g_sim);
trace_only_of(&t, JOB_A);
for (i = 0; i < 900; i++) {
sched_trace_attempt_reasons(t.d.trace, NULL, 0, NULL, 0, NULL, 0);
}
cap_begin();
drain_one(&t.d);
out = cap_end();
TEST_ASSERT_NOT_NULL(strstr(out, "[sim] trace: write to /dev/full failed: "));
TEST_ASSERT_NOT_NULL(strstr(out, strerror(ENOSPC)));
free(out);
repl_trace_session_end(&g_sim);
}
int main(void)
{
UNITY_BEGIN();
RUN_TEST(test_session_begin_publishes_next_with_stamp);
RUN_TEST(test_session_begin_with_tracing_off_publishes_nothing);
RUN_TEST(test_command_during_session_applies_next_session);
RUN_TEST(test_replaced_file_kept_until_session_end);
RUN_TEST(test_step_mode_trace_follows_echo);
RUN_TEST(test_continue_mode_flush_echo_then_trace);
RUN_TEST(test_continue_mode_trace_only_first_flushes_without_echo);
RUN_TEST(test_flags_reset_after_round_ended_in_step_mode);
RUN_TEST(test_step_mode_round_flags_reset_at_next_begin);
RUN_TEST(test_file_copy_byte_equal_to_console);
RUN_TEST(test_file_write_error_inside_render_reports_cause);
RUN_TEST(test_file_write_error_reported_once_per_session);
return UNITY_END();
}
Reference in a new issue View git blame Copy permalink