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);