From 52d67726ec78562c75eea727595fc5874423bb92 Mon Sep 17 00:00:00 2001 From: Rongsong Shen Date: Thu, 7 May 2026 04:37:18 +0000 Subject: [PATCH] fix select eval ctx share --- src/include/resreq.h | 5 +- src/lib/xce_select.c | 82 +++++++++++++------- src/scheduler/include/sched/types.h | 11 +-- src/scheduler/include/scheduler.h | 8 +- src/scheduler/src/main.c | 3 - src/scheduler/src/plugins/seqjob_allocator.c | 10 ++- src/scheduler/src/resource.c | 48 +++++++++--- src/scheduler/src/schedobj.c | 1 + src/scheduler/src/scheduler.c | 13 ++++ 9 files changed, 123 insertions(+), 58 deletions(-) diff --git a/src/include/resreq.h b/src/include/resreq.h index b40a6801..4943957c 100644 --- a/src/include/resreq.h +++ b/src/include/resreq.h @@ -109,8 +109,9 @@ extern int sel_get_optimize_level(void); extern void sel_register_alloc_thread(void); /* Intern table lifecycle */ -extern void sel_node_table_init(void); -extern void sel_node_table_fini(void); +extern void sel_node_table_init(void); +extern void sel_node_table_fini(void); +extern uint32_t sel_get_node_count(void); /* Test-only reset (requires #define UNIT_TEST before this header) */ #ifdef UNIT_TEST diff --git a/src/lib/xce_select.c b/src/lib/xce_select.c index f8b5f39d..39b907ac 100644 --- a/src/lib/xce_select.c +++ b/src/lib/xce_select.c @@ -497,6 +497,11 @@ void select_eval_ctx_fini(select_eval_ctx_t* ctx) free(ctx); } +uint32_t sel_get_node_count(void) +{ + return atomic_load_explicit(&sel_node_next_id, memory_order_acquire); +} + /* ── 2-bit bitmap helpers (EVAL_NONE=0 so memset(0) clears correctly) ── */ #define EVAL_NONE 0 @@ -735,48 +740,72 @@ static int eval_select_inner(const xce_select_node_t* node, if (l < 0) return l; return eval_select(node->u.binop.right, ctx); } - case SEL_GT: - eval_select(node->u.binop.left, ctx); - eval_select(node->u.binop.right, ctx); + case SEL_GT: { + int lr = eval_select(node->u.binop.left, ctx); + int rr = eval_select(node->u.binop.right, ctx); + if (lr < 0) return lr; + if (rr < 0) return rr; return ctx_num(ctx, node->u.binop.left) > ctx_num(ctx, node->u.binop.right) ? 1 : 0; - case SEL_LT: - eval_select(node->u.binop.left, ctx); - eval_select(node->u.binop.right, ctx); + } + case SEL_LT: { + int lr = eval_select(node->u.binop.left, ctx); + int rr = eval_select(node->u.binop.right, ctx); + if (lr < 0) return lr; + if (rr < 0) return rr; return ctx_num(ctx, node->u.binop.left) < ctx_num(ctx, node->u.binop.right) ? 1 : 0; - case SEL_GE: - eval_select(node->u.binop.left, ctx); - eval_select(node->u.binop.right, ctx); + } + case SEL_GE: { + int lr = eval_select(node->u.binop.left, ctx); + int rr = eval_select(node->u.binop.right, ctx); + if (lr < 0) return lr; + if (rr < 0) return rr; return ctx_num(ctx, node->u.binop.left) >= ctx_num(ctx, node->u.binop.right) ? 1 : 0; - case SEL_LE: - eval_select(node->u.binop.left, ctx); - eval_select(node->u.binop.right, ctx); + } + case SEL_LE: { + int lr = eval_select(node->u.binop.left, ctx); + int rr = eval_select(node->u.binop.right, ctx); + if (lr < 0) return lr; + if (rr < 0) return rr; return ctx_num(ctx, node->u.binop.left) <= ctx_num(ctx, node->u.binop.right) ? 1 : 0; + } case SEL_EQ: if (is_str_cmp(node)) { - eval_select(node->u.binop.left, ctx); - eval_select(node->u.binop.right, ctx); + int lr = eval_select(node->u.binop.left, ctx); + int rr = eval_select(node->u.binop.right, ctx); + if (lr < 0) return lr; + if (rr < 0) return rr; return xce_str_match(ctx_str(ctx, node->u.binop.left), ctx_str(ctx, node->u.binop.right)); } - eval_select(node->u.binop.left, ctx); - eval_select(node->u.binop.right, ctx); - return ctx_num(ctx, node->u.binop.left) == - ctx_num(ctx, node->u.binop.right) ? 1 : 0; + { + int lr = eval_select(node->u.binop.left, ctx); + int rr = eval_select(node->u.binop.right, ctx); + if (lr < 0) return lr; + if (rr < 0) return rr; + return ctx_num(ctx, node->u.binop.left) == + ctx_num(ctx, node->u.binop.right) ? 1 : 0; + } case SEL_NE: if (is_str_cmp(node)) { - eval_select(node->u.binop.left, ctx); - eval_select(node->u.binop.right, ctx); + int lr = eval_select(node->u.binop.left, ctx); + int rr = eval_select(node->u.binop.right, ctx); + if (lr < 0) return lr; + if (rr < 0) return rr; return !xce_str_match(ctx_str(ctx, node->u.binop.left), ctx_str(ctx, node->u.binop.right)) ? 1 : 0; } - eval_select(node->u.binop.left, ctx); - eval_select(node->u.binop.right, ctx); - return ctx_num(ctx, node->u.binop.left) != - ctx_num(ctx, node->u.binop.right) ? 1 : 0; + { + int lr = eval_select(node->u.binop.left, ctx); + int rr = eval_select(node->u.binop.right, ctx); + if (lr < 0) return lr; + if (rr < 0) return rr; + return ctx_num(ctx, node->u.binop.left) != + ctx_num(ctx, node->u.binop.right) ? 1 : 0; + } } ERROR("eval_select_inner: unknown node type %d", (int)node->type); abort(); @@ -879,11 +908,6 @@ void gc_sweep_select_nodes(const xce_sched_job_t** live_jobs, size_t njobs) if (sel_node_table.table == NULL) { return; } - if (atomic_load_explicit(&sel_alloc_thread_registered, memory_order_acquire) - && !pthread_equal(pthread_self(), sel_alloc_thread_id)) { - ERROR("gc_sweep_select_nodes called from non-allocator thread"); - return; - } int caps = sel_node_table.nodes.caps; diff --git a/src/scheduler/include/sched/types.h b/src/scheduler/include/sched/types.h index 7af23b7e..2cdf6773 100644 --- a/src/scheduler/include/sched/types.h +++ b/src/scheduler/include/sched/types.h @@ -96,11 +96,12 @@ typedef struct xce_sched_job_ { } xce_sched_job_t; typedef struct { - char* name; - int idx; - uint32_t status; - zarray_t resvals; - zarray_t resInUse; + char* name; + int idx; + uint32_t status; + zarray_t resvals; + zarray_t resInUse; + host_eval_ctx_t* host_eval_ctx; /* cached per-session; NULL until first use */ } xce_sched_worker_t; typedef struct { diff --git a/src/scheduler/include/scheduler.h b/src/scheduler/include/scheduler.h index 3c35f2f6..f0fc3a54 100644 --- a/src/scheduler/include/scheduler.h +++ b/src/scheduler/include/scheduler.h @@ -84,9 +84,10 @@ typedef struct { cbuf_t wbuf; /* action to trigger scheduling */ cbuf_t jbuf; /* job picked for allocation */ cbuf_t decisions; /* scheduler decisions */ - _Atomic int session_active; /* 1 during scheduling session, 0 between */ - _Atomic int eval_in_progress; /* 1 while eval_select loop runs */ - _Atomic int staging_merge_complete; /* 1 after staging merge completes */ + _Atomic int session_active; /* 1 during scheduling session, 0 between */ + _Atomic int eval_in_progress; /* 1 while eval_select loop runs */ + _Atomic int staging_merge_complete; /* 1 after staging merge completes */ + select_eval_ctx_t* sel_eval_ctx; /* session-level memoisation context; NULL between sessions */ } xce_scheduler_t; extern int sched_init(xce_scheduler_t*, xce_client_config_t*, @@ -112,6 +113,7 @@ extern int sched_queue_is_ready(xce_sched_queue_t*); extern int sched_commit_decision(xce_scheduler_t*, xce_decision_t*); extern int sched_publish_pend_reason(xce_scheduler_t*, xce_preason_set_t*); extern int sched_publish_dispatch(xce_scheduler_t*, xce_job_dispatch_t*); +extern void sched_worker_release_host_ctx(xce_sched_worker_t*); extern int sched_meet_host_selection(xce_scheduler_t*, xce_sched_worker_t*, xce_select_t*); extern int sched_check_shared_satisfy(xce_scheduler_t*, xvec_t*, xvec_idval_t*, diff --git a/src/scheduler/src/main.c b/src/scheduler/src/main.c index 29d4bb36..c4da571f 100644 --- a/src/scheduler/src/main.c +++ b/src/scheduler/src/main.c @@ -117,9 +117,6 @@ int main(int argc, char* argv[]) INFO("xce scheduler is running"); - /* Register this thread as the sole eval/GC allocator thread (write-once) */ - sel_register_alloc_thread(); - while (1) { sched_update(&sched); sched_begin_session(&sched); diff --git a/src/scheduler/src/plugins/seqjob_allocator.c b/src/scheduler/src/plugins/seqjob_allocator.c index 16ab6cef..4d3e1018 100644 --- a/src/scheduler/src/plugins/seqjob_allocator.c +++ b/src/scheduler/src/plugins/seqjob_allocator.c @@ -110,14 +110,16 @@ static int my_alloc(xce_sched_plugin_t* plugin, xce_sched_ctx_t* ctx) xce_sched_obj_t* schedobj = ctx->schedobj; int rc; xlist_t workers; - xce_pendreason_t* reason = NULL; + xce_pendreason_t* reason = ctx->reason; xvec_t* uncomit = NULL; xce_sched_worker_t* w = NULL; - reason = make_pendreason(0); - ctx->reason = reason; if (reason == NULL) { - return -ENOMEM; + reason = make_pendreason(0); + ctx->reason = reason; + if (reason == NULL) { + return -ENOMEM; + } } rc = sched_ctx_get_uncomit_shared(&uncomit, ctx); diff --git a/src/scheduler/src/resource.c b/src/scheduler/src/resource.c index e0d9f418..aa773911 100644 --- a/src/scheduler/src/resource.c +++ b/src/scheduler/src/resource.c @@ -177,25 +177,40 @@ free_host_eval_ctx(host_eval_ctx_t* ctx) free(ctx); } +void sched_worker_release_host_ctx(xce_sched_worker_t* w) +{ + if (w == NULL) return; + free_host_eval_ctx(w->host_eval_ctx); + w->host_eval_ctx = NULL; +} + int sched_meet_host_selection(xce_scheduler_t* sched, xce_sched_worker_t* w, xce_select_t* s) { - (void)sched; - if (s == NULL || s->root == NULL) return 1; - int num_res = (int)w->resvals.caps; - - host_eval_ctx_t* hctx = build_host_eval_ctx(w, num_res); - if (hctx == NULL) { - /* OOM: conservatively pass host */ - return 1; + /* Build host context lazily — reused for all jobs against this host. */ + if (w->host_eval_ctx == NULL) { + int num_res = (int)w->resvals.caps; + w->host_eval_ctx = build_host_eval_ctx(w, num_res); + if (w->host_eval_ctx == NULL) { + /* OOM: conservatively pass host */ + return 1; + } } - int rc = eval_select_simple(s->root, hctx); - free_host_eval_ctx(hctx); + /* Fast path: use session-level memoisation context when available. */ + if (sched->sel_eval_ctx != NULL) { + /* Reset memoisation bitmap only when moving to a different host. */ + if (sched->sel_eval_ctx->host != w->host_eval_ctx) + select_eval_ctx_reset(sched->sel_eval_ctx, w->host_eval_ctx); - /* eval_select_simple: 1=match, 0=no-match, <0=error (treat as no-match) */ + int rc = eval_select(s->root, sched->sel_eval_ctx); + return (rc > 0) ? 1 : 0; + } + + /* Fallback (no session context — e.g. unit tests or OOM at session start). */ + int rc = eval_select_simple(s->root, w->host_eval_ctx); return (rc > 0) ? 1 : 0; } @@ -435,7 +450,15 @@ static int sched_try_alloc_on_host_(xce_sched_worker_t* w, va_list ap) rc = sched_meet_host_selection(sched, w, primary->select); if (rc == 0) { - add_job_preason(schedobj->reason->reasons, PREASON_MATCH_FAIL, w->name, + if (ctx->reason == NULL) { + ctx->reason = make_pendreason(0); + } + + if (ctx->reason == NULL) { + return -errno; + } + + add_job_preason(ctx->reason->reasons, PREASON_MATCH_FAIL, w->name, NULL); return 0; @@ -464,6 +487,7 @@ static int sched_try_alloc_on_host_(xce_sched_worker_t* w, va_list ap) if (rc == 0) { int i; + if (ctx) for (i = 0; i < ctx->unmeet.count; i++) { xce_resource_t* r = NULL; diff --git a/src/scheduler/src/schedobj.c b/src/scheduler/src/schedobj.c index 3ab21359..253ba463 100644 --- a/src/scheduler/src/schedobj.c +++ b/src/scheduler/src/schedobj.c @@ -234,6 +234,7 @@ void kill_sched_worker(void* arg) if (arg == NULL) { return; } + sched_worker_release_host_ctx((xce_sched_worker_t*)arg); } void kill_sched_resval(xce_val_t* val) diff --git a/src/scheduler/src/scheduler.c b/src/scheduler/src/scheduler.c index a6fba60d..4749860f 100644 --- a/src/scheduler/src/scheduler.c +++ b/src/scheduler/src/scheduler.c @@ -220,6 +220,7 @@ static void* sched_allocator(void* arg) { xce_scheduler_t* sched = (xce_scheduler_t*)arg; + sel_register_alloc_thread(); xce_use(sched); while (1) { uint64_t jobid; @@ -694,6 +695,9 @@ static int sched_reset_worker_(xce_scheduler_t* sched, xce_sched_worker_t* w) return 0; } + /* Release the per-session host snapshot so the next session rebuilds it. */ + sched_worker_release_host_ctx(w); + return zarray_iter(&(w->resInUse), (zarray_iter_fn_t)my_reset_resval_, sched); } @@ -769,6 +773,12 @@ 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); + /* Allocate a fresh memoisation context sized to the current interned-node + * population. sched_reset_workers (called below) frees each worker's + * stale host_eval_ctx so they are rebuilt lazily during this session. */ + select_eval_ctx_fini(sched->sel_eval_ctx); + sched->sel_eval_ctx = select_eval_ctx_init(sel_get_node_count()); + plugin_begin_session(sched); sched_reset_shared_resval(sched); @@ -785,6 +795,9 @@ void sched_end_session(xce_scheduler_t* sched) { atomic_store_explicit(&sched->session_active, 0, memory_order_release); + select_eval_ctx_fini(sched->sel_eval_ctx); + sched->sel_eval_ctx = NULL; + plugin_end_session(sched); /* Guard: GC requires the evaluation pipeline to be idle */