save works

src/scheduler/include/plugin.h

@@ -21,8 +21,11 @@ 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*);
+typedef void* (*plugin_job_election_fn_t)(void*);
 
+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*);
 
@@ -34,14 +37,16 @@ typedef struct {
 } plugin_hooks_t;
 
 typedef struct {
-    char*                    name;
+    char*                     name;
-    void*                    sched;
+    void*                     sched;
-    plugin_init_fn_t         init_fn;
+    plugin_init_fn_t          init_fn;
-    plugin_fini_fn_t         fini_fn;
+    plugin_fini_fn_t          fini_fn;
-    plugin_job_election_fn_t job_elect_fn;
+    plugin_begin_session_fn_t begin_session_fn;
-    plugin_job_allocate_fn_t job_alloc_fn;
+    plugin_end_session_fn_t   end_session_fn;
-    plugin_hooks_t           hooks;
+    plugin_job_election_fn_t  job_elect_fn;
-    plugin_data_t            data;
+    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*);
@@ -50,6 +55,8 @@ 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  plugin_begin_session(void*);
+extern int  plugin_end_session(void*);
 
 extern int register_fcfs(void*);
 

src/scheduler/include/scheduler.h

@@ -66,6 +66,8 @@ extern int                  sched_init(xce_scheduler_t*, xce_client_config_t*,
 extern void                 sched_fini(xce_scheduler_t*);
 extern int                  sched_update(xce_scheduler_t*);
 extern void                 sched_go(xce_scheduler_t*);
+extern void                 sched_begin_session(xce_scheduler_t*);
+extern void                 sched_end_session(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*);
@@ -77,5 +79,6 @@ 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*);
+extern int sched_queue_is_ready(xce_sched_queue_t*);
 
 #endif

src/scheduler/src/main.c

@@ -71,7 +71,7 @@ static void stop_scheduler(xce_scheduler_t* sched)
     INFO("stop to scheduling jobs...");
 
     sched->stop = 1;
-    cbuf_put(&(sched->decisions), &decision);
+    cbuf_tryput(&(sched->decisions), &decision);
 }
 
 static void sig_handler(int signo)
@@ -118,6 +118,7 @@ int main(int argc, char* argv[])
 
     while (1) {
         sched_update(&sched);
+        sched_begin_session(&sched);
         sched_go(&sched);
 
         while (1) {
@@ -146,6 +147,8 @@ int main(int argc, char* argv[])
             }
         }
 
+        sched_end_session(&sched);
+        
         if (sched.stop) {
             break;
         }

src/scheduler/src/plugin.c

@@ -186,3 +186,69 @@ int plugin_worker_update(void* arg, xce_sched_worker_t* worker)
 
     return rc;
 }
+
+static int plugin_begin_session_(void* arg1, void* arg2)
+{
+    xce_sched_plugin_t* plugin = (xce_sched_plugin_t*)arg2;
+
+    if (plugin == NULL) {
+        return 0;
+    }
+
+    if (plugin->begin_session_fn != NULL) {
+        return plugin->begin_session_fn(plugin);
+    }
+
+    return 0;
+}
+
+int plugin_begin_session(void* arg)
+{
+    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_begin_session_);
+    rc = kvmgr_iter(&(sched->plugins), NULL, &lst, NULL);
+    xlist_fini(&lst);
+
+    return rc;
+}
+
+static int plugin_end_session_(void* arg1, void* arg2)
+{
+    xce_sched_plugin_t* plugin = (xce_sched_plugin_t*)arg2;
+
+    if (plugin == NULL) {
+        return 0;
+    }
+
+    if (plugin->end_session_fn != NULL) {
+        return plugin->end_session_fn(plugin);
+    }
+
+    return 0;
+}
+
+int plugin_end_session(void* arg)
+{
+    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_end_session_);
+    rc = kvmgr_iter(&(sched->plugins), NULL, &lst, NULL);
+    xlist_fini(&lst);
+
+    return rc;
+}

src/scheduler/src/plugin_fcfs.c

@@ -20,6 +20,7 @@ typedef struct {
 
 typedef struct {
     ordq_t queues;
+    int    new_session;
 } fcfs_plugin_data_t;
 
 static void fcfs_job_killer(void* arg)
@@ -81,6 +82,7 @@ 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);
+    data->new_session = 0;
     return 0;
 }
 
@@ -92,15 +94,8 @@ static int fcfs_update_config(void* arg, xce_conf4sched_t* config)
     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);
+    xce_use(privdata);
-        plugin->data = privdata;
-    }
 
     return 0;
 }
@@ -194,73 +189,113 @@ static int fcfs_update_job(void* arg, xce_sched_job_t* j)
     return 0;
 }
 
-static int skip_no_pend_job_queue(void* arg1, va_list ap)
+static void* fcfs_elect_job(xce_sched_plugin_t* plugin)
 {
-    fcfs_queue_t* q = (fcfs_queue_t*)arg1;
+    xce_scheduler_t*    sched    = plugin->sched;
+    fcfs_plugin_data_t* privdata = (fcfs_plugin_data_t*)plugin->data;
 
-    if (q == NULL) {
+    while (1) {
-        return 1;
+        ordq_t*      queues = &(privdata->queues);
+        ordq_item_t* it     = NULL;
+        int          go     = 0;
+
+        pthread_testcancel();
+
+        cbuf_get(&(sched->wbuf), &go);
+        privdata->new_session = 0;
+
+        for (it = queues->head; it != NULL; it = it->next) {
+            fcfs_queue_t* q        = (fcfs_queue_t*)it->data;
+            ordq_t*       pendJobs = NULL;
+            ordq_item_t*  it2      = NULL;
+
+            pthread_testcancel();
+
+            if (!sched_queue_is_ready(q->queue)) {
+                continue;
+            }
+
+            pendJobs = &(q->pendJobs);
+            for (it2 = pendJobs->head; it2 != NULL; it2 = it2->next) {
+                fcfs_job_t* j = (fcfs_job_t*)it2->data;
+
+                pthread_testcancel();
+
+                if (sched_job_is_ready(j->job)) {
+                    uint64_t jobid = j->job->jobid;
+
+                    TRACE("%s jobid %ld", __func__, jobid);
+                    cbuf_put(&(sched->jbuf), &jobid);
+                }
+
+                if (privdata->new_session) {
+                    break;
+                }
+            }
+
+            if (privdata->new_session) {
+                break;
+            }
+        }
+
+        sleep(1);
     }
 
-    return (ordq_peek(&(q->pendJobs), NULL) == NULL);
+    return NULL;
 }
 
-static int skip_not_ready_job(void* arg1, va_list ap)
+static int fcfs_end_session(void* arg)
-{
-    fcfs_job_t* job = (fcfs_job_t*)arg1;
-
-    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__);
+    privdata->new_session = 1;
 
-    it = ordq_peek(&(privdata->queues), (ordq_skip_fn_t)skip_no_pend_job_queue);
+    return 0;
-    if (it == NULL) {
+}
-        TRACE("%s no job to be elected", __func__);
 
-        return 0;
+static int fcfs_init(void* arg)
+{
+    xce_sched_plugin_t* plugin   = (xce_sched_plugin_t*)arg;
+    fcfs_plugin_data_t* privdata = NULL;
+
+    privdata = (fcfs_plugin_data_t*)calloc(1, sizeof(*privdata));
+    if (privdata == NULL) {
+        return -ENOMEM;
     }
 
-    q  = (fcfs_queue_t*)it->data;
+    fcfs_priv_data_init(privdata);
-    it = ordq_get(&(q->pendJobs), skip_not_ready_job);
+    plugin->data = privdata;
-    if (it == NULL) {
-        TRACE("%s no job to be elected", __func__);
 
-        return 0;
+    return 1;
+}
+
+static void fcfs_fini(void* arg)
+{
+    xce_sched_plugin_t* plugin   = (xce_sched_plugin_t*)arg;
+    fcfs_plugin_data_t* privdata = (fcfs_plugin_data_t*)plugin->data;
+
+    if (privdata == NULL) {
+        return;
     }
 
-    job   = (fcfs_job_t*)it->data;
+    ordq_fini(&(privdata->queues));
-    jobid = job->job->jobid;
+    free(privdata);
-    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",
+    .name             = "xce_sched_fcfs",
-    .hooks        = { .config_hook = fcfs_update_config,
+    .hooks            = { .config_hook = fcfs_update_config,
-                     .queue_hook  = fcfs_update_queue,
+                         .queue_hook  = fcfs_update_queue,
-                     .job_hook    = fcfs_update_job,
+                         .job_hook    = fcfs_update_job,
-                     .worker_hook = NULL },
+                         .worker_hook = NULL },
-    .job_elect_fn = fcfs_elect_job,
+    .init_fn          = (plugin_init_fn_t)fcfs_init,
-    .job_alloc_fn = NULL,
+    .fini_fn          = (plugin_fini_fn_t)fcfs_fini,
-    .data         = NULL
+    .begin_session_fn = NULL,
+    .end_session_fn   = (plugin_end_session_fn_t)fcfs_end_session,
+    .job_elect_fn     = (plugin_job_election_fn_t)fcfs_elect_job,
+    .job_alloc_fn     = NULL,
+    .data             = NULL
 };
 
 int register_fcfs(void* sched) { return plugin_register(sched, &fcfs_plugin); }

src/scheduler/src/scheduler.c

@@ -13,54 +13,18 @@
 #include "scheduler.h"
 #include "utils.h"
 
-static uint64_t pick_job(xce_scheduler_t* sched)
-{
-    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;
-}
-
 static void* sched_job_picker(void* arg)
 {
-    xce_scheduler_t* sched = (xce_scheduler_t*)arg;
+    xce_sched_plugin_t* job_picker      = NULL;
+    xce_scheduler_t*    sched           = (xce_scheduler_t*)arg;
+    char*               job_pick_plugin = "xce_sched_fcfs";
 
-    xce_use(sched);
+    job_picker = kvmgr_lock(&(sched->plugins), job_pick_plugin, OBJREF_NOP);
-
+    if ((job_picker == NULL) || (job_picker->job_elect_fn == NULL)) {
-    while (1) {
+        return NULL;
-        int      go    = 0;
-        uint64_t jobid = 0;
-
-        pthread_testcancel();
-
-        cbuf_get(&(sched->wbuf), &go);
-        if (go == 0) {
-            break;
-        }
-
-        /* pick job */
-        jobid = pick_job(sched);
-
-        cbuf_put(&(sched->jbuf), &jobid);
     }
 
-    return NULL;
+    return job_picker->job_elect_fn(job_picker);
 }
 
 static xce_decision_t* calc_decision(xce_scheduler_t* sched, uint64_t jobid)
@@ -335,6 +299,20 @@ void sched_go(xce_scheduler_t* sched)
     return;
 }
 
+void sched_begin_session(xce_scheduler_t* sched)
+{
+    plugin_begin_session(sched);
+
+    return;
+}
+
+void sched_end_session(xce_scheduler_t* sched)
+{
+    plugin_end_session(sched);
+
+    return;
+}
+
 int sched_add_job(xce_scheduler_t* sched, xce_sched_job_t* j)
 {
     return kvmgr_add(&(sched->jobs), (void*)j->jobid, j);
@@ -386,3 +364,5 @@ 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; }