diff --git a/src/controller/apiserver.c b/src/controller/apiserver.c index 33960e9e..e62207a5 100644 --- a/src/controller/apiserver.c +++ b/src/controller/apiserver.c @@ -3,6 +3,7 @@ #include "cluster.h" #include "controller.capnp.h" +#include "objmgr.h" #include "request.h" #include "xce/types.h" @@ -50,7 +51,7 @@ static int xce_apiserver_push_workers(cluster_t* c, endpoint_t* ep) deref_objs(workers.n_total, (void**)workers.workers); FREEUP(workers.workers); - return 0; + return rc; } static int check_job_(xce_job_t* j) @@ -81,30 +82,430 @@ static int xce_apiserver_push_jobs(cluster_t* c, endpoint_t* ep) deref_objs(jobs.n_total, (void**)jobs.jobs); FREEUP(jobs.jobs); + return rc; +} + +static int apiserver_push_joblist(cluster_t* c, xlist_t* l) +{ + int i; + for (i = 0; i < N_APISERVER_MAX; i++) { + if ((c->apiservers[i].name != NULL) && c->apiservers[i].ep != NULL) { + int rc; + xce_data_t data; + xce_jobs_data_t jobs = { .n_total = l->count, + .pos = 0, + .count = l->count, + .jobs = (xce_job_t**)l->data }; + + memset(&data, 0x0, sizeof(data)); + data.table = Data_data_schedjobs; + data.data.jobs = &jobs; + + rc = xce_push_data_req(c->apiservers[i].ep, c->myhost, &data, + check_controller, c, 100); + if (rc < 0) { + WARN("%s push job list to api server %s has problem, rc %d", + __func__, c->apiservers[i].name, rc); + } + } + } + + return 0; +} + +static int apiserver_job_new_fun(int type, int situation, void* data, + va_list ap) +{ + cluster_t* c = va_arg(ap, cluster_t*); + xlist_t* l = va_arg(ap, xlist_t*); + int rc = 0; + + switch (situation) { + case HOOK_AFTER_EVENT: + rc = -EINVAL; + if (l != NULL) { + rc = apiserver_push_joblist(c, l); + } + break; + default: + break; + } + + return rc; +} + +static int apiserver_job_start_fun(int type, int situation, void* data, + va_list ap) +{ + cluster_t* c = va_arg(ap, cluster_t*); + job_pidinfo_t* info = va_arg(ap, job_pidinfo_t*); + int i; + + switch (situation) { + case HOOK_AFTER_EVENT: + for (i = 0; i < N_APISERVER_MAX; i++) { + if ((c->apiservers[i].name != NULL) + && (c->apiservers[i].ep != NULL)) { + int rc; + xce_notify_data_t xdata + = { .event = NotifyData_data_schedStartJob, + .data = { .sched_start_job = info->jobid } }; + + /* FIXME: add job pid to API server? */ + rc = notify(c, c->apiservers[i].ep, &xdata); + if (rc < 0) { + WARN("%s notify job start to api server %s fail, rc %d", + __func__, c->apiservers[i].name, rc); + } + } + } + break; + default: + break; + } + + return 0; +} + +static int apiserver_job_exit_fun(int type, int situation, void* data, + va_list ap) +{ + cluster_t* c = va_arg(ap, cluster_t*); + xce_jobfinish_t* info = va_arg(ap, xce_jobfinish_t*); + int i; + + switch (situation) { + case HOOK_AFTER_EVENT: + for (i = 0; i < N_APISERVER_MAX; i++) { + if ((c->apiservers[i].name != NULL) + && (c->apiservers[i].ep != NULL)) { + int rc; + xce_notify_data_t xdata + = { .event = NotifyData_data_schedExitJob, + .data = { .sched_exit_job = info->jobid } }; + + /* FIXME: we need to send exit code of job to API server */ + rc = notify(c, c->apiservers[i].ep, &xdata); + if (rc < 0) { + WARN("%s notify job exit to api server %s fail, rc %d", + __func__, c->apiservers[i].name, rc); + } + } + } + break; + default: + break; + } + + return 0; +} + +static int apiserver_job_fail_fun(int type, int situation, void* data, + va_list ap) +{ + cluster_t* c = va_arg(ap, cluster_t*); + job_failinfo_t* info = va_arg(ap, job_failinfo_t*); + int i; + + switch (situation) { + case HOOK_AFTER_EVENT: + for (i = 0; i < N_APISERVER_MAX; i++) { + if ((c->apiservers[i].name != NULL) + && (c->apiservers[i].ep != NULL)) { + int rc; + xce_notify_data_t xdata + = { .event = NotifyData_data_schedFailJob, + .data = { .sched_fail_job = info->jobid } }; + + /* FIXME: we need to add fail reason of job to api server */ + rc = notify(c, c->apiservers[i].ep, &xdata); + if (rc < 0) { + WARN("%s notify job fail to api server %s fail, rc %d", + __func__, c->apiservers[i].name, rc); + } + } + } + break; + default: + break; + } + + return 0; +} + +static int apiserver_job_update_fun(int type, int situation, void* data, + va_list ap) +{ + cluster_t* c = va_arg(ap, cluster_t*); + job_t* job = va_arg(ap, job_t*); + int i; + switch (situation) { + case HOOK_AFTER_EVENT: + for (i = 0; i < N_APISERVER_MAX; i++) { + if ((c->apiservers[i].name != NULL) + && (c->apiservers[i].ep != NULL)) { + int rc; + xce_notify_data_t xdata + = { .event = NotifyData_data_schedUpdateJob, + .data = { .sched_update_job = &(job->job) } }; + + rc = notify(c, c->apiservers[i].ep, &xdata); + if (rc < 0) { + WARN("%s notify job update to api server %s fail, rc %d", + __func__, c->apiservers[i].name, rc); + } + } + } + break; + default: + break; + } + + return 0; +} + +static int apiserver_job_rusage_update_fun(int type, int situation, void* data, + va_list ap) +{ + cluster_t* c = va_arg(ap, cluster_t*); + xce_job_rusages_t* rusages = va_arg(ap, xce_job_rusages_t*); + int i; + + switch (situation) { + case HOOK_AFTER_EVENT: + for (i = 0; i < N_APISERVER_MAX; i++) { + if ((c->apiservers[i].name != NULL) + && (c->apiservers[i].ep != NULL)) { + int rc; + xce_notify_data_t xdata + = { .event = NotifyData_data_schedJobRusages, + .data = { .sched_job_rusages = rusages } }; + rc = notify(c, c->apiservers[i].ep, &xdata); + if (rc < 0) { + WARN("%s notify rusage update to api server %s fail, rc %d", + __func__, c->apiservers[i].name, rc); + } + } + } + break; + default: + break; + } + return 0; } static int apiserver_job_hook_fun(int type, int situation, int detail, void* data, va_list ap) { + switch (detail) { + case HOOK_DETAIL_JOB_NEW: + return apiserver_job_new_fun(type, situation, data, ap); + case HOOK_DETAIL_JOB_START: + return apiserver_job_start_fun(type, situation, data, ap); + case HOOK_DETAIL_JOB_EXIT: + return apiserver_job_exit_fun(type, situation, data, ap); + case HOOK_DETAIL_JOB_UPDATE: + return apiserver_job_update_fun(type, situation, data, ap); + case HOOK_DETAIL_JOB_FAIL: + return apiserver_job_fail_fun(type, situation, data, ap); + case HOOK_DETAIL_JOB_RUSAGE_UPDATE: + return apiserver_job_rusage_update_fun(type, situation, data, ap); + } + return 0; } +static int apiserver_push_workerlist(cluster_t* c, endpoint_t* ep, xlist_t* l) +{ + xce_data_t data; + xce_workers_data_t workers = { .n_total = l->count, + .pos = 0, + .count = l->count, + .workers = (xce_worker_t**)l->data }; + int rc = 0; + + memset(&data, 0x0, sizeof(data)); + + data.table = Data_data_schedworkers; + data.data.workers = &workers; + + rc = xce_push_data_req(ep, c->myhost, &data, check_controller, c, 100); + return rc; +} + +static int apiserver_worker_change(int type, int situation, void* data, + cluster_t* c, worker_t* w) +{ + int i; + + TRACE("%s worker %s idx %d situation %d", __func__, w->host.name, w->idx, + situation); + + collect_resval_with_idx(&(w->host.resvals), &(w->host.nresvals_to_sched), + &(w->host.resvals_to_sched)); + + switch (situation) { + case HOOK_AFTER_EVENT: + for (i = 0; i < N_APISERVER_MAX; i++) { + if ((c->apiservers[i].name != NULL) + && (c->apiservers[i].ep != NULL)) { + xlist_t wlist; + int rc; + + xlist_init(&wlist, 2); + xlist_store(&wlist, w); + rc = apiserver_push_workerlist(c, c->apiservers[i].ep, &wlist); + if (rc < 0) { + WARN("%s notify worker change failed to api server %s, rc " + "%d", + __func__, c->apiservers[i].name, rc); + } + xlist_fini(&wlist); + } + } + break; + default: + break; + } + + free_resval_with_idx_list(&(w->host.resvals_to_sched), + w->host.nresvals_to_sched); + + return 0; +} + +static int apiserver_worker_join(int type, int situation, void* data, + va_list ap) +{ + cluster_t* c = va_arg(ap, cluster_t*); + char* name = va_arg(ap, char*); + worker_t* w = NULL; + int rc; + + if (situation == HOOK_BEFORE_EVENT) { + /* nothing to do */ + return 0; + } + + w = workermgr_lock(&(c->workermgr), name, OBJREF_READ); + + rc = apiserver_worker_change(type, situation, data, c, w); + + /* FIXME: need to mark the jobs in this worker */ + obj_deref(w); + + return rc; +} + +static int apiserver_worker_dead(int type, int situation, void* data, + va_list ap) +{ + cluster_t* c = va_arg(ap, cluster_t*); + worker_t* w = va_arg(ap, worker_t*); + + /* FIXME: need to handle job status in this worker */ + if (situation == HOOK_BEFORE_EVENT) { + return 0; + } + + return apiserver_worker_change(type, situation, data, c, w); +} + +static int apiserver_worker_update(int type, int situation, void* data, + va_list ap) +{ + cluster_t* c = va_arg(ap, cluster_t*); + worker_t* w = va_arg(ap, worker_t*); + + if (situation == HOOK_BEFORE_EVENT) { + return 0; + } + + return apiserver_worker_change(type, situation, data, c, w); +} + static int apiserver_worker_hook_fun(int type, int situation, int detail, void* data, va_list ap) { + switch (detail) { + case HOOK_DETAIL_WORKER_JOIN: + return apiserver_worker_join(type, situation, data, ap); + case HOOK_DETAIL_WORKER_DEAD: + return apiserver_worker_dead(type, situation, data, ap); + case HOOK_DETAIL_WORKER_UPDATE: + return apiserver_worker_update(type, situation, data, ap); + } + return 0; } static int apiserver_resource_hook_fun(int type, int situation, int detail, void* data, va_list ap) { + cluster_t* c = va_arg(ap, cluster_t*); + xce_resval_list_t* resvals = va_arg(ap, xce_resval_list_t*); + int i; + + if (detail != HOOK_DETAIL_RESOURCE_UPDATE) { + return 0; + } + + switch (situation) { + case HOOK_AFTER_EVENT: + for (i = 0; i < N_APISERVER_MAX; i++) { + if ((c->apiservers[i].name != NULL) + && (c->apiservers[i].ep != NULL)) { + int rc; + xce_notify_data_t xdata + = { .event = NotifyData_data_schedUpdateResvals, + .data = { .sched_update_resvals = resvals } }; + + rc = notify(c, c->sched.ep, &xdata); + if (rc < 0) { + WARN("%s notify api server %s resource update fail, rc %d", + __func__, c->apiservers[i].name, rc); + } + } + } + break; + default: + break; + } + return 0; } static int apiserver_queue_hook_fun(int type, int situation, int detail, void* data, va_list ap) { + cluster_t* c = va_arg(ap, cluster_t*); + queue_t* q = va_arg(ap, queue_t*); + int i; + + switch (situation) { + case HOOK_AFTER_EVENT: + for (i = 0; i < N_APISERVER_MAX; i++) { + if ((c->apiservers[i].name != NULL) + && (c->apiservers[i].ep != NULL)) { + int rc; + xce_notify_data_t xdata + = { .event = NotifyData_data_schedUpdateQueue, + .data = { .sched_update_queue = &(q->queue) } }; + + rc = notify(c, c->sched.ep, &xdata); + if (rc < 0) { + WARN( + "%s notify queue change to api server %s failed, rc %d", + __func__, c->apiservers[i].name, rc); + } + } + } + break; + default: + break; + } + return 0; } @@ -134,6 +535,53 @@ int on_apiserver_join(cluster_t* c, endpoint_t* ep) int handle_apiserver_leave(cluster_t* c, endpoint_t* ep) { - /* FIXME */ + ep_private_t* priv; + int i; + + if (ep == NULL) { + return -EINVAL; + } + + priv = (ep_private_t*)ep->data; + if (priv == NULL) { + return -EINVAL; + } + + i = priv->u.idx; + + if (c->apiservers[i].name != NULL) { + obj_deref(c->apiservers[i].name); + c->apiservers[i].ep = NULL; + } + + /* FIXME: kill ep? */ return 0; } + +int add_apiserver(cluster_t* c, char* name, endpoint_t* ep) +{ + int i; + + for (i = 0; i < N_APISERVER_MAX; i++) { + if ((c->apiservers[i].name != NULL) + && (strcmp(c->apiservers[i].name, name) == 0)) { + if (c->apiservers[i].ep != NULL) { + /* FIXME: we need to delete 1 */ + } + c->apiservers[i].ep = ep; + return i; + } + } + + for (i = 0; i < N_APISERVER_MAX; i++) { + if (c->apiservers[i].name == NULL) { + c->apiservers[i].name + = create_obj_by_typename(&(c->objmgr), "xstring", name); + c->apiservers[i].ep = ep; + return i; + } + } + + /* better error code */ + return -1; +} diff --git a/src/controller/include/cluster.h b/src/controller/include/cluster.h index b1edd541..6f3a2639 100644 --- a/src/controller/include/cluster.h +++ b/src/controller/include/cluster.h @@ -137,6 +137,6 @@ extern int collect_resval_with_idx(zarray_t*, int*, xce_val_with_idx_t***); 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*); #endif diff --git a/src/controller/join.c b/src/controller/join.c index 0dc2ffa0..8e016ef9 100644 --- a/src/controller/join.c +++ b/src/controller/join.c @@ -117,21 +117,33 @@ static int handle_controller_join(cluster_t* c, endpoint_t* ep, return rc; } -static int handle_apiserver_join(cluster_t *c, endpoint_t *ep, xce_request_t *request) { - xce_reply_t reply; - ep_private_t *priv = (ep_private_t *)ep->data; - int rc = 0; +static int handle_apiserver_join(cluster_t* c, endpoint_t* ep, + xce_request_t* request) +{ + xce_reply_t reply; + xce_peerinfo_t* from = request->from; + int rc = 0; TRACE("%s api server join, ep %p", __func__, ep); init_reply_by_req(&reply, request); - rc = on_apiserver_join(c, ep); + rc = add_apiserver(c, from->name, ep); + if (rc >= 0) { + ep_private_t* priv = (ep_private_t*)ep->data; - reply.status = rc; + priv->u.idx = rc; + on_apiserver_join(c, ep); + } + + reply.status = rc; reply.data_which = RequestReply_data_nodata; xce_reply(ep, c->myhost, &reply); + if (rc < 0) { + ERROR("api server %s join failed, rc %d", from->name, rc); + return rc; + } return 0; } diff --git a/src/include/xce/cluster_def.h b/src/include/xce/cluster_def.h index 9d712ddd..851ac3cd 100644 --- a/src/include/xce/cluster_def.h +++ b/src/include/xce/cluster_def.h @@ -20,6 +20,7 @@ #include "zarray.h" #define N_CONTROLLER_MAX (16) +#define N_APISERVER_MAX (16) enum { XCE_CONTROLLER_HOOK_JOB = 0, @@ -82,31 +83,37 @@ typedef struct { } scheduler_t; typedef struct { - myconf_t* myconf; - xce_peerinfo_t* myhost; - hooksys_t hooksys; - iomgr_t iomgr; - objmgr_t objmgr; - zarray_t endpoints; - endpoint_t ep; - zarray_t configs; - zarray_t sharedinst; - controller_t controllers[N_CONTROLLER_MAX]; - int myidx; - int master; - int ncontrollers; - delaymgr_t delaymgr; - scheduler_t sched; - jobmgr_t jobmgr; - resmgr_t resmgr; - workermgr_t workermgr; - qmgr_t qmgr; - kvmgr_t configmgr; - kvmgr_t ugmgr; - kvmgr_t hgmgr; - vtaskmgr_t vtaskmgr; - cbuf_t wbuf; - int running; + char* name; + endpoint_t* ep; +} cluster_apiserver_t; + +typedef struct { + myconf_t* myconf; + xce_peerinfo_t* myhost; + hooksys_t hooksys; + iomgr_t iomgr; + objmgr_t objmgr; + zarray_t endpoints; + endpoint_t ep; + zarray_t configs; + zarray_t sharedinst; + controller_t controllers[N_CONTROLLER_MAX]; + cluster_apiserver_t apiservers[N_APISERVER_MAX]; + int myidx; + int master; + int ncontrollers; + delaymgr_t delaymgr; + scheduler_t sched; + jobmgr_t jobmgr; + resmgr_t resmgr; + workermgr_t workermgr; + qmgr_t qmgr; + kvmgr_t configmgr; + kvmgr_t ugmgr; + kvmgr_t hgmgr; + vtaskmgr_t vtaskmgr; + cbuf_t wbuf; + int running; } cluster_t; #endif /* XCE_CLUSTER_DEF_H */