save works

src/scheduler/include/decision.h

@@ -0,0 +1,21 @@
+#if !defined(_DECISION_H_)
+
+#define _DECISION_H_ 1
+
+#include "xce/types.h"
+
+typedef enum {
+    XCE_DECISION_NONE = 0,
+    XCE_DECISION_DISPATCH,
+    XCE_DECISION_PEND,
+    XCE_DECISION_NEW_SESSION
+} xce_decision_op_t;
+
+typedef struct {
+    xce_decision_op_t op;
+    union {
+        xce_job_alloc_t* alloc;
+    } u;
+} xce_decision_t;
+
+#endif

src/scheduler/include/scheduler.h

@@ -4,26 +4,46 @@
 
 #include <pthread.h>
 
+#include "decision.h"
 #include "client.h"
+#include "dataq.h"
 #include "mylock.h"
 #include "xce/types.h"
 #include "xlist.h"
 
 typedef enum {
-    XCE_DECISION_NONE = 0,
+    XCE_SCHED_OP_NONE = 0,
-    XCE_DECISION_DISPATCH,
+    XCE_SCHED_OP_UPDATE_CONFIG,
-    XCE_DECISION_PEND
+    XCE_SCHED_OP_UPDATE_JOB,
-} xce_decision_op_t;
+    XCE_SCHED_OP_UPDATE_WORKER
+} xce_sched_op_t;
 
 typedef struct {
-    xce_decision_op_t op;
+    data_item_t    link;
+    xce_sched_op_t op;
     union {
-        xce_job_alloc_t* alloc;
+        xce_conf4sched_t* config;
+        struct {
+            int         njobs;
+            xce_job_t** jobs;
+        } jobupdate;
+        struct {
+            int            nworkers;
+            xce_worker_t** workers;
+        } workerupdate;
     } u;
-} xce_decision_t;
+} xce_sched_update_request_t;
+
+#define ITPTR_TO_UPDATE_REQ(p)                                                 \
+    ((xce_sched_update_request_t*)(((char*)p)                                  \
+                                   - offsetof(xce_sched_update_request_t,      \
+                                              link)))
 
 typedef struct {
+    mylock_t     lock;
+    int          stop;
     xcecli_ctx_t apictx;
+    dataq_t*     pend_updates[2];
     pthread_t    jobpicker;
     pthread_t    allocator;
     cbuf_t       wbuf;      /* action to trigger scheduling */

src/scheduler/meson.build

@@ -2,6 +2,7 @@ scheduler_srcs = [
   'src/main.c',
   'src/handler.c',
   'src/scheduler.c',
+  'src/decision.c',
 ]
 
 incdir = include_directories('src',

src/scheduler/src/decision.c

@@ -0,0 +1,5 @@
+#include <stdio.h>
+#include <stdlib.h>
+
+#include "decision.h"
+

src/scheduler/src/handler.c

@@ -4,16 +4,37 @@
 #include <unistd.h>
 
 #include "controller.capnp.h"
+#include "dataq.h"
 #include "xce/types.h"
 
 #include "scheduler.h"
 #include "utils.h"
 
+static void enq_pend_update(xce_scheduler_t*            sched,
+                            xce_sched_update_request_t* req)
+{
+    mylock_lock(&(sched->lock));
+    dataq_enq(sched->pend_updates[0], &(req->link));
+    mylock_unlock(&(sched->lock));
+}
+
 static int sched_handle_config(xce_scheduler_t* sched, xce_conf4sched_t* config)
 {
+    xce_sched_update_request_t* req;
+
     TRACE("%s...", __func__);
     TRACE("my cluster name %s", config->clustername);
 
+    req = (xce_sched_update_request_t*)calloc(1, sizeof(*req));
+    if (req == NULL) {
+        return -ENOMEM;
+    }
+
+    req->op       = XCE_SCHED_OP_UPDATE_CONFIG;
+    req->u.config = config;
+
+    enq_pend_update(sched, req);
+
     return 0;
 }
 
@@ -34,12 +55,38 @@ static int sched_handle_workerpush(xce_scheduler_t*    sched,
 
 static int sched_on_jobs(xce_scheduler_t* sched, int njobs, xce_job_t** jobs)
 {
+    xce_sched_update_request_t* req;
+
+    req = (xce_sched_update_request_t*)calloc(1, sizeof(*req));
+    if (req == NULL) {
+        return -ENOMEM;
+    }
+
+    req->op                = XCE_SCHED_OP_UPDATE_JOB;
+    req->u.jobupdate.njobs = njobs;
+    req->u.jobupdate.jobs  = jobs;
+
+    enq_pend_update(sched, req);
+
     return 0;
 }
 
 static int sched_on_workers(xce_scheduler_t* sched, int nworkers,
                             xce_worker_t** workers)
 {
+    xce_sched_update_request_t* req;
+
+    req = (xce_sched_update_request_t*)calloc(1, sizeof(*req));
+    if (req == NULL) {
+        return -ENOMEM;
+    }
+
+    req->op                      = XCE_SCHED_OP_UPDATE_WORKER;
+    req->u.workerupdate.nworkers = nworkers;
+    req->u.workerupdate.workers  = workers;
+
+    enq_pend_update(sched, req);
+
     return 0;
 }
 
@@ -52,10 +99,21 @@ static int sched_handle_dataset(xce_scheduler_t* sched, dataset_t* dataset)
     switch (dataset->table) {
     case Data_data_schedjobs:
         rc = sched_on_jobs(sched, dataset->nrows, (xce_job_t**)dataset->rows);
+        if (rc >= 0) {
+            /* jobs will be put in pend update request queue */
+            dataset->nrows = 0;
+            dataset->rows  = NULL;
+        }
+
         break;
     case Data_data_schedworkers:
         rc = sched_on_workers(sched, dataset->nrows,
                               (xce_worker_t**)dataset->rows);
+        if (rc >= 0) {
+            /* workers will be put in pend update request queue */
+            dataset->nrows = 0;
+            dataset->rows  = NULL;
+        }
         break;
     case Data_data_resources:
         break;

src/scheduler/src/main.c

@@ -1,4 +1,5 @@
 #include <argp.h>
+#include <signal.h>
 #include <stdio.h>
 #include <stdlib.h>
 #include <unistd.h>
@@ -61,6 +62,29 @@ static int init_config(xce_client_config_t* config)
     return 0;
 }
 
+static xce_scheduler_t sched;
+
+static void stop_scheduler(xce_scheduler_t* sched)
+{
+    xce_decision_t *decision = NULL;
+    
+    INFO("stop to scheduling jobs...");
+
+    sched->stop = 1;
+    cbuf_put(&(sched->decisions), &decision);
+}
+
+static void sig_handler(int signo)
+{
+    switch (signo) {
+    case SIGINT:
+        stop_scheduler(&sched);
+        break;
+    default:
+        break;
+    }
+}
+
 int main(int argc, char* argv[])
 {
     struct arguments    arguments;
@@ -69,7 +93,6 @@ int main(int argc, char* argv[])
     int                 logsize  = 64 * 1024 * 1024;
     int                 logcount = 10;
     xce_client_config_t config;
-    xce_scheduler_t     sched;
 
     memset(&arguments, 0x0, sizeof(arguments));
 
@@ -89,6 +112,8 @@ int main(int argc, char* argv[])
 
     sched_init(&sched, &config, scheduler_req_handler);
 
+    signal(SIGINT, sig_handler);
+
     INFO("xce scheduler is running");
 
     while (1) {
@@ -100,6 +125,8 @@ int main(int argc, char* argv[])
 
             cbuf_get(&(sched.decisions), &decision);
 
+            TRACE("%s decision = %p", __func__, decision);
+
             if (decision == NULL) {
                 /* start new session */
                 break;
@@ -119,6 +146,10 @@ int main(int argc, char* argv[])
             }
         }
 
+        if (sched.stop) {
+            break;
+        }
+
         sleep(1);
     }
 

src/scheduler/src/scheduler.c

@@ -6,6 +6,7 @@
 #include <string.h>
 
 #include "client.h"
+#include "controller.capnp.h"
 #include "scheduler.h"
 #include "utils.h"
 
@@ -61,7 +62,8 @@ static void* sched_allocator(void* arg)
         cbuf_get(&(sched->jbuf), &jobid);
         if (jobid == 0) {
             /* end of scheduling session */
-            cbuf_put(&(sched->decisions), NULL);
+            decision = NULL;
+            cbuf_put(&(sched->decisions), &decision);
             continue;
         }
 
@@ -74,6 +76,40 @@ static void* sched_allocator(void* arg)
     return NULL;
 }
 
+static void sched_pend_update_killer(data_item_t* it)
+{
+    xce_sched_update_request_t* req;
+    int                         i;
+
+    if (it == NULL) {
+        return;
+    }
+
+    req = ITPTR_TO_UPDATE_REQ(it);
+    switch (req->op) {
+    case XCE_SCHED_OP_UPDATE_CONFIG:
+        break;
+    case XCE_SCHED_OP_UPDATE_JOB:
+        for (i = 0; i < req->u.jobupdate.njobs; i++) {
+            free_Job_ptr(&(req->u.jobupdate.jobs[i]));
+        }
+
+        free(req->u.jobupdate.jobs);
+        break;
+    case XCE_SCHED_OP_UPDATE_WORKER:
+        for (i = 0; i < req->u.workerupdate.nworkers; i++) {
+            free_Worker_ptr(&(req->u.workerupdate.workers[i]));
+        }
+
+        free(req->u.workerupdate.workers);
+        break;
+    default:
+        break;
+    }
+
+    free(req);
+}
+
 int sched_init(xce_scheduler_t* sched, xce_client_config_t* config,
                xce_request_handler_t handler)
 {
@@ -87,10 +123,17 @@ int sched_init(xce_scheduler_t* sched, xce_client_config_t* config,
 
     memset(sched, 0x0, sizeof(*sched));
 
+    mylock_init(&(sched->lock));
     cbuf_init(&(sched->wbuf), 1, sizeof(int));
     cbuf_init(&(sched->jbuf), njobs, sizeof(uint64_t));
     cbuf_init(&(sched->decisions), ndecisions, sizeof(xce_decision_t*));
 
+    sched->pend_updates[0] = (dataq_t*)calloc(1, sizeof(dataq_t));
+    dataq_init(sched->pend_updates[0], sched_pend_update_killer);
+
+    sched->pend_updates[1] = (dataq_t*)calloc(1, sizeof(dataq_t));
+    dataq_init(sched->pend_updates[1], sched_pend_update_killer);
+
     rc = xcecli_ctx_init(&(sched->apictx), config);
     if (rc < 0) {
         return rc;
@@ -140,10 +183,14 @@ void sched_fini(xce_scheduler_t* sched)
     cbuf_fini(&(sched->jbuf));
     cbuf_fini(&(sched->wbuf));
 
+    dataq_fini(sched->pend_updates[0]);
+    dataq_fini(sched->pend_updates[1]);
+    
+    mylock_fini(&(sched->lock));
+
     return;
 }
 
-#if 0
 static int sched_update_config(xce_scheduler_t* sched, xce_conf4sched_t* config)
 {
     return 0;
@@ -160,9 +207,55 @@ static int sched_update_workers(xce_scheduler_t* sched, int nworkers,
 {
     return 0;
 }
-#endif
 
-int sched_update(xce_scheduler_t* sched) { return 0; }
+int sched_update(xce_scheduler_t* sched)
+{
+    dataq_t* q = NULL;
+
+    /* swap dataq */
+    mylock_lock(&(sched->lock));
+    q                      = sched->pend_updates[0];
+    sched->pend_updates[0] = sched->pend_updates[1];
+    sched->pend_updates[1] = q;
+    mylock_unlock(&(sched->lock));
+
+    while (1) {
+        data_item_t*                it;
+        xce_sched_update_request_t* req;
+        int                         rc;
+
+        it = dataq_deq(q, NULL);
+        if (it == NULL) {
+            break;
+        }
+
+        req = ITPTR_TO_UPDATE_REQ(it);
+        switch (req->op) {
+        case XCE_SCHED_OP_UPDATE_CONFIG:
+            rc = sched_update_config(sched, req->u.config);
+            break;
+        case XCE_SCHED_OP_UPDATE_JOB:
+            rc = sched_update_jobs(sched, req->u.jobupdate.njobs,
+                                   req->u.jobupdate.jobs);
+            break;
+        case XCE_SCHED_OP_UPDATE_WORKER:
+            rc = sched_update_workers(sched, req->u.workerupdate.nworkers,
+                                      req->u.workerupdate.workers);
+            break;
+        default:
+            rc = -EINVAL;
+            break;
+        }
+
+        sched_pend_update_killer(it);
+        
+        if (rc < 0) {
+            /* something wrong? */
+        }
+    }
+
+    return 0;
+}
 
 void sched_go(xce_scheduler_t* sched)
 {