add more helper functions more example code

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

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

src/include/helper.h

@@ -0,0 +1,21 @@
+#if !defined(_HELPER_H_)
+
+#define _HELPER_H_ 1
+
+#include <stdlib.h>
+
+#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

src/lib/helper.c

@@ -0,0 +1,251 @@
+#include <errno.h>
+#include <stdlib.h>
+
+#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;
+}

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')

src/nixSupport/xce.nix

@@ -22,6 +22,7 @@ stdenv.mkDerivation {
     pkg-config
     cmake
     go
+    delve
     (openssl.overrideAttrs (oldAttrs: {
       separateDebugInfo = true;
     }))