scheduler framework

src/scheduler/src/scheduler.c

@@ -1,4 +1,5 @@
 #include <errno.h>
+#include <pthread.h>
 #include <signal.h>
 #include <stdio.h>
 #include <stdlib.h>
@@ -6,11 +7,78 @@
 
 #include "client.h"
 #include "scheduler.h"
+#include "utils.h"
+
+static uint64_t pick_job(xce_scheduler_t* sched)
+{
+    uint64_t jobid = 0;
+
+    return jobid;
+}
+
+static void* sched_job_picker(void* arg)
+{
+    xce_scheduler_t* sched = (xce_scheduler_t*)arg;
+
+    xce_use(sched);
+
+    while (1) {
+        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;
+}
+
+static xce_decision_t *calc_decision(xce_scheduler_t *sched, uint64_t jobid) {
+    return NULL;
+}
+
+static void* sched_allocator(void* arg)
+{
+    xce_scheduler_t* sched = (xce_scheduler_t*)arg;
+
+    xce_use(sched);
+    while (1) {
+        uint64_t        jobid;
+        xce_decision_t* decision = NULL;
+
+        pthread_testcancel();
+
+        cbuf_get(&(sched->jbuf), &jobid);
+        if (jobid == 0) {
+            /* end of scheduling session */
+            cbuf_put(&(sched->decisions), NULL);
+            continue;
+        }
+
+        /* calculate decision for jobid */
+        decision = calc_decision(sched, jobid);
+        
+        cbuf_put(&(sched->decisions), &decision);
+    }
+
+    return NULL;
+}
 
 int sched_init(xce_scheduler_t* sched, xce_client_config_t* config,
                xce_request_handler_t handler)
 {
-    int rc = 0;
+    int rc         = 0;
+    int ndecisions = 5; /* FIXME: make this configurable */
+    int njobs      = 1; /* FIXME: make this configurable */
 
     if ((sched == NULL) || (config == NULL) || (handler == NULL)) {
         return -EINVAL;
@@ -18,8 +86,9 @@ int sched_init(xce_scheduler_t* sched, xce_client_config_t* config,
 
     memset(sched, 0x0, sizeof(*sched));
 
-    /* FIXME: configurable delta for xlist_init()? */
-    xlist_init(&(sched->decisions), 32);
+    cbuf_init(&(sched->wbuf), 1, sizeof(int));
+    cbuf_init(&(sched->jbuf), njobs, sizeof(uint64_t));
+    cbuf_init(&(sched->decisions), ndecisions, sizeof(xce_decision_t*));
 
     rc = xcecli_ctx_init(&(sched->apictx), config);
     if (rc < 0) {
@@ -33,16 +102,43 @@ int sched_init(xce_scheduler_t* sched, xce_client_config_t* config,
         return rc;
     }
 
-    return xcecli_join(sched->apictx, HostRole_scheduler, NULL);
+    rc = xcecli_join(sched->apictx, HostRole_scheduler, NULL);
+    if (rc < 0) {
+        return rc;
+    }
+
+    rc = pthread_create(&(sched->jobpicker), NULL, &sched_job_picker, sched);
+    if (rc < 0) {
+        return rc;
+    }
+
+    rc = pthread_create(&(sched->allocator), NULL, &sched_allocator, sched);
+
+    return rc;
 }
 
 void sched_fini(xce_scheduler_t* sched)
 {
+    void* res;
+
+    if (sched == NULL) {
+        return;
+    }
+
+    pthread_cancel(sched->jobpicker);
+    pthread_join(sched->jobpicker, &res);
+
+    pthread_cancel(sched->allocator);
+    pthread_join(sched->allocator, &res);
+
     xcecli_wakeup(sched->apictx, (int)-EINTR);
     xcecli_stop(sched->apictx);
     xcecli_ctx_fini(sched->apictx);
 
-    xlist_fini(&(sched->decisions));
+    cbuf_fini(&(sched->decisions));
+    cbuf_fini(&(sched->jbuf));
+    cbuf_fini(&(sched->wbuf));
+
     return;
 }
 
@@ -67,35 +163,11 @@ static int sched_update_workers(xce_scheduler_t* sched, int nworkers,
 
 int sched_update(xce_scheduler_t* sched) { return 0; }
 
-static xce_job_t* pick_job(xce_scheduler_t* sched) { return NULL; }
-
-static int calc_job_decision(xce_scheduler_t* sched, xce_job_t* job,
-                             xce_decision_t** decision)
+void sched_go(xce_scheduler_t* sched)
 {
-    *decision = NULL;
+    int go = 1;
 
-    return 0;
-}
-
-int sched_go(xce_scheduler_t* sched)
-{
-    uint32_t njobs = 0;
-
-    while (1) {
-        xce_job_t*      job      = NULL;
-        xce_decision_t* decision = NULL;
-        int             rc       = 0;
-
-        job = pick_job(sched);
-        if (job == NULL) {
-            return njobs;
-        }
-
-        rc = calc_job_decision(sched, job, &decision);
-        if (rc > 0) {
-            njobs++;
-            /* put decision of output list */
-        }
-    }
-    return 0;
+    cbuf_put(&(sched->wbuf), &go);
+
+    return;
 }