wdgzyjx/myxce
1
0
Fork 0
forked from lei.chen/myxce
Code Pull requests Activity

refactor(controller): move manager init to cluster_init_phase1

Browse source 
Extract cluster_init_managers() which initializes configmgr, jobmgr,
resmgr, workermgr, qmgr, sharedinst, xstrings, endpoints, and job_obj.
Call it from cluster_init_phase1 (after init_controller_tunables) so
all managers exist before election.

This fixes the heap-buffer-overflow in handle_connection: incoming
OP_requestVote SSL connections arrive during the pre-election phase
and call zarray_store(&c->endpoints) — which requires endpoints to
be initialized first.

cluster_init_phase2 is simplified: it no longer re-initializes these
managers and only sets up iomgr/SSL, opens the listener, and loads
cluster-config-dependent data.
This commit is contained in:
Rongsong Shen 2026-05-30 01:20:22 +00:00
parent c42e1e0cb4
commit b2faf41703
1 changed files with 55 additions and 39 deletions
Download patch file Download diff file Expand all files Collapse all files

94
src/controller/cluster.c
View file

@ -656,6 +656,57 @@ static void config_obj_killer(void* arg)
* TODO: wire these to cluster_config fields once the config generation tool
* supports jobmgr/workermgr/qmgr/resmgr/xstrings (@13–@17).
*/
/*
* cluster_init_managers: allocate and initialize the manager subsystems.
*
* Called from cluster_init_phase1 so these are ready before election —
* handle_connection() arrives during the pre-election phase and may call
* zarray_store on c->endpoints (via endpoint tracking) which requires the
* managers to exist. cluster_init_phase2 skips re-init if already done.
*/
static int cluster_init_managers(cluster_t *c)
{
kvmgr_params_t config_mgr_params
= { .key_type = KVMGR_KEY_STRING,
.nslots = 128,
.delta = 128,
.datakiller = (destroy_fn_t)config_obj_killer };
kvmgr_params_t resmgr_params
= { .key_type = KVMGR_KEY_STRING,
.nslots = c->tunables.resmgr_nhash,
.delta = c->tunables.resmgr_delta,
.datakiller = (destroy_fn_t)res_obj_killer };
kvmgr_params_t jobmgr_params
= { .key_type = KVMGR_KEY_INT,
.nslots = c->tunables.jobmgr_nhash,
.delta = c->tunables.jobmgr_delta,
.datakiller = (destroy_fn_t)job_obj_killer };
kvmgr_params_t workermgr_params
= { .key_type = KVMGR_KEY_STRING,
.nslots = c->tunables.workermgr_nhash,
.delta = c->tunables.workermgr_delta,
.datakiller = (destroy_fn_t)worker_obj_killer };
kvmgr_params_t qmgr_params
= { .key_type = KVMGR_KEY_STRING,
.nslots = c->tunables.qmgr_nhash,
.delta = c->tunables.qmgr_delta,
.datakiller = (destroy_fn_t)queue_obj_killer };
c->configmgr = makeInstance(&xkvmgr_type, &config_mgr_params);
c->jobmgr = makeInstance(&xkvmgr_type, &jobmgr_params);
c->resmgr = makeInstance(&xkvmgr_type, &resmgr_params);
zarray_init(&(c->sharedinst), c->tunables.resmgr_inst,
(zarray_item_killer_t)resval_obj_killer);
c->workermgr = makeInstance(&xkvmgr_type, &workermgr_params);
c->qmgr = makeInstance(&xkvmgr_type, &qmgr_params);
xstring_table_init(c->tunables.xstrings_nhash);
job_obj_init(5381, 5381);
zarray_init(&(c->endpoints), 16, endpoint_killer);
return 0;
}
static void init_controller_tunables(cluster_t *c)
{
/* Jobs can be numerous; workers/queues/resources are typically smaller.
@ -755,6 +806,7 @@ int cluster_init_phase1(cluster_t* c)
c->ep.fd = -1;
init_controller_tunables(c);
cluster_init_managers(c);
return 0;
}
@ -805,24 +857,7 @@ int cluster_open_listener(cluster_t* c)
int cluster_init_phase2(cluster_t* c)
{
int i, rc;
kvmgr_params_t config_mgr_params
= { .key_type = KVMGR_KEY_STRING,
.nslots = 128,
.delta = 128,
.datakiller = (destroy_fn_t)config_obj_killer };
kvmgr_params_t resmgr_params
= { .key_type = KVMGR_KEY_STRING,
.datakiller = (destroy_fn_t)res_obj_killer };
kvmgr_params_t jobmgr_params
= { .key_type = KVMGR_KEY_INT,
.datakiller = (destroy_fn_t)job_obj_killer };
kvmgr_params_t workermgr_params
= { .key_type = KVMGR_KEY_STRING,
.datakiller = (destroy_fn_t)worker_obj_killer };
kvmgr_params_t qmgr_params
= { .key_type = KVMGR_KEY_STRING,
.datakiller = (destroy_fn_t)queue_obj_killer };
int i, rc;
if (c == NULL)
return -EINVAL;
@ -848,22 +883,8 @@ int cluster_init_phase2(cluster_t* c)
mylock_init(&(c->sched.lock));
c->sched.ep = NULL;
resmgr_params.nslots = c->tunables.resmgr_nhash;
resmgr_params.delta = c->tunables.resmgr_delta;
jobmgr_params.nslots = c->tunables.jobmgr_nhash;
jobmgr_params.delta = c->tunables.jobmgr_delta;
workermgr_params.nslots = c->tunables.workermgr_nhash;
workermgr_params.delta = c->tunables.workermgr_delta;
qmgr_params.nslots = c->tunables.qmgr_nhash;
qmgr_params.delta = c->tunables.qmgr_delta;
c->configmgr = makeInstance(&xkvmgr_type, &config_mgr_params);
c->jobmgr = makeInstance(&xkvmgr_type, &jobmgr_params);
c->resmgr = makeInstance(&xkvmgr_type, &resmgr_params);
zarray_init(&(c->sharedinst), c->tunables.resmgr_inst,
(zarray_item_killer_t)resval_obj_killer);
c->workermgr = makeInstance(&xkvmgr_type, &workermgr_params);
c->qmgr = makeInstance(&xkvmgr_type, &qmgr_params);
/* Managers, xstrings, endpoints, job_obj are initialized early in
* cluster_init_phase1 via cluster_init_managers(). Skip re-init here. */
/* iomgr and vtaskmgr are initialised in cluster_open_listener() /
* myapp_run() before election. Guard against re-initialisation when
@ -876,11 +897,6 @@ int cluster_init_phase2(cluster_t* c)
iomgr_set_ssl_ctx(&(c->iomgr), ssl_ctx);
}
xstring_table_init(c->tunables.xstrings_nhash);
job_obj_init(5381, 5381);
zarray_init(&(c->endpoints), 16, endpoint_killer);
/* Open the listener socket (idempotent — skipped if already opened before
* election via cluster_open_listener()). */
rc = cluster_open_listener(c);