myxce/src/controller/job.c

#include <errno.h>
#include <pthread.h>
#include <signal.h>
#include <time.h>

#include "cluster.h"
#include "jobj.h"
#include "preason.h"
#include "request.h"
#include "xce/cluster_def.h"
#include "xce/errnox.h"
#include "xce/types.h"
#include "xstring.h"

static int mark_job_start(cluster_t*, job_t*, job_pidinfo_t*);
static int mark_job_finish(cluster_t*, job_t*, xce_jobfinish_t*);
static int mark_job_fail(cluster_t*, job_t*, job_failinfo_t*);
static int mark_job_update(cluster_t*, job_t*, int);

#define NJOBS (100)

uint64_t jobid = 100;

static job_t* create_job(cluster_t* c, xce_job_t* j)
{
    job_t* job = NULL;

    job = makeJob(c->jobmgr, j);

    return job;
}

uint64_t job_submit(void* c, xce_job_submit_t* submit)
{
    job_t*    job     = NULL;
    uint64_t  base_id = submit->jobid;
    xce_job_t j       = { .submit = submit->submit };
    time_t    now     = time(NULL);
    int       i;

    j.sub_time     = now;
    j.queue_time   = now;
    j.start_time   = 0;
    j.finish_time  = 0;
    j.pidinfo      = NULL;
    j.state        = JOB_STAT_PEND;
    j.allocs       = NULL;
    j.run_time     = 0;
    j.run_usage    = NULL;
    j.exit_info    = NULL;
    j.ckptinfo     = NULL;
    j.pending_ctrl = 0;

    for (i = 0; i < submit->array_size; i++) {
        j.jobid = base_id + i;
        job     = create_job(c, &j);
        if (job == NULL) {
            return -errno;
        }
    }

    return base_id;
}

int load_jobs(void* c)
{
#if 0
        job_t* job;
        int    i;
        char   cmdline[1024];

        for (i = 1; i < NJOBS; i++) {
            sprintf(cmdline, "echo normal; sleep %d", i);
            job = create_job(c, jobid++, "normal", "rshen", "/home/rshen", "/tmp",
                            "rshen", (char*)cmdline, NULL, 1, 1);
            if (job == NULL) {
                return -errno;
            }
        }

        for (i = 1; i < NJOBS; i++) {
            job = create_job(c, jobid++, "xce", "rshen", "/home/rshen", "/tmp",
                            "rshen", "echo xce; sleep 30", NULL, 1, 1);
            if (job == NULL) {
                return -errno;
            }
        }
#endif

    return 0;
}

static int get_nslots_in_alloc(xce_host_alloc_t* alloc)
{
    int i;

    for (i = 0; i < alloc->nallocs; i++) {
        if (strcmp(alloc->allocs[i]->resname, "slots") == 0) {
            return (int)alloc->allocs[i]->value;
        }
    }

    return 1;
}

static void free_launchinfo(xce_launch_t* launch)
{
    int i;

    for (i = 0; i < launch->nhosts; i++) {
        FREEUP(launch->hosts[i]);
    }

    FREEUP(launch->hosts);
}

static int mk_launchinfo(xce_launch_t* launch, xce_job_alloc_t* alloc)
{
    int i;
    int rc = 0;

    memset(launch, 0x0, sizeof(*launch));

    launch->hosts = (xce_host_launch_t**)calloc(alloc->nhosts,
                                                sizeof(xce_host_launch_t*));
    if (launch->hosts == NULL) {
        return -ENOMEM;
    }

    launch->nhosts = alloc->nhosts;
    for (i = 0; i < alloc->nhosts; i++) {
        launch->hosts[i]
            = (xce_host_launch_t*)calloc(1, sizeof(xce_host_launch_t));
        if (launch->hosts[i] == NULL) {
            rc = -ENOMEM;
            break;
        }

        launch->hosts[i]->host   = alloc->allocs[i]->host;
        launch->hosts[i]->nranks = get_nslots_in_alloc(alloc->allocs[i]);
    }

    if (rc < 0) {
        free_launchinfo(launch);
    }

    return rc;
}

static int dispatch_a_job_(cluster_t* c, endpoint_t* ep, xce_job_alloc_t* alloc)
{
    int                rc = 0;
    job_t*             j  = NULL;
    worker_t*          w  = NULL;
    xce_host_alloc_t** allocs;
    int                nc;
    xce_job_exec_t     einfo;
    xce_launch_t       launchinfo;

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

    if (j == NULL) {
        PERROR("jobmgr_lockjob");

        return -errno;
    }

    nc = alloc->nhosts;
    if (nc <= 0) {
        dereference(j);

        return -EINVAL;
    }

    /* FIXME: change case of nc > 1 just hard coded mpi as
     * the real job type later
     */
    mk_launchinfo(&launchinfo, 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;
    }

    if (w == NULL) {
        free_launchinfo(&launchinfo);
        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;
    }

    /* FIXME: we may consider to optimize when a job has enter a queue? */
    if (1) {
        char* q_pre_exec;
        char* q_post_exec;

        rc = get_queue_exec(c, j->job.submit->queue, &q_pre_exec, &q_post_exec);
        if (rc < 0) {
            return rc;
        }

        if (j->job.submit->start_info->q_pre_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;
    }

    einfo.jobid      = alloc->jobid;
    einfo.start_info = j->job.submit->start_info;
    einfo.launchinfo = &launchinfo;

    rc = xce_startjob_req(w->ep, &einfo);

    unlock(w);

    dereference(w);

    free_launchinfo(&launchinfo);

    if (j->job.allocs != NULL) {
        free_JobAlloc_ptr(&(j->job.allocs));
    }
    j->job.allocs        = (xce_job_alloc_t*)calloc(1, sizeof(xce_job_alloc_t));
    j->job.allocs->jobid = alloc->jobid;
    j->job.allocs->nhosts = nc;
    MOVE_PTR(j->job.allocs->allocs, allocs);
    alloc->allocs = NULL;

    if (rc < 0) {
        SET_ERROR(-rc);
        PERROR("xce_startjob_req");
        return rc;
    }

    return rc;
}

int job_ctrl(cluster_t* c, xce_jobctrl_t* ctrinfo)
{
    uint64_t jobid = ctrinfo->jobid;
    job_t*   j     = NULL;
    int      rc    = 0;

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

    if (j == NULL) {
        PERROR("jobmgr_lockjob");
        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;

        /* 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);
                if (rc < 0) {
                    ERROR("job ctrl %ld on host %s returns %d", jobid,
                          w->host.name, rc);
                }
            } else {
                ERROR("worker %s is not in correct state, status %x ep %p",
                      w->host.name, w->host.status, w->ep);
                rc = -errno;
            }
            unlock(w);

            dereference(w);
        } else {
            rc = -errno;
        }
    } else {
        /* job has not been started yet */
        job_failinfo_t failinfo = {
            .jobid = jobid,
        };

        xce_use(&failinfo);

        switch (ctrinfo->action) {
        case JobControl_u_siginfo:
            switch (ctrinfo->u.siginfo->signo) {
            case SIGCONT:
                if (j->job.state
                    & (JOB_STAT_PSUSP | JOB_STAT_USUSP | JOB_STAT_SSUSP)) {
                    int state = j->job.state;

                    /* resume suspend job? */
                    state &= ~(JOB_STAT_PSUSP | JOB_STAT_USUSP);
                    mark_job_update(c, j, state);
                }
                break;
            case SIGSTOP:
            case SIGTSTP:
                if (!(j->job.state
                      & (JOB_STAT_PSUSP | JOB_STAT_USUSP | JOB_STAT_SSUSP))) {
                    int state = j->job.state;

                    state |= JOB_STAT_PSUSP;
                    mark_job_update(c, j, state);
                }
                break;
            case SIGKILL:
            case SIGTERM:
            default:
                if (j->job.state
                    & (JOB_STAT_DONE | JOB_STAT_EXIT | JOB_STAT_ERR)) {
                    break;
                }
                /* FIXME: */
                failinfo.code   = 128 + ctrinfo->u.siginfo->signo;
                failinfo.detail = "Job killed by signal";

                mark_job_fail(c, j, &failinfo);
                break;
            }
            break;
        case JobControl_u_ckptoption:
            if (j->job.state & (JOB_STAT_DONE | JOB_STAT_EXIT | JOB_STAT_ERR)) {
                ERROR("job <%ld> has been finished, no checkpoint can be done.",
                      jobid);
            } else {
                ERROR("job %ld has not been started, no checkpoint can be done",
                      jobid);
            }
            rc = -EINVAL;
            break;
        default:
            rc = -EINVAL;
            break;
        }
    }

    unlock(j);

    return rc;
}

static void notify_dispatch(cluster_t* c, endpoint_t* ep, uint64_t jobid,
                            int code)
{
    /* clang-format off */
    job_failinfo_t info = {
        .jobid = jobid,
        .code  = code
    };
    xce_notify_data_t data = {
        .event = NotifyData_data_jobDispatchFail,
        .data  = {
             .dispatchfail = &info
	}
    };
    /* clang-format on */

    notify(c, ep, &data);
}

int handle_jobdispatch(cluster_t* c, endpoint_t* ep, xce_request_t* request)
{
    int                 rc = 0;
    int                 i;
    xce_job_dispatch_t* dispatchmsg = request->body.dispatchmsg;
    xce_reply_t         reply;

    /* reply job dispatch that we'll handle request */
    init_reply_by_req(&reply, request);
    reply.status     = 0;
    reply.data_which = RequestReply_data_nodata;

    xce_reply(ep, c->myhost, &reply);

    for (i = 0; i < dispatchmsg->njobs; i++) {
        int              rc1;
        xce_job_alloc_t* alloc = NULL;

        MOVE_PTR(alloc, dispatchmsg->allocs[i]);

        rc1 = dispatch_a_job_(c, ep, alloc);
        if (rc1 < 0) {
            /* notify dispatch fail */
            notify_dispatch(c, ep, alloc->jobid, rc1);
        } else {
            /* notify dispatch ok */
            notify_dispatch(c, ep, alloc->jobid, 0);
        }

        free_JobAlloc_ptr(&(alloc));
    }
    return rc;
}

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)
{
    uint64_t          jobid;
    xce_job_submit_t* submit;
    xce_reply_t       reply;
    xlist_t           jlist;
    job_t*            j = NULL;
    int               i;
    int               rc;

    submit = request->body.jobsubmsg;

    init_reply_by_req(&reply, request);

    if (NULLSTR(submit->submit->user) || NULLSTR(submit->submit->queue)
        || (submit->submit->start_info == NULL)
        || NULLSTR(submit->submit->start_info->cmdline)) {
        jobid = 0;
        errno = EINVAL;
    } else {
        xce_run_hooks(&(c->hooksys), XCE_CONTROLLER_HOOK_JOB, HOOK_BEFORE_EVENT,
                      HOOK_DETAIL_JOB_NEW, c, submit);

        jobid = job_submit(c, submit);
    }
    if (jobid == 0) {
        reply.status     = -errno;
        reply.data_which = RequestReply_data_nodata;

        xce_reply(ep, c->myhost, &reply);
        return -errno;
    }

    reply.status     = 0;
    reply.data_which = RequestReply_data_jobid;
    reply.data.jobid = jobid;

    xce_reply(ep, c->myhost, &reply);

    xlist_init(&jlist, submit->array_size);

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

            fmt_jobid(idstring, sizeof(idstring) - 1, jobid);
            ERROR("%s fail to lock job %s (%lu)", __func__, idstring, jobid);
        } else {
            if (xlist_store(&jlist, j) < 0) {
                PERROR("xlist_store");
            }
        }
    }

    xce_run_hooks(&(c->hooksys), XCE_CONTROLLER_HOOK_JOB, HOOK_AFTER_EVENT,
                  HOOK_DETAIL_JOB_NEW, c, &jlist);

    xlist_iter(&jlist, (xlist_iter_fn_t)xlist_deref_job_item_, NULL);

    xlist_fini(&jlist);

    return 0;
}

static int mark_job_start(cluster_t* c, job_t* job, job_pidinfo_t* info)
{
    xce_run_hooks(&(c->hooksys), XCE_CONTROLLER_HOOK_JOB, HOOK_BEFORE_EVENT,
                  HOOK_DETAIL_JOB_START, c, info, job);
    /* update job information */
    job->job.state      = JOB_STAT_RUN;
    job->job.start_time = time(NULL);
    if (job->job.pidinfo == NULL) {
        job->job.pidinfo = (job_pidinfo_t*)calloc(1, sizeof(job_pidinfo_t));
    }

    job->job.pidinfo->npids  = copy_int_list((int**)&(job->job.pidinfo->pids),
                                             (int*)info->pids, info->npids);
    job->job.pidinfo->npgids = copy_int_list((int**)&(job->job.pidinfo->pgids),
                                             (int*)info->pgids, info->npgids);
    if (info->exdata != NULL) {
        job->job.pidinfo->exdata = makeString(info->exdata);
    } else {
        job->job.pidinfo->exdata = NULL;
    }

    xce_run_hooks(&(c->hooksys), XCE_CONTROLLER_HOOK_JOB, HOOK_AFTER_EVENT,
                  HOOK_DETAIL_JOB_START, c, info, job);

    return 0;
}

int on_job_start(cluster_t* c, job_pidinfo_t* info)
{
    job_t* job = NULL;
    int    rc;

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

static int mark_job_finish(cluster_t* c, job_t* job, xce_jobfinish_t* finish)
{
    xce_run_hooks(&(c->hooksys), XCE_CONTROLLER_HOOK_JOB, HOOK_BEFORE_EVENT,
                  HOOK_DETAIL_JOB_EXIT, c, finish, job);
    /* update job information */
    if (finish->exit_code == 0) {
        job->job.state = JOB_STAT_DONE;
    } else {
        job->job.state = JOB_STAT_EXIT;
    }
    job->job.finish_time = time(NULL);

    /* fill job run time/rusage information */

    if (job->job.exit_info == NULL) {
        job->job.exit_info = (job_exitinfo_t*)calloc(1, sizeof(job_exitinfo_t));
    }

    if (job->job.exit_info != NULL) {
        job->job.exit_info->exitcode = finish->exit_code;
    }

    job->job.run_time = job->job.finish_time - job->job.start_time;

    xce_run_hooks(&(c->hooksys), XCE_CONTROLLER_HOOK_JOB, HOOK_AFTER_EVENT,
                  HOOK_DETAIL_JOB_EXIT, c, finish, job);

    return 0;
}

int on_job_finish(cluster_t* c, xce_jobfinish_t* finish)
{
    job_t* job = NULL;

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

static int mark_job_fail(cluster_t* c, job_t* job, job_failinfo_t* failinfo)
{
    xce_run_hooks(&(c->hooksys), XCE_CONTROLLER_HOOK_JOB, HOOK_BEFORE_EVENT,
                  HOOK_DETAIL_JOB_FAIL, c, failinfo, job);
    /* update job information */

    switch (failinfo->stage) {
    case JOB_EXEC_STAGE_Q_POST_EXEC:
    case JOB_EXEC_STAGE_POST_EXEC:
        if ((job->job.state & (JOB_STAT_EXIT | JOB_STAT_DONE)) == 0) {
            WARN("job %ld got post exec fail notification but job's exit "
                 "status has not been set",
                 job->job.jobid);
            job->job.state |= JOB_STAT_EXIT;
        }
        job->job.state |= JOB_STAT_PERR;
        break;
    default:
        job->job.state = JOB_STAT_EXIT;
        break;
    }

    job->job.finish_time = time(NULL);

    xce_run_hooks(&(c->hooksys), XCE_CONTROLLER_HOOK_JOB, HOOK_AFTER_EVENT,
                  HOOK_DETAIL_JOB_FAIL, c, failinfo, job);

    return 0;
}

static int mark_job_update(cluster_t* c, job_t* job, int state)
{
    xce_run_hooks(&(c->hooksys), XCE_CONTROLLER_HOOK_JOB, HOOK_BEFORE_EVENT,
                  HOOK_DETAIL_JOB_UPDATE, c, job);
    /* update job state */
    job->job.state = state;

    xce_run_hooks(&(c->hooksys), XCE_CONTROLLER_HOOK_JOB, HOOK_AFTER_EVENT,
                  HOOK_DETAIL_JOB_UPDATE, c, job);

    return 0;
}

int on_job_fail(cluster_t* c, job_failinfo_t* failinfo)
{
    job_t* job = NULL;

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

int on_job_stop(cluster_t* c, uint64_t jobid)
{
    job_t* job = NULL;
    int    state;

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

int on_job_resume(cluster_t* c, uint64_t jobid)
{
    job_t* job = NULL;
    int    state;

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

int on_job_postdone(cluster_t* c, endpoint_t* ep, uint64_t jobid)
{
    job_t* job = NULL;
    int    state;

    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 "
             "has not been set",
             job->job.jobid);
        state |= JOB_STAT_EXIT;
        /* ensure that job finish time has been set */
        job->job.finish_time = time(NULL);
    }

    state |= JOB_STAT_PDONE;
    mark_job_update(c, job, state);
    unlock(job);

    dereference(job);

    return 0;
}

int on_job_ckpt(cluster_t* c, endpoint_t* ep, xce_ckptinfo_t* ckptinfo)
{
    uint64_t jobid;
    char*    info = NULL;
    job_t*   j    = NULL;

    if (ckptinfo == NULL) {
        ERROR("%s ckptinfo should not be NULL", __func__);
        return -EINVAL;
    }

    jobid = ckptinfo->jobid;
    info  = ckptinfo->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;
}

int do_jobcontrol_op(cluster_t* c, endpoint_t* ep, xce_request_t* request)
{
    int         rc = 0;
    xce_reply_t reply;

    init_reply_by_req(&reply, request);

    rc               = job_ctrl(c, request->body.jobctrlmsg);
    reply.status     = rc;
    reply.data_which = RequestReply_data_nodata;

    xce_reply(ep, c->myhost, &reply);
    return rc;
}

int update_job_rusages(cluster_t* c, endpoint_t* ep, xce_request_t* request)
{
    xce_job_rusages_t* rusages = request->body.updatemsg->data.jobrusages;
    job_t*             j       = NULL;
    worker_t*          w       = NULL;
    char*              host    = NULL;

    TRACE("%s trying to update job(%ld) rusage", __func__, rusages->jobid);

    xce_run_hooks(&(c->hooksys), XCE_CONTROLLER_HOOK_JOB, HOOK_BEFORE_EVENT,
                  HOOK_DETAIL_JOB_RUSAGE_UPDATE, c, rusages);

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

    w = get_ep_worker(c, ep);
    if (w == NULL) {
        return errno;
    }

    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,
                         rusages->usages);

    update_pids(&(j->job.pidinfo), rusages->npids, rusages->pids);

    update_pgids(&(j->job.pidinfo), rusages->npgids, rusages->pgids);

    /* FIXME
     * update other fields of job rusage later
     */
    xce_run_hooks(&(c->hooksys), XCE_CONTROLLER_HOOK_JOB, HOOK_AFTER_EVENT,
                  HOOK_DETAIL_JOB_UPDATE, c, j);

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

static int set_preason(cluster_t* c, xce_pendreason_t* reasons)
{
    job_t*     j   = NULL;
    xce_job_t* job = NULL;

    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);
    if (job->pend_reasons != NULL) {
        erase_job_preason(job->pend_reasons);
        FREEUP(job->pend_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;
}

int do_set_preason(cluster_t* c, endpoint_t* ep, xce_request_t* request)
{
    int                i;
    xce_preason_set_t* reasons = request->body.preasonmsg;
    xce_reply_t        reply;

    init_reply_by_req(&reply, request);
    reply.status     = 0;
    reply.data_which = RequestReply_data_nodata;

    xce_reply(ep, c->myhost, &reply);

    for (i = 0; i < reasons->nreasons; i++) {
        int rc = 0;

        if (reasons->reasons[i] == NULL) {
            continue;
        }

        rc = set_preason(c, reasons->reasons[i]);
        if (rc < 0) {
            PERROR("set_preason");
        }
    }

    return 0;
}

int do_clr_preason(cluster_t* c, endpoint_t* ep, xce_request_t* request)
{
    int                 i;
    xce_reply_t         reply;
    xce_clear_reason_t* clreason = request->body.clrpreasonmsg;

    init_reply_by_req(&reply, request);
    reply.status     = 0;
    reply.data_which = RequestReply_data_nodata;

    xce_reply(ep, c->myhost, &reply);

    for (i = 0; i < clreason->njobs; i++) {
        uint64_t   jobid = clreason->jobs[i];
        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);

        job = &(j->job);
        if (job->pend_reasons != NULL) {
            erase_job_preason(job->pend_reasons);
            FREEUP(job->pend_reasons);
        }

        xce_run_hooks(&(c->hooksys), XCE_CONTROLLER_HOOK_JOB, HOOK_BEFORE_EVENT,
                      HOOK_DETAIL_CLRPREASON, c, j);
        unlock(j);

        dereference(j);
    }

    return 0;
}

/* ---- fetchfile pending-request infrastructure ---- */

typedef struct fetchfile_pending {
    uint64_t               jobid;
    pthread_mutex_t        mu;
    pthread_cond_t         cv;
    int                    done;
    int                    status;
    xce_fetchfile_reply_t* reply;
    struct fetchfile_pending* next;
} fetchfile_pending_t;

static pthread_mutex_t    g_fetchfile_mu   = PTHREAD_MUTEX_INITIALIZER;
static fetchfile_pending_t* g_fetchfile_list = NULL;

static void fetchfile_register(fetchfile_pending_t* p)
{
    pthread_mutex_lock(&g_fetchfile_mu);
    p->next        = g_fetchfile_list;
    g_fetchfile_list = p;
    pthread_mutex_unlock(&g_fetchfile_mu);
}

static void fetchfile_unregister(fetchfile_pending_t* p)
{
    fetchfile_pending_t** pp;
    pthread_mutex_lock(&g_fetchfile_mu);
    for (pp = &g_fetchfile_list; *pp != NULL; pp = &(*pp)->next) {
        if (*pp == p) {
            *pp = p->next;
            break;
        }
    }
    pthread_mutex_unlock(&g_fetchfile_mu);
}

fetchfile_pending_t* fetchfile_find(uint64_t jobid)
{
    fetchfile_pending_t* p;
    pthread_mutex_lock(&g_fetchfile_mu);
    for (p = g_fetchfile_list; p != NULL; p = p->next) {
        if (p->jobid == jobid) break;
    }
    pthread_mutex_unlock(&g_fetchfile_mu);
    return p;
}

/* Signal a pending fetchfile request with the worker's reply. */
void fetchfile_signal(uint64_t jobid, int status, xce_fetchfile_reply_t* reply)
{
    fetchfile_pending_t* p = fetchfile_find(jobid);
    if (p == NULL) return;

    pthread_mutex_lock(&p->mu);
    p->status = status;
    p->reply  = reply;
    p->done   = 1;
    pthread_cond_signal(&p->cv);
    pthread_mutex_unlock(&p->mu);
}

/* ---- on_fetchfile: proxy fetchfile from apiserver to worker ---- */

int on_fetchfile(cluster_t* c, endpoint_t* caller_ep, xce_request_t* request)
{
    xce_fetchfile_t* msg = request->body.fetchfilemsg;
    xce_reply_t      reply;
    job_t*           j        = NULL;
    worker_t*        w        = NULL;
    int              rc       = 0;
    char*            exechost = NULL;
    struct timespec  deadline;

    fetchfile_pending_t pending = {
        .mu   = PTHREAD_MUTEX_INITIALIZER,
        .cv   = PTHREAD_COND_INITIALIZER,
        .done = 0, .status = -ETIMEDOUT, .reply = NULL, .next = NULL
    };

    init_reply_by_req(&reply, request);

    if (msg == NULL || msg->filepath == NULL || msg->filepath[0] == '\0') {
        reply.status     = -EINVAL;
        reply.data_which = RequestReply_data_nodata;
        xce_reply(caller_ep, c->myhost, &reply);
        return -EINVAL;
    }

    pending.jobid = msg->jobid;

    /* Look up the job */
    lock(c->jobmgr, RTTI_LOCK_READ);
    key_find(c->jobmgr, (void*)msg->jobid, (void**)&j);
    if (j != NULL) reference(j);
    unlock(c->jobmgr);

    if (j == NULL) {
        reply.status     = -ENOENT;
        reply.data_which = RequestReply_data_nodata;
        xce_reply(caller_ep, c->myhost, &reply);
        return -ENOENT;
    }

    /* Get the first exec-host from job allocs */
    lock(j, RTTI_LOCK_READ);
    if (j->job.allocs != NULL && j->job.allocs->nhosts > 0
        && j->job.allocs->allocs != NULL && j->job.allocs->allocs[0] != NULL
        && j->job.allocs->allocs[0]->host != NULL) {
        exechost = strdup(j->job.allocs->allocs[0]->host);
    }
    unlock(j);
    dereference(j);

    if (exechost == NULL) {
        reply.status     = -ENOENT;
        reply.data_which = RequestReply_data_nodata;
        xce_reply(caller_ep, c->myhost, &reply);
        return -ENOENT;
    }

    /* Find the worker */
    lock(c->workermgr, RTTI_LOCK_READ);
    rc = key_find(c->workermgr, (void*)exechost, (void**)&w);
    if (w != NULL) reference(w);
    unlock(c->workermgr);
    free(exechost);

    if (w == NULL) {
        reply.status     = -ENOENT;
        reply.data_which = RequestReply_data_nodata;
        xce_reply(caller_ep, c->myhost, &reply);
        return -ENOENT;
    }

    lock(w, RTTI_LOCK_READ);
    if (w->ep == NULL) {
        unlock(w);
        dereference(w);
        reply.status     = -ENOENT;
        reply.data_which = RequestReply_data_nodata;
        xce_reply(caller_ep, c->myhost, &reply);
        return -ENOENT;
    }

    /* Register pending before sending to avoid a race with the reply */
    fetchfile_register(&pending);
    rc = xce_fetchfile_req(w->ep, msg);
    unlock(w);
    dereference(w);

    if (rc < 0) {
        fetchfile_unregister(&pending);
        reply.status     = rc;
        reply.data_which = RequestReply_data_nodata;
        xce_reply(caller_ep, c->myhost, &reply);
        return rc;
    }

    /* Wait up to 10 seconds for the worker to respond */
    clock_gettime(CLOCK_REALTIME, &deadline);
    deadline.tv_sec += 10;

    pthread_mutex_lock(&pending.mu);
    while (!pending.done) {
        rc = pthread_cond_timedwait(&pending.cv, &pending.mu, &deadline);
        if (rc == ETIMEDOUT) break;
    }
    pthread_mutex_unlock(&pending.mu);

    fetchfile_unregister(&pending);

    /* Relay the reply back to the apiserver */
    if (pending.done && pending.status >= 0 && pending.reply != NULL) {
        reply.status              = 0;
        reply.data_which          = RequestReply_data_fetchfilereply;
        reply.data.fetchfilereply = pending.reply;
    } else {
        reply.status     = pending.done ? pending.status : -ETIMEDOUT;
        reply.data_which = RequestReply_data_nodata;
        FREEUP(pending.reply);
    }
    xce_reply(caller_ep, c->myhost, &reply);
    return 0;
}