save works

src/scheduler/include/decision.h

@@ -3,6 +3,7 @@
 #define _DECISION_H_ 1
 
 #include "xce/types.h"
+#include <stdint.h>
 
 typedef enum {
     XCE_DECISION_NONE = 0,
@@ -13,9 +14,13 @@ typedef enum {
 
 typedef struct {
     xce_decision_op_t op;
+    uint64_t          jobid;
     union {
         xce_job_alloc_t* alloc;
     } u;
 } xce_decision_t;
 
+extern xce_decision_t* make_pend_decision(uint64_t);
+extern xce_decision_t* make_dispatch_decision(xce_job_alloc_t*);
+
 #endif

src/scheduler/include/plugin.h

@@ -5,6 +5,7 @@
 #include <stdint.h>
 #include <stdlib.h>
 
+#include "decision.h"
 #include "sched/types.h"
 
 typedef void* plugin_data_t;
@@ -14,6 +15,7 @@ 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_job_hook_t)(void*, xce_sched_job_t*);
 typedef int (*plugin_update_worker_hook_t)(void*, xce_sched_worker_t*);
+typedef int (*plugin_commit_decision_hook_t)(void*, xce_decision_t*);
 
 /* initialize function of plugin */
 typedef int (*plugin_init_fn_t)(void*);
@@ -27,13 +29,14 @@ typedef int (*plugin_begin_session_fn_t)(void*);
 
 typedef int (*plugin_end_session_fn_t)(void*);
 /* allocation function of job */
-typedef xce_job_alloc_t* (*plugin_job_allocate_fn_t)(void*, int64_t*);
+typedef xce_job_alloc_t* (*plugin_job_allocate_fn_t)(void*, int64_t);
 
 typedef struct {
-    plugin_update_config_hook_t config_hook;
+    plugin_update_config_hook_t   config_hook;
-    plugin_update_queue_hook_t  queue_hook;
+    plugin_update_queue_hook_t    queue_hook;
-    plugin_update_job_hook_t    job_hook;
+    plugin_update_job_hook_t      job_hook;
-    plugin_update_worker_hook_t worker_hook;
+    plugin_update_worker_hook_t   worker_hook;
+    plugin_commit_decision_hook_t commit_hook;
 } plugin_hooks_t;
 
 typedef struct {
@@ -59,5 +62,6 @@ extern int  plugin_begin_session(void*);
 extern int  plugin_end_session(void*);
 
 extern int register_fcfs(void*);
+extern int register_default_allocator(void*);
 
 #endif

src/scheduler/include/sched/types.h

@@ -45,6 +45,7 @@ typedef struct {
     xce_sched_queue_t*   queue;
     xce_sched_user_t*    user;
     xce_sched_project_t* project;
+    void*                allocator;
     uint32_t             mintasks;
     uint32_t             maxtasks;
     xce_job_alloc_t*     allocs;

src/scheduler/include/scheduler.h

@@ -80,5 +80,6 @@ extern int sched_add_worker(xce_scheduler_t*, xce_sched_worker_t*);
 extern int sched_add_project(xce_scheduler_t*, xce_sched_project_t*);
 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*);
 
 #endif

src/scheduler/meson.build

@@ -6,6 +6,7 @@ scheduler_srcs = [
   'src/schedobj.c',
   'src/plugin.c',
   'src/plugin_fcfs.c',
+  'src/plugin_default_allocator.c',
 ]
 
 incdir = include_directories('src',

src/scheduler/src/decision.c

@@ -2,3 +2,33 @@
 #include <stdlib.h>
 
 #include "decision.h"
+
+xce_decision_t* make_pend_decision(uint64_t jobid)
+{
+    xce_decision_t* decision;
+
+    decision = (xce_decision_t*)calloc(1, sizeof(*decision));
+    if (decision == NULL) {
+        return NULL;
+    }
+
+    decision->op    = XCE_DECISION_PEND;
+    decision->jobid = jobid;
+
+    return decision;
+}
+
+xce_decision_t* make_dispatch_decision(xce_job_alloc_t* alloc)
+{
+    xce_decision_t* decision;
+    decision = (xce_decision_t*)calloc(1, sizeof(*decision));
+    if (decision == NULL) {
+        return NULL;
+    }
+
+    decision->op      = XCE_DECISION_DISPATCH;
+    decision->u.alloc = alloc;
+    decision->jobid   = alloc->jobid;
+
+    return decision;
+}

src/scheduler/src/plugin_default_allocator.c

@@ -0,0 +1,41 @@
+#include <errno.h>
+#include <pthread.h>
+#include <stdio.h>
+#include <stdlib.h>
+#include <string.h>
+
+#include "plugin.h"
+#include "scheduler.h"
+#include "utils.h"
+
+static int default_allocator_init(void* arg) { return 1; }
+
+static void default_allocator_fini(void* arg) { return; }
+
+static int my_update_job(xce_sched_plugin_t *plugin, xce_sched_job_t *j) {
+    /* FIXME: we need to add logic to check whether the job should be handled by me */
+    j->allocator = plugin;
+    return 1;
+}
+
+static xce_job_alloc_t* my_alloc(xce_sched_plugin_t* plugin, int64_t jobid)
+{
+    return NULL;
+}
+
+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 },
+    .init_fn          = (plugin_init_fn_t)default_allocator_init,
+    .fini_fn          = (plugin_fini_fn_t)default_allocator_fini,
+    .begin_session_fn = NULL,
+    .end_session_fn   = NULL,
+    .job_elect_fn     = NULL,
+    .job_alloc_fn     = (plugin_job_allocate_fn_t)my_alloc,
+    .data             = NULL
+};
+
+int register_default_allocator(void* sched)
+{
+    return plugin_register(sched, &default_allocator_plugin);
+}

src/scheduler/src/scheduler.c

@@ -7,6 +7,7 @@
 
 #include "client.h"
 #include "controller.capnp.h"
+#include "decision.h"
 #include "kvmgr.h"
 #include "objmgr.h"
 #include "plugin.h"
@@ -29,7 +30,37 @@ static void* sched_job_picker(void* arg)
 
 static xce_decision_t* calc_decision(xce_scheduler_t* sched, uint64_t jobid)
 {
-    return NULL;
+    xce_sched_job_t*    job    = sched_get_job(sched, jobid);
+    xce_sched_plugin_t* plugin = NULL;
+    xce_job_alloc_t*    alloc  = NULL;
+
+    if (job == NULL) {
+        ERROR("%s job %lu can not be found", __func__, jobid);
+        return NULL;
+    }
+
+    if (job->allocator == NULL) {
+        ERROR("%s job %lu has invalid allocator", __func__, jobid);
+        return NULL;
+    }
+
+    plugin = (xce_sched_plugin_t*)job->allocator;
+    if (plugin->job_alloc_fn == NULL) {
+        ERROR("%s plugin %s has no allocation function", __func__,
+              plugin->name);
+        return NULL;
+    }
+
+    alloc = plugin->job_alloc_fn(job->allocator, jobid);
+    if (alloc == NULL) {
+        return make_pend_decision(jobid);
+    }
+
+    return make_dispatch_decision(alloc);
+}
+
+static int commit_decision(xce_scheduler_t *sched, xce_decision_t *decision) {
+    return sched_commit_decision(sched, decision);
 }
 
 static void* sched_allocator(void* arg)
@@ -54,6 +85,8 @@ static void* sched_allocator(void* arg)
         /* calculate decision for jobid */
         decision = calc_decision(sched, jobid);
 
+        commit_decision(sched, decision);
+        
         cbuf_put(&(sched->decisions), &decision);
     }
 
@@ -98,6 +131,7 @@ static void sched_pend_update_killer(data_item_t* it)
 static int sched_register_plugins(xce_scheduler_t* sched)
 {
     register_fcfs(sched);
+    register_default_allocator(sched);
 
     return 0;
 }
@@ -366,3 +400,7 @@ xce_sched_project_t* sched_get_project(xce_scheduler_t* sched, char* name)
 int sched_job_is_ready(xce_sched_job_t* j) { return 1; }
 
 int sched_queue_is_ready(xce_sched_queue_t* q) { return 1; }
+
+int sched_commit_decision(xce_scheduler_t *sched, xce_decision_t *decision) {
+    return 0;
+}