scheduling job with fake resource

src/include/preason.h

@@ -5,6 +5,7 @@
 #include "xce/types.h"
 
 #define PREASON_BAD_RESREQ  (1)
+#define PREASON_NO_RESOURCE (2)
 
 extern void              erase_job_preason(xce_job_preason_t*);
 extern int               add_job_preason(xce_job_preason_t*, int, char*, char*);

src/include/xcealloc.h

@@ -0,0 +1,15 @@
+#if !defined(_XCEALLOC_H_)
+
+#define _XCEALLOC_H_ 1
+
+#include <stdarg.h>
+
+#include "xce/types.h"
+
+extern xce_alloc_t*       make_alloc(char*, char*, int, float, int);
+extern xce_host_alloc_t*  make_host_alloc(char*, int, xce_alloc_t**);
+extern xce_job_alloc_t*   make_job_alloc(uint64_t, int, xce_host_alloc_t**);
+extern xce_alloc_t**      make_allocs(int, ...);
+extern xce_host_alloc_t** make_host_allocs(int, ...);
+
+#endif

src/lib/meson.build

@@ -30,7 +30,8 @@ libxce_src = [ gen_srcs[0],
                'idgen.c',
                'delay.c',
                'ordq.c',
-               'preason.c'
+               'preason.c',
+               'xcealloc.c',
              ]
 
 incdir = include_directories('../include')

src/lib/xcealloc.c

@@ -0,0 +1,119 @@
+#include <errno.h>
+#include <stdarg.h>
+#include <stdio.h>
+#include <stdlib.h>
+#include <string.h>
+
+#include "xce/types.h"
+#include "xcealloc.h"
+
+xce_alloc_t* make_alloc(char* resname, char* reserve, int instid, float value,
+                        int duration)
+{
+    xce_alloc_t* alloc = NULL;
+
+    alloc = (xce_alloc_t*)calloc(1, sizeof(*alloc));
+    if (alloc == NULL) {
+        return NULL;
+    }
+
+    alloc->resname = strdup(resname);
+    if (reserve != NULL) {
+        alloc->reserve = strdup(reserve);
+    } else {
+        alloc->reserve = strdup("");
+    }
+
+    alloc->instid   = instid;
+    alloc->value    = value;
+    alloc->from     = time(NULL);
+    alloc->duration = duration;
+
+    return alloc;
+}
+
+xce_alloc_t** make_allocs(int nallocs, ...)
+{
+    xce_alloc_t** allocs = NULL;
+    int           i;
+    va_list       ap;
+
+    if (nallocs <= 0) {
+        return NULL;
+    }
+
+    allocs = (xce_alloc_t**)calloc(nallocs, sizeof(*allocs));
+    if (allocs == NULL) {
+        return NULL;
+    }
+
+    va_start(ap, nallocs);
+    for (i = 0; i < nallocs; i++) {
+        xce_alloc_t* alloc = va_arg(ap, xce_alloc_t*);
+
+        allocs[i] = alloc;
+    }
+    va_end(ap);
+
+    return allocs;
+}
+
+xce_host_alloc_t* make_host_alloc(char* host, int nalloc, xce_alloc_t** allocs)
+{
+    xce_host_alloc_t* halloc = NULL;
+
+    halloc = (xce_host_alloc_t*)calloc(1, sizeof(*halloc));
+    if (halloc == NULL) {
+        return NULL;
+    }
+
+    halloc->nallocs = nalloc;
+    halloc->allocs  = allocs;
+    halloc->host    = strdup(host);
+
+    return halloc;
+}
+
+xce_host_alloc_t** make_host_allocs(int nhosts, ...)
+{
+    xce_host_alloc_t** hallocs = NULL;
+    int                i;
+    va_list            ap;
+
+    if (nhosts == NULL) {
+        return NULL;
+    }
+
+    hallocs = (xce_host_alloc_t**)calloc(1, sizeof(*hallocs));
+    if (hallocs == NULL) {
+        return NULL;
+    }
+
+    va_start(ap, nhosts);
+    for (i = 0; i < nhosts; i++) {
+        xce_host_alloc_t* halloc = va_arg(ap, xce_host_alloc_t*);
+
+        hallocs[i] = halloc;
+    }
+
+    va_end(ap);
+
+    return hallocs;
+}
+
+xce_job_alloc_t* make_job_alloc(uint64_t jobid, int nhost,
+                                xce_host_alloc_t** hallocs)
+{
+    xce_job_alloc_t* jalloc = NULL;
+
+    jalloc = (xce_job_alloc_t*)calloc(1, sizeof(*jalloc));
+    if (jalloc == NULL) {
+        return NULL;
+    }
+
+    jalloc->jobid  = jobid;
+    jalloc->nhosts = nhost;
+    jalloc->allocs = hallocs;
+
+    return jalloc;
+}

src/scheduler/include/plugin.h

@@ -12,9 +12,12 @@ typedef void* plugin_data_t;
 
 /* hooker of plugin, the first parameter should be xce_sched_plugin_t */
 typedef int (*plugin_update_config_hook_t)(void*, xce_conf4sched_t*);
-typedef int (*plugin_update_queue_hook_t)(void*, xce_sched_queue_t*);
+typedef int (*plugin_update_queue_hook_t)(void*, xce_sched_queue_t*,
-typedef int (*plugin_update_job_hook_t)(void*, xce_sched_job_t*);
+                                          xce_sched_queue_t*);
-typedef int (*plugin_update_worker_hook_t)(void*, xce_sched_worker_t*);
+typedef int (*plugin_update_job_hook_t)(void*, xce_sched_job_t*,
+                                        xce_sched_job_t*);
+typedef int (*plugin_update_worker_hook_t)(void*, xce_sched_worker_t*,
+                                           xce_sched_worker_t*);
 typedef int (*plugin_commit_decision_hook_t)(void*, xce_decision_t*);
 
 /* initialize function of plugin */
@@ -56,9 +59,10 @@ typedef struct {
 extern int  plugin_register(void*, void*);
 extern void plugin_unregister(void*);
 extern int  plugin_config_update(void*, xce_conf4sched_t*);
-extern int  plugin_queue_update(void*, xce_sched_queue_t*);
+extern int  plugin_queue_update(void*, xce_sched_queue_t*, xce_sched_queue_t*);
-extern int  plugin_job_update(void*, xce_sched_job_t*);
+extern int  plugin_job_update(void*, xce_sched_job_t*, xce_sched_job_t*);
-extern int  plugin_worker_update(void*, xce_sched_worker_t*);
+extern int  plugin_worker_update(void*, xce_sched_worker_t*,
+                                 xce_sched_worker_t*);
 extern int  plugin_begin_session(void*);
 extern int  plugin_end_session(void*);
 extern int  plugin_commit_decision(void*, xce_decision_t*);

src/scheduler/include/sched/types.h

@@ -41,6 +41,7 @@ typedef struct {
     time_t               queue_time;
     time_t               start_time;
     unsigned int         state;
+    unsigned int         starting : 1;
     int32_t              priority;
     xce_sched_queue_t*   queue;
     xce_sched_user_t*    user;

src/scheduler/include/scheduler.h

@@ -82,5 +82,6 @@ extern int sched_job_is_ready(xce_sched_job_t*);
 extern int sched_queue_is_ready(xce_sched_queue_t*);
 extern int sched_commit_decision(xce_scheduler_t*, xce_decision_t*);
 extern int sched_publish_pend_reason(xce_scheduler_t*, xce_preason_set_t*);
+extern int sched_publish_dispatch(xce_scheduler_t*, xce_job_dispatch_t*);
 
 #endif

src/scheduler/src/main.c

@@ -122,13 +122,12 @@ int main(int argc, char* argv[])
         sched_go(&sched);
 
         while (1) {
-            xce_decision_t* decision = NULL;
+            xce_decision_t*    decision = NULL;
-            int rc = 0;
+            int                rc       = 0;
-            xce_pendreason_t *preasons[1];
+            xce_pendreason_t*  preasons[1];
-            xce_preason_set_t reasons = {
+            xce_preason_set_t  reasons = { .nreasons = 1, .reasons = preasons };
-                .nreasons = 1,
+            xce_job_alloc_t*   allocs[1];
-                .reasons = preasons
+            xce_job_dispatch_t jdispatch = { .njobs = 1, .allocs = allocs };
-            };
 
             cbuf_get(&(sched.decisions), &decision);
 
@@ -144,11 +143,13 @@ int main(int argc, char* argv[])
                 break;
             case XCE_DECISION_DISPATCH:
                 /* tell controller to dispatch job */
+                allocs[0] = decision->u.alloc;
+                rc        = sched_publish_dispatch(&sched, &jdispatch);
                 break;
             case XCE_DECISION_PEND:
                 /* set pend reason for jobs */
-                preasons[ 0] = decision->u.reason;
+                preasons[0] = decision->u.reason;
-                rc = sched_publish_pend_reason(&sched, &reasons);
+                rc          = sched_publish_pend_reason(&sched, &reasons);
                 break;
             default:
                 break;
@@ -158,7 +159,7 @@ int main(int argc, char* argv[])
         }
 
         sched_end_session(&sched);
-        
+
         if (sched.stop) {
             break;
         }

src/scheduler/src/plugin.c

@@ -7,6 +7,11 @@
 #include "scheduler.h"
 #include "xlist.h"
 
+typedef struct {
+    void* old;
+    void* new;
+} update_args_t;
+
 int plugin_register(void* arg1, void* arg2)
 {
     xce_scheduler_t*    sched  = (xce_scheduler_t*)arg1;
@@ -88,23 +93,26 @@ int plugin_config_update(void* arg, xce_conf4sched_t* config)
 static int plugin_job_update_(void* arg1, void* arg2)
 {
     xce_sched_plugin_t* plugin = (xce_sched_plugin_t*)arg2;
+    update_args_t*      args   = (update_args_t*)arg1;
 
     if ((arg1 == NULL) || (plugin == NULL)) {
         return 0;
     }
 
     if (plugin->hooks.job_hook != NULL) {
-        return plugin->hooks.job_hook(plugin, arg1);
+        return plugin->hooks.job_hook(plugin, (xce_sched_job_t*)args->old,
+                                      (xce_sched_job_t*)args->new);
     }
 
     return 0;
 }
 
-int plugin_job_update(void* arg, xce_sched_job_t* job)
+int plugin_job_update(void* arg, xce_sched_job_t* jold, xce_sched_job_t* job)
 {
     xce_scheduler_t* sched = (xce_scheduler_t*)arg;
     xlist_t          lst;
     int              rc;
+    update_args_t    args = { .old = jold, .new = job };
 
     if (sched == NULL) {
         return -EINVAL;
@@ -112,7 +120,7 @@ int plugin_job_update(void* arg, xce_sched_job_t* job)
 
     xlist_init(&lst, 3);
     xlist_store(&lst, plugin_job_update_);
-    rc = kvmgr_iter(&(sched->plugins), NULL, &lst, job);
+    rc = kvmgr_iter(&(sched->plugins), NULL, &lst, &args);
 
     xlist_fini(&lst);
 
@@ -122,23 +130,27 @@ int plugin_job_update(void* arg, xce_sched_job_t* job)
 static int plugin_queue_update_(void* arg1, void* arg2)
 {
     xce_sched_plugin_t* plugin = (xce_sched_plugin_t*)arg2;
+    update_args_t*      args   = (update_args_t*)arg1;
 
     if ((arg1 == NULL) || (plugin == NULL)) {
         return 0;
     }
 
     if (plugin->hooks.queue_hook != NULL) {
-        return plugin->hooks.queue_hook(plugin, arg1);
+        return plugin->hooks.queue_hook(plugin, (xce_sched_queue_t*)args->old,
+                                        (xce_sched_queue_t*)args->new);
     }
 
     return 0;
 }
 
-int plugin_queue_update(void* arg, xce_sched_queue_t* queue)
+int plugin_queue_update(void* arg, xce_sched_queue_t* qold,
+                        xce_sched_queue_t* queue)
 {
     xce_scheduler_t* sched = (xce_scheduler_t*)arg;
     xlist_t          lst;
     int              rc;
+    update_args_t    args = { .old = qold, .new = queue };
 
     if (sched == NULL) {
         return -EINVAL;
@@ -146,7 +158,7 @@ int plugin_queue_update(void* arg, xce_sched_queue_t* queue)
 
     xlist_init(&lst, 3);
     xlist_store(&lst, plugin_queue_update_);
-    rc = kvmgr_iter(&(sched->plugins), NULL, &lst, queue);
+    rc = kvmgr_iter(&(sched->plugins), NULL, &lst, &args);
 
     xlist_fini(&lst);
 
@@ -156,23 +168,27 @@ int plugin_queue_update(void* arg, xce_sched_queue_t* queue)
 static int plugin_worker_update_(void* arg1, void* arg2)
 {
     xce_sched_plugin_t* plugin = (xce_sched_plugin_t*)arg2;
+    update_args_t*      args   = (update_args_t*)arg1;
 
     if ((arg1 == NULL) || (plugin == NULL)) {
         return 0;
     }
 
     if (plugin->hooks.worker_hook != NULL) {
-        return plugin->hooks.worker_hook(plugin, arg1);
+        return plugin->hooks.worker_hook(plugin, (xce_sched_worker_t*)args->old,
+                                         (xce_sched_worker_t*)args->new);
     }
 
     return 0;
 }
 
-int plugin_worker_update(void* arg, xce_sched_worker_t* worker)
+int plugin_worker_update(void* arg, xce_sched_worker_t* wold,
+                         xce_sched_worker_t* worker)
 {
     xce_scheduler_t* sched = (xce_scheduler_t*)arg;
     xlist_t          lst;
     int              rc;
+    update_args_t    args = { .old = wold, .new = worker };
 
     if (sched == NULL) {
         return -EINVAL;
@@ -180,7 +196,7 @@ int plugin_worker_update(void* arg, xce_sched_worker_t* worker)
 
     xlist_init(&lst, 3);
     xlist_store(&lst, plugin_worker_update_);
-    rc = kvmgr_iter(&(sched->plugins), NULL, &lst, worker);
+    rc = kvmgr_iter(&(sched->plugins), NULL, &lst, &args);
 
     xlist_fini(&lst);
 
@@ -253,8 +269,9 @@ int plugin_end_session(void* arg)
     return rc;
 }
 
-static int plugin_commit_decision_(void* arg1, void* arg2) {
+static int plugin_commit_decision_(void* arg1, void* arg2)
-    xce_sched_plugin_t *plugin = (xce_sched_plugin_t *)arg2;
+{
+    xce_sched_plugin_t* plugin = (xce_sched_plugin_t*)arg2;
 
     if (plugin == NULL) {
         return 0;
@@ -263,7 +280,7 @@ static int plugin_commit_decision_(void* arg1, void* arg2) {
     if (plugin->hooks.commit_hook != NULL) {
         return plugin->hooks.commit_hook(plugin, arg1);
     }
-    
+
     return 0;
 }
 

src/scheduler/src/plugins/default_allocator.c

@@ -10,33 +10,93 @@
 #include "scheduler.h"
 #include "utils.h"
 #include "xce/types.h"
+#include "xcealloc.h"
+
+static uint32_t maxslots   = 8;
+static uint32_t slotsInUse = 0;
 
 static int default_allocator_init(xce_sched_plugin_t* plugin) { return 1; }
 
-static void default_allocator_fini(xce_sched_plugin_t* plugin) {
+static void default_allocator_fini(xce_sched_plugin_t* plugin)
+{
     if (plugin == NULL) {
         return;
     }
     FREEUP(plugin);
 }
 
-static int my_update_job(xce_sched_plugin_t* plugin, xce_sched_job_t* j)
+static int my_update_job(xce_sched_plugin_t* plugin, xce_sched_job_t* jold,
+                         xce_sched_job_t* j)
 {
     /* FIXME: we need to add logic to check whether the job should be handled by
      * me */
     j->allocator = plugin;
+    switch (j->state) {
+    case JOB_STAT_PEND:
+        break;
+    case JOB_STAT_RUN:
+        break;
+    case JOB_STAT_DONE:
+        break;
+    default:
+        break;
+    }
     return 1;
 }
 
 static xce_job_alloc_t* my_alloc(xce_sched_plugin_t* plugin, int64_t jobid,
                                  xce_pendreason_t** preason)
 {
-    xce_pendreason_t* reason = make_pendreason(jobid);
+    xce_job_alloc_t* jalloc = NULL;
 
-    add_job_preason(reason->reasons, PREASON_BAD_RESREQ, NULL, NULL);
+    *preason = NULL;
 
-    *preason = reason;
+    if (slotsInUse < maxslots) {
-    return NULL;
+        xce_alloc_t*       alloc   = NULL;
+        xce_alloc_t**      allocs  = NULL;
+        xce_host_alloc_t*  halloc  = NULL;
+        xce_host_alloc_t** hallocs = NULL;
+
+        alloc   = make_alloc("slots", "", -1, 1.0, -1);
+        allocs  = make_allocs(1, alloc);
+        halloc  = make_host_alloc("pc13v", 1, allocs);
+        hallocs = make_host_allocs(1, halloc);
+
+        jalloc = make_job_alloc(jobid, 1, hallocs);
+    } else {
+        xce_pendreason_t* reason = NULL;
+
+        reason = make_pendreason(jobid);
+        add_job_preason(reason->reasons, PREASON_NO_RESOURCE, "pc13v", "slots");
+
+        *preason = reason;
+    }
+
+    return jalloc;
+}
+
+static int my_commit(xce_sched_plugin_t* plugin, xce_decision_t* decision)
+{
+    int              rc    = 0;
+    uint64_t         jobid = 0;
+    xce_sched_job_t* job   = NULL;
+
+    switch (decision->op) {
+    case XCE_DECISION_DISPATCH:
+        /* FIXME: */
+        jobid = decision->u.alloc->jobid;
+        job   = sched_get_job(plugin->sched, jobid);
+        if (job != NULL) {
+            slotsInUse++;
+            job->starting = 1;
+        }
+
+        break;
+    default:
+        break;
+    }
+
+    return rc;
 }
 
 int register_default_allocator(void* sched)
@@ -44,7 +104,8 @@ int register_default_allocator(void* sched)
     static xce_sched_plugin_t default_allocator_plugin = {
         .name             = "xce_sched_default_allocator",
         .hooks            = { .config_hook = NULL,
-                             .job_hook    = (plugin_update_job_hook_t)my_update_job },
+                             .job_hook    = (plugin_update_job_hook_t)my_update_job,
+                             .commit_hook = (plugin_commit_decision_hook_t)my_commit },
         .init_fn          = (plugin_init_fn_t)default_allocator_init,
         .fini_fn          = (plugin_fini_fn_t)default_allocator_fini,
         .begin_session_fn = NULL,
@@ -54,9 +115,9 @@ int register_default_allocator(void* sched)
         .data             = NULL
     };
 
-    xce_sched_plugin_t *plugin = NULL;
+    xce_sched_plugin_t* plugin = NULL;
 
-    plugin= (xce_sched_plugin_t *)calloc(1, sizeof(*plugin));
+    plugin = (xce_sched_plugin_t*)calloc(1, sizeof(*plugin));
     if (plugin == NULL) {
         return -ENOMEM;
     }

src/scheduler/src/plugins/fcfs.c

@@ -124,7 +124,8 @@ static int job_match(void* arg1, va_list ap)
     return (j->job->jobid == jobid);
 }
 
-static int fcfs_update_queue(xce_sched_plugin_t* plugin, xce_sched_queue_t* q)
+static int fcfs_update_queue(xce_sched_plugin_t* plugin,
+                             xce_sched_queue_t* qold, xce_sched_queue_t* q)
 {
     fcfs_plugin_data_t* privdata = (fcfs_plugin_data_t*)plugin->data;
     ordq_item_t*        it;
@@ -164,7 +165,8 @@ static fcfs_queue_t* fcfs_get_queue(fcfs_plugin_data_t* data, char* name)
     return (fcfs_queue_t*)it->data;
 }
 
-static int fcfs_update_job(xce_sched_plugin_t* plugin, xce_sched_job_t* j)
+static int fcfs_update_job(xce_sched_plugin_t* plugin, xce_sched_job_t* jold,
+                           xce_sched_job_t* j)
 {
     fcfs_plugin_data_t* privdata = (fcfs_plugin_data_t*)plugin->data;
     fcfs_queue_t*       q        = NULL;

src/scheduler/src/scheduler.c

@@ -206,7 +206,8 @@ int sched_init(xce_scheduler_t* sched, xce_client_config_t* config,
 
 void sched_fini(xce_scheduler_t* sched)
 {
-    void* res;
+    void*    res;
+    uint64_t jobid;
 
     if (sched == NULL) {
         return;
@@ -216,6 +217,8 @@ void sched_fini(xce_scheduler_t* sched)
     pthread_join(sched->jobpicker, &res);
 
     pthread_cancel(sched->allocator);
+    /* try to wake up allocator */
+    cbuf_tryput(&(sched->jbuf), &jobid);
     pthread_join(sched->allocator, &res);
 
     xcecli_wakeup(sched->apictx, (int)-EINTR);
@@ -241,9 +244,11 @@ static int sched_update_config(xce_scheduler_t* sched, xce_conf4sched_t* config)
     plugin_config_update(sched, config);
 
     for (i = 0; i < config->n_queues; i++) {
+        xce_sched_queue_t* qold
+            = sched_get_queue(sched, config->queues[i]->queue);
         xce_sched_queue_t* q = make_sched_queue(sched, config->queues[i]);
 
-        plugin_queue_update(sched, q);
+        plugin_queue_update(sched, qold, q);
     }
 
     return 0;
@@ -255,9 +260,10 @@ static int sched_update_jobs(xce_scheduler_t* sched, int njobs,
     int i;
 
     for (i = 0; i < njobs; i++) {
-        xce_sched_job_t* j = make_sched_job(sched, jobs[i]);
+        xce_sched_job_t* jold = sched_get_job(sched, jobs[i]->jobid);
+        xce_sched_job_t* j    = make_sched_job(sched, jobs[i]);
 
-        plugin_job_update(sched, j);
+        plugin_job_update(sched, jold, j);
     }
 
     return 0;
@@ -269,9 +275,10 @@ static int sched_update_workers(xce_scheduler_t* sched, int nworkers,
     int i;
 
     for (i = 0; i < nworkers; i++) {
-        xce_sched_worker_t* w = make_sched_worker(sched, workers[i]);
+        xce_sched_worker_t* wold = sched_get_worker(sched, workers[i]->name);
+        xce_sched_worker_t* w    = make_sched_worker(sched, workers[i]);
 
-        plugin_worker_update(sched, w);
+        plugin_worker_update(sched, wold,w);
     }
 
     return 0;
@@ -413,3 +420,8 @@ int sched_publish_pend_reason(xce_scheduler_t*   sched,
 {
     return xcecli_set_pend_reason(sched->apictx, reasons);
 }
+
+int sched_publish_dispatch(xce_scheduler_t* sched, xce_job_dispatch_t* dispatch)
+{
+    return xcecli_jobdispatch(sched->apictx, dispatch);
+}