save

src/lib/kvmgr.c

@@ -50,6 +50,10 @@ static void* kvmgr_lock_(kvmgr_t* mgr, void* key, int op)
         return NULL;
     }
 
+    if (op == OBJREF_NOP) {
+        return ptr;
+    }
+    
     return obj_ref(ptr, op);
 }
 
@@ -101,6 +105,10 @@ void* kvmgr_lock_byid(kvmgr_t* mgr, int idx, int op)
         return NULL;
     }
 
+    if (op == OBJREF_NOP) {
+        return ptr;
+    }
+    
     return obj_ref(ptr, op);
 }
 

src/scheduler/include/plugin.h

@@ -0,0 +1,56 @@
+#if !defined(_PLUGIN_H_)
+
+#define _PLUGIN_H_ 1
+
+#include <stdint.h>
+#include <stdlib.h>
+
+#include "sched/types.h"
+
+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_job_hook_t)(void*, xce_sched_job_t*);
+typedef int (*plugin_update_worker_hook_t)(void*, xce_sched_worker_t*);
+
+/* initialize function of plugin */
+typedef int (*plugin_init_fn_t)(void*);
+
+typedef void (*plugin_fini_fn_t)(void*);
+
+/* plugin function to elect job to be scheduler */
+typedef int64_t (*plugin_job_election_fn_t)(void*);
+
+/* allocation function of job */
+typedef xce_job_alloc_t* (*plugin_job_allocate_fn_t)(void*, int64_t*);
+
+typedef struct {
+    plugin_update_config_hook_t config_hook;
+    plugin_update_queue_hook_t  queue_hook;
+    plugin_update_job_hook_t    job_hook;
+    plugin_update_worker_hook_t worker_hook;
+} plugin_hooks_t;
+
+typedef struct {
+    char*                    name;
+    void*                    sched;
+    plugin_init_fn_t         init_fn;
+    plugin_fini_fn_t         fini_fn;
+    plugin_job_election_fn_t job_elect_fn;
+    plugin_job_allocate_fn_t job_alloc_fn;
+    plugin_hooks_t           hooks;
+    plugin_data_t            data;
+} xce_sched_plugin_t;
+
+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_job_update(void*, xce_sched_job_t*);
+extern int  plugin_worker_update(void*, xce_sched_worker_t*);
+
+extern int register_fcfs(void*);
+
+#endif

src/scheduler/include/sched/types.h

@@ -0,0 +1,69 @@
+#if !defined(_SCHED_TYPES_H_)
+
+#define _SCHED_TYPES_H_ 1
+
+#include <stdint.h>
+#include <stdlib.h>
+
+#include "typemap.h"
+#include "xce/types.h"
+
+typedef void* host_selection_t;
+
+typedef struct {
+    host_selection_t* select;
+    int               nres;
+    uint32_t*         resources;
+    float*            amount;
+} rank_req_t;
+
+typedef struct {
+    char* name;
+} xce_sched_user_t;
+
+typedef struct {
+    char* name;
+} xce_sched_project_t;
+
+typedef struct {
+    char*    name;
+    uint32_t attrs;
+    uint32_t status;
+    int32_t  priority;
+    int32_t  accept_interval;
+    int32_t  sched_delay;
+    char*    resreq;
+} xce_sched_queue_t;
+
+typedef struct {
+    uint64_t             jobid;
+    time_t               sub_time;
+    time_t               queue_time;
+    time_t               start_time;
+    unsigned int         state;
+    int32_t              priority;
+    xce_sched_queue_t*   queue;
+    xce_sched_user_t*    user;
+    xce_sched_project_t* project;
+    uint32_t             mintasks;
+    uint32_t             maxtasks;
+    xce_job_alloc_t*     allocs;
+} xce_sched_job_t;
+
+typedef struct {
+    char*      name;
+    xce_val_t* res_avail;  /* resources in worker */
+    xce_val_t* res_in_use; /* used resources in worker */
+} xce_sched_worker_t;
+
+extern xce_sched_job_t*     make_sched_job(void*, xce_job_t*);
+extern xce_sched_queue_t*   make_sched_queue(void*, xce_queue_t*);
+extern xce_sched_worker_t*  make_sched_worker(void*, xce_worker_t*);
+extern xce_sched_user_t*    make_sched_user(void*, char*);
+extern xce_sched_project_t* make_sched_project(void*, char*);
+extern void                 kill_sched_job(void*);
+extern void                 kill_sched_queue(void*);
+extern void                 kill_sched_worker(void*);
+extern void                 kill_sched_user(void*);
+extern void                 kill_sched_project(void*);
+#endif

src/scheduler/include/scheduler.h

@@ -9,7 +9,9 @@
 #include "client.h"
 #include "dataq.h"
 #include "decision.h"
+#include "kvmgr.h"
 #include "mylock.h"
+#include "plugin.h"
 #include "xce/types.h"
 #include "xlist.h"
 
@@ -41,50 +43,17 @@ typedef struct {
                                    - offsetof(xce_sched_update_request_t,      \
                                               link)))
 
-typedef void* host_selection_t;
-
-typedef struct {
-    host_selection_t* select;
-    int               nres;
-    uint32_t*         resources;
-    float*            amount;
-} rank_req_t;
-
-typedef struct {
-    uint64_t         jobid;
-    time_t           sub_time;
-    time_t           queue_time;
-    time_t           start_time;
-    unsigned int     state;
-    char*            queue;
-    char*            user;
-    char*            project;
-    uint32_t         mintasks;
-    uint32_t         maxtasks;
-    xce_job_alloc_t* allocs;
-} xce_sched_job_t;
-
-typedef struct {
-    char*    name;
-    uint32_t attrs;
-    uint32_t status;
-    int32_t  priority;
-    int32_t  accept_interval;
-    int32_t  sched_delay;
-    char*    resreq;
-} xce_sched_queue_t;
-
-typedef struct {
-    char*      name;
-    xce_val_t* res_avail;  /* resources in worker */
-    xce_val_t* res_in_use; /* used resources in worker */
-} xce_sched_worker_t;
-
 typedef struct {
     mylock_t     lock;
     int          stop;
     xcecli_ctx_t apictx;
+    kvmgr_t      plugins;
     dataq_t*     pend_updates[2];
+    kvmgr_t      jobs;
+    kvmgr_t      queues;
+    kvmgr_t      workers;
+    kvmgr_t      users;
+    kvmgr_t      projects;
     pthread_t    jobpicker;
     pthread_t    allocator;
     cbuf_t       wbuf;      /* action to trigger scheduling */
@@ -92,10 +61,21 @@ typedef struct {
     cbuf_t       decisions; /* scheduler decisions */
 } xce_scheduler_t;
 
-extern int  sched_init(xce_scheduler_t*, xce_client_config_t*,
-                       xce_request_handler_t);
-extern void sched_fini(xce_scheduler_t*);
-extern int  sched_update(xce_scheduler_t*);
-extern void sched_go(xce_scheduler_t*);
+extern int                  sched_init(xce_scheduler_t*, xce_client_config_t*,
+                                       xce_request_handler_t);
+extern void                 sched_fini(xce_scheduler_t*);
+extern int                  sched_update(xce_scheduler_t*);
+extern void                 sched_go(xce_scheduler_t*);
+extern xce_sched_job_t*     sched_get_job(xce_scheduler_t*, uint64_t);
+extern xce_sched_queue_t*   sched_get_queue(xce_scheduler_t*, char*);
+extern xce_sched_worker_t*  sched_get_worker(xce_scheduler_t*, char*);
+extern xce_sched_user_t*    sched_get_user(xce_scheduler_t*, char*);
+extern xce_sched_project_t* sched_get_project(xce_scheduler_t*, char*);
+extern int                  sched_add_job(xce_scheduler_t*, xce_sched_job_t*);
+extern int sched_add_queue(xce_scheduler_t*, xce_sched_queue_t*);
+extern int sched_add_user(xce_scheduler_t*, xce_sched_user_t*);
+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*);
 
 #endif

src/scheduler/meson.build

@@ -3,6 +3,9 @@ scheduler_srcs = [
   'src/handler.c',
   'src/scheduler.c',
   'src/decision.c',
+  'src/schedobj.c',
+  'src/plugin.c',
+  'src/plugin_fcfs.c',
 ]
 
 incdir = include_directories('src',

src/scheduler/src/plugin.c

@@ -0,0 +1,188 @@
+#include <errno.h>
+#include <stdint.h>
+#include <stdio.h>
+#include <stdlib.h>
+
+#include "plugin.h"
+#include "scheduler.h"
+#include "xlist.h"
+
+int plugin_register(void* arg1, void* arg2)
+{
+    xce_scheduler_t*    sched  = (xce_scheduler_t*)arg1;
+    xce_sched_plugin_t* plugin = (xce_sched_plugin_t*)arg2;
+    int                 rc     = 0;
+
+    if ((sched == NULL) && (plugin == NULL)) {
+        return -EINVAL;
+    }
+
+    plugin->sched = sched;
+    if (plugin->init_fn != NULL) {
+        rc = plugin->init_fn(plugin);
+    }
+
+    if (rc < 0) {
+        return rc;
+    }
+
+    rc = kvmgr_add(&(sched->plugins), (void*)plugin->name, (void*)plugin);
+
+    return rc;
+}
+
+void plugin_unregister(void* arg)
+{
+    xce_sched_plugin_t* plugin = (xce_sched_plugin_t*)arg;
+    xce_scheduler_t*    sched  = NULL;
+
+    if (plugin == NULL) {
+        return;
+    }
+
+    sched = (xce_scheduler_t*)plugin->sched;
+    if (sched == NULL) {
+        return;
+    }
+
+    kvmgr_del(&(sched->plugins), plugin->name);
+    if (plugin->fini_fn != NULL) {
+        plugin->fini_fn(plugin);
+    }
+}
+
+static int plugin_config_update_(void* arg1, void* arg2)
+{
+    xce_sched_plugin_t* plugin = (xce_sched_plugin_t*)arg2;
+
+    if ((arg1 == NULL) || (plugin == NULL)) {
+        return 0;
+    }
+
+    if (plugin->hooks.config_hook != NULL) {
+        return plugin->hooks.config_hook(plugin, arg1);
+    }
+
+    return 0;
+}
+
+int plugin_config_update(void* arg, xce_conf4sched_t* config)
+{
+    xce_scheduler_t* sched = (xce_scheduler_t*)arg;
+    xlist_t          lst;
+    int              rc;
+
+    if (sched == NULL) {
+        return -EINVAL;
+    }
+
+    xlist_init(&lst, 3);
+    xlist_store(&lst, plugin_config_update_);
+    rc = kvmgr_iter(&(sched->plugins), NULL, &lst, config);
+
+    xlist_fini(&lst);
+
+    return rc;
+}
+
+static int plugin_job_update_(void* arg1, void* arg2)
+{
+    xce_sched_plugin_t* plugin = (xce_sched_plugin_t*)arg2;
+
+    if ((arg1 == NULL) || (plugin == NULL)) {
+        return 0;
+    }
+
+    if (plugin->hooks.job_hook != NULL) {
+        return plugin->hooks.job_hook(plugin, arg1);
+    }
+
+    return 0;
+}
+
+int plugin_job_update(void* arg, xce_sched_job_t* job)
+{
+    xce_scheduler_t* sched = (xce_scheduler_t*)arg;
+    xlist_t          lst;
+    int              rc;
+
+    if (sched == NULL) {
+        return -EINVAL;
+    }
+
+    xlist_init(&lst, 3);
+    xlist_store(&lst, plugin_job_update_);
+    rc = kvmgr_iter(&(sched->plugins), NULL, &lst, job);
+
+    xlist_fini(&lst);
+
+    return rc;
+}
+
+static int plugin_queue_update_(void* arg1, void* arg2)
+{
+    xce_sched_plugin_t* plugin = (xce_sched_plugin_t*)arg2;
+
+    if ((arg1 == NULL) || (plugin == NULL)) {
+        return 0;
+    }
+
+    if (plugin->hooks.queue_hook != NULL) {
+        return plugin->hooks.queue_hook(plugin, arg1);
+    }
+
+    return 0;
+}
+
+int plugin_queue_update(void* arg, xce_sched_queue_t* queue)
+{
+    xce_scheduler_t* sched = (xce_scheduler_t*)arg;
+    xlist_t          lst;
+    int              rc;
+
+    if (sched == NULL) {
+        return -EINVAL;
+    }
+
+    xlist_init(&lst, 3);
+    xlist_store(&lst, plugin_queue_update_);
+    rc = kvmgr_iter(&(sched->plugins), NULL, &lst, queue);
+
+    xlist_fini(&lst);
+
+    return rc;
+}
+
+static int plugin_worker_update_(void* arg1, void* arg2)
+{
+    xce_sched_plugin_t* plugin = (xce_sched_plugin_t*)arg2;
+
+    if ((arg1 == NULL) || (plugin == NULL)) {
+        return 0;
+    }
+
+    if (plugin->hooks.worker_hook != NULL) {
+        return plugin->hooks.worker_hook(plugin, arg1);
+    }
+
+    return 0;
+}
+
+int plugin_worker_update(void* arg, xce_sched_worker_t* worker)
+{
+    xce_scheduler_t* sched = (xce_scheduler_t*)arg;
+    xlist_t          lst;
+    int              rc;
+
+    if (sched == NULL) {
+        return -EINVAL;
+    }
+
+    xlist_init(&lst, 3);
+    xlist_store(&lst, plugin_worker_update_);
+    rc = kvmgr_iter(&(sched->plugins), NULL, &lst, worker);
+
+    xlist_fini(&lst);
+
+    return rc;
+}

src/scheduler/src/plugin_fcfs.c

@@ -0,0 +1,266 @@
+#include <errno.h>
+#include <stdint.h>
+#include <stdio.h>
+#include <stdlib.h>
+
+#include "ordq.h"
+#include "plugin.h"
+#include "sched/types.h"
+#include "scheduler.h"
+#include "utils.h"
+
+typedef struct {
+    xce_sched_queue_t* queue;
+    ordq_t             pendJobs;
+} fcfs_queue_t;
+
+typedef struct {
+    xce_sched_job_t* job;
+} fcfs_job_t;
+
+typedef struct {
+    ordq_t queues;
+} fcfs_plugin_data_t;
+
+static void fcfs_job_killer(void* arg)
+{
+    if (arg == NULL) {
+        return;
+    }
+
+    free(arg);
+}
+
+static void fcfs_queue_killer(void* arg)
+{
+    fcfs_queue_t* q = (fcfs_queue_t*)arg;
+
+    if (q == NULL) {
+        return;
+    }
+
+    ordq_fini(&(q->pendJobs));
+    q->queue = NULL;
+
+    FREEUP(q);
+}
+
+static int fcfs_queue_cmp(void* ctx, void* arg1, void* arg2)
+{
+    fcfs_queue_t* q1 = (fcfs_queue_t*)arg1;
+    fcfs_queue_t* q2 = (fcfs_queue_t*)arg2;
+
+    if (q1->queue->priority > q2->queue->priority) {
+        return -1;
+    } else if (q1->queue->priority < q2->queue->priority) {
+        return 1;
+    }
+
+    return 0;
+}
+
+static int fcfs_job_cmp(void* ctx, void* arg1, void* arg2)
+{
+    fcfs_job_t* j1 = (fcfs_job_t*)arg1;
+    fcfs_job_t* j2 = (fcfs_job_t*)arg2;
+
+    if (j1->job->priority > j2->job->priority) {
+        return -1;
+    } else if (j1->job->priority < j2->job->priority) {
+        return 1;
+    } else if (j1->job->queue_time < j2->job->queue_time) {
+        return -1;
+    } else if (j1->job->queue_time > j2->job->queue_time) {
+        return 1;
+    }
+
+    return 0;
+}
+
+static int fcfs_priv_data_init(fcfs_plugin_data_t* data)
+{
+    ordq_init(&(data->queues), (ordq_data_killer_t)fcfs_queue_killer,
+              (ordq_cmp_fn_t)fcfs_queue_cmp);
+    return 0;
+}
+
+static int fcfs_update_config(void* arg, xce_conf4sched_t* config)
+{
+    xce_sched_plugin_t* plugin   = (xce_sched_plugin_t*)arg;
+    fcfs_plugin_data_t* privdata = NULL;
+
+    TRACE("%s ...", __func__);
+
+    privdata = (fcfs_plugin_data_t*)plugin->data;
+    if (privdata == NULL) {
+        privdata = (fcfs_plugin_data_t*)calloc(1, sizeof(*privdata));
+        if (privdata == NULL) {
+            return -ENOMEM;
+        }
+
+        fcfs_priv_data_init(privdata);
+        plugin->data = privdata;
+    }
+
+    return 0;
+}
+
+static int skip_not_matched_queue(void* arg1, void* arg2)
+{
+    char*         name = (char*)arg1;
+    fcfs_queue_t* q    = (fcfs_queue_t*)arg2;
+
+    if ((name == NULL) || (q == NULL)) {
+        return 1;
+    }
+
+    return (strcmp(name, q->queue->name) != 0);
+}
+
+static int fcfs_update_queue(void* arg, xce_sched_queue_t* q)
+{
+    xce_sched_plugin_t* plugin   = (xce_sched_plugin_t*)arg;
+    fcfs_plugin_data_t* privdata = (fcfs_plugin_data_t*)plugin->data;
+    ordq_item_t*        it;
+    fcfs_queue_t*       queue = NULL;
+
+    TRACE("%s queue %s...", __func__, q->name);
+
+    privdata->queues.ctx = q->name;
+    it = ordq_get(&(privdata->queues), (ordq_skip_fn_t)skip_not_matched_queue);
+    if (it == NULL) {
+        queue = (fcfs_queue_t*)calloc(1, sizeof(*queue));
+        if (queue == NULL) {
+            return -ENOMEM;
+        }
+
+        queue->queue = q;
+        ordq_init(&(queue->pendJobs), (ordq_data_killer_t)fcfs_job_killer,
+                  (ordq_cmp_fn_t)fcfs_job_cmp);
+    } else {
+        queue = (fcfs_queue_t*)it->data;
+        free(it);
+    }
+
+    ordq_put(&(privdata->queues), (void*)queue);
+
+    return 0;
+}
+
+static fcfs_queue_t* fcfs_get_queue(fcfs_plugin_data_t* data, char* name)
+{
+    ordq_item_t* it = NULL;
+
+    data->queues.ctx = name;
+    it = ordq_peek(&(data->queues), (ordq_skip_fn_t)skip_not_matched_queue);
+    if (it == NULL) {
+        return NULL;
+    }
+
+    return (fcfs_queue_t*)it->data;
+}
+
+static int fcfs_update_job(void* arg, xce_sched_job_t* j)
+{
+    xce_sched_plugin_t* plugin   = (xce_sched_plugin_t*)arg;
+    fcfs_plugin_data_t* privdata = (fcfs_plugin_data_t*)plugin->data;
+    fcfs_queue_t*       q        = NULL;
+    fcfs_job_t*         job      = NULL;
+
+    TRACE("%s...", __func__);
+    if (privdata == NULL) {
+        return -EINVAL;
+    }
+
+    if ((j->state & JOB_STAT_PEND) == 0) {
+        /* skip non pend job */
+        return 0;
+    }
+
+    q = fcfs_get_queue(privdata, j->queue->name);
+    if (q == NULL) {
+        return -EINVAL;
+    }
+
+    job = (fcfs_job_t*)calloc(1, sizeof(*job));
+    if (job == NULL) {
+        return -ENOMEM;
+    }
+
+    job->job = j;
+
+    ordq_put(&(q->pendJobs), job);
+
+    return 0;
+}
+
+static int skip_no_pend_job_queue(void* arg1, void* arg2)
+{
+    fcfs_queue_t* q = (fcfs_queue_t*)arg2;
+
+    if (q == NULL) {
+        return 1;
+    }
+
+    return (ordq_peek(&(q->pendJobs), NULL) == NULL);
+}
+
+static int skip_not_ready_job(void* arg1, void* arg2)
+{
+    fcfs_job_t* job = (fcfs_job_t*)arg2;
+
+    return !sched_job_is_ready(job->job);
+}
+
+static int64_t fcfs_elect_job(void* arg)
+{
+    xce_sched_plugin_t* plugin   = (xce_sched_plugin_t*)arg;
+    fcfs_plugin_data_t* privdata = (fcfs_plugin_data_t*)plugin->data;
+    ordq_item_t*        it       = NULL;
+    fcfs_queue_t*       q        = NULL;
+    fcfs_job_t*         job      = NULL;
+    uint64_t            jobid;
+
+    TRACE("%s ...", __func__);
+
+    it = ordq_peek(&(privdata->queues), (ordq_skip_fn_t)skip_no_pend_job_queue);
+    if (it == NULL) {
+        TRACE("%s no job to be elected", __func__);
+
+        return 0;
+    }
+
+    q  = (fcfs_queue_t*)it->data;
+    it = ordq_get(&(q->pendJobs), skip_not_ready_job);
+    if (it == NULL) {
+        TRACE("%s no job to be elected", __func__);
+
+        return 0;
+    }
+
+    job   = (fcfs_job_t*)it->data;
+    jobid = job->job->jobid;
+    free(job);
+    free(it);
+
+    if (1) {
+        char idstring[128];
+
+        fmt_jobid(idstring, sizeof(idstring) - 1, jobid);
+        TRACE("%s job %s has been elected", __func__, idstring);
+    }
+    return jobid;
+}
+
+static xce_sched_plugin_t fcfs_plugin = {
+    .name         = "xce_sched_fcfs",
+    .hooks        = { .config_hook = fcfs_update_config,
+                     .queue_hook  = fcfs_update_queue,
+                     .job_hook    = fcfs_update_job,
+                     .worker_hook = NULL },
+    .job_elect_fn = fcfs_elect_job,
+    .job_alloc_fn = NULL,
+    .data         = NULL
+};
+
+int register_fcfs(void* sched) { return plugin_register(sched, &fcfs_plugin); }

src/scheduler/src/schedobj.c

@@ -0,0 +1,131 @@
+#include <stdlib.h>
+#include <string.h>
+
+#include "sched/types.h"
+#include "scheduler.h"
+#include "xce/types.h"
+
+void kill_sched_job(void* arg)
+{
+    if (arg == NULL) {
+        return;
+    }
+}
+
+xce_sched_job_t* make_sched_job(void* arg, xce_job_t* j)
+{
+    xce_sched_job_t* job   = (xce_sched_job_t*)calloc(1, sizeof(*job));
+    xce_scheduler_t* sched = (xce_scheduler_t*)arg;
+
+    if ((sched == NULL) || (job == NULL)) {
+        return NULL;
+    }
+
+    job->jobid      = j->jobid;
+    job->sub_time   = j->sub_time;
+    job->queue_time = j->queue_time;
+    job->start_time = j->start_time;
+    job->state      = j->state;
+    job->user       = sched_get_user(sched, j->submit->user);
+    job->queue      = sched_get_queue(sched, j->submit->queue);
+    job->project    = sched_get_project(sched, j->submit->project_name);
+    job->mintasks   = j->submit->mintasks;
+    job->maxtasks   = j->submit->maxtasks;
+    job->priority   = 0; /* FIXME: support job level priority */
+
+    sched_add_job(sched, job);
+
+    return job;
+}
+
+void kill_sched_queue(void* arg)
+{
+    if (arg == NULL) {
+        return;
+    }
+}
+
+xce_sched_queue_t* make_sched_queue(void* arg, xce_queue_t* q)
+{
+    xce_scheduler_t*   sched = (xce_scheduler_t*)arg;
+    xce_sched_queue_t* queue = NULL;
+
+    if ((sched == NULL) || (q == NULL)) {
+        return NULL;
+    }
+
+    queue = (xce_sched_queue_t*)calloc(1, sizeof(*queue));
+    if (queue == NULL) {
+        return NULL;
+    }
+
+    queue->name            = strdup(q->queue);
+    queue->attrs           = q->attrs;
+    queue->status          = q->status;
+    queue->priority        = q->priority;
+    queue->accept_interval = q->accept_interval;
+    queue->sched_delay     = q->sched_delay;
+
+    sched_add_queue(sched, queue);
+    
+    return queue;
+}
+
+void kill_sched_worker(void* arg)
+{
+    if (arg == NULL) {
+        return;
+    }
+}
+
+xce_sched_worker_t* make_sched_worker(void* arg, xce_worker_t* w)
+{
+    xce_scheduler_t*    sched  = (xce_scheduler_t*)arg;
+    xce_sched_worker_t* worker = NULL;
+
+    if ((sched == NULL) || (w == NULL)) {
+        return NULL;
+    }
+
+    worker = (xce_sched_worker_t*)calloc(1, sizeof(*worker));
+    if (worker == NULL) {
+        return NULL;
+    }
+
+    worker->name = strdup(w->name);
+    /* FIXME: */
+
+    sched_add_worker(sched, worker);
+
+    return worker;
+}
+
+void kill_sched_user(void* arg)
+{
+    if (arg == NULL) {
+        return;
+    }
+}
+
+xce_sched_user_t* make_sched_user(void* arg, char* name)
+{
+    xce_scheduler_t*  sched = (xce_scheduler_t*)arg;
+    xce_sched_user_t* user  = NULL;
+
+    return user;
+}
+
+void kill_sched_project(void* arg)
+{
+    if (arg == NULL) {
+        return;
+    }
+}
+
+xce_sched_project_t* make_sched_project(void* arg, char* name)
+{
+    xce_scheduler_t*     sched   = (xce_scheduler_t*)arg;
+    xce_sched_project_t* project = NULL;
+
+    return project;
+}

src/scheduler/src/scheduler.c

@@ -7,12 +7,32 @@
 
 #include "client.h"
 #include "controller.capnp.h"
+#include "kvmgr.h"
+#include "objmgr.h"
+#include "plugin.h"
 #include "scheduler.h"
 #include "utils.h"
 
 static uint64_t pick_job(xce_scheduler_t* sched)
 {
-    uint64_t jobid = 0;
+    static xce_sched_plugin_t* job_picker = NULL;
+    uint64_t                   jobid      = 0;
+
+    if (job_picker == NULL) {
+        char* job_picker_plugin = "xce_sched_fcfs";
+
+        /* set job_pick_plugin according configuration */
+        job_picker
+            = kvmgr_lock(&(sched->plugins), job_picker_plugin, OBJREF_NOP);
+        if ((job_picker == NULL) || (job_picker->job_elect_fn == NULL)) {
+            /* something wrong */
+            return 0;
+        }
+    }
+
+    if (job_picker->job_elect_fn != NULL) {
+        jobid = job_picker->job_elect_fn(job_picker);
+    }
 
     return jobid;
 }
@@ -111,6 +131,13 @@ static void sched_pend_update_killer(data_item_t* it)
     free(req);
 }
 
+static int sched_register_plugins(xce_scheduler_t* sched)
+{
+    register_fcfs(sched);
+
+    return 0;
+}
+
 int sched_init(xce_scheduler_t* sched, xce_client_config_t* config,
                xce_request_handler_t handler)
 {
@@ -135,6 +162,21 @@ int sched_init(xce_scheduler_t* sched, xce_client_config_t* config,
     sched->pend_updates[1] = (dataq_t*)calloc(1, sizeof(dataq_t));
     dataq_init(sched->pend_updates[1], sched_pend_update_killer);
 
+    kvmgr_init(&(sched->plugins), string_hash_key, 8, 8,
+               (zarray_item_killer_t)plugin_unregister);
+    kvmgr_init(&(sched->jobs), int_hash_key, 1000, 1000,
+               (zarray_item_killer_t)kill_sched_job);
+    kvmgr_init(&(sched->queues), string_hash_key, 8, 8,
+               (zarray_item_killer_t)kill_sched_queue);
+    kvmgr_init(&(sched->workers), string_hash_key, 256, 256,
+               (zarray_item_killer_t)kill_sched_worker);
+    kvmgr_init(&(sched->users), string_hash_key, 256, 256,
+               (zarray_item_killer_t)kill_sched_user);
+    kvmgr_init(&(sched->projects), string_hash_key, 8, 8,
+               (zarray_item_killer_t)kill_sched_project);
+
+    sched_register_plugins(sched);
+
     rc = xcecli_ctx_init(&(sched->apictx), config);
     if (rc < 0) {
         return rc;
@@ -194,18 +236,44 @@ void sched_fini(xce_scheduler_t* sched)
 
 static int sched_update_config(xce_scheduler_t* sched, xce_conf4sched_t* config)
 {
+    int i;
+
+    plugin_config_update(sched, config);
+
+    for (i = 0; i < config->n_queues; i++) {
+        xce_sched_queue_t* q = make_sched_queue(sched, config->queues[i]);
+
+        plugin_queue_update(sched, q);
+    }
+
     return 0;
 }
 
 static int sched_update_jobs(xce_scheduler_t* sched, int njobs,
                              xce_job_t** jobs)
 {
+    int i;
+
+    for (i = 0; i < njobs; i++) {
+        xce_sched_job_t* j = make_sched_job(sched, jobs[i]);
+
+        plugin_job_update(sched, j);
+    }
+
     return 0;
 }
 
 static int sched_update_workers(xce_scheduler_t* sched, int nworkers,
                                 xce_worker_t** workers)
 {
+    int i;
+
+    for (i = 0; i < nworkers; i++) {
+        xce_sched_worker_t* w = make_sched_worker(sched, workers[i]);
+
+        plugin_worker_update(sched, w);
+    }
+
     return 0;
 }
 
@@ -266,3 +334,55 @@ void sched_go(xce_scheduler_t* sched)
 
     return;
 }
+
+int sched_add_job(xce_scheduler_t* sched, xce_sched_job_t* j)
+{
+    return kvmgr_add(&(sched->jobs), (void*)j->jobid, j);
+}
+
+xce_sched_job_t* sched_get_job(xce_scheduler_t* sched, uint64_t jobid)
+{
+    return kvmgr_lock(&(sched->jobs), (void*)jobid, OBJREF_NOP);
+}
+
+int sched_add_queue(xce_scheduler_t* sched, xce_sched_queue_t* q)
+{
+    return kvmgr_add(&(sched->queues), q->name, q);
+}
+
+xce_sched_queue_t* sched_get_queue(xce_scheduler_t* sched, char* name)
+{
+    return kvmgr_lock(&(sched->queues), (void*)name, OBJREF_NOP);
+}
+
+int sched_add_worker(xce_scheduler_t* sched, xce_sched_worker_t* w)
+{
+    return kvmgr_add(&(sched->workers), w->name, w);
+}
+
+xce_sched_worker_t* sched_get_worker(xce_scheduler_t* sched, char* name)
+{
+    return kvmgr_lock(&(sched->workers), (void*)name, OBJREF_NOP);
+}
+
+int sched_add_user(xce_scheduler_t* sched, xce_sched_user_t* u)
+{
+    return kvmgr_add(&(sched->users), u->name, u);
+}
+
+xce_sched_user_t* sched_get_user(xce_scheduler_t* sched, char* name)
+{
+    return kvmgr_lock(&(sched->users), (void*)name, OBJREF_NOP);
+}
+
+int sched_add_project(xce_scheduler_t* sched, xce_sched_project_t* p)
+{
+    return kvmgr_add(&(sched->projects), p->name, p);
+}
+
+xce_sched_project_t* sched_get_project(xce_scheduler_t* sched, char* name)
+{
+    return kvmgr_lock(&(sched->projects), (void*)name, OBJREF_NOP);
+}
+
+int sched_job_is_ready(xce_sched_job_t* j) { return 1; }