From 3f74f146870dbe5979e95dd47e1c584a1ce4eb48 Mon Sep 17 00:00:00 2001 From: Rongsong Shen Date: Wed, 4 Mar 2026 18:18:53 +0800 Subject: [PATCH] save works --- src/controller/apiserver.c | 20 ++- src/controller/cluster.c | 49 +++++- src/controller/dataex.c | 60 ++++++- src/controller/job.c | 161 +++++++++++++++++-- src/controller/join.c | 2 + src/controller/misc.c | 6 + src/controller/opget.c | 17 +- src/controller/queue.c | 23 +++ src/controller/sched.c | 17 +- src/controller/task.c | 18 +++ src/controller/worker.c | 51 ++++++ src/docs/xce.md | 114 +++---------- src/flake.lock | 8 +- src/flake.nix | 12 +- src/include/kvmgr.h | 1 - src/include/xce/cluster_def.h | 2 + src/lib/configobj.c | 14 +- src/lib/group.c | 116 ++----------- src/lib/jobj.c | 77 +++++---- src/lib/kvmgr.c | 108 +++++-------- src/lib/qobj.c | 36 +++-- src/lib/resobj.c | 33 ++-- src/lib/resreq.c | 17 +- src/lib/testsuite/test_jobj.c | 1 - src/lib/testsuite/test_kvmgr.c | 5 +- src/lib/testsuite/test_resreq.c | 21 +++ src/lib/wobj.c | 33 ++-- src/lib/xcelimit.c | 12 ++ src/lib/xstring.c | 22 +-- src/lib/zarray.c | 32 ++-- src/meson.build | 3 + src/scheduler/include/plugin.h | 3 +- src/scheduler/src/plugin.c | 8 +- src/scheduler/src/plugins/limit.c | 26 ++- src/scheduler/src/plugins/seqjob_allocator.c | 3 + src/scheduler/src/resource.c | 5 + src/scheduler/src/scheduler.c | 52 ++++-- src/tools/appimage/entry.sh | 11 +- 38 files changed, 751 insertions(+), 448 deletions(-) diff --git a/src/controller/apiserver.c b/src/controller/apiserver.c index 825f62a..6b86f66 100644 --- a/src/controller/apiserver.c +++ b/src/controller/apiserver.c @@ -52,7 +52,7 @@ static int xce_apiserver_push_workers(cluster_t* c, endpoint_t* ep) } unlock_collect_workers(&wlist); - + FREEUP(workers.workers); return rc; @@ -88,7 +88,7 @@ static int xce_apiserver_push_jobs(cluster_t* c, endpoint_t* ep) rc = xce_push_data_req(ep, c->myhost, &data, check_controller, c, 100); unlock_collect_jobs(&jlist); - + FREEUP(jobs.jobs); return rc; @@ -394,10 +394,8 @@ static int apiserver_worker_change(int type, int situation, void* data, break; } -#if 0 - free_resval_with_idx_list(&(w->host.resvals_to_sched), - w->host.nresvals_to_sched); -#endif + w->host.nresvals_to_sched = 0; + w->host.resvals_to_sched = NULL; xlist_fini(&resvalist); return 0; @@ -416,7 +414,13 @@ static int apiserver_worker_join(int type, int situation, void* data, return 0; } + lock(c->workermgr, RTTI_LOCK_WRITE); rc = key_find(c->workermgr, (void*)name, (void**)&w); + if (w != NULL) { + reference(w); + } + unlock(c->workermgr); + if (rc < 0) { return rc; } @@ -425,9 +429,11 @@ static int apiserver_worker_join(int type, int situation, void* data, return -ENOENT; } + lock(w, RTTI_LOCK_READ); rc = apiserver_worker_change(type, situation, data, c, w); + unlock(w); - /* FIXME: need to mark the jobs in this worker */ + dereference(w); return rc; } diff --git a/src/controller/cluster.c b/src/controller/cluster.c index 509eff6..aa978fd 100644 --- a/src/controller/cluster.c +++ b/src/controller/cluster.c @@ -17,12 +17,12 @@ #include "xce/types.h" +#include "jobj.h" #include "resobj.h" #include "utils.h" #include "worker.h" #include "xce/errnox.h" #include "xstring.h" -#include "jobj.h" static xce_peerinfo_t from_controller = { .name = "controller", .key = "keepSecret" }; @@ -376,6 +376,14 @@ int cluster_build_resources(cluster_t* c) if (res == NULL) { return -errno; } + + if ((res->flags & XCE_RESF_GLOBAL) != 0) { + res->idx = c->n_shared_res; + c->n_shared_res++; + } else { + res->idx = c->n_host_res; + c->n_host_res++; + } } return 0; @@ -509,25 +517,20 @@ int cluster_init(cluster_t* c) 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, .datakiller = (destroy_fn_t)res_obj_killer }; kvmgr_params_t jobmgr_params = { .key_type = KVMGR_KEY_INT, - .dynamic = 1, .datakiller = (destroy_fn_t)job_obj_killer }; kvmgr_params_t workermgr_params = { .key_type = KVMGR_KEY_STRING, - .dynamic = 1, .datakiller = (destroy_fn_t)worker_obj_killer }; kvmgr_params_t qmgr_params = { .key_type = KVMGR_KEY_STRING, - .dynamic = 0, .datakiller = (destroy_fn_t)queue_obj_killer }; if (c == NULL) { @@ -681,9 +684,17 @@ static int check_workers_timeout(void* args) iterator_t it; cluster_t* c = (cluster_t*)args; int rc = 0; + xlist_t l; + int i; + xlist_init(&l, 1024); + + lock(c->workermgr, RTTI_LOCK_READ); rc = iter_make(&it, c->workermgr); if (rc < 0) { + unlock(c->workermgr); + xlist_fini(&l); + return rc; } @@ -695,7 +706,12 @@ static int check_workers_timeout(void* args) break; } - check_worker_timeout_(c, w); + /* check_worker_timeout_(c, w); */ + if (w != NULL) { + reference(w); + + xlist_store(&l, w); + } } if (rc == RTTI_ITER_NO_DATA) { @@ -703,6 +719,23 @@ static int check_workers_timeout(void* args) } iter_destroy(&it); + unlock(c->workermgr); + + for (i = 0; i < l.count; i++) { + worker_t* w = xlist_get(&l, i); + + if (w == NULL) { + continue; + } + + lock(w, RTTI_LOCK_WRITE); + check_worker_timeout_(c, w); + unlock(w); + + dereference(w); + } + + xlist_fini(&l); return rc; } @@ -781,7 +814,7 @@ void cluster_fini(cluster_t* c) cleanup_openssl(); job_obj_fini(); - + xstring_table_fini(); return; diff --git a/src/controller/dataex.c b/src/controller/dataex.c index f74cd31..2f770d7 100644 --- a/src/controller/dataex.c +++ b/src/controller/dataex.c @@ -72,13 +72,29 @@ int collect_resval_with_idx(zarray_t* array, xlist_t* l) return rc; } +static void lock_objs(xlist_t* l, int op) +{ + int i; + + for (i = 0; i < l->count; i++) { + var obj = xlist_get(l, i); + + if (obj != NULL) { + lock(obj, op); + } + } +} + void collect_confvars(cluster_t* c, collect_check_fn_t fn, xlist_t* l) { iterator_t it; int rc; + lock(c->configmgr, RTTI_LOCK_READ); rc = iter_make(&it, c->configmgr); if (rc < 0) { + unlock(c->configmgr); + return; } @@ -95,11 +111,15 @@ void collect_confvars(cluster_t* c, collect_check_fn_t fn, xlist_t* l) } if ((fn == NULL) || (fn(config))) { + reference(config); xlist_store(l, config); } } iter_destroy(&it); + unlock(c->configmgr); + + lock_objs(l, RTTI_LOCK_READ); return; } @@ -109,8 +129,11 @@ void collect_queues(cluster_t* c, collect_check_fn_t fn, xlist_t* l) iterator_t it; int rc; + lock(c->qmgr, RTTI_LOCK_READ); rc = iter_make(&it, c->qmgr); if (rc < 0) { + unlock(c->qmgr); + return; } @@ -125,13 +148,16 @@ void collect_queues(cluster_t* c, collect_check_fn_t fn, xlist_t* l) if (q == NULL) { continue; } - if ((fn != NULL) || fn(q)) { - lock(q, RTTI_LOCK_READ); + if ((fn == NULL) || fn(q)) { + reference(q); xlist_store(l, &(q->queue)); } } iter_destroy(&it); + unlock(c->qmgr); + + lock_objs(l, RTTI_LOCK_READ); } void collect_resources(cluster_t* c, collect_check_fn_t fn, xlist_t* l) @@ -139,8 +165,11 @@ void collect_resources(cluster_t* c, collect_check_fn_t fn, xlist_t* l) iterator_t it; int rc; + lock(c->resmgr, RTTI_LOCK_READ); rc = iter_make(&it, c->resmgr); if (rc < 0) { + unlock(c->resmgr); + return; } @@ -157,11 +186,15 @@ void collect_resources(cluster_t* c, collect_check_fn_t fn, xlist_t* l) } if ((fn == NULL) || (fn(r))) { + reference(r); xlist_store(l, r); } } iter_destroy(&it); + unlock(c->resmgr); + + lock_objs(l, RTTI_LOCK_READ); return; } @@ -171,8 +204,11 @@ void collect_jobs(cluster_t* c, collect_check_fn_t fn, xlist_t* l) iterator_t it; int rc = 0; + lock(c->jobmgr, RTTI_LOCK_READ); rc = iter_make(&it, c->jobmgr); if (rc < 0) { + unlock(c->jobmgr); + return; } @@ -189,12 +225,16 @@ void collect_jobs(cluster_t* c, collect_check_fn_t fn, xlist_t* l) } if ((fn == NULL) || (fn(j))) { - lock(j, RTTI_LOCK_READ); + reference(j); xlist_store(l, &(j->job)); } } iter_destroy(&it); + unlock(c->jobmgr); + + lock_objs(l, RTTI_LOCK_READ); + return; } @@ -203,8 +243,11 @@ void collect_workers(cluster_t* c, collect_check_fn_t fn, xlist_t* l) iterator_t it; int rc = 0; + lock(c->workermgr, RTTI_LOCK_READ); rc = iter_make(&it, c->workermgr); if (rc < 0) { + unlock(c->workermgr); + return; } @@ -221,12 +264,15 @@ void collect_workers(cluster_t* c, collect_check_fn_t fn, xlist_t* l) } if ((fn == NULL) || (fn(w))) { - lock(w, RTTI_LOCK_READ); + reference(w); xlist_store(l, &(w->host)); } } iter_destroy(&it); + unlock(c->workermgr); + + lock_objs(l, RTTI_LOCK_READ); return; } @@ -258,6 +304,8 @@ void unlock_collect_queues(xlist_t* l) q1 = CONTAINER_OF(q, queue_t, queue); unlock(q1); + + dereference(q1); } } @@ -275,6 +323,8 @@ void unlock_collect_workers(xlist_t* l) w1 = CONTAINER_OF(w, worker_t, host); unlock(w1); + + dereference(w1); } } @@ -292,5 +342,7 @@ void unlock_collect_jobs(xlist_t* l) j1 = CONTAINER_OF(j, job_t, job); unlock(j1); + + dereference(j1); } } diff --git a/src/controller/job.c b/src/controller/job.c index 4d16074..c975c1b 100644 --- a/src/controller/job.c +++ b/src/controller/job.c @@ -157,7 +157,13 @@ static int dispatch_a_job_(cluster_t* c, endpoint_t* ep, xce_job_alloc_t* alloc) TRACE("%s jobid %ld", __func__, alloc->jobid); + lock(c->jobmgr, RTTI_LOCK_READ); rc = key_find(c->jobmgr, (void*)alloc->jobid, (void**)&j); + if (j != NULL) { + reference(j); + } + unlock(c->jobmgr); + if (rc < 0) { return rc; } @@ -170,6 +176,8 @@ static int dispatch_a_job_(cluster_t* c, endpoint_t* ep, xce_job_alloc_t* alloc) nc = alloc->nhosts; if (nc <= 0) { + dereference(j); + return -EINVAL; } @@ -180,7 +188,12 @@ 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 */ + lock(c->workermgr, RTTI_LOCK_READ); rc = key_find(c->workermgr, (void*)allocs[0]->host, (void**)&w); + if (w != NULL) { + reference(w); + } + unlock(c->workermgr); if (rc < 0) { return rc; } @@ -190,12 +203,18 @@ static int dispatch_a_job_(cluster_t* c, endpoint_t* ep, xce_job_alloc_t* alloc) return -errno; } + lock(w, RTTI_LOCK_READ); 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); free_launchinfo(&launchinfo); SET_ERROR(E_XCE_BAD_WORKER); + + unlock(w); + + dereference(w); + return -errno; } @@ -209,11 +228,9 @@ static int dispatch_a_job_(cluster_t* c, endpoint_t* ep, xce_job_alloc_t* alloc) return rc; } - if (j->job.submit->start_info->q_pre_exec != NULL) { - } + if (j->job.submit->start_info->q_pre_exec != NULL) { } - if (j->job.submit->start_info->q_post_exec != NULL) { - } + if (j->job.submit->start_info->q_post_exec != NULL) { } j->job.submit->start_info->q_pre_exec = q_pre_exec; j->job.submit->start_info->q_post_exec = q_post_exec; @@ -225,6 +242,10 @@ static int dispatch_a_job_(cluster_t* c, endpoint_t* ep, xce_job_alloc_t* alloc) rc = xce_startjob_req(w->ep, &einfo); + unlock(w); + + dereference(w); + free_launchinfo(&launchinfo); if (j->job.allocs != NULL) { @@ -236,7 +257,6 @@ static int dispatch_a_job_(cluster_t* c, endpoint_t* ep, xce_job_alloc_t* alloc) MOVE_PTR(j->job.allocs->allocs, allocs); alloc->allocs = NULL; - if (rc < 0) { SET_ERROR(-rc); PERROR("xce_startjob_req"); @@ -254,7 +274,13 @@ int job_ctrl(cluster_t* c, xce_jobctrl_t* ctrinfo) TRACE("%s job_ctrl job %ld", __func__, jobid); + lock(c->jobmgr, RTTI_LOCK_READ); rc = key_find(c->jobmgr, (void*)jobid, (void**)&j); + if (j != NULL) { + reference(j); + } + unlock(c->jobmgr); + if (rc < 0) { return rc; } @@ -264,6 +290,8 @@ int job_ctrl(cluster_t* c, xce_jobctrl_t* ctrinfo) return -ENOENT; } + /* FIXME: eliminate held multiple locks */ + lock(j, RTTI_LOCK_READ); if (j->job.allocs != NULL) { xce_host_alloc_t** allocs = j->job.allocs->allocs; worker_t* w = NULL; @@ -271,11 +299,19 @@ int job_ctrl(cluster_t* c, xce_jobctrl_t* ctrinfo) /* job control command will be sent to first node * for parallel jobs */ + + lock(c->workermgr, RTTI_LOCK_READ); rc = key_find(c->workermgr, (void*)allocs[0]->host, (void**)&w); + if (w != NULL) { + reference(w); + } + unlock(c->workermgr); + if (rc < 0) { return rc; } if (w != NULL) { + lock(w, RTTI_LOCK_READ); /* FIXME: better method to check host status */ if ((w->ep != NULL) && (w->host.status == 0)) { rc = xce_jobctrl_req(w->ep, ctrinfo); @@ -288,6 +324,9 @@ int job_ctrl(cluster_t* c, xce_jobctrl_t* ctrinfo) w->host.name, w->host.status, w->ep); rc = -errno; } + unlock(w); + + dereference(w); } else { rc = -errno; } @@ -353,6 +392,8 @@ int job_ctrl(cluster_t* c, xce_jobctrl_t* ctrinfo) } } + unlock(j); + return rc; } @@ -409,10 +450,7 @@ int handle_jobdispatch(cluster_t* c, endpoint_t* ep, xce_request_t* request) return rc; } -static int xlist_deref_job_item_(job_t* j, void* arg) -{ - return 0; -} +static int xlist_deref_job_item_(job_t* j, void* arg) { return 0; } int handle_jobsubmit(cluster_t* c, endpoint_t* ep, xce_request_t* request) { @@ -457,7 +495,13 @@ 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 */ + lock(c->jobmgr, RTTI_LOCK_READ); rc = key_find(c->jobmgr, (void*)(jobid + i), (void**)&j); + if (j != NULL) { + reference(j); + } + unlock(c->jobmgr); + if ((rc <= 0) || (j == NULL)) { char idstring[128]; @@ -514,14 +558,23 @@ int on_job_start(cluster_t* c, job_pidinfo_t* info) TRACE("%s jobid %ld", __func__, info->jobid); + lock(c->jobmgr, RTTI_LOCK_READ); rc = key_find(c->jobmgr, (void*)info->jobid, (void**)&job); + if (job != NULL) { + reference(job); + } + unlock(c->jobmgr); if ((rc <= 0) || (job == NULL)) { ERROR("can not find job %ld", info->jobid); return -errno; } + lock(job, RTTI_LOCK_WRITE); mark_job_start(c, job, info); + unlock(job); + + dereference(job); return 0; } @@ -563,14 +616,23 @@ 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); + lock(c->jobmgr, RTTI_LOCK_READ); key_find(c->jobmgr, (void*)finish->jobid, (void**)&job); + if (job != NULL) { + reference(job); + } + unlock(c->jobmgr); if (job == NULL) { ERROR("can not find job %ld", finish->jobid); return -errno; } + lock(job, RTTI_LOCK_WRITE); mark_job_finish(c, job, finish); + unlock(job); + + dereference(job); return 0; } @@ -625,13 +687,23 @@ int on_job_fail(cluster_t* c, job_failinfo_t* failinfo) TRACE("%s job fail jobid %ld code %d", __func__, failinfo->jobid, failinfo->code); + lock(c->jobmgr, RTTI_LOCK_READ); key_find(c->jobmgr, (void*)failinfo->jobid, (void**)&job); + if (job != NULL) { + reference(job); + } + unlock(c->jobmgr); + if (job == NULL) { ERROR("can not find job %ld", failinfo->jobid); return -errno; } + lock(job, RTTI_LOCK_WRITE); mark_job_fail(c, job, failinfo); + unlock(job); + + dereference(job); return 0; } @@ -643,16 +715,26 @@ int on_job_stop(cluster_t* c, uint64_t jobid) TRACE("%s job stop jobid %ld", __func__, jobid); + lock(c->jobmgr, RTTI_LOCK_READ); key_find(c->jobmgr, (void*)jobid, (void**)&job); + if (job != NULL) { + reference(job); + } + unlock(c->jobmgr); + if (job == NULL) { ERROR("can not find job %ld", jobid); return -errno; } + lock(job, RTTI_LOCK_WRITE); state = job->job.state; state |= JOB_STAT_USUSP; mark_job_update(c, job, state); + unlock(job); + + dereference(job); return 0; } @@ -664,15 +746,25 @@ int on_job_resume(cluster_t* c, uint64_t jobid) TRACE("%s job resume jobid %ld", __func__, jobid); + lock(c->jobmgr, RTTI_LOCK_READ); key_find(c->jobmgr, (void*)jobid, (void**)&job); + if (job != NULL) { + reference(job); + } + unlock(c->jobmgr); + if (job == NULL) { ERROR("can not find job %ld", jobid); return -errno; } + lock(job, RTTI_LOCK_WRITE); state = job->job.state; state &= ~JOB_STAT_USUSP; mark_job_update(c, job, state); + unlock(job); + + dereference(job); return 0; } @@ -684,12 +776,19 @@ int on_job_postdone(cluster_t* c, endpoint_t* ep, uint64_t jobid) TRACE("%s job post done jobid %ld", __func__, jobid); + lock(c->jobmgr, RTTI_LOCK_READ); key_find(c->jobmgr, (void*)jobid, (void**)&job); + if (job != NULL) { + reference(job); + } + unlock(c->jobmgr); + if (job == NULL) { ERROR("can not find job %ld", jobid); return -errno; } + lock(job, RTTI_LOCK_WRITE); 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 " @@ -702,6 +801,9 @@ int on_job_postdone(cluster_t* c, endpoint_t* ep, uint64_t jobid) state |= JOB_STAT_PDONE; mark_job_update(c, job, state); + unlock(job); + + dereference(job); return 0; } @@ -722,22 +824,32 @@ int on_job_ckpt(cluster_t* c, endpoint_t* ep, xce_ckptinfo_t* ckptinfo) TRACE("%s job %ld ckptinfo %s", __func__, jobid, info); + lock(c->jobmgr, RTTI_LOCK_READ); key_find(c->jobmgr, (void*)jobid, (void**)&j); + if (j != NULL) { + reference(j); + } + unlock(c->jobmgr); if (j == NULL) { ERROR("%s can not find job %ld", __func__, jobid); return -errno; } + lock(j, RTTI_LOCK_WRITE); xce_run_hooks(&(c->hooksys), XCE_CONTROLLER_HOOK_JOB, HOOK_BEFORE_EVENT, HOOK_DETAIL_JOB_CKPT, c, j); if (j->job.ckptinfo != NULL) { + dereference(j->job.ckptinfo); } j->job.ckptinfo = makeString(info); xce_run_hooks(&(c->hooksys), XCE_CONTROLLER_HOOK_JOB, HOOK_AFTER_EVENT, HOOK_DETAIL_JOB_CKPT, c, j); + unlock(j); + + dereference(j); return 0; } @@ -769,7 +881,13 @@ 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_RUSAGE_UPDATE, c, rusages); + lock(c->jobmgr, RTTI_LOCK_READ); key_find(c->jobmgr, (void*)rusages->jobid, (void**)&j); + if (j != NULL) { + reference(j); + } + unlock(c->jobmgr); + if (j == NULL) { PERROR("jobmgr_lockjob"); return -errno; @@ -782,12 +900,15 @@ int update_job_rusages(cluster_t* c, endpoint_t* ep, xce_request_t* request) host = makeString(w->host.name); if (host == NULL) { + dereference(j); + return -ENOMEM; } TRACE("%s update job (%ld) rusage for host %s", __func__, rusages->jobid, host); + lock(j, RTTI_LOCK_WRITE); 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, @@ -805,6 +926,9 @@ int update_job_rusages(cluster_t* c, endpoint_t* ep, xce_request_t* request) xce_run_hooks(&(c->hooksys), XCE_CONTROLLER_HOOK_JOB, HOOK_AFTER_EVENT, HOOK_DETAIL_JOB_RUSAGE_UPDATE, c, rusages); + unlock(j); + + dereference(j); return 0; } @@ -816,12 +940,19 @@ static int set_preason(cluster_t* c, xce_pendreason_t* reasons) TRACE("%s jobid %ld", __func__, reasons->jobid); + lock(c->jobmgr, RTTI_LOCK_READ); key_find(c->jobmgr, (void*)reasons->jobid, (void**)&j); + if (j != NULL) { + reference(j); + } + unlock(c->jobmgr); + if (j == NULL) { PERROR("jobmgr_lockjob"); return -ENOENT; } + lock(j, RTTI_LOCK_WRITE); xce_run_hooks(&(c->hooksys), XCE_CONTROLLER_HOOK_JOB, HOOK_BEFORE_EVENT, HOOK_DETAIL_SETPREASON, c, j); job = &(j->job); @@ -833,6 +964,9 @@ 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); + unlock(j); + + dereference(j); return 0; } @@ -882,12 +1016,19 @@ int do_clr_preason(cluster_t* c, endpoint_t* ep, xce_request_t* request) job_t* j; xce_job_t* job; + lock(c->jobmgr, RTTI_LOCK_READ); key_find(c->jobmgr, (void*)jobid, (void**)&j); + if (j != NULL) { + reference(j); + } + unlock(c->jobmgr); + if (j == NULL) { PERROR("jobmgr_lockjob"); continue; } + lock(j, RTTI_LOCK_WRITE); xce_run_hooks(&(c->hooksys), XCE_CONTROLLER_HOOK_JOB, HOOK_BEFORE_EVENT, HOOK_DETAIL_CLRPREASON, c, j); @@ -899,7 +1040,9 @@ 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); + unlock(j); + dereference(j); } return 0; diff --git a/src/controller/join.c b/src/controller/join.c index 8e016ef..2fe38a7 100644 --- a/src/controller/join.c +++ b/src/controller/join.c @@ -55,6 +55,7 @@ static int handle_worker_join(cluster_t* c, endpoint_t* ep, joinmsg = request->body.joinmsg; + lock(c->workermgr, RTTI_LOCK_WRITE); xce_run_hooks(&(c->hooksys), XCE_CONTROLLER_HOOK_WORKER, HOOK_BEFORE_EVENT, HOOK_DETAIL_WORKER_JOIN, c, from->name, 0); idx = add_worker(c, ep, from->name, joinmsg->data.attrs); @@ -65,6 +66,7 @@ static int handle_worker_join(cluster_t* c, endpoint_t* ep, xce_run_hooks(&(c->hooksys), XCE_CONTROLLER_HOOK_WORKER, HOOK_AFTER_EVENT, HOOK_DETAIL_WORKER_JOIN, c, from->name, idx); + unlock(c->workermgr); if (idx < 0) { ERROR("worker %s join failed, rc %d", from->name, idx); diff --git a/src/controller/misc.c b/src/controller/misc.c index 24e29c1..f31150a 100644 --- a/src/controller/misc.c +++ b/src/controller/misc.c @@ -64,7 +64,13 @@ worker_t* get_ep_worker(cluster_t* c, endpoint_t* ep) return NULL; } + lock(c->workermgr, RTTI_LOCK_READ); rc = key_find(c->workermgr, (void*)idx, (void**)&w); + if (w != NULL) { + reference(w); + } + unlock(c->workermgr); + if (rc < 0) { return NULL; } diff --git a/src/controller/opget.c b/src/controller/opget.c index 771d692..f9a94a2 100644 --- a/src/controller/opget.c +++ b/src/controller/opget.c @@ -129,7 +129,7 @@ static int do_reply_workerlist(cluster_t* c, endpoint_t* ep, } unlock_collect_workers(&wlist); - + xlist_fini(&wlist); return rc; @@ -148,7 +148,7 @@ static int do_reply_joblist(cluster_t* c, endpoint_t* ep, init_reply_by_req(&reply, request); - xlist_init(&jlist, 1024*1024); + xlist_init(&jlist, 1024 * 1024); if (match->type == MatchCondition_condition_matchjid) { job_t* j = NULL; @@ -156,10 +156,18 @@ static int do_reply_joblist(cluster_t* c, endpoint_t* ep, jobdata.n_total = 0; jobdata.jobs = NULL; + lock(c->jobmgr, RTTI_LOCK_READ); rc = key_find(c->jobmgr, (void*)match->condition.jobid, (void**)&j); + if (j != NULL) { + reference(j); + } + unlock(c->jobmgr); + if (rc > 0) { - lock(j, RTTI_LOCK_WRITE); + lock(j, RTTI_LOCK_READ); xlist_store(&jlist, &(j->job)); + } else { + dereference(j); } } else { collect_jobs(c, NULL, &jlist); @@ -174,9 +182,8 @@ static int do_reply_joblist(cluster_t* c, endpoint_t* ep, rc = xce_reply_get(ep, c->myhost, &reply, check_controller, c, 100); - unlock_collect_jobs(&jlist); - + FREEUP(jobdata.jobs); return rc; diff --git a/src/controller/queue.c b/src/controller/queue.c index eb90eb5..77953cf 100644 --- a/src/controller/queue.c +++ b/src/controller/queue.c @@ -55,7 +55,14 @@ int queue_ctrl(cluster_t* c, xce_queuectrl_t* ctrl) return -EINVAL; } TRACE("%s queue_ctrl queue %s", __func__, ctrl->queue); + + lock(c->qmgr, RTTI_LOCK_READ); rc = key_find(c->qmgr, (void*)ctrl->queue, (void**)&q); + if (q != NULL) { + reference(q); + } + unlock(c->qmgr); + if (rc < 0) { return rc; } @@ -65,6 +72,7 @@ int queue_ctrl(cluster_t* c, xce_queuectrl_t* ctrl) return -ENOENT; } + lock(q, RTTI_LOCK_WRITE); switch (ctrl->action) { case QUEUE_OPEN: if (!(q->queue.status & QUEUE_STAT_CLOSED)) { @@ -98,7 +106,12 @@ int queue_ctrl(cluster_t* c, xce_queuectrl_t* ctrl) xce_run_hooks(&(c->hooksys), XCE_CONTROLLER_HOOK_QUEUE, HOOK_AFTER_EVENT, HOOK_DETAIL_QUEUE_UPDATE, c, q); } + clean: + unlock(q); + + dereference(q); + return rc; } @@ -131,7 +144,13 @@ int get_queue_exec(cluster_t* c, char* queue, char** pre_exec, char** post_exec) TRACE("%s queue %s", __func__, queue); + lock(c->qmgr, RTTI_LOCK_READ); rc = key_find(c->qmgr, (void*)queue, (void**)&q); + if (q != NULL) { + reference(q); + } + unlock(c->qmgr); + if (rc < 0) { return rc; } @@ -141,6 +160,7 @@ int get_queue_exec(cluster_t* c, char* queue, char** pre_exec, char** post_exec) return -ENOENT; } + lock(q, RTTI_LOCK_READ); if (q->queue.pre_exec != NULL) { *pre_exec = makeString(q->queue.pre_exec); } else { @@ -152,6 +172,9 @@ int get_queue_exec(cluster_t* c, char* queue, char** pre_exec, char** post_exec) } else { *post_exec = NULL; } + unlock(q); + + dereference(q); return 0; } diff --git a/src/controller/sched.c b/src/controller/sched.c index 2e4cc9c..f6cd22b 100644 --- a/src/controller/sched.c +++ b/src/controller/sched.c @@ -148,7 +148,7 @@ static int xcesched_push_configs(cluster_t* c) FREEUP(configdata.data.conf4sched); unlock_collect_queues(&qlist); - + xlist_fini(&clist); xlist_fini(&qlist); xlist_fini(&rlist); @@ -187,7 +187,9 @@ static int xcesched_push_jobs(cluster_t* c) rc = xce_push_data_req(c->sched.ep, c->myhost, &data, check_controller, c, 100); - FREEUP(jobs.jobs); + unlock_collect_jobs(&jlist); + + xlist_fini(&jlist); return rc; } @@ -224,7 +226,9 @@ static int xcesched_push_workers(cluster_t* c) free_resval_with_idx_list(&(w->resvals_to_sched), w->nresvals_to_sched); } - FREEUP(workers.workers); + unlock_collect_workers(&wlist); + + xlist_fini(&wlist); return rc; } @@ -476,7 +480,12 @@ static int sched_worker_join(int type, int situation, void* data, va_list ap) return 0; } + lock(c->workermgr, RTTI_LOCK_READ); rc = key_find(c->workermgr, (void*)name, (void**)&w); + if (w != NULL) { + reference(w); + } + unlock(c->workermgr); if (rc < 0) { return rc; } @@ -487,6 +496,8 @@ static int sched_worker_join(int type, int situation, void* data, va_list ap) rc = sched_worker_change(type, situation, data, c, w); + dereference(w); + return rc; } diff --git a/src/controller/task.c b/src/controller/task.c index 50d430b..1e9942f 100644 --- a/src/controller/task.c +++ b/src/controller/task.c @@ -17,7 +17,12 @@ static int run_tasks_on_host(cluster_t* c, endpoint_t* ep, uint64_t jobid, .clientidx = ep->idx, .taskhosts = (xce_taskhost_t*[]) { info } }; + lock(c->workermgr, RTTI_LOCK_READ); rc = key_find(c->workermgr, (void*)info->host, (void**)&w); + if (w != NULL) { + reference(w); + } + unlock(c->workermgr); if (rc < 0) { return rc; } @@ -27,10 +32,13 @@ static int run_tasks_on_host(cluster_t* c, endpoint_t* ep, uint64_t jobid, } if ((w->ep == NULL) || (IS_WORKER_UNAVAIL(w->host.status))) { + dereference(w); + return -EINVAL; } rc = xce_runtask_req(w->ep, &runmsg); + dereference(w); return rc; } @@ -43,8 +51,14 @@ static int check_task_hosts(cluster_t* c, xce_taskrun_t* runmsg) worker_t* w; int rc = 0; + lock(c->workermgr, RTTI_LOCK_READ); rc = key_find(c->workermgr, (void*)runmsg->taskhosts[i]->host, (void**)&w); + if (w != NULL) { + reference(w); + } + unlock(c->workermgr); + if (rc < 0) { return rc; } @@ -57,8 +71,12 @@ static int check_task_hosts(cluster_t* c, xce_taskrun_t* runmsg) } if ((w->ep == NULL) || IS_WORKER_UNAVAIL(w->host.status)) { + dereference(w); + return -ENOENT; } + + dereference(w); } return 0; diff --git a/src/controller/worker.c b/src/controller/worker.c index cc11e80..6e282df 100644 --- a/src/controller/worker.c +++ b/src/controller/worker.c @@ -101,7 +101,10 @@ static int update_resval(cluster_t* c, worker_t* w, xce_resval_t* val) xce_val_t* v = NULL; int rc; + lock(c->resmgr, RTTI_LOCK_READ); rc = key_find(c->resmgr, (void*)val->name, (void**)&r); + unlock(c->resmgr); + if (rc < 0) { return rc; } @@ -161,7 +164,10 @@ static int mark_host_res(cluster_t* c, worker_t* w, xce_config_t* config) xce_val_t val; /* clang-format-on */ + lock(c->resmgr, RTTI_LOCK_READ); rc = key_find(c->resmgr, (void*)n, (void**)&r); + unlock(c->resmgr); + if (rc < 0) { return rc; } @@ -298,6 +304,9 @@ int add_worker(void* cls, endpoint_t* ep, const char* host, hostattrs_t* attrs) /* we should hold the update lock */ /* If admin makes the host in locked or closed, we need keep this */ + + lock(w, RTTI_LOCK_WRITE); + w->host.status &= ~XCE_WORKER_UNAVAIL; /* status OK */ w->ep = ep; @@ -350,6 +359,8 @@ int add_worker(void* cls, endpoint_t* ep, const char* host, hostattrs_t* attrs) mark_host_res(c, w, hconfig->resources[i]); } + unlock(w); + return idx; } @@ -367,10 +378,15 @@ int restore_worker(void* cls, xce_worker_t host) if (w == NULL) { return -errno; } + + lock(w, RTTI_LOCK_WRITE); + add_hostattrs_resvals(c, w); idx = w->idx; w->host.status |= XCE_WORKER_UNAVAIL; + unlock(w); + return idx; } @@ -440,7 +456,10 @@ int get_resval(cluster_t* c, char* resname, char* host, xce_val_t* val) worker_t* w = NULL; int rc = 0; + lock(c->resmgr, RTTI_LOCK_READ); rc = key_find(c->resmgr, (void*)resname, (void**)&r); + unlock(c->resmgr); + if (rc < 0) { return rc; } @@ -455,7 +474,12 @@ int get_resval(cluster_t* c, char* resname, char* host, xce_val_t* val) return -EINVAL; } + lock(c->workermgr, RTTI_LOCK_READ); rc = key_find(c->workermgr, (void*)host, (void**)&w); + if (w != NULL) { + reference(w); + } + unlock(c->workermgr); if (rc < 0) { return rc; } @@ -473,10 +497,19 @@ int get_resval(cluster_t* c, char* resname, char* host, xce_val_t* val) } if (v == NULL) { + if (w != NULL) { + dereference(w); + } + return -ENOENT; } memcpy(val, v, sizeof(*val)); + + if (w != NULL) { + dereference(w); + } + return 1; } @@ -484,6 +517,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; xce_run_hooks(&(c->hooksys), XCE_CONTROLLER_HOOK_WORKER, HOOK_AFTER_EVENT, @@ -495,6 +529,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; xce_run_hooks(&(c->hooksys), XCE_CONTROLLER_HOOK_WORKER, HOOK_AFTER_EVENT, @@ -530,10 +565,14 @@ int handle_worker_leave(cluster_t* c, int idx) return -errno; } + lock(w, RTTI_LOCK_WRITE); + w->ep = NULL; mark_worker_unavail(c, w); + unlock(w); + return 0; } @@ -548,7 +587,13 @@ int worker_ctrl(cluster_t* c, xce_hostctrl_t* ctrl) } TRACE("%s worker_ctrl worker %s", __func__, ctrl->hostname); + lock(c->workermgr, RTTI_LOCK_READ); rc = key_find(c->workermgr, (void*)ctrl->hostname, (void**)&w); + if (w != NULL) { + reference(w); + } + unlock(c->workermgr); + if (rc < 0) { return rc; } @@ -558,6 +603,7 @@ int worker_ctrl(cluster_t* c, xce_hostctrl_t* ctrl) return -ENOENT; } + lock(w, RTTI_LOCK_WRITE); switch (ctrl->action) { case HOST_OPEN: if (!(w->host.status & XCE_WORKER_CLOSED)) { @@ -579,6 +625,11 @@ int worker_ctrl(cluster_t* c, xce_hostctrl_t* ctrl) xce_run_hooks(&(c->hooksys), XCE_CONTROLLER_HOOK_WORKER, HOOK_AFTER_EVENT, HOOK_DETAIL_WORKER_UPDATE, c, w); } + + unlock(w); + + dereference(w); + clean: return rc; } diff --git a/src/docs/xce.md b/src/docs/xce.md index 5849134..f4beb16 100644 --- a/src/docs/xce.md +++ b/src/docs/xce.md @@ -1,108 +1,32 @@ -# 1. commit 规则 -- 每个人 fork 自己的 branch， 从自己的 branch 做 PR，review 完再 commit +Help of XCE +=========== -# 2. code format -## 2.1. C -推荐使用 clang-format，公共配置放在源码里面 xce/.clang-format +This is the entrypoint of xce tool. Currently, we have following sub-commands available: -（理论上直接使用源码里面 xce/.clang-format 文件格式不会变） +- controller +- worker +- scheduler +- apiserver +- shell -### 2.1.1. vscode 可以在配置中加上如下内容 -``` -"C_Cpp.clang_format_style": "WebKit", -"C_Cpp.clang_format_fallbackStyle": "WebKit", -``` +If you run xce without any parameter or with parameter which is not in above, then it will show this help screen -### 2.1.2. vim & emacs +## controller -clang-format -style="{BasedOnStyle: webkit}" -i XX.c +This command with start xce controller with given parameters after controller sub-command -## 2.2. golang +## scheduler -使用默认 gofmt +This command with start xce scheduler with given parameters after scheduler sub-command -# 3. build environment and build (example of ubuntu) -## 3.1 build environment - -- Install nix package management tool +## worker -```sh -sh <(curl -L https://nixos.org/nix/install) --daemon -``` +This command will start xcecd with given parameters after worker sub-command -- Edit /etc/nix/nix.conf to make sure +## apiserver -``` -experimental-features = nix-command flakes -sandbox = true -sandbox-fallback = false -substituters = https://mirrors.ustc.edu.cn/nix-channels/store https://cache.nixos.org https://cache.nixos.org/ -trusted-users = root -``` +This command will start xce apiserver with given parameters after apiserver sub-command -make sure that you have restarted nix-daemon service after you have changed /etc/nix/nix.conf +## shell -- Install socat and config ssh (for clone by ssh protocol) - -```sh -nix profile add nixpkgs#socat -``` - -.ssh/config -``` -Host gitea.shenrs.eu - HostName gitea.shenrs.eu - IdentityFile ~/.ssh/id_ed25519 - ProxyCommand socat - openssl-connect:gitea.shenrs.eu:2022,snihost=gitea.shenrs.eu,verify=0 -``` - -## 3.2 checkout source code - -## 3.3 build (for package and release) - -```sh -cd xce/src - -# for build debug version -nix build --option sandbox false .#xcecore_debug - -# for normal build -nix build --option sandbox false .#xcecore -``` - -You can find the generated binaries at =result= directory - -## 3.4 build (for daily develop) - -```sh -cd xce/src -nix develop - -# using meson to generated build configuration (debug and address sanitize) - -meson setup build --buildtype=debug -Db_sanitize=address - -# run meson to build -meson compile -C build - -# The generated files will be in build/ -``` - -## 3.5 build AppImage package - -```sh -cd xce/src -nix bundle --bundler github:ralismark/nix-appimage --option sandbox false .#xce -``` - -This will create all in one AppImage file which contains everything for package delivery. - -# 4. misc -## 4.1. Windows remap ctrl and alt key -create a reg file, e.g. remapkey.reg -``` -Windows Registry Editor Version 5.00 -[HKEY_LOCAL_MACHINE\SYSTEM\CurrentControlSet\Control\Keyboard Layout] -"Scancode Map"=hex:00,00,00,00,00,00,00,00,05,00,00,00,38,00,1d,00,1d,00,38,00,38,e0,1d,e0,1d,e0,38,e0,00,00,00,00 -``` -then double click remapkey.reg to import it. Restart OS. +An interactive bash shell to start, this is just for trouble shoot diff --git a/src/flake.lock b/src/flake.lock index baeb7be..dc04ca1 100644 --- a/src/flake.lock +++ b/src/flake.lock @@ -117,11 +117,11 @@ "xbuild": "xbuild" }, "locked": { - "lastModified": 1772074796, - "narHash": "sha256-zcEQs+slX7r7C1u9VFvWXTU8BIRt7RmaDu6/mNHHrLc=", + "lastModified": 1772607564, + "narHash": "sha256-kBYbiOvM7bBZdIFq2idjQ3eidlAT3w6WS3GRoxHM67s=", "ref": "master", - "rev": "f66de934830db47db651e83f6c8e466cca19dd54", - "revCount": 18, + "rev": "10c32f769b4431cb657c3c8f23158f643fad74b4", + "revCount": 19, "type": "git", "url": "ssh://forgejo@gitea.shenrs.eu/rshen/librtti" }, diff --git a/src/flake.nix b/src/flake.nix index f499458..198ed8e 100644 --- a/src/flake.nix +++ b/src/flake.nix @@ -32,21 +32,23 @@ name = "xce"; runtimeInputs = with pkgs; [ xpkgs.xcecore + glibcLocales coreutils-full which bash gzip glow + less haproxy ]; text = appImageScriptContent; - derivationArgs = { - postInstall = '' - mkdir -p $out/share - cp docs/* $out/share/ - ''; + runtimeEnv = { + LOCALE_ARCHIVE="${pkgs.glibcLocales}/lib/locale/locale-archive"; + LANG="en_US.UTF-8"; + LESS="-R"; }; + meta = { mainProgram = "xce"; }; diff --git a/src/include/kvmgr.h b/src/include/kvmgr.h index 72d768c..4981870 100644 --- a/src/include/kvmgr.h +++ b/src/include/kvmgr.h @@ -15,7 +15,6 @@ typedef enum { typedef struct { kvmgr_key_type_t key_type; - unsigned int dynamic : 1; uint32_t nslots; uint32_t delta; destroy_fn_t datakiller; diff --git a/src/include/xce/cluster_def.h b/src/include/xce/cluster_def.h index cd6f35b..b5a84fd 100644 --- a/src/include/xce/cluster_def.h +++ b/src/include/xce/cluster_def.h @@ -102,6 +102,8 @@ typedef struct { int myidx; int master; int ncontrollers; + int n_shared_res; + int n_host_res; delaymgr_t delaymgr; scheduler_t sched; var jobmgr; diff --git a/src/lib/configobj.c b/src/lib/configobj.c index 1324dc8..6e421f8 100644 --- a/src/lib/configobj.c +++ b/src/lib/configobj.c @@ -48,13 +48,14 @@ static typeinfo_t xconfig_type = { .destroy = (obj_destroy_fn_t)xconfig_destroy, }; -xce_config_t* makeConfig(var configmgr, xce_config_t* c) +static xce_config_t* makeConfig_(var configmgr, xce_config_t* c) { xce_config_t* c1 = NULL; int rc = 0; rc = key_find(configmgr, (void*)c->name, (void**)&c1); if ((rc < 0) && (rc != -ENOENT)) { + errno = -rc; return NULL; } @@ -77,3 +78,14 @@ xce_config_t* makeConfig(var configmgr, xce_config_t* c) return c1; } + +xce_config_t* makeConfig(var configmgr, xce_config_t* c) +{ + xce_config_t* c1 = NULL; + + lock(configmgr, RTTI_LOCK_WRITE); + c1 = makeConfig_(configmgr, c); + unlock(configmgr); + + return c1; +} diff --git a/src/lib/group.c b/src/lib/group.c index 505c5d9..b4c70e2 100644 --- a/src/lib/group.c +++ b/src/lib/group.c @@ -156,7 +156,10 @@ static int group_init(meta_obj_t* obj, va_list ap) return -EINVAL; } + lock(mgr, RTTI_LOCK_WRITE); rc = key_add(mgr, (void*)g_name, (void*)g); + unlock(mgr); + if (rc < 0) { return rc; } @@ -210,7 +213,10 @@ xce_group_t* makeGroup(var mgr, xce_group_config_t* config, return NULL; } + lock(mgr, RTTI_LOCK_READ); rc = key_find(mgr, (void*)g_name, (void**)&g); + unlock(mgr); + if ((rc < 0) && (rc != -ENOENT)) { errno = -rc; return NULL; @@ -262,7 +268,6 @@ int groupmgr_init(var* mgr, int nslots, int delta) xce_group_config_t empty = { .name = "empty" }; xce_group_t* g = NULL; kvmgr_params_t params = { .key_type = KVMGR_KEY_STRING, - .dynamic = 1, .nslots = nslots, .delta = delta, .datakiller = destroy }; @@ -291,7 +296,9 @@ int groupmgr_add(var mgr, xce_group_t* g) { int rc = 0; + lock(mgr, RTTI_LOCK_WRITE); rc = key_add(mgr, (void*)g->name, (void*)g); + unlock(mgr); return rc; } @@ -301,7 +308,10 @@ xce_group_t* groupmgr_lock(var mgr, char* name, int op) int rc; xce_group_t* g = NULL; + lock(mgr, RTTI_LOCK_READ); rc = key_find(mgr, (void*)name, (void**)&g); + unlock(mgr); + if (rc <= 0) { return NULL; } @@ -311,7 +321,9 @@ xce_group_t* groupmgr_lock(var mgr, char* name, int op) void groupmgr_del(var mgr, xce_group_t* g) { + lock(mgr, RTTI_LOCK_WRITE); key_del(mgr, (void*)g->name); + unlock(mgr); return; } @@ -325,7 +337,10 @@ xce_group_t* get_group(var mgr, char* name) return NULL; } + lock(mgr, RTTI_LOCK_READ); rc = key_find(mgr, (void*)name, (void**)&g); + unlock(mgr); + if (rc <= 0) { return NULL; } @@ -381,105 +396,6 @@ int fmt_group(char* buf, int len, xce_group_t* g) return strlen(buf); } -xce_group_t* build_group(var mgr, xce_group_config_t* g_config, - char* default_group_name) -{ - xce_group_t* g = NULL; - char* g_name; - int i; - - if ((mgr == NULL) || (g_config == NULL)) { - errno = EINVAL; - return NULL; - } - - g_name = g_config->name; - if ((g_name == NULL) || (strlen(g_name) == 0)) { - g_name = default_group_name; - } - - if (g_name == NULL) { - return NULL; - } - - g = get_group(mgr, g_name); - if (g != NULL) { - for (i = 0; i < g->members.count; i++) { - xce_g_member_t* m = (xce_g_member_t*)xlist_get(&(g->members), i); - if (m == NULL) { - continue; - } - if (m->is_group == 0) { - FREEUP(m->u.name); - } - - FREEUP(m); - } - g->members.count = 0; - - for (i = 0; i < g->ex_members.count; i++) { - xce_g_member_t* m = (xce_g_member_t*)xlist_get(&(g->ex_members), i); - if (m == NULL) { - continue; - } - if (m->is_group == 0) { - FREEUP(m->u.name); - } - - FREEUP(m); - } - g->ex_members.count = 0; - g->dirty = 1; - } else { - g = (xce_group_t*)calloc(1, sizeof(*g)); - if (g == NULL) { - return NULL; - } - - xlist_init(&(g->affecteds), 4); - xlist_init(&(g->members), 4); - xlist_init(&(g->ex_members), 4); - g->name = makeString(g_name); - g->dirty = 0; - } - - for (i = 0; i < g_config->members.count; i++) { - char* g_name = g_config->members.data[i]; - xce_g_member_t* m = make_g_member(mgr, g_name); - - if (m == NULL) { - continue; - } - - xlist_store(&(g->members), m); - if (m->is_group) { - xlist_store(&(m->u.g->affecteds), g); - } - } - - for (i = 0; i < g_config->ex_members.count; i++) { - char* g_name = g_config->ex_members.data[i]; - xce_g_member_t* m = make_g_member(mgr, g_name); - - if (m == NULL) { - continue; - } - - xlist_store(&(g->ex_members), m); - if (m->is_group) { - xlist_store(&(m->u.g->affecteds), g); - } - } - - groupmgr_add(mgr, g); - - if (g->dirty) { - update_group_dirty(g); - } - - return g; -} - static int check_in_member(xce_g_member_t* m, va_list ap) { char* name = va_arg(ap, char*); diff --git a/src/lib/jobj.c b/src/lib/jobj.c index 61c4ea2..dee9c4b 100644 --- a/src/lib/jobj.c +++ b/src/lib/jobj.c @@ -1,6 +1,7 @@ #include #include #include +#include #include #include @@ -103,8 +104,7 @@ static void jobsub_table_fini() } typedef struct { - mylock_t lock; - uint64_t refcount; + atomic_ulong refcount; } job_submit_private_t; static size_t jobsub_datasize(va_list ap) @@ -154,7 +154,6 @@ static int jobsub_init(meta_obj_t* obj, va_list ap) return -ENOMEM; } - mylock_init(&(priv->lock)); priv->refcount = 1; return 1; @@ -176,7 +175,6 @@ static void jobsub_destroy(meta_obj_t* obj) dereference(jobsub->start_info); } - mylock_fini(&(priv->lock)); memset(priv, 0x0, sizeof(*priv)); } @@ -184,9 +182,7 @@ static int jobsub_reference(meta_obj_t* obj) { job_submit_private_t* priv = (job_submit_private_t*)RTTI_OBJ_TO_PRIV(obj); - mylock_lock(&(priv->lock)); - priv->refcount++; - mylock_unlock(&(priv->lock)); + atomic_fetch_add(&(priv->refcount), 1); return 1; } @@ -197,12 +193,9 @@ static void jobsub_dereference(meta_obj_t* obj) job_submit_data_t* data = RTTI_OBJ_TO_DATA(job_submit_data_t, obj); uint64_t count; - mylock_lock(&(priv->lock)); - priv->refcount--; - count = priv->refcount; - mylock_unlock(&(priv->lock)); + count = atomic_fetch_sub(&(priv->refcount), 1); - if (count == 0) { + if (count == 1) { mylock_lock(&(jobsub_table.lock)); key_del(jobsub_table.sub_table, data); mylock_unlock(&(jobsub_table.lock)); @@ -268,8 +261,7 @@ job_submit_data_t* makeJobsub(job_submit_data_t* subdata) } typedef struct { - mylock_t lock; - uint64_t refcount; + atomic_ulong refcount; } startinfo_private_t; static void startinfo_destroy(meta_obj_t* obj) @@ -305,7 +297,6 @@ static void startinfo_destroy(meta_obj_t* obj) dereference(startinfo->job_post_exec); } - mylock_fini(&(priv->lock)); memset(priv, 0x0, sizeof(*priv)); return; } @@ -327,7 +318,6 @@ static int startinfo_init(meta_obj_t* obj, va_list ap) return -EINVAL; } - mylock_init(&(priv->lock)); priv->refcount = 1; dst->exechome = makeString(src->exechome); @@ -371,9 +361,7 @@ static int startinfo_reference(meta_obj_t* obj) { startinfo_private_t* priv = (startinfo_private_t*)RTTI_OBJ_TO_PRIV(obj); - mylock_lock(&(priv->lock)); - priv->refcount++; - mylock_unlock(&(priv->lock)); + atomic_fetch_add(&(priv->refcount), 1); return 1; } @@ -384,16 +372,14 @@ static void startinfo_dereference(meta_obj_t* obj) startinfo_private_t* priv = (startinfo_private_t*)RTTI_OBJ_TO_PRIV(obj); uint64_t count; - mylock_lock(&(priv->lock)); - priv->refcount--; - count = priv->refcount; - mylock_unlock(&(priv->lock)); + count = atomic_fetch_sub(&(priv->refcount), 1); - if (count == 0) { + if (count == 1) { mylock_lock(&(startinfo_table.lock)); key_del(startinfo_table.startinfo_table, startinfo); mylock_unlock(&(startinfo_table.lock)); } + return; } @@ -531,8 +517,8 @@ static void startinfo_table_fini() } typedef struct { - my_rwlock_t lock; - uint64_t refcount; + my_rwlock_t lock; + atomic_ulong refcount; } job_private_t; static size_t job_datasize(va_list ap) @@ -606,9 +592,7 @@ static int job_reference(meta_obj_t* obj) { job_private_t* priv = (job_private_t*)RTTI_OBJ_TO_PRIV(obj); - my_rwlock_wlock(&(priv->lock)); - priv->refcount++; - my_rwlock_unlock(&(priv->lock)); + atomic_fetch_add(&(priv->refcount), 1); return 1; } @@ -619,12 +603,9 @@ static void job_dereference(meta_obj_t* obj) job_t* j = RTTI_OBJ_TO_DATA(job_t, obj); uint64_t count; - my_rwlock_wlock(&(priv->lock)); - priv->refcount--; - count = priv->refcount; - my_rwlock_unlock(&(priv->lock)); + count = atomic_fetch_sub(&(priv->refcount), 1); - if (count == 0) { + if (count == 1) { destroy_(j, (obj_destroy_fn_t)job_destroy); } @@ -674,12 +655,13 @@ static typeinfo_t job_type = { .unlock = (unlock_fn_t)job_unlock } } }; -job_t* makeJob(var jobmgr, xce_job_t* j) +static job_t* makeJob_(var jobmgr, xce_job_t* j) { job_t* job = NULL; - int rc = 0; + int rc = 0; + + rc = key_find(jobmgr, (void*)j->jobid, (void**)&job); - rc = key_find(jobmgr, (void *)j->jobid, (void **)&job); if ((rc < 0) && (rc != -ENOENT)) { errno = -rc; return NULL; @@ -689,10 +671,11 @@ job_t* makeJob(var jobmgr, xce_job_t* j) errno = EEXIST; return NULL; } - + job = makeInstance(&job_type, j); if (job != NULL) { - rc = key_add(jobmgr, (void *)job->job.jobid, (void **)job); + rc = key_add(jobmgr, (void*)job->job.jobid, (void**)job); + if (rc < 0) { destroy(job); @@ -700,6 +683,22 @@ job_t* makeJob(var jobmgr, xce_job_t* j) return NULL; } } + + return job; +} + +job_t* makeJob(var jobmgr, xce_job_t* j) +{ + job_t* job = NULL; + int rc = 0; + + lock(jobmgr, RTTI_LOCK_WRITE); + job = makeJob_(jobmgr, j); + rc = errno; + unlock(jobmgr); + + errno = rc; + return job; } diff --git a/src/lib/kvmgr.c b/src/lib/kvmgr.c index 194dbff..0525295 100644 --- a/src/lib/kvmgr.c +++ b/src/lib/kvmgr.c @@ -19,7 +19,6 @@ typedef struct { typedef struct { kvmgr_key_type_t key_type; - unsigned int dynamic : 1; uint32_t nslots; var kvtab; var data; @@ -34,10 +33,11 @@ static size_t xkvmgr_datasize(va_list ap) static int xkvmgr_init(meta_obj_t* obj, va_list ap) { - xkvmgr_t* kvm = RTTI_OBJ_TO_DATA(xkvmgr_t, obj); - kvmgr_params_t* params = va_arg(ap, kvmgr_params_t*); - kv_params_t kv_params; - xzarray_params_t data_params; + xkvmgr_t* kvm = RTTI_OBJ_TO_DATA(xkvmgr_t, obj); + kvmgr_params_t* params = va_arg(ap, kvmgr_params_t*); + kv_params_t kv_params; + xzarray_params_t data_params; + xkvmgr_private_t* priv = RTTI_OBJ_TO_PRIV(obj); if ((params == NULL) || ((params->key_type != KVMGR_KEY_INT) @@ -76,20 +76,16 @@ static int xkvmgr_init(meta_obj_t* obj, va_list ap) kvm->nslots = params->nslots; kvm->key_type = params->key_type; - kvm->dynamic = params->dynamic; - if (params->dynamic) { - xkvmgr_private_t* priv = RTTI_OBJ_TO_PRIV(obj); - - my_rwlock_init(&(priv->lock)); - } + my_rwlock_init(&(priv->lock)); return 1; } static void xkvmgr_destroy(meta_obj_t* obj) { - xkvmgr_t* kvm = RTTI_OBJ_TO_DATA(xkvmgr_t, obj); + xkvmgr_t* kvm = RTTI_OBJ_TO_DATA(xkvmgr_t, obj); + xkvmgr_private_t* priv = RTTI_OBJ_TO_PRIV(obj); if (kvm->data != NULL) { destroy(kvm->data); @@ -99,11 +95,7 @@ static void xkvmgr_destroy(meta_obj_t* obj) destroy(kvm->kvtab); } - if (kvm->dynamic) { - xkvmgr_private_t* priv = RTTI_OBJ_TO_PRIV(obj); - - my_rwlock_fini(&(priv->lock)); - } + my_rwlock_fini(&(priv->lock)); memset(kvm, 0x0, sizeof(*kvm)); return; @@ -116,10 +108,6 @@ static int xkvmgr_add(meta_obj_t* obj, void* key, void* value) xkvmgr_private_t* priv = RTTI_OBJ_TO_PRIV(obj); int idx; - if (kvm->dynamic) { - my_rwlock_wlock(&(priv->lock)); - } - rc = vstore(kvm->data, value); if (rc >= 0) { @@ -134,10 +122,6 @@ static int xkvmgr_add(meta_obj_t* obj, void* key, void* value) } } - if (kvm->dynamic) { - my_rwlock_unlock(&(priv->lock)); - } - return rc; } @@ -150,10 +134,6 @@ static int xkvmgr_find(meta_obj_t* obj, void* key, void** value) *value = NULL; - if (kvm->dynamic) { - my_rwlock_rlock(&(priv->lock)); - } - rc = key_find(kvm->kvtab, key, &data); if (rc >= 0) { int idx = (int)(long)data; @@ -165,10 +145,6 @@ static int xkvmgr_find(meta_obj_t* obj, void* key, void** value) } } - if (kvm->dynamic) { - my_rwlock_unlock(&(priv->lock)); - } - return rc; } @@ -184,15 +160,9 @@ static int xkvmgr_make_iter(iterator_t* iter, va_list ap) return -ENOMEM; } - if (kvm->dynamic) { - my_rwlock_rlock(&(priv->lock)); - } rc = iter_make(it, kvm->data); if (rc < 0) { FREEUP(it); - if (kvm->dynamic) { - my_rwlock_unlock(&(priv->lock)); - } return rc; } @@ -230,10 +200,6 @@ static void xkvmgr_iter_destroy(iterator_t* iter) iter->data = NULL; - if (kvm->dynamic) { - my_rwlock_unlock(&(priv->lock)); - } - return; } @@ -243,15 +209,8 @@ static int xkvmgr_get_vec(meta_obj_t* obj, int count, int* idxs, void** data) xkvmgr_private_t* priv = RTTI_OBJ_TO_PRIV(obj); int rc = 0; - if (kvm->dynamic) { - my_rwlock_rlock(&(priv->lock)); - } rc = vget_vec(kvm->data, count, idxs, data); - if (kvm->dynamic) { - my_rwlock_unlock(&(priv->lock)); - } - return rc; } @@ -261,15 +220,8 @@ static int xkvmgr_get(meta_obj_t* obj, int idx, void** data) xkvmgr_private_t* priv = RTTI_OBJ_TO_PRIV(obj); int rc = 0; - if (kvm->dynamic) { - my_rwlock_rlock(&(priv->lock)); - } rc = vget(kvm->data, idx, data); - if (kvm->dynamic) { - my_rwlock_unlock(&(priv->lock)); - } - return rc; } @@ -279,15 +231,8 @@ static int xkvmgr_put_vec(meta_obj_t* obj, int count, int* idxs, void** data) xkvmgr_private_t* priv = RTTI_OBJ_TO_PRIV(obj); int rc = 0; - if (kvm->dynamic) { - my_rwlock_wlock(&(priv->lock)); - } rc = vput_vec(kvm->data, count, idxs, data); - if (kvm->dynamic) { - my_rwlock_unlock(&(priv->lock)); - } - return rc; } @@ -297,18 +242,39 @@ static int xkvmgr_put(meta_obj_t* obj, int idx, void* data) xkvmgr_private_t* priv = RTTI_OBJ_TO_PRIV(obj); int rc = 0; - if (kvm->dynamic) { - my_rwlock_wlock(&(priv->lock)); - } rc = vput(kvm->data, idx, data); - if (kvm->dynamic) { - my_rwlock_unlock(&(priv->lock)); + return rc; +} + +static int xkvmgr_lock(meta_obj_t* obj, uint64_t option) +{ + xkvmgr_private_t* priv = (xkvmgr_private_t*)RTTI_OBJ_TO_PRIV(obj); + int rc = -EINVAL; + + switch (option) { + case RTTI_LOCK_READ: + rc = my_rwlock_rlock(&(priv->lock)); + break; + case RTTI_LOCK_WRITE: + rc = my_rwlock_wlock(&(priv->lock)); + break; + default: + break; } return rc; } +static void xkvmgr_unlock(meta_obj_t* obj) +{ + xkvmgr_private_t* priv = (xkvmgr_private_t*)RTTI_OBJ_TO_PRIV(obj); + + my_rwlock_unlock(&(priv->lock)); + + return; +} + typeinfo_t xkvmgr_type = { .name = "xkvmgr", .privateSize = sizeof(xkvmgr_private_t), @@ -330,5 +296,7 @@ typeinfo_t xkvmgr_type = { .next = (iter_next_fn_t)xkvmgr_iter_next, .reset = (iter_reset_fn_t)xkvmgr_iter_reset, .destroy - = (iter_destroy_fn_t)xkvmgr_iter_destroy } } + = (iter_destroy_fn_t)xkvmgr_iter_destroy }, + .lock = &(intf_lock_t) { .lock = (lock_fn_t)xkvmgr_lock, + .unlock = (unlock_fn_t)xkvmgr_unlock } } }; diff --git a/src/lib/qobj.c b/src/lib/qobj.c index 04b5749..334b46e 100644 --- a/src/lib/qobj.c +++ b/src/lib/qobj.c @@ -1,6 +1,7 @@ #include #include #include +#include #include #include @@ -10,8 +11,8 @@ #include "xstring.h" typedef struct { - my_rwlock_t lock; - uint64_t refcount; + my_rwlock_t lock; + atomic_ulong refcount; } queue_private_t; static size_t queue_datasize(va_list ap) @@ -42,6 +43,7 @@ static int queue_init(meta_obj_t* obj, va_list ap) MOVE_PTR(dst->queue.hosts, src->hosts); dst->queue.nthresholds = src->nthresholds; MOVE_PTR(dst->queue.thresholds, src->thresholds); + src->nthresholds = 0; MOVE_PTR(dst->queue.runwindow, src->runwindow); MOVE_PTR(dst->queue.dispatchwindow, src->dispatchwindow); @@ -95,9 +97,7 @@ static int queue_reference(meta_obj_t* obj) { queue_private_t* priv = (queue_private_t*)RTTI_OBJ_TO_PRIV(obj); - my_rwlock_wlock(&(priv->lock)); - priv->refcount++; - my_rwlock_unlock(&(priv->lock)); + atomic_fetch_add(&(priv->refcount), 1); return 1; } @@ -108,12 +108,9 @@ static void queue_dereference(meta_obj_t* obj) queue_t* w = RTTI_OBJ_TO_DATA(queue_t, obj); uint64_t count = 0; - my_rwlock_wlock(&(priv->lock)); - priv->refcount--; - count = priv->refcount; - my_rwlock_unlock(&(priv->lock)); + count = atomic_fetch_sub(&(priv->refcount), 1); - if (count == 0) { + if (count == 1) { destroy_(w, (obj_destroy_fn_t)queue_destroy); } @@ -163,12 +160,13 @@ static typeinfo_t queue_type = { .unlock = (unlock_fn_t)queue_unlock } } }; -queue_t* makeQueue(var qmgr, xce_queue_t* q) +static queue_t* makeQueue_(var qmgr, xce_queue_t* q) { queue_t* queue = NULL; int rc = 0; rc = key_find(qmgr, (void*)q->queue, (void**)&queue); + if ((rc < 0) && (rc != -ENOENT)) { errno = -rc; return NULL; @@ -181,8 +179,8 @@ queue_t* makeQueue(var qmgr, xce_queue_t* q) queue = makeInstance(&queue_type, q); if (queue != NULL) { - reference(queue); rc = key_add(qmgr, (void*)q->queue, (void*)queue); + if (rc < 0) { destroy(queue); @@ -193,3 +191,17 @@ queue_t* makeQueue(var qmgr, xce_queue_t* q) return queue; } + +queue_t* makeQueue(var qmgr, xce_queue_t* q) +{ + queue_t* queue = NULL; + int rc; + + lock(qmgr, RTTI_LOCK_WRITE); + queue = makeQueue_(qmgr, q); + rc = errno; + unlock(qmgr); + errno = rc; + + return queue; +} diff --git a/src/lib/resobj.c b/src/lib/resobj.c index 5821797..fb9358c 100644 --- a/src/lib/resobj.c +++ b/src/lib/resobj.c @@ -1,6 +1,7 @@ #include #include #include +#include #include #include @@ -10,8 +11,7 @@ #include "xstring.h" typedef struct { - mylock_t lock; - uint64_t refcount; + atomic_ulong refcount; } resobj_private_t; static size_t resobj_datasize(va_list ap) @@ -52,7 +52,6 @@ static int resobj_init(meta_obj_t* obj, va_list ap) dst->hbased_defined = NULL; #endif - mylock_init(&(priv->lock)); priv->refcount = 1; return 1; @@ -66,7 +65,6 @@ static void resobj_destroy(meta_obj_t* obj) dereference(r->name); dereference(r->desc); - mylock_fini(&(priv->lock)); memset(priv, 0x0, sizeof(*priv)); return; @@ -76,9 +74,7 @@ static int resobj_reference(meta_obj_t* obj) { resobj_private_t* priv = (resobj_private_t*)RTTI_OBJ_TO_PRIV(obj); - mylock_lock(&(priv->lock)); - priv->refcount++; - mylock_unlock(&(priv->lock)); + atomic_fetch_add(&(priv->refcount), 1); return 1; } @@ -89,12 +85,9 @@ static void resobj_dereference(meta_obj_t* obj) xce_resource_t* w = RTTI_OBJ_TO_DATA(xce_resource_t, obj); uint64_t count = 0; - mylock_lock(&(priv->lock)); - priv->refcount--; - count = priv->refcount; - mylock_unlock(&(priv->lock)); + count = atomic_fetch_sub(&(priv->refcount), 1); - if (count == 0) { + if (count == 1) { destroy_(w, (obj_destroy_fn_t)resobj_destroy); } @@ -112,12 +105,13 @@ static typeinfo_t resobj_type .reference = (reference_fn_t)resobj_reference, .dereference = (dereference_fn_t)resobj_dereference } } }; -xce_resource_t* makeResource(var resmgr, xce_resource_t* r) +static xce_resource_t* makeResource_(var resmgr, xce_resource_t* r) { xce_resource_t* resobj = NULL; int rc; rc = key_find(resmgr, (void*)r->name, (void**)&resobj); + if ((rc < 0) && (rc != -ENOENT)) { errno = -rc; @@ -131,8 +125,8 @@ xce_resource_t* makeResource(var resmgr, xce_resource_t* r) resobj = makeInstance(&resobj_type, r); if (resobj != NULL) { - reference(resobj); rc = key_add(resmgr, (void*)resobj->name, (void*)resobj); + if (rc < 0) { destroy(resobj); @@ -142,3 +136,14 @@ xce_resource_t* makeResource(var resmgr, xce_resource_t* r) } return resobj; } + +xce_resource_t* makeResource(var resmgr, xce_resource_t* r) +{ + xce_resource_t* r1 = NULL; + + lock(resmgr, RTTI_LOCK_WRITE); + r1 = makeResource_(resmgr, r); + unlock(resmgr); + + return r1; +} diff --git a/src/lib/resreq.c b/src/lib/resreq.c index 9aaf559..25bf5e5 100644 --- a/src/lib/resreq.c +++ b/src/lib/resreq.c @@ -169,11 +169,14 @@ int parse_resreq_primary(xce_primary_resreq_t* primary, char* resreqString, } xlist_init(&(primary->rusages), 8); - rc = parse_rusage(&(primary->rusages), data.rusageString, errmsg); - if (rc < 0) { - kill_resreq_data(&data); - return rc; + if ((data.rusageString != NULL) && (strlen(data.rusageString) != 0)) { + rc = parse_rusage(&(primary->rusages), data.rusageString, errmsg); + if (rc < 0) { + kill_resreq_data(&data); + + return rc; + } } /* FIXME: add others later */ @@ -451,7 +454,10 @@ int calc_wanted_global_resvals(xvec_idval_t* resvals, var resmgr, continue; } + lock(resmgr, RTTI_LOCK_READ); rc = key_find(resmgr, (void*)v->res, (void**)&r); + unlock(resmgr); + if (rc <= 0) { continue; } @@ -507,7 +513,10 @@ int calc_wanted_host_resvals(xvec_idval_t* resvals, var resmgr, continue; } + lock(resmgr, RTTI_LOCK_READ); rc = key_find(resmgr, (void*)v->res, (void**)&r); + unlock(resmgr); + if (rc <= 0) { continue; } diff --git a/src/lib/testsuite/test_jobj.c b/src/lib/testsuite/test_jobj.c index a50f195..6ddc979 100644 --- a/src/lib/testsuite/test_jobj.c +++ b/src/lib/testsuite/test_jobj.c @@ -208,7 +208,6 @@ static void test_job_create(void) { int rc; kvmgr_params_t jobmgr_params = { .key_type = KVMGR_KEY_INT, - .dynamic = 1, .nslots = 32, .delta = 32, .datakiller = destroy }; diff --git a/src/lib/testsuite/test_kvmgr.c b/src/lib/testsuite/test_kvmgr.c index 292cae9..d663a77 100644 --- a/src/lib/testsuite/test_kvmgr.c +++ b/src/lib/testsuite/test_kvmgr.c @@ -15,7 +15,6 @@ typedef struct { static void test_kvmgr_create(void) { kvmgr_params_t params = { .key_type = KVMGR_KEY_STRING, - .dynamic = 0, .nslots = 4, .delta = 4, .datakiller = NULL }; @@ -46,7 +45,6 @@ static void my_user_killer(void* data) static void test_kvmgr_add(void) { kvmgr_params_t params = { .key_type = KVMGR_KEY_STRING, - .dynamic = 0, .nslots = 4, .delta = 4, .datakiller = my_user_killer }; @@ -79,7 +77,6 @@ static void test_kvmgr_add(void) static void test_kvmgr_iter(void) { kvmgr_params_t params = { .key_type = KVMGR_KEY_INT, - .dynamic = 0, .nslots = 4, .delta = 4, .datakiller = NULL }; @@ -116,7 +113,7 @@ static void test_kvmgr_iter(void) TEST_ASSERT_TRUE(i == 10); iter_destroy(&it); - + destroy(kvmgr); } diff --git a/src/lib/testsuite/test_resreq.c b/src/lib/testsuite/test_resreq.c index 08b515d..49f4fd6 100644 --- a/src/lib/testsuite/test_resreq.c +++ b/src/lib/testsuite/test_resreq.c @@ -212,6 +212,26 @@ void test_resreq_primary(void) } } +void test_resreq_primary2(void) +{ + char* resreqString + = "select[type==any] "; + char* errmsg = NULL; + int rc = 0; + xce_primary_resreq_t resreq; + + rc = parse_resreq_primary(&resreq, resreqString, &errmsg); + TEST_ASSERT_TRUE(rc >= 0); + if (rc >= 0) { + int i; + xlist_t* rusages = &(resreq.rusages); + + TEST_ASSERT_TRUE(rusages->count == 0); + + xlist_fini(rusages); + } +} + void test_resreq1(void) { char* resreqString @@ -337,6 +357,7 @@ void test_resreq(void) RUN_TEST(test_parse_rusage1); RUN_TEST(test_parse_rusage2); RUN_TEST(test_resreq_primary); + RUN_TEST(test_resreq_primary2); RUN_TEST(test_resreq1); RUN_TEST(test_resreq2); RUN_TEST(test_resreq3); diff --git a/src/lib/wobj.c b/src/lib/wobj.c index da8fdf9..57fdfb2 100644 --- a/src/lib/wobj.c +++ b/src/lib/wobj.c @@ -1,6 +1,7 @@ #include #include #include +#include #include #include @@ -10,8 +11,8 @@ #include "xstring.h" typedef struct { - my_rwlock_t lock; - uint64_t refcount; + my_rwlock_t lock; + atomic_ulong refcount; } worker_private_t; static size_t worker_datasize(va_list ap) @@ -91,9 +92,7 @@ static int worker_reference(meta_obj_t* obj) { worker_private_t* priv = (worker_private_t*)RTTI_OBJ_TO_PRIV(obj); - my_rwlock_wlock(&(priv->lock)); - priv->refcount++; - my_rwlock_unlock(&(priv->lock)); + atomic_fetch_add(&(priv->refcount), 1); return 1; } @@ -104,12 +103,9 @@ static void worker_dereference(meta_obj_t* obj) worker_t* w = RTTI_OBJ_TO_DATA(worker_t, obj); uint64_t count = 0; - my_rwlock_wlock(&(priv->lock)); - priv->refcount--; - count = priv->refcount; - my_rwlock_unlock(&(priv->lock)); + count = atomic_fetch_sub(&(priv->refcount), 1); - if (count == 0) { + if (count == 1) { destroy_(w, (obj_destroy_fn_t)worker_destroy); } @@ -159,12 +155,13 @@ static typeinfo_t worker_type = { .unlock = (unlock_fn_t)worker_unlock } } }; -worker_t* makeWorker(var workermgr, xce_worker_t* w) +static worker_t* makeWorker_(var workermgr, xce_worker_t* w) { worker_t* worker = NULL; int rc = 0; rc = key_find(workermgr, (void*)w->name, (void**)&worker); + if ((rc < 0) && (rc != -ENOENT)) { errno = -rc; @@ -178,8 +175,8 @@ worker_t* makeWorker(var workermgr, xce_worker_t* w) worker = makeInstance(&worker_type, w); if (worker != NULL) { - reference(worker); rc = key_add(workermgr, (void*)worker->host.name, (void*)worker); + if (rc < 0) { destroy(worker); @@ -187,5 +184,17 @@ worker_t* makeWorker(var workermgr, xce_worker_t* w) return NULL; } } + + return worker; +} + +worker_t* makeWorker(var workermgr, xce_worker_t* w) +{ + worker_t* worker = NULL; + + lock(workermgr, RTTI_LOCK_WRITE); + worker = makeWorker_(workermgr, w); + unlock(workermgr); + return worker; } diff --git a/src/lib/xcelimit.c b/src/lib/xcelimit.c index 86540ad..65e12eb 100644 --- a/src/lib/xcelimit.c +++ b/src/lib/xcelimit.c @@ -377,7 +377,10 @@ int is_host_based_limit(var resmgr, xce_limit_t* limit) return -EINVAL; } + lock(resmgr, RTTI_LOCK_READ); rc = key_find(resmgr, (void*)limit->val.resname, (void**)&r); + unlock(resmgr); + if (rc <= 0) { return -ENOENT; } @@ -420,7 +423,10 @@ int add_vres(xce_resource_t** pvres, var resmgr, char* name, int global) vres->flags |= XCE_RESF_DEFAULT_HOST; } + lock(resmgr, RTTI_LOCK_WRITE); rc = key_add(resmgr, (void*)vres->name, (void*)vres); + unlock(resmgr); + if (rc < 0) { return rc; } @@ -460,7 +466,10 @@ int instantiate_limit(xce_vres_t** pvres, var resmgr, xce_limit_t* limit, return rc; } + lock(resmgr, RTTI_LOCK_READ); rc = key_find(resmgr, (void*)vresname, (void**)&(vres->vres)); + unlock(resmgr); + if (rc < 0) { return rc; } @@ -475,7 +484,10 @@ int instantiate_limit(xce_vres_t** pvres, var resmgr, xce_limit_t* limit, FREEUP(vresname); + lock(resmgr, RTTI_LOCK_READ); rc = key_find(resmgr, (void*)limit->val.resname, (void**)&(vres->res)); + unlock(resmgr); + if (rc < 0) { return rc; } diff --git a/src/lib/xstring.c b/src/lib/xstring.c index 2a7b406..9f17567 100644 --- a/src/lib/xstring.c +++ b/src/lib/xstring.c @@ -1,6 +1,7 @@ #include #include #include +#include #include #include @@ -16,10 +17,9 @@ static struct { } xstring_table; typedef struct { - mylock_t lock; - uint64_t refcount; - uint32_t hashcode; - size_t len; + atomic_ulong refcount; + uint32_t hashcode; + size_t len; } xstring_private_t; static typeinfo_t xstring_type; @@ -78,7 +78,6 @@ static int xstring_init(meta_obj_t* obj, va_list ap) strcpy(dest, src); - mylock_init(&(priv->lock)); priv->refcount = 1; priv->hashcode = hash_string(src); priv->len = strlen(src); @@ -90,7 +89,6 @@ static void xstring_destroy(meta_obj_t* obj) { xstring_private_t* priv = (xstring_private_t*)RTTI_OBJ_TO_PRIV(obj); - mylock_fini(&(priv->lock)); memset(priv, 0x0, sizeof(*priv)); return; @@ -100,9 +98,7 @@ static int xstring_reference(meta_obj_t* obj) { xstring_private_t* priv = (xstring_private_t*)RTTI_OBJ_TO_PRIV(obj); - mylock_lock(&(priv->lock)); - priv->refcount++; - mylock_unlock(&(priv->lock)); + atomic_fetch_add(&(priv->refcount), 1); return 0; } @@ -113,13 +109,9 @@ static void xstring_dereference(meta_obj_t* obj) char* v = RTTI_OBJ_TO_DATA(char, obj); uint64_t count; - mylock_lock(&(priv->lock)); - priv->refcount--; + count = atomic_fetch_sub(&(priv->refcount), 1); - count = priv->refcount; - mylock_unlock(&(priv->lock)); - - if (count == 0) { + if (count == 1) { mylock_lock(&(xstring_table.lock)); key_del(xstring_table.xstr_table, v); mylock_unlock(&(xstring_table.lock)); diff --git a/src/lib/zarray.c b/src/lib/zarray.c index c6a136e..4f7b377 100644 --- a/src/lib/zarray.c +++ b/src/lib/zarray.c @@ -456,7 +456,6 @@ static int xzarray_store(meta_obj_t* obj, void* data) xzarray_private_t* priv = (xzarray_private_t*)RTTI_OBJ_TO_PRIV(obj); int rc = 0; - mylock_lock(&(priv->lock)); rc = xzarray_ensure_items(a, a->count + 1); if (rc >= 0) { int idx, i; @@ -484,7 +483,6 @@ static int xzarray_store(meta_obj_t* obj, void* data) a->pos = (rc + 1) % a->caps; } } - mylock_unlock(&(priv->lock)); return rc; } @@ -499,7 +497,6 @@ static int xzarray_get_vec(meta_obj_t* obj, int count, int* idxs, void** data) return -EINVAL; } - mylock_lock(&(priv->lock)); for (i = 0; i < count; i++) { int idx = idxs[i]; @@ -516,7 +513,6 @@ static int xzarray_get_vec(meta_obj_t* obj, int count, int* idxs, void** data) data[i] = a->data[idx].data; } } - mylock_unlock(&(priv->lock)); return rc; } @@ -537,7 +533,6 @@ static int xzarray_put_vec(meta_obj_t* obj, int count, int* idxs, void** data) return -EINVAL; } - mylock_lock(&(priv->lock)); maxid = 0; rc = 0; @@ -570,8 +565,6 @@ static int xzarray_put_vec(meta_obj_t* obj, int count, int* idxs, void** data) } } - mylock_unlock(&(priv->lock)); - return rc; } @@ -588,7 +581,6 @@ static int xzarray_clear(meta_obj_t* obj, int idx) xzarray_private_t* priv = (xzarray_private_t*)RTTI_OBJ_TO_PRIV(obj); int rc = -EINVAL; - mylock_lock(&(priv->lock)); if (idx < a->caps) { rc = 0; if (a->data[idx].inuse) { @@ -601,7 +593,6 @@ static int xzarray_clear(meta_obj_t* obj, int idx) a->data[idx].data = NULL; a->pos = idx; /* we can find the available idx next time */ } - mylock_unlock(&(priv->lock)); return rc; } @@ -610,7 +601,6 @@ static int xzarray_make_iter(iterator_t* iter, va_list ap) { xzarray_private_t* priv = (xzarray_private_t*)RTTI_OBJ_TO_PRIV(iter->obj); - mylock_lock(&(priv->lock)); iter->data = (void*)(long)0; return RTTI_ITER_OK; @@ -640,12 +630,26 @@ static int xzarray_iter_reset(iterator_t* iter) return RTTI_ITER_OK; } +static int xzarray_lock(meta_obj_t* obj, uint64_t option) +{ + xzarray_private_t* priv = (xzarray_private_t*)RTTI_OBJ_TO_PRIV(obj); + + return mylock_lock(&(priv->lock)); +} + +static void xzarray_unlock(meta_obj_t* obj) +{ + xzarray_private_t* priv = (xzarray_private_t*)RTTI_OBJ_TO_PRIV(obj); + + mylock_unlock(&(priv->lock)); + + return; +} + static void xzarray_iter_destroy(iterator_t* iter) { xzarray_private_t* priv = (xzarray_private_t*)RTTI_OBJ_TO_PRIV(iter->obj); - mylock_unlock(&(priv->lock)); - return; } @@ -667,5 +671,7 @@ typeinfo_t xzarray_type = { .next = (iter_next_fn_t)xzarray_iter_next, .reset = (iter_reset_fn_t)xzarray_iter_reset, .destroy - = (iter_destroy_fn_t)xzarray_iter_destroy } } + = (iter_destroy_fn_t)xzarray_iter_destroy }, + .lock = &(intf_lock_t) { .lock = (lock_fn_t)xzarray_lock, + .unlock = (unlock_fn_t)xzarray_unlock } } }; diff --git a/src/meson.build b/src/meson.build index 88d96bc..76d989c 100644 --- a/src/meson.build +++ b/src/meson.build @@ -69,3 +69,6 @@ configure_file( install_dir: 'bin', install_mode: 'rwxr-xr-x' ) + +install_data('docs/xce.md', + install_dir: 'share/docs') diff --git a/src/scheduler/include/plugin.h b/src/scheduler/include/plugin.h index 77d8dbd..89be6e9 100644 --- a/src/scheduler/include/plugin.h +++ b/src/scheduler/include/plugin.h @@ -6,8 +6,8 @@ #include #include "decision.h" -#include "sched/types.h" #include "resreq.h" +#include "sched/types.h" typedef void* plugin_data_t; @@ -79,5 +79,6 @@ extern int plugin_job_election(void*, xce_sched_job_t*); extern int register_fcfs(void*); extern int register_limit(void*); extern int register_seqjob_allocator(void*); +extern int register_parjob_allocator(void*); #endif diff --git a/src/scheduler/src/plugin.c b/src/scheduler/src/plugin.c index 449f194..c9f19b3 100644 --- a/src/scheduler/src/plugin.c +++ b/src/scheduler/src/plugin.c @@ -28,7 +28,10 @@ int plugin_register(void* arg1, void* arg2) return rc; } + lock(sched->plugins, RTTI_LOCK_WRITE); rc = key_add(sched->plugins, (void*)plugin->name, (void*)plugin); + unlock(sched->plugins); + if (rc >= 0) { plugin->plugin_id = rc; } @@ -50,7 +53,10 @@ void plugin_unregister(void* arg) return; } + lock(sched->plugins, RTTI_LOCK_WRITE); key_del(sched->plugins, plugin->name); + unlock(sched->plugins); + if (plugin->fini_fn != NULL) { plugin->fini_fn(plugin); } @@ -163,7 +169,7 @@ int plugin_queue_update(void* arg, xce_sched_queue_t* queue) } iter_destroy(&it); - + return rc; } diff --git a/src/scheduler/src/plugins/limit.c b/src/scheduler/src/plugins/limit.c index f34fcd4..8a170e3 100644 --- a/src/scheduler/src/plugins/limit.c +++ b/src/scheduler/src/plugins/limit.c @@ -21,14 +21,12 @@ static int limit_priv_data_init(limit_plugin_data_t* data) { kvmgr_params_t limits_params = { .key_type = KVMGR_KEY_STRING, - .dynamic = 1, .nslots = 1021, .delta = 64, .datakiller = (destroy_fn_t)kill_limit, }; kvmgr_params_t instances_params = { .key_type = KVMGR_KEY_STRING, - .dynamic = 1, .nslots = 1021, .delta = 64, .datakiller = (destroy_fn_t)kill_limit, @@ -63,13 +61,16 @@ static int limit_update_config(xce_sched_plugin_t* plugin, int rc; xce_limit_t* limit; - rc = build_limit(&limit, (sched->ugmgr), (sched->qgmgr), - (sched->pgmgr), (sched->hgmgr), limits[i]); + rc = build_limit(&limit, (sched->ugmgr), (sched->qgmgr), (sched->pgmgr), + (sched->hgmgr), limits[i]); if (rc < 0) { return rc; } + lock(privdata->limits, RTTI_LOCK_WRITE); rc = key_add(privdata->limits, (void*)limit->name, (void*)limit); + unlock(privdata->limits); + if (rc < 0) { kill_limit(limit); @@ -182,7 +183,10 @@ static int match_job_global_limit_(xce_sched_plugin_t* plugin, return rc; } + lock(privdata->instances, RTTI_LOCK_READ); rc = key_find(privdata->instances, (void*)vresname, (void*)&vres); + unlock(privdata->instances); + if (rc <= 0) { return rc; } @@ -196,7 +200,10 @@ static int match_job_global_limit_(xce_sched_plugin_t* plugin, return rc; } + lock(privdata->instances, RTTI_LOCK_WRITE); rc = key_add(privdata->instances, (void*)vres->vres->name, vres); + unlock(privdata->instances); + if (rc < 0) { return rc; } @@ -324,7 +331,10 @@ static int match_job_host_limit_(xce_sched_plugin_t* plugin, return rc; } + lock(privdata->instances, RTTI_LOCK_READ); rc = key_find(privdata->instances, (void*)vresname, (void*)&vres); + unlock(privdata->instances); + if (rc < 0) { return rc; } @@ -337,7 +347,9 @@ static int match_job_host_limit_(xce_sched_plugin_t* plugin, return rc; } + lock(privdata->instances, RTTI_LOCK_WRITE); rc = key_add(privdata->instances, vres->vres->name, vres); + unlock(privdata->instances); if (rc < 0) { return rc; } @@ -398,8 +410,8 @@ static int build_limit_vres_requests(xce_scheduler_t* sched, static int limit_prepare_allocation(void* arg, xce_sched_ctx_t* ctx) { - xce_sched_plugin_t* plugin = (xce_sched_plugin_t*)arg; - xce_sched_worker_t* worker = NULL; + xce_sched_plugin_t* plugin = (xce_sched_plugin_t*)arg; + xce_sched_worker_t* worker = NULL; int rc = 0; limit_plugin_data_t* privdata = NULL; iterator_t it; @@ -439,7 +451,7 @@ static int limit_prepare_allocation(void* arg, xce_sched_ctx_t* ctx) } iter_destroy(&it); - + return rc; } diff --git a/src/scheduler/src/plugins/seqjob_allocator.c b/src/scheduler/src/plugins/seqjob_allocator.c index dfd4487..16ab6ce 100644 --- a/src/scheduler/src/plugins/seqjob_allocator.c +++ b/src/scheduler/src/plugins/seqjob_allocator.c @@ -82,7 +82,10 @@ static int seqjob_prepare_allocation(void* arg, xce_sched_ctx_t* ctx) } /* add slots into requested resources */ + lock(sched->resmgr, RTTI_LOCK_READ); rc = key_find(sched->resmgr, (void*)"slots", (void**)&r); + unlock(sched->resmgr); + if (rc < 0) { return rc; } diff --git a/src/scheduler/src/resource.c b/src/scheduler/src/resource.c index 4fcd4d6..345d866 100644 --- a/src/scheduler/src/resource.c +++ b/src/scheduler/src/resource.c @@ -301,7 +301,10 @@ static int sched_update_res_accounts(xce_scheduler_t* sched, zarray_t* array, xce_resource_t* r = NULL; int rc = 0; + lock(sched->resmgr, RTTI_LOCK_READ); rc = key_find(sched->resmgr, (void*)acct->resname, (void**)&r); + unlock(sched->resmgr); + if (rc < 0) { return rc; } @@ -315,7 +318,9 @@ static int sched_update_res_accounts(xce_scheduler_t* sched, zarray_t* array, return rc; } + lock(sched->resmgr, RTTI_LOCK_READ); rc = key_find(sched->resmgr, (void*)acct->resname, (void**)&r); + unlock(sched->resmgr); if (rc < 0) { return rc; } diff --git a/src/scheduler/src/scheduler.c b/src/scheduler/src/scheduler.c index 0aaacb0..ec34e26 100644 --- a/src/scheduler/src/scheduler.c +++ b/src/scheduler/src/scheduler.c @@ -16,6 +16,7 @@ #include "scheduler.h" #include "utils.h" #include "xcealloc.h" +#include "xstring.h" static void* sched_job_picker(void* arg) { @@ -24,7 +25,10 @@ static void* sched_job_picker(void* arg) char* job_pick_plugin = "xce_sched_fcfs"; int rc = 0; + lock(sched->plugins, RTTI_LOCK_READ); rc = key_find(sched->plugins, job_pick_plugin, (void**)&job_picker); + unlock(sched->plugins); + if (rc < 0) { errno = -rc; @@ -290,6 +294,7 @@ static int sched_register_plugins(xce_scheduler_t* sched) register_fcfs(sched); register_limit(sched); register_seqjob_allocator(sched); + register_parjob_allocator(sched); return 0; } @@ -297,7 +302,6 @@ static int sched_register_plugins(xce_scheduler_t* sched) static int sched_plugin_init(xce_scheduler_t* sched) { kvmgr_params_t params = { .key_type = KVMGR_KEY_STRING, - .dynamic = 1, .nslots = 16, .delta = 8, .datakiller = plugin_unregister }; @@ -317,36 +321,30 @@ int sched_init(xce_scheduler_t* sched, xce_client_config_t* config, int ndecisions = 5; /* FIXME: make this configurable */ int njobs = 1; /* FIXME: make this configurable */ kvmgr_params_t jobs_params = { .key_type = KVMGR_KEY_INT, - .dynamic = 1, .nslots = 5381, /* FIXME */ .delta = 1024, .datakiller = kill_sched_job }; kvmgr_params_t queues_params = { .key_type = KVMGR_KEY_STRING, - .dynamic = 0, .nslots = 59, .delta = 16, .datakiller = (destroy_fn_t)kill_sched_queue }; kvmgr_params_t workers_params = { .key_type = KVMGR_KEY_STRING, - .dynamic = 1, .nslots = 5381, .delta = 64, .datakiller = (destroy_fn_t)kill_sched_worker }; kvmgr_params_t users_params = { .key_type = KVMGR_KEY_STRING, - .dynamic = 1, .nslots = 5381, .delta = 64, .datakiller = (destroy_fn_t)kill_sched_user }; kvmgr_params_t projects_params = { .key_type = KVMGR_KEY_STRING, - .dynamic = 1, .nslots = 31, .delta = 8, .datakiller = (destroy_fn_t)kill_sched_project }; kvmgr_params_t resmgr_params = { .key_type = KVMGR_KEY_STRING, - .dynamic = 1, .nslots = 64, .delta = 8, .datakiller = NULL }; @@ -360,6 +358,8 @@ int sched_init(xce_scheduler_t* sched, xce_client_config_t* config, sched->nextid = 1; sched->configver = 0; + xstring_table_init(1024*1024); + mylock_init(&(sched->lock)); cbuf_init(&(sched->wbuf), 1, sizeof(int)); cbuf_init(&(sched->jbuf), njobs, sizeof(uint64_t)); @@ -467,7 +467,9 @@ static int sched_init_resources(xce_scheduler_t* sched) for (i = 0; i < config->n_resources; i++) { xce_resource_t* r = config->resources[i]; + lock(sched->resmgr, RTTI_LOCK_WRITE); key_add(sched->resmgr, (void*)r->name, (void*)r); + unlock(sched->resmgr); } return 0; } @@ -747,7 +749,13 @@ void sched_end_session(xce_scheduler_t* sched) int sched_add_job(xce_scheduler_t* sched, xce_sched_job_t* j) { - return key_add(sched->jobs, (void*)j->jobid, j); + int rc; + + lock(sched->jobs, RTTI_LOCK_WRITE); + rc = key_add(sched->jobs, (void*)j->jobid, j); + unlock(sched->jobs); + + return rc; } xce_sched_job_t* sched_get_job(xce_scheduler_t* sched, uint64_t jobid) @@ -755,7 +763,10 @@ xce_sched_job_t* sched_get_job(xce_scheduler_t* sched, uint64_t jobid) int rc = 0; xce_sched_job_t* job = NULL; + lock(sched->jobs, RTTI_LOCK_READ); rc = key_find(sched->jobs, (void*)jobid, (void**)&job); + unlock(sched->jobs); + if (rc <= 0) { return NULL; } @@ -765,7 +776,13 @@ xce_sched_job_t* sched_get_job(xce_scheduler_t* sched, uint64_t jobid) int sched_add_queue(xce_scheduler_t* sched, xce_sched_queue_t* q) { - return key_add(sched->queues, (void*)q->name, q); + int rc; + + lock(sched->queues, RTTI_LOCK_WRITE); + rc = key_add(sched->queues, (void*)q->name, q); + unlock(sched->queues); + + return rc; } xce_sched_queue_t* sched_get_queue(xce_scheduler_t* sched, char* name) @@ -773,7 +790,10 @@ xce_sched_queue_t* sched_get_queue(xce_scheduler_t* sched, char* name) xce_sched_queue_t* q = NULL; int rc = 0; + lock(sched->queues, RTTI_LOCK_READ); rc = key_find(sched->queues, (void*)name, (void**)&q); + unlock(sched->queues); + if (rc <= 0) { return NULL; } @@ -785,7 +805,9 @@ int sched_add_worker(xce_scheduler_t* sched, xce_sched_worker_t* w) { int rc; + lock(sched->workers, RTTI_LOCK_WRITE); rc = key_add(sched->workers, (void*)w->name, (void*)w); + unlock(sched->workers); if (rc >= 0) { w->idx = rc; } @@ -798,7 +820,10 @@ xce_sched_worker_t* sched_get_worker(xce_scheduler_t* sched, char* name) xce_sched_worker_t* w = NULL; int rc = 0; + lock(sched->workers, RTTI_LOCK_READ); rc = key_find(sched->workers, (void*)name, (void**)&w); + unlock(sched->workers); + if (rc <= 0) { return NULL; } @@ -810,7 +835,9 @@ int sched_add_user(xce_scheduler_t* sched, xce_sched_user_t* u) { int rc = 0; + lock(sched->users, RTTI_LOCK_WRITE); rc = key_add(sched->users, (void*)u->name, (void*)u); + unlock(sched->users); return rc; } @@ -820,7 +847,9 @@ xce_sched_user_t* sched_get_user(xce_scheduler_t* sched, char* name) xce_sched_user_t* user = NULL; int rc = 0; + lock(sched->users, RTTI_LOCK_READ); rc = key_find(sched->users, (void*)name, (void**)&user); + unlock(sched->users); if (rc < 0) { return NULL; } @@ -835,7 +864,10 @@ int sched_add_project(xce_scheduler_t* sched, xce_sched_project_t* p) { int rc; + lock(sched->projects, RTTI_LOCK_WRITE); rc = key_add(sched->projects, (void*)p->name, (void*)p); + unlock(sched->projects); + return rc; } @@ -848,7 +880,9 @@ xce_sched_project_t* sched_get_project(xce_scheduler_t* sched, char* name) name = "default"; } + lock(sched->projects, RTTI_LOCK_READ); rc = key_find(sched->projects, (void*)name, (void**)&project); + unlock(sched->projects); if (rc < 0) { return NULL; } diff --git a/src/tools/appimage/entry.sh b/src/tools/appimage/entry.sh index 0ded498..f86c89a 100644 --- a/src/tools/appimage/entry.sh +++ b/src/tools/appimage/entry.sh @@ -1,16 +1,14 @@ #!/usr/bin/env bash -APP="$0" - usage() { local _f - _f=$(dirname "$APP")/share/xce.md + _f=$(dirname "$(which xce_controller)")/../share/docs/xce.md if [ ! -f "$_f" ]; then echo no document found else - glow -p ${_f} + glow -p "${_f}" fi } @@ -20,6 +18,11 @@ if [ "${debug}" = "yes" ]; then set -x fi +if [ $# -lt 1 ]; then + usage + exit 1 +fi + case "$1" in controller) shift