create framework of scheduler

src/include/client.h

@@ -9,10 +9,6 @@
 #include "xce/types.h"
 #include "xceapi.h"
 
-#if 0
-typedef int (*xce_request_handler_t)(xce_request_t*, void*);
-#endif
-
 typedef void* xcecli_ctx_t;
 
 #define N_MAX_MASTERS (3)

src/scheduler/include/scheduler.h

@@ -1,4 +1,33 @@
 #if !defined(_SCHEDULER_H_)
 
 #define _SCHEDULER_H_ 1
+
+#include "client.h"
+#include "xce/types.h"
+#include "xlist.h"
+
+typedef enum {
+    XCE_DECISION_NONE = 0,
+    XCE_DECISION_DISPATCH,
+    XCE_DECISION_PEND
+} xce_decision_op_t;
+
+typedef struct {
+    xce_decision_op_t op;
+    union {
+        xce_job_alloc_t *alloc;
+    } u;
+} xce_decision_t;
+
+typedef struct {
+    xcecli_ctx_t apictx;
+    xlist_t      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 int  sched_go(xce_scheduler_t*);
+
 #endif

src/scheduler/meson.build

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

src/scheduler/src/handler.c

@@ -9,37 +9,41 @@
 #include "scheduler.h"
 #include "utils.h"
 
-static int sched_handle_config(xce_conf4sched_t* config)
+static int sched_handle_config(xce_scheduler_t* sched, xce_conf4sched_t* config)
 {
     TRACE("%s...", __func__);
-    TRACE("my cluster name %s",  config->clustername);
+    TRACE("my cluster name %s", config->clustername);
 
     return 0;
 }
 
-static int sched_handle_jobpush(xce_jobs_data_t* jobs)
+static int sched_handle_jobpush(xce_scheduler_t* sched, xce_jobs_data_t* jobs)
 {
     TRACE("%s...", __func__);
 
     return 0;
 }
 
-static int sched_handle_workerpush(xce_workers_data_t* workers)
+static int sched_handle_workerpush(xce_scheduler_t*    sched,
+                                   xce_workers_data_t* workers)
 {
     TRACE("%s...", __func__);
 
     return 0;
 }
 
-static int sched_on_jobs(int njobs, xce_job_t **jobs) {
+static int sched_on_jobs(xce_scheduler_t* sched, int njobs, xce_job_t** jobs)
+{
     return 0;
 }
 
-static int sched_on_workers(int nworkers, xce_worker_t **workers) {
+static int sched_on_workers(xce_scheduler_t* sched, int nworkers,
+                            xce_worker_t** workers)
+{
     return 0;
 }
 
-static int sched_handle_dataset(dataset_t* dataset)
+static int sched_handle_dataset(xce_scheduler_t* sched, dataset_t* dataset)
 {
     int rc = 0;
 
@@ -47,10 +51,11 @@ static int sched_handle_dataset(dataset_t* dataset)
 
     switch (dataset->table) {
     case Data_data_schedjobs:
-        rc = sched_on_jobs(dataset->nrows, (xce_job_t **)dataset->rows);
+        rc = sched_on_jobs(sched, dataset->nrows, (xce_job_t**)dataset->rows);
         break;
     case Data_data_schedworkers:
-        rc = sched_on_workers(dataset->nrows, (xce_worker_t **)dataset->rows);
+        rc = sched_on_workers(sched, dataset->nrows,
+                              (xce_worker_t**)dataset->rows);
         break;
     case Data_data_resources:
         break;
@@ -61,9 +66,10 @@ static int sched_handle_dataset(dataset_t* dataset)
     return rc;
 }
 
-int scheduler_req_handler(xce_request_t* request)
+int scheduler_req_handler(xce_request_t* request, void* args)
 {
-    int rc = 0;
+    int              rc    = 0;
+    xce_scheduler_t* sched = (xce_scheduler_t*)args;
 
     switch (request->opcode) {
     case OPCode_push:
@@ -71,24 +77,25 @@ int scheduler_req_handler(xce_request_t* request)
         case Request_body_configpushmsg:
             /* configuration arriving */
             rc = sched_handle_config(
-                request->body.configpushmsg->data.conf4sched);
+                sched, request->body.configpushmsg->data.conf4sched);
             break;
         case Request_body_pushmsg:
             /* data arriving */
             switch (request->body.pushmsg->table) {
             case Data_data_schedjobs:
                 /* job arriving */
-                rc = sched_handle_jobpush(request->body.pushmsg->data.jobs);
+                rc = sched_handle_jobpush(sched,
+                                          request->body.pushmsg->data.jobs);
                 break;
             case Data_data_schedworkers:
                 /* workers arriving */
                 rc = sched_handle_workerpush(
-                    request->body.pushmsg->data.workers);
+                    sched, request->body.pushmsg->data.workers);
                 break;
             }
             break;
         case Request_body_datamsg:
-            rc = sched_handle_dataset(request->body.datamsg);
+            rc = sched_handle_dataset(sched, request->body.datamsg);
             break;
         default:
             rc = -EINVAL;

src/scheduler/src/main.c

@@ -61,31 +61,6 @@ static int init_config(xce_client_config_t* config)
     return 0;
 }
 
-static int scheduler_init(xcecli_ctx_t* ctx, xce_client_config_t* config)
-{
-    int rc = 0;
-
-    rc = xcecli_ctx_init(ctx, config);
-    if (rc < 0) {
-        return rc;
-    }
-
-    xcecli_set_handler(*ctx, scheduler_req_handler, NULL);
-    rc = xcecli_start(*ctx);
-    if (rc < 0) {
-        return rc;
-    }
-
-    rc = xcecli_join(*ctx, HostRole_scheduler, NULL);
-    if (rc < 0) {
-        return rc;
-    }
-
-    return 0;
-}
-
-static void scheduler_fini(xcecli_ctx_t* ctx) { return; }
-
 int main(int argc, char* argv[])
 {
     struct arguments    arguments;
@@ -93,8 +68,9 @@ int main(int argc, char* argv[])
     char*               loglevel = "info";
     int                 logsize  = 64 * 1024 * 1024;
     int                 logcount = 10;
-    xcecli_ctx_t        ctx;
     xce_client_config_t config;
+    xce_scheduler_t     sched;
+    int                 njobs = 0;
 
     memset(&arguments, 0x0, sizeof(arguments));
 
@@ -109,17 +85,26 @@ int main(int argc, char* argv[])
     }
 
     mylog_init(logfile, loglevel, logsize, logcount);
-    
+
     init_config(&config);
-    scheduler_init(&ctx, &config);
+
+    sched_init(&sched, &config, scheduler_req_handler);
 
     INFO("xce scheduler is running");
 
     while (1) {
+        sched_update(&sched);
+        njobs = sched_go(&sched);
+
+        if (njobs > 0) {
+            /* flush decision list to controller */
+        }
+
+        /* handle scheduler core outputs */
         sleep(1);
     }
 
-    scheduler_fini(&ctx);
+    sched_fini(&sched);
 
     INFO("xce scheduler has been shutdown");
 

src/scheduler/src/scheduler.c

@@ -0,0 +1,101 @@
+#include <errno.h>
+#include <signal.h>
+#include <stdio.h>
+#include <stdlib.h>
+#include <string.h>
+
+#include "client.h"
+#include "scheduler.h"
+
+int sched_init(xce_scheduler_t* sched, xce_client_config_t* config,
+               xce_request_handler_t handler)
+{
+    int rc = 0;
+
+    if ((sched == NULL) || (config == NULL) || (handler == NULL)) {
+        return -EINVAL;
+    }
+
+    memset(sched, 0x0, sizeof(*sched));
+
+    /* FIXME: configurable delta for xlist_init()? */
+    xlist_init(&(sched->decisions), 32);
+
+    rc = xcecli_ctx_init(&(sched->apictx), config);
+    if (rc < 0) {
+        return rc;
+    }
+
+    xcecli_set_handler(sched->apictx, handler, sched);
+
+    rc = xcecli_start(sched->apictx);
+    if (rc < 0) {
+        return rc;
+    }
+
+    return xcecli_join(sched->apictx, HostRole_scheduler, NULL);
+}
+
+void sched_fini(xce_scheduler_t* sched)
+{
+    xcecli_wakeup(sched->apictx, (int)-EINTR);
+    xcecli_stop(sched->apictx);
+    xcecli_ctx_fini(sched->apictx);
+
+    xlist_fini(&(sched->decisions));
+    return;
+}
+
+#if 0
+static int sched_update_config(xce_scheduler_t* sched, xce_conf4sched_t* config)
+{
+    return 0;
+}
+
+static int sched_update_jobs(xce_scheduler_t* sched, int njobs,
+                             xce_job_t** jobs)
+{
+    return 0;
+}
+
+static int sched_update_workers(xce_scheduler_t* sched, int nworkers,
+                                xce_worker_t** workers)
+{
+    return 0;
+}
+#endif
+
+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)
+{
+    *decision = NULL;
+
+    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;
+}