From 5267d36a8436900461cf9aaeed2ee18b6b9b9f3b Mon Sep 17 00:00:00 2001 From: Rongsong Shen Date: Thu, 30 Apr 2026 05:44:46 +0000 Subject: [PATCH] =?UTF-8?q?feat:=20drive=20GC=20from=20sched=5Fend=5Fsessi?= =?UTF-8?q?on=20=E2=80=94=20live=5Fjobs=20mark,=20seq=5Fcst=20fence,=20gc?= =?UTF-8?q?=5Fsweep=20(6-8,=206-12)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- src/scheduler/src/scheduler.c | 97 +++++++++++++++++++++++++++++++++- src/scheduler/testsuite/test.c | 25 +++++++++ 2 files changed, 121 insertions(+), 1 deletion(-) diff --git a/src/scheduler/src/scheduler.c b/src/scheduler/src/scheduler.c index f5d4dd50..7214ca13 100644 --- a/src/scheduler/src/scheduler.c +++ b/src/scheduler/src/scheduler.c @@ -760,6 +760,9 @@ static int sched_account_jobs(xce_scheduler_t* sched) void sched_begin_session(xce_scheduler_t* sched) { + atomic_store_explicit(&sched->session_active, 1, memory_order_release); + atomic_store_explicit(&sched->staging_merge_complete, 0, memory_order_release); + plugin_begin_session(sched); sched_reset_shared_resval(sched); @@ -772,9 +775,101 @@ void sched_begin_session(xce_scheduler_t* sched) void sched_end_session(xce_scheduler_t* sched) { + atomic_store_explicit(&sched->session_active, 0, memory_order_release); + plugin_end_session(sched); - return; + /* Guard: GC requires the evaluation pipeline to be idle */ + if (sched_eval_in_progress(sched) || !sched_staging_merge_complete(sched)) { + TRACE("%s: GC skipped — eval_in_progress=%d staging_merge_complete=%d", + __func__, + sched_eval_in_progress(sched), + sched_staging_merge_complete(sched)); + return; + } + + /* Collect live jobs into a flat array for the GC mark phase */ + iterator_t it; + xce_sched_job_t* job = NULL; + int njobs = 0; + int cap = 64; + xce_sched_job_t** live_jobs = calloc(cap, sizeof(*live_jobs)); + + if (live_jobs != NULL) { + int rc = iter_make(&it, sched->jobs); + if (rc >= 0) { + for (;;) { + rc = iter_next(&it, (void**)&job); + if ((rc < 0) || (rc == RTTI_ITER_NO_DATA)) { + break; + } + if (njobs >= cap) { + int newcap = cap * 2; + xce_sched_job_t** tmp = realloc(live_jobs, + newcap * sizeof(*live_jobs)); + if (tmp == NULL) { + break; + } + live_jobs = tmp; + cap = newcap; + } + live_jobs[njobs++] = job; + } + if (rc == RTTI_ITER_NO_DATA) { + rc = 0; + } + iter_destroy(&it); + } + } + + /* Mark phase */ + gc_clear_marks(); + for (int i = 0; i < njobs; i++) { + xce_resreq_t* rr = live_jobs[i]->resreq; + if (rr == NULL) { + continue; + } + if (rr->tag == XCE_RESREQ_PRIMARY) { + xce_primary_resreq_t* pr = rr->u.primary; + if (pr && pr->select) { + gc_mark_select_tree(pr->select->root); + } + } else if (rr->tag == XCE_RESREQ_COMPOUND) { + xce_compound_resreq_t* cp = rr->u.compound; + if (cp != NULL) { + for (int j = 0; j < cp->primaries.count; j++) { + xce_primary_resreq_t* pr = xlist_get(&cp->primaries, j); + if (pr && pr->select) { + gc_mark_select_tree(pr->select->root); + } + } + } + } else if (rr->tag == XCE_RESREQ_ALTER) { + xce_alter_resreq_t* alt = rr->u.alter; + if (alt != NULL) { + for (int j = 0; j < alt->compounds.count; j++) { + xce_compound_resreq_t* cp = xlist_get(&alt->compounds, j); + if (cp == NULL) { + continue; + } + for (int k = 0; k < cp->primaries.count; k++) { + xce_primary_resreq_t* pr = xlist_get(&cp->primaries, k); + if (pr && pr->select) { + gc_mark_select_tree(pr->select->root); + } + } + } + } + } + } + + /* Fence: mark writes visible before sweep reads them */ + atomic_thread_fence(memory_order_seq_cst); + + gc_sweep_select_nodes(NULL, 0); + free(live_jobs); + + atomic_store_explicit(&sched->staging_merge_complete, 1, memory_order_release); } int sched_add_job(xce_scheduler_t* sched, xce_sched_job_t* j) diff --git a/src/scheduler/testsuite/test.c b/src/scheduler/testsuite/test.c index 81085ff1..817c13a2 100644 --- a/src/scheduler/testsuite/test.c +++ b/src/scheduler/testsuite/test.c @@ -196,6 +196,30 @@ test_meet_host_selection_always_false_fails(void) sel_optimize_test_reset(); } +static void +test_gc_predicates_idle_invariant(void) +{ + xce_scheduler_t sched; + memset(&sched, 0, sizeof(sched)); + /* staging_merge_complete=1 means GC is safe to run */ + atomic_store(&sched.staging_merge_complete, 1); + + sel_optimize_set(1); + sel_node_table_init(); + + /* Verify the atomic fields are readable — testing the predicate contract */ + TEST_ASSERT_EQUAL_INT(0, atomic_load(&sched.session_active)); + TEST_ASSERT_EQUAL_INT(0, atomic_load(&sched.eval_in_progress)); + TEST_ASSERT_EQUAL_INT(1, atomic_load(&sched.staging_merge_complete)); + + /* GC operations on an empty node table must not crash */ + gc_clear_marks(); + gc_sweep_select_nodes(NULL, 0); + + sel_node_table_fini(); + sel_optimize_test_reset(); +} + #define TEST(x) \ do { \ if (check_test(&arguments, #x)) { \ @@ -215,6 +239,7 @@ int main(int argc, char* argv[]) TEST(test_meet_host_selection_null_select_passes); TEST(test_meet_host_selection_always_true_passes); TEST(test_meet_host_selection_always_false_fails); + TEST(test_gc_predicates_idle_invariant); return UNITY_END(); }