use nix build & meson & zlog
This commit is contained in:
Rongsong Shen
parent
0123623742
commit
346768a144
60 changed files with 13739 additions and 0 deletions
371
src/lib/xcectx.c
Normal file
371
src/lib/xcectx.c
Normal file
|
|
@ -0,0 +1,371 @@
|
|||
#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)
|
||||
{
|
||||
endpoint_t* ep = task->ctx.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 = ioctx_decode(&(task->ctx), &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_input(endpoint_t* ep)
|
||||
{
|
||||
xce_ctx_t* ctx = (xce_ctx_t*)ep->owner;
|
||||
vtask_t* task;
|
||||
int rc = 0;
|
||||
|
||||
task = vtask_alloc();
|
||||
if (task == NULL) {
|
||||
return -ENOMEM;
|
||||
}
|
||||
|
||||
rc = vtask_init(task, (vtask_func_t)&vtask_entry, ep);
|
||||
if (rc < 0) {
|
||||
SET_ERROR(-rc);
|
||||
|
||||
if (rc == -EPIPE) {
|
||||
/* connection broken */
|
||||
TRACE("broken pipe, ep %p", ep);
|
||||
}
|
||||
|
||||
FREEUP(task);
|
||||
return rc;
|
||||
}
|
||||
|
||||
rc = vtask_enq(&(ctx->vtaskmgr), task);
|
||||
if (rc < 0) {
|
||||
return rc;
|
||||
}
|
||||
|
||||
return 0;
|
||||
}
|
||||
|
||||
static int handle_output(endpoint_t* ep)
|
||||
{
|
||||
data_item_t* it;
|
||||
ioctx_t* ctx;
|
||||
int rc;
|
||||
|
||||
it = endpoint_deq(ep);
|
||||
if (it == NULL) {
|
||||
return 0;
|
||||
}
|
||||
|
||||
ctx = ITPTR_TO_IOCTX(it);
|
||||
|
||||
rc = ioctx_flush(ctx, ep->fd);
|
||||
ioctx_deref(ctx, 1);
|
||||
|
||||
return rc;
|
||||
}
|
||||
|
||||
static int handle_io(struct epoll_event* p)
|
||||
{
|
||||
endpoint_t* ep = (endpoint_t*)p->data.ptr;
|
||||
int rc = 0;
|
||||
|
||||
if ((p->events & EPOLLIN) != 0) {
|
||||
rc = handle_input(ep);
|
||||
}
|
||||
|
||||
if (rc >= 0) {
|
||||
if ((p->events & EPOLLOUT) != 0) {
|
||||
rc = handle_output(ep);
|
||||
}
|
||||
}
|
||||
|
||||
if (rc < 0) {
|
||||
xce_ctx_t* ctx = (xce_ctx_t*)ep->owner;
|
||||
iomgr_t* iomgr = &(ctx->iomgr);
|
||||
iomgr_unmonitor(iomgr, ep);
|
||||
endpoint_deref(ep);
|
||||
|
||||
xce_try_wakeup(ctx, -EIO);
|
||||
|
||||
ctx->state = XCE_CTX_DOWN;
|
||||
|
||||
return rc;
|
||||
}
|
||||
|
||||
return 0;
|
||||
}
|
||||
|
||||
int xce_ctx_init(xce_ctx_t* ctx,
|
||||
char* name,
|
||||
char* key)
|
||||
{
|
||||
int rc;
|
||||
|
||||
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;
|
||||
}
|
||||
|
||||
rc = cbuf_init(&(ctx->wbuf), 1, sizeof(int));
|
||||
if (rc < 0) {
|
||||
FREEUP(ctx->myinfo.name);
|
||||
FREEUP(ctx->myinfo.key);
|
||||
return rc;
|
||||
}
|
||||
|
||||
iomgr_init(&(ctx->iomgr));
|
||||
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);
|
||||
|
||||
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;
|
||||
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_handler(ctx->ep,
|
||||
handle_io, ctx);
|
||||
rc = iomgr_monitor(&(ctx->iomgr),
|
||||
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));
|
||||
}
|
||||
Loading…
Add table
Add a link
Reference in a new issue