diff --git a/src/controller/apiserver.c b/src/controller/apiserver.c index 958239c1..058cfffd 100644 --- a/src/controller/apiserver.c +++ b/src/controller/apiserver.c @@ -89,6 +89,7 @@ static int apiserver_push_joblist(cluster_t* c, xlist_t* l) { int i; for (i = 0; i < N_APISERVER_MAX; i++) { + mylock_lock(&(c->apiservers[i].lock)); if ((c->apiservers[i].name != NULL) && c->apiservers[i].ep != NULL) { int rc; xce_data_t data; @@ -108,6 +109,7 @@ static int apiserver_push_joblist(cluster_t* c, xlist_t* l) __func__, c->apiservers[i].name, rc); } } + mylock_unlock(&(c->apiservers[i].lock)); } return 0; @@ -144,6 +146,7 @@ static int apiserver_job_start_fun(int type, int situation, void* data, switch (situation) { case HOOK_AFTER_EVENT: for (i = 0; i < N_APISERVER_MAX; i++) { + mylock_lock(&(c->apiservers[i].lock)); if ((c->apiservers[i].name != NULL) && (c->apiservers[i].ep != NULL)) { int rc; @@ -157,6 +160,7 @@ static int apiserver_job_start_fun(int type, int situation, void* data, __func__, c->apiservers[i].name, rc); } } + mylock_unlock(&(c->apiservers[i].lock)); } break; default: @@ -176,6 +180,7 @@ static int apiserver_job_exit_fun(int type, int situation, void* data, switch (situation) { case HOOK_AFTER_EVENT: for (i = 0; i < N_APISERVER_MAX; i++) { + mylock_lock(&(c->apiservers[i].lock)); if ((c->apiservers[i].name != NULL) && (c->apiservers[i].ep != NULL)) { int rc; @@ -190,6 +195,7 @@ static int apiserver_job_exit_fun(int type, int situation, void* data, __func__, c->apiservers[i].name, rc); } } + mylock_unlock(&(c->apiservers[i].lock)); } break; default: @@ -209,6 +215,7 @@ static int apiserver_job_fail_fun(int type, int situation, void* data, switch (situation) { case HOOK_AFTER_EVENT: for (i = 0; i < N_APISERVER_MAX; i++) { + mylock_lock(&(c->apiservers[i].lock)); if ((c->apiservers[i].name != NULL) && (c->apiservers[i].ep != NULL)) { int rc; @@ -223,6 +230,7 @@ static int apiserver_job_fail_fun(int type, int situation, void* data, __func__, c->apiservers[i].name, rc); } } + mylock_lock(&(c->apiservers[i].lock)); } break; default: @@ -241,6 +249,7 @@ static int apiserver_job_update_fun(int type, int situation, void* data, switch (situation) { case HOOK_AFTER_EVENT: for (i = 0; i < N_APISERVER_MAX; i++) { + mylock_lock(&(c->apiservers[i].lock)); if ((c->apiservers[i].name != NULL) && (c->apiservers[i].ep != NULL)) { int rc; @@ -254,6 +263,7 @@ static int apiserver_job_update_fun(int type, int situation, void* data, __func__, c->apiservers[i].name, rc); } } + mylock_unlock(&(c->apiservers[i].lock)); } break; default: @@ -273,6 +283,7 @@ static int apiserver_job_rusage_update_fun(int type, int situation, void* data, switch (situation) { case HOOK_AFTER_EVENT: for (i = 0; i < N_APISERVER_MAX; i++) { + mylock_lock(&(c->apiservers[i].lock)); if ((c->apiservers[i].name != NULL) && (c->apiservers[i].ep != NULL)) { int rc; @@ -285,6 +296,7 @@ static int apiserver_job_rusage_update_fun(int type, int situation, void* data, __func__, c->apiservers[i].name, rc); } } + mylock_unlock(&(c->apiservers[i].lock)); } break; default: @@ -347,6 +359,7 @@ static int apiserver_worker_change(int type, int situation, void* data, switch (situation) { case HOOK_AFTER_EVENT: for (i = 0; i < N_APISERVER_MAX; i++) { + mylock_lock(&(c->apiservers[i].lock)); if ((c->apiservers[i].name != NULL) && (c->apiservers[i].ep != NULL)) { xlist_t wlist; @@ -362,6 +375,7 @@ static int apiserver_worker_change(int type, int situation, void* data, } xlist_fini(&wlist); } + mylock_unlock(&(c->apiservers[i].lock)); } break; default: @@ -453,6 +467,7 @@ static int apiserver_resource_hook_fun(int type, int situation, int detail, switch (situation) { case HOOK_AFTER_EVENT: for (i = 0; i < N_APISERVER_MAX; i++) { + mylock_lock(&(c->apiservers[i].lock)); if ((c->apiservers[i].name != NULL) && (c->apiservers[i].ep != NULL)) { int rc; @@ -466,6 +481,7 @@ static int apiserver_resource_hook_fun(int type, int situation, int detail, __func__, c->apiservers[i].name, rc); } } + mylock_unlock(&(c->apiservers[i].lock)); } break; default: @@ -485,6 +501,7 @@ static int apiserver_queue_hook_fun(int type, int situation, int detail, switch (situation) { case HOOK_AFTER_EVENT: for (i = 0; i < N_APISERVER_MAX; i++) { + mylock_lock(&(c->apiservers[i].lock)); if ((c->apiservers[i].name != NULL) && (c->apiservers[i].ep != NULL)) { int rc; @@ -499,6 +516,7 @@ static int apiserver_queue_hook_fun(int type, int situation, int detail, __func__, c->apiservers[i].name, rc); } } + mylock_unlock(&(c->apiservers[i].lock)); } break; default: @@ -512,11 +530,12 @@ int apiserver_init(cluster_t* c) { int i; - for(i = 0; i < N_APISERVER_MAX; i ++) { + for (i = 0; i < N_APISERVER_MAX; i++) { + mylock_init(&(c->apiservers[i].lock)); c->apiservers[i].name = NULL; - c->apiservers[i].ep = 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, @@ -553,6 +572,10 @@ int handle_apiserver_leave(cluster_t* c, endpoint_t* ep) } i = priv->u.idx; + if (i < 0) { + SET_ERROR(EINVAL); + return -errno; + } if (c->apiservers[i].name != NULL) { obj_deref(c->apiservers[i].name); diff --git a/src/include/xce/cluster_def.h b/src/include/xce/cluster_def.h index 851ac3cd..e77b8df9 100644 --- a/src/include/xce/cluster_def.h +++ b/src/include/xce/cluster_def.h @@ -83,6 +83,7 @@ typedef struct { } scheduler_t; typedef struct { + mylock_t lock; char* name; endpoint_t* ep; } cluster_apiserver_t; diff --git a/src/lib/request.c b/src/lib/request.c index aaf8496c..a3f11753 100644 --- a/src/lib/request.c +++ b/src/lib/request.c @@ -372,9 +372,11 @@ 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: FREEUP(data1.data.jobs); break; case Data_data_schedworkers: + case Data_data_workers: FREEUP(data1.data.workers); break; default: