#include "config.h" #include #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" /* startup_validation.c — forward declaration */ int validate_heartbeat_config(uint32_t heartbeat_timeout_ms, uint32_t heartbeat_interval_ms, char *err_buf, size_t err_len); 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; } /* handle_wal_catchup_entry: dispatch handler for OPCode_walCatchupEntry. * Delegates to ha_apply_wal_catchup_entry (ha.c); on error logs and returns 0 * (skip, no crash). Standby remains in CTRL_STANDBY throughout (Req 3.4). * Task 3.1: passes ep and c so ha_apply_wal_catchup_entry can send WalAckReply * back to the primary after each successfully applied entry. */ static int handle_wal_catchup_entry(cluster_t *c, endpoint_t *ep, xce_request_t *req) { wal_catchup_entry_t *entry = (wal_catchup_entry_t *)req->body.walcatchupentrymsg; /* body may be NULL if the sender used Request_body_nomsg; skip gracefully */ if (!entry) return 0; /* Req 1.3 (ha-heartbeat-liveness): any msg from primary resets standby * liveness and steps down from CTRL_ELECTING. */ { ctrl_role_t role = (ctrl_role_t)atomic_load_explicit( &c->ha.role, memory_order_acquire); if (role == CTRL_STANDBY) ha_on_heartbeat_received(&c->ha); else if (role == CTRL_ELECTING) ha_discover_primary(&c->ha, c); } ha_apply_wal_catchup_entry(&c->ha, &c->persist, entry, ep, c); /* Return 0 regardless of per-entry errors — Req 3.3: skip, don't crash */ return 0; } /* handle_snapshot_transfer: dispatch handler for OPCode_snapshotTransfer. * Delegates to ha_handle_snapshot_chunk (ha.c). * On mid-transfer reconnect the caller (iomgr reconnect path) must delete * snap.part before the next transfer starts — this handler does not track * cross-chunk state itself (Req 4.4). * No reply is sent — primary learns of completion via next heartbeat reply. */ static int handle_snapshot_transfer(cluster_t *c, xce_request_t *req) { snapshot_transfer_t *chunk = (snapshot_transfer_t *)req->body.snapshottransfermsg; /* body may be NULL if sender used Request_body_nomsg; skip gracefully */ if (!chunk) return 0; /* Req 1.3 (ha-heartbeat-liveness): any msg from primary resets standby * liveness and steps down from CTRL_ELECTING. */ { ctrl_role_t role = (ctrl_role_t)atomic_load_explicit( &c->ha.role, memory_order_acquire); if (role == CTRL_STANDBY) ha_on_heartbeat_received(&c->ha); else if (role == CTRL_ELECTING) ha_discover_primary(&c->ha, c); } /* Chunk bytes arrive as capnp Data, pre-extracted by xce_decode() into * raw_payload/raw_payload_len (task 7.2). */ int rc = ha_handle_snapshot_chunk(&c->ha, &c->persist, c->persist.snap_path, chunk, chunk->raw_payload, chunk->raw_payload_len); if (rc != 0) ERROR("handle_snapshot_transfer: chunk %u rc=%d", chunk->chunk_index, rc); /* Return 0 — errors are logged; primary retries on next ctrlhello (Req 4.4) */ 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); /* Req 4.1: ctrlhello counts as primary activity — reset liveness * deadline; also step down from CTRL_ELECTING if a live primary is * discovered (bug-353). */ { ctrl_role_t role = (ctrl_role_t)atomic_load_explicit( &c->ha.role, memory_order_acquire); if (role == CTRL_STANDBY) ha_on_heartbeat_received(&c->ha); else if (role == CTRL_ELECTING) ha_discover_primary(&c->ha, c); } } 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: { /* Task 2.3: standby populates stateReport in every heartbeat reply so * the primary can track convergence (Req 1.1, 1.3, 1.4). */ state_report_t sr; memset(&sr, 0, sizeof(sr)); reply.status = 0; ha_nop_reply_state_report(&c->ha, c->myconf, &sr, &reply); xce_reply(ep, c->myhost, &reply); /* Req 1.3: any NOP from primary resets liveness deadline on standby. * Also steps down from CTRL_ELECTING if a live primary is discovered. */ { ctrl_role_t role = (ctrl_role_t)atomic_load_explicit( &c->ha.role, memory_order_acquire); if (role == CTRL_STANDBY) ha_on_heartbeat_received(&c->ha); else if (role == CTRL_ELECTING) ha_discover_primary(&c->ha, c); } 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: { /* Task 6.3: delegate vote-request handling to raft library. * * The freshness gate (config_tag / log_lsn comparison) is applied here * BEFORE calling raft_on_vote_request because raft_cb_vote_eligible * (on_vote_eligible) intentionally skips the freshness check — it * relies on the caller (this dispatcher) to enforce it first. * * raft_on_vote_request handles: * - stale-term deny (term < current_term) * - term upgrade (term > current_term → clear voted_for) * - role gate: RAFT_LEADER denies; COMMITTING gate in on_vote_eligible * - one-vote-per-term invariant (voted_for != -1 && voted_for != from_idx) * - on_persist before granting (durability guarantee) * * reply_term is set by raft_on_vote_request to the current term * (after any term upgrade), so vr_rep.term reflects the latest term. */ 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); /* Freshness gate: deny if candidate is not at least as fresh as us. * (Req 2.4: config_tag must be ≥ ours; if equal, log_lsn must be ≥ ours.) */ int freshness_ok = !(cand_tag < my_tag || (cand_tag == my_tag && cand_lsn < my_lsn)); /* Map candidate name to integer index for raft_on_vote_request. */ int from_idx = -1; const char *cand_id = vrq ? vrq->candidate_id : ""; if (cand_id && cand_id[0] != '\0') { const xce_cluster_config_t *pcfg = &c->myconf->peers_cfg; for (int i = 0; i < pcfg->candidates_count; i++) { if (strcmp(pcfg->candidates[i].name, cand_id) == 0) { from_idx = i; break; } } } uint64_t reply_term = 0; int granted = 0; if (freshness_ok && from_idx >= 0) { int r = raft_on_vote_request(&c->ha.raft, from_idx, cand_term, &reply_term); granted = (r == RAFT_GRANT) ? 1 : 0; } else { /* Freshness deny or unknown candidate: read current term for reply */ reply_term = atomic_load_explicit(&c->ha.term, memory_order_acquire); } request_vote_reply_t vr_rep; memset(&vr_rep, 0, sizeof(vr_rep)); vr_rep.term = reply_term; vr_rep.vote_granted = granted; 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); /* Inbound vote connections are for voting only. After the reply is * sent the connection is left open until the remote side closes it. * The primary will connect outbound to the newcomer from the heartbeat * task if the peer is missing from standby_endpoints. */ rc = 0; break; } case OPCode_replConfig: rc = handle_config_request(c, ep, &request); break; case OPCode_replConfigPush: /* Primary pushes config to this standby — save and reload */ rc = handle_repl_config_push(c, ep, &request); break; case OPCode_walCatchupEntry: /* Task 2.1: standby applies a single WAL catch-up entry from the * primary. applied_lsn advances on success; errors are logged and * skipped (Req 3.1, 3.3, 3.4, 3.5). No reply is sent — primary * learns of progress via the next heartbeat reply's stateReport. */ rc = handle_wal_catchup_entry(c, ep, &request); break; case OPCode_snapshotTransfer: /* Task 3.5c: standby receives one snapshot chunk from the primary. * Writes chunk payload to snap.part; on isLast renames and loads. * No reply — primary learns of completion via next heartbeat reply. * Req 4.2, 4.3, 4.4. */ rc = handle_snapshot_transfer(c, &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; /* Populate myhost_info so replies identify this node by its real name * (e.g. ctrl-1) rather than the generic "controller" placeholder. */ c->myhost_info.name = (c->myconf && c->myconf->name) ? c->myconf->name : from_controller.name; c->myhost_info.key = from_controller.key; c->myhost = &c->myhost_info; 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); } /* Req 2.4: reject misconfigured heartbeat timing before ha_init() */ { char hb_err[256]; if (validate_heartbeat_config(ha_cfg.heartbeat_timeout_ms, ha_cfg.heartbeat_interval_ms, hb_err, sizeof(hb_err)) != 0) { ERROR("startup validation failed: %s", hb_err); fprintf(stderr, "xce_controller: %s\n", hb_err); return -EINVAL; } } 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). * File lives under {workdir}/{myname}/data/ alongside WAL and snapshot. */ if (c->myconf->workdir && c->myconf->name) { char term_dir[1024]; snprintf(term_dir, sizeof(term_dir), "%s/%s/data", c->myconf->workdir, c->myconf->name); rc = ha_restore_term(&c->ha, term_dir); if (rc == 0) { INFO("HA: restored term=%lu from %s/ha-term.bin", (unsigned long)atomic_load(&c->ha.term), term_dir); } else if (rc != -ENOENT) { WARN("HA: ha_restore_term failed: %d", rc); } } /* ha_install_sigusr1() removed — failover_requested replaced by ha_arm_heartbeat_timeout_vtask */ /* 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; } } /* Role set by ha_elect_evaluate_and_signal() inside cluster_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; } /* ha_start_watchdog() removed — watchdog pthread replaced by * ha_arm_heartbeat_timeout_vtask (delaymgr) */ 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; } /* ── 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; }