#include "config.h" #include #include #include #include #include #include "cluster.h" #include "config_repl.h" #include "controller.capnp.h" #include "dataq.h" #include "htab.h" #include "kvmgr.h" #include "mylock.h" #include "request.h" #include "vtaskmgr.h" #include "xce/cluster_def.h" #include "zarray.h" #include "xce/types.h" #include "jobj.h" #include "resobj.h" #include "utils.h" #include "worker.h" #include "xce/errnox.h" #include "xstring.h" #include "helper.h" static xce_peerinfo_t from_controller = { .name = "controller", .key = "keepSecret" }; static ep_private_t* ep_get_private() { ep_private_t* priv; priv = (ep_private_t*)calloc(1, sizeof(*priv)); return priv; } static int handle_client_leave(cluster_t* c) { return 0; } static void handle_endpoint_error(endpoint_t* ep) { cluster_t* c = (cluster_t*)ep->owner; ep_private_t* priv = NULL; int role = -1; if (c == NULL) { ERROR("endpoint %p has no owner cluster", ep); return; } priv = (ep_private_t*)ep->data; TRACE("endpoint %p priv data %p", ep, priv); if (priv != NULL) { role = priv->role; switch (role) { case HostRole_worker: if (priv->u.idx >= 0) { handle_worker_leave(c, priv->u.idx); } else { ERROR("incorrect worker idx %d", priv->u.idx); } break; case HostRole_scheduler: 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); break; case HostRole_apiserver: handle_apiserver_leave(c, ep); break; default: ERROR("unknown role %d for endpoint %p", priv->role, ep); break; } } if (ep->idx >= 0) { /* zarray_del will finalize ep */ zarray_del(&(c->endpoints), ep->idx); return; } endpoint_fini(ep); FREEUP(ep); return; } static int do_notify(cluster_t* c, endpoint_t* ep, xce_request_t* req) { xce_notify_data_t* data = req->body.notifymsg; switch (data->event) { case NotifyData_data_jobPidInfo: on_job_start(c, data->data.pidinfo); break; case NotifyData_data_jobFailInfo: on_job_fail(c, data->data.failinfo); break; case NotifyData_data_jobFinishInfo: on_job_finish(c, data->data.finishinfo); break; case NotifyData_data_stoppedJob: on_job_stop(c, data->data.stopped_job); break; case NotifyData_data_resumedJob: on_job_resume(c, data->data.resumed_job); break; case NotifyData_data_ckptinfo: on_job_ckpt(c, ep, data->data.ckptinfo); break; case NotifyData_data_postDoneJob: on_job_postdone(c, ep, data->data.pdone_job); break; case NotifyData_data_taskPidInfo: on_task_start(c, data->data.taskinfo); break; case NotifyData_data_taskFailInfo: on_task_fail(c, data->data.taskfailinfo); break; case NotifyData_data_taskFinish: on_task_finish(c, data->data.taskfinish); break; default: break; } return 0; } static int myreq_handler(vtask_t* task) { vtask_data_t* data = (vtask_data_t*)task->data; endpoint_t* ep = data->u.iodata.ep; cluster_t* c = (cluster_t*)ep->owner; int stop = 0; int rc = 0; xce_request_t request; xce_reply_t reply; /* xce_decode() extracts sha256+bytes for OPCode_upgradeConfig into * request.body.upgradeconfigrequestmsg->raw_sha256/raw_bytes before * capn_free(), so no pre-decode raw-packet copy is needed. */ rc = xce_decode(data->u.iodata.pkt, &request); TRACE("%s cpu#%d opcode %s", __func__, task->vcpu, xce_request_op_name(&request)); init_reply_by_req(&reply, &request); /* Dispatch gate: reject ops not allowed in current FSM state (req 1.3, 1.4) */ rc = ha_dispatch_gate(&c->ha, request.opcode); if (rc != 0) { reply.status = rc; reply.data_which = RequestReply_data_nodata; xce_reply(ep, c->myhost, &reply); xce_free_request(&request); return rc; } switch (request.opcode) { case OPCode_shutdown: reply.status = 0; reply.data_which = RequestReply_data_nodata; xce_reply(ep, c->myhost, &reply); stop = 1; cbuf_put(&(c->wbuf), &stop); break; case OPCode_hello: if (request.body_which == Request_body_ctrlhellomsg) { /* Extended OP_ctrlHello with CtrlHelloRequest body — state sync (Req 6.5–6.7) */ rc = handle_ctrlhello(c, ep, &request); } else if (request.body_which == Request_body_replymsg && request.body.replymsg && request.body.replymsg->data_which == RequestReply_data_ctrlhelloreplymsg) { /* Standby receives CtrlHelloReply from primary — update last known * committed LSN and config tag for use in RDHUP failover path. */ ctrl_hello_reply_t *chrep = request.body.replymsg->data.ctrlhelloreplymsg; if (chrep) { atomic_store_explicit(&c->ha.last_known_committed_lsn, chrep->committed_lsn, memory_order_release); c->ha.last_known_committed_config_tag = chrep->config_tag; INFO("HA: CtrlHelloReply from primary: committed_lsn=%lu config_tag=%lu", (unsigned long)chrep->committed_lsn, (unsigned long)chrep->config_tag); } } else { /* Legacy plain OP_hello (nop path) */ reply.status = 0; reply.data_which = RequestReply_data_nodata; xce_reply(ep, c->myhost, &reply); } break; case OPCode_nop: reply.status = 0; reply.data_which = RequestReply_data_nodata; xce_reply(ep, c->myhost, &reply); break; case OPCode_join: rc = handle_join(c, ep, &request); break; case OPCode_jobdispatch: rc = handle_jobdispatch(c, ep, &request); break; case OPCode_jobsub: rc = handle_jobsubmit(c, ep, &request); break; case OPCode_get: rc = do_get_op(c, ep, &request); break; case OPCode_notify: rc = do_notify(c, ep, &request); break; case OPCode_clusterconf: rc = handle_clusterconf(c, ep, &request); break; case OPCode_update: rc = do_update_op(c, ep, &request); break; case OPCode_jobcontrol: rc = do_jobcontrol_op(c, ep, &request); break; case OPCode_queuecontrol: rc = do_queuecontrol_op(c, ep, &request); break; case OPCode_hostcontrol: rc = do_hostcontrol_op(c, ep, &request); break; case OPCode_runtasks: rc = do_runtasks(c, ep, &request); break; case OPCode_setpreason: rc = do_set_preason(c, ep, &request); break; case OPCode_clrpreason: rc = do_clr_preason(c, ep, &request); break; case OPCode_requestVote: { request_vote_req_t *vrq = request.body.requestvotemsg; uint64_t cand_tag = vrq ? vrq->config_tag : 0; uint64_t cand_lsn = vrq ? vrq->last_log_lsn : 0; uint64_t cand_term = vrq ? vrq->term : 0; uint64_t my_tag = c->myconf->config_version_tag; uint64_t my_lsn = (uint64_t)atomic_load_explicit( &c->ha.applied_lsn, memory_order_acquire); int granted = handle_request_vote( &c->ha, cand_term, vrq ? vrq->candidate_id : "", cand_tag, cand_lsn, my_tag, my_lsn, 0 /* in_catch_up_flag: not implemented here */); request_vote_reply_t vr_rep; memset(&vr_rep, 0, sizeof(vr_rep)); vr_rep.term = (uint64_t)atomic_load_explicit( &c->ha.term, memory_order_acquire); vr_rep.vote_granted = (granted > 0) ? 1 : 0; vr_rep.peer_config_tag = my_tag; vr_rep.peer_last_log_lsn = my_lsn; reply.status = 0; reply.data_which = RequestReply_data_requestvotereplymsg; reply.data.requestvotereplymsg = &vr_rep; xce_reply(ep, c->myhost, &reply); rc = 0; break; } case OPCode_replConfig: rc = handle_config_request(c, ep, &request); break; case OPCode_upgradeConfig: #if CONF_ENABLE_UPGRADE_CONFIG /* OPCode_upgradeConfig is sent BY the primary TO standbys to push config. * Standbys must apply it; only a primary sending this to itself is invalid. */ if (ha_is_primary(&c->ha)) { /* Primary should not receive its own config push — ignore */ 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); reply.status = rc; reply.data_which = RequestReply_data_nodata; xce_reply(ep, c->myhost, &reply); break; } xce_free_request(&request); return rc; } static int vtask_entry(vtask_t* task) { int rc; rc = myreq_handler(task); return rc; } static int handle_msg(endpoint_t* ep) { cluster_t* c = NULL; if (ep == NULL) { return -EINVAL; } c = (cluster_t*)ep->owner; if (c == NULL) { return -EINVAL; } return vtask_handle_msg(&(c->vtaskmgr), (vtask_func_t)vtask_entry, (void*)ep); } static int handle_connection(struct epoll_event* p) { endpoint_t* ep = (endpoint_t*)p->data.ptr; endpoint_t* ep2; cluster_t* c = (cluster_t*)ep->owner; iomgr_t* iomgr = &(c->iomgr); int idx; ep_private_t* priv = NULL; ep2 = endpoint_get(); if (ep2 == NULL) { PERROR("endpoint_get"); return -ENOMEM; } endpoint_accept(ep, ep2); if (ep2->fd < 0) { PERROR("endpoint_accept"); return -1; } TRACE("server sock %d accepted %d\n", ep->fd, ep2->fd); priv = ep_get_private(); if (priv == NULL) { PERROR("ep_get_private"); endpoint_fini(ep2); FREEUP(ep2); return -ENOMEM; } /* Assume client as default */ priv->role = HostRole_client; priv->u.idx = -1; idx = zarray_store(&(c->endpoints), ep2); if (idx < 0) { SET_ERROR(-idx); PERROR("zarray_store"); FREEUP(priv); endpoint_fini(ep2); FREEUP(ep2); return idx; } ep2->idx = idx; ep2->data = priv; ep2->iomgr = iomgr; endpoint_set_owner(ep2, c); endpoint_set_peer(ep2, c->myhost); endpoint_set_msg_handler(ep2, (msg_handler_t)handle_msg); endpoint_set_err_handler(ep2, (err_handler_t)handle_endpoint_error); iomgr_monitor(ep2); return 0; } static int controller_init(controller_t* c, cluster_t* cluster) { c->ep = NULL; return 0; } static void endpoint_killer(void* p) { endpoint_t* ep = (endpoint_t*)p; cluster_t* c = (cluster_t*)ep->owner; if (ep->idx >= 0) { zarray_clear(&(c->endpoints), ep->idx); ep->idx = -1; } endpoint_fini(ep); FREEUP(ep); return; } int cluster_build_config(cluster_t* c) { char valbuf[128]; int i; snprintf(valbuf, sizeof(valbuf) - 1, "%d", c->myconf->cluster_config->heartbeat_interval); add_config(c, "heartbeat_interval", valbuf); snprintf(valbuf, sizeof(valbuf) - 1, "%d", c->myconf->cluster_config->inactive_heartbeat_count); add_config(c, "inactive_heartbeat_count", valbuf); snprintf(valbuf, sizeof(valbuf) - 1, "%d", c->myconf->cluster_config->ncandidates); add_config(c, "ncandidates", valbuf); for (i = 0; i < c->myconf->cluster_config->ncandidates; i++) { char keybuf[128]; sprintf(keybuf, "candidate_%d_name", i); sprintf(valbuf, "%s", c->myconf->cluster_config->candidates[i]->name); add_config(c, keybuf, valbuf); sprintf(keybuf, "candidate_%d_host", i); sprintf(valbuf, "%s", c->myconf->cluster_config->candidates[i]->host); add_config(c, keybuf, valbuf); sprintf(keybuf, "candidate_%d_port", i); sprintf(valbuf, "%d", c->myconf->cluster_config->candidates[i]->port); add_config(c, keybuf, valbuf); } return 0; } int cluster_build_resources(cluster_t* c) { int i; if (c->myconf == NULL) { return -EINVAL; } for (i = 0; i < c->myconf->cluster_config->nres; i++) { xce_resource_t* res; res = makeResource(c->resmgr, c->myconf->cluster_config->resources[i]); if (res == NULL) { return -errno; } } return 0; } int cluster_load_jobs(cluster_t* c) { load_jobs(c); return 0; } int cluster_build_queues(cluster_t* c) { build_queues(c); return 0; } int cluster_build_groups(cluster_t* c) { /* TODO: implement group building */ return 0; } static int xce_config_init(int type, int situation, void* data, va_list ap) { int rc = 0; cluster_t* c = va_arg(ap, cluster_t*); if (situation == HOOK_BEFORE_EVENT) { c->myconf->cluster_config = get_default_conf(c); if (c->myconf->cluster_config == NULL) { return -EINVAL; } return 0; } xce_use(c); rc = cluster_build_config(c); if (rc < 0) { return rc; } rc = cluster_build_resources(c); if (rc < 0) { return rc; } rc = cluster_build_queues(c); if (rc < 0) { return rc; } rc = cluster_build_groups(c); if (rc < 0) { return rc; } rc = cluster_load_jobs(c); if (rc < 0) { return rc; } tlog_info("xce controller is running"); return rc; } static int cluster_hook_fun(int type, int situation, int detail, void* data, va_list ap) { if (detail == HOOK_DETAIL_CLUSTER_INIT) { return xce_config_init(type, situation, data, ap); } return 0; } static int xce_install_hooks(cluster_t* c) { xce_add_hook(&(c->hooksys), XCE_CONTROLLER_HOOK_CLUSTER, (hook_fn_t)cluster_hook_fun, NULL); return 0; } static int run_delay_tasks(delay_task_entry_fn_t entry, void* arg) { return entry(arg); } static void res_obj_killer(xce_resource_t* r) { if (r == NULL) { return; } } static void job_obj_killer(void* arg) { if (arg == NULL) { return; } } static void worker_obj_killer(void* arg) { if (arg == NULL) { return; } } static void queue_obj_killer(void* arg) { if (arg == NULL) { return; } } static void resval_obj_killer(void* arg) { if (arg == NULL) { return; } } static void config_obj_killer(void* arg) { if (arg == NULL) { return; } } /* * init_controller_tunables: set performance-tuning defaults on c->tunables. * * Called early in cluster_init_phase1 so the values are available throughout * the startup sequence (including pre-election and phase2). * * 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. * TODO: replace with config-driven values once the config generation tool * supports cluster_config fields @13–@17 (jobmgr/workermgr/qmgr/resmgr/xstrings). */ c->tunables.jobmgr_nhash = 512; c->tunables.jobmgr_delta = 128; c->tunables.workermgr_nhash = 128; c->tunables.workermgr_delta = 32; c->tunables.qmgr_nhash = 128; c->tunables.qmgr_delta = 32; c->tunables.resmgr_nhash = 128; c->tunables.resmgr_delta = 32; c->tunables.resmgr_inst = 64; c->tunables.xstrings_nhash = 512; } 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, .committing_timeout_ms = HA_DEFAULT_COMMITTING_TIMEOUT_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; } /* Restore persisted election term + voted_for from ha-term.bin (Req 9.2). * ha_persist_term() already writes on vote grant; this reads it back at * startup so the restarted node remembers its last term. */ if (c->myconf->workdir) { rc = ha_restore_term(&c->ha, c->myconf->workdir); if (rc == 0) { INFO("HA: restored term=%lu from %s/ha-term.bin", (unsigned long)atomic_load(&c->ha.term), c->myconf->workdir); } else if (rc != -ENOENT) { WARN("HA: ha_restore_term failed: %d", 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)); /* Sentinel: -1 means iomgr not yet initialised. * cluster_init_phase2() uses this to skip re-init when myapp_run() * has already called iomgr_init() before election. */ c->iomgr.epoll_fd = -1; /* Sentinel: fd=-1 means the listener socket is not yet open. * cluster_open_listener() uses this for its idempotency guard. */ c->ep.fd = -1; init_controller_tunables(c); cluster_init_managers(c); return 0; } int cluster_open_listener(cluster_t* c) { int rc; if (c == NULL) return -EINVAL; /* Idempotent: if the listener socket is already open, do nothing. */ if (c->ep.fd >= 0) return 0; /* Determine listen port: prefer per-candidate port from --peers list, * fall back to cluster_config->port if config is already loaded. */ 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) { if (c->myconf->cluster_config != NULL) { my_port = (uint32_t)c->myconf->cluster_config->port; } else { ERROR("cluster_open_listener: no port in --peers and cluster_config " "not yet loaded; specify port in --peers list"); return -EINVAL; } } rc = endpoint_init(&(c->ep), AF_INET, SOCK_STREAM, (int)my_port, 5); if (rc < 0) return rc; endpoint_set_peer(&(c->ep), c->myhost); return 0; } int cluster_init_phase2(cluster_t* c) { int i, rc; if (c == NULL) return -EINVAL; if (c->myconf->cluster_config == NULL) { ERROR("cluster_init_phase2: cluster_config not loaded"); return -EINVAL; } 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; /* 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 * cluster_init() (legacy wrapper) calls cluster_init_phase2() directly. */ if (c->iomgr.epoll_fd < 0) { iomgr_init(&(c->iomgr)); iomgr_set_ssl_ctx(&(c->iomgr), ssl_ctx); } else { /* ssl_ctx may have been set before phase1; always refresh it here. */ iomgr_set_ssl_ctx(&(c->iomgr), ssl_ctx); } /* Open the listener socket (idempotent — skipped if already opened before * election via cluster_open_listener()). */ rc = cluster_open_listener(c); if (rc < 0) return rc; for (i = 0; i < c->ncontrollers; i++) { controller_init(&(c->controllers[i]), c); } zarray_init(&(c->configs), 64, NULL /* FIXME */); /* Req 4.3: use --vcpus CLI value; fall back to built-in default (8). * The cluster config 'vcpus' field has been removed (Req 4.4 deprecation). * vtaskmgr_init() is guarded by n_vcpus==0: myapp_run() calls it before * election; the guard here covers the legacy cluster_init() path where * it has not been called yet. */ if (c->vtaskmgr.n_vcpus == 0) { int nvcpus = (c->myconf->vcpus > 0) ? c->myconf->vcpus : 8; vtaskmgr_init(&(c->vtaskmgr), nvcpus); } /* Guard: myapp_run() calls delaymgr_init + delaymgr_start before election * (needed for election timeout watchdog via delay_task_ms). * Skip re-init on the two-phase path; run on the legacy cluster_init() path * where delaymgr has not been initialized yet (precision_ms == 0). */ if (c->delaymgr.precision_ms == 0) { delaymgr_init(&(c->delaymgr), 200 /* precision_ms */, run_delay_tasks); } apiserver_init(c); /* Persistence: derive paths and init WAL/snapshot directories. * Recovery is deferred until after cluster_build_resources() runs inside * the HOOK_AFTER_EVENT below so that resmgr is fully populated before * restore_worker() tries to register per-host resource values. * * Guard: myapp_run() calls persist_set_workdir + persist_init + persist_start * before election (so WAL is ready when cluster_init_phase2 fires from vtask * context after election). Skip re-init if wal_path is already set. */ if (c->persist.wal_path[0] == '\0') { const char* workdir = c->myconf->workdir ? c->myconf->workdir : DEFAULT_WORKDIR; const char* name = c->myconf->name ? c->myconf->name : "controller"; persist_config_t pcfg = persist_config_defaults(); persist_set_workdir(&c->persist, workdir, name); rc = persist_init(&c->persist, &pcfg, c); if (rc != 0) { ERROR("persist_init failed: %d", rc); return rc; } } /* 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); if (rc != 0) { ERROR("persist_recover failed: %d", rc); persist_fini(&c->persist); return rc; } 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"); cluster_fini(c); 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) { cluster_fini(c); return rc; } } rc = cluster_init_phase2(c); if (rc != 0) { cluster_fini(c); return rc; } return 0; } static int make_heartbeat(cluster_t* c, endpoint_t* ep) { return xce_nop_req(ep); } static int check_worker_timeout_(cluster_t* c, worker_t* w) { worker_t* w1 = NULL; endpoint_t* ep; time_t now = time(NULL); int timeout; cluster_config_t* config; if (w == NULL) { return 0; } if (c == NULL) { return -EINVAL; } ep = w->ep; if (ep == NULL) { return 0; } /* grab lock on worker */ config = c->myconf->cluster_config; timeout = config->heartbeat_interval * config->inactive_heartbeat_count; if (now - ep->ts_last_in > timeout) { TRACE("%s worker %s becomes UNAVAIL as heartbeat timeout", __func__, w->host.name); mark_worker_unavail(c, w); return 0; } else { if (IS_WORKER_UNAVAIL(w->host.status)) { mark_worker_back(c, w); } } if (now - ep->ts_last_out >= config->heartbeat_interval) { if (DATAQ_EMPTY(&(ep->outq))) { int rc = 0; TRACE("%s send heartbeat to %s", __func__, w->host.name); rc = make_heartbeat(c, ep); if (rc < 0) { ERROR("%s make_heartbeat to %s failed", __func__, w->host.name); return rc; } } } return 0; } static int check_workers_timeout(void* args) { iterator_t it; cluster_t* c = (cluster_t*)args; int rc = 0; xlist_t l; int i; xlist_init(&l, 1024); lock(c->workermgr, RTTI_LOCK_READ); rc = iter_make(&it, c->workermgr); if (rc < 0) { unlock(c->workermgr); xlist_fini(&l); return rc; } for (;;) { worker_t* w = NULL; rc = iter_next(&it, (void**)&w); if ((rc < 0) || (rc == RTTI_ITER_NO_DATA)) { break; } if (w->ep == NULL) { continue; } /* check_worker_timeout_(c, w); */ if (w != NULL) { reference(w); xlist_store(&l, w); } } if (rc == RTTI_ITER_NO_DATA) { rc = 0; } iter_destroy(&it); unlock(c->workermgr); for (i = 0; i < l.count; i++) { worker_t* w = xlist_get(&l, i); if (w == NULL) { continue; } lock(w, RTTI_LOCK_WRITE); check_worker_timeout_(c, w); unlock(w); dereference(w); } xlist_fini(&l); return rc; } /* * cluster_register_listener: wire c->ep into iomgr's epoll watch. * * Called before election from myapp_run() so that incoming OP_requestVote * SSL connections are accepted and dispatched during the election phase. * cluster_start() is idempotent — it calls iomgr_monitor again but * iomgr_monitor is safe to call twice (epoll EPOLL_CTL_MOD handles it). */ int cluster_register_listener(cluster_t *c) { endpoint_set_owner(&(c->ep), c); endpoint_set_conn_handler(&(c->ep), (conn_handler_t)handle_connection); c->ep.iomgr = &(c->iomgr); return iomgr_monitor(&(c->ep)); } int cluster_start(cluster_t* c) { int rc; endpoint_set_owner(&(c->ep), c); endpoint_set_conn_handler(&(c->ep), (conn_handler_t)handle_connection); c->ep.iomgr = &(c->iomgr); /* Skip iomgr_monitor if cluster_register_listener already registered * this fd with epoll (ep->flags is set on successful registration). */ if (c->ep.flags == 0) { rc = iomgr_monitor(&(c->ep)); if (rc < 0) return rc; } /* Guard: myapp_run() starts iomgr and vtaskmgr before election so that * incoming OP_requestVote RPCs are accepted during the election phase. * Only start here if they were not already started (legacy cluster_init path). */ if (c->iomgr.worker == 0) iomgr_start(&(c->iomgr)); if (c->vtaskmgr.dispatcher == 0) vtaskmgr_start(&(c->vtaskmgr)); delaymgr_start(&(c->delaymgr)); /* Start persistence fsync thread and register WAL hooks */ rc = persist_start(&c->persist); if (rc != 0) { ERROR("persist_start failed: %d", rc); return rc; } /* Start HA heartbeat watchdog thread — P1-5 */ rc = ha_start_watchdog(&c->ha); if (rc != 0) { ERROR("ha_start_watchdog failed: %d", rc); return rc; } delay_task(&(c->delaymgr), c->myconf->cluster_config->heartbeat_interval, -1, (delay_task_entry_fn_t)check_workers_timeout, (void*)c); c->running = 1; return 0; } int cluster_run(cluster_t* c) { for (;;) { int stop; cbuf_get(&(c->wbuf), &stop); if (stop) { break; } } xce_run_hooks(&(c->hooksys), XCE_CONTROLLER_HOOK_CLUSTER, HOOK_BEFORE_EVENT, HOOK_DETAIL_CLUSTER_SHUTDOWN, c); c->running = 0; delaymgr_stop(&(c->delaymgr)); 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; } /* xce_request: defined in request.c, not yet in any public header */ extern int xce_request(endpoint_t *ep, xce_request_t *req); /* ── Task 4.2: ha_send_ctrlhello_to_all_standbys ─────────────────────────── * * Send OP_hello CtrlHelloRequest to every connected standby endpoint. * This triggers handle_ctrlhello on the receiving standby, which runs the * existing catch-up replication sequence (WAL streaming, config push). * * Must be called from the COMMITTING path in ha_elect_evaluate_and_signal * after ha->role is set to CTRL_PRIMARY_COMMITTING (Req 7.2). * * Thread safety: acquires ha->lock around standby_endpoints iteration. * Best-effort: failures per-endpoint are logged but do not abort the loop. */ void ha_send_ctrlhello_to_all_standbys(cluster_t *c) { if (!c) return; ha_state_t *ha = &c->ha; if (!ha->standby_endpoints) return; uint64_t my_lsn = atomic_load_explicit(&c->persist.committed_lsn, memory_order_acquire); uint64_t my_tag = c->myconf ? c->myconf->config_version_tag : 0; char *my_name = c->myconf ? c->myconf->name : ""; pthread_mutex_lock(&ha->lock); int n = ha->standby_endpoints->count; for (int i = 0; i < n; i++) { endpoint_t *ep = (endpoint_t *)zarray_get(ha->standby_endpoints, i); if (!ep) continue; ctrl_hello_req_t hello = { .name = my_name, .key = NULL, .applied_lsn = my_lsn, .config_tag = my_tag, }; xce_request_t req; memset(&req, 0, sizeof(req)); req.opcode = OPCode_hello; req.body_which = Request_body_ctrlhellomsg; req.body.ctrlhellomsg = &hello; int rc = xce_request(ep, &req); if (rc < 0) { WARN("ha_send_ctrlhello_to_all_standbys: xce_request to standby[%d] " "failed: %d", i, rc); } } pthread_mutex_unlock(&ha->lock); } 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; } /* iomgr owns the SSL_CTX lifetime; null ha's copy so ha_fini doesn't double-free */ c->ha.ssl_ctx = NULL; /* Stop HA watchdog thread before tearing down other subsystems — P1-5 */ ha_fini(&c->ha); /* Stop persistence before tearing down the rest */ persist_stop(&c->persist); persist_fini(&c->persist); delaymgr_fini(&(c->delaymgr)); vtaskmgr_fini(&(c->vtaskmgr)); endpoint_fini(&(c->ep)); xce_run_hooks(&(c->hooksys), XCE_CONTROLLER_HOOK_CLUSTER, HOOK_AFTER_EVENT, HOOK_DETAIL_CLUSTER_SHUTDOWN, c); cbuf_fini(&(c->wbuf)); iomgr_stop(&(c->iomgr)); /* join worker thread before freeing SSL_CTX */ iomgr_fini(&(c->iomgr)); destroy(c->jobmgr); destroy(c->workermgr); destroy(c->qmgr); destroy(c->resmgr); destroy(c->configmgr); mylock_fini(&(c->sched.lock)); zarray_fini(&(c->endpoints)); zarray_fini(&(c->configs)); xce_hook_fini(&(c->hooksys)); cleanup_openssl(); job_obj_fini(); xstring_table_fini(); return; }