forked from lei.chen/myxce
Rongsong Shen
6429ef0bff
Four bugs fixed: 1. persist.c: array job WAL loop aborted after index [0] when replication returned -ENODEV (no standbys). Loop guard now uses the return value of persist_wal_sync_append() which is 0 on local-disk success regardless of replication status. 2. join.c: WAL-recovered worker slot (status=UNAVAIL) caused add_worker() to return -EEXIST on reconnect. New EEXIST path in handle_worker_join() looks up the existing slot by name, updates its endpoint, and clears UNAVAIL so the normal hook/reply/on_worker_join flow proceeds. 3. xcectx.c: cbuf_get() in xce_hello() blocked indefinitely when the TCP connection dropped because no error handler was registered. Added handle_disconnect() which sets ctx->state=XCE_CTX_ERR and calls cbuf_tryput(-ECONNRESET) to unblock the waiter, allowing checking_connection() to detect the failure and rejoin. 4. cluster.c: persist_recover() was called before cluster_build_resources() populated resmgr, causing restore_worker() → add_hostattrs_resvals() to fail with EEXIST noise. Moved persist_recover() to after HOOK_AFTER_EVENT so resmgr is fully populated before WAL replay.
338 lines
7 KiB
C
338 lines
7 KiB
C
#include <errno.h>
|
|
#include <string.h>
|
|
#include <sys/socket.h>
|
|
|
|
#include "dataq.h"
|
|
#include "mylock.h"
|
|
#include "netio.h"
|
|
#include "typemap.h"
|
|
#include "utils.h"
|
|
#include "xcectx.h"
|
|
|
|
static int handle_get_reply(xce_ctx_t* ctx, xce_reply_t* reply)
|
|
{
|
|
xce_data_t* getdata = reply->data.getdata;
|
|
dataset_t* dataset = ctx->reply.u.get;
|
|
|
|
return extract_data(dataset, getdata);
|
|
}
|
|
|
|
static int handle_sub_reply(xce_ctx_t* ctx, xce_reply_t* reply)
|
|
{
|
|
if (reply->data_which != RequestReply_data_jobid) {
|
|
return -1;
|
|
}
|
|
|
|
ctx->reply.u.jobid = reply->data.jobid;
|
|
return 1;
|
|
}
|
|
|
|
static int handle_reply(xce_ctx_t* ctx, xce_reply_t* reply)
|
|
{
|
|
int rc = 0;
|
|
|
|
if (reply->status < 0) {
|
|
return reply->status;
|
|
}
|
|
|
|
switch (reply->opcode) {
|
|
case OPCode_get:
|
|
rc = handle_get_reply(ctx, reply);
|
|
break;
|
|
case OPCode_jobsub:
|
|
rc = handle_sub_reply(ctx, reply);
|
|
break;
|
|
case OPCode_clusterconf:
|
|
rc = -1;
|
|
if (reply->data_which == RequestReply_data_clusterconf) {
|
|
MOVE_PTR(ctx->reply.u.clusterconf, reply->data.clusterconf);
|
|
rc = 1;
|
|
}
|
|
break;
|
|
default:
|
|
rc = -ENOENT;
|
|
break;
|
|
}
|
|
|
|
return rc;
|
|
}
|
|
|
|
static int vtask_entry(vtask_t* task)
|
|
{
|
|
io_task_data_t* data = (io_task_data_t*)&(task->data->u.iodata);
|
|
endpoint_t* ep = data->ep;
|
|
xce_ctx_t* ctx = NULL;
|
|
int rc = 0;
|
|
xce_request_t request;
|
|
|
|
if (ep == NULL) {
|
|
return -EINVAL;
|
|
}
|
|
|
|
ctx = (xce_ctx_t*)ep->owner;
|
|
if (ctx == NULL) {
|
|
return -EINVAL;
|
|
}
|
|
|
|
rc = xce_decode(data->pkt, &request);
|
|
if (rc < 0) {
|
|
return rc;
|
|
}
|
|
|
|
if (request.opcode == OPCode_reply) {
|
|
int need_wake = 1;
|
|
int status = request.body.replymsg->status;
|
|
|
|
need_wake = handle_reply(ctx, request.body.replymsg);
|
|
if (need_wake) {
|
|
rc = xce_wakeup(ctx, (int)status);
|
|
}
|
|
} else {
|
|
int callnext = 1;
|
|
|
|
if ((request.opcode == OPCode_push)
|
|
&& (request.body_which == Request_body_pushmsg)) {
|
|
|
|
if (ctx->dataset == NULL) {
|
|
ctx->dataset = (dataset_t*)calloc(1, sizeof(dataset_t));
|
|
ctx->dataset->table = request.body.pushmsg->table;
|
|
}
|
|
|
|
callnext = extract_data(ctx->dataset, request.body.pushmsg);
|
|
if (callnext) {
|
|
FREEUP(request.body.pushmsg);
|
|
request.body_which = Request_body_datamsg;
|
|
MOVE_PTR(request.body.datamsg, ctx->dataset);
|
|
}
|
|
}
|
|
|
|
if (callnext) {
|
|
rc = (ctx->handler)(&request, ctx->handler_args);
|
|
}
|
|
}
|
|
|
|
free_Request(&request);
|
|
|
|
return 0;
|
|
}
|
|
|
|
static int handle_msg(endpoint_t* ep)
|
|
{
|
|
xce_ctx_t* ctx = NULL;
|
|
|
|
if (ep == NULL) {
|
|
return -EINVAL;
|
|
}
|
|
|
|
ctx = (xce_ctx_t*)ep->owner;
|
|
if (ctx == NULL) {
|
|
return -EINVAL;
|
|
}
|
|
|
|
return vtask_handle_msg(&(ctx->vtaskmgr), (vtask_func_t)vtask_entry,
|
|
(void*)ep);
|
|
}
|
|
|
|
/* Called by netio when the TCP connection drops (EPOLLHUP / EPOLLERR / read EOF).
|
|
* Unblocks any thread stuck in cbuf_get (e.g. xce_hello waiting for a reply)
|
|
* so that checking_connection can detect the failure and attempt to rejoin. */
|
|
static int handle_disconnect(endpoint_t* ep)
|
|
{
|
|
if (ep == NULL)
|
|
return 0;
|
|
|
|
xce_ctx_t* ctx = (xce_ctx_t*)ep->owner;
|
|
if (ctx == NULL)
|
|
return 0;
|
|
|
|
ctx->state = XCE_CTX_ERR;
|
|
|
|
/* Inject an error into wbuf so any blocked cbuf_get returns immediately.
|
|
* cbuf_tryput returns -EAGAIN if the buffer already has an item — that is
|
|
* fine: the existing item will wake the waiter. */
|
|
int err = -ECONNRESET;
|
|
cbuf_tryput(&ctx->wbuf, &err);
|
|
|
|
return 0;
|
|
}
|
|
|
|
int xce_ctx_init(xce_ctx_t* ctx, char* name, char* key)
|
|
{
|
|
int rc;
|
|
SSL_CTX* ssl_ctx = NULL;
|
|
|
|
if ((ctx == NULL) || (name == NULL) || (key == NULL)) {
|
|
return -EINVAL;
|
|
}
|
|
|
|
memset(ctx, 0x0, sizeof(*ctx));
|
|
ctx->state = XCE_CTX_ERR;
|
|
ctx->myinfo.name = strdup(name);
|
|
ctx->myinfo.key = strdup(key);
|
|
|
|
if ((ctx->myinfo.name == NULL) || (ctx->myinfo.key == NULL)) {
|
|
FREEUP(ctx->myinfo.name);
|
|
FREEUP(ctx->myinfo.key);
|
|
return -ENOMEM;
|
|
}
|
|
|
|
initialize_openssl();
|
|
ssl_ctx = create_ssl_context(NULL, NULL);
|
|
if (ssl_ctx == NULL) {
|
|
return -errno;
|
|
}
|
|
|
|
rc = cbuf_init(&(ctx->wbuf), 1, sizeof(int));
|
|
if (rc < 0) {
|
|
FREEUP(ctx->myinfo.name);
|
|
FREEUP(ctx->myinfo.key);
|
|
cleanup_openssl();
|
|
|
|
return rc;
|
|
}
|
|
|
|
iomgr_init(&(ctx->iomgr));
|
|
iomgr_set_ssl_ctx(&(ctx->iomgr), ssl_ctx);
|
|
|
|
vtaskmgr_init(&(ctx->vtaskmgr), 4);
|
|
|
|
ctx->ep = NULL;
|
|
ctx->state = XCE_CTX_INIT;
|
|
return 0;
|
|
}
|
|
|
|
void xce_ctx_fini(xce_ctx_t* ctx)
|
|
{
|
|
if (ctx == NULL) {
|
|
return;
|
|
}
|
|
|
|
xce_ctx_stop(ctx);
|
|
|
|
vtaskmgr_fini(&(ctx->vtaskmgr));
|
|
iomgr_fini(&(ctx->iomgr));
|
|
cbuf_fini(&(ctx->wbuf));
|
|
|
|
FREEUP(ctx->myinfo.name);
|
|
FREEUP(ctx->myinfo.key);
|
|
|
|
cleanup_openssl();
|
|
|
|
return;
|
|
}
|
|
|
|
int xce_ctx_set_handler(xce_ctx_t* ctx, xce_request_handler_t handler,
|
|
void* args)
|
|
{
|
|
if ((ctx == NULL) || (handler == NULL)) {
|
|
return -EINVAL;
|
|
}
|
|
|
|
ctx->handler = handler;
|
|
ctx->handler_args = args;
|
|
|
|
return 0;
|
|
}
|
|
|
|
int xce_ctx_set_endpoint(xce_ctx_t* ctx, endpoint_t* ep)
|
|
{
|
|
if ((ctx == NULL) || (ep == NULL)) {
|
|
return -EINVAL;
|
|
}
|
|
|
|
ctx->ep = ep;
|
|
ctx->ep->iomgr = &(ctx->iomgr);
|
|
|
|
return 0;
|
|
}
|
|
|
|
int xce_ctx_start(xce_ctx_t* ctx)
|
|
{
|
|
int rc;
|
|
|
|
if (ctx == NULL) {
|
|
return -EINVAL;
|
|
}
|
|
|
|
if ((ctx->state != XCE_CTX_INIT) || (ctx->ep == NULL)) {
|
|
return -EINVAL;
|
|
}
|
|
|
|
endpoint_set_owner(ctx->ep, ctx);
|
|
endpoint_set_msg_handler(ctx->ep, (msg_handler_t)handle_msg);
|
|
endpoint_set_err_handler(ctx->ep, (err_handler_t)handle_disconnect);
|
|
rc = iomgr_monitor(ctx->ep);
|
|
if (rc < 0) {
|
|
ctx->state = XCE_CTX_ERR;
|
|
return rc;
|
|
}
|
|
|
|
iomgr_start(&(ctx->iomgr));
|
|
vtaskmgr_start(&(ctx->vtaskmgr));
|
|
|
|
ctx->state = XCE_CTX_RUN;
|
|
|
|
return 0;
|
|
}
|
|
|
|
int xce_ctx_stop(xce_ctx_t* ctx)
|
|
{
|
|
if (ctx == NULL) {
|
|
return -EINVAL;
|
|
}
|
|
|
|
if (ctx->ep != NULL) {
|
|
iomgr_unmonitor(&(ctx->iomgr), ctx->ep);
|
|
ctx->ep = NULL;
|
|
}
|
|
iomgr_stop(&(ctx->iomgr));
|
|
|
|
ctx->state = XCE_CTX_DOWN;
|
|
|
|
return 0;
|
|
}
|
|
|
|
int xce_wakeup(xce_ctx_t* ctx, int status)
|
|
{
|
|
int rc = 0;
|
|
|
|
if (ctx == NULL) {
|
|
return -EINVAL;
|
|
}
|
|
|
|
rc = cbuf_tryput(&(ctx->wbuf), &status);
|
|
if (rc == -EAGAIN) {
|
|
/* pending data in wbuf, this is normal */
|
|
rc = 0;
|
|
}
|
|
|
|
return rc;
|
|
}
|
|
|
|
int xce_try_wakeup(xce_ctx_t* ctx, int status)
|
|
{
|
|
if (ctx == NULL) {
|
|
return -EINVAL;
|
|
}
|
|
|
|
return cbuf_tryput(&(ctx->wbuf), &status);
|
|
}
|
|
|
|
void xce_block_tasks(xce_ctx_t* ctx)
|
|
{
|
|
if (ctx == NULL) {
|
|
return;
|
|
}
|
|
|
|
endpoint_block_tasks(ctx->ep);
|
|
}
|
|
|
|
void xce_unblock_tasks(xce_ctx_t* ctx)
|
|
{
|
|
if (ctx == NULL) {
|
|
return;
|
|
}
|
|
|
|
endpoint_unblock_tasks(ctx->ep);
|
|
vtaskmgr_wakeup(&(ctx->vtaskmgr));
|
|
}
|