From 8f796b310965ed35acd28848332c1580a3ce21f0 Mon Sep 17 00:00:00 2001 From: Rongsong Shen Date: Tue, 18 Nov 2025 11:21:24 +0800 Subject: [PATCH] security communication with openssl and simplify netio --- src/controller/cluster.c | 175 +++--- src/controller/dataex.c | 9 +- src/controller/job.c | 28 +- src/controller/main.c | 6 + src/controller/task.c | 9 +- src/controller/testsuite/test.c | 2 +- src/include/myconf.h | 2 + src/include/netio.h | 114 +++- src/include/request.h | 52 +- src/include/vtaskmgr.h | 26 +- src/lib/client.c | 13 +- src/lib/meson.build | 3 +- src/lib/netio.c | 912 +++++++++++++++++++++++--------- src/lib/request.c | 294 +++++----- src/lib/testsuite/test_idgen.c | 2 +- src/lib/vtaskmgr.c | 92 +++- src/lib/xceapi.c | 184 ++----- src/lib/xcectx.c | 119 ++--- src/meson.build | 8 +- src/nixSupport/xce.nix | 48 +- 20 files changed, 1176 insertions(+), 922 deletions(-) diff --git a/src/controller/cluster.c b/src/controller/cluster.c index 34e46674..49f021ad 100644 --- a/src/controller/cluster.c +++ b/src/controller/cluster.c @@ -48,8 +48,6 @@ static void handle_endpoint_error(endpoint_t* ep) return; } - iomgr_unmonitor(&(c->iomgr), ep); - priv = (ep_private_t*)ep->data; TRACE("endpoint %p priv data %p", ep, priv); @@ -79,11 +77,13 @@ static void handle_endpoint_error(endpoint_t* ep) } if (ep->idx >= 0) { + /* zarray_del will finalize ep */ zarray_del(&(c->endpoints), ep->idx); return; } - endpoint_deref(ep); + endpoint_fini(ep); + FREEUP(ep); return; } @@ -130,13 +130,15 @@ static int do_notify(cluster_t* c, endpoint_t* ep, xce_request_t* req) } static int myreq_handler(vtask_t* task) { - cluster_t* c = (cluster_t*)task->ctx.ep->owner; + 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; - rc = ioctx_decode(&(task->ctx), &request); + rc = xce_decode(data->u.iodata.pkt, &request); TRACE("%s cpu#%d opcode %d", __func__, task->vcpu, request.opcode); @@ -146,7 +148,7 @@ static int myreq_handler(vtask_t* task) case OPCode_shutdown: reply.status = 0; reply.data_which = RequestReply_data_nodata; - xce_reply(task->ctx.ep, c->myhost, &reply); + xce_reply(ep, c->myhost, &reply); stop = 1; cbuf_put(&(c->wbuf), &stop); break; @@ -154,44 +156,44 @@ static int myreq_handler(vtask_t* task) case OPCode_nop: reply.status = 0; reply.data_which = RequestReply_data_nodata; - xce_reply(task->ctx.ep, c->myhost, &reply); + xce_reply(ep, c->myhost, &reply); break; case OPCode_join: - rc = handle_join(c, task->ctx.ep, &request); + rc = handle_join(c, ep, &request); break; case OPCode_jobdispatch: - rc = handle_jobdispatch(c, task->ctx.ep, &request); + rc = handle_jobdispatch(c, ep, &request); break; case OPCode_jobsub: - rc = handle_jobsubmit(c, task->ctx.ep, &request); + rc = handle_jobsubmit(c, ep, &request); break; case OPCode_get: - rc = do_get_op(c, task->ctx.ep, &request); + rc = do_get_op(c, ep, &request); break; case OPCode_notify: - rc = do_notify(c, task->ctx.ep, &request); + rc = do_notify(c, ep, &request); break; case OPCode_clusterconf: - rc = handle_clusterconf(c, task->ctx.ep, &request); + rc = handle_clusterconf(c, ep, &request); break; case OPCode_update: - rc = do_update_op(c, task->ctx.ep, &request); + rc = do_update_op(c, ep, &request); break; case OPCode_jobcontrol: - rc = do_jobcontrol_op(c, task->ctx.ep, &request); + rc = do_jobcontrol_op(c, ep, &request); break; case OPCode_queuecontrol: - rc = do_queuecontrol_op(c, task->ctx.ep, &request); + rc = do_queuecontrol_op(c, ep, &request); break; case OPCode_runtasks: - rc = do_runtasks(c, task->ctx.ep, &request); + rc = do_runtasks(c, ep, &request); break; default: rc = -E_XCE_BAD_OPCODE; ERROR("Unknown opcode: %d", request.opcode); reply.status = rc; reply.data_which = RequestReply_data_nodata; - xce_reply(task->ctx.ep, c->myhost, &reply); + xce_reply(ep, c->myhost, &reply); break; } @@ -210,93 +212,21 @@ static int vtask_entry(vtask_t* task) return rc; } -static int handle_input(endpoint_t* ep) +static int handle_msg(endpoint_t* ep) { - cluster_t* c = (cluster_t*)ep->owner; - time_t now; - vtask_t* task; - int rc = 0; + cluster_t* c = NULL; - task = vtask_alloc(); - if (task == NULL) { - PERROR("vtask_alloc"); - - return -ENOMEM; + if (ep == NULL) { + return -EINVAL; } - rc = vtask_init(task, (vtask_func_t)&vtask_entry, ep); - if (rc < 0) { - SET_ERROR(-rc); - - INFO("vtask_init failed, maybe peer side closed the connection"); - - if (rc == -EPIPE) { - /* connection broken */ - TRACE("broken pipe, ep %p", ep); - - handle_endpoint_error(ep); - } - - FREEUP(task); - return rc; + c = (cluster_t*)ep->owner; + if (c == NULL) { + return -EINVAL; } - rc = vtask_enq(&(c->vtaskmgr), task); - if (rc < 0) { - PERROR("vtask_enq"); - - return rc; - } - - now = time(NULL); - npkts++; - - if (now - last_print_time > 10) { - TRACE("%ld npkts:%ld iops: %f\n", now, npkts, - (npkts - npkts_old) * 1.0 / (now - last_print_time)); - npkts_old = npkts; - last_print_time = now; - } - - return rc; -} - -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); - } - } - - return rc; + return vtask_handle_msg(&(c->vtaskmgr), (vtask_func_t)vtask_entry, + (void*)ep); } static int handle_connection(struct epoll_event* p) @@ -327,7 +257,9 @@ static int handle_connection(struct epoll_event* p) if (priv == NULL) { PERROR("ep_get_private"); - endpoint_deref(ep2); + endpoint_fini(ep2); + FREEUP(ep2); + return -ENOMEM; } @@ -342,17 +274,21 @@ static int handle_connection(struct epoll_event* p) PERROR("zarray_store"); FREEUP(priv); - endpoint_deref(ep2); + endpoint_fini(ep2); + FREEUP(ep2); return idx; } - ep2->idx = idx; - ep2->data = priv; + ep2->idx = idx; + ep2->data = priv; + ep2->iomgr = iomgr; - endpoint_set_handler(ep2, handle_io, c); + endpoint_set_owner(ep2, c); + endpoint_set_msg_handler(ep2, (msg_handler_t)handle_msg); + endpoint_set_err_handler(ep2, (err_handler_t)handle_endpoint_error); - iomgr_monitor(iomgr, ep2); + iomgr_monitor(ep2); return 0; } @@ -372,7 +308,8 @@ static void endpoint_killer(void* p) zarray_clear(&(c->endpoints), ep->idx); ep->idx = -1; } - endpoint_deref(ep); + endpoint_fini(ep); + FREEUP(ep); return; } @@ -511,7 +448,8 @@ static int xce_install_hooks(cluster_t* c) int cluster_init(cluster_t* c) { - int i, rc; + int i, rc; + SSL_CTX* ssl_ctx = NULL; if (c == NULL) { return -EINVAL; @@ -535,6 +473,12 @@ int cluster_init(cluster_t* c) mylock_init(&(c->sched.lock)); c->sched.ep = NULL; + initialize_openssl(); + ssl_ctx = create_ssl_context(c->myconf->certfile, c->myconf->keyfile); + if (ssl_ctx == NULL) { + return -errno; + } + kvmgr_init(&(c->configmgr), string_hash_key, 128, 128, (zarray_item_killer_t)config_obj_killer); @@ -550,6 +494,9 @@ int cluster_init(cluster_t* c) qmgr_init(&(c->qmgr), c->myconf->cluster_config->qmgr->nhash_slots, c->myconf->cluster_config->qmgr->delta); iomgr_init(&(c->iomgr)); + + iomgr_set_ssl_ctx(&(c->iomgr), ssl_ctx); + register_xstring(&(c->objmgr), c->myconf->cluster_config->xstrings->nhash_slots, c->myconf->cluster_config->xstrings->delta); @@ -560,9 +507,12 @@ int cluster_init(cluster_t* c) rc = endpoint_init(&(c->ep), AF_INET, SOCK_STREAM, c->myconf->cluster_config->port, 5); if (rc < 0) { + cleanup_openssl(); return rc; } + endpoint_set_peer(&(c->ep), c->myhost); + for (i = 0; i < c->ncontrollers; i++) { controller_init(&(c->controllers[i]), c); } @@ -583,8 +533,11 @@ int cluster_start(cluster_t* c) { int rc; - endpoint_set_handler(&(c->ep), handle_connection, c); - rc = iomgr_monitor(&(c->iomgr), &(c->ep)); + 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; @@ -620,7 +573,7 @@ int cluster_run(cluster_t* c) void cluster_fini(cluster_t* c) { vtaskmgr_fini(&(c->vtaskmgr)); - endpoint_deref(&(c->ep)); + endpoint_fini(&(c->ep)); xce_run_hooks(&(c->hooksys), XCE_CONTROLLER_HOOK_CLUSTER, HOOK_AFTER_EVENT, HOOK_DETAIL_CLUSTER_SHUTDOWN, c); @@ -641,5 +594,7 @@ void cluster_fini(cluster_t* c) xce_hook_fini(&(c->hooksys)); + cleanup_openssl(); + return; } diff --git a/src/controller/dataex.c b/src/controller/dataex.c index 8daed292..323ca36e 100644 --- a/src/controller/dataex.c +++ b/src/controller/dataex.c @@ -486,12 +486,5 @@ void deref_objs(int count, void** objs) int notify(cluster_t* c, endpoint_t* ep, xce_notify_data_t* data) { - ioctx_t* ctx = NULL; - - ctx = ioctx_alloc(ep, c->myhost); - if (ctx == NULL) { - return -ENOMEM; - } - - return xce_notify_req(ctx, data); + return xce_notify_req(ep, data); } diff --git a/src/controller/job.c b/src/controller/job.c index 2dc70bd2..bcbd10c8 100644 --- a/src/controller/job.c +++ b/src/controller/job.c @@ -128,7 +128,6 @@ static int dispatch_a_job_(cluster_t* c, endpoint_t* ep, xce_job_alloc_t* alloc) int rc = 0; job_t* j; worker_t* w; - ioctx_t* ioctx; xce_host_alloc_t** allocs; int nc; xce_job_exec_t einfo; @@ -201,21 +200,11 @@ static int dispatch_a_job_(cluster_t* c, endpoint_t* ep, xce_job_alloc_t* alloc) j->job.start_info->q_post_exec = q_post_exec; } - ioctx = ioctx_alloc(w->ep, c->myhost); - if (ioctx == NULL) { - ERROR("memory allocate failed"); - free_launchinfo(&launchinfo); - obj_deref(j); - DEREF_OBJ(w); - - return -ENOMEM; - } - einfo.jobid = alloc->jobid; einfo.start_info = j->job.start_info; einfo.launchinfo = &launchinfo; - rc = xce_startjob_req(ioctx, &einfo); + rc = xce_startjob_req(w->ep, &einfo); free_launchinfo(&launchinfo); @@ -256,7 +245,6 @@ int job_ctrl(cluster_t* c, xce_jobctrl_t* ctrinfo) if (j->job.allocs != NULL) { xce_host_alloc_t** allocs = j->job.allocs->allocs; worker_t* w; - ioctx_t* ioctx; /* job control command will be sent to first node * for parallel jobs @@ -265,16 +253,10 @@ int job_ctrl(cluster_t* c, xce_jobctrl_t* ctrinfo) if (w != NULL) { /* FIXME: better method to check host status */ if ((w->ep != NULL) && (w->host.status == 0)) { - ioctx = ioctx_alloc(w->ep, c->myhost); - if (ioctx != NULL) { - rc = xce_jobctrl_req(ioctx, ctrinfo); - if (rc < 0) { - ERROR("job ctrl %ld on host %s returns %d", jobid, - w->host.name, rc); - } - } else { - ERROR("memory allocate failed"); - rc = -ENOMEM; + rc = xce_jobctrl_req(w->ep, ctrinfo); + if (rc < 0) { + ERROR("job ctrl %ld on host %s returns %d", jobid, + w->host.name, rc); } } else { ERROR("worker %s is not in correct state, status %x ep %p", diff --git a/src/controller/main.c b/src/controller/main.c index de4e0a36..ce2cd500 100644 --- a/src/controller/main.c +++ b/src/controller/main.c @@ -89,6 +89,8 @@ int main(int argc, char* argv[]) int logsize = 64 * 1024 * 1024; int logcount = 10; char conf_file[XCE_MAXPATH_LEN + 1]; + char key_file[XCE_MAXPATH_LEN + 1]; + char cert_file[XCE_MAXPATH_LEN + 1]; char myname[XCE_MAXNAME_LEN + 1]; memset(&arguments, 0x0, sizeof(arguments)); @@ -124,9 +126,13 @@ int main(int argc, char* argv[]) memset(&conf_file, 0x0, sizeof(conf_file)); snprintf(conf_file, XCE_MAXPATH_LEN, "%s/%s/conf/current_config", arguments.workdir, myname); + snprintf(key_file, XCE_MAXPATH_LEN, "%s/certs/xce.key", arguments.workdir); + snprintf(cert_file, XCE_MAXPATH_LEN, "%s/certs/xce.crt", arguments.workdir); myconf.cluster_config = NULL; myconf.conffile = conf_file; + myconf.certfile = cert_file; + myconf.keyfile = key_file; myapp_run(&myconf); diff --git a/src/controller/task.c b/src/controller/task.c index 509a0e0b..7f089634 100644 --- a/src/controller/task.c +++ b/src/controller/task.c @@ -11,7 +11,6 @@ static int run_tasks_on_host(cluster_t* c, endpoint_t* ep, uint64_t jobid, xce_taskhost_t* info) { worker_t* w; - ioctx_t* ioctx; int rc; xce_taskrun_t runmsg = { .nhosts = 1, .jobid = jobid, @@ -28,13 +27,7 @@ static int run_tasks_on_host(cluster_t* c, endpoint_t* ep, uint64_t jobid, return -EINVAL; } - ioctx = ioctx_alloc(w->ep, c->myhost); - if (ioctx == NULL) { - obj_deref(w); - return -ENOMEM; - } - - rc = xce_runtask_req(ioctx, &runmsg); + rc = xce_runtask_req(w->ep, &runmsg); obj_deref(w); diff --git a/src/controller/testsuite/test.c b/src/controller/testsuite/test.c index e3b0b445..9f6cbeb0 100644 --- a/src/controller/testsuite/test.c +++ b/src/controller/testsuite/test.c @@ -527,7 +527,7 @@ int cmd_reslist(int argc, char* argv[]) res[i]->interval, res[i]->desc); } - free(res); + free_Resource_list(rc, res); return 0; } diff --git a/src/include/myconf.h b/src/include/myconf.h index 6e10f95a..61b93b91 100644 --- a/src/include/myconf.h +++ b/src/include/myconf.h @@ -16,6 +16,8 @@ struct logging { typedef struct { cluster_config_t* cluster_config; char* conffile; + char* certfile; + char* keyfile; } myconf_t; extern cluster_config_t* get_default_conf(void*); diff --git a/src/include/netio.h b/src/include/netio.h index 682614e2..a56e142d 100644 --- a/src/include/netio.h +++ b/src/include/netio.h @@ -2,6 +2,8 @@ #define _NETIO_H_ 1 +#include +#include #include #include #include @@ -9,63 +11,127 @@ #include "dataq.h" #include "mylock.h" +#include "xce/types.h" -typedef int (*msghandler_t)(struct epoll_event*); +typedef int (*conn_handler_t)(struct epoll_event*); +typedef int (*msg_handler_t)(void*); +typedef int (*err_handler_t)(void*); #define IO_MON_INPUT (0x01) #define IO_MON_OUTPUT (0x02) typedef struct { - uint16_t sz; - char* data; + data_item_t link; + uint16_t sz; + char* data; } io_packet_t; +#define ITPTR_TO_PACKET(it) \ + ((io_packet_t*)(((char*)it) - offsetof(io_packet_t, link))) + typedef struct { mylock_t lock; int epoll_fd; struct epoll_event ev; pthread_t worker; + SSL_CTX* ssl_ctx; } iomgr_t; +typedef enum { + FSM_IO_READ = 0, + FSM_IO_WRITE +} io_fsm_op_t; + +typedef enum { + FSM_STATE_INIT = 0, + FSM_STATE_READ_PKT_HEAD, + FSM_STATE_READ_PKT, + FSM_STATE_WRITE_PKT_HEAD, + FSM_STATE_WRITE_PKT, + FSM_STATE_DONE, + FSM_STATE_ERROR +} io_fsm_state_t; + typedef struct { - mylock_t lock; - iomgr_t* iomgr; - int fd; - int idx; - int flags; - unsigned int busy : 1; - unsigned int needfree : 1; - uint32_t refcount; - dataq_t outq; - uint64_t seq; - msghandler_t handler; - void* owner; - void* data; + void* ep; + io_fsm_op_t op; + io_fsm_state_t state; + io_packet_t* pkt; + size_t iocount; +} io_fsm_t; + +typedef struct { + mylock_t lock; + iomgr_t* iomgr; + SSL* ssl; + int fd; + int idx; + int flags; + unsigned int busy : 1; + unsigned int ssl_handshake_done : 1; + dataq_t outq; + uint64_t seq; + conn_handler_t conn_handler; + void* owner; + void* data; + io_fsm_t fsm_in; + io_fsm_t fsm_out; + xce_peerinfo_t* peer; + msg_handler_t msg_handler; + err_handler_t err_handler; } endpoint_t; -extern int netio_write(int, char*, int); -extern int netio_read(int, char*, int); -extern int write_packet(int, io_packet_t*); +#define INVOKE_MSG_HANDLER(ep) \ + do { \ + TRACE("%s msg handler %p", __func__, (ep)->msg_handler); \ + if ((ep)->msg_handler != NULL) { \ + (ep)->msg_handler(ep); \ + /* passing ownership of packet to upper layer */ \ + (ep)->fsm_in.pkt = NULL; \ + } \ + } while (0) + +#define INVOKE_ERR_HANDLER(ep) \ + do { \ + if ((ep)->err_handler != NULL) { \ + (ep)->err_handler(ep); \ + } \ + } while (0) + +typedef struct { + endpoint_t* ep; + io_packet_t* pkt; +} io_task_data_t; + extern int alloc_packet(io_packet_t*); -extern int read_packet(int, io_packet_t*); extern void free_packet(io_packet_t*); extern endpoint_t* endpoint_get(void); extern int endpoint_init(endpoint_t*, int, int, int, int); extern int endpoint_connect(endpoint_t*, char*, int); extern int endpoint_accept(endpoint_t*, endpoint_t*); -extern int endpoint_set_handler(endpoint_t*, msghandler_t, void*); +extern int endpoint_set_conn_handler(endpoint_t*, conn_handler_t); extern int endpoint_enq(endpoint_t*, data_item_t*); extern data_item_t* endpoint_deq(endpoint_t*); -extern endpoint_t* endpoint_ref(endpoint_t*); -extern void endpoint_deref(endpoint_t*); extern uint64_t endpoint_getseq(endpoint_t*); extern void endpoint_block_tasks(endpoint_t*); extern void endpoint_unblock_tasks(endpoint_t*); extern int endpoint_is_task_blocked(endpoint_t*); +extern int endpoint_set_owner(endpoint_t*, void*); +extern int endpoint_set_msg_handler(endpoint_t*, msg_handler_t); +extern int endpoint_set_err_handler(endpoint_t*, err_handler_t); +extern void endpoint_fini(endpoint_t*); +extern int endpoint_set_peer(endpoint_t*, xce_peerinfo_t*); + +extern int io_fsm_start(io_fsm_t*, io_packet_t*); + +extern void initialize_openssl(void); +extern void cleanup_openssl(void); +extern SSL_CTX* create_ssl_context(char*, char*); +extern void iomgr_set_ssl_ctx(iomgr_t*, SSL_CTX*); extern int iomgr_init(iomgr_t*); -extern int iomgr_monitor(iomgr_t*, endpoint_t*); +extern int iomgr_monitor(endpoint_t*); extern int iomgr_unmonitor(iomgr_t*, endpoint_t*); extern int iomgr_start(iomgr_t*); extern int iomgr_fini(iomgr_t*); diff --git a/src/include/request.h b/src/include/request.h index 48eb9d21..aa784a7b 100644 --- a/src/include/request.h +++ b/src/include/request.h @@ -15,61 +15,45 @@ typedef struct { int wait; } reqbar_t; -typedef struct { - data_item_t link; - endpoint_t* ep; - xce_peerinfo_t* from; - io_packet_t pkt; - struct capn c; - struct Request rq; - uint32_t nrefs; -} ioctx_t; - -#define ITPTR_TO_IOCTX(p) ((ioctx_t*)(((char*)p) - offsetof(ioctx_t, link))) - typedef int (*xce_check_fn_t)(void*); -extern ioctx_t* ioctx_alloc(endpoint_t*, xce_peerinfo_t*); -extern int ioctx_ref(ioctx_t*); -extern void ioctx_deref(ioctx_t*, int); -extern int ioctx_flush(ioctx_t*, int); -extern int ioctx_encode(ioctx_t*, xce_request_t*); -extern int ioctx_decode(ioctx_t*, xce_request_t*); +extern int xce_encode(xce_request_t*, io_packet_t*); +extern int xce_decode(io_packet_t*, xce_request_t*); extern int xce_reply(endpoint_t*, xce_peerinfo_t*, xce_reply_t*); -extern int xce_hello_req(ioctx_t*, char*); -extern int xce_nop_req(ioctx_t*); -extern int xce_shutdown_req(ioctx_t*); +extern int xce_hello_req(endpoint_t*, char*); +extern int xce_nop_req(endpoint_t*); +extern int xce_shutdown_req(endpoint_t*); extern void init_reply_by_req(xce_reply_t*, xce_request_t*); -extern int xce_join_req(ioctx_t*, int, hostattrs_t*); +extern int xce_join_req(endpoint_t*, int, hostattrs_t*); extern int xce_pushconfig_req(endpoint_t*, xce_peerinfo_t*, xce_configdata_t*); extern int xce_push_data_req(endpoint_t*, xce_peerinfo_t*, xce_data_t*, xce_check_fn_t, void*, int); -extern int xce_workerlist_req(ioctx_t*, xce_peerinfo_t*); +extern int xce_workerlist_req(endpoint_t*, xce_peerinfo_t*); extern int xce_reply_workerlist(endpoint_t*, uint64_t, int, xce_peerinfo_t*, xce_workerlist_t*); -extern int xce_jobdispatch_req(ioctx_t*, xce_job_dispatch_t*); -extern int xce_jobsubmit_req(ioctx_t*, xce_job_submit_t*); +extern int xce_jobdispatch_req(endpoint_t*, xce_job_dispatch_t*); +extern int xce_jobsubmit_req(endpoint_t*, xce_job_submit_t*); extern int xce_reply_jobdispatch(endpoint_t*, uint64_t, int32_t, xce_peerinfo_t*); extern int xce_reply_jobid(endpoint_t*, int, uint64_t, uint64_t, xce_peerinfo_t*); -extern int xce_joblist_req(ioctx_t*, xce_peerinfo_t*); +extern int xce_joblist_req(endpoint_t*, xce_peerinfo_t*); extern int xce_reply_joblist(endpoint_t*, uint64_t, int, xce_peerinfo_t*, xce_job_list_t*); -extern int xce_startjob_req(ioctx_t*, xce_job_exec_t*); -extern int xce_clusterconf_req(ioctx_t*); +extern int xce_startjob_req(endpoint_t*, xce_job_exec_t*); +extern int xce_clusterconf_req(endpoint_t*); extern int xce_reply_clusterconf(endpoint_t*, uint64_t, int, xce_peerinfo_t*, xce_clusterconf_t*); -extern int xce_get_req(ioctx_t*, xce_get_t*); +extern int xce_get_req(endpoint_t*, xce_get_t*); extern int xce_reply_get(endpoint_t*, xce_peerinfo_t*, xce_reply_t*, xce_check_fn_t, void*, int); -extern int xce_notify_req(ioctx_t*, xce_notify_data_t*); +extern int xce_notify_req(endpoint_t*, xce_notify_data_t*); extern int extract_data(dataset_t*, xce_data_t*); -extern int xce_update_req(ioctx_t*, xce_update_data_t*); -extern int xce_jobctrl_req(ioctx_t*, xce_jobctrl_t*); -extern int xce_queuectrl_req(ioctx_t*, xce_queuectrl_t*); -extern int xce_runtask_req(ioctx_t*, xce_taskrun_t*); +extern int xce_update_req(endpoint_t*, xce_update_data_t*); +extern int xce_jobctrl_req(endpoint_t*, xce_jobctrl_t*); +extern int xce_queuectrl_req(endpoint_t*, xce_queuectrl_t*); +extern int xce_runtask_req(endpoint_t*, xce_taskrun_t*); #endif diff --git a/src/include/vtaskmgr.h b/src/include/vtaskmgr.h index f9ba05d8..433f7748 100644 --- a/src/include/vtaskmgr.h +++ b/src/include/vtaskmgr.h @@ -12,15 +12,26 @@ typedef int (*vtask_func_t)(void*); +#define VTASK_IO 0x01 +#define VTASK_GENERIC 0x02 + typedef struct { - void* owner; - ioctx_t ctx; - int vcpu; - vtask_func_t entry; + int vtype; + union { + void* generic; + io_task_data_t iodata; + } u; +} vtask_data_t; + +typedef struct { + data_item_t link; + void* owner; + vtask_data_t* data; + int vcpu; + vtask_func_t entry; } vtask_t; -#define ITPTR_TO_VTASK(p) \ - ((vtask_t*)(((char*)p) - offsetof(ioctx_t, link) - offsetof(vtask_t, ctx))) +#define ITPTR_TO_VTASK(p) ((vtask_t*)(((char*)p) - offsetof(vtask_t, link))) typedef enum { VCPU_IDLE = 0, @@ -47,10 +58,11 @@ typedef struct { } vtaskmgr_t; extern vtask_t* vtask_alloc(); -extern int vtask_init(vtask_t*, vtask_func_t, endpoint_t*); +extern int vtask_init(vtask_t*, vtask_func_t, void*); extern void vtask_fini(vtask_t*); extern int vtask_enq(vtaskmgr_t*, vtask_t*); +extern int vtask_handle_msg(vtaskmgr_t*, vtask_func_t, void* arg); extern int vtaskmgr_init(vtaskmgr_t*, int); extern int vtaskmgr_start(vtaskmgr_t*); diff --git a/src/lib/client.c b/src/lib/client.c index 7673decb..f993de36 100644 --- a/src/lib/client.c +++ b/src/lib/client.c @@ -53,6 +53,9 @@ int xcecli_ctx_init(xcecli_ctx_t* c, xce_client_config_t* cfg) endpoint_init(&(ctx->ep), AF_INET, SOCK_STREAM, 0, 0); + /* set information for endpoint */ + endpoint_set_peer(&(ctx->ep), &(ctx->ctx.myinfo)); + xce_ctx_set_endpoint(&(ctx->ctx), &(ctx->ep)); *c = (void*)ctx; @@ -68,7 +71,7 @@ void xcecli_ctx_fini(xcecli_ctx_t c) } xce_ctx_fini(&(ctx->ctx)); - endpoint_deref(&(ctx->ep)); + endpoint_fini(&(ctx->ep)); free(ctx); } @@ -118,11 +121,13 @@ int xcecli_start(xcecli_ctx_t c) rc = endpoint_connect(&(ctx->ep), ctx->conf->masters[0], ctx->conf->master_ports[0]); if (rc < 0) { - endpoint_deref(&(ctx->ep)); + endpoint_fini(&(ctx->ep)); return rc; } - return xce_ctx_start(&(ctx->ctx)); + rc = xce_ctx_start(&(ctx->ctx)); + + return rc; } int xcecli_stop(xcecli_ctx_t c) @@ -134,7 +139,7 @@ int xcecli_stop(xcecli_ctx_t c) } xce_ctx_stop(&(ctx->ctx)); - endpoint_deref(&(ctx->ep)); + endpoint_fini(&(ctx->ep)); return 0; } diff --git a/src/lib/meson.build b/src/lib/meson.build index 846f7832..63cf70ed 100644 --- a/src/lib/meson.build +++ b/src/lib/meson.build @@ -41,8 +41,9 @@ thread_dep = dependency('threads') c_capnproto_dep = dependency('c-capnproto') tinylog_dep = dependency('tinylog') unity_dep = dependency('unity') +ssl_dep = dependency('openssl') -libxce_deps = [c_capnproto_dep, tinylog_dep, thread_dep] +libxce_deps = [c_capnproto_dep, ssl_dep, tinylog_dep, thread_dep] libxce = static_library('xce', libxce_src, diff --git a/src/lib/netio.c b/src/lib/netio.c index 6dea37df..fb8f1843 100644 --- a/src/lib/netio.c +++ b/src/lib/netio.c @@ -2,6 +2,8 @@ #include #include #include +#include +#include #include #include #include @@ -14,8 +16,187 @@ #include "request.h" #include "utils.h" -static int iomgr_update_poll(iomgr_t*, endpoint_t*, int); -static void endpoint_fini(endpoint_t*); +static int iomgr_update_poll(iomgr_t*, endpoint_t*, int); +static int endpoint_update_epoll(endpoint_t*); + +static int io_fsm_init(io_fsm_t* fsm, endpoint_t* ep, io_fsm_op_t op) +{ + if ((fsm == NULL) || (ep == NULL) + || ((op != FSM_IO_READ) && (op != FSM_IO_WRITE))) { + return -EINVAL; + } + + fsm->ep = ep; + fsm->op = op; + fsm->state = FSM_STATE_INIT; + fsm->pkt = NULL; + fsm->iocount = 0; + + return 0; +} + +int io_fsm_start(io_fsm_t* fsm, io_packet_t* pkt) +{ + endpoint_t* ep; + + if ((fsm == NULL) || (pkt == NULL)) { + return -EINVAL; + } + + ep = (endpoint_t*)fsm->ep; + if ((ep == NULL) || (ep->ssl == NULL)) { + return -EINVAL; + } + + if ((fsm->state != FSM_STATE_INIT) && (fsm->state != FSM_STATE_DONE)) { + return -EINVAL; + } + + switch (fsm->op) { + case FSM_IO_READ: + fsm->state = FSM_STATE_READ_PKT_HEAD; + break; + case FSM_IO_WRITE: + fsm->state = FSM_STATE_WRITE_PKT_HEAD; + break; + } + + fsm->pkt = pkt; + fsm->iocount = 0; + + return 0; +} + +static int io_fsm_do(io_fsm_t* fsm, char* data, size_t len) +{ + int rc = 0; + size_t count; + endpoint_t* ep = (endpoint_t*)fsm->ep; + + count = len - fsm->iocount; + + switch (fsm->op) { + case FSM_IO_READ: + rc = SSL_read(ep->ssl, data + fsm->iocount, count); + break; + case FSM_IO_WRITE: + rc = SSL_write(ep->ssl, data + fsm->iocount, count); + break; + } + + if (rc <= 0) { + int err = 0; + int ecode = 0; + + ecode = errno; + err = SSL_get_error(ep->ssl, rc); + switch (err) { + case SSL_ERROR_WANT_READ: + case SSL_ERROR_WANT_WRITE: + rc = 0; + break; + case SSL_ERROR_SYSCALL: + if ((rc < 0) && ((ecode == EAGAIN) || (ecode == EWOULDBLOCK))) { + rc = 0; + } else { + fsm->state = FSM_STATE_ERROR; + rc = -errno; + } + if (rc == 0) { + /* we meet EOF */ + fsm->state = FSM_STATE_ERROR; + } + break; + default: + fsm->state = FSM_STATE_ERROR; + rc = -errno; + break; + } + + return rc; + } + + fsm->iocount += rc; + count -= rc; + if (count == 0) { + rc = fsm->iocount; + + switch (fsm->state) { + case FSM_STATE_READ_PKT_HEAD: + fsm->state = FSM_STATE_READ_PKT; + fsm->iocount = 0; + if (alloc_packet(fsm->pkt) < 0) { + fsm->state = FSM_STATE_ERROR; + rc = -ENOMEM; + } + break; + case FSM_STATE_READ_PKT: + case FSM_STATE_WRITE_PKT: + fsm->state = FSM_STATE_DONE; + fsm->iocount = 0; + break; + case FSM_STATE_WRITE_PKT_HEAD: + fsm->state = FSM_STATE_WRITE_PKT; + fsm->iocount = 0; + break; + case FSM_STATE_DONE: + break; + case FSM_STATE_INIT: + case FSM_STATE_ERROR: + /* should never go here and make compiler happy */ + rc = -EINVAL; + break; + } + } + + return rc; +} + +static int io_fsm_perform(io_fsm_t* fsm) +{ + char* data; + size_t len; + + if ((fsm == NULL) || (fsm->state == FSM_STATE_INIT) + || (fsm->state == FSM_STATE_DONE) || (fsm->state == FSM_STATE_ERROR)) { + return -EINVAL; + } + + switch (fsm->state) { + case FSM_STATE_READ_PKT_HEAD: + case FSM_STATE_WRITE_PKT_HEAD: + data = (char*)&(fsm->pkt->sz); + len = sizeof(fsm->pkt->sz); + break; + case FSM_STATE_READ_PKT: + case FSM_STATE_WRITE_PKT: + data = fsm->pkt->data; + len = fsm->pkt->sz; + break; + default: + return -EINVAL; + } + + return io_fsm_do(fsm, data, len); +} + +static int set_fd_nonblocking(int fd) +{ + int flags = 0; + int rc = 0; + + flags = fcntl(fd, F_GETFL, 0); + if (flags < 0) { + return -errno; + } + + rc = fcntl(fd, F_SETFL, flags | O_NONBLOCK); + if (rc < 0) { + return -errno; + } + + return rc; +} static int socket_create(int domain, int type, int port, int backlog) { @@ -127,70 +308,6 @@ static int socket_accept(int sock) return s; } -int netio_write(int fd, char* data, int sz) -{ - int nc = 0; - - while (sz > 0) { - int rc; - rc = write(fd, data + nc, sz); - if (rc < 0) { - SET_ERROR(errno); - - return -errno; - } - - sz -= rc; - nc += rc; - } - - return nc; -} - -int netio_read(int fd, char* data, int sz) -{ - int nc = 0; - - while (sz > 0) { - int rc; - - rc = read(fd, data + nc, sz); - if (rc < 0) { - SET_ERROR(errno); - - return -errno; - } - - if (rc == 0) { - /* network read return 0, it seems remote - * size has tear down the connection - */ - return -EPIPE; - } - - sz -= rc; - nc += rc; - } - - return nc; -} - -int write_packet(int fd, io_packet_t* pkt) -{ - int rc; - - rc = netio_write(fd, (char*)&(pkt->sz), sizeof(pkt->sz)); - if (rc < 0) { - return -errno; - } - - rc = netio_write(fd, pkt->data, pkt->sz); - if (rc < 0) { - return -errno; - } - return rc; -} - int alloc_packet(io_packet_t* pkt) { if (pkt->sz <= 0) { @@ -208,37 +325,168 @@ int alloc_packet(io_packet_t* pkt) return 1; } -int read_packet(int fd, io_packet_t* pkt) + +void free_packet(io_packet_t* pkt) +{ + if (pkt != NULL) { + FREEUP(pkt->data); + } +} + +#define MAX_EPOLL_EVENTS (5) + +static int handle_input(endpoint_t* ep) +{ + int rc = 0; + + if ((ep->fsm_in.state == FSM_STATE_INIT) + || (ep->fsm_in.state == FSM_STATE_DONE)) { + io_packet_t* pkt; + + pkt = ep->fsm_in.pkt; + if (pkt == NULL) { + pkt = (io_packet_t*)calloc(1, sizeof(*pkt)); + if (pkt == NULL) { + return -ENOMEM; + } + } else { + free_packet(pkt); + ep->fsm_in.pkt = NULL; + } + + io_fsm_start(&(ep->fsm_in), pkt); + } + + rc = io_fsm_perform(&(ep->fsm_in)); + if (rc < 0) { + return rc; + } + + if (ep->fsm_in.state == FSM_STATE_DONE) { + INVOKE_MSG_HANDLER(ep); + } + + return 0; +} + +static int handle_output(endpoint_t* ep) +{ + int rc = 0; + + if ((ep->fsm_out.state == FSM_STATE_INIT) + || (ep->fsm_out.state == FSM_STATE_DONE)) { + data_item_t* it = NULL; + io_packet_t* pkt = NULL; + + it = endpoint_deq(ep); + if (it == NULL) { + return 0; + } + + pkt = ITPTR_TO_PACKET(it); + rc = io_fsm_start(&(ep->fsm_out), pkt); + if (rc < 0) { + return rc; + } + } + + rc = io_fsm_perform(&(ep->fsm_out)); + if (rc < 0) { + return rc; + } + + if (ep->fsm_out.state == FSM_STATE_DONE) { + /* handle output done */ + io_packet_t* pkt = ep->fsm_out.pkt; + + if (pkt != NULL) { + free_packet(pkt); + FREEUP(pkt); + } + + ep->fsm_out.pkt = NULL; + + endpoint_update_epoll(ep); + } + + return 0; +} + +static int handle_error(endpoint_t* ep) +{ + iomgr_t* iomgr = (iomgr_t*)ep->owner; + + if (iomgr != NULL) { + iomgr_unmonitor(iomgr, ep); + } + + INVOKE_ERR_HANDLER(ep); + + return 0; +} + +static int handle_ssl_handshake(endpoint_t* ep) { int rc; + int err; + int ecode = 0; - rc = netio_read(fd, (char*)&(pkt->sz), sizeof(pkt->sz)); - if (rc < 0) { - return rc; + rc = SSL_do_handshake(ep->ssl); + if (rc == 1) { + ep->ssl_handshake_done = 1; + return 1; } - rc = alloc_packet(pkt); - if (rc < 0) { - return rc; - } - - rc = netio_read(fd, pkt->data, pkt->sz); - if (rc < 0) { - free_packet(pkt); + ecode = errno; + err = SSL_get_error(ep->ssl, rc); + switch (err) { + case SSL_ERROR_WANT_READ: + case SSL_ERROR_WANT_WRITE: + rc = 0; + break; + default: + errno = ecode; + perror("SSL_do_handshake"); + rc = handle_error(ep); + break; } return rc; } -void free_packet(io_packet_t* pkt) +static int io_handler(struct epoll_event* p) { - if (pkt->data != NULL) { - free(pkt->data); - pkt->data = NULL; - } -} + endpoint_t* ep = (endpoint_t*)p->data.ptr; + int rc = 0; -#define MAX_EPOLL_EVENTS (5) + if (ep == NULL) { + return -EINVAL; + } + + if (!ep->ssl_handshake_done) { + return handle_ssl_handshake(ep); + } + + if ((p->events & EPOLLIN) != 0) { + if (ep->msg_handler != NULL) { + rc = handle_input(ep); + } else if (ep->conn_handler != NULL) { + rc = ep->conn_handler(p); + } + } + + if (rc >= 0) { + if ((p->events & EPOLLOUT) != 0) { + rc = handle_output(ep); + } + } + + if (rc < 0) { + rc = handle_error(ep); + } + + return rc; +} static void* io_worker(void* args) { @@ -261,11 +509,7 @@ static void* io_worker(void* args) } for (i = 0; i < nfds; i++) { - endpoint_t* p = (endpoint_t*)events[i].data.ptr; - - if (p->handler != NULL) { - (p->handler)(&events[i]); - } + io_handler(&events[i]); } pthread_testcancel(); @@ -278,131 +522,32 @@ static int endpoint_calc_flags(endpoint_t* ep) { int flags = EPOLLIN; - if (ep->outq.head != NULL) { + if (!ep->ssl_handshake_done) { flags = flags | EPOLLOUT; + } else if (ep->outq.head != NULL) { + flags = flags | EPOLLOUT; + } else { + io_fsm_t* fsm = &(ep->fsm_out); + + switch (fsm->state) { + /* we are in the middle stages of output */ + case FSM_STATE_WRITE_PKT_HEAD: + case FSM_STATE_WRITE_PKT: + flags = flags | EPOLLOUT; + break; + default: + break; + } } return flags; } -endpoint_t* endpoint_get() +static int endpoint_update_epoll(endpoint_t* ep) { - endpoint_t* ep; - - ep = (endpoint_t*)calloc(1, sizeof(*ep)); - if (ep == NULL) { - SET_ERROR(errno); - - return ep; - } - - ep->fd = -1; - ep->idx = -1; - ep->iomgr = NULL; - ep->busy = 0; - ep->needfree = 1; - - mylock_init(&(ep->lock)); - - return endpoint_ref(ep); -} - -static void outq_item_killer(data_item_t* it) -{ - if (it != NULL) { - ioctx_t* ctx = ITPTR_TO_IOCTX(it); - ioctx_deref(ctx, 1); - } -} -int endpoint_init(endpoint_t* ep, int domain, int type, int port, int backlog) -{ - int fd = -1; - - if (ep == NULL) { - SET_ERROR(EINVAL); - - return -EINVAL; - } - - fd = socket_create(domain, type, port, backlog); - if (fd < 0) { - return -errno; - } - - ep->flags = 0; - ep->fd = fd; - ep->owner = NULL; - ep->data = NULL; - ep->iomgr = NULL; - ep->busy = 0; - ep->needfree = 0; - - dataq_init(&(ep->outq), outq_item_killer); - mylock_init(&(ep->lock)); - - endpoint_ref(ep); - - return fd; -} - -int endpoint_connect(endpoint_t* ep, char* host, int port) -{ - int rc = -1; - - rc = socket_connect(ep->fd, host, port); - if (rc < 0) { - return rc; - } - - ep->flags = 0; - return rc; -} - -int endpoint_accept(endpoint_t* ep, endpoint_t* ep2) -{ - int sock = -1; - - sock = socket_accept(ep->fd); - if (sock < 0) { - return sock; - } - - ep2->fd = sock; - ep2->flags = 0; - - return sock; -} - -int endpoint_set_handler(endpoint_t* ep, msghandler_t h, void* owner) -{ - if (ep->fd < 0) { - SET_ERROR(EINVAL); - - return -EINVAL; - } - - mylock_lock(&(ep->lock)); - ep->handler = h; - ep->owner = owner; - mylock_unlock(&(ep->lock)); - - return 1; -} - -int endpoint_enq(endpoint_t* ep, data_item_t* it) -{ - int rc = 0; int flags; - if ((ep == NULL) || (it == NULL)) { - SET_ERROR(EINVAL); - - return -EINVAL; - } - mylock_lock(&(ep->lock)); - rc = dataq_enq(&(ep->outq), it); - flags = endpoint_calc_flags(ep); if (flags != ep->flags) { if (ep->iomgr != NULL) { @@ -419,7 +564,222 @@ int endpoint_enq(endpoint_t* ep, data_item_t* it) } } } + mylock_unlock(&(ep->lock)); + return 0; +} + +static void outq_item_killer(data_item_t* it) +{ + if (it != NULL) { + io_packet_t* pkt = ITPTR_TO_PACKET(it); + + free_packet(pkt); + FREEUP(pkt); + } +} + +static int do_endpoint_init(endpoint_t* ep) +{ + if (ep == NULL) { + return -EINVAL; + } + + ep->flags = 0; + ep->fd = -1; + ep->owner = NULL; + ep->data = NULL; + ep->iomgr = NULL; + ep->ssl = NULL; + ep->busy = 0; + ep->ssl_handshake_done = 0; + ep->msg_handler = NULL; + ep->err_handler = NULL; + + dataq_init(&(ep->outq), outq_item_killer); + mylock_init(&(ep->lock)); + + io_fsm_init(&(ep->fsm_in), ep, FSM_IO_READ); + io_fsm_init(&(ep->fsm_out), ep, FSM_IO_WRITE); + + return 1; +} +endpoint_t* endpoint_get() +{ + endpoint_t* ep; + + ep = (endpoint_t*)calloc(1, sizeof(*ep)); + if (ep == NULL) { + SET_ERROR(errno); + + return ep; + } + + do_endpoint_init(ep); + + return ep; +} + +int endpoint_init(endpoint_t* ep, int domain, int type, int port, int backlog) +{ + int fd = -1; + + if (ep == NULL) { + SET_ERROR(EINVAL); + + return -EINVAL; + } + + fd = socket_create(domain, type, port, backlog); + if (fd < 0) { + return -errno; + } + + do_endpoint_init(ep); + ep->fd = fd; + /* no need to ssl handle shake in server endpoint just for init connection + */ + ep->ssl_handshake_done = 1; + + return fd; +} + +int endpoint_connect(endpoint_t* ep, char* host, int port) +{ + int rc = -1; + iomgr_t* iomgr = (iomgr_t*)ep->iomgr; + + rc = socket_connect(ep->fd, host, port); + if (rc < 0) { + return rc; + } + + ep->ssl = SSL_new(iomgr->ssl_ctx); + + SSL_set_fd(ep->ssl, ep->fd); + + SSL_set_connect_state(ep->ssl); + + set_fd_nonblocking(ep->fd); + + ep->flags = 0; + ep->ssl_handshake_done = 0; + + handle_ssl_handshake(ep); + + return rc; +} + +int endpoint_accept(endpoint_t* ep, endpoint_t* ep2) +{ + int sock = -1; + int rc = 0; + iomgr_t* iomgr = (iomgr_t*)ep->iomgr; + + sock = socket_accept(ep->fd); + if (sock < 0) { + return sock; + } + + ep2->ssl = SSL_new(iomgr->ssl_ctx); + + rc = SSL_set_fd(ep2->ssl, sock); + if (rc <= 0) { + int err = SSL_get_error(ep2->ssl, rc); + + ERROR("SSL_set_fd returns error: %d", err); + close(sock); + return -errno; + } + + SSL_set_accept_state(ep2->ssl); + + set_fd_nonblocking(sock); + + ep2->fd = sock; + ep2->flags = 0; + + return sock; +} + +int endpoint_set_owner(endpoint_t* ep, void* owner) +{ + if ((ep == NULL) || (owner == NULL)) { + return -EINVAL; + } + + mylock_lock(&(ep->lock)); + ep->owner = owner; + mylock_unlock(&(ep->lock)); + + return 1; +} + +int endpoint_set_peer(endpoint_t* ep, xce_peerinfo_t* peer) +{ + mylock_lock(&(ep->lock)); + ep->peer = peer; + mylock_unlock(&(ep->lock)); + + return 0; +} + +int endpoint_set_conn_handler(endpoint_t* ep, conn_handler_t h) +{ + if (ep->fd < 0) { + SET_ERROR(EINVAL); + + return -EINVAL; + } + + mylock_lock(&(ep->lock)); + ep->conn_handler = h; + mylock_unlock(&(ep->lock)); + + return 1; +} + +int endpoint_set_msg_handler(endpoint_t* ep, msg_handler_t h) +{ + if ((ep == NULL) || (h == NULL)) { + return -EINVAL; + } + + mylock_lock(&(ep->lock)); + ep->msg_handler = h; + mylock_unlock(&(ep->lock)); + + return 1; +} + +int endpoint_set_err_handler(endpoint_t* ep, err_handler_t h) +{ + if ((ep == NULL) || (h == NULL)) { + return -EINVAL; + } + + mylock_lock(&(ep->lock)); + ep->err_handler = h; + mylock_unlock(&(ep->lock)); + + return 1; +} + +int endpoint_enq(endpoint_t* ep, data_item_t* it) +{ + int rc = 0; + + if ((ep == NULL) || (it == NULL)) { + SET_ERROR(EINVAL); + + return -EINVAL; + } + + mylock_lock(&(ep->lock)); + rc = dataq_enq(&(ep->outq), it); + mylock_unlock(&(ep->lock)); + + endpoint_update_epoll(ep); return rc; } @@ -427,7 +787,6 @@ int endpoint_enq(endpoint_t* ep, data_item_t* it) data_item_t* endpoint_deq(endpoint_t* ep) { data_item_t* it = NULL; - int flags; if (ep == NULL) { SET_ERROR(EINVAL); @@ -436,50 +795,12 @@ data_item_t* endpoint_deq(endpoint_t* ep) } mylock_lock(&(ep->lock)); - it = dataq_deq(&(ep->outq), NULL); - flags = endpoint_calc_flags(ep); - if (flags != ep->flags) { - if (ep->iomgr != NULL) { - int rc; - - mylock_lock(&(ep->iomgr->lock)); - rc = iomgr_update_poll(ep->iomgr, ep, flags); - mylock_unlock(&(ep->iomgr->lock)); - - if (rc >= 0) { - ep->flags = flags; - } - } - } + it = dataq_deq(&(ep->outq), NULL); mylock_unlock(&(ep->lock)); return it; } -endpoint_t* endpoint_ref(endpoint_t* ep) -{ - mylock_lock(&(ep->lock)); - ep->refcount++; - mylock_unlock(&(ep->lock)); - - return ep; -} - -void endpoint_deref(endpoint_t* ep) -{ - uint32_t refcount; - - mylock_lock(&(ep->lock)); - ep->refcount--; - - refcount = ep->refcount; - mylock_unlock(&(ep->lock)); - - if (refcount == 0) { - endpoint_fini(ep); - } -} - uint64_t endpoint_getseq(endpoint_t* ep) { uint64_t seq; @@ -522,10 +843,8 @@ int endpoint_is_task_blocked(endpoint_t* ep) return (busy != 0); } -static void endpoint_fini(endpoint_t* ep) +void endpoint_fini(endpoint_t* ep) { - int needfree = ep->needfree; - dataq_fini(&(ep->outq)); FREEUP(ep->data); @@ -533,13 +852,72 @@ static void endpoint_fini(endpoint_t* ep) close(ep->fd); } - ep->flags = 0; - ep->handler = NULL; - ep->owner = NULL; - - if (needfree) { - FREEUP(ep); + if (ep->ssl != NULL) { + SSL_free(ep->ssl); + ep->ssl = NULL; } + + if (ep->fsm_in.pkt != NULL) { + free_packet(ep->fsm_in.pkt); + + FREEUP(ep->fsm_in.pkt); + } + + if (ep->fsm_out.pkt != NULL) { + free_packet(ep->fsm_out.pkt); + + FREEUP(ep->fsm_out.pkt); + } + + ep->flags = 0; + ep->msg_handler = NULL; + ep->err_handler = NULL; + ep->owner = NULL; +} + +void initialize_openssl() +{ + SSL_load_error_strings(); + OpenSSL_add_ssl_algorithms(); +} + +void cleanup_openssl() { EVP_cleanup(); } + +SSL_CTX* create_ssl_context(char* cert_file, char* key_file) +{ + const SSL_METHOD* method; + SSL_CTX* ctx; + + /* any of cert_file or key_file is NULL, just as client */ + if ((cert_file == NULL) || (key_file == NULL)) { + method = TLS_client_method(); + ctx = SSL_CTX_new(method); + return ctx; + } + + method = TLS_server_method(); + + ctx = SSL_CTX_new(method); + if (ctx == NULL) { + return NULL; + } + + if (SSL_CTX_use_certificate_chain_file(ctx, cert_file) <= 0) { + SSL_CTX_free(ctx); + return NULL; + } + + if (SSL_CTX_use_PrivateKey_file(ctx, key_file, SSL_FILETYPE_PEM) <= 0) { + SSL_CTX_free(ctx); + return NULL; + } + + if (!SSL_CTX_check_private_key(ctx)) { + SSL_CTX_free(ctx); + return NULL; + } + + return ctx; } int iomgr_init(iomgr_t* iomgr) @@ -555,6 +933,15 @@ int iomgr_init(iomgr_t* iomgr) return 0; } +void iomgr_set_ssl_ctx(iomgr_t* iomgr, SSL_CTX* ssl_ctx) +{ + if ((iomgr == NULL) || (ssl_ctx == NULL)) { + return; + } + + iomgr->ssl_ctx = ssl_ctx; +} + static int iomgr_update_poll(iomgr_t* iomgr, endpoint_t* ep, int flags) { struct epoll_event ev; @@ -565,11 +952,16 @@ static int iomgr_update_poll(iomgr_t* iomgr, endpoint_t* ep, int flags) return epoll_ctl(iomgr->epoll_fd, EPOLL_CTL_MOD, ep->fd, &ev); } -int iomgr_monitor(iomgr_t* iomgr, endpoint_t* ep) +int iomgr_monitor(endpoint_t* ep) { struct epoll_event ev; int rc; int flags; + iomgr_t* iomgr = (iomgr_t*)ep->iomgr; + + if (iomgr->ssl_ctx == NULL) { + return -EINVAL; + } flags = EPOLLIN; ev.events = flags; @@ -614,7 +1006,15 @@ int iomgr_start(iomgr_t* iomgr) return rc; } -int iomgr_fini(iomgr_t* iomgr) { return 0; } +int iomgr_fini(iomgr_t* iomgr) +{ + if (iomgr->ssl_ctx != NULL) { + SSL_CTX_free(iomgr->ssl_ctx); + iomgr->ssl_ctx = NULL; + } + + return 0; +} int iomgr_stop(iomgr_t* iomgr) { diff --git a/src/lib/request.c b/src/lib/request.c index 1f75969c..47422ca8 100644 --- a/src/lib/request.c +++ b/src/lib/request.c @@ -11,134 +11,94 @@ #include "utils.h" #include "xce/types.h" -ioctx_t* ioctx_alloc(endpoint_t* ep, xce_peerinfo_t* peer) -{ - ioctx_t* ctx; - - ctx = (ioctx_t*)calloc(1, sizeof(*ctx)); - if (ctx == NULL) { - errno = -ENOMEM; - return NULL; - } - - ctx->ep = endpoint_ref(ep); - ctx->from = peer; - ctx->nrefs++; - return ctx; -} - -int ioctx_ref(ioctx_t* ctx) -{ - int old = ctx->nrefs; - - ctx->nrefs++; - - return old; -} - -static void ioctx_kill(ioctx_t* ctx, int free_ctx) -{ - if (ctx == NULL) { - return; - } - - endpoint_deref(ctx->ep); - - free_packet(&(ctx->pkt)); - capn_free(&(ctx->c)); - if (free_ctx) { - free(ctx); - } -} - -void ioctx_deref(ioctx_t* ctx, int free_ctx) -{ - if (ctx == NULL) { - return; - } - - if (ctx->nrefs == 0) { - WARN("Something wrong. Have you forgotten to call ioctx_ref()?"); - return; - } - ctx->nrefs--; - if (ctx->nrefs == 0) { - ioctx_kill(ctx, free_ctx); - } -} - -int ioctx_encode(ioctx_t* ctx, xce_request_t* req) +int xce_encode(xce_request_t* req, io_packet_t* pkt) { + struct capn c; + struct Request rq; struct capn_segment* cs; Request_ptr rp; int rc; - capn_init_malloc(&(ctx->c)); - cs = capn_root(&(ctx->c)).seg; + TRACE("%s request op %d seq %ld", __func__, req->opcode, req->seq); - encode_Request(cs, &(ctx->rq), req); + capn_init_malloc(&c); + cs = capn_root(&c).seg; - cs = capn_root(&(ctx->c)).seg; + encode_Request(cs, &rq, req); + + cs = capn_root(&c).seg; rp = new_Request(cs); - write_Request(&(ctx->rq), rp); + write_Request(&rq, rp); - capn_setp(capn_root(&(ctx->c)), 0, rp.p); + capn_setp(capn_root(&c), 0, rp.p); - ctx->pkt.sz = capn_size(&(ctx->c)); - rc = alloc_packet(&(ctx->pkt)); + pkt->sz = capn_size(&c); + rc = alloc_packet(pkt); if (rc < 0) { - return rc; + capn_free(&c); + return -errno; } - capn_write_mem(&(ctx->c), (uint8_t*)ctx->pkt.data, ctx->pkt.sz, 0); - return 0; + capn_write_mem(&c, (uint8_t*)pkt->data, pkt->sz, 0); + capn_free(&c); + + return 1; } -int ioctx_decode(ioctx_t* ctx, xce_request_t* req) +int xce_decode(io_packet_t* pkt, xce_request_t* req) { - Request_ptr rp; + struct capn c; + struct Request rq; + Request_ptr rp; - capn_init_mem(&(ctx->c), (const uint8_t*)ctx->pkt.data, ctx->pkt.sz, 0); - rp.p = capn_getp(capn_root(&(ctx->c)), 0, 1); + capn_init_mem(&c, (const uint8_t*)pkt->data, pkt->sz, 0); + rp.p = capn_getp(capn_root(&c), 0, 1); - read_Request(&(ctx->rq), rp); + read_Request(&rq, rp); - decode_Request(req, &(ctx->rq)); + decode_Request(req, &rq); + + TRACE("%s request opcode %d seq %ld", __func__, req->opcode, req->seq); + capn_free(&c); + + free_packet(pkt); + FREEUP(pkt); return 0; } -int ioctx_flush(ioctx_t* ctx, int fd) -{ - int rc; - - rc = write_packet(fd, &(ctx->pkt)); - return rc; -} - void init_reply_by_req(xce_reply_t* reply, xce_request_t* req) { reply->seq = req->seq; reply->opcode = req->opcode; } -int xce_request(ioctx_t* ctx, xce_request_t* req) +int xce_request(endpoint_t* ep, xce_request_t* req) { - int rc; + int rc; + io_packet_t* pkt; - req->seq = endpoint_getseq(ctx->ep); - req->from = ctx->from; + pkt = (io_packet_t*)calloc(1, sizeof(*pkt)); + if (pkt == NULL) { + return -ENOMEM; + } - rc = ioctx_encode(ctx, req); + req->seq = endpoint_getseq(ep); + req->from = ep->peer; + + rc = xce_encode(req, pkt); if (rc < 0) { + free_packet(pkt); + free(pkt); + return rc; } - endpoint_enq(ctx->ep, &(ctx->link)); + endpoint_enq(ep, &(pkt->link)); return 0; } -int xce_reply_with_ctx(ioctx_t* ctx, xce_peerinfo_t* from, xce_reply_t* reply) +int xce_reply(endpoint_t* ep, xce_peerinfo_t* from, xce_reply_t* reply) { /* clang-format off */ xce_request_t request = { @@ -146,50 +106,47 @@ int xce_reply_with_ctx(ioctx_t* ctx, xce_peerinfo_t* from, xce_reply_t* reply) .body_which = Request_body_replymsg, }; /* clang-format on */ - int rc; + int rc; + io_packet_t* pkt; TRACE("%s seq %lu opcode %d status %d data which %d", __func__, reply->seq, reply->opcode, reply->status, reply->data_which); - request.from = from; - request.body.replymsg = reply; - - request.seq = endpoint_getseq(ctx->ep); - - rc = ioctx_encode(ctx, &request); - if (rc < 0) { - return rc; - } - - endpoint_enq(ctx->ep, &(ctx->link)); - return 0; -} - -int xce_reply(endpoint_t* ep, xce_peerinfo_t* from, xce_reply_t* reply) -{ - ioctx_t* ctx; - - ctx = ioctx_alloc(ep, from); - if (ctx == NULL) { + pkt = (io_packet_t*)calloc(1, sizeof(*pkt)); + if (pkt == NULL) { return -ENOMEM; } - return xce_reply_with_ctx(ctx, from, reply); + request.from = from; + request.body.replymsg = reply; + + request.seq = endpoint_getseq(ep); + + rc = xce_encode(&request, pkt); + if (rc < 0) { + free_packet(pkt); + free(pkt); + + return rc; + } + + endpoint_enq(ep, &(pkt->link)); + return 0; } -int xce_nop_req(ioctx_t* ctx) +int xce_nop_req(endpoint_t* ep) { xce_request_t request = { .opcode = OPCode_nop, .body_which = Request_body_nomsg }; - if (ctx == NULL) { + if (ep == NULL) { return -EINVAL; } - return xce_request(ctx, &request); + return xce_request(ep, &request); } -int xce_hello_req(ioctx_t* ctx, char* msg) +int xce_hello_req(endpoint_t* ep, char* msg) { /* clang-format off */ xce_request_t request = { @@ -201,13 +158,13 @@ int xce_hello_req(ioctx_t* ctx, char* msg) }; /* clang-format on */ - if ((ctx == NULL) || (msg == NULL)) { + if ((ep == NULL) || (msg == NULL)) { return -EINVAL; } - return xce_request(ctx, &request); + return xce_request(ep, &request); } -int xce_join_req(ioctx_t* ctx, int role, hostattrs_t* attrs) +int xce_join_req(endpoint_t* ep, int role, hostattrs_t* attrs) { /* clang-format off */ xce_joindata_t joinmsg = { @@ -222,8 +179,8 @@ int xce_join_req(ioctx_t* ctx, int role, hostattrs_t* attrs) }; if ((role != HostRole_client) && - (role != HostRole_worker) && - (role != HostRole_scheduler)) { + (role != HostRole_worker) && + (role != HostRole_scheduler)) { return -EINVAL; } /* clang-format on */ @@ -236,26 +193,26 @@ int xce_join_req(ioctx_t* ctx, int role, hostattrs_t* attrs) joinmsg.data_which = JoinData_data_attrs; joinmsg.data.attrs = attrs; } - return xce_request(ctx, &request); + return xce_request(ep, &request); } -int xce_shutdown_req(ioctx_t* ctx) +int xce_shutdown_req(endpoint_t* ep) { int rc; dataq_t* outq; xce_request_t request = { .opcode = OPCode_shutdown, .body_which = Request_body_nomsg }; - if (ctx == NULL) { + if (ep == NULL) { return -EINVAL; } - rc = xce_request(ctx, &request); + rc = xce_request(ep, &request); if (rc < 0) { return rc; } - outq = &(ctx->ep->outq); + outq = &(ep->outq); while (!DATAQ_EMPTY(outq)) { sleep(1); @@ -267,30 +224,25 @@ int xce_shutdown_req(ioctx_t* ctx) int xce_pushconfig_req(endpoint_t* ep, xce_peerinfo_t* from, xce_configdata_t* config) { - int rc = 0; + int rc = 0; + /* clang-format off */ xce_request_t request = { .opcode = OPCode_push, .body_which = Request_body_configpushmsg }; - ioctx_t* ctx = NULL; + /* clang-format on */ if ((ep == NULL) || (config == NULL)) { return -EINVAL; } - ctx = ioctx_alloc(ep, from); - if (ctx == NULL) { - return -ENOMEM; - } - request.body.configpushmsg = config; - rc = xce_request(ctx, &request); + rc = xce_request(ep, &request); return rc; } static int xce_push_req(endpoint_t* ep, xce_peerinfo_t* from, xce_data_t* data) { - int rc = 0; - ioctx_t* ctx = NULL; + int rc = 0; xce_request_t request = { .opcode = OPCode_push, .body_which = Request_body_pushmsg }; @@ -298,14 +250,9 @@ static int xce_push_req(endpoint_t* ep, xce_peerinfo_t* from, xce_data_t* data) return -EINVAL; } - ctx = ioctx_alloc(ep, from); - if (ctx == NULL) { - return -ENOMEM; - } - request.body.pushmsg = data; - rc = xce_request(ctx, &request); + rc = xce_request(ep, &request); return rc; } @@ -416,7 +363,7 @@ int xce_push_data_req(endpoint_t* ep, xce_peerinfo_t* from, xce_data_t* data, return rc; } -int xce_get_req(ioctx_t* ctx, xce_get_t* get) +int xce_get_req(endpoint_t* ep, xce_get_t* get) { /* clang-format off */ xce_request_t request = { @@ -428,7 +375,7 @@ int xce_get_req(ioctx_t* ctx, xce_get_t* get) }; /* clang-format on */ - return xce_request(ctx, &request); + return xce_request(ep, &request); } int xce_reply_get(endpoint_t* ep, xce_peerinfo_t* from, xce_reply_t* reply, @@ -699,7 +646,7 @@ int extract_data(dataset_t* dataset, xce_data_t* getdata) return 1; } -int xce_jobdispatch_req(ioctx_t* ctx, xce_job_dispatch_t* info) +int xce_jobdispatch_req(endpoint_t* ep, xce_job_dispatch_t* info) { /* clang-format off */ xce_request_t request = { @@ -712,16 +659,16 @@ int xce_jobdispatch_req(ioctx_t* ctx, xce_job_dispatch_t* info) /* clang-format on */ /* clang-format off */ - if ((ctx == NULL) || - (info == NULL)) { + if ((ep == NULL) || + (info == NULL)) { return -EINVAL; } /* clang-format on */ - return xce_request(ctx, &request); + return xce_request(ep, &request); } -int xce_notify_req(ioctx_t* ctx, xce_notify_data_t* data) +int xce_notify_req(endpoint_t* ep, xce_notify_data_t* data) { /* clang-format off */ xce_request_t request = { @@ -731,17 +678,16 @@ int xce_notify_req(ioctx_t* ctx, xce_notify_data_t* data) .notifymsg = data } }; - /* clang-forn on */ + /* clang-format on */ - if ((ctx == NULL) || (data == NULL)) { - return -EINVAL; + if ((ep == NULL) || (data == NULL)) { + return -EINVAL; } - return xce_request(ctx, &request); + return xce_request(ep, &request); } -int xce_jobsubmit_req(ioctx_t* ctx, - xce_job_submit_t* submit) +int xce_jobsubmit_req(endpoint_t* ep, xce_job_submit_t* submit) { /* clang-format off */ xce_request_t request = { @@ -754,16 +700,16 @@ int xce_jobsubmit_req(ioctx_t* ctx, /*clang-format on*/ /* clang-format off */ - if ((ctx == NULL) || - (submit == NULL)) { - return -EINVAL; + if ((ep == NULL) || + (submit == NULL)) { + return -EINVAL; } /* clang-format on */ - return xce_request(ctx, &request); + return xce_request(ep, &request); } -int xce_startjob_req(ioctx_t* ctx, xce_job_exec_t* execinfo) +int xce_startjob_req(endpoint_t* ep, xce_job_exec_t* execinfo) { /* clang-format off */ xce_request_t request = { @@ -775,10 +721,10 @@ int xce_startjob_req(ioctx_t* ctx, xce_job_exec_t* execinfo) }; /* clang-format on */ - return xce_request(ctx, &request); + return xce_request(ep, &request); } -int xce_clusterconf_req(ioctx_t* ctx) +int xce_clusterconf_req(endpoint_t* ep) { /* clang-format off */ xce_request_t request = { @@ -787,10 +733,10 @@ int xce_clusterconf_req(ioctx_t* ctx) }; /* clang-format on */ - return xce_request(ctx, &request); + return xce_request(ep, &request); } -int xce_update_req(ioctx_t* ctx, xce_update_data_t* data) +int xce_update_req(endpoint_t* ep, xce_update_data_t* data) { /* clang-format off */ xce_request_t request = { @@ -802,14 +748,14 @@ int xce_update_req(ioctx_t* ctx, xce_update_data_t* data) }; /* clang-format on */ - if ((ctx == NULL) || (data == NULL)) { + if ((ep == NULL) || (data == NULL)) { return -EINVAL; } - return xce_request(ctx, &request); + return xce_request(ep, &request); } -int xce_jobctrl_req(ioctx_t* ctx, xce_jobctrl_t* ctrinfo) +int xce_jobctrl_req(endpoint_t* ep, xce_jobctrl_t* ctrinfo) { /* clang-format off */ xce_request_t request = { @@ -821,7 +767,7 @@ int xce_jobctrl_req(ioctx_t* ctx, xce_jobctrl_t* ctrinfo) }; /* clang-format on */ - if ((ctx == NULL) || (ctrinfo == NULL)) { + if ((ep == NULL) || (ctrinfo == NULL)) { return -EINVAL; } @@ -841,10 +787,10 @@ int xce_jobctrl_req(ioctx_t* ctx, xce_jobctrl_t* ctrinfo) return -EINVAL; break; } - return xce_request(ctx, &request); + return xce_request(ep, &request); } -int xce_queuectrl_req(ioctx_t* ctx, xce_queuectrl_t* ctrlinfo) +int xce_queuectrl_req(endpoint_t* ep, xce_queuectrl_t* ctrlinfo) { /* clang-format off */ xce_request_t request = { @@ -856,7 +802,7 @@ int xce_queuectrl_req(ioctx_t* ctx, xce_queuectrl_t* ctrlinfo) }; /* clang-format on */ - if ((ctx == NULL) || (ctrlinfo == NULL)) { + if ((ep == NULL) || (ctrlinfo == NULL)) { return -EINVAL; } @@ -872,10 +818,10 @@ int xce_queuectrl_req(ioctx_t* ctx, xce_queuectrl_t* ctrlinfo) return -EINVAL; } - return xce_request(ctx, &request); + return xce_request(ep, &request); } -int xce_runtask_req(ioctx_t* ctx, xce_taskrun_t* runinfo) +int xce_runtask_req(endpoint_t* ep, xce_taskrun_t* runinfo) { /* clang-format off */ xce_request_t request = { @@ -887,7 +833,7 @@ int xce_runtask_req(ioctx_t* ctx, xce_taskrun_t* runinfo) }; /* clang-format on */ - if ((ctx == NULL) || (runinfo == NULL)) { + if ((ep == NULL) || (runinfo == NULL)) { return -EINVAL; } @@ -895,5 +841,5 @@ int xce_runtask_req(ioctx_t* ctx, xce_taskrun_t* runinfo) return -EINVAL; } - return xce_request(ctx, &request); + return xce_request(ep, &request); } diff --git a/src/lib/testsuite/test_idgen.c b/src/lib/testsuite/test_idgen.c index 2c7ab03a..2fb27c88 100644 --- a/src/lib/testsuite/test_idgen.c +++ b/src/lib/testsuite/test_idgen.c @@ -58,7 +58,7 @@ static void test_idgen_array(void) for (k = 0; k < 5; k++) { rc = idgen(&ctx, 0x0, 0x0, i + 1, &idx); TEST_ASSERT_TRUE(rc >= 0); - if ( i >= 1) { + if (i >= 1) { TEST_ASSERT_TRUE((idx & MAX_ARRAY_IDX) == 0); } printf("\tidx = %lx\n", idx); diff --git a/src/lib/vtaskmgr.c b/src/lib/vtaskmgr.c index 314185f0..abc2e47e 100644 --- a/src/lib/vtaskmgr.c +++ b/src/lib/vtaskmgr.c @@ -18,22 +18,14 @@ vtask_t* vtask_alloc() return task; } -int vtask_init(vtask_t* task, vtask_func_t entry, endpoint_t* ep) +int vtask_init(vtask_t* task, vtask_func_t entry, void* data) { - int rc; + task->entry = entry; + task->vcpu = -1; + task->data = data; + task->owner = NULL; - task->entry = entry; - task->vcpu = -1; - task->ctx.ep = endpoint_ref(ep); - - rc = read_packet(ep->fd, &(task->ctx.pkt)); - if (rc < 0) { - free_packet(&(task->ctx.pkt)); - endpoint_deref(task->ctx.ep); - task->ctx.ep = NULL; - } - - return rc; + return 0; } void vtask_fini(vtask_t* task) @@ -42,7 +34,7 @@ void vtask_fini(vtask_t* task) return; } - ioctx_deref(&(task->ctx), 0); + FREEUP(task->data); return; } @@ -59,13 +51,22 @@ void vcpu_idle(vcpu_t* vcpu) void run_task(vtask_t* task) { - vtaskmgr_t* vtaskmgr; + vtaskmgr_t* vtaskmgr; + vtask_data_t* data; if (task->entry != NULL) { (task->entry)(task); } - endpoint_unblock_tasks(task->ctx.ep); + data = task->data; + if ((data != NULL) && (data->vtype == VTASK_IO)) { + endpoint_t* ep = data->u.iodata.ep; + + if (ep != NULL) { + endpoint_unblock_tasks(ep); + } + } + vtaskmgr = (vtaskmgr_t*)task->owner; if (vtaskmgr != NULL) { /* wake up dispatcher */ @@ -201,15 +202,21 @@ static vcpu_t* pick_idle_vcpu(vtaskmgr_t* vtaskmgr) static int vtask_eligible(data_item_t* data) { - vtask_t* task = ITPTR_TO_VTASK(data); - int eligible = 0; + vtask_t* task = ITPTR_TO_VTASK(data); + int eligible = 0; + vtask_data_t* vdata; - if (task->ctx.ep == NULL) { + vdata = task->data; + if ((vdata == NULL) || (vdata->vtype != VTASK_IO)) { + return 1; + } + + if (vdata->u.iodata.ep == NULL) { WARN("no endpoint for task 0x%p", task); return 0; } - eligible = !(endpoint_is_task_blocked(task->ctx.ep)); + eligible = !(endpoint_is_task_blocked(vdata->u.iodata.ep)); TRACE("%s vtask %p is %s", __func__, task, eligible ? "eligible" : "not eligible"); @@ -236,7 +243,11 @@ static int bind_vtask(vtask_t* task, vcpu_t* vcpu) TRACE("binding task %p to cpu#%d\n", task, vcpu->cpuid); if (task != NULL) { - endpoint_block_tasks(task->ctx.ep); + vtask_data_t* data = (vtask_data_t*)task->data; + + if ((data->vtype == VTASK_IO) && (data->u.iodata.ep != NULL)) { + endpoint_block_tasks(data->u.iodata.ep); + } task->vcpu = vcpu->cpuid; } @@ -331,9 +342,8 @@ void vtaskmgr_fini(vtaskmgr_t* vtaskmgr) int vtask_enq(vtaskmgr_t* vtaskmgr, vtask_t* task) { int rc = 0; - ioctx_ref(&(task->ctx)); - rc = dataq_enq(&(vtaskmgr->taskq), &(task->ctx.link)); + rc = dataq_enq(&(vtaskmgr->taskq), &(task->link)); if (rc >= 0) { /* wake up task dispatcher */ task->owner = vtaskmgr; @@ -342,3 +352,37 @@ int vtask_enq(vtaskmgr_t* vtaskmgr, vtask_t* task) return rc; } + +int vtask_handle_msg(vtaskmgr_t* mgr, vtask_func_t entry, void* arg) +{ + endpoint_t* ep = (endpoint_t*)arg; + vtask_data_t* data; + vtask_t* task; + + if (ep == NULL) { + return -EINVAL; + } + + if ((ep->fsm_in.state != FSM_STATE_DONE) || (ep->fsm_in.pkt == NULL)) { + return -EINVAL; + } + + data = (vtask_data_t*)calloc(1, sizeof(*data)); + if (data == NULL) { + return -ENOMEM; + } + + task = (vtask_t*)calloc(1, sizeof(*task)); + if (task == NULL) { + free(data); + return -ENOMEM; + } + + data->vtype = VTASK_IO; + data->u.iodata.ep = ep; + data->u.iodata.pkt = ep->fsm_in.pkt; + + vtask_init(task, entry, data); + + return vtask_enq(mgr, task); +} diff --git a/src/lib/xceapi.c b/src/lib/xceapi.c index 31b51731..81ee7f80 100644 --- a/src/lib/xceapi.c +++ b/src/lib/xceapi.c @@ -5,8 +5,7 @@ int xce_hello(xce_ctx_t* ctx, char* msg) { - ioctx_t* ioctx; - int rc, status; + int rc, status; if (ctx == NULL) { return -EINVAL; @@ -16,12 +15,7 @@ int xce_hello(xce_ctx_t* ctx, char* msg) return -EINVAL; } - ioctx = ioctx_alloc(ctx->ep, &(ctx->myinfo)); - if (ioctx == NULL) { - return -ENOMEM; - } - - rc = xce_hello_req(ioctx, msg); + rc = xce_hello_req(ctx->ep, msg); if (rc < 0) { return rc; } @@ -35,9 +29,7 @@ int xce_hello(xce_ctx_t* ctx, char* msg) int xce_shutdown(xce_ctx_t* ctx) { - ioctx_t* ioctx; - int rc; - int status; + int rc; if (ctx == NULL) { return -EINVAL; @@ -47,27 +39,14 @@ int xce_shutdown(xce_ctx_t* ctx) return -EINVAL; } - ioctx = ioctx_alloc(ctx->ep, &(ctx->myinfo)); - if (ioctx == NULL) { - return -ENOMEM; - } + rc = xce_shutdown_req(ctx->ep); - rc = xce_shutdown_req(ioctx); - if (rc < 0) { - return rc; - } - - rc = cbuf_get(&(ctx->wbuf), &status); - if (rc < 0) { - return rc; - } - - return status; + /* do not need wait controller back as it will be shutdown */ + return rc; } int xce_get(xce_ctx_t* ctx, int table, void*** data) { - ioctx_t* ioctx; int rc; int status; xce_match_t condition = { .type = MatchCondition_condition_matchall }; @@ -82,13 +61,8 @@ int xce_get(xce_ctx_t* ctx, int table, void*** data) return -EINVAL; } - ioctx = ioctx_alloc(ctx->ep, &(ctx->myinfo)); - if (ioctx == NULL) { - return -ENOMEM; - } - ctx->reply.u.get = &getreply; - rc = xce_get_req(ioctx, &get); + rc = xce_get_req(ctx->ep, &get); if (rc < 0) { return rc; } @@ -119,28 +93,27 @@ int xce_reslist(xce_ctx_t* ctx, xce_resource_t*** res) int xce_resval(xce_ctx_t* ctx, char* host, char* res, xce_resval_t*** vals) { - ioctx_t* ioctx; - int rc; - int status; + int rc; + int status; /* clang-format off */ xce_reskey_t reskey = { - .resname = res, - .hostname = host + .resname = res, + .hostname = host }; xce_match_t condition = { - .type = MatchCondition_condition_matchreskey, - .condition = { - .reskey = &reskey - } + .type = MatchCondition_condition_matchreskey, + .condition = { + .reskey = &reskey + } }; xce_get_t get = { - .table = Data_data_resvals, - .condition = &condition + .table = Data_data_resvals, + .condition = &condition }; dataset_t getreply = { - .table = Data_data_resvals, - .nrows = 0, - .rows = NULL + .table = Data_data_resvals, + .nrows = 0, + .rows = NULL }; /* clang-format on */ @@ -152,13 +125,8 @@ int xce_resval(xce_ctx_t* ctx, char* host, char* res, xce_resval_t*** vals) return -EINVAL; } - ioctx = ioctx_alloc(ctx->ep, &(ctx->myinfo)); - if (ioctx == NULL) { - return -ENOMEM; - } - ctx->reply.u.get = &getreply; - rc = xce_get_req(ioctx, &get); + rc = xce_get_req(ctx->ep, &get); if (rc < 0) { return rc; } @@ -182,8 +150,7 @@ int xce_resval(xce_ctx_t* ctx, char* host, char* res, xce_resval_t*** vals) int xce_join(xce_ctx_t* ctx, int role, hostattrs_t* attrs) { - ioctx_t* ioctx; - int rc, status; + int rc, status; if (ctx == NULL) { return -EINVAL; @@ -193,12 +160,7 @@ int xce_join(xce_ctx_t* ctx, int role, hostattrs_t* attrs) return -EINVAL; } - ioctx = ioctx_alloc(ctx->ep, &(ctx->myinfo)); - if (ioctx == NULL) { - return -ENOMEM; - } - - rc = xce_join_req(ioctx, role, attrs); + rc = xce_join_req(ctx->ep, role, attrs); if (rc < 0) { return rc; } @@ -221,8 +183,7 @@ int xce_free_workers(int nworkers, xce_worker_t* workers) { return 0; } int xce_jobdispatch(xce_ctx_t* ctx, xce_job_dispatch_t* info) { - ioctx_t* ioctx; - int rc, status; + int rc, status; if (ctx == NULL) { return -EINVAL; @@ -232,12 +193,7 @@ int xce_jobdispatch(xce_ctx_t* ctx, xce_job_dispatch_t* info) return -EINVAL; } - ioctx = ioctx_alloc(ctx->ep, &(ctx->myinfo)); - if (ioctx == NULL) { - return -ENOMEM; - } - - rc = xce_jobdispatch_req(ioctx, info); + rc = xce_jobdispatch_req(ctx->ep, info); if (rc < 0) { return rc; } @@ -252,8 +208,7 @@ int xce_jobdispatch(xce_ctx_t* ctx, xce_job_dispatch_t* info) int xce_notify(xce_ctx_t* ctx, xce_notify_data_t* data) { - ioctx_t* ioctx; - int rc; + int rc; if (ctx == NULL) { return -EINVAL; @@ -263,20 +218,14 @@ int xce_notify(xce_ctx_t* ctx, xce_notify_data_t* data) return -EINVAL; } - ioctx = ioctx_alloc(ctx->ep, &(ctx->myinfo)); - if (ioctx == NULL) { - return -ENOMEM; - } - - rc = xce_notify_req(ioctx, data); + rc = xce_notify_req(ctx->ep, data); return rc; } int xce_jobsubmit(xce_ctx_t* ctx, xce_job_submit_t* submit, uint64_t* jobid) { - ioctx_t* ioctx; - int rc, status; + int rc, status; if (ctx == NULL) { return -EINVAL; @@ -286,12 +235,7 @@ int xce_jobsubmit(xce_ctx_t* ctx, xce_job_submit_t* submit, uint64_t* jobid) return -EINVAL; } - ioctx = ioctx_alloc(ctx->ep, &(ctx->myinfo)); - if (ioctx == NULL) { - return -ENOMEM; - } - - rc = xce_jobsubmit_req(ioctx, submit); + rc = xce_jobsubmit_req(ctx->ep, submit); if (rc < 0) { return rc; } @@ -393,8 +337,7 @@ int xce_free_groups(int ngroups, xce_group_t* groups) int xce_clusterconf(xce_ctx_t* ctx, xce_clusterconf_t** conf) { - ioctx_t* ioctx; - int rc, status; + int rc, status; if ((ctx == NULL) || (conf == NULL)) { return -EINVAL; @@ -404,12 +347,7 @@ int xce_clusterconf(xce_ctx_t* ctx, xce_clusterconf_t** conf) return -EINVAL; } - ioctx = ioctx_alloc(ctx->ep, &(ctx->myinfo)); - if (ioctx == NULL) { - return -ENOMEM; - } - - rc = xce_clusterconf_req(ioctx); + rc = xce_clusterconf_req(ctx->ep); if (rc < 0) { return rc; } @@ -429,9 +367,8 @@ int xce_clusterconf(xce_ctx_t* ctx, xce_clusterconf_t** conf) int xce_update_resvals(xce_ctx_t* ctx, int count, xce_resval_t** vals) { - ioctx_t* ioctx; - int rc; - int status; + int rc; + int status; /* clang-format off */ xce_resval_list_t valist = { @@ -454,12 +391,7 @@ int xce_update_resvals(xce_ctx_t* ctx, int count, xce_resval_t** vals) return -EINVAL; } - ioctx = ioctx_alloc(ctx->ep, &(ctx->myinfo)); - if (ioctx == NULL) { - return -ENOMEM; - } - - rc = xce_update_req(ioctx, &data); + rc = xce_update_req(ctx->ep, &data); if (rc < 0) { return rc; } @@ -475,9 +407,8 @@ int xce_update_resvals(xce_ctx_t* ctx, int count, xce_resval_t** vals) int xce_update_jusage(xce_ctx_t* ctx, uint64_t jobid, int npids, int* pids, int ngids, int* pgids, int nusage, xce_rusage_t** usages) { - ioctx_t* ioctx; - int rc; - int status; + int rc; + int status; /* clang-format off */ xce_job_rusages_t rusages = { @@ -503,12 +434,7 @@ int xce_update_jusage(xce_ctx_t* ctx, uint64_t jobid, int npids, int* pids, return -EINVAL; } - ioctx = ioctx_alloc(ctx->ep, &(ctx->myinfo)); - if (ioctx == NULL) { - return -ENOMEM; - } - - rc = xce_update_req(ioctx, &data); + rc = xce_update_req(ctx->ep, &data); if (rc < 0) { return rc; } @@ -523,9 +449,8 @@ int xce_update_jusage(xce_ctx_t* ctx, uint64_t jobid, int npids, int* pids, int xce_jobctrl(xce_ctx_t* ctx, xce_jobctrl_t* ctrl) { - ioctx_t* ioctx; - int rc; - int status; + int rc; + int status; if (ctx == NULL) { return -EINVAL; @@ -535,12 +460,7 @@ int xce_jobctrl(xce_ctx_t* ctx, xce_jobctrl_t* ctrl) return -EINVAL; } - ioctx = ioctx_alloc(ctx->ep, &(ctx->myinfo)); - if (ioctx == NULL) { - return -ENOMEM; - } - - rc = xce_jobctrl_req(ioctx, ctrl); + rc = xce_jobctrl_req(ctx->ep, ctrl); if (rc < 0) { return rc; } @@ -555,9 +475,8 @@ int xce_jobctrl(xce_ctx_t* ctx, xce_jobctrl_t* ctrl) int xce_queuectrl(xce_ctx_t* ctx, xce_queuectrl_t* ctrl) { - ioctx_t* ioctx; - int rc; - int status; + int rc; + int status; if (ctx == NULL) { return -EINVAL; @@ -567,12 +486,7 @@ int xce_queuectrl(xce_ctx_t* ctx, xce_queuectrl_t* ctrl) return -EINVAL; } - ioctx = ioctx_alloc(ctx->ep, &(ctx->myinfo)); - if (ioctx == NULL) { - return -ENOMEM; - } - - rc = xce_queuectrl_req(ioctx, ctrl); + rc = xce_queuectrl_req(ctx->ep, ctrl); if (rc < 0) { return rc; } @@ -587,9 +501,8 @@ int xce_queuectrl(xce_ctx_t* ctx, xce_queuectrl_t* ctrl) int xce_runtask(xce_ctx_t* ctx, xce_taskrun_t* runinfo) { - ioctx_t* ioctx; - int rc; - int status; + int rc; + int status; if (ctx == NULL) { return -EINVAL; @@ -599,12 +512,7 @@ int xce_runtask(xce_ctx_t* ctx, xce_taskrun_t* runinfo) return -EINVAL; } - ioctx = ioctx_alloc(ctx->ep, &(ctx->myinfo)); - if (ioctx == NULL) { - return -ENOMEM; - } - - rc = xce_runtask_req(ioctx, runinfo); + rc = xce_runtask_req(ctx->ep, runinfo); if (rc < 0) { return rc; } diff --git a/src/lib/xcectx.c b/src/lib/xcectx.c index e42079e2..502357cb 100644 --- a/src/lib/xcectx.c +++ b/src/lib/xcectx.c @@ -59,10 +59,11 @@ static int handle_reply(xce_ctx_t* ctx, xce_reply_t* reply) 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; + 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; @@ -73,7 +74,7 @@ static int vtask_entry(vtask_t* task) return -EINVAL; } - rc = ioctx_decode(&(task->ctx), &request); + rc = xce_decode(data->pkt, &request); if (rc < 0) { return rc; } @@ -114,91 +115,28 @@ static int vtask_entry(vtask_t* task) return 0; } -static int handle_input(endpoint_t* ep) + +static int handle_msg(endpoint_t* ep) { - xce_ctx_t* ctx = (xce_ctx_t*)ep->owner; - vtask_t* task; - int rc = 0; + xce_ctx_t* ctx = NULL; - task = vtask_alloc(); - if (task == NULL) { - return -ENOMEM; + if (ep == NULL) { + return -EINVAL; } - 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; + ctx = (xce_ctx_t*)ep->owner; + if (ctx == NULL) { + return -EINVAL; } - 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; + 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; + int rc; + SSL_CTX* ssl_ctx = NULL; if ((ctx == NULL) || (name == NULL) || (key == NULL)) { return -EINVAL; @@ -215,14 +153,24 @@ int xce_ctx_init(xce_ctx_t* ctx, char* name, char* 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; @@ -245,6 +193,8 @@ void xce_ctx_fini(xce_ctx_t* ctx) FREEUP(ctx->myinfo.name); FREEUP(ctx->myinfo.key); + cleanup_openssl(); + return; } @@ -264,7 +214,9 @@ int xce_ctx_set_endpoint(xce_ctx_t* ctx, endpoint_t* ep) return -EINVAL; } - ctx->ep = ep; + ctx->ep = ep; + ctx->ep->iomgr = &(ctx->iomgr); + return 0; } @@ -280,8 +232,9 @@ int xce_ctx_start(xce_ctx_t* ctx) return -EINVAL; } - endpoint_set_handler(ctx->ep, handle_io, ctx); - rc = iomgr_monitor(&(ctx->iomgr), ctx->ep); + 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; diff --git a/src/meson.build b/src/meson.build index 3bf244b6..62b96580 100644 --- a/src/meson.build +++ b/src/meson.build @@ -4,13 +4,17 @@ project('xce', meson_version: '>=1.5', default_options: [ 'c_std=gnu17', - 'c_args=-DDEBUG -O1']) + 'c_args=-DDEBUG -g -ggdb']) with_debug=false -if get_option('with_debug') +if get_option('buildtype') == 'debug' # add -DWITH_DEBUG=1 add_project_arguments('-DWITH_DEBUG=1', language: 'c') + add_project_arguments('-Wno-cpp', language: ['c', 'cpp']) + # add_project_arguments('-Og', language: 'c') + add_project_arguments('-fno-inline', language: 'c') + add_project_arguments('-fno-omit-frame-pointer', language: 'c') with_debug=true endif diff --git a/src/nixSupport/xce.nix b/src/nixSupport/xce.nix index 4a4f4aea..408b7888 100644 --- a/src/nixSupport/xce.nix +++ b/src/nixSupport/xce.nix @@ -5,7 +5,6 @@ myCommitId, withDebug? false, withAsan? false, - withoutDb? false, ... }: @@ -16,27 +15,29 @@ stdenv.mkDerivation { src = ./..; - nativeBuildInputs = [ - pkgs.meson - pkgs.ninja - pkgs.capnproto - pkgs.pkg-config - pkgs.cmake - pkgs.go - pkgs.openssl - pkgs.libtirpc - pkgs.readline - pkgs.ncurses - pkgs.capnpc - pkgs.which - pkgs.findutils - pkgs.libuv - pkgs.elogind - pkgs.sqlite - pkgs.hwloc - pkgs.linux-pam - pkgs.tinylog - pkgs.unity_test_with_color + nativeBuildInputs = with pkgs; [ + meson + ninja + capnproto + pkg-config + cmake + go + (openssl.overrideAttrs (oldAttrs: { + separateDebugInfo = true; + })) + libtirpc + readline + ncurses + capnpc + which + findutils + libuv + elogind + sqlite + hwloc + linux-pam + tinylog + unity_test_with_color ]; build-system = [ pkgs.meson ]; @@ -47,9 +48,8 @@ stdenv.mkDerivation { impureEnvVars = [ "HOME" ]; mesonFlags =[ - (lib.optional withDebug "-Dwith_debug=true") + (lib.optional withDebug "--buildtype=debug") (lib.optional withAsan "-Db_sanitize=address") - (lib.optional withoutDb "-Dwithout_db=true") ("-Dcommit_id=" + myCommitId) # (lib.optional (!with_debug) "-Dbuildtype=release") ];