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

feat: split cluster_init() into phase1/phase2; update myapp_run() for two-phase startup with am_i_primary gating

Browse source 
- cluster_init_phase1(): hooks, HA init (priority from --peers), standby_endpoints zarray, cbuf
- cluster_init_phase2(): SSL ctx from ha.ssl_ctx, cluster_config guard, kvmgr sizing, iomgr, endpoint, persist, role promotion, create_lost_and_found, persist_recover
- cluster_init(): legacy wrapper calling both phases with inline SSL init for backward compat
- myapp_run(): phase1 → SSL init → am_i_primary → load_primary_config/fetch_config_from_primary → phase2 → start → run
- myreq_handler(): add OPCode_replConfig and OPCode_upgradeConfig dispatch
- handle_endpoint_error(): prune standby_endpoints on HostRole_controller disconnect
- cluster_run(): log restart_requested on exit
- config_repl.c: add create_lost_and_found, handle_config_request, handle_upgrade_config stubs
- config_repl.h: add declarations for new stubs; include netio.h + request.h
- cluster.h: declare cluster_init_phase1 and cluster_init_phase2
- myconf.c: fix FATAL → ERROR (FATAL macro undefined); add fetch_config_from_primary stub
This commit is contained in:
Rongsong Shen 2026-05-13 17:06:45 +00:00
parent ec3cab40c3
commit 6b06aa02d9
6 changed files with 307 additions and 92 deletions
Download patch file Download diff file Expand all files Collapse all files

277
src/controller/cluster.c
View file

@ -2,10 +2,12 @@
#include <errno.h>
#include <rtti.h>
#include <stdatomic.h>
#include <string.h>
#include <unistd.h>
#include "cluster.h"
#include "config_repl.h"
#include "dataq.h"
#include "htab.h"
#include "kvmgr.h"
@ -67,6 +69,19 @@ static void handle_endpoint_error(endpoint_t* ep)
handle_scheduler_leave(c);
break;
case HostRole_controller:
if (c->ha.standby_endpoints) {
pthread_mutex_lock(&c->ha.lock);
for (int idx = 0;
idx < c->ha.standby_endpoints->count; idx++) {
endpoint_t *se = (endpoint_t *)zarray_get(
c->ha.standby_endpoints, idx);
if (se == ep) {
zarray_del(c->ha.standby_endpoints, idx);
break;
}
}
pthread_mutex_unlock(&c->ha.lock);
}
break;
case HostRole_client:
handle_client_leave(c);
@ -201,6 +216,34 @@ static int myreq_handler(vtask_t* task)
case OPCode_clrpreason:
rc = do_clr_preason(c, ep, &request);
break;
case OPCode_replConfig:
rc = handle_config_request(c, ep, &request);
break;
case OPCode_upgradeConfig:
#if CONF_ENABLE_UPGRADE_CONFIG
if (!ha_is_primary(&c->ha)) {
/* Standby cannot handle upgrade — send error reply */
xce_reply_t ureply;
init_reply_by_req(&ureply, &request);
ureply.status = -EAGAIN;
ureply.data_which = RequestReply_data_nodata;
xce_reply(ep, c->myhost, &ureply);
rc = -EAGAIN;
break;
}
rc = handle_upgrade_config(c, ep, &request);
break;
#else
{
xce_reply_t ureply;
init_reply_by_req(&ureply, &request);
ureply.status = -ENOSYS;
ureply.data_which = RequestReply_data_nodata;
xce_reply(ep, c->myhost, &ureply);
rc = -ENOSYS;
}
break;
#endif
default:
rc = -E_XCE_BAD_OPCODE;
ERROR("Unknown opcode: %d", request.opcode);
@ -510,10 +553,70 @@ static void config_obj_killer(void* arg)
}
}
int cluster_init(cluster_t* c)
int cluster_init_phase1(cluster_t* c)
{
int rc;
if (c == NULL)
return -EINVAL;
c->myhost = &from_controller;
c->ncontrollers = 1;
c->myidx = 0;
xce_hook_init(&(c->hooksys), XCE_CONTROLLER_HOOK_MAX);
xce_install_hooks(c);
/* HA config and priority — derived from --peers list before config loads */
ha_config_t ha_cfg = {
.heartbeat_interval_ms = HA_DEFAULT_HEARTBEAT_INTERVAL_MS,
.heartbeat_timeout_ms = HA_DEFAULT_HEARTBEAT_TIMEOUT_MS,
.election_timeout_ms = HA_DEFAULT_ELECTION_TIMEOUT_MS,
.probe_connect_timeout_ms = HA_DEFAULT_PROBE_CONNECT_TIMEOUT_MS,
.max_promotion_retries = HA_DEFAULT_MAX_PROMOTION_RETRIES,
.election_jitter_max_ms = HA_DEFAULT_ELECTION_JITTER_MAX_MS,
.election_priority_step_ms = HA_DEFAULT_ELECTION_PRIORITY_STEP_MS,
};
int my_priority = 0;
const xce_cluster_config_t* pcfg = &c->myconf->peers_cfg;
if (pcfg->candidates_count > 0) {
my_priority = pcfg->candidates_count - 1; /* default: last = lowest */
for (int i = 0; i < pcfg->candidates_count; i++) {
const char* key = pcfg->candidates[i].name[0]
? pcfg->candidates[i].name
: pcfg->candidates[i].hostname;
if (strncmp(key, c->myconf->name, XCE_MAXNAME_LEN - 1) == 0) {
my_priority = i;
break;
}
}
INFO("HA peers: %d candidates, my_priority=%d (name=%s)",
pcfg->candidates_count, my_priority, c->myconf->name);
}
rc = ha_init(&c->ha, &ha_cfg, my_priority);
if (rc != 0) {
ERROR("ha_init failed: %d", rc);
return rc;
}
ha_install_sigusr1(&c->ha);
/* Initialize standby endpoint tracking */
c->ha.standby_endpoints = (zarray_t *)calloc(1, sizeof(zarray_t));
if (c->ha.standby_endpoints) {
zarray_init(c->ha.standby_endpoints, 4, NULL);
}
/* cbuf used by cluster_run() to signal restart/stop */
cbuf_init(&(c->wbuf), 1, sizeof(int));
return 0;
}
int cluster_init_phase2(cluster_t* c)
{
int i, rc;
SSL_CTX* ssl_ctx = NULL;
kvmgr_params_t config_mgr_params
= { .key_type = KVMGR_KEY_STRING,
.nslots = 128,
@ -532,34 +635,30 @@ int cluster_init(cluster_t* c)
= { .key_type = KVMGR_KEY_STRING,
.datakiller = (destroy_fn_t)queue_obj_killer };
if (c == NULL) {
if (c == NULL)
return -EINVAL;
if (c->myconf->cluster_config == NULL) {
ERROR("cluster_init_phase2: cluster_config not loaded");
return -EINVAL;
}
c->myhost = &from_controller;
/* FIXME: */
c->ncontrollers = 1;
c->myidx = 0;
xce_hook_init(&(c->hooksys), XCE_CONTROLLER_HOOK_MAX);
xce_install_hooks(c);
rc = xce_run_hooks(&(c->hooksys), XCE_CONTROLLER_HOOK_CLUSTER,
HOOK_BEFORE_EVENT, HOOK_DETAIL_CLUSTER_INIT, c);
if (rc < 0) {
return rc;
SSL_CTX* ssl_ctx = (SSL_CTX*)c->ha.ssl_ctx;
if (ssl_ctx == NULL) {
ERROR("cluster_init_phase2: ssl_ctx not set (myapp_run must set "
"c->ha.ssl_ctx before calling phase2)");
return -EINVAL;
}
/* HOOK_BEFORE_EVENT for CLUSTER_INIT loads cluster_config from disk via
* xce_config_init(). In the two-phase path, cluster_config is already
* populated by load_primary_config() / fetch_config_from_primary() before
* phase2 is called — so we do NOT re-run the hook here.
* The legacy cluster_init() wrapper runs it before calling phase2. */
mylock_init(&(c->sched.lock));
c->sched.ep = NULL;
initialize_openssl();
ssl_ctx = create_ssl_context(c->myconf->certfile, c->myconf->keyfile);
if (ssl_ctx == NULL) {
return -errno;
}
resmgr_params.nslots = c->myconf->cluster_config->resmgr->nhash_slots;
resmgr_params.delta = c->myconf->cluster_config->resmgr->delta;
jobmgr_params.nslots = c->myconf->cluster_config->jobmgr->nhash_slots;
@ -570,9 +669,8 @@ int cluster_init(cluster_t* c)
qmgr_params.delta = c->myconf->cluster_config->qmgr->delta;
c->configmgr = makeInstance(&xkvmgr_type, &config_mgr_params);
c->jobmgr = makeInstance(&xkvmgr_type, &jobmgr_params);
c->resmgr = makeInstance(&xkvmgr_type, &resmgr_params);
c->jobmgr = makeInstance(&xkvmgr_type, &jobmgr_params);
c->resmgr = makeInstance(&xkvmgr_type, &resmgr_params);
zarray_init(&(c->sharedinst), c->myconf->cluster_config->resmgr->inst_delta,
(zarray_item_killer_t)resval_obj_killer);
c->workermgr = makeInstance(&xkvmgr_type, &workermgr_params);
@ -586,10 +684,26 @@ int cluster_init(cluster_t* c)
zarray_init(&(c->endpoints), 16, endpoint_killer);
rc = endpoint_init(&(c->ep), AF_INET, SOCK_STREAM,
c->myconf->cluster_config->port, 5);
/* Determine listen port: prefer per-candidate port from --peers list,
* fall back to cluster_config->port */
uint32_t my_port = 0;
{
const xce_cluster_config_t* pcfg = &c->myconf->peers_cfg;
for (int j = 0; j < pcfg->candidates_count; j++) {
const char* key = pcfg->candidates[j].name[0]
? pcfg->candidates[j].name
: pcfg->candidates[j].hostname;
if (strncmp(key, c->myconf->name, XCE_MAXNAME_LEN - 1) == 0) {
my_port = pcfg->candidates[j].port;
break;
}
}
}
if (my_port == 0)
my_port = (uint32_t)c->myconf->cluster_config->port;
rc = endpoint_init(&(c->ep), AF_INET, SOCK_STREAM, (int)my_port, 5);
if (rc < 0) {
cleanup_openssl();
return rc;
}
@ -599,8 +713,6 @@ int cluster_init(cluster_t* c)
controller_init(&(c->controllers[i]), c);
}
cbuf_init(&(c->wbuf), 1, sizeof(int));
zarray_init(&(c->configs), 64, NULL /* FIXME */);
vtaskmgr_init(&(c->vtaskmgr), c->myconf->cluster_config->vcpus);
@ -628,59 +740,19 @@ int cluster_init(cluster_t* c)
}
}
/* HA initialisation */
{
ha_config_t ha_cfg = {
.heartbeat_interval_ms = HA_DEFAULT_HEARTBEAT_INTERVAL_MS,
.heartbeat_timeout_ms = HA_DEFAULT_HEARTBEAT_TIMEOUT_MS,
.election_timeout_ms = HA_DEFAULT_ELECTION_TIMEOUT_MS,
.probe_connect_timeout_ms = HA_DEFAULT_PROBE_CONNECT_TIMEOUT_MS,
.max_promotion_retries = HA_DEFAULT_MAX_PROMOTION_RETRIES,
};
/* Derive priority from position in --peers list.
* Index 0 = highest priority (becomes primary on startup).
* If the controller's name is not found in the list, fall back to
* the last index (lowest priority = standby). */
int my_priority = 0;
const xce_cluster_config_t* pcfg = &c->myconf->peers_cfg;
if (pcfg->candidates_count > 0) {
my_priority = pcfg->candidates_count - 1; /* default: last = lowest */
for (int i = 0; i < pcfg->candidates_count; i++) {
/* Match by logical name first (supports multiple controllers
* on the same dev host via name=host:port syntax). Fall back
* to hostname match when no name was specified. */
const char* key = pcfg->candidates[i].name[0]
? pcfg->candidates[i].name
: pcfg->candidates[i].hostname;
if (strncmp(key, c->myconf->name, XCE_MAXNAME_LEN - 1) == 0) {
my_priority = i;
break;
}
}
INFO("HA peers: %d candidates, my_priority=%d (name=%s)",
pcfg->candidates_count, my_priority, c->myconf->name);
}
rc = ha_init(&c->ha, &ha_cfg, my_priority);
if (rc != 0) {
ERROR("ha_init failed: %d", rc);
persist_fini(&c->persist);
return rc;
}
ha_install_sigusr1(&c->ha);
/* Determine initial role: priority-0 node starts as primary */
if (am_i_primary(c))
ha_promote_to_primary(c);
else
atomic_store_explicit(&c->ha.role, (int32_t)CTRL_STANDBY,
memory_order_release);
}
/* Determine initial role based on am_i_primary() result set by myapp_run */
if (ha_is_primary(&c->ha))
ha_promote_to_primary(c);
else
atomic_store_explicit(&c->ha.role, (int32_t)CTRL_STANDBY,
memory_order_release);
xce_run_hooks(&(c->hooksys), XCE_CONTROLLER_HOOK_CLUSTER, HOOK_AFTER_EVENT,
HOOK_DETAIL_CLUSTER_INIT, c);
/* Ensure lost_and_found synthetic queue/host exist before recovery */
create_lost_and_found(c);
/* Recover WAL/snapshot now that resmgr, jobmgr, and workermgr are
* fully populated by cluster_build_resources() above. */
rc = persist_recover(&c->persist);
@ -693,6 +765,42 @@ int cluster_init(cluster_t* c)
return 0;
}
int cluster_init(cluster_t* c)
{
int rc;
/* Legacy wrapper: run both phases with SSL init inline.
* Used by tests and any caller that provides a fully populated myconf_t
* (including cluster_config) before calling cluster_init(). */
if (c == NULL)
return -EINVAL;
rc = cluster_init_phase1(c);
if (rc != 0)
return rc;
/* Create SSL context for the wrapper path */
initialize_openssl();
SSL_CTX* ssl_ctx = create_ssl_context(c->myconf->certfile,
c->myconf->keyfile);
if (ssl_ctx == NULL) {
ERROR("create_ssl_context failed");
return -errno;
}
c->ha.ssl_ctx = ssl_ctx;
/* For the legacy path, run the before-event hook to populate
* cluster_config via xce_config_init() if it isn't already set */
if (c->myconf->cluster_config == NULL) {
rc = xce_run_hooks(&(c->hooksys), XCE_CONTROLLER_HOOK_CLUSTER,
HOOK_BEFORE_EVENT, HOOK_DETAIL_CLUSTER_INIT, c);
if (rc < 0)
return rc;
}
return cluster_init_phase2(c);
}
static int make_heartbeat(cluster_t* c, endpoint_t* ep)
{
return xce_nop_req(ep);
@ -876,11 +984,22 @@ int cluster_run(cluster_t* c)
iomgr_unmonitor(&(c->iomgr), &(c->ep));
iomgr_stop(&(c->iomgr));
if (atomic_load(&c->restart_requested)) {
INFO("Config upgrade restart requested — exiting for service manager restart");
}
return 0;
}
void cluster_fini(cluster_t* c)
{
/* Free heap-allocated standby_endpoints array (created in phase1) */
if (c->ha.standby_endpoints) {
zarray_fini(c->ha.standby_endpoints);
free(c->ha.standby_endpoints);
c->ha.standby_endpoints = NULL;
}
/* Stop HA watchdog thread before tearing down other subsystems — P1-5 */
ha_fini(&c->ha);

26
src/controller/config_repl.c
View file

@ -388,6 +388,32 @@ int load_primary_config_for_test(cluster_t *c)
}
#endif
/* ── create_lost_and_found ──────────────────────── */
void create_lost_and_found(cluster_t *c) { (void)c; /* Task 10 */ }
/* ── handle_config_request ──────────────────────── */
int handle_config_request(cluster_t *c, endpoint_t *ep, xce_request_t *req)
{
(void)c;
(void)ep;
(void)req;
/* Task 9: full implementation — read versioned blob, send reply */
return -ENOSYS;
}
/* ── handle_upgrade_config ──────────────────────── */
int handle_upgrade_config(cluster_t *c, endpoint_t *ep, xce_request_t *req)
{
(void)c;
(void)ep;
(void)req;
/* Task 9: full implementation — validate, store, fan-out, restart */
return -ENOSYS;
}
/* ── sanitise_cn ────────────────────────────────── */
char *sanitise_cn(const char *src, size_t maxlen)

22
src/controller/config_repl.h
View file

@ -7,6 +7,8 @@
#include "myconf.h"
#include "clusterconfig.h"
#include "netio.h"
#include "request.h"
/* Max config blob size — reject before allocating */
#define CONF_MAX_SIZE (1 * 1024 * 1024) /* 1 MiB */
@ -84,6 +86,26 @@ void conf_trim_old_versions(const char *workdir, int keep_n);
*/
int config_contains_reserved_name(cluster_config_t *cfg);
/*
* create_lost_and_found: ensure the synthetic "lost_and_found" queue and
* host exist before persist_recover() runs. Stub for Task 10.
*/
void create_lost_and_found(cluster_t *c);
/*
* handle_config_request: handle an OP_replConfig request from a standby.
* Reads the current versioned config blob and sends it back.
* Stub for Task 8 (full implementation in Task 9).
*/
int handle_config_request(cluster_t *c, endpoint_t *ep, xce_request_t *req);
/*
* handle_upgrade_config: handle an OP_upgradeConfig request from an admin tool.
* Validates, stores, and fans out the new config blob; triggers restart.
* Stub for Task 9. Only callable when CONF_ENABLE_UPGRADE_CONFIG is set.
*/
int handle_upgrade_config(cluster_t *c, endpoint_t *ep, xce_request_t *req);
/*
* sanitise_cn: copy at most maxlen printable ASCII chars from src to a
* static buffer, replacing non-printable/control chars with '?'.

2
src/controller/include/cluster.h
View file

@ -22,6 +22,8 @@
typedef int (*collect_check_fn_t)(void*);
extern int check_controller(void*);
extern int cluster_init_phase1(cluster_t*);
extern int cluster_init_phase2(cluster_t*);
extern int cluster_init(cluster_t*);
extern int cluster_start(cluster_t*);
extern int cluster_run(cluster_t*);

50
src/controller/myapp.c
View file

@ -23,17 +23,51 @@ int myapp_run(myconf_t* myconf)
int rc;
memset(&c, 0x0, sizeof(c));
c.myconf = myconf;
rc = cluster_init(&c);
/* Phase 1: pre-config init (hooks, HA priority, standby_endpoints, cbuf) */
rc = cluster_init_phase1(&c);
if (rc < 0) {
SET_ERROR(-rc);
ERROR("cluster_init: %s", strerror(-rc));
ERROR("cluster_init_phase1: %s", strerror(-rc));
cluster_fini(&c);
return rc;
}
/* SSL context — created after Phase 1, used by election probing and Phase 2 */
initialize_openssl();
SSL_CTX* ssl_ctx = create_ssl_context(myconf->certfile, myconf->keyfile);
if (!ssl_ctx) {
ERROR("create_ssl_context failed");
cluster_fini(&c);
return -EINVAL;
}
c.ha.ssl_ctx = ssl_ctx;
/* Election: determine role before loading config */
if (am_i_primary(&c)) {
/* Primary: load config from conf/current (FATAL if missing) */
rc = load_primary_config(&c);
if (rc != 0) {
cluster_fini(&c);
myconf_fini(myconf);
return rc;
}
} else {
/* Standby: fetch config from primary before Phase 2 */
rc = fetch_config_from_primary(&c);
if (rc != 0) {
cluster_fini(&c);
myconf_fini(myconf);
return rc;
}
}
/* Phase 2: full init (requires cluster_config) */
rc = cluster_init_phase2(&c);
if (rc < 0) {
ERROR("cluster_init_phase2: %s", strerror(-rc));
cluster_fini(&c);
myconf_fini(myconf);
return rc;
}
@ -42,13 +76,13 @@ int myapp_run(myconf_t* myconf)
SET_ERROR(-rc);
perror("cluster_start");
cluster_fini(&c);
myconf_fini(myconf);
return rc;
}
rc = cluster_run(&c);
cluster_fini(&c);
myconf_fini(myconf);
return rc;
}

22
src/controller/myconf.c
View file

@ -166,7 +166,7 @@ int load_primary_config(cluster_t *c)
char fname[256] = {0};
int rc = conf_read_current_filename(conf_dir, fname, sizeof(fname));
if (rc != 0) {
FATAL("Primary has no config at '%s/conf/current' (rc=%d) — "
ERROR("Primary has no config at '%s/conf/current' (rc=%d) — "
"provision the config before starting",
conf_dir, rc);
return -ENOENT;
@ -183,7 +183,7 @@ int load_primary_config(cluster_t *c)
size_t blen = 0;
rc = conf_read_raw(myconf->conffile, &bytes, &blen);
if (rc != 0) {
FATAL("Primary config '%s' unreadable: %d", myconf->conffile, rc);
ERROR("Primary config '%s' unreadable: %d", myconf->conffile, rc);
free(myconf->conffile);
myconf->conffile = NULL;
return rc;
@ -205,7 +205,7 @@ int load_primary_config(cluster_t *c)
/* Decode Cap'n Proto directly into myconf->cluster_config */
rc = load_cluster_config_from_file(myconf->conffile, &myconf->cluster_config);
if (rc != 0) {
FATAL("Primary config '%s' failed Cap'n Proto decode: %d",
ERROR("Primary config '%s' failed Cap'n Proto decode: %d",
myconf->conffile, rc);
free(bytes);
free(myconf->conffile);
@ -214,7 +214,7 @@ int load_primary_config(cluster_t *c)
}
if (myconf->cluster_config == NULL) {
FATAL("Primary config '%s' decoded to NULL", myconf->conffile);
ERROR("Primary config '%s' decoded to NULL", myconf->conffile);
free(bytes);
free(myconf->conffile);
myconf->conffile = NULL;
@ -227,7 +227,7 @@ int load_primary_config(cluster_t *c)
/* Check for reserved lost_and_found name */
if (config_contains_reserved_name(myconf->cluster_config)) {
FATAL("Config '%s' contains reserved name 'lost_and_found'",
ERROR("Config '%s' contains reserved name 'lost_and_found'",
myconf->conffile);
free_ClusterConfig(myconf->cluster_config);
free(myconf->cluster_config);
@ -253,6 +253,18 @@ int load_primary_config(cluster_t *c)
return 0;
}
/* ── fetch_config_from_primary ──────────────────── */
int fetch_config_from_primary(cluster_t *c)
{
(void)c;
/* Task 9: full implementation — connect to primary, send OP_replConfig,
* receive blob, verify SHA-256, call conf_save_config(), decode into
* myconf->cluster_config. */
ERROR("fetch_config_from_primary: not yet implemented (Task 9)");
return -ENOSYS;
}
/* ── myconf_fini ────────────────────────────────── */
void myconf_fini(myconf_t *myconf)