lei.chen/myxce
1
0
Fork 1
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

save works

Browse source
This commit is contained in:
Rongsong Shen 2026-02-28 17:55:50 +08:00
parent b1faeb94b6
commit b748e30d20
35 changed files with 1247 additions and 2149 deletions
Download patch file Download diff file Expand all files Collapse all files

24
src/controller/apiserver.c
View file

@ -256,7 +256,7 @@ static int apiserver_job_update_fun(int type, int situation, void* data,
int rc;
xce_notify_data_t xdata
= { .event = NotifyData_data_updateJob,
.data = { .update_job = &(job->job) } };
.data = { .update_job = job->job } };
rc = notify(c, c->apiservers[i].ep, &xdata);
if (rc < 0) {
@ -351,11 +351,11 @@ static int apiserver_worker_change(int type, int situation, void* data,
{
int i;
TRACE("%s worker %s idx %d situation %d", __func__, w->host.name, w->idx,
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));
collect_resval_with_idx(&(w->host->resvals), &(w->host->nresvals_to_sched),
&(w->host->resvals_to_sched));
switch (situation) {
case HOOK_AFTER_EVENT:
@ -383,8 +383,8 @@ static int apiserver_worker_change(int type, int situation, void* data,
break;
}
free_resval_with_idx_list(&(w->host.resvals_to_sched),
w->host.nresvals_to_sched);
free_resval_with_idx_list(&(w->host->resvals_to_sched),
w->host->nresvals_to_sched);
return 0;
}
@ -402,12 +402,18 @@ static int apiserver_worker_join(int type, int situation, void* data,
return 0;
}
w = workermgr_lock(&(c->workermgr), name, OBJREF_READ);
rc = key_find(c->workermgr, (void *)name, (void **)&w);
if (rc < 0) {
return rc;
}
if (rc == 0) {
return -ENOENT;
}
rc = apiserver_worker_change(type, situation, data, c, w);
/* FIXME: need to mark the jobs in this worker */
obj_deref(w);
return rc;
}
@ -508,7 +514,7 @@ static int apiserver_queue_hook_fun(int type, int situation, int detail,
int rc;
xce_notify_data_t xdata
= { .event = NotifyData_data_schedUpdateQueue,
.data = { .sched_update_queue = &(q->queue) } };
.data = { .sched_update_queue = q->queue } };
rc = notify(c, c->apiservers[i].ep, &xdata);
if (rc < 0) {

167
src/controller/cluster.c
View file

@ -1,12 +1,14 @@
#include "config.h"
#include <errno.h>
#include <rtti.h>
#include <string.h>
#include <unistd.h>
#include "cluster.h"
#include "dataq.h"
#include "htab.h"
#include "kvmgr.h"
#include "mylock.h"
#include "request.h"
#include "vtaskmgr.h"
@ -15,6 +17,7 @@
#include "xce/types.h"
#include "resobj.h"
#include "utils.h"
#include "worker.h"
#include "xce/errnox.h"
@ -368,8 +371,7 @@ int cluster_build_resources(cluster_t* c)
for (i = 0; i < c->myconf->cluster_config->nres; i++) {
xce_resource_t* res;
res = create_obj_by_typename(&(c->objmgr), "xresource",
c->myconf->cluster_config->resources[i]);
res = makeResource(c->myconf->cluster_config->resources[i]);
if (res == NULL) {
return -errno;
}
@ -458,10 +460,82 @@ static int run_delay_tasks(delay_task_entry_fn_t entry, void* arg)
return entry(arg);
}
static void res_obj_killer(xce_resource_t* r)
{
if (r == NULL) {
return;
}
}
static void job_obj_killer(void* arg)
{
if (arg == NULL) {
return;
}
}
static void worker_obj_killer(void* arg)
{
if (arg == NULL) {
return;
}
}
static void queue_obj_killer(void* arg)
{
if (arg == NULL) {
return;
}
}
static void resval_obj_killer(void* arg)
{
if (arg == NULL) {
return;
}
}
static void config_obj_killer(void* arg)
{
if (arg == NULL) {
return;
}
}
int cluster_init(cluster_t* c)
{
int i, rc;
SSL_CTX* ssl_ctx = NULL;
int i, rc;
SSL_CTX* ssl_ctx = NULL;
kvmgr_params_t config_mgr_params
= { .key_type = KVMGR_KEY_STRING,
.dynamic = 1,
.nslots = 128,
.delta = 128,
.datakiller = (destroy_fn_t)config_obj_killer };
kvmgr_params_t resmgr_params
= { .key_type = KVMGR_KEY_STRING,
.dynamic = 1,
.nslots = c->myconf->cluster_config->resmgr->nhash_slots,
.delta = c->myconf->cluster_config->resmgr->delta,
.datakiller = (destroy_fn_t)res_obj_killer };
kvmgr_params_t jobmgr_params
= { .key_type = KVMGR_KEY_INT,
.dynamic = 1,
.nslots = c->myconf->cluster_config->jobmgr->nhash_slots,
.delta = c->myconf->cluster_config->jobmgr->delta,
.datakiller = (destroy_fn_t)job_obj_killer };
kvmgr_params_t workermgr_params
= { .key_type = KVMGR_KEY_STRING,
.dynamic = 1,
.nslots = c->myconf->cluster_config->workermgr->nhash_slots,
.delta = c->myconf->cluster_config->workermgr->delta,
.datakiller = (destroy_fn_t)worker_obj_killer };
kvmgr_params_t qmgr_params
= { .key_type = KVMGR_KEY_STRING,
.dynamic = 0,
.nslots = c->myconf->cluster_config->qmgr->nhash_slots,
.delta = c->myconf->cluster_config->qmgr->delta,
.datakiller = (destroy_fn_t)queue_obj_killer };
if (c == NULL) {
return -EINVAL;
@ -491,36 +565,21 @@ int cluster_init(cluster_t* c)
return -errno;
}
kvmgr_init(&(c->configmgr), string_hash_key, 128, 128,
(zarray_item_killer_t)config_obj_killer);
c->configmgr = makeInstance(&xkvmgr_type, &config_mgr_params);
objmgr_init(&(c->objmgr));
jobmgr_init(&(c->jobmgr), c->myconf->cluster_config->jobmgr->nhash_slots,
c->myconf->cluster_config->jobmgr->delta,
(zarray_item_killer_t)job_obj_killer);
resmgr_init(&(c->resmgr), c->myconf->cluster_config->resmgr->nhash_slots,
c->myconf->cluster_config->resmgr->delta,
(zarray_item_killer_t)res_obj_killer);
c->jobmgr = makeInstance(&xkvmgr_type, &jobmgr_params);
c->resmgr = makeInstance(&xkvmgr_type, &resmgr_params);
zarray_init(&(c->sharedinst), c->myconf->cluster_config->resmgr->inst_delta,
(zarray_item_killer_t)resval_obj_killer);
workermgr_init(&(c->workermgr),
c->myconf->cluster_config->workermgr->nhash_slots,
c->myconf->cluster_config->workermgr->delta,
(zarray_item_killer_t)worker_obj_killer);
qmgr_init(&(c->qmgr), c->myconf->cluster_config->qmgr->nhash_slots,
c->myconf->cluster_config->qmgr->delta,
(zarray_item_killer_t)queue_obj_killer);
c->workermgr = makeInstance(&xkvmgr_type, &workermgr_params);
c->qmgr = makeInstance(&xkvmgr_type, &qmgr_params);
iomgr_init(&(c->iomgr));
iomgr_set_ssl_ctx(&(c->iomgr), ssl_ctx);
xstring_table_init(c->myconf->cluster_config->xstrings->nhash_slots);
register_xstring(&(c->objmgr),
c->myconf->cluster_config->xstrings->nhash_slots,
c->myconf->cluster_config->xstrings->delta);
register_types(c);
zarray_init(&(c->endpoints), 16, endpoint_killer);
rc = endpoint_init(&(c->ep), AF_INET, SOCK_STREAM,
@ -557,10 +616,8 @@ static int make_heartbeat(cluster_t* c, endpoint_t* ep)
return xce_nop_req(ep);
}
static int check_worker_timeout_(void* arg1, void* arg2)
static int check_worker_timeout_(cluster_t* c, worker_t* w)
{
cluster_t* c = (cluster_t*)arg1;
worker_t* w = (worker_t*)arg2;
worker_t* w1 = NULL;
endpoint_t* ep;
time_t now = time(NULL);
@ -581,24 +638,19 @@ static int check_worker_timeout_(void* arg1, void* arg2)
}
/* grab lock on worker */
w1 = obj_ref(w, OBJREF_UPDATE);
if (w1 == NULL) {
return -EINVAL;
}
config = c->myconf->cluster_config;
timeout = config->heartbeat_interval * config->inactive_heartbeat_count;
if (now - ep->ts_last_in > timeout) {
TRACE("%s worker %s becomes UNAVAIL as heartbeat timeout", __func__,
w->host.name);
w->host->name);
mark_worker_unavail(c, w);
obj_deref(w1);
return 0;
} else {
if (IS_WORKER_UNAVAIL(w->host.status)) {
if (IS_WORKER_UNAVAIL(w->host->status)) {
mark_worker_back(c, w);
}
}
@ -607,34 +659,49 @@ static int check_worker_timeout_(void* arg1, void* arg2)
if (DATAQ_EMPTY(&(ep->outq))) {
int rc = 0;
TRACE("%s send heartbeat to %s", __func__, w->host.name);
TRACE("%s send heartbeat to %s", __func__, w->host->name);
rc = make_heartbeat(c, ep);
if (rc < 0) {
ERROR("%s make_heartbeat to %s failed", __func__, w->host.name);
ERROR("%s make_heartbeat to %s failed", __func__,
w->host->name);
obj_deref(w1);
return rc;
}
}
}
obj_deref(w1);
return 0;
}
static int check_workers_timeout(void* args)
{
xlist_t lst;
int rc;
cluster_t* c = (cluster_t*)args;
iterator_t it;
cluster_t* c = (cluster_t*)args;
int rc = 0;
xlist_init(&lst, 3);
xlist_store(&lst, check_worker_timeout_);
rc = workermgr_iter(&(c->workermgr), &lst, c);
rc = iter_make(&it, c->workermgr);
if (rc < 0) {
return rc;
}
xlist_fini(&lst);
for (;;) {
worker_t* w = NULL;
rc = iter_next(&it, (void**)&w);
if ((rc < 0) || (rc == RTTI_ITER_NO_DATA)) {
break;
}
check_worker_timeout_(c, w);
}
if (rc == RTTI_ITER_NO_DATA) {
rc = 0;
}
iter_destroy(&it);
return rc;
}
@ -698,11 +765,11 @@ void cluster_fini(cluster_t* c)
cbuf_fini(&(c->wbuf));
iomgr_fini(&(c->iomgr));
jobmgr_fini(&(c->jobmgr));
workermgr_fini(&(c->workermgr));
qmgr_fini(&(c->qmgr));
resmgr_fini(&(c->resmgr));
kvmgr_fini(&(c->configmgr));
destroy(c->jobmgr);
destroy(c->workermgr);
destroy(c->qmgr);
destroy(c->resmgr);
destroy(c->configmgr);
mylock_fini(&(c->sched.lock));
zarray_fini(&(c->endpoints));

496
src/controller/dataex.c
View file

@ -18,22 +18,6 @@ int check_controller(void* arg)
return 1;
}
struct intern_array_ {
int count;
int idx;
void* data;
check_fn_t checkfn;
};
int not_null_check(void* arg)
{
if (arg == NULL) {
return 0;
}
return 1;
}
static int collect_resval_with_idx_(zarray_t* array, int* nres,
xce_val_with_idx_t*** resvals)
{
@ -118,345 +102,309 @@ int collect_resval_with_idx(zarray_t* array, int* nres,
return rc;
}
static int count_(void* arg1, void* arg2)
{
struct intern_array_* arg = (struct intern_array_*)arg1;
if ((arg == NULL) || (arg->checkfn == NULL)) {
return -EINVAL;
}
if (!(arg->checkfn(arg2))) {
return 0;
}
arg->count++;
return 0;
}
static int add_confvar_(void* arg1, void* arg2)
{
struct intern_array_* arg = (struct intern_array_*)arg1;
xce_config_t** p = NULL;
if ((arg == NULL) || (arg->count < 0)) {
return -EINVAL;
}
if (arg->count == 0) {
return 0;
}
if (!(arg->checkfn(arg2))) {
return 0;
}
p = (xce_config_t**)arg->data;
if (p == NULL) {
p = (xce_config_t**)calloc(arg->count, sizeof(xce_config_t*));
if (p == NULL) {
return -ENOMEM;
}
arg->data = p;
}
p[arg->idx] = (xce_config_t*)obj_ref(arg2, OBJREF_READ);
arg->idx++;
return 1;
}
void collect_confvars(cluster_t* c, collect_check_fn_t fn, int* nconfig,
xce_config_t*** configs)
{
xlist_t lst;
struct intern_array_ args
= { .count = 0,
.idx = 0,
.data = NULL,
.checkfn = (fn == NULL) ? not_null_check : fn };
int rc;
iterator_t it;
int nc = 0, idx;
int rc;
*nconfig = 0;
*configs = NULL;
xlist_init(&lst, 3);
xlist_store(&lst, count_);
xlist_store(&lst, add_confvar_);
rc = kvmgr_iter(&(c->configmgr), NULL, &lst, &args);
xlist_fini(&lst);
if (rc >= 0) {
*nconfig = args.count;
*configs = args.data;
}
}
static int add_queue_(void* arg1, void* arg2)
{
struct intern_array_* arg = (struct intern_array_*)arg1;
xce_queue_t** p = NULL;
if (arg == NULL) {
return -EINVAL;
rc = iter_make(&it, c->configmgr);
if (rc < 0) {
return;
}
if (arg->count < 0) {
return -EINVAL;
}
nc = 0;
for (;;) {
xce_config_t* config = NULL;
if (arg->count == 0) {
return 0;
}
if (!(arg->checkfn(arg2))) {
return 0;
}
p = (xce_queue_t**)arg->data;
if (p == NULL) {
p = (xce_queue_t**)calloc(arg->count, sizeof(xce_queue_t*));
if (p == NULL) {
return -ENOMEM;
rc = iter_next(&it, (void**)&config);
if ((rc < 0) || (rc == RTTI_ITER_NO_DATA)) {
break;
}
arg->data = p;
if (config != NULL) {
nc++;
}
}
p[arg->idx] = (xce_queue_t*)obj_ref(arg2, OBJREF_READ);
arg->idx++;
if (rc < 0) {
iter_destroy(&it);
return;
}
return 1;
if (nc == 0) {
iter_destroy(&it);
return;
}
iter_reset(&it);
*configs = (xce_config_t**)calloc(nc, sizeof(xce_config_t*));
if ((*configs) == NULL) {
return;
}
idx = 0;
for (;;) {
xce_config_t* config = NULL;
rc = iter_next(&it, (void**)&config);
if ((rc < 0) || (rc == RTTI_ITER_NO_DATA)) {
break;
}
if (config != NULL) {
(*configs)[idx] = config;
idx++;
}
}
iter_destroy(&it);
}
void collect_queues(cluster_t* c, collect_check_fn_t fn, int* nqueues,
xce_queue_t*** queues)
{
xlist_t lst;
struct intern_array_ args
= { .count = 0,
.idx = 0,
.data = NULL,
.checkfn = (fn == NULL) ? not_null_check : fn };
int rc;
iterator_t it;
int count;
int rc;
int nc;
*nqueues = 0;
*queues = NULL;
xlist_init(&lst, 3);
xlist_store(&lst, count_);
xlist_store(&lst, add_queue_);
rc = qmgr_iter(&(c->qmgr), &lst, &args);
xlist_fini(&lst);
if (rc >= 0) {
*nqueues = args.count;
*queues = args.data;
}
}
static int add_resource_(void* arg1, void* arg2)
{
struct intern_array_* arg = (struct intern_array_*)arg1;
xce_resource_t** p = NULL;
if (arg == NULL) {
return -EINVAL;
rc = iter_make(&it, c->qmgr);
if (rc < 0) {
return;
}
if (arg->count < 0) {
return -EINVAL;
}
count = 0;
for (;;) {
xce_queue_t* q = NULL;
if (arg->count == 0) {
return 0;
}
if (!(arg->checkfn(arg2))) {
return 0;
}
p = (xce_resource_t**)arg->data;
if (p == NULL) {
p = (xce_resource_t**)calloc(arg->count, sizeof(xce_resource_t*));
if (p == NULL) {
return -ENOMEM;
rc = iter_next(&it, (void**)&q);
if ((rc < 0) || (rc == RTTI_ITER_NO_DATA)) {
break;
}
arg->data = p;
if (q != NULL) {
count++;
}
}
p[arg->idx] = (xce_resource_t*)obj_ref(arg2, OBJREF_READ);
if (rc < 0) {
return;
}
arg->idx++;
*queues = (xce_queue_t**)calloc(count, sizeof(xce_queue_t*));
if ((*queues) == NULL) {
iter_destroy(&it);
return;
}
return 1;
iter_reset(&it);
nc = 0;
for (;;) {
xce_queue_t* q = NULL;
rc = iter_next(&it, (void**)&q);
if ((rc < 0) || (rc == RTTI_ITER_NO_DATA)) {
break;
}
if (q != NULL) {
(*queues)[nc] = q;
nc++;
}
}
*nqueues = nc;
iter_destroy(&it);
}
void collect_resources(cluster_t* c, collect_check_fn_t fn, int* nresources,
xce_resource_t*** resources)
{
xlist_t lst;
struct intern_array_ args
= { .count = 0,
.idx = 0,
.data = NULL,
.checkfn = (fn == NULL) ? not_null_check : fn };
int rc;
iterator_t it;
int rc;
int i, nc;
*nresources = 0;
*resources = NULL;
xlist_init(&lst, 3);
xlist_store(&lst, count_);
xlist_store(&lst, add_resource_);
rc = resmgr_iter(&(c->resmgr), &lst, &args);
xlist_fini(&lst);
if (rc >= 0) {
*nresources = args.count;
*resources = args.data;
}
}
static int add_job_(void* arg1, void* arg2)
{
struct intern_array_* arg = (struct intern_array_*)arg1;
xce_job_t** p = NULL;
if (arg == NULL) {
return -EINVAL;
rc = iter_make(&it, c->resmgr);
if (rc < 0) {
return;
}
if (arg->count < 0) {
return -EINVAL;
}
nc = 0;
for (;;) {
xce_resource_t* r = NULL;
if (arg->count == 0) {
return 0;
}
if (!(arg->checkfn(arg2))) {
return 0;
}
p = (xce_job_t**)arg->data;
if (p == NULL) {
p = (xce_job_t**)calloc(arg->count, sizeof(xce_job_t*));
if (p == NULL) {
return -ENOMEM;
rc = iter_next(&it, (void**)&r);
if ((rc < 0) || (rc == RTTI_ITER_NO_DATA)) {
break;
}
arg->data = p;
if (r != NULL) {
nc++;
}
}
p[arg->idx] = (xce_job_t*)obj_ref(arg2, OBJREF_READ);
arg->idx++;
if ((rc < 0) || (nc == 0)) {
iter_destroy(&it);
return;
}
return 1;
*resources = calloc(nc, sizeof(xce_resource_t));
if ((*resources) == NULL) {
iter_destroy(&it);
return;
}
iter_reset(&it);
i = 0;
for (;;) {
xce_resource_t* r = NULL;
rc = iter_next(&it, (void**)&r);
if ((rc < 0) || (rc == RTTI_ITER_NO_DATA)) {
break;
}
if (r != NULL) {
(*resources)[i] = r;
i++;
}
}
*nresources = i;
iter_destroy(&it);
return;
}
void collect_jobs(cluster_t* c, collect_check_fn_t fn, int* njobs,
xce_job_t*** jobs)
{
xlist_t lst;
struct intern_array_ args
= { .count = 0, .idx = 0, .data = NULL, .checkfn = (check_fn_t)fn };
int rc = 0;
iterator_t it;
int nc, i;
int rc = 0;
*njobs = 0;
*jobs = NULL;
xlist_init(&lst, 3);
xlist_store(&lst, count_);
xlist_store(&lst, add_job_);
rc = jobmgr_iter(&(c->jobmgr), &lst, &args);
xlist_fini(&lst);
if (rc >= 0) {
*njobs = args.count;
*jobs = args.data;
rc = iter_make(&it, c->jobmgr);
if (rc < 0) {
return;
}
return;
}
nc = 0;
for (;;) {
xce_job_t* j = NULL;
static int add_worker_(void* arg1, void* arg2)
{
struct intern_array_* arg = (struct intern_array_*)arg1;
xce_worker_t** p = NULL;
xce_worker_t* w;
if (arg == NULL) {
return -EINVAL;
}
if (arg->count < 0) {
return -EINVAL;
}
if (arg->count == 0) {
return 0;
}
if (!(arg->checkfn(arg2))) {
return 0;
}
p = (xce_worker_t**)arg->data;
if (p == NULL) {
p = (xce_worker_t**)calloc(arg->count, sizeof(xce_worker_t*));
if (p == NULL) {
return -ENOMEM;
rc = iter_next(&it, (void**)&j);
if ((rc < 0) || (rc == RTTI_ITER_NO_DATA)) {
break;
}
arg->data = p;
if (j != NULL) {
nc++;
}
}
w = (xce_worker_t*)obj_ref(arg2, OBJREF_READ);
if ((rc < 0) || (nc == 0)) {
iter_destroy(&it);
return;
}
collect_resval_with_idx(&(w->resvals), &(w->nresvals_to_sched),
&(w->resvals_to_sched));
iter_reset(&it);
i = 0;
p[arg->idx] = w;
arg->idx++;
for (;;) {
xce_job_t* j = NULL;
return 1;
rc = iter_next(&it, (void**)&j);
if ((rc < 0) || (rc == RTTI_ITER_NO_DATA)) {
break;
}
if (j != NULL) {
(*jobs)[i] = j;
i++;
}
}
*njobs = i;
iter_destroy(&it);
return;
}
void collect_workers(cluster_t* c, collect_check_fn_t fn, int* nworkers,
xce_worker_t*** workers)
{
xlist_t lst;
struct intern_array_ args
= { .count = 0, .idx = 0, .data = NULL, .checkfn = (check_fn_t)fn };
int rc = 0;
iterator_t it;
int i, nc;
int rc = 0;
*nworkers = 0;
*workers = NULL;
xlist_init(&lst, 3);
xlist_store(&lst, count_);
xlist_store(&lst, add_worker_);
rc = workermgr_iter(&(c->workermgr), &lst, &args);
xlist_fini(&lst);
if (rc >= 0) {
*nworkers = args.count;
*workers = args.data;
rc = iter_make(&it, c->workermgr);
if (rc < 0) {
return;
}
nc = 0;
for (;;) {
xce_worker_t* w = NULL;
rc = iter_next(&it, (void**)&w);
if ((rc < 0) || (rc == RTTI_ITER_NO_DATA)) {
break;
}
if (w != NULL) {
nc++;
}
}
if ((rc < 0) || (nc == 0)) {
iter_destroy(&it);
return;
}
iter_reset(&it);
i = 0;
for (;;) {
xce_worker_t* w = NULL;
rc = iter_next(&it, (void**)&w);
if ((rc < 0) || (rc == RTTI_ITER_NO_DATA)) {
break;
}
if (w != NULL) {
(*workers)[i] = w;
i++;
}
}
*nworkers = i;
iter_destroy(&it);
return;
}

12
src/controller/include/cluster.h
View file

@ -34,12 +34,8 @@ extern int add_to_xlist(xlist_t*, void*);
extern int add_worker_to_xlist(xlist_t*, void*);
extern int add_queue_to_xlist(xlist_t*, void*);
extern int add_job_to_xlist(xlist_t*, void*);
extern void queue_obj_killer(queue_t*);
extern void res_obj_killer(xce_resource_t*);
extern void resval_obj_killer(xce_resval_t*);
extern void resource_killer(void*);
extern int add_resource(void*, char*, char*, int, xce_order_t, int, int);
extern void config_obj_killer(xce_config_t*);
extern int add_resource(void*, char*, char*, int, xce_order_t, int, int);
extern int update_resvals(cluster_t*, worker_t*, int, xce_resval_t*);
@ -69,8 +65,6 @@ extern int get_resval(cluster_t*, char*, char*, xce_val_t*);
extern job_t* jobmgr_lockjob(jobmgr_t*, uint64_t, int);
extern int32_t jobmgr_add_job(jobmgr_t*, job_t*);
extern void jobmgr_del_job(jobmgr_t*, job_t*);
extern void job_obj_killer(job_t*);
extern void worker_obj_killer(worker_t*);
extern int handle_push(struct capn*, cluster_t*, endpoint_t*, struct Request*);
extern int do_get_op(cluster_t*, endpoint_t*, xce_request_t*);
extern int do_update_op(cluster_t*, endpoint_t*, xce_request_t*);
@ -106,7 +100,6 @@ extern int on_job_resume(cluster_t*, uint64_t);
extern int on_job_postdone(cluster_t*, endpoint_t*, uint64_t);
extern int on_job_ckpt(cluster_t*, endpoint_t*, xce_ckptinfo_t*);
extern int notify(cluster_t*, endpoint_t*, xce_notify_data_t*);
extern int not_null_check(void*);
extern int check_worker_(xce_worker_t*);
extern int update_resval_list(cluster_t*, worker_t*, xce_resval_list_t*);
extern int do_jobcontrol_op(cluster_t*, endpoint_t*, xce_request_t*);
@ -117,9 +110,8 @@ extern int update_host_runusage(xce_job_runusage_t**, char*, int,
extern int update_pids(job_pidinfo_t**, int, int*);
extern int update_pgids(job_pidinfo_t**, int, int*);
extern int get_ep_role(endpoint_t*);
extern worker_t* get_ep_worker(cluster_t*, endpoint_t*, int);
extern worker_t* get_ep_worker(cluster_t*, endpoint_t*);
extern int not_null_check(void*);
extern int check_worker_(xce_worker_t*);
extern int update_resval_list(cluster_t*, worker_t*, xce_resval_list_t*);
extern int do_jobcontrol_op(cluster_t*, endpoint_t*, xce_request_t*);

285
src/controller/job.c
View file

@ -3,10 +3,12 @@
#include <time.h>
#include "cluster.h"
#include "jobj.h"
#include "preason.h"
#include "xce/cluster_def.h"
#include "xce/errnox.h"
#include "xce/types.h"
#include "xstring.h"
static int mark_job_start(cluster_t*, job_t*, job_pidinfo_t*);
static int mark_job_finish(cluster_t*, job_t*, xce_jobfinish_t*);
@ -19,20 +21,15 @@ uint64_t jobid = 100;
static job_t* create_job(cluster_t* c, xce_job_t* j)
{
job_t* job = NULL;
static int typeid = -1;
job_t* job = NULL;
if (typeid < 0) {
typeid = typeid_of_name(&(c->objmgr), "xjob");
TRACE("%s type %s id %d", __func__, "xjob", typeid);
}
if (typeid < 0) {
SET_ERROR(-typeid);
job = calloc(1, sizeof(*job));
if (job == NULL) {
return NULL;
}
job = create_obj_by_typeid(&(c->objmgr), typeid, j);
/* FIXME: */
job->job = makeJob(j);
return job;
}
@ -72,26 +69,26 @@ uint64_t job_submit(void* c, xce_job_submit_t* submit)
int load_jobs(void* c)
{
#if 0
job_t* job;
int i;
char cmdline[1024];
job_t* job;
int i;
char cmdline[1024];
for (i = 1; i < NJOBS; i++) {
sprintf(cmdline, "echo normal; sleep %d", i);
job = create_job(c, jobid++, "normal", "rshen", "/home/rshen", "/tmp",
"rshen", (char*)cmdline, NULL, 1, 1);
if (job == NULL) {
return -errno;
for (i = 1; i < NJOBS; i++) {
sprintf(cmdline, "echo normal; sleep %d", i);
job = create_job(c, jobid++, "normal", "rshen", "/home/rshen", "/tmp",
"rshen", (char*)cmdline, NULL, 1, 1);
if (job == NULL) {
return -errno;
}
}
}
for (i = 1; i < NJOBS; i++) {
job = create_job(c, jobid++, "xce", "rshen", "/home/rshen", "/tmp",
"rshen", "echo xce; sleep 30", NULL, 1, 1);
if (job == NULL) {
return -errno;
for (i = 1; i < NJOBS; i++) {
job = create_job(c, jobid++, "xce", "rshen", "/home/rshen", "/tmp",
"rshen", "echo xce; sleep 30", NULL, 1, 1);
if (job == NULL) {
return -errno;
}
}
}
#endif
return 0;
@ -157,8 +154,8 @@ static int mk_launchinfo(xce_launch_t* launch, xce_job_alloc_t* alloc)
static int dispatch_a_job_(cluster_t* c, endpoint_t* ep, xce_job_alloc_t* alloc)
{
int rc = 0;
job_t* j;
worker_t* w;
job_t* j = NULL;
worker_t* w = NULL;
xce_host_alloc_t** allocs;
int nc;
xce_job_exec_t einfo;
@ -166,7 +163,10 @@ static int dispatch_a_job_(cluster_t* c, endpoint_t* ep, xce_job_alloc_t* alloc)
TRACE("%s jobid %ld", __func__, alloc->jobid);
j = jobmgr_lockjob(&(c->jobmgr), alloc->jobid, OBJREF_UPDATE);
rc = key_find(c->jobmgr, (void*)alloc->jobid, (void**)&j);
if (rc < 0) {
return rc;
}
if (j == NULL) {
PERROR("jobmgr_lockjob");
@ -187,17 +187,20 @@ static int dispatch_a_job_(cluster_t* c, endpoint_t* ep, xce_job_alloc_t* alloc)
allocs = (xce_host_alloc_t**)alloc->allocs;
/* using head node of job */
w = workermgr_lock(&(c->workermgr), allocs[0]->host, OBJREF_READ);
rc = key_find(c->workermgr, (void*)allocs[0]->host, (void**)&w);
if (rc < 0) {
return rc;
}
if (w == NULL) {
PERROR("workermgr_lock");
free_launchinfo(&launchinfo);
obj_deref(j);
return -errno;
}
if ((w->ep == NULL) || (w->host.status != 0)) {
if ((w->ep == NULL) || (w->host->status != 0)) {
ERROR("worker %s is not in correct state, status %x ep %p",
w->host.name, w->host.status, w->ep);
w->host->name, w->host->status, w->ep);
free_launchinfo(&launchinfo);
obj_deref(w);
@ -212,40 +215,41 @@ static int dispatch_a_job_(cluster_t* c, endpoint_t* ep, xce_job_alloc_t* alloc)
char* q_pre_exec;
char* q_post_exec;
rc = get_queue_exec(c, j->job.submit->queue, &q_pre_exec, &q_post_exec);
rc = get_queue_exec(c, j->job->submit->queue, &q_pre_exec,
&q_post_exec);
if (rc < 0) {
obj_deref(j);
obj_deref(w);
return rc;
}
if (j->job.submit->start_info->q_pre_exec != NULL) {
obj_deref(j->job.submit->start_info->q_pre_exec);
if (j->job->submit->start_info->q_pre_exec != NULL) {
obj_deref(j->job->submit->start_info->q_pre_exec);
}
if (j->job.submit->start_info->q_post_exec != NULL) {
obj_deref(j->job.submit->start_info->q_post_exec);
if (j->job->submit->start_info->q_post_exec != NULL) {
obj_deref(j->job->submit->start_info->q_post_exec);
}
j->job.submit->start_info->q_pre_exec = q_pre_exec;
j->job.submit->start_info->q_post_exec = q_post_exec;
j->job->submit->start_info->q_pre_exec = q_pre_exec;
j->job->submit->start_info->q_post_exec = q_post_exec;
}
einfo.jobid = alloc->jobid;
einfo.start_info = j->job.submit->start_info;
einfo.start_info = j->job->submit->start_info;
einfo.launchinfo = &launchinfo;
rc = xce_startjob_req(w->ep, &einfo);
free_launchinfo(&launchinfo);
if (j->job.allocs != NULL) {
free_JobAlloc_ptr(&(j->job.allocs));
if (j->job->allocs != NULL) {
free_JobAlloc_ptr(&(j->job->allocs));
}
j->job.allocs = (xce_job_alloc_t*)calloc(1, sizeof(xce_job_alloc_t));
j->job.allocs->jobid = alloc->jobid;
j->job.allocs->nhosts = nc;
MOVE_PTR(j->job.allocs->allocs, allocs);
j->job->allocs = (xce_job_alloc_t*)calloc(1, sizeof(xce_job_alloc_t));
j->job->allocs->jobid = alloc->jobid;
j->job->allocs->nhosts = nc;
MOVE_PTR(j->job->allocs->allocs, allocs);
alloc->allocs = NULL;
DEREF_OBJ(w);
@ -263,40 +267,47 @@ static int dispatch_a_job_(cluster_t* c, endpoint_t* ep, xce_job_alloc_t* alloc)
int job_ctrl(cluster_t* c, xce_jobctrl_t* ctrinfo)
{
uint64_t jobid = ctrinfo->jobid;
job_t* j;
int rc = 0;
job_t* j = NULL;
int rc = 0;
TRACE("%s job_ctrl job %ld", __func__, jobid);
j = jobmgr_lockjob(&(c->jobmgr), jobid, OBJREF_UPDATE);
rc = key_find(c->jobmgr, (void*)jobid, (void**)&j);
if (rc < 0) {
return rc;
}
if (j == NULL) {
PERROR("jobmgr_lockjob");
return -ENOENT;
}
if (j->job.allocs != NULL) {
xce_host_alloc_t** allocs = j->job.allocs->allocs;
worker_t* w;
if (j->job->allocs != NULL) {
xce_host_alloc_t** allocs = j->job->allocs->allocs;
worker_t* w = NULL;
/* job control command will be sent to first node
* for parallel jobs
*/
w = workermgr_lock(&(c->workermgr), allocs[0]->host, OBJREF_READ);
rc = key_find(c->workermgr, (void*)allocs[0]->host, (void**)&w);
if (rc < 0) {
return rc;
}
if (w != NULL) {
/* FIXME: better method to check host status */
if ((w->ep != NULL) && (w->host.status == 0)) {
if ((w->ep != NULL) && (w->host->status == 0)) {
rc = xce_jobctrl_req(w->ep, ctrinfo);
if (rc < 0) {
ERROR("job ctrl %ld on host %s returns %d", jobid,
w->host.name, rc);
w->host->name, rc);
}
} else {
ERROR("worker %s is not in correct state, status %x ep %p",
w->host.name, w->host.status, w->ep);
w->host->name, w->host->status, w->ep);
rc = -errno;
}
obj_deref(w);
} else {
PERROR("workermgr_lock");
rc = -errno;
}
} else {
@ -311,9 +322,9 @@ int job_ctrl(cluster_t* c, xce_jobctrl_t* ctrinfo)
case JobControl_u_siginfo:
switch (ctrinfo->u.siginfo->signo) {
case SIGCONT:
if (j->job.state
if (j->job->state
& (JOB_STAT_PSUSP | JOB_STAT_USUSP | JOB_STAT_SSUSP)) {
int state = j->job.state;
int state = j->job->state;
/* resume suspend job? */
state &= ~(JOB_STAT_PSUSP | JOB_STAT_USUSP);
@ -322,9 +333,9 @@ int job_ctrl(cluster_t* c, xce_jobctrl_t* ctrinfo)
break;
case SIGSTOP:
case SIGTSTP:
if (!(j->job.state
if (!(j->job->state
& (JOB_STAT_PSUSP | JOB_STAT_USUSP | JOB_STAT_SSUSP))) {
int state = j->job.state;
int state = j->job->state;
state |= JOB_STAT_PSUSP;
mark_job_update(c, j, state);
@ -333,7 +344,7 @@ int job_ctrl(cluster_t* c, xce_jobctrl_t* ctrinfo)
case SIGKILL:
case SIGTERM:
default:
if (j->job.state
if (j->job->state
& (JOB_STAT_DONE | JOB_STAT_EXIT | JOB_STAT_ERR)) {
break;
}
@ -346,7 +357,8 @@ int job_ctrl(cluster_t* c, xce_jobctrl_t* ctrinfo)
}
break;
case JobControl_u_ckptoption:
if (j->job.state & (JOB_STAT_DONE | JOB_STAT_EXIT | JOB_STAT_ERR)) {
if (j->job->state
& (JOB_STAT_DONE | JOB_STAT_EXIT | JOB_STAT_ERR)) {
ERROR("job <%ld> has been finished, no checkpoint can be done.",
jobid);
} else {
@ -433,6 +445,7 @@ int handle_jobsubmit(cluster_t* c, endpoint_t* ep, xce_request_t* request)
xlist_t jlist;
job_t* j = NULL;
int i;
int rc;
submit = request->body.jobsubmsg;
@ -467,8 +480,8 @@ int handle_jobsubmit(cluster_t* c, endpoint_t* ep, xce_request_t* request)
for (i = 0; i < submit->array_size; i++) {
/* Run hooks to tell scheduler */
j = jobmgr_lockjob(&(c->jobmgr), jobid + i, OBJREF_READ);
if (j == NULL) {
rc = key_find(c->jobmgr, (void*)(jobid + i), (void**)&j);
if ((rc <= 0) || (j == NULL)) {
char idstring[128];
fmt_jobid(idstring, sizeof(idstring) - 1, jobid);
@ -496,21 +509,20 @@ static int mark_job_start(cluster_t* c, job_t* job, job_pidinfo_t* info)
xce_run_hooks(&(c->hooksys), XCE_CONTROLLER_HOOK_JOB, HOOK_BEFORE_EVENT,
HOOK_DETAIL_JOB_START, c, info, job);
/* update job information */
job->job.state = JOB_STAT_RUN;
job->job.start_time = time(NULL);
if (job->job.pidinfo == NULL) {
job->job.pidinfo = (job_pidinfo_t*)calloc(1, sizeof(job_pidinfo_t));
job->job->state = JOB_STAT_RUN;
job->job->start_time = time(NULL);
if (job->job->pidinfo == NULL) {
job->job->pidinfo = (job_pidinfo_t*)calloc(1, sizeof(job_pidinfo_t));
}
job->job.pidinfo->npids = copy_int_list((int**)&(job->job.pidinfo->pids),
(int*)info->pids, info->npids);
job->job.pidinfo->npgids = copy_int_list((int**)&(job->job.pidinfo->pgids),
(int*)info->pgids, info->npgids);
job->job->pidinfo->npids = copy_int_list((int**)&(job->job->pidinfo->pids),
(int*)info->pids, info->npids);
job->job->pidinfo->npgids = copy_int_list(
(int**)&(job->job->pidinfo->pgids), (int*)info->pgids, info->npgids);
if (info->exdata != NULL) {
job->job.pidinfo->exdata
= create_obj_by_typename(&(c->objmgr), "xstring", info->exdata);
job->job->pidinfo->exdata = makeString(info->exdata);
} else {
job->job.pidinfo->exdata = NULL;
job->job->pidinfo->exdata = NULL;
}
xce_run_hooks(&(c->hooksys), XCE_CONTROLLER_HOOK_JOB, HOOK_AFTER_EVENT,
@ -522,20 +534,19 @@ static int mark_job_start(cluster_t* c, job_t* job, job_pidinfo_t* info)
int on_job_start(cluster_t* c, job_pidinfo_t* info)
{
job_t* job = NULL;
int rc;
TRACE("%s jobid %ld", __func__, info->jobid);
job = jobmgr_lockjob(&(c->jobmgr), info->jobid, OBJREF_UPDATE);
if (job == NULL) {
rc = key_find(c->jobmgr, (void*)info->jobid, (void**)&job);
if ((rc <= 0) || (job == NULL)) {
ERROR("can not find job %ld", info->jobid);
return -errno;
}
mark_job_start(c, job, info);
/* unlock the job */
DEREF_OBJ(job);
return 0;
}
@ -545,23 +556,24 @@ static int mark_job_finish(cluster_t* c, job_t* job, xce_jobfinish_t* finish)
HOOK_DETAIL_JOB_EXIT, c, finish, job);
/* update job information */
if (finish->exit_code == 0) {
job->job.state = JOB_STAT_DONE;
job->job->state = JOB_STAT_DONE;
} else {
job->job.state = JOB_STAT_EXIT;
job->job->state = JOB_STAT_EXIT;
}
job->job.finish_time = time(NULL);
job->job->finish_time = time(NULL);
/* fill job run time/rusage information */
if (job->job.exit_info == NULL) {
job->job.exit_info = (job_exitinfo_t*)calloc(1, sizeof(job_exitinfo_t));
if (job->job->exit_info == NULL) {
job->job->exit_info
= (job_exitinfo_t*)calloc(1, sizeof(job_exitinfo_t));
}
if (job->job.exit_info != NULL) {
job->job.exit_info->exitcode = finish->exit_code;
if (job->job->exit_info != NULL) {
job->job->exit_info->exitcode = finish->exit_code;
}
job->job.run_time = job->job.finish_time - job->job.start_time;
job->job->run_time = job->job->finish_time - job->job->start_time;
xce_run_hooks(&(c->hooksys), XCE_CONTROLLER_HOOK_JOB, HOOK_AFTER_EVENT,
HOOK_DETAIL_JOB_EXIT, c, finish, job);
@ -576,7 +588,8 @@ int on_job_finish(cluster_t* c, xce_jobfinish_t* finish)
TRACE("%s job finish jobid %ld exit code %d", __func__, finish->jobid,
finish->exit_code);
job = jobmgr_lockjob(&(c->jobmgr), finish->jobid, OBJREF_UPDATE);
key_find(c->jobmgr, (void*)finish->jobid, (void**)&job);
if (job == NULL) {
ERROR("can not find job %ld", finish->jobid);
return -errno;
@ -584,9 +597,6 @@ int on_job_finish(cluster_t* c, xce_jobfinish_t* finish)
mark_job_finish(c, job, finish);
/* unlock the job */
DEREF_OBJ(job);
return 0;
}
@ -599,20 +609,20 @@ static int mark_job_fail(cluster_t* c, job_t* job, job_failinfo_t* failinfo)
switch (failinfo->stage) {
case JOB_EXEC_STAGE_Q_POST_EXEC:
case JOB_EXEC_STAGE_POST_EXEC:
if ((job->job.state & (JOB_STAT_EXIT | JOB_STAT_DONE)) == 0) {
if ((job->job->state & (JOB_STAT_EXIT | JOB_STAT_DONE)) == 0) {
WARN("job %ld got post exec fail notification but job's exit "
"status has not been set",
job->job.jobid);
job->job.state |= JOB_STAT_EXIT;
job->job->jobid);
job->job->state |= JOB_STAT_EXIT;
}
job->job.state |= JOB_STAT_PERR;
job->job->state |= JOB_STAT_PERR;
break;
default:
job->job.state = JOB_STAT_EXIT;
job->job->state = JOB_STAT_EXIT;
break;
}
job->job.finish_time = time(NULL);
job->job->finish_time = time(NULL);
xce_run_hooks(&(c->hooksys), XCE_CONTROLLER_HOOK_JOB, HOOK_AFTER_EVENT,
HOOK_DETAIL_JOB_FAIL, c, failinfo, job);
@ -625,7 +635,7 @@ static int mark_job_update(cluster_t* c, job_t* job, int state)
xce_run_hooks(&(c->hooksys), XCE_CONTROLLER_HOOK_JOB, HOOK_BEFORE_EVENT,
HOOK_DETAIL_JOB_UPDATE, c, job);
/* update job state */
job->job.state = state;
job->job->state = state;
xce_run_hooks(&(c->hooksys), XCE_CONTROLLER_HOOK_JOB, HOOK_AFTER_EVENT,
HOOK_DETAIL_JOB_UPDATE, c, job);
@ -640,7 +650,7 @@ int on_job_fail(cluster_t* c, job_failinfo_t* failinfo)
TRACE("%s job fail jobid %ld code %d", __func__, failinfo->jobid,
failinfo->code);
job = jobmgr_lockjob(&(c->jobmgr), failinfo->jobid, OBJREF_UPDATE);
key_find(c->jobmgr, (void*)failinfo->jobid, (void**)&job);
if (job == NULL) {
ERROR("can not find job %ld", failinfo->jobid);
return -errno;
@ -648,9 +658,6 @@ int on_job_fail(cluster_t* c, job_failinfo_t* failinfo)
mark_job_fail(c, job, failinfo);
/* unlock the job */
DEREF_OBJ(job);
return 0;
}
@ -661,20 +668,17 @@ int on_job_stop(cluster_t* c, uint64_t jobid)
TRACE("%s job stop jobid %ld", __func__, jobid);
job = jobmgr_lockjob(&(c->jobmgr), jobid, OBJREF_UPDATE);
key_find(c->jobmgr, (void*)jobid, (void**)&job);
if (job == NULL) {
ERROR("can not find job %ld", jobid);
return -errno;
}
state = job->job.state;
state = job->job->state;
state |= JOB_STAT_USUSP;
mark_job_update(c, job, state);
/* unlock the job */
DEREF_OBJ(job);
return 0;
}
@ -685,19 +689,16 @@ int on_job_resume(cluster_t* c, uint64_t jobid)
TRACE("%s job resume jobid %ld", __func__, jobid);
job = jobmgr_lockjob(&(c->jobmgr), jobid, OBJREF_UPDATE);
key_find(c->jobmgr, (void*)jobid, (void**)&job);
if (job == NULL) {
ERROR("can not find job %ld", jobid);
return -errno;
}
state = job->job.state;
state = job->job->state;
state &= ~JOB_STAT_USUSP;
mark_job_update(c, job, state);
/* unlock the job */
DEREF_OBJ(job);
return 0;
}
@ -708,28 +709,25 @@ int on_job_postdone(cluster_t* c, endpoint_t* ep, uint64_t jobid)
TRACE("%s job post done jobid %ld", __func__, jobid);
job = jobmgr_lockjob(&(c->jobmgr), jobid, OBJREF_UPDATE);
key_find(c->jobmgr, (void*)jobid, (void**)&job);
if (job == NULL) {
ERROR("can not find job %ld", jobid);
return -errno;
}
state = job->job.state;
state = job->job->state;
if ((state & (JOB_STAT_EXIT | JOB_STAT_DONE)) == 0) {
WARN("job %ld got post exec fail notification but job's exit status "
"has not been set",
job->job.jobid);
job->job->jobid);
state |= JOB_STAT_EXIT;
/* ensure that job finish time has been set */
job->job.finish_time = time(NULL);
job->job->finish_time = time(NULL);
}
state |= JOB_STAT_PDONE;
mark_job_update(c, job, state);
/* unlock the job */
DEREF_OBJ(job);
return 0;
}
@ -737,7 +735,7 @@ int on_job_ckpt(cluster_t* c, endpoint_t* ep, xce_ckptinfo_t* ckptinfo)
{
uint64_t jobid;
char* info = NULL;
job_t* j;
job_t* j = NULL;
if (ckptinfo == NULL) {
ERROR("%s ckptinfo should not be NULL", __func__);
@ -749,7 +747,8 @@ int on_job_ckpt(cluster_t* c, endpoint_t* ep, xce_ckptinfo_t* ckptinfo)
TRACE("%s job %ld ckptinfo %s", __func__, jobid, info);
j = jobmgr_lockjob(&(c->jobmgr), jobid, OBJREF_UPDATE);
key_find(c->jobmgr, (void*)jobid, (void**)&j);
if (j == NULL) {
ERROR("%s can not find job %ld", __func__, jobid);
return -errno;
@ -757,11 +756,12 @@ int on_job_ckpt(cluster_t* c, endpoint_t* ep, xce_ckptinfo_t* ckptinfo)
xce_run_hooks(&(c->hooksys), XCE_CONTROLLER_HOOK_JOB, HOOK_BEFORE_EVENT,
HOOK_DETAIL_JOB_CKPT, c, j);
if (j->job.ckptinfo != NULL) {
obj_deref(j->job.ckptinfo);
if (j->job->ckptinfo != NULL) {
obj_deref(j->job->ckptinfo);
}
j->job.ckptinfo = create_obj_by_typename(&(c->objmgr), "xstring", info);
j->job->ckptinfo = makeString(info);
xce_run_hooks(&(c->hooksys), XCE_CONTROLLER_HOOK_JOB, HOOK_AFTER_EVENT,
HOOK_DETAIL_JOB_CKPT, c, j);
@ -788,28 +788,28 @@ int do_jobcontrol_op(cluster_t* c, endpoint_t* ep, xce_request_t* request)
int update_job_rusages(cluster_t* c, endpoint_t* ep, xce_request_t* request)
{
xce_job_rusages_t* rusages = request->body.updatemsg->data.jobrusages;
job_t* j;
worker_t* w;
char* host = NULL;
job_t* j = NULL;
worker_t* w = NULL;
char* host = NULL;
TRACE("%s trying to update job(%ld) rusage", __func__, rusages->jobid);
xce_run_hooks(&(c->hooksys), XCE_CONTROLLER_HOOK_JOB, HOOK_BEFORE_EVENT,
HOOK_DETAIL_JOB_RUSAGE_UPDATE, c, rusages);
j = jobmgr_lockjob(&(c->jobmgr), rusages->jobid, OBJREF_UPDATE);
key_find(c->jobmgr, (void*)rusages->jobid, (void**)&j);
if (j == NULL) {
PERROR("jobmgr_lockjob");
return -errno;
}
w = get_ep_worker(c, ep, OBJREF_READ);
w = get_ep_worker(c, ep);
if (w == NULL) {
obj_deref(j);
return errno;
}
host = create_obj_by_typename(&(c->objmgr), "xstring", w->host.name);
host = makeString(w->host->name);
obj_deref(w);
if (host == NULL) {
@ -821,12 +821,12 @@ int update_job_rusages(cluster_t* c, endpoint_t* ep, xce_request_t* request)
xce_run_hooks(&(c->hooksys), XCE_CONTROLLER_HOOK_JOB, HOOK_BEFORE_EVENT,
HOOK_DETAIL_JOB_UPDATE, c, j);
update_host_runusage(&(j->job.run_usage), host, rusages->count,
update_host_runusage(&(j->job->run_usage), host, rusages->count,
rusages->usages);
update_pids(&(j->job.pidinfo), rusages->npids, rusages->pids);
update_pids(&(j->job->pidinfo), rusages->npids, rusages->pids);
update_pgids(&(j->job.pidinfo), rusages->npgids, rusages->pgids);
update_pgids(&(j->job->pidinfo), rusages->npgids, rusages->pgids);
/* FIXME
* update other fields of job rusage later
@ -843,12 +843,12 @@ int update_job_rusages(cluster_t* c, endpoint_t* ep, xce_request_t* request)
static int set_preason(cluster_t* c, xce_pendreason_t* reasons)
{
job_t* j;
xce_job_t* job;
job_t* j = NULL;
xce_job_t* job = NULL;
TRACE("%s jobid %ld", __func__, reasons->jobid);
j = jobmgr_lockjob(&(c->jobmgr), reasons->jobid, OBJREF_UPDATE);
key_find(c->jobmgr, (void*)reasons->jobid, (void**)&j);
if (j == NULL) {
PERROR("jobmgr_lockjob");
return -ENOENT;
@ -856,7 +856,7 @@ static int set_preason(cluster_t* c, xce_pendreason_t* reasons)
xce_run_hooks(&(c->hooksys), XCE_CONTROLLER_HOOK_JOB, HOOK_BEFORE_EVENT,
HOOK_DETAIL_SETPREASON, c, j);
job = &(j->job);
job = j->job;
if (job->pend_reasons != NULL) {
erase_job_preason(job->pend_reasons);
FREEUP(job->pend_reasons);
@ -865,7 +865,6 @@ static int set_preason(cluster_t* c, xce_pendreason_t* reasons)
MOVE_PTR(job->pend_reasons, reasons->reasons);
xce_run_hooks(&(c->hooksys), XCE_CONTROLLER_HOOK_JOB, HOOK_AFTER_EVENT,
HOOK_DETAIL_SETPREASON, c, j);
obj_deref(j);
return 0;
}
@ -915,7 +914,7 @@ int do_clr_preason(cluster_t* c, endpoint_t* ep, xce_request_t* request)
job_t* j;
xce_job_t* job;
j = jobmgr_lockjob(&(c->jobmgr), jobid, OBJREF_UPDATE);
key_find(c->jobmgr, (void*)jobid, (void**)&j);
if (j == NULL) {
PERROR("jobmgr_lockjob");
continue;
@ -924,7 +923,7 @@ int do_clr_preason(cluster_t* c, endpoint_t* ep, xce_request_t* request)
xce_run_hooks(&(c->hooksys), XCE_CONTROLLER_HOOK_JOB, HOOK_BEFORE_EVENT,
HOOK_DETAIL_CLRPREASON, c, j);
job = &(j->job);
job = j->job;
if (job->pend_reasons != NULL) {
erase_job_preason(job->pend_reasons);
FREEUP(job->pend_reasons);

1
src/controller/meson.build
View file

@ -14,7 +14,6 @@ controller_src = [
'sched.c',
'task.c',
'worker.c',
'xceobjs.c',
'apiserver.c',
'myconf.c',
]

11
src/controller/misc.c
View file

@ -47,10 +47,11 @@ int get_ep_role(endpoint_t* ep)
return priv->role;
}
worker_t* get_ep_worker(cluster_t* c, endpoint_t* ep, int op)
worker_t* get_ep_worker(cluster_t* c, endpoint_t* ep)
{
int role, idx;
worker_t* w = NULL;
worker_t* w = NULL;
int rc = 0;
role = get_ep_role(ep);
if (role != HostRole_worker) {
@ -64,7 +65,11 @@ worker_t* get_ep_worker(cluster_t* c, endpoint_t* ep, int op)
return NULL;
}
w = workermgr_lock_byid(&(c->workermgr), idx, op);
rc = key_find(c->workermgr, (void*)idx, (void**)&w);
if (rc < 0) {
return NULL;
}
if (w == NULL) {
ERROR("can not find worker with id (%d)", idx);
}

3
src/controller/myconf.c
View file

@ -6,6 +6,7 @@
#include "controller.capnp.h"
#include "myconf.h"
#include "configobj.h"
#include "objmgr.h"
#include "utils.h"
#include "xce/cluster_def.h"
@ -17,7 +18,7 @@ int add_config(void* c, char* name, char* value)
xce_config_t cvar = { .name = name, .value = value };
xce_config_t* cv = NULL;
cv = create_obj_by_typename(&(cls->objmgr), "xconfig", &cvar);
cv = makeConfig(&cvar);
if (cv == NULL) {
return -errno;
}

8
src/controller/opget.c
View file

@ -148,16 +148,16 @@ static int do_reply_joblist(cluster_t* c, endpoint_t* ep,
init_reply_by_req(&reply, request);
if (match->type == MatchCondition_condition_matchjid) {
job_t* j;
job_t* j = NULL;
jobdata.n_total = 0;
jobdata.jobs = NULL;
j = jobmgr_lockjob(&(c->jobmgr), match->condition.jobid, OBJREF_READ);
if (j != NULL) {
rc = key_find(c->jobmgr, (void*)match->condition.jobid, (void**)&j);
if (rc > 0) {
jobdata.jobs = (xce_job_t**)calloc(1, sizeof(xce_job_t*));
if (jobdata.jobs != NULL) {
jobdata.jobs[0] = &(j->job);
jobdata.jobs[0] = j->job;
jobdata.n_total = 1;
}
}

4
src/controller/opupd.c
View file

@ -23,9 +23,9 @@ static int do_update_resvals(cluster_t* c, endpoint_t* ep,
int idx = priv->u.idx;
if ((priv->role == HostRole_worker) && (idx >= 0)) {
worker_t* w;
worker_t* w = NULL;
w = workermgr_lock_byid(&(c->workermgr), idx, OBJREF_UPDATE);
rc = vget(c->workermgr, idx, (void**)&w);
if (w == NULL) {
rc = -E_XCE_BAD_WORKER;
} else {

51
src/controller/queue.c
View file

@ -3,8 +3,10 @@
#include <string.h>
#include "cluster.h"
#include "qobj.h"
#include "xce/cluster_def.h"
#include "xce/types.h"
#include "xstring.h"
void queue_fini(queue_t* q)
{
@ -15,12 +17,14 @@ void queue_fini(queue_t* q)
queue_t* create_queue(cluster_t* c, xce_queue_t* p)
{
queue_t q1;
queue_t* q;
memcpy(&(q1.queue), p, sizeof(q1.queue));
q = calloc(1, sizeof(*q));
if (q == NULL) {
return NULL;
}
q = create_obj_by_typename(&(c->objmgr), "xqueue", &q1);
q->queue = makeQueue(p);
return q;
}
@ -56,7 +60,11 @@ int queue_ctrl(cluster_t* c, xce_queuectrl_t* ctrl)
return -EINVAL;
}
TRACE("%s queue_ctrl queue %s", __func__, ctrl->queue);
q = qmgr_lockq(&(c->qmgr), ctrl->queue, OBJREF_UPDATE);
rc = key_find(c->qmgr, (void*)ctrl->queue, (void**)&q);
if (rc < 0) {
return rc;
}
if (q == NULL) {
PERROR("qmgr_lockq");
return -ENOENT;
@ -64,28 +72,28 @@ int queue_ctrl(cluster_t* c, xce_queuectrl_t* ctrl)
switch (ctrl->action) {
case QUEUE_OPEN:
if (!(q->queue.status & QUEUE_STAT_CLOSED)) {
if (!(q->queue->status & QUEUE_STAT_CLOSED)) {
goto clean;
}
q->queue.status &= ~QUEUE_STAT_CLOSED;
q->queue->status &= ~QUEUE_STAT_CLOSED;
break;
case QUEUE_CLOSE:
if (q->queue.status & QUEUE_STAT_CLOSED) {
if (q->queue->status & QUEUE_STAT_CLOSED) {
goto clean;
}
q->queue.status |= QUEUE_STAT_CLOSED;
q->queue->status |= QUEUE_STAT_CLOSED;
break;
case QUEUE_INACTIVE:
if (q->queue.status & QUEUE_STAT_INACT_ADMIN) {
if (q->queue->status & QUEUE_STAT_INACT_ADMIN) {
goto clean;
}
q->queue.status |= QUEUE_STAT_INACT_ADMIN;
q->queue->status |= QUEUE_STAT_INACT_ADMIN;
break;
case QUEUE_ACTIVE:
if (!(q->queue.status & QUEUE_STAT_INACT_ADMIN)) {
if (!(q->queue->status & QUEUE_STAT_INACT_ADMIN)) {
goto clean;
}
q->queue.status &= ~QUEUE_STAT_INACT_ADMIN;
q->queue->status &= ~QUEUE_STAT_INACT_ADMIN;
break;
default:
rc = -1;
@ -117,7 +125,8 @@ int do_queuecontrol_op(cluster_t* c, endpoint_t* ep, xce_request_t* request)
int get_queue_exec(cluster_t* c, char* queue, char** pre_exec, char** post_exec)
{
queue_t* q = NULL;
queue_t* q = NULL;
int rc = 0;
TRACE("entering %s", __func__);
@ -128,22 +137,24 @@ int get_queue_exec(cluster_t* c, char* queue, char** pre_exec, char** post_exec)
TRACE("%s queue %s", __func__, queue);
q = qmgr_lockq(&(c->qmgr), queue, OBJREF_READ);
rc = key_find(c->qmgr, (void*)queue, (void**)&q);
if (rc < 0) {
return rc;
}
if (q == NULL) {
PERROR("qmgr_lockq");
return -ENOENT;
}
if (q->queue.pre_exec != NULL) {
*pre_exec = create_obj_by_typename(&(c->objmgr), "xstring",
q->queue.pre_exec);
if (q->queue->pre_exec != NULL) {
*pre_exec = makeString(q->queue->pre_exec);
} else {
*pre_exec = NULL;
}
if (q->queue.post_exec != NULL) {
*post_exec = create_obj_by_typename(&(c->objmgr), "xstring",
q->queue.post_exec);
if (q->queue->post_exec != NULL) {
*post_exec = makeString(q->queue->post_exec);
} else {
*post_exec = NULL;
}

87
src/controller/sched.c
View file

@ -114,8 +114,8 @@ static int xcesched_push_configs(cluster_t* c)
configdata.data.conf4sched->n_hgroups = c->myconf->cluster_config->nhgroups;
configdata.data.conf4sched->hgroups = c->myconf->cluster_config->hgroups;
#if 0 /* FIXME: we use builtin limit currently */
configdata.data.conf4sched->nlimits = c->myconf->cluster_config->nlimits;
configdata.data.conf4sched->limits = c->myconf->cluster_config->limits;
configdata.data.conf4sched->nlimits = c->myconf->cluster_config->nlimits;
configdata.data.conf4sched->limits = c->myconf->cluster_config->limits;
#else
configdata.data.conf4sched->nlimits = nlimits;
configdata.data.conf4sched->limits = &(limits[0]);
@ -204,7 +204,7 @@ static int xcesched_push_workers(cluster_t* c)
100);
/* free up collected resource value which used
passed to scheduler
passed to scheduler
*/
for (i = 0; i < workers.n_total; i++) {
xce_worker_t* w = workers.workers[i];
@ -248,12 +248,12 @@ static int sched_job_start_fun(int type, int situation, void* data, va_list ap)
job_pidinfo_t* info = va_arg(ap, job_pidinfo_t*);
int rc = 0;
/* clang-format off */
xce_notify_data_t xdata = {
.event = NotifyData_data_schedStartJob,
.data = {
.sched_start_job = info->jobid
}
};
xce_notify_data_t xdata = {
.event = NotifyData_data_schedStartJob,
.data = {
.sched_start_job = info->jobid
}
};
/* clang-format on */
mylock_lock(&(c->sched.lock));
@ -278,12 +278,12 @@ static int sched_job_exit_fun(int type, int situation, void* data, va_list ap)
xce_jobfinish_t* info = va_arg(ap, xce_jobfinish_t*);
int rc = 0;
/* clang-format off */
xce_notify_data_t xdata = {
.event = NotifyData_data_schedExitJob,
.data = {
.sched_exit_job = info->jobid
}
};
xce_notify_data_t xdata = {
.event = NotifyData_data_schedExitJob,
.data = {
.sched_exit_job = info->jobid
}
};
/* clang-format on */
mylock_lock(&(c->sched.lock));
@ -308,12 +308,12 @@ static int sched_job_fail_fun(int type, int situation, void* data, va_list ap)
job_failinfo_t* info = va_arg(ap, job_failinfo_t*);
int rc = 0;
/* clang-format off */
xce_notify_data_t xdata = {
.event = NotifyData_data_schedFailJob,
.data = {
.sched_fail_job = info->jobid
}
};
xce_notify_data_t xdata = {
.event = NotifyData_data_schedFailJob,
.data = {
.sched_fail_job = info->jobid
}
};
/* clang-format on */
mylock_lock(&(c->sched.lock));
@ -338,12 +338,12 @@ static int sched_job_update_fun(int type, int situation, void* data, va_list ap)
job_t* job = va_arg(ap, job_t*);
int rc = 0;
/* clang-format off */
xce_notify_data_t xdata = {
.event = NotifyData_data_schedUpdateJob,
.data = {
.sched_update_job = &(job->job)
}
};
xce_notify_data_t xdata = {
.event = NotifyData_data_schedUpdateJob,
.data = {
.sched_update_job = job->job
}
};
/* clang-format on */
mylock_lock(&(c->sched.lock));
@ -369,12 +369,12 @@ static int sched_job_rusage_update_fun(int type, int situation, void* data,
xce_job_rusages_t* rusages = va_arg(ap, xce_job_rusages_t*);
int rc = 0;
/* clang-format off */
xce_notify_data_t xdata = {
.event = NotifyData_data_schedJobRusages,
.data = {
.sched_job_rusages = rusages
}
};
xce_notify_data_t xdata = {
.event = NotifyData_data_schedJobRusages,
.data = {
.sched_job_rusages = rusages
}
};
/* clang-format on */
mylock_lock(&(c->sched.lock));
@ -419,11 +419,11 @@ static int sched_worker_change(int type, int situation, void* data,
int rc = 0;
xlist_t wlist;
TRACE("%s worker %s idx %d situation %d", __func__, w->host.name, w->idx,
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));
collect_resval_with_idx(&(w->host->resvals), &(w->host->nresvals_to_sched),
&(w->host->resvals_to_sched));
mylock_lock(&(c->sched.lock));
if (c->sched.ep != NULL) {
@ -440,8 +440,8 @@ static int sched_worker_change(int type, int situation, void* data,
}
mylock_unlock(&(c->sched.lock));
free_resval_with_idx_list(&(w->host.resvals_to_sched),
w->host.nresvals_to_sched);
free_resval_with_idx_list(&(w->host->resvals_to_sched),
w->host->nresvals_to_sched);
return rc;
}
@ -458,7 +458,14 @@ static int sched_worker_join(int type, int situation, void* data, va_list ap)
return 0;
}
w = workermgr_lock(&(c->workermgr), name, OBJREF_READ);
rc = key_find(c->workermgr, (void *)name, (void **)&w);
if (rc < 0) {
return rc;
}
if (rc == 0) {
return -ENOENT;
}
rc = sched_worker_change(type, situation, data, c, w);
@ -551,7 +558,7 @@ static int sched_queue_update(int type,
xce_notify_data_t xdata = {
.event = NotifyData_data_schedUpdateQueue,
.data = {
.sched_update_queue = &(q->queue)
.sched_update_queue = q->queue
}
};
/* clang-format on */

25
src/controller/task.c
View file

@ -17,12 +17,16 @@ static int run_tasks_on_host(cluster_t* c, endpoint_t* ep, uint64_t jobid,
.clientidx = ep->idx,
.taskhosts = (xce_taskhost_t*[]) { info } };
w = workermgr_lock(&(c->workermgr), info->host, OBJREF_READ);
rc = key_find(c->workermgr, (void*)info->host, (void**)&w);
if (rc < 0) {
return rc;
}
if (w == NULL) {
return -ENOENT;
}
if ((w->ep == NULL) || (IS_WORKER_UNAVAIL(w->host.status))) {
if ((w->ep == NULL) || (IS_WORKER_UNAVAIL(w->host->status))) {
obj_deref(w);
return -EINVAL;
}
@ -40,19 +44,24 @@ static int check_task_hosts(cluster_t* c, xce_taskrun_t* runmsg)
for (i = 0; i < runmsg->nhosts; i++) {
worker_t* w;
int rc = 0;
w = workermgr_lock(&(c->workermgr), runmsg->taskhosts[i]->host,
OBJREF_READ);
rc = key_find(c->workermgr, (void*)runmsg->taskhosts[i]->host,
(void**)&w);
if (rc < 0) {
return rc;
}
if (rc == 0) {
return -ENOENT;
}
if (w == NULL) {
return -ENOENT;
}
if ((w->ep == NULL) || IS_WORKER_UNAVAIL(w->host.status)) {
obj_deref(w);
if ((w->ep == NULL) || IS_WORKER_UNAVAIL(w->host->status)) {
return -ENOENT;
}
obj_deref(w);
}
return 0;

145
src/controller/worker.c
View file

@ -2,9 +2,11 @@
#include <stdlib.h>
#include "cluster.h"
#include "wobj.h"
#include "worker.h"
#include "xce/cluster_def.h"
#include "xce/errnox.h"
#include "xstring.h"
static int hostmatched(const char* pattern, const char* name)
{
@ -68,7 +70,7 @@ static int update_val_res(cluster_t* c, worker_t* w, xce_resource_t* r,
array = &(c->sharedinst);
} else {
/* update host based resource */
array = &(w->host.resvals);
array = &(w->host->resvals);
}
if (idx < array->caps) {
@ -95,11 +97,14 @@ static int update_val_res(cluster_t* c, worker_t* w, xce_resource_t* r,
static int update_resval(cluster_t* c, worker_t* w, xce_resval_t* val)
{
xce_resource_t* r;
xce_resource_t* r = NULL;
xce_val_t* v = NULL;
int rc;
r = resmgr_lockres(&(c->resmgr), val->name, OBJREF_READ);
rc = key_find(c->resmgr, (void*)val->name, (void**)&r);
if (rc < 0) {
return rc;
}
if (r == NULL) {
ERROR("resource `%s' can not be found", val->name);
return -errno;
@ -114,8 +119,6 @@ static int update_resval(cluster_t* c, worker_t* w, xce_resval_t* val)
rc = update_val_res(c, w, r, v);
obj_deref(r);
return rc;
}
@ -151,14 +154,18 @@ int update_resval_list(cluster_t* c, worker_t* w, xce_resval_list_t* l)
static int mark_host_res(cluster_t* c, worker_t* w, xce_config_t* config)
{
xce_resource_t* r;
xce_resource_t* r = NULL;
char* n = config->name;
int rc;
/* clang-format-off */
xce_val_t val;
/* clang-format-on */
r = resmgr_lockres(&(c->resmgr), n, OBJREF_READ);
rc = key_find(c->resmgr, (void*)n, (void**)&r);
if (rc < 0) {
return rc;
}
if (r == NULL) {
ERROR("resource `%s' can not be found", n);
return -ENOENT;
@ -199,7 +206,6 @@ static int mark_host_res(cluster_t* c, worker_t* w, xce_config_t* config)
}
rc = update_val_res(c, w, r, &val);
obj_deref(r);
return rc;
}
@ -207,16 +213,17 @@ static int add_hostattrs_resvals(cluster_t* c, worker_t* w)
{
#define N_HOST_ATTRS (8)
xce_val_t values[N_HOST_ATTRS] = {
{ .which = Value_u_svalue, .u = { .svalue = w->host.attrs->hosttype } },
{ .which = Value_u_svalue,
.u = { .svalue = w->host.attrs->hostmodel } },
.u = { .svalue = w->host->attrs->hosttype } },
{ .which = Value_u_svalue,
.u = { .svalue = w->host->attrs->hostmodel } },
{ .which = Value_u_fvalue,
.u = { .fvalue = w->host.attrs->cpufactor } },
{ .which = Value_u_fvalue, .u = { .fvalue = w->host.attrs->maxcpus } },
{ .which = Value_u_fvalue, .u = { .fvalue = w->host.attrs->maxmem } },
{ .which = Value_u_fvalue, .u = { .fvalue = w->host.attrs->maxswap } },
{ .which = Value_u_fvalue, .u = { .fvalue = w->host.attrs->maxtmp } },
{ .which = Value_u_fvalue, .u = { .fvalue = w->host.attrs->ndisks } }
.u = { .fvalue = w->host->attrs->cpufactor } },
{ .which = Value_u_fvalue, .u = { .fvalue = w->host->attrs->maxcpus } },
{ .which = Value_u_fvalue, .u = { .fvalue = w->host->attrs->maxmem } },
{ .which = Value_u_fvalue, .u = { .fvalue = w->host->attrs->maxswap } },
{ .which = Value_u_fvalue, .u = { .fvalue = w->host->attrs->maxtmp } },
{ .which = Value_u_fvalue, .u = { .fvalue = w->host->attrs->ndisks } }
};
xce_resval_t resvals[N_HOST_ATTRS] = {
{ .name = "type", .val = &values[0] },
@ -259,7 +266,7 @@ int add_worker(void* cls, endpoint_t* ep, const char* host, hostattrs_t* attrs)
cluster_t* c = (cluster_t*)cls;
/* clang-format off */
worker_t worker = {
.host = {
.host = &(xce_worker_t){
.name = (char*)host,
}
};
@ -273,35 +280,38 @@ int add_worker(void* cls, endpoint_t* ep, const char* host, hostattrs_t* attrs)
return -EINVAL;
}
w = calloc(1, sizeof(*w));
if (w == NULL) {
return -ENOMEM;
}
hconfig = find_matched_hostconfig(c, host);
if (hconfig == NULL) {
return -errno;
}
worker.host.attrs = attrs;
worker.host->attrs = attrs;
w = create_obj_by_typename(&(c->objmgr), "xworker", &worker);
if (w == NULL) {
w->host = makeWorker(worker.host);
if (w->host == NULL) {
return -errno;
}
/* we should hold the update lock */
/* If admin makes the host in locked or closed, we need keep this */
w->host.status &= ~XCE_WORKER_UNAVAIL; /* status OK */
w->host->status &= ~XCE_WORKER_UNAVAIL; /* status OK */
w->ep = ep;
/* override host type? */
if (strcmp(hconfig->hosttype, "!") != 0) {
obj_deref(w->host.attrs->hosttype);
w->host.attrs->hosttype = create_obj_by_typename(
&(c->objmgr), "xstring", hconfig->hosttype);
dereference(w->host->attrs->hosttype);
w->host->attrs->hosttype = makeString(hconfig->hosttype);
}
/* override host model */
if (strcmp(hconfig->hostmodel, "!") != 0) {
obj_deref(w->host.attrs->hostmodel);
w->host.attrs->hostmodel = create_obj_by_typename(
&(c->objmgr), "xstring", hconfig->hostmodel);
dereference(w->host->attrs->hostmodel);
w->host->attrs->hostmodel = makeString(hconfig->hostmodel);
}
add_hostattrs_resvals(c, w);
idx = w->idx;
@ -309,31 +319,31 @@ int add_worker(void* cls, endpoint_t* ep, const char* host, hostattrs_t* attrs)
/* using auto detected or pre configured MXJ? */
nslots = hconfig->mxj;
if (nslots < 0) {
nslots = w->host.attrs->maxcpus;
nslots = w->host->attrs->maxcpus;
}
set_host_slots(c, w, nslots);
if (w->host.dispatch_window_config != NULL) {
free_timewindow_config(w->host.dispatch_window_config);
w->host.dispatch_window_config = NULL;
if (w->host->dispatch_window_config != NULL) {
free_timewindow_config(w->host->dispatch_window_config);
w->host->dispatch_window_config = NULL;
}
if (hconfig->dispatchwindow != NULL) {
w->host.dispatch_window_config
w->host->dispatch_window_config
= dup_timewindow_config(hconfig->dispatchwindow);
}
if (w->host.thresholds != NULL) {
free_resource_thresholds(w->host.thresholds, w->host.nthresholds);
w->host.nthresholds = 0;
w->host.thresholds = NULL;
if (w->host->thresholds != NULL) {
free_resource_thresholds(w->host->thresholds, w->host->nthresholds);
w->host->nthresholds = 0;
w->host->thresholds = NULL;
}
if (hconfig->thresholds != NULL) {
w->host.thresholds = dup_resource_thresholds(hconfig->thresholds,
hconfig->nthresholds);
w->host.nthresholds = hconfig->nthresholds;
w->host->thresholds = dup_resource_thresholds(hconfig->thresholds,
hconfig->nthresholds);
w->host->nthresholds = hconfig->nthresholds;
}
/* add bool resources to host */
@ -354,20 +364,24 @@ int restore_worker(void* cls, xce_worker_t host)
/* do work in add_worker() doing for normal worker join */
/* clang-format off */
worker_t worker = {
.host = {
.host = &(xce_worker_t){
.name = (char*)host.name,
.status = host.status,
},
};
/* clang-format on */
w = create_obj_by_typename(&(c->objmgr), "xworker", &worker);
w = calloc(1, sizeof(*w));
if (w == NULL) {
return -ENOMEM;
}
w->host = makeWorker(worker.host);
if (w->host == NULL) {
return -errno;
}
add_hostattrs_resvals(c, w);
idx = w->idx;
w->host.status |= XCE_WORKER_UNAVAIL;
w->host->status |= XCE_WORKER_UNAVAIL;
DEREF_OBJ(w);
@ -432,12 +446,17 @@ int on_worker_join(void* cls, endpoint_t* ep)
int get_resval(cluster_t* c, char* resname, char* host, xce_val_t* val)
{
xce_resource_t* r;
xce_resource_t* r = NULL;
zarray_t* array = NULL;
xce_val_t* v = NULL;
worker_t* w = NULL;
int rc = 0;
rc = key_find(c->resmgr, (void*)resname, (void**)&r);
if (rc < 0) {
return rc;
}
r = resmgr_lockres(&(c->resmgr), resname, OBJREF_READ);
if (r == NULL) {
return -ENOENT;
}
@ -449,13 +468,15 @@ int get_resval(cluster_t* c, char* resname, char* host, xce_val_t* val)
return -EINVAL;
}
w = workermgr_lock(&(c->workermgr), host, OBJREF_READ);
rc = key_find(c->workermgr, (void*)host, (void**)&w);
if (rc < 0) {
return rc;
}
if (w == NULL) {
DEREF_OBJ(r);
return -ENOENT;
}
array = &(w->host.resvals);
array = &(w->host->resvals);
} else {
array = &(c->sharedinst);
}
@ -464,15 +485,11 @@ int get_resval(cluster_t* c, char* resname, char* host, xce_val_t* val)
v = (xce_val_t*)array->data[r->idx];
}
obj_deref(r);
if (v == NULL) {
DEREF_OBJ(w);
return -ENOENT;
}
memcpy(val, v, sizeof(*val));
DEREF_OBJ(w);
return 1;
}
@ -480,7 +497,7 @@ int mark_worker_unavail(cluster_t* c, worker_t* w)
{
xce_run_hooks(&(c->hooksys), XCE_CONTROLLER_HOOK_WORKER, HOOK_BEFORE_EVENT,
HOOK_DETAIL_WORKER_DEAD, c, w);
w->host.status |= XCE_WORKER_UNAVAIL;
w->host->status |= XCE_WORKER_UNAVAIL;
xce_run_hooks(&(c->hooksys), XCE_CONTROLLER_HOOK_WORKER, HOOK_AFTER_EVENT,
HOOK_DETAIL_WORKER_DEAD, c, w);
@ -491,7 +508,7 @@ int mark_worker_back(cluster_t* c, worker_t* w)
{
xce_run_hooks(&(c->hooksys), XCE_CONTROLLER_HOOK_WORKER, HOOK_BEFORE_EVENT,
HOOK_DETAIL_WORKER_UPDATE, c, w);
w->host.status &= ~XCE_WORKER_UNAVAIL;
w->host->status &= ~XCE_WORKER_UNAVAIL;
xce_run_hooks(&(c->hooksys), XCE_CONTROLLER_HOOK_WORKER, HOOK_AFTER_EVENT,
HOOK_DETAIL_WORKER_UPDATE, c, w);
@ -501,6 +518,7 @@ int mark_worker_back(cluster_t* c, worker_t* w)
int handle_worker_leave(cluster_t* c, int idx)
{
worker_t* w = NULL;
int rc;
TRACE("%s worker %d broken connection", __func__, idx);
@ -514,7 +532,11 @@ int handle_worker_leave(cluster_t* c, int idx)
return -errno;
}
w = workermgr_lock_byid(&(c->workermgr), idx, OBJREF_UPDATE);
rc = vget(c->workermgr, idx, (void**)&w);
if (rc < 0) {
return rc;
}
if (w == NULL) {
ERROR("fail to find worker by id %d", idx);
SET_ERROR(E_XCE_BAD_WORKER);
@ -539,7 +561,12 @@ int worker_ctrl(cluster_t* c, xce_hostctrl_t* ctrl)
return -EINVAL;
}
TRACE("%s worker_ctrl worker %s", __func__, ctrl->hostname);
w = workermgr_lock(&(c->workermgr), ctrl->hostname, OBJREF_UPDATE);
rc = key_find(c->workermgr, (void*)ctrl->hostname, (void**)&w);
if (rc < 0) {
return rc;
}
if (w == NULL) {
PERROR("workermgr_lock");
return -ENOENT;
@ -547,16 +574,16 @@ int worker_ctrl(cluster_t* c, xce_hostctrl_t* ctrl)
switch (ctrl->action) {
case HOST_OPEN:
if (!(w->host.status & XCE_WORKER_CLOSED)) {
if (!(w->host->status & XCE_WORKER_CLOSED)) {
goto clean;
}
w->host.status &= ~XCE_WORKER_CLOSED;
w->host->status &= ~XCE_WORKER_CLOSED;
break;
case HOST_CLOSE:
if (w->host.status & XCE_WORKER_CLOSED) {
if (w->host->status & XCE_WORKER_CLOSED) {
goto clean;
}
w->host.status |= XCE_WORKER_CLOSED;
w->host->status |= XCE_WORKER_CLOSED;
break;
default:
rc = -1;

1011
src/controller/xceobjs.c
View file

File diff suppressed because it is too large Load diff