myxce/src/controller/sched.c

#include <errno.h>
#include <stdlib.h>

#include "cluster.h"
#include "controller.capnp.h"
#include "request.h"
#include "worker.h"
#include "xce/cluster_def.h"
#include "xce/types.h"
#include "xcelimit.h"

extern void collect_shared_resvals(cluster_t*, collect_check_fn_t, xlist_t*);

static int xcesched_push_joblist(cluster_t* c, xlist_t* l)
{
    xce_data_t      data;
    xce_jobs_data_t jobs = { .n_total = l->count,
                             .pos     = 0,
                             .count   = l->count,
                             .jobs    = (xce_job_t**)l->data };
    int             rc   = 0;

    memset(&data, 0x0, sizeof(data));

    data.table     = Data_data_schedjobs;
    data.data.jobs = &jobs;

    rc = xce_push_data_req(c->sched.ep, c->myhost, &data, check_controller, c,
                           100);
    return rc;
}

static int xcesched_push_workerlist(cluster_t* c, xlist_t* l)
{
    xce_data_t         data;
    xce_workers_data_t workers = { .n_total = l->count,
                                   .pos     = 0,
                                   .count   = l->count,
                                   .workers = (xce_worker_t**)l->data };
    int                rc      = 0;

    memset(&data, 0x0, sizeof(data));

    data.table        = Data_data_schedworkers;
    data.data.workers = &workers;

    rc = xce_push_data_req(c->sched.ep, c->myhost, &data, check_controller, c,
                           100);
    return rc;
}

static int xcesched_push_configs(cluster_t* c)
{
    xce_configdata_t   configdata;
    int                rc            = 0;
    xce_limit_config_t limit1_config = {
        .name        = "limit1",
        .per_user    = 0,
        .per_queue   = 0,
        .per_project = 0,
        .per_host    = 0,
        .users
        = &(xce_group_config_t) { .members
                                  = { .count = 1, .data = (char*[]) { "all" } },
                                 .ex_members
                                  = { .count = 1,
                                      .data  = (char*[]) { "rshen" } } },
        .queues   = NULL,
        .projects = NULL,
        .hosts    = NULL,
        .val
        = &(xce_limit_val_t) { .resname = "slots", .percent = 0, .val = 3.0 }
    };
    xce_limit_config_t limit2_config = {
        .name        = "limits2",
        .per_user    = 1,
        .per_queue   = 1,
        .per_project = 1,
        .per_host    = 1,
        .users       = &(xce_group_config_t) { .members
                                               = { .count = 1,
                                                   .data  = (char*[]) { "rshen" } } },
        .queues
        = &(xce_group_config_t) { .members = { .count = 1,
                                               .data  = (char*[]) { "xce" } } },
        .projects = NULL,
        .hosts    = NULL,
        .val = &(xce_limit_val_t) { .resname = "mem", .percent = 1, .val = 0.2 }
    };
    xce_limit_config_t* limits[] = { &limit1_config, &limit2_config };
    int                 nlimits  = 2;
    xlist_t             clist;
    xlist_t             qlist;
    xlist_t             rlist;
    xlist_t             xreslist;

    memset(&configdata, 0x0, sizeof(configdata));

    configdata.data.conf4sched
        = (xce_conf4sched_t*)calloc(1, sizeof(xce_conf4sched_t));
    if (configdata.data.conf4sched == NULL) {
        return -ENOMEM;
    }

    xlist_init(&clist, 64);
    xlist_init(&qlist, 64);
    xlist_init(&rlist, 256);
    xlist_init(&xreslist, 256);
    collect_confvars(c, NULL, &clist);
    configdata.data.conf4sched->n_confvars = clist.count;
    configdata.data.conf4sched->confvars   = (xce_config_t**)clist.data;
    collect_queues(c, NULL, &qlist);
    configdata.data.conf4sched->n_queues = qlist.count;
    configdata.data.conf4sched->queues   = (xce_queue_t**)qlist.data;
    collect_resources(c, NULL, &rlist);
    configdata.data.conf4sched->n_resources = rlist.count;
    configdata.data.conf4sched->resources   = (xce_resource_t**)rlist.data;
    collect_shared_resvals(c, NULL, &xreslist);
    configdata.data.conf4sched->nresvals = xreslist.count;
    configdata.data.conf4sched->resvals  = (xce_val_with_idx_t**)xreslist.data;

    configdata.data.conf4sched->n_ugroups = c->myconf->cluster_config->nugroups;
    configdata.data.conf4sched->ugroups   = c->myconf->cluster_config->ugroups;
    configdata.data.conf4sched->n_hgroups = c->myconf->cluster_config->nhgroups;
    configdata.data.conf4sched->hgroups   = c->myconf->cluster_config->hgroups;
#if 0 /* FIXME:  we use builtin limit currently */
        configdata.data.conf4sched->nlimits   = c->myconf->cluster_config->nlimits;
        configdata.data.conf4sched->limits    = c->myconf->cluster_config->limits;
#else
    configdata.data.conf4sched->nlimits = nlimits;
    configdata.data.conf4sched->limits  = &(limits[0]);
#endif
    configdata.data.conf4sched->nhproviders
        = c->myconf->cluster_config->nhproviders;
    configdata.data.conf4sched->hproviders
        = c->myconf->cluster_config->hproviders;

    /* add resources */
    configdata.data_which = ConfigData_data_schedconfig;
    configdata.data.conf4sched->clustername
        = c->myconf->cluster_config->clustername;

    rc = xce_pushconfig_req(c->sched.ep, c->myhost, &configdata);

    free_resval_with_idx_list(&(configdata.data.conf4sched->resvals),
                              configdata.data.conf4sched->nresvals);

    FREEUP(configdata.data.conf4sched);

    unlock_collect_queues(&qlist);
    
    xlist_fini(&clist);
    xlist_fini(&qlist);
    xlist_fini(&rlist);
    xlist_fini(&xreslist);

    return rc;
}

static int check_job_(xce_job_t* j)
{
    if (j == NULL) {
        return 0;
    }

    return 1;
}

static int xcesched_push_jobs(cluster_t* c)
{
    xce_data_t      data;
    xce_jobs_data_t jobs = { .n_total = 0, .pos = 0, .count = 0, .jobs = NULL };
    int             rc   = 0;
    xlist_t         jlist;

    memset(&data, 0x0, sizeof(data));

    xlist_init(&jlist, 1024 * 1024);

    data.table     = Data_data_schedjobs;
    data.data.jobs = &jobs;

    collect_jobs(c, (collect_check_fn_t)check_job_, &jlist);
    jobs.n_total = jlist.count;
    jobs.jobs    = (xce_job_t**)jlist.data;
    jobs.count   = jobs.n_total;

    rc = xce_push_data_req(c->sched.ep, c->myhost, &data, check_controller, c,
                           100);
    FREEUP(jobs.jobs);

    return rc;
}

static int xcesched_push_workers(cluster_t* c)
{
    xce_data_t         data;
    xce_workers_data_t workers
        = { .n_total = 0, .pos = 0, .count = 0, .workers = NULL };
    int     rc = 0;
    int     i;
    xlist_t wlist;

    memset(&data, 0x0, sizeof(data));
    xlist_init(&wlist, 8192);

    data.table        = Data_data_schedworkers;
    data.data.workers = &workers;

    collect_workers(c, (collect_check_fn_t)check_worker_, &wlist);
    workers.n_total = wlist.count;
    workers.workers = (xce_worker_t**)wlist.data;
    workers.count   = workers.n_total;

    rc = xce_push_data_req(c->sched.ep, c->myhost, &data, check_controller, c,
                           100);

    /* free up collected resource value which used
    passed to scheduler
    */
    for (i = 0; i < workers.n_total; i++) {
        xce_worker_t* w = workers.workers[i];

        free_resval_with_idx_list(&(w->resvals_to_sched), w->nresvals_to_sched);
    }

    FREEUP(workers.workers);

    return rc;
}

static int sched_job_new_fun(int type, int situation, void* data, va_list ap)
{
    cluster_t* c  = va_arg(ap, cluster_t*);
    xlist_t*   l  = va_arg(ap, xlist_t*);
    int        rc = 0;

    mylock_lock(&(c->sched.lock));
    if (c->sched.ep != NULL) {
        switch (situation) {
        case HOOK_AFTER_EVENT:
            rc = -EINVAL;
            if (l != NULL) {
                rc = xcesched_push_joblist(c, l);
            }
            break;
        default:
            break;
        }
    }
    mylock_unlock(&(c->sched.lock));

    return rc;
}

static int sched_job_start_fun(int type, int situation, void* data, va_list ap)
{
    cluster_t*     c    = va_arg(ap, cluster_t*);
    job_pidinfo_t* info = va_arg(ap, job_pidinfo_t*);
    int            rc   = 0;
    /* clang-format off */
        xce_notify_data_t xdata = {
            .event = NotifyData_data_schedStartJob,
            .data  = {
                .sched_start_job = info->jobid
        }
        };
    /* clang-format on */

    mylock_lock(&(c->sched.lock));
    if (c->sched.ep != NULL) {
        switch (situation) {
        case HOOK_AFTER_EVENT:
            rc = notify(c, c->sched.ep, &xdata);
            break;
        default:
            break;
        }
    }

    mylock_unlock(&(c->sched.lock));

    return rc;
}

static int sched_job_exit_fun(int type, int situation, void* data, va_list ap)
{
    cluster_t*       c    = va_arg(ap, cluster_t*);
    xce_jobfinish_t* info = va_arg(ap, xce_jobfinish_t*);
    int              rc   = 0;
    /* clang-format off */
        xce_notify_data_t xdata = {
            .event = NotifyData_data_schedExitJob,
            .data  = {
                .sched_exit_job = info->jobid
        }
        };
    /* clang-format on */

    mylock_lock(&(c->sched.lock));
    if (c->sched.ep != NULL) {
        switch (situation) {
        case HOOK_AFTER_EVENT:
            rc = notify(c, c->sched.ep, &xdata);
            break;
        default:
            break;
        }
    }

    mylock_unlock(&(c->sched.lock));

    return rc;
}

static int sched_job_fail_fun(int type, int situation, void* data, va_list ap)
{
    cluster_t*      c    = va_arg(ap, cluster_t*);
    job_failinfo_t* info = va_arg(ap, job_failinfo_t*);
    int             rc   = 0;
    /* clang-format off */
        xce_notify_data_t xdata = {
            .event = NotifyData_data_schedFailJob,
            .data  = {
                .sched_fail_job = info->jobid
        }
        };
    /* clang-format on */

    mylock_lock(&(c->sched.lock));
    if (c->sched.ep != NULL) {
        switch (situation) {
        case HOOK_AFTER_EVENT:
            rc = notify(c, c->sched.ep, &xdata);
            break;
        default:
            break;
        }
    }

    mylock_unlock(&(c->sched.lock));

    return rc;
}

static int sched_job_update_fun(int type, int situation, void* data, va_list ap)
{
    cluster_t* c   = va_arg(ap, cluster_t*);
    job_t*     job = va_arg(ap, job_t*);
    int        rc  = 0;
    /* clang-format off */
        xce_notify_data_t xdata = {
            .event = NotifyData_data_schedUpdateJob,
            .data  = {
                .sched_update_job = &(job->job)
        }
        };
    /* clang-format on */

    mylock_lock(&(c->sched.lock));
    if (c->sched.ep != NULL) {
        switch (situation) {
        case HOOK_AFTER_EVENT:
            rc = notify(c, c->sched.ep, &xdata);
            break;
        default:
            break;
        }
    }

    mylock_unlock(&(c->sched.lock));

    return rc;
}

static int sched_job_rusage_update_fun(int type, int situation, void* data,
                                       va_list ap)
{
    cluster_t*         c       = va_arg(ap, cluster_t*);
    xce_job_rusages_t* rusages = va_arg(ap, xce_job_rusages_t*);
    int                rc      = 0;
    /* clang-format off */
        xce_notify_data_t xdata = {
            .event = NotifyData_data_schedJobRusages,
            .data  = {
            .sched_job_rusages = rusages 
        }
        };
    /* clang-format on */

    mylock_lock(&(c->sched.lock));
    if (c->sched.ep != NULL) {
        switch (situation) {
        case HOOK_AFTER_EVENT:
            rc = notify(c, c->sched.ep, &xdata);
            break;
        default:
            break;
        }
    }

    mylock_unlock(&(c->sched.lock));

    return rc;
}

static int sched_job_hook_fun(int type, int situation, int detail, void* data,
                              va_list ap)
{
    switch (detail) {
    case HOOK_DETAIL_JOB_NEW:
        return sched_job_new_fun(type, situation, data, ap);
    case HOOK_DETAIL_JOB_START:
        return sched_job_start_fun(type, situation, data, ap);
    case HOOK_DETAIL_JOB_EXIT:
        return sched_job_exit_fun(type, situation, data, ap);
    case HOOK_DETAIL_JOB_UPDATE:
        return sched_job_update_fun(type, situation, data, ap);
    case HOOK_DETAIL_JOB_FAIL:
        return sched_job_fail_fun(type, situation, data, ap);
    case HOOK_DETAIL_JOB_RUSAGE_UPDATE:
        return sched_job_rusage_update_fun(type, situation, data, ap);
    }

    return 0;
}
static int sched_worker_change(int type, int situation, void* data,
                               cluster_t* c, worker_t* w)
{
    int     rc = 0;
    xlist_t wlist;
    xlist_t resvalist;

    TRACE("%s worker %s idx %d situation %d", __func__, w->host.name, w->idx,
          situation);

    xlist_init(&resvalist, 256);
    collect_resval_with_idx(&(w->host.resvals), &resvalist);
    w->host.nresvals_to_sched = resvalist.count;
    w->host.resvals_to_sched  = (xce_val_with_idx_t**)resvalist.data;

    mylock_lock(&(c->sched.lock));
    if (c->sched.ep != NULL) {
        switch (situation) {
        case HOOK_AFTER_EVENT:
            xlist_init(&wlist, 2);
            xlist_store(&wlist, w);
            rc = xcesched_push_workerlist(c, &wlist);
            xlist_fini(&wlist);
            break;
        default:
            break;
        }
    }
    mylock_unlock(&(c->sched.lock));

#if 0
    /*FIXME*/
    free_resval_with_idx_list(&(w->host.resvals_to_sched),
                              w->host.nresvals_to_sched);
#endif
    xlist_fini(&resvalist);

    return rc;
}

static int sched_worker_join(int type, int situation, void* data, va_list ap)
{
    cluster_t* c    = va_arg(ap, cluster_t*);
    char*      name = va_arg(ap, char*);
    worker_t*  w    = NULL;
    int        rc;

    if (situation == HOOK_BEFORE_EVENT) {
        /* nothing to do  */
        return 0;
    }

    rc = key_find(c->workermgr, (void*)name, (void**)&w);
    if (rc < 0) {
        return rc;
    }

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

    rc = sched_worker_change(type, situation, data, c, w);

    return rc;
}

static int sched_worker_dead(int type, int situation, void* data, va_list ap)
{
    cluster_t* c = va_arg(ap, cluster_t*);
    worker_t*  w = va_arg(ap, worker_t*);

    /* FIXME: need to handle job status in this worker */
    if (situation == HOOK_BEFORE_EVENT) {
        return 0;
    }

    return sched_worker_change(type, situation, data, c, w);
}

static int sched_worker_update(int type, int situation, void* data, va_list ap)
{
    cluster_t* c = va_arg(ap, cluster_t*);
    worker_t*  w = va_arg(ap, worker_t*);

    if (situation == HOOK_BEFORE_EVENT) {
        return 0;
    }

    return sched_worker_change(type, situation, data, c, w);
}
static int sched_worker_hook_fun(int type, int situation, int detail,
                                 void* data, va_list ap)
{
    switch (detail) {
    case HOOK_DETAIL_WORKER_JOIN:
        return sched_worker_join(type, situation, data, ap);
    case HOOK_DETAIL_WORKER_DEAD:
        return sched_worker_dead(type, situation, data, ap);
    case HOOK_DETAIL_WORKER_UPDATE:
        return sched_worker_update(type, situation, data, ap);
    }

    return 0;
}

static int sched_resource_hook_fun(int type, int situation, int detail,
                                   void* data, va_list ap)
{
    cluster_t*         c       = va_arg(ap, cluster_t*);
    xce_resval_list_t* resvals = va_arg(ap, xce_resval_list_t*);
    /* clang-format on */
    xce_notify_data_t xdata = { .event = NotifyData_data_schedUpdateResvals,
                                .data  = { .sched_update_resvals = resvals } };
    /* clang-format off */
    int rc = 0;

    if (detail != HOOK_DETAIL_RESOURCE_UPDATE) {
        return 0;
    }

    xce_use(resvals);

    mylock_lock(&(c->sched.lock));
    if (c->sched.ep != NULL) {
        switch (situation) {
        case HOOK_AFTER_EVENT:
            rc = notify(c, c->sched.ep, &xdata);
            break;
        default:
            break;
        }
    }
    mylock_unlock(&(c->sched.lock));

    return rc;
}

static int sched_queue_update(int     type,
                              int     situation,
                              void*   data,
                              va_list ap)
{
    cluster_t*   c = va_arg(ap, cluster_t*);
    queue_t* q = va_arg(ap, queue_t*);
    int          rc = 0;
    /* clang-format off */
    xce_notify_data_t xdata = {
	.event = NotifyData_data_schedUpdateQueue,
	.data = {
	    .sched_update_queue = &(q->queue)
	}
    };
    /* clang-format on */

    mylock_lock(&(c->sched.lock));
    if (c->sched.ep != NULL) {
        switch (situation) {
        case HOOK_AFTER_EVENT:
            rc = notify(c, c->sched.ep, &xdata);
            break;
        default:
            break;
        }
    }
    mylock_unlock(&(c->sched.lock));

    return rc;
}
static int sched_queue_hook_fun(int type, int situation, int detail, void* data,
                                va_list ap)
{
    switch (detail) {
    case HOOK_DETAIL_QUEUE_UPDATE:
        return sched_queue_update(type, situation, data, ap);
    default:
        break;
    }

    return 0;
}

int on_scheduler_join(cluster_t* c)
{
    static int hook_installed = 0;

    mylock_lock(&(c->sched.lock));
    if (c->sched.ep != NULL) {
        xcesched_push_configs(c);
        xcesched_push_workers(c);
        xcesched_push_jobs(c);
    }
    mylock_unlock(&(c->sched.lock));

    if (!hook_installed) {
        hook_installed = 1;

        xce_add_hook(&(c->hooksys), XCE_CONTROLLER_HOOK_JOB,
                     (hook_fn_t)sched_job_hook_fun, NULL);
        xce_add_hook(&(c->hooksys), XCE_CONTROLLER_HOOK_WORKER,
                     (hook_fn_t)sched_worker_hook_fun, NULL);
        xce_add_hook(&(c->hooksys), XCE_CONTROLLER_HOOK_RESOURCE,
                     (hook_fn_t)sched_resource_hook_fun, NULL);
        xce_add_hook(&(c->hooksys), XCE_CONTROLLER_HOOK_QUEUE,
                     (hook_fn_t)sched_queue_hook_fun, NULL);
    }
    return 0;
}

int handle_scheduler_leave(cluster_t* c)
{
    /* we need to mark the cluster status */
    TRACE("%s scheduler leave", __func__);
    mylock_lock(&(c->sched.lock));
    c->sched.ep = NULL;
    mylock_unlock(&(c->sched.lock));
    return 0;
}