forked from lei.chen/myxce
1182 lines
32 KiB
C
1182 lines
32 KiB
C
#include "config.h"
|
||
|
||
#include <errno.h>
|
||
#include <rtti.h>
|
||
#include <stdatomic.h>
|
||
#include <string.h>
|
||
#include <unistd.h>
|
||
|
||
#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;
|
||
|
||
/* Peek at opcode to decide whether to save raw bytes before xce_decode()
|
||
* frees the capnp segment. OPCode_upgradeConfig requires the raw packet
|
||
* for xce_upgrade_config_req_extract_raw(). */
|
||
uint8_t *raw_pkt_copy = NULL;
|
||
size_t raw_pkt_sz = 0;
|
||
{
|
||
io_packet_t *pkt = data->u.iodata.pkt;
|
||
if (pkt && pkt->data && pkt->sz > 0) {
|
||
struct capn peek_c;
|
||
struct Request peek_rq;
|
||
Request_ptr peek_rp;
|
||
capn_init_mem(&peek_c, (const uint8_t *)pkt->data, (size_t)pkt->sz, 0);
|
||
peek_rp.p = capn_getp(capn_root(&peek_c), 0, 1);
|
||
read_Request(&peek_rq, peek_rp);
|
||
capn_free(&peek_c);
|
||
|
||
if (peek_rq.opcode == OPCode_upgradeConfig) {
|
||
raw_pkt_copy = (uint8_t *)malloc((size_t)pkt->sz);
|
||
if (raw_pkt_copy) {
|
||
memcpy(raw_pkt_copy, pkt->data, (size_t)pkt->sz);
|
||
raw_pkt_sz = (size_t)pkt->sz;
|
||
}
|
||
}
|
||
}
|
||
}
|
||
|
||
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);
|
||
free_Request(&request);
|
||
free(raw_pkt_copy);
|
||
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_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, raw_pkt_copy, raw_pkt_sz);
|
||
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;
|
||
}
|
||
|
||
free_Request(&request);
|
||
free(raw_pkt_copy); /* NULL-safe: free(NULL) is a no-op */
|
||
|
||
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;
|
||
}
|
||
}
|
||
|
||
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;
|
||
}
|
||
|
||
/* 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;
|
||
|
||
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;
|
||
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 };
|
||
|
||
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;
|
||
|
||
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;
|
||
jobmgr_params.delta = c->myconf->cluster_config->jobmgr->delta;
|
||
workermgr_params.nslots = c->myconf->cluster_config->workermgr->nhash_slots;
|
||
workermgr_params.delta = c->myconf->cluster_config->workermgr->delta;
|
||
qmgr_params.nslots = c->myconf->cluster_config->qmgr->nhash_slots;
|
||
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);
|
||
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);
|
||
c->qmgr = makeInstance(&xkvmgr_type, &qmgr_params);
|
||
|
||
/* 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);
|
||
}
|
||
|
||
xstring_table_init(c->myconf->cluster_config->xstrings->nhash_slots);
|
||
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);
|
||
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);
|
||
}
|
||
|
||
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. */
|
||
{
|
||
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;
|
||
}
|
||
|
||
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);
|
||
|
||
rc = iomgr_monitor(&(c->ep));
|
||
|
||
if (rc < 0) {
|
||
return rc;
|
||
}
|
||
|
||
iomgr_start(&(c->iomgr));
|
||
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;
|
||
}
|
||
|
||
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_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;
|
||
}
|