lei.chen/myxce
1
0
Fork 1
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

security communication with openssl and simplify netio

Browse source
This commit is contained in:
Rongsong Shen 2025-11-18 11:21:24 +08:00
parent 2a7b7c39bb
commit 8f796b3109
20 changed files with 1176 additions and 922 deletions
Download patch file Download diff file Expand all files Collapse all files

175
src/controller/cluster.c
View file

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

9
src/controller/dataex.c
View file

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

28
src/controller/job.c
View file

@ -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",

6
src/controller/main.c
View file

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

9
src/controller/task.c
View file

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

2
src/controller/testsuite/test.c
View file

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

2
src/include/myconf.h
View file

@ -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*);

114
src/include/netio.h
View file

@ -2,6 +2,8 @@
#define _NETIO_H_ 1
#include <openssl/err.h>
#include <openssl/ssl.h>
#include <stdint.h>
#include <stdio.h>
#include <sys/epoll.h>
@ -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*);

52
src/include/request.h
View file

@ -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

26
src/include/vtaskmgr.h
View file

@ -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*);

13
src/lib/client.c
View file

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

3
src/lib/meson.build
View file

@ -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,

912
src/lib/netio.c
View file

File diff suppressed because it is too large Load diff

294
src/lib/request.c
View file

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

2
src/lib/testsuite/test_idgen.c
View file

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

92
src/lib/vtaskmgr.c
View file

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

184
src/lib/xceapi.c
View file

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

119
src/lib/xcectx.c
View file

@ -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;

8
src/meson.build
View file

@ -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

48
src/nixSupport/xce.nix
View file

@ -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")
];