myxce/src/controller/task.c

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

#include "cluster.h"
#include "xce/cluster_def.h"
#include "xce/errnox.h"
#include "xce/types.h"

static int run_tasks_on_host(cluster_t* c, endpoint_t* ep, uint64_t jobid,
                             xce_taskhost_t* info)
{
    worker_t*     w;
    int           rc;
    xce_taskrun_t runmsg = { .nhosts    = 1,
                             .jobid     = 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;
    }

    if (w == NULL) {
        return -ENOENT;
    }

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

static int check_task_hosts(cluster_t* c, xce_taskrun_t* runmsg)
{
    int i;

    for (i = 0; i < runmsg->nhosts; i++) {
        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;
        }

        if (rc == 0) {
            return -ENOENT;
        }
        if (w == NULL) {
            return -ENOENT;
        }

        if ((w->ep == NULL) || IS_WORKER_UNAVAIL(w->host.status)) {
            dereference(w);

            return -ENOENT;
        }

        dereference(w);
    }

    return 0;
}

int do_runtasks(cluster_t* c, endpoint_t* ep, xce_request_t* req)
{
    int            rc = 0;
    xce_reply_t    reply;
    int            i;
    xce_taskrun_t* runmsg = req->body.runtaskmsg;

    init_reply_by_req(&reply, req);
    reply.data_which = RequestReply_data_nodata;
    rc               = check_task_hosts(c, runmsg);

    reply.status = rc;
    xce_reply(ep, c->myhost, &reply);

    if (rc < 0) {
        return rc;
    }

    rc = 0;

    for (i = 0; i < runmsg->nhosts; i++) {
        int rc1;

        rc1 = run_tasks_on_host(c, ep, runmsg->jobid, runmsg->taskhosts[i]);
        if (rc1 < 0) {
            xce_task_failinfo_t failinfo
                = { .jobid   = runmsg->jobid,
                    .host    = runmsg->taskhosts[i]->host,
                    .taskidx = -1, /* all tasks failed */
                    .code    = rc1,
                    .detail  = NULL };
            xce_notify_data_t data = { .event = NotifyData_data_taskFailInfo,
                                       .data  = { .taskfailinfo = &failinfo } };

            /* notify client that run tasks on host fail */
            notify(c, ep, &data);

            break;
        }
    }

    return rc;
}

int on_task_start(cluster_t* c, xce_task_pidinfo_t* info)
{
    endpoint_t*       ep;
    xce_notify_data_t data      = { .event = NotifyData_data_taskPidInfo,
                                    .data  = { .taskinfo = info } };
    int               clientidx = info->clientidx;

    if (clientidx < 0) {
        ERROR("incorrect client idx %d", clientidx);
        return -EINVAL;
    }

    ep = (endpoint_t*)zarray_get(&(c->endpoints), clientidx);
    if (ep == NULL) {
        ERROR("incorrect client idx %d", clientidx);
        return -EINVAL;
    }

    notify(c, ep, &data);

    return 0;
}

int on_task_fail(cluster_t* c, xce_task_failinfo_t* info)
{
    endpoint_t*       ep;
    xce_notify_data_t data      = { .event = NotifyData_data_taskFailInfo,
                                    .data  = { .taskfailinfo = info } };
    int               clientidx = info->clientidx;

    if (clientidx < 0) {
        ERROR("incorrect client idx %d", clientidx);
        return -EINVAL;
    }

    ep = (endpoint_t*)zarray_get(&(c->endpoints), clientidx);
    if (ep == NULL) {
        ERROR("incorrect client idx %d", clientidx);
        return -EINVAL;
    }

    notify(c, ep, &data);

    return 0;
}

int on_task_finish(cluster_t* c, xce_task_finish_t* info)
{
    endpoint_t*       ep;
    xce_notify_data_t data      = { .event = NotifyData_data_taskFinish,
                                    .data  = { .taskfinish = info } };
    int               clientidx = info->clientidx;

    if (clientidx < 0) {
        ERROR("incorrect client idx %d", clientidx);
        return -EINVAL;
    }

    ep = (endpoint_t*)zarray_get(&(c->endpoints), clientidx);
    if (ep == NULL) {
        ERROR("incorrect client idx %d", clientidx);
        return -EINVAL;
    }

    notify(c, ep, &data);

    return 0;
}