From c89e4e45209e11eb65caa5d8c555d7a7ab2aeef3 Mon Sep 17 00:00:00 2001 From: Rongsong Shen Date: Tue, 29 Sep 2026 08:55:34 +0000 Subject: [PATCH] fix(scheduler): release replay-event payloads after the simulator pushes them (bug-1884, part 2) 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) --- src/scheduler/src/sim_loop.c | 106 ++++++++++++++---- .../test_maxtasks_wiring_sim_driver_parity.c | 7 +- .../testsuite/test_sim_discrete_clock.c | 5 +- src/scheduler/testsuite/test_sim_event_loop.c | 6 +- 4 files changed, 101 insertions(+), 23 deletions(-) diff --git a/src/scheduler/src/sim_loop.c b/src/scheduler/src/sim_loop.c index 0becda34..42e4515d 100644 --- a/src/scheduler/src/sim_loop.c +++ b/src/scheduler/src/sim_loop.c @@ -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); diff --git a/src/scheduler/testsuite/test_maxtasks_wiring_sim_driver_parity.c b/src/scheduler/testsuite/test_maxtasks_wiring_sim_driver_parity.c index 492dd178..367b7cdd 100644 --- a/src/scheduler/testsuite/test_maxtasks_wiring_sim_driver_parity.c +++ b/src/scheduler/testsuite/test_maxtasks_wiring_sim_driver_parity.c @@ -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"); } diff --git a/src/scheduler/testsuite/test_sim_discrete_clock.c b/src/scheduler/testsuite/test_sim_discrete_clock.c index ddf14ae9..dd13252d 100644 --- a/src/scheduler/testsuite/test_sim_discrete_clock.c +++ b/src/scheduler/testsuite/test_sim_discrete_clock.c @@ -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)), diff --git a/src/scheduler/testsuite/test_sim_event_loop.c b/src/scheduler/testsuite/test_sim_event_loop.c index e6813325..43b42a2b 100644 --- a/src/scheduler/testsuite/test_sim_event_loop.c +++ b/src/scheduler/testsuite/test_sim_event_loop.c @@ -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);