myxce/src/lib/xcectx.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);
        }
    }

    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);
}

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)
{
    if ((ctx == NULL) || (handler == NULL)) {
        return -EINVAL;
    }

    ctx->handler = handler;
    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);
    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));
}