lei.chen/myxce
1
0
Fork 1
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

fix(scheduler): release replay-event payloads after the simulator pushes them (bug-1884, part 2)

Browse source 
sim_process_replay_event_message() now hands the decoded replay event to
the caller, which keeps it beside the translated core event and frees it
after xce_core_push_events() with driver.c's release_translated_payload()
rules: a pushed config is kept by the core, and a rusages report is kept
by the job it found. An untranslatable event is freed at once, and events
after the last session are freed at the end. The payloads were an
"accepted leak" per event; exit-time leak on the reproduction replay
75,495 -> 25,807 bytes.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
This commit is contained in:
Rongsong Shen 2026-09-29 08:55:34 +00:00
parent 880dc51007
commit c89e4e4520
Signed by untrusted user: rshen
GPG key ID: 80533536B0EB651C
4 changed files with 101 additions and 23 deletions
Download patch file Download diff file Expand all files Collapse all files

106
src/scheduler/src/sim_loop.c
View file

@ -363,16 +363,12 @@ static int replay_event_to_core_event(const replay_event_t* src,
* message IS the whole-stream termination marker (ReplayEvent_body_endSession,
* Req 5 AC 20) -- the caller must stop reading, not accumulate anything.
*
* Ownership: the decoded replay_event_t's payload is never freed here (nor
* by the caller) -- it is aliased directly into *out_ev's pointers, which
* must stay valid until xce_core_push_events() consumes the accumulated
* batch at the next ReplaySession boundary. Deliberately not tracked for
* later release: unlike driver.c's release_translated_payload() (which
* exists because the LIVE scheduler runs indefinitely and must not leak
* per live round), a `--sim` run processes one bounded fixture file and
* exits -- this mirrors sim.c's own already-established "accepted leak,
* detect_leaks=0 by design" precedent for the snapshot-load path
* (sim_locate_and_load_snapshot()'s doc comment).
* Ownership: *out_ev's pointers alias the decoded replay_event_t's payload,
* which must stay valid until xce_core_push_events() consumes the batch at
* the next ReplaySession boundary. On a return of 1 the decoded event is
* handed to the caller in *out_decoded, which releases it after the push
* with sim_release_pushed_event() (bug-1884); on any other return it is
* already freed and *out_decoded is zeroed.
*
* `*out_record_rc` (task 11.5, Req 5 AC 19) is 0 unless the --sim-output
* pass-through write fails, in which case it holds the write's own
@ -389,13 +385,16 @@ static int replay_event_to_core_event(const replay_event_t* src,
static int sim_process_replay_event_message(xce_sched_sim_driver_t* sim,
struct capn* c, capn_ptr root,
xce_core_event_t* out_ev,
replay_event_t* out_decoded,
int* out_record_rc)
{
ReplayEvent_ptr ep;
struct ReplayEvent wire;
replay_event_t decoded;
int translated;
*out_record_rc = 0;
memset(out_decoded, 0, sizeof(*out_decoded));
ep.p = root;
read_ReplayEvent(&wire, ep);
@ -420,7 +419,54 @@ static int sim_process_replay_event_message(xce_sched_sim_driver_t* sim,
sim_advance_virtual_clock(sim, decoded.ts);
return replay_event_to_core_event(&decoded, out_ev);
translated = replay_event_to_core_event(&decoded, out_ev);
if (translated > 0) {
*out_decoded = decoded; /* the caller releases it after the push */
} else {
free_ReplayEvent(&decoded);
}
return translated;
}
/*
* sim_release_pushed_event - bug-1884: frees a decoded replay event whose
* translation xce_core_push_events() has consumed, with the same two
* adoption exceptions as driver.c's release_translated_payload(): the core
* keeps a pushed config (core->config) and a job keeps a rusages report it
* found its job for. Every other payload is copied by the core (JOB_UPDATE
* MOVE_PTR()s the allocs out and leaves NULL behind). `applied` is false
* for an event whose batch was never pushed or whose push failed.
*/
static void sim_release_pushed_event(xce_sched_core_t* core,
replay_event_t* decoded, int applied)
{
if (applied && (decoded->body_which == ReplayEvent_body_configPush)) {
decoded->body.config_push = NULL;
}
if (applied && (decoded->body_which == ReplayEvent_body_jobRusage)
&& (decoded->body.job_rusage != NULL)) {
xce_sched_job_t* job
= sched_get_job(core, decoded->body.job_rusage->jobid);
if ((job != NULL) && (job->rusages == decoded->body.job_rusage)) {
decoded->body.job_rusage = NULL;
}
}
free_ReplayEvent(decoded);
}
static void sim_release_pushed_events(xce_sched_core_t* core,
replay_event_t* decoded, int n,
int applied)
{
int i;
for (i = 0; i < n; i++) {
sim_release_pushed_event(core, &(decoded[i]), applied);
}
}
/* sim_bootstrap_picker - Req 5 AC 17's "core.h-scoped plugin registration
@ -1335,9 +1381,10 @@ static void sim_reconcile_unaccounted_jobs(xce_sched_sim_driver_t* sim,
int sched_sim_driver_run(xce_sched_sim_driver_t* sim,
const sched_sim_args_t* args)
{
xce_core_event_t* buf_events = NULL;
int buf_cap = 0;
int buf_n = 0;
xce_core_event_t* buf_events = NULL;
replay_event_t* buf_decoded = NULL;
int buf_cap = 0;
int buf_n = 0;
int rc;
rc = sched_sim_driver_start(sim, args);
@ -1431,9 +1478,10 @@ int sched_sim_driver_run(xce_sched_sim_driver_t* sim,
if (msg_root_has_shape(&c, &root, ReplayEvent_word_count,
ReplayEvent_pointer_count)) {
xce_core_event_t ev;
replay_event_t decoded;
int record_rc = 0;
int translated = sim_process_replay_event_message(sim, &c, root,
&ev, &record_rc);
int translated = sim_process_replay_event_message(
sim, &c, root, &ev, &decoded, &record_rc);
if (record_rc < 0) {
ERROR("%s failed to write --sim-output pass-through event: "
@ -1443,6 +1491,8 @@ int sched_sim_driver_run(xce_sched_sim_driver_t* sim,
fclose(sim->record_file);
sim->record_file = NULL;
}
sim_release_pushed_events(&(sim->core), buf_decoded, buf_n, 0);
free(buf_decoded);
free(buf_events);
xce_core_set_active_for_logging(NULL);
return record_rc;
@ -1459,20 +1509,32 @@ int sched_sim_driver_run(xce_sched_sim_driver_t* sim,
int newcap = buf_cap ? (buf_cap * 2) : 16;
xce_core_event_t* tmp = (xce_core_event_t*)realloc(
buf_events, (size_t)newcap * sizeof(*tmp));
replay_event_t* dtmp = NULL;
if (tmp == NULL) {
if (tmp != NULL) {
buf_events = tmp;
dtmp = (replay_event_t*)realloc(
buf_decoded, (size_t)newcap * sizeof(*dtmp));
}
if (dtmp == NULL) {
ERROR("%s out of memory growing the accumulated "
"event batch (buf_n=%d)",
__func__, buf_n);
free_ReplayEvent(&decoded);
sim_release_pushed_events(&(sim->core), buf_decoded,
buf_n, 0);
free(buf_decoded);
free(buf_events);
xce_core_set_active_for_logging(NULL);
return -ENOMEM;
}
buf_events = tmp;
buf_cap = newcap;
buf_decoded = dtmp;
buf_cap = newcap;
}
buf_decoded[buf_n] = decoded;
buf_events[buf_n++] = ev;
}
@ -1525,6 +1587,8 @@ int sched_sim_driver_run(xce_sched_sim_driver_t* sim,
ERROR("%s xce_core_push_events failed rc=%d (buf_n=%d)",
__func__, push_rc, buf_n);
}
sim_release_pushed_events(&(sim->core), buf_decoded, buf_n,
push_rc >= 0);
buf_n = 0;
sim_advance_virtual_clock(sim, begin_round);
@ -1674,6 +1738,7 @@ int sched_sim_driver_run(xce_sched_sim_driver_t* sim,
__func__, (unsigned long long)begin_round, record_rc);
fclose(sim->record_file);
sim->record_file = NULL;
free(buf_decoded);
free(buf_events);
xce_core_set_active_for_logging(NULL);
return record_rc;
@ -1717,6 +1782,9 @@ int sched_sim_driver_run(xce_sched_sim_driver_t* sim,
sim->record_file = NULL;
}
/* events after the last session are never pushed */
sim_release_pushed_events(&(sim->core), buf_decoded, buf_n, 0);
free(buf_decoded);
free(buf_events);
xce_core_set_active_for_logging(NULL);

7
src/scheduler/testsuite/test_maxtasks_wiring_sim_driver_parity.c
View file

@ -571,6 +571,7 @@ test_sim_process_replay_event_jobs_push_wires_real_task_count_into_resreq_primar
struct capn c;
capn_ptr root;
xce_core_event_t translated_ev;
replay_event_t translated_decoded;
int record_rc = 0;
int translated;
xce_sched_job_t* sjob;
@ -631,7 +632,9 @@ test_sim_process_replay_event_jobs_push_wires_real_task_count_into_resreq_primar
* not a mirror of it. record_rc is always 0 here: sim.record_file is
* NULL for this whole fixture (no --sim-output involved). ────────── */
translated = sim_process_replay_event_message(&sim, &c, root,
&translated_ev, &record_rc);
&translated_ev,
&translated_decoded,
&record_rc);
fclose(f);
TEST_ASSERT_EQUAL_INT(0, record_rc);
@ -693,6 +696,8 @@ test_sim_process_replay_event_jobs_push_wires_real_task_count_into_resreq_primar
"a hand-built mirror of it, must have carried the job's real "
"maxTasks through to the dispatched standalone primary");
free_ReplayEvent(&translated_decoded);
printf("test_sim_process_replay_event_jobs_push_wires_real_task_count_"
"into_resreq_primary: PASS\n");
}

5
src/scheduler/testsuite/test_sim_discrete_clock.c
View file

@ -757,15 +757,18 @@ static void test_events_earlier_than_start_snapshot_set_exact(void)
struct capn c;
capn_ptr root;
xce_core_event_t ev;
replay_event_t decoded;
int record_rc = 0;
TEST_ASSERT_EQUAL_INT(0, capn_init_fp(&c, start_sim.replay_file, 0));
TEST_ASSERT_TRUE(msg_root_has_shape(&c, &root, ReplayEvent_word_count,
ReplayEvent_pointer_count));
TEST_ASSERT_TRUE(sim_process_replay_event_message(&start_sim, &c, root,
&ev, &record_rc)
&ev, &decoded,
&record_rc)
>= 0);
TEST_ASSERT_EQUAL_INT(0, record_rc);
free_ReplayEvent(&decoded);
usleep(WAIT_US / 5);
TEST_ASSERT_EQUAL_UINT64_MESSAGE(ev_ts[i],
xce_core_now(&(start_sim.core)),

6
src/scheduler/testsuite/test_sim_event_loop.c
View file

@ -720,6 +720,7 @@ test_process_replay_event_advances_virtual_clock_to_exact_event_ts_in_file_order
struct capn c;
capn_ptr root;
xce_core_event_t ev;
replay_event_t decoded;
int record_rc = 0;
int translated;
@ -731,8 +732,8 @@ test_process_replay_event_advances_virtual_clock_to_exact_event_ts_in_file_order
* (sim_loop.c) -- always 0 here, sim.record_file is NULL for this
* whole fixture (no --sim-output involved in this task's own
* Observable). */
translated
= sim_process_replay_event_message(&sim, &c, root, &ev, &record_rc);
translated = sim_process_replay_event_message(&sim, &c, root, &ev,
&decoded, &record_rc);
TEST_ASSERT_EQUAL_INT(0, record_rc);
TEST_ASSERT_EQUAL_INT(1, translated);
@ -745,6 +746,7 @@ test_process_replay_event_advances_virtual_clock_to_exact_event_ts_in_file_order
* (see this file's own header comment for why a picker-hook spy
* cannot see per-event granularity). */
TEST_ASSERT_EQUAL_UINT64(ts_values[i], xce_core_now(&(sim.core)));
free_ReplayEvent(&decoded);
}
fclose(f);