lei.chen/myxce
1
0
Fork 1
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

fix select eval ctx share

Browse source
This commit is contained in:
Rongsong Shen 2026-05-07 04:37:18 +00:00
parent 3ed7ef011a
commit 52d67726ec
9 changed files with 123 additions and 58 deletions
Download patch file Download diff file Expand all files Collapse all files

5
src/include/resreq.h
View file

@ -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

82
src/lib/xce_select.c
View file

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

11
src/scheduler/include/sched/types.h
View file

@ -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 {

8
src/scheduler/include/scheduler.h
View file

@ -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*,

3
src/scheduler/src/main.c
View file

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

10
src/scheduler/src/plugins/seqjob_allocator.c
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;
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);

48
src/scheduler/src/resource.c
View file

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

1
src/scheduler/src/schedobj.c
View file

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

13
src/scheduler/src/scheduler.c
View file

@ -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 */