diff --git a/src/apiserver/cmd/main.go b/src/apiserver/cmd/main.go index 8c3c90e8..58934038 100644 --- a/src/apiserver/cmd/main.go +++ b/src/apiserver/cmd/main.go @@ -8,68 +8,61 @@ package main #include "client.h" #include "netio.h" #include "controller.capnp.h" + #include "helper.h" - static int init_config(xce_client_config_t *config) { - config->nmasters = 1; - config->masters[0] = "localhost"; - config->master_ports[0] = 9990; - config->from = "xce_apiserver"; - config->key = "keepsecret"; - return 0; - } - - static xce_client_config_t config; - static struct { - xcecli_ctx_t ctx; - } xce_api_server_ctx; + typedef struct { + xcecli_ctx_t ctx; + uintptr_t go_handler; + void* go_apiserver; + } apiserver_ctx_t; - static uintptr_t go_handler = 0; - - extern int go_callback(uintptr_t, xce_request_t*, void *); + extern int go_callback(uintptr_t, xce_request_t*, apiserver_ctx_t *); static int xce_apiserver_handler(xce_request_t *req, void *arg) { int rc = -EINVAL; + apiserver_ctx_t *ctx = (apiserver_ctx_t *)arg; - if (go_handler != 0) { - rc = go_callback(go_handler, req, arg); + if (ctx == NULL) { + return -EINVAL; + } + + if (ctx->go_handler != 0) { + rc = go_callback(ctx->go_handler, req, arg); } return rc; } - static int xce_apiserver_init(uintptr_t handler) { + static int xce_apiserver_init(xce_client_config_t *config, apiserver_ctx_t *ctx) { int rc; - init_config(&config); - - go_handler = handler; + if ((config == NULL) || (ctx == NULL)) { + return -EINVAL; + } - rc = xcecli_ctx_init(&(xce_api_server_ctx.ctx), &config); + rc = xcecli_ctx_init(&(ctx->ctx), config); if (rc < 0) { return rc; } - xcecli_set_handler(xce_api_server_ctx.ctx, (xce_request_handler_t)xce_apiserver_handler, NULL); + xcecli_set_handler(ctx->ctx, (xce_request_handler_t)xce_apiserver_handler, ctx); - rc = xcecli_start(xce_api_server_ctx.ctx); + rc = xcecli_start(ctx->ctx); if (rc < 0) { return rc; } - rc = xcecli_join(xce_api_server_ctx.ctx, HostRole_apiserver, NULL); + rc = xcecli_join(ctx->ctx, HostRole_apiserver, NULL); return rc; } - static void* xce_request_body(xce_request_t *req) { - switch (req->body_which) { - case Request_body_configpushmsg: - return req->body.configpushmsg; - case Request_body_datamsg: - return req->body.datamsg; - default: - return NULL; - } + static xce_job_t **dataset_jobs(dataset_t *d) { + return (xce_job_t **)d->rows; + } + + static xce_worker_t **dataset_workers(dataset_t *d) { + return (xce_worker_t **)d->rows; } */ import "C" @@ -78,18 +71,28 @@ import ( "fmt" "unsafe" "runtime/cgo" + "log" + "net/http" ) -//export go_callback -func go_callback(h C.uintptr_t, req *C.xce_request_t, args unsafe.Pointer) C.int { - fn := cgo.Handle(h).Value().(func(*C.xce_request_t,unsafe.Pointer) C.int) - return fn(req, args); +type ApiServer struct { + // add data here } -func handle_config(config *C.xce_configdata_t) C.int { +//export go_callback +func go_callback(h C.uintptr_t, req *C.xce_request_t, ctx *C.apiserver_ctx_t) C.int { + fn := cgo.Handle(h).Value().(func(*C.xce_request_t,*C.apiserver_ctx_t) C.int) + return fn(req, ctx); +} + +func (apiserver *ApiServer) handle_config(config *C.xce_configdata_t) C.int { switch (config.data_which) { case C.ConfigData_data_clusterconfig: fmt.Printf("cluster configuration received\n") + // Extract configuration information from config + // - queues + // - resources information + // - and other information default: // fmt.Printf("unhandled config request %d\n", config.data_which) @@ -97,26 +100,70 @@ func handle_config(config *C.xce_configdata_t) C.int { return 0 } -func handle_dataset(dataset *C.dataset_t) C.int { - fmt.Printf("handle dataset table %d\n", dataset.table) +func print_job(job *C.xce_job_t) { + fmt.Printf("jobid `%v' user `%s' queue `%s'\n",uint64(job.jobid), C.GoString(job.submit.user), C.GoString(job.submit.queue)) +} + +func print_worker(w *C.xce_worker_t) { + fmt.Printf("worker `%s'\n", C.GoString(w.name)) +} + +func (apiserver *ApiServer) handle_job_push(jobs []*C.xce_job_t) C.int { + // add all jobs to apiserver + for i := range jobs { + print_job(jobs[i]) + } + return 0 } -func go_req_handler(request *C.xce_request_t, arg unsafe.Pointer) C.int { + +func (apiserver *ApiServer) handle_worker_push(workers []*C.xce_worker_t) C.int { + // add all workers to api server + for i := range workers { + print_worker(workers[i]) + } + return 0 +} + +func (apiserver *ApiServer)handle_dataset(dataset *C.dataset_t) C.int { + + fmt.Printf("handle dataset table %d\n", dataset.table) + switch(dataset.table) { + case C.Data_data_schedjobs: + fmt.Printf("%d jobs arrived\n", dataset.nrows) + var theJobs **C.xce_job_t = C.dataset_jobs(dataset) + + jobs := unsafe.Slice(theJobs, dataset.nrows) + + return apiserver.handle_job_push(jobs) + case C.Data_data_schedworkers: + fmt.Printf("%d workers arrived\n", dataset.nrows) + var theWorkers **C.xce_worker_t = C.dataset_workers(dataset) + + workers := unsafe.Slice(theWorkers, dataset.nrows) + return apiserver.handle_worker_push(workers) + default: + // + } + return 0 +} +func go_req_handler(request *C.xce_request_t, ctx *C.apiserver_ctx_t) C.int { var from = C.GoString(request.from.name) var body_which = request.body_which + var apiserver = (*ApiServer)(unsafe.Pointer(ctx.go_apiserver)) fmt.Println("api server handler called") - fmt.Printf("opcode %d body_which %d from %s\n", request.opcode, body_which, from) - + fmt.Printf("op `%s' body_which %d from %s\n", C.GoString(C.xce_request_op_name(request)), body_which, from) + switch (request.opcode) { case C.OPCode_push: switch (body_which) { case C.Request_body_configpushmsg: - // handle push config - return handle_config((*C.xce_configdata_t)(unsafe.Pointer(C.xce_request_body(request)))) + // handle push config + return apiserver.handle_config((*C.xce_configdata_t)(unsafe.Pointer(C.xce_request_body(request)))) case C.Request_body_datamsg: // handle pushed data msg - return handle_dataset((*C.dataset_t)(unsafe.Pointer(C.xce_request_body(request)))) + return apiserver.handle_dataset((*C.dataset_t)(unsafe.Pointer(C.xce_request_body(request)))) default: // something wrong } @@ -128,9 +175,37 @@ func go_req_handler(request *C.xce_request_t, arg unsafe.Pointer) C.int { return 0 } -func main() { - h := cgo.NewHandle(go_req_handler) - rc := C.xce_apiserver_init(C.uintptr_t(h)) - fmt.Printf("rc %d\n", rc); - h.Delete() +func init_config(config *C.xce_client_config_t) C.int { + config.nmasters =1 + config.masters[0] = C.CString("localhost") + config.master_ports[0] = C.int(9990); + config.from = C.CString("xce_apiserver") + config.key = C.CString("keepsecret$") + + return 0 +} + +func (apiserver *ApiServer) handle_job_query(w http.ResponseWriter, r *http.Request) { + fmt.Fprintf(w, "URL.Path = %q\n", r.URL.Path) +} + +func main() { + var config C.xce_client_config_t + var ctx C.apiserver_ctx_t + var apiserver ApiServer + + init_config(&config) + + h := cgo.NewHandle(go_req_handler) + defer h.Delete() + + ctx.go_handler = C.uintptr_t(h) + ctx.go_apiserver = unsafe.Pointer(&apiserver) + + rc := C.xce_apiserver_init(&config, &ctx) + fmt.Printf("rc %d\n", rc); + + http.HandleFunc("/jobs", (&apiserver).handle_job_query) + + log.Fatal(http.ListenAndServe("localhost:8000", nil)) } diff --git a/src/controller/apiserver.c b/src/controller/apiserver.c index a5d06b32..33960e9e 100644 --- a/src/controller/apiserver.c +++ b/src/controller/apiserver.c @@ -77,8 +77,7 @@ static int xce_apiserver_push_jobs(cluster_t* c, endpoint_t* ep) &(jobs.jobs)); jobs.count = jobs.n_total; - rc = xce_push_data_req(c->sched.ep, c->myhost, &data, check_controller, c, - 100); + rc = xce_push_data_req(ep, c->myhost, &data, check_controller, c, 100); deref_objs(jobs.n_total, (void**)jobs.jobs); FREEUP(jobs.jobs); diff --git a/src/include/helper.h b/src/include/helper.h new file mode 100644 index 00000000..850af76b --- /dev/null +++ b/src/include/helper.h @@ -0,0 +1,21 @@ +#if !defined(_HELPER_H_) + +#define _HELPER_H_ 1 + +#include + +#include "typemap.h" +#include "xce/types.h" + +extern void* xce_request_body(xce_request_t*); +extern char* xce_request_op_name(xce_request_t*); +extern void* xce_val_ptr(xce_val_t*); +extern void* xce_val_ptr(xce_val_t*); +extern void* xce_configdata_ptr(xce_configdata_t*); +extern void* xce_data_ptr(xce_data_t*); +extern void* xce_update_data_ptr(xce_update_data_t*); +extern void* xce_notify_data_ptr(xce_notify_data_t*); +extern void* xce_jobctrl_data_ptr(xce_jobctrl_t*); +extern void* xce_reply_data_ptr(xce_reply_t*); + +#endif diff --git a/src/lib/helper.c b/src/lib/helper.c new file mode 100644 index 00000000..b765a850 --- /dev/null +++ b/src/lib/helper.c @@ -0,0 +1,251 @@ +#include +#include + +#include "controller.capnp.h" +#include "helper.h" +#include "typemap.h" +#include "xce/types.h" + +void* xce_request_body(xce_request_t* req) +{ + switch (req->body_which) { + case Request_body_nomsg: + break; + case Request_body_replymsg: + return req->body.replymsg; + case Request_body_hellomsg: + return req->body.hellomsg; + case Request_body_joinmsg: + return req->body.joinmsg; + case Request_body_configpushmsg: + return req->body.configpushmsg; + case Request_body_pushmsg: + return req->body.pushmsg; + case Request_body_getmsg: + return req->body.getmsg; + case Request_body_datamsg: + return req->body.datamsg; + case Request_body_dispatchmsg: + return req->body.dispatchmsg; + case Request_body_jobstartmsg: + return req->body.jobstartmsg; + case Request_body_notifymsg: + return req->body.notifymsg; + case Request_body_jobsubmsg: + return req->body.jobsubmsg; + case Request_body_updatemsg: + return req->body.updatemsg; + case Request_body_jobctrlmsg: + return req->body.jobctrlmsg; + case Request_body_queuectrlmsg: + return req->body.queuectrlmsg; + case Request_body_runtaskmsg: + return req->body.runtaskmsg; + case Request_body_preasonmsg: + return req->body.preasonmsg; + case Request_body_clrpreasonmsg: + return req->body.clrpreasonmsg; + default: + break; + } + + return NULL; +} + +char* xce_request_op_name(xce_request_t* req) +{ + switch (req->opcode) { + case OPCode_nop: + return "nop"; + case OPCode_reply: + return "reply"; + case OPCode_hello: + return "hello"; + case OPCode_join: + return "join"; + case OPCode_shutdown: + return "shutdown"; + case OPCode_get: + return "get"; + case OPCode_push: + return "push"; + case OPCode_update: + return "update"; + case OPCode_jobdispatch: + return "jobdispatch"; + case OPCode_startjob: + return "startjob"; + case OPCode_notify: + return "notify"; + case OPCode_jobsub: + return "jobsub"; + case OPCode_jobstatus: + return "jobstatus"; + case OPCode_jobrusage: + return "jobrusage"; + case OPCode_hostcontrol: + return "hostcontrol"; + case OPCode_jobcontrol: + return "jobcontrol"; + case OPCode_clusterconf: + return "clusterconf"; + case OPCode_queuecontrol: + return "queuecontrol"; + case OPCode_runtasks: + return "runtasks"; + case OPCode_setpreason: + return "setpreason"; + case OPCode_clrpreason: + return "clrpreason"; + default: + break; + } + + return "unknown"; +} + +void *xce_val_ptr(xce_val_t *v) { + switch(v->which) { + case Value_u_ivalue32: + return &(v->u.ivalue32); + case Value_u_uvalue32: + return &(v->u.uvalue32); + case Value_u_ivalue64: + return &(v->u.ivalue64); + case Value_u_uvalue64: + return &(v->u.uvalue64); + case Value_u_svalue: + return v->u.svalue; + case Value_u_fvalue: + return &(v->u.fvalue); + case Value_u_bvalue: + return &(v->u.bvalue); + default: + break; + } + + return NULL; +} + +void* xce_configdata_ptr(xce_configdata_t *d) { + switch(d->data_which) { + case ConfigData_data_schedconfig: + return d->data.conf4sched; + case ConfigData_data_workerconfig: + return d->data.conf4worker; + case ConfigData_data_clusterconfig: + return d->data.clusterconfig; + default: + break; + } + + return NULL; +} + +void* xce_data_ptr(xce_data_t *d) { + switch(d->table) { + case Data_data_schedjobs: + return d->data.jobs; + case Data_data_schedworkers: + return d->data.workers; + case Data_data_jobs: + return d->data.jobs; + case Data_data_workers: + return d->data.workers; + case Data_data_resources: + return d->data.resources; + case Data_data_resvals: + return d->data.resvals; + default: + break; + } + return NULL; +} + +void* xce_update_data_ptr(xce_update_data_t *d) { + switch(d->table) { + case Update_data_nodata: + break; + case Update_data_resvals: + return d->data.resvals; + case Update_data_jobrusages: + return d->data.jobrusages; + default: + break; + } + return NULL; +} + +void* xce_notify_data_ptr(xce_notify_data_t *d) { + switch(d->event) { + case NotifyData_data_jobPidInfo: + return d->data.pidinfo; + case NotifyData_data_jobFailInfo: + return d->data.failinfo; + case NotifyData_data_jobDispatchFail: + return d->data.dispatchfail; + case NotifyData_data_jobFinishInfo: + return d->data.finishinfo; + case NotifyData_data_schedStartJob: + return (void *)d->data.sched_start_job; + case NotifyData_data_schedExitJob: + return (void *)d->data.sched_exit_job; + case NotifyData_data_schedFailJob: + return (void *)d->data.sched_fail_job; + case NotifyData_data_schedUpdateJob: + return d->data.sched_update_job; + case NotifyData_data_schedUpdateResvals: + return d->data.sched_update_resvals; + case NotifyData_data_stoppedJob: + return (void *)d->data.stopped_job; + case NotifyData_data_resumedJob: + return (void *)d->data.resumed_job; + case NotifyData_data_ckptinfo: + return d->data.ckptinfo; + case NotifyData_data_schedUpdateQueue: + return d->data.sched_update_queue; + case NotifyData_data_taskPidInfo: + return d->data.taskinfo; + case NotifyData_data_taskFailInfo: + return d->data.taskfailinfo; + case NotifyData_data_taskFinish: + return d->data.taskfinish; + case NotifyData_data_postDoneJob: + return (void *)d->data.pdone_job; + case NotifyData_data_schedJobRusages: + return d->data.sched_job_rusages; + default: + break; + } + return NULL; +} + +void* xce_jobctrl_data_ptr(xce_jobctrl_t *d) { + switch(d->action) { + case JobControl_u_siginfo: + return d->u.siginfo; + case JobControl_u_ckptoption: + return d->u.ckptoption; + default: + break; + } + return NULL; +} + +void* xce_reply_data_ptr(xce_reply_t *reply) { + switch(reply->data_which) { + case RequestReply_data_nodata: + break; + case RequestReply_data_debugmsg: + return reply->data.debugmsg; + case RequestReply_data_getdata: + return reply->data.getdata; + case RequestReply_data_jobid: + return (void *)reply->data.jobid; + case RequestReply_data_clusterconf: + return reply->data.clusterconf; + default: + break; + } + return NULL; +} diff --git a/src/lib/meson.build b/src/lib/meson.build index 8d9bafda..00826df7 100644 --- a/src/lib/meson.build +++ b/src/lib/meson.build @@ -36,6 +36,7 @@ libxce_src = [ gen_srcs[0], 'group.c', 'xcelimit.c', 'xvec.c', + 'helper.c', ] incdir = include_directories('../include') diff --git a/src/nixSupport/xce.nix b/src/nixSupport/xce.nix index 7c4a5c35..792a8497 100644 --- a/src/nixSupport/xce.nix +++ b/src/nixSupport/xce.nix @@ -22,6 +22,7 @@ stdenv.mkDerivation { pkg-config cmake go + delve (openssl.overrideAttrs (oldAttrs: { separateDebugInfo = true; }))