save works

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;
 }

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;

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);
     }
 }

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;

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);
 

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;
     }

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;

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;
 }

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;
 }
 

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;

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;
 }