diff --git a/src/apiserver/cmd/main.go b/src/apiserver/cmd/main.go index 58934038..19358971 100644 --- a/src/apiserver/cmd/main.go +++ b/src/apiserver/cmd/main.go @@ -127,16 +127,16 @@ func (apiserver *ApiServer) handle_worker_push(workers []*C.xce_worker_t) C.int func (apiserver *ApiServer)handle_dataset(dataset *C.dataset_t) C.int { - fmt.Printf("handle dataset table %d\n", dataset.table) + fmt.Printf("handle dataset table `%s'\n", C.GoString(C.xce_dataset_table(dataset.table))) switch(dataset.table) { - case C.Data_data_schedjobs: + case C.Data_data_jobs: 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: + case C.Data_data_workers: fmt.Printf("%d workers arrived\n", dataset.nrows) var theWorkers **C.xce_worker_t = C.dataset_workers(dataset) @@ -153,7 +153,7 @@ func go_req_handler(request *C.xce_request_t, ctx *C.apiserver_ctx_t) C.int { var apiserver = (*ApiServer)(unsafe.Pointer(ctx.go_apiserver)) fmt.Println("api server handler called") - fmt.Printf("op `%s' body_which %d from %s\n", C.GoString(C.xce_request_op_name(request)), body_which, from) + fmt.Printf("op `%s' msg `%s' from %s\n", C.GoString(C.xce_request_op_name(request)), C.GoString(C.xce_request_msg_name(request)), from) switch (request.opcode) { case C.OPCode_push: diff --git a/src/controller/apiserver.c b/src/controller/apiserver.c index 9921edcb..958239c1 100644 --- a/src/controller/apiserver.c +++ b/src/controller/apiserver.c @@ -147,9 +147,8 @@ static int apiserver_job_start_fun(int type, int situation, void* data, if ((c->apiservers[i].name != NULL) && (c->apiservers[i].ep != NULL)) { int rc; - xce_notify_data_t xdata - = { .event = NotifyData_data_jobPidInfo, - .data = { .pidinfo = info } }; + xce_notify_data_t xdata = { .event = NotifyData_data_jobPidInfo, + .data = { .pidinfo = info } }; /* FIXME: add job pid to API server? */ rc = notify(c, c->apiservers[i].ep, &xdata); @@ -461,7 +460,7 @@ static int apiserver_resource_hook_fun(int type, int situation, int detail, = { .event = NotifyData_data_schedUpdateResvals, .data = { .sched_update_resvals = resvals } }; - rc = notify(c, c->sched.ep, &xdata); + rc = notify(c, c->apiservers[i].ep, &xdata); if (rc < 0) { WARN("%s notify api server %s resource update fail, rc %d", __func__, c->apiservers[i].name, rc); @@ -509,27 +508,33 @@ static int apiserver_queue_hook_fun(int type, int situation, int detail, return 0; } +int apiserver_init(cluster_t* c) +{ + int i; + + for(i = 0; i < N_APISERVER_MAX; i ++) { + c->apiservers[i].name = NULL; + c->apiservers[i].ep = NULL; + } + + xce_add_hook(&(c->hooksys), XCE_CONTROLLER_HOOK_JOB, + (hook_fn_t)apiserver_job_hook_fun, NULL); + xce_add_hook(&(c->hooksys), XCE_CONTROLLER_HOOK_WORKER, + (hook_fn_t)apiserver_worker_hook_fun, NULL); + xce_add_hook(&(c->hooksys), XCE_CONTROLLER_HOOK_RESOURCE, + (hook_fn_t)apiserver_resource_hook_fun, NULL); + xce_add_hook(&(c->hooksys), XCE_CONTROLLER_HOOK_QUEUE, + (hook_fn_t)apiserver_queue_hook_fun, NULL); + + return 0; +} + int on_apiserver_join(cluster_t* c, endpoint_t* ep) { - /* FIXME: use lock to protect this later */ - static int hook_installed = 0; - xce_apiserver_push_configs(c, ep); xce_apiserver_push_workers(c, ep); xce_apiserver_push_jobs(c, ep); - if (!hook_installed) { - hook_installed = 1; - - xce_add_hook(&(c->hooksys), XCE_CONTROLLER_HOOK_JOB, - (hook_fn_t)apiserver_job_hook_fun, NULL); - xce_add_hook(&(c->hooksys), XCE_CONTROLLER_HOOK_WORKER, - (hook_fn_t)apiserver_worker_hook_fun, NULL); - xce_add_hook(&(c->hooksys), XCE_CONTROLLER_HOOK_RESOURCE, - (hook_fn_t)apiserver_resource_hook_fun, NULL); - xce_add_hook(&(c->hooksys), XCE_CONTROLLER_HOOK_QUEUE, - (hook_fn_t)apiserver_queue_hook_fun, NULL); - } return 0; } diff --git a/src/controller/cluster.c b/src/controller/cluster.c index 9d8c688f..9c9e48e2 100644 --- a/src/controller/cluster.c +++ b/src/controller/cluster.c @@ -542,6 +542,8 @@ int cluster_init(cluster_t* c) delaymgr_init(&(c->delaymgr), 2, run_delay_tasks); + apiserver_init(c); + xce_run_hooks(&(c->hooksys), XCE_CONTROLLER_HOOK_CLUSTER, HOOK_AFTER_EVENT, HOOK_DETAIL_CLUSTER_INIT, c); diff --git a/src/controller/include/cluster.h b/src/controller/include/cluster.h index 6f3a2639..273c1237 100644 --- a/src/controller/include/cluster.h +++ b/src/controller/include/cluster.h @@ -138,5 +138,6 @@ extern int do_set_preason(cluster_t*, endpoint_t*, xce_request_t*); extern int do_clr_preason(cluster_t*, endpoint_t*, xce_request_t*); extern int handle_apiserver_leave(cluster_t*, endpoint_t*); extern int add_apiserver(cluster_t*, char*, endpoint_t*); +extern int apiserver_init(cluster_t*); #endif diff --git a/src/include/helper.h b/src/include/helper.h index 850af76b..b28bd09a 100644 --- a/src/include/helper.h +++ b/src/include/helper.h @@ -9,6 +9,7 @@ extern void* xce_request_body(xce_request_t*); extern char* xce_request_op_name(xce_request_t*); +extern char* xce_request_msg_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*); @@ -17,5 +18,6 @@ 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*); +extern char* xce_dataset_table(int); #endif diff --git a/src/lib/helper.c b/src/lib/helper.c index b765a850..33e7d2ae 100644 --- a/src/lib/helper.c +++ b/src/lib/helper.c @@ -52,6 +52,52 @@ void* xce_request_body(xce_request_t* req) return NULL; } +char* xce_request_msg_name(xce_request_t* req) +{ + switch (req->body_which) { + case Request_body_nomsg: + return "nomsg"; + case Request_body_replymsg: + return "replymsg"; + case Request_body_hellomsg: + return "hellomsg"; + case Request_body_joinmsg: + return "joinmsg"; + case Request_body_configpushmsg: + return "configmsg"; + case Request_body_pushmsg: + return "pushmsg"; + case Request_body_getmsg: + return "getmsg"; + case Request_body_datamsg: + return "datamsg"; + case Request_body_dispatchmsg: + return "dispatchmsg"; + case Request_body_jobstartmsg: + return "jobstartmsg"; + case Request_body_notifymsg: + return "notifymsg"; + case Request_body_jobsubmsg: + return "jobsubmsg"; + case Request_body_updatemsg: + return "updatemsg"; + case Request_body_jobctrlmsg: + return "jobctrlmsg"; + case Request_body_queuectrlmsg: + return "queuectrlmsg"; + case Request_body_runtaskmsg: + return "runtaskmsg"; + case Request_body_preasonmsg: + return "preasonmsg"; + case Request_body_clrpreasonmsg: + return "clrpreasonmsg"; + default: + break; + } + + return "unknown"; +} + char* xce_request_op_name(xce_request_t* req) { switch (req->opcode) { @@ -104,8 +150,9 @@ char* xce_request_op_name(xce_request_t* req) return "unknown"; } -void *xce_val_ptr(xce_val_t *v) { - switch(v->which) { +void* xce_val_ptr(xce_val_t* v) +{ + switch (v->which) { case Value_u_ivalue32: return &(v->u.ivalue32); case Value_u_uvalue32: @@ -127,8 +174,9 @@ void *xce_val_ptr(xce_val_t *v) { return NULL; } -void* xce_configdata_ptr(xce_configdata_t *d) { - switch(d->data_which) { +void* xce_configdata_ptr(xce_configdata_t* d) +{ + switch (d->data_which) { case ConfigData_data_schedconfig: return d->data.conf4sched; case ConfigData_data_workerconfig: @@ -142,8 +190,9 @@ void* xce_configdata_ptr(xce_configdata_t *d) { return NULL; } -void* xce_data_ptr(xce_data_t *d) { - switch(d->table) { +void* xce_data_ptr(xce_data_t* d) +{ + switch (d->table) { case Data_data_schedjobs: return d->data.jobs; case Data_data_schedworkers: @@ -162,8 +211,9 @@ void* xce_data_ptr(xce_data_t *d) { return NULL; } -void* xce_update_data_ptr(xce_update_data_t *d) { - switch(d->table) { +void* xce_update_data_ptr(xce_update_data_t* d) +{ + switch (d->table) { case Update_data_nodata: break; case Update_data_resvals: @@ -176,8 +226,9 @@ void* xce_update_data_ptr(xce_update_data_t *d) { return NULL; } -void* xce_notify_data_ptr(xce_notify_data_t *d) { - switch(d->event) { +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: @@ -187,19 +238,19 @@ void* xce_notify_data_ptr(xce_notify_data_t *d) { case NotifyData_data_jobFinishInfo: return d->data.finishinfo; case NotifyData_data_schedStartJob: - return (void *)d->data.sched_start_job; + return (void*)d->data.sched_start_job; case NotifyData_data_schedExitJob: - return (void *)d->data.sched_exit_job; + return (void*)d->data.sched_exit_job; case NotifyData_data_schedFailJob: - return (void *)d->data.sched_fail_job; + 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; + return (void*)d->data.stopped_job; case NotifyData_data_resumedJob: - return (void *)d->data.resumed_job; + return (void*)d->data.resumed_job; case NotifyData_data_ckptinfo: return d->data.ckptinfo; case NotifyData_data_schedUpdateQueue: @@ -211,7 +262,7 @@ void* xce_notify_data_ptr(xce_notify_data_t *d) { case NotifyData_data_taskFinish: return d->data.taskfinish; case NotifyData_data_postDoneJob: - return (void *)d->data.pdone_job; + return (void*)d->data.pdone_job; case NotifyData_data_schedJobRusages: return d->data.sched_job_rusages; default: @@ -220,8 +271,9 @@ void* xce_notify_data_ptr(xce_notify_data_t *d) { return NULL; } -void* xce_jobctrl_data_ptr(xce_jobctrl_t *d) { - switch(d->action) { +void* xce_jobctrl_data_ptr(xce_jobctrl_t* d) +{ + switch (d->action) { case JobControl_u_siginfo: return d->u.siginfo; case JobControl_u_ckptoption: @@ -232,8 +284,9 @@ void* xce_jobctrl_data_ptr(xce_jobctrl_t *d) { return NULL; } -void* xce_reply_data_ptr(xce_reply_t *reply) { - switch(reply->data_which) { +void* xce_reply_data_ptr(xce_reply_t* reply) +{ + switch (reply->data_which) { case RequestReply_data_nodata: break; case RequestReply_data_debugmsg: @@ -241,7 +294,7 @@ void* xce_reply_data_ptr(xce_reply_t *reply) { case RequestReply_data_getdata: return reply->data.getdata; case RequestReply_data_jobid: - return (void *)reply->data.jobid; + return (void*)reply->data.jobid; case RequestReply_data_clusterconf: return reply->data.clusterconf; default: @@ -249,3 +302,25 @@ void* xce_reply_data_ptr(xce_reply_t *reply) { } return NULL; } + +char* xce_dataset_table(int table) +{ + switch (table) { + case Data_data_schedjobs: + return "schedjobs"; + case Data_data_jobs: + return "jobs"; + case Data_data_workers: + return "workers"; + case Data_data_schedworkers: + return "schedworkers"; + case Data_data_resources: + return "resources"; + case Data_data_resvals: + return "resvals"; + default: + break; + } + + return "unknown"; +} diff --git a/src/lib/request.c b/src/lib/request.c index 80e159cf..aaf8496c 100644 --- a/src/lib/request.c +++ b/src/lib/request.c @@ -281,6 +281,7 @@ int xce_push_data_req(endpoint_t* ep, xce_peerinfo_t* from, xce_data_t* data, data1.table = data->table; switch (data->table) { case Data_data_schedjobs: + case Data_data_jobs: nrows = data->data.jobs->n_total; data1.data.jobs = (xce_jobs_data_t*)calloc(1, sizeof(xce_jobs_data_t)); if (data1.data.jobs == NULL) { @@ -288,6 +289,7 @@ int xce_push_data_req(endpoint_t* ep, xce_peerinfo_t* from, xce_data_t* data, } break; case Data_data_schedworkers: + case Data_data_workers: nrows = data->data.workers->n_total; data1.data.workers = (xce_workers_data_t*)calloc(1, sizeof(xce_workers_data_t)); @@ -295,6 +297,10 @@ int xce_push_data_req(endpoint_t* ep, xce_peerinfo_t* from, xce_data_t* data, rc = -ENOMEM; } break; + case Data_data_resources: + break; + case Data_data_resvals: + break; default: rc = -EINVAL; break; @@ -324,6 +330,7 @@ int xce_push_data_req(endpoint_t* ep, xce_peerinfo_t* from, xce_data_t* data, switch (data->table) { case Data_data_schedjobs: + case Data_data_jobs: data1.data.jobs->n_total = data->data.jobs->n_total; data1.data.jobs->pos = pos; data1.data.jobs->count = nc; @@ -334,6 +341,7 @@ int xce_push_data_req(endpoint_t* ep, xce_peerinfo_t* from, xce_data_t* data, } break; case Data_data_schedworkers: + case Data_data_workers: data1.data.workers->n_total = data->data.workers->n_total; data1.data.workers->pos = pos; data1.data.workers->count = nc; @@ -344,6 +352,10 @@ int xce_push_data_req(endpoint_t* ep, xce_peerinfo_t* from, xce_data_t* data, = &(data->data.workers->workers[pos]); } break; + case Data_data_resources: + break; + case Data_data_resvals: + break; default: break; }