fix(scheduler): bug-1513 — rejoin crash root-caused to 3 rejoin re-push bugs, not a lib race

The "libxce reconnect heap corruption" (ASAN bad-free, free(): invalid
pointer) was NOT a libxce bug — xce_ctx_reconnect is fine. The crashes are
in the scheduler's rejoin state re-sync path (config/dataset re-push),
fixed with the reconnect/rejoin feature now integrated:

1. group.c makeGroup refresh: FREEUP(m->u.name) raw-freed shared interned
   xstrings (makeString is a global refcounted table); the correct release
   is dereference(), matching group_killer. → dereference().
2. limit.c limit_update_config: after key_add returns -EEXIST (rejoin
   re-push — limit name already present) it called kill_limit(limit), but
   xkvmgr_add's vclear already freed the limit via the kvmgr datakiller
   (=kill_limit). → skip on -EEXIST (idempotent re-push).
3. handler.c/scheduler.c: the config push was applied inline on the vcpu
   thread (sched_init_config adds queues/plugin queues) while job updates
   flow through pend_updates on the main loop — concurrent rejoin re-push
   let fcfs_update_job see a NULL j->queue (SIGSEGV). Route the config
   push through pend_updates (XCE_SCHED_OP_UPDATE_CONFIG, killer case
   already existed) and process config requests FIRST in sched_update via
   a staging queue so jobs always see their queues/resources.

Verified: 295 lib unit tests pass under ASAN (incl. 2 new group-refresh
regression tests), scheduler survives 3 consecutive controller restarts
(each rejoin ok), full job lifecycle (submit→dispatch→DONE) on the
cluster. Also fixed stale build deps (libxce_deps missing librtti_dep,
scheduler_deps missing ssl_dep/c_capnproto_dep, unity include path).

Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
lei.chen 2026-08-14 07:29:32 +00:00
parent 2044dedb2f
commit 7f8c07ba85
10 changed files with 253 additions and 27 deletions

View file

@ -117,6 +117,8 @@ typedef struct {
_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 needs_rejoin; /* 1 when the lib reconnected and the
* scheduler session must be re-joined */
select_eval_ctx_t* sel_eval_ctx; /* session-level memoisation context; NULL between sessions */
xce_cluster_config_t cluster_cfg; /* HA candidate list — empty means single-host mode */
} xce_scheduler_t;
@ -129,6 +131,7 @@ extern int sched_update(xce_scheduler_t*);
extern void sched_go(xce_scheduler_t*);
extern void sched_begin_session(xce_scheduler_t*);
extern void sched_end_session(xce_scheduler_t*);
extern int sched_rejoin_if_needed(xce_scheduler_t*);
extern xce_sched_job_t* sched_get_job(xce_scheduler_t*, uint64_t);
extern int sched_requeue_pend_job(xce_scheduler_t*, uint64_t);
extern xce_sched_queue_t* sched_get_queue(xce_scheduler_t*, char*);

View file

@ -22,7 +22,9 @@ scheduler_deps = [
librtti_dep,
math_dep,
tinylog_dep,
thread_dep
thread_dep,
ssl_dep,
c_capnproto_dep
]
scheduler = executable('xce_scheduler',

View file

@ -592,17 +592,33 @@ int scheduler_req_handler(xce_request_t* request, void* args)
switch (request->opcode) {
case OPCode_push:
switch (request->body_which) {
case Request_body_configpushmsg:
/* configuration packet, it will only arrive when scheduler has
* register itself to controller, we can handle it synchronized
*/
rc = sched_init_config(
sched, request->body.configpushmsg->data.conf4sched);
if (rc >= 0) {
/* configuration will be saved in sched->config */
request->body.configpushmsg->data.conf4sched = NULL;
case Request_body_configpushmsg: {
/* Configuration packet. Route it through the pend-update queue so
* sched_init_config runs in the MAIN loop, serialised with the
* job/worker/queue updates it co-arrives with on a controller
* rejoin (bug-1513). Applying it inline on the vcpu thread races
* sched_update's job processing (fcfs_update_queue's ordq_put vs
* fcfs_update_job's ordq_peek on the same plugin queues), and the
* queues may not exist yet when the job update looks them up →
* NULL j->queue → SIGSEGV. */
xce_sched_update_request_t* req;
xce_conf4sched_t* config = request->body.configpushmsg->data.conf4sched;
req = (xce_sched_update_request_t*)calloc(1, sizeof(*req));
if (req == NULL) {
rc = -ENOMEM;
break;
}
req->op = XCE_SCHED_OP_UPDATE_CONFIG;
req->u.config = config;
enq_pend_update(sched, req);
/* Ownership of config moved to the queued request. */
request->body.configpushmsg->data.conf4sched = NULL;
rc = 0;
break;
}
case Request_body_datamsg:
rc = sched_handle_dataset(sched, request->body.datamsg);
break;

View file

@ -1,5 +1,6 @@
#include <argp.h>
#include <signal.h>
#include <stdatomic.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
@ -156,7 +157,21 @@ int main(int argc, char* argv[])
INFO("xce scheduler is running");
int hb_was_down = 0;
while (1) {
sched_rejoin_if_needed(&sched);
{
int hb = xcecli_heartbeat(sched.apictx, 1);
if ((hb == 0) && hb_was_down) {
INFO("controller connection restored — requesting scheduler rejoin");
atomic_store_explicit(&sched.needs_rejoin, 1,
memory_order_release);
}
hb_was_down = (hb < 0);
}
sched_update(&sched);
sched_begin_session(&sched);
sched_go(&sched);

View file

@ -73,8 +73,16 @@ static int limit_update_config(xce_sched_plugin_t* plugin,
unlock(privdata->limits);
if (rc < 0) {
kill_limit(limit);
if (rc == -EEXIST) {
/* Rejoin re-push: the limit name already exists from the
* first config, and xkvmgr_add already rolled back — vclear
* ran the kvmgr's datakiller (kill_limit) and freed this
* freshly-built duplicate. Never free it again here (that
* double-free was the rejoin heap-use-after-free crash,
* bug-1513). The existing limit is still valid — skip and
* keep the config push idempotent. */
continue;
}
return rc;
}
}

View file

@ -2,6 +2,7 @@
#include <pthread.h>
#include <rtti.h>
#include <signal.h>
#include <stdatomic.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
@ -375,6 +376,45 @@ static int sched_plugin_init(xce_scheduler_t* sched)
return 0;
}
static int sched_reconnect_cb(xcecli_ctx_t c, void* arg)
{
xce_scheduler_t* sched = (xce_scheduler_t*)arg;
(void)c;
if (sched == NULL) {
return -EINVAL;
}
atomic_store_explicit(&sched->needs_rejoin, 1, memory_order_release);
return 0;
}
int sched_rejoin_if_needed(xce_scheduler_t* sched)
{
int rc;
if (sched == NULL) {
return -EINVAL;
}
if (!atomic_load_explicit(&sched->needs_rejoin, memory_order_acquire)) {
return 0;
}
atomic_store_explicit(&sched->needs_rejoin, 0, memory_order_release);
INFO("%s: controller reconnected — re-joining as scheduler", __func__);
rc = xcecli_join(sched->apictx, HostRole_scheduler, NULL);
if (rc < 0) {
WARN("%s: rejoin failed rc=%d; will re-try on the next reconnect",
__func__, rc);
return rc;
}
INFO("%s: rejoin ok", __func__);
return 0;
}
int sched_init(xce_scheduler_t* sched, xce_client_config_t* config,
const xce_cluster_config_t* cluster_cfg,
xce_request_handler_t handler)
@ -466,6 +506,7 @@ int sched_init(xce_scheduler_t* sched, xce_client_config_t* config,
}
xcecli_set_handler(sched->apictx, handler, sched);
xcecli_set_reconnect_handler(sched->apictx, sched_reconnect_cb, sched);
/* HA mode: iterate candidates until one accepts as primary */
if (sched->cluster_cfg.candidates_count > 0) {
@ -495,6 +536,8 @@ int sched_init(xce_scheduler_t* sched, xce_client_config_t* config,
continue;
}
xcecli_set_handler(sched->apictx, handler, sched);
xcecli_set_reconnect_handler(sched->apictx, sched_reconnect_cb,
sched);
rc = xcecli_start(sched->apictx);
if (rc < 0) {
@ -822,21 +865,56 @@ int sched_update(xce_scheduler_t* sched)
sched->pend_updates[1] = q;
mylock_unlock(&(sched->lock));
/* Apply any queued config update FIRST so job/worker/queue updates
* always see the queues/resources they reference (join AND rejoin re-push
* ordering, bug-1513). The config and dataset pushes are separate wire
* messages dispatched to different vcpu threads, so their enqueue order
* is not guaranteed — a job update processed before its config would find
* NULL j->queue and crash. Park non-config items on a staging queue. */
dataq_t staging;
dataq_init(&staging, NULL);
while (1) {
data_item_t* it;
xce_sched_update_request_t* req;
int rc;
it = dataq_deq(q, NULL);
if (it == NULL) {
break;
}
req = ITPTR_TO_UPDATE_REQ(it);
if (req->op != XCE_SCHED_OP_UPDATE_CONFIG) {
dataq_enq(&staging, it);
continue;
}
int rc = sched_init_config(sched, req->u.config);
if (rc >= 0) {
/* ownership passed to sched->config */
req->u.config = NULL;
}
sched_pend_update_killer(it);
if (rc < 0) {
WARN("%s: config update failed: %s (%d)", __func__,
strerror(-rc), rc);
}
}
while (1) {
data_item_t* it;
xce_sched_update_request_t* req;
int rc;
it = dataq_deq(&staging, NULL);
if (it == NULL) {
break;
}
req = ITPTR_TO_UPDATE_REQ(it);
switch (req->op) {
case XCE_SCHED_OP_UPDATE_CONFIG:
/* this will only be handled when scheduler has been started, so
* should never get here */
/* handled in the pre-pass above; cannot reach here */
rc = -EINVAL;
break;
case XCE_SCHED_OP_UPDATE_JOB:
@ -957,6 +1035,8 @@ int sched_update(xce_scheduler_t* sched)
}
}
dataq_fini(&staging);
return 0;
}