working of scheduler to handle configuration

src/controller/cluster.c

@@ -20,10 +20,6 @@
 #include "xce/errnox.h"
 #include "xstring.h"
 
-static time_t        last_print_time = 0;
-static unsigned long npkts_old       = 0;
-static unsigned long npkts           = 0;
-
 static xce_peerinfo_t from_controller
     = { .name = "controller", .key = "keepSecret" };
 

src/include/xce/types.h

@@ -358,10 +358,11 @@ typedef struct {
     0x20 /**< Fairshare queue with null user share defined*/
 
 /* FIXME: improve following later */
-#define JOB_IS_PEND(j) ((j->state & JOB_STAT_PEND) != 0)
+#define JOB_IS_PEND(state)   (((state) & JOB_STAT_PEND) != 0)
-#define JOB_IS_RUN(j)  ((j->state & JOB_STAT_RUN) != 0)
+#define JOB_IS_RUN(state)    (((state) & JOB_STAT_RUN) != 0)
-#define JOB_IS_EXIT(j) ((j->state & JOB_STAT_EXIT) != 0)
+#define JOB_IS_EXIT(state)   (((state) & JOB_STAT_EXIT) != 0)
-#define JOB_IS_DONE(j) ((j->state & JOB_STAT_DONE) != 0)
+#define JOB_IS_DONE(state)   (((state) & JOB_STAT_DONE) != 0)
+#define JOB_IS_FINISH(state) (JOB_IS_EXIT(state) || JOB_IS_DONE(state))
 
 typedef enum {
     NORMAL = 0,

src/scheduler/include/plugin.h

@@ -11,8 +11,8 @@
 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_config_hook_t)(void*, xce_sched_config_t*,
-                                           xce_conf4sched_t*);
+                                           xce_sched_config_t*);
 typedef int (*plugin_update_queue_hook_t)(void*, xce_sched_queue_t*,
                                           xce_sched_queue_t*);
 typedef int (*plugin_update_job_hook_t)(void*, xce_sched_job_t*,
@@ -62,7 +62,7 @@ typedef struct {
 
 extern int  plugin_register(void*, void*);
 extern void plugin_unregister(void*);
-extern int  plugin_config_update(void*, xce_conf4sched_t*, xce_conf4sched_t*);
+extern int  plugin_config_update(void*, xce_sched_config_t*, xce_sched_config_t*);
 extern int  plugin_queue_update(void*, xce_sched_queue_t*, xce_sched_queue_t*);
 extern int  plugin_job_update(void*, xce_sched_job_t*, xce_sched_job_t*);
 extern int  plugin_worker_update(void*, xce_sched_worker_t*,

src/scheduler/include/sched/types.h

@@ -5,6 +5,7 @@
 #include <stdint.h>
 #include <stdlib.h>
 
+#include "kvmgr.h"
 #include "typemap.h"
 #include "xce/types.h"
 
@@ -54,15 +55,22 @@ typedef struct {
 
 typedef struct {
     char*      name;
+    uint32_t   status;
     xce_val_t* res_avail;  /* resources in worker */
     xce_val_t* res_in_use; /* used resources in worker */
 } xce_sched_worker_t;
 
+typedef struct {
+    char*    clustername;
+    resmgr_t resmgr;
+} xce_sched_config_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 xce_sched_config_t*  make_sched_config(void*, xce_conf4sched_t*);
 extern void                 kill_sched_job(void*);
 extern void                 kill_sched_queue(void*);
 extern void                 kill_sched_worker(void*);

src/scheduler/include/scheduler.h

@@ -49,22 +49,22 @@ typedef struct {
                                               link)))
 
 typedef struct {
-    mylock_t          lock;
+    mylock_t            lock;
-    int               stop;
+    int                 stop;
-    xcecli_ctx_t      apictx;
+    xcecli_ctx_t        apictx;
-    xce_conf4sched_t* config;
+    xce_sched_config_t* config;
-    kvmgr_t           plugins;
+    kvmgr_t             plugins;
-    dataq_t*          pend_updates[2];
+    dataq_t*            pend_updates[2];
-    kvmgr_t           jobs;
+    kvmgr_t             jobs;
-    kvmgr_t           queues;
+    kvmgr_t             queues;
-    kvmgr_t           workers;
+    kvmgr_t             workers;
-    kvmgr_t           users;
+    kvmgr_t             users;
-    kvmgr_t           projects;
+    kvmgr_t             projects;
-    pthread_t         jobpicker;
+    pthread_t           jobpicker;
-    pthread_t         allocator;
+    pthread_t           allocator;
-    cbuf_t            wbuf;      /* action to trigger scheduling */
+    cbuf_t              wbuf;      /* action to trigger scheduling */
-    cbuf_t            jbuf;      /* job picked for allocation */
+    cbuf_t              jbuf;      /* job picked for allocation */
-    cbuf_t            decisions; /* scheduler decisions */
+    cbuf_t              decisions; /* scheduler decisions */
 } xce_scheduler_t;
 
 extern int                  sched_init(xce_scheduler_t*, xce_client_config_t*,

src/scheduler/src/handler.c

@@ -127,23 +127,23 @@ static int sched_handle_dataset(xce_scheduler_t* sched, dataset_t* dataset)
 static int sched_job_state_change(xce_scheduler_t* sched, uint64_t jobid,
                                   uint32_t state)
 {
-    char jobidString[128];
+    char                        jobidString[128];
-    xce_sched_update_request_t *req = NULL;
+    xce_sched_update_request_t* req = NULL;
 
     fmt_jobid(jobidString, sizeof(jobidString) - 1, jobid);
     TRACE("%s %s state %u ", __func__, jobidString, state);
 
-    req = (xce_sched_update_request_t *)calloc(1, sizeof(*req));
+    req = (xce_sched_update_request_t*)calloc(1, sizeof(*req));
     if (req == NULL) {
         return -ENOMEM;
     }
 
-    req->op = XCE_SCHED_OP_JSTATE_CHANGE;
+    req->op                       = XCE_SCHED_OP_JSTATE_CHANGE;
     req->u.job_state_change.jobid = jobid;
     req->u.job_state_change.state = state;
 
     enq_pend_update(sched, req);
-    
+
     return 0;
 }
 int sched_handle_notify(xce_scheduler_t* sched, xce_notify_data_t* notify)
@@ -216,9 +216,6 @@ int scheduler_req_handler(xce_request_t* request, void* args)
         case Request_body_datamsg:
             rc = sched_handle_dataset(sched, request->body.datamsg);
             break;
-        case Request_body_notifymsg:
-            rc = sched_handle_notify(sched, request->body.notifymsg);
-            break;
         default:
             rc = -EINVAL;
             break;

src/scheduler/src/plugin.c

@@ -72,8 +72,8 @@ static int plugin_config_update_(void* arg1, void* arg2)
     return 0;
 }
 
-int plugin_config_update(void* arg, xce_conf4sched_t* old,
+int plugin_config_update(void* arg, xce_sched_config_t* old,
-                         xce_conf4sched_t* config)
+                         xce_sched_config_t* config)
 {
     xce_scheduler_t* sched = (xce_scheduler_t*)arg;
     xlist_t          lst;

src/scheduler/src/plugins/fcfs.c

@@ -86,8 +86,9 @@ static int fcfs_priv_data_init(fcfs_plugin_data_t* data)
     return 0;
 }
 
-static int fcfs_update_config(xce_sched_plugin_t* plugin, xce_conf4sched_t* old,
+static int fcfs_update_config(xce_sched_plugin_t* plugin,
-                              xce_conf4sched_t* config)
+                              xce_sched_config_t* old,
+                              xce_sched_config_t* config)
 {
     fcfs_plugin_data_t* privdata = NULL;
 

src/scheduler/src/plugins/simple_allocator.c

@@ -33,11 +33,11 @@ static int my_update_job(xce_sched_plugin_t* plugin, xce_sched_job_t* jold,
     j->allocator = plugin;
 
     if (jold == NULL) {
-        if (JOB_IS_RUN(j)) {
+        if (JOB_IS_RUN(j->state)) {
             slotsInUse++;
         }
     } else {
-        if (JOB_IS_RUN(jold) && (JOB_IS_EXIT(j) || (JOB_IS_DONE(j)))) {
+        if (JOB_IS_RUN(jold->state) && JOB_IS_FINISH(j->state)) {
             slotsInUse--;
         }
     }
@@ -103,8 +103,7 @@ static int my_commit(xce_sched_plugin_t* plugin, xce_decision_t* decision)
 static int my_jstate_changed(xce_sched_plugin_t* plugin, xce_sched_job_t* job,
                              uint32_t state)
 {
-    if (JOB_IS_RUN(job)
+    if (JOB_IS_RUN(job->state) && JOB_IS_FINISH(state)) {
-        && ((state == JOB_STAT_EXIT) || (state == JOB_STAT_DONE))) {
         slotsInUse--;
     }
 

src/scheduler/src/schedobj.c

@@ -89,7 +89,8 @@ xce_sched_worker_t* make_sched_worker(void* arg, xce_worker_t* w)
         return NULL;
     }
 
-    worker->name = strdup(w->name);
+    worker->name   = strdup(w->name);
+    worker->status = w->status;
     /* FIXME: */
 
     return worker;
@@ -124,3 +125,42 @@ xce_sched_project_t* make_sched_project(void* arg, char* name)
 
     return project;
 }
+
+xce_sched_config_t* make_sched_config(void* arg, xce_conf4sched_t* c)
+{
+    xce_scheduler_t*    sched  = (xce_scheduler_t*)arg;
+    xce_sched_config_t* config = NULL;
+    int i;
+
+    if ((sched == NULL) || (c == NULL)) {
+        errno = EINVAL;
+        
+        return NULL;
+    }
+
+    config = (xce_sched_config_t *)calloc(1, sizeof(*config));
+    if (config == NULL) {
+        errno = ENOMEM;
+        return NULL;
+    }
+
+    /* FIXME: */
+    resmgr_init(&(config->resmgr), /* slots */128, /* delta */ 128, NULL);
+    config->clustername = strdup(c->clustername);
+
+    for(i = 0; i < c->n_resources; i ++) {
+        xce_resource_t *r = NULL;
+
+        MOVE_PTR(r, c->resources[i]);
+        resmgr_add_res(&(config->resmgr), r);
+    }
+
+    return config;
+}
+
+void kill_sched_config(void* arg)
+{
+    if (arg == NULL) {
+        return;
+    }
+}

src/scheduler/src/scheduler.c

@@ -239,11 +239,23 @@ void sched_fini(xce_scheduler_t* sched)
     return;
 }
 
+static int sched_merge_config(xce_scheduler_t *sched, xce_sched_config_t *config) {
+    if (sched->config == NULL) {
+        sched->config = config;
+        return 0;
+    }
+
+    /* FIXME: merge sched->config and config and free unused */
+    return 1;
+}
 static int sched_update_config(xce_scheduler_t* sched, xce_conf4sched_t* config)
 {
     int i;
+    xce_sched_config_t *config2 = make_sched_config(sched, config);
 
-    plugin_config_update(sched, sched->config, config);
+    plugin_config_update(sched, sched->config, config2);
+
+    sched_merge_config(sched, config2);
 
     for (i = 0; i < config->n_queues; i++) {
         xce_sched_queue_t* qold