Merge branch 'share_host_ctx' into merge_api_server

This commit is contained in:
Rongsong Shen 2026-05-07 04:38:01 +00:00
commit c772d7e601
9 changed files with 123 additions and 58 deletions

View file

@ -109,8 +109,9 @@ extern int sel_get_optimize_level(void);
extern void sel_register_alloc_thread(void); extern void sel_register_alloc_thread(void);
/* Intern table lifecycle */ /* Intern table lifecycle */
extern void sel_node_table_init(void); extern void sel_node_table_init(void);
extern void sel_node_table_fini(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) */ /* Test-only reset (requires #define UNIT_TEST before this header) */
#ifdef UNIT_TEST #ifdef UNIT_TEST

View file

@ -497,6 +497,11 @@ void select_eval_ctx_fini(select_eval_ctx_t* ctx)
free(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) ── */ /* ── 2-bit bitmap helpers (EVAL_NONE=0 so memset(0) clears correctly) ── */
#define EVAL_NONE 0 #define EVAL_NONE 0
@ -735,48 +740,72 @@ static int eval_select_inner(const xce_select_node_t* node,
if (l < 0) return l; if (l < 0) return l;
return eval_select(node->u.binop.right, ctx); return eval_select(node->u.binop.right, ctx);
} }
case SEL_GT: case SEL_GT: {
eval_select(node->u.binop.left, ctx); int lr = eval_select(node->u.binop.left, ctx);
eval_select(node->u.binop.right, 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) > return ctx_num(ctx, node->u.binop.left) >
ctx_num(ctx, node->u.binop.right) ? 1 : 0; ctx_num(ctx, node->u.binop.right) ? 1 : 0;
case SEL_LT: }
eval_select(node->u.binop.left, ctx); case SEL_LT: {
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 ctx_num(ctx, node->u.binop.left) < return ctx_num(ctx, node->u.binop.left) <
ctx_num(ctx, node->u.binop.right) ? 1 : 0; ctx_num(ctx, node->u.binop.right) ? 1 : 0;
case SEL_GE: }
eval_select(node->u.binop.left, ctx); case SEL_GE: {
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 ctx_num(ctx, node->u.binop.left) >= return ctx_num(ctx, node->u.binop.left) >=
ctx_num(ctx, node->u.binop.right) ? 1 : 0; ctx_num(ctx, node->u.binop.right) ? 1 : 0;
case SEL_LE: }
eval_select(node->u.binop.left, ctx); case SEL_LE: {
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 ctx_num(ctx, node->u.binop.left) <= return ctx_num(ctx, node->u.binop.left) <=
ctx_num(ctx, node->u.binop.right) ? 1 : 0; ctx_num(ctx, node->u.binop.right) ? 1 : 0;
}
case SEL_EQ: case SEL_EQ:
if (is_str_cmp(node)) { if (is_str_cmp(node)) {
eval_select(node->u.binop.left, ctx); int lr = eval_select(node->u.binop.left, ctx);
eval_select(node->u.binop.right, 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), return xce_str_match(ctx_str(ctx, node->u.binop.left),
ctx_str(ctx, node->u.binop.right)); ctx_str(ctx, node->u.binop.right));
} }
eval_select(node->u.binop.left, ctx); {
eval_select(node->u.binop.right, ctx); int lr = eval_select(node->u.binop.left, ctx);
return ctx_num(ctx, node->u.binop.left) == int rr = eval_select(node->u.binop.right, ctx);
ctx_num(ctx, node->u.binop.right) ? 1 : 0; 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: case SEL_NE:
if (is_str_cmp(node)) { if (is_str_cmp(node)) {
eval_select(node->u.binop.left, ctx); int lr = eval_select(node->u.binop.left, ctx);
eval_select(node->u.binop.right, 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), return !xce_str_match(ctx_str(ctx, node->u.binop.left),
ctx_str(ctx, node->u.binop.right)) ? 1 : 0; ctx_str(ctx, node->u.binop.right)) ? 1 : 0;
} }
eval_select(node->u.binop.left, ctx); {
eval_select(node->u.binop.right, ctx); int lr = eval_select(node->u.binop.left, ctx);
return ctx_num(ctx, node->u.binop.left) != int rr = eval_select(node->u.binop.right, ctx);
ctx_num(ctx, node->u.binop.right) ? 1 : 0; 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); ERROR("eval_select_inner: unknown node type %d", (int)node->type);
abort(); 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) { if (sel_node_table.table == NULL) {
return; 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; int caps = sel_node_table.nodes.caps;

View file

@ -96,11 +96,12 @@ typedef struct xce_sched_job_ {
} xce_sched_job_t; } xce_sched_job_t;
typedef struct { typedef struct {
char* name; char* name;
int idx; int idx;
uint32_t status; uint32_t status;
zarray_t resvals; zarray_t resvals;
zarray_t resInUse; zarray_t resInUse;
host_eval_ctx_t* host_eval_ctx; /* cached per-session; NULL until first use */
} xce_sched_worker_t; } xce_sched_worker_t;
typedef struct { typedef struct {

View file

@ -84,9 +84,10 @@ typedef struct {
cbuf_t wbuf; /* action to trigger scheduling */ cbuf_t wbuf; /* action to trigger scheduling */
cbuf_t jbuf; /* job picked for allocation */ cbuf_t jbuf; /* job picked for allocation */
cbuf_t decisions; /* scheduler decisions */ cbuf_t decisions; /* scheduler decisions */
_Atomic int session_active; /* 1 during scheduling session, 0 between */ _Atomic int session_active; /* 1 during scheduling session, 0 between */
_Atomic int eval_in_progress; /* 1 while eval_select loop runs */ _Atomic int eval_in_progress; /* 1 while eval_select loop runs */
_Atomic int staging_merge_complete; /* 1 after staging merge completes */ _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; } xce_scheduler_t;
extern int sched_init(xce_scheduler_t*, xce_client_config_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_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_pend_reason(xce_scheduler_t*, xce_preason_set_t*);
extern int sched_publish_dispatch(xce_scheduler_t*, xce_job_dispatch_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*, extern int sched_meet_host_selection(xce_scheduler_t*, xce_sched_worker_t*,
xce_select_t*); xce_select_t*);
extern int sched_check_shared_satisfy(xce_scheduler_t*, xvec_t*, xvec_idval_t*, extern int sched_check_shared_satisfy(xce_scheduler_t*, xvec_t*, xvec_idval_t*,

View file

@ -117,9 +117,6 @@ int main(int argc, char* argv[])
INFO("xce scheduler is running"); INFO("xce scheduler is running");
/* Register this thread as the sole eval/GC allocator thread (write-once) */
sel_register_alloc_thread();
while (1) { while (1) {
sched_update(&sched); sched_update(&sched);
sched_begin_session(&sched); sched_begin_session(&sched);

View file

@ -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; xce_sched_obj_t* schedobj = ctx->schedobj;
int rc; int rc;
xlist_t workers; xlist_t workers;
xce_pendreason_t* reason = NULL; xce_pendreason_t* reason = ctx->reason;
xvec_t* uncomit = NULL; xvec_t* uncomit = NULL;
xce_sched_worker_t* w = NULL; xce_sched_worker_t* w = NULL;
reason = make_pendreason(0);
ctx->reason = reason;
if (reason == NULL) { 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); rc = sched_ctx_get_uncomit_shared(&uncomit, ctx);

View file

@ -177,25 +177,40 @@ free_host_eval_ctx(host_eval_ctx_t* ctx)
free(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, int sched_meet_host_selection(xce_scheduler_t* sched, xce_sched_worker_t* w,
xce_select_t* s) xce_select_t* s)
{ {
(void)sched;
if (s == NULL || s->root == NULL) return 1; if (s == NULL || s->root == NULL) return 1;
int num_res = (int)w->resvals.caps; /* Build host context lazily — reused for all jobs against this host. */
if (w->host_eval_ctx == NULL) {
host_eval_ctx_t* hctx = build_host_eval_ctx(w, num_res); int num_res = (int)w->resvals.caps;
if (hctx == NULL) { w->host_eval_ctx = build_host_eval_ctx(w, num_res);
/* OOM: conservatively pass host */ if (w->host_eval_ctx == NULL) {
return 1; /* OOM: conservatively pass host */
return 1;
}
} }
int rc = eval_select_simple(s->root, hctx); /* Fast path: use session-level memoisation context when available. */
free_host_eval_ctx(hctx); 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; 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); rc = sched_meet_host_selection(sched, w, primary->select);
if (rc == 0) { 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); NULL);
return 0; return 0;
@ -464,6 +487,7 @@ static int sched_try_alloc_on_host_(xce_sched_worker_t* w, va_list ap)
if (rc == 0) { if (rc == 0) {
int i; int i;
if (ctx)
for (i = 0; i < ctx->unmeet.count; i++) { for (i = 0; i < ctx->unmeet.count; i++) {
xce_resource_t* r = NULL; xce_resource_t* r = NULL;

View file

@ -234,6 +234,7 @@ void kill_sched_worker(void* arg)
if (arg == NULL) { if (arg == NULL) {
return; return;
} }
sched_worker_release_host_ctx((xce_sched_worker_t*)arg);
} }
void kill_sched_resval(xce_val_t* val) void kill_sched_resval(xce_val_t* val)

View file

@ -220,6 +220,7 @@ static void* sched_allocator(void* arg)
{ {
xce_scheduler_t* sched = (xce_scheduler_t*)arg; xce_scheduler_t* sched = (xce_scheduler_t*)arg;
sel_register_alloc_thread();
xce_use(sched); xce_use(sched);
while (1) { while (1) {
uint64_t jobid; uint64_t jobid;
@ -694,6 +695,9 @@ static int sched_reset_worker_(xce_scheduler_t* sched, xce_sched_worker_t* w)
return 0; 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_, return zarray_iter(&(w->resInUse), (zarray_iter_fn_t)my_reset_resval_,
sched); 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->session_active, 1, memory_order_release);
atomic_store_explicit(&sched->staging_merge_complete, 0, 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); plugin_begin_session(sched);
sched_reset_shared_resval(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); 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); plugin_end_session(sched);
/* Guard: GC requires the evaluation pipeline to be idle */ /* Guard: GC requires the evaluation pipeline to be idle */