diff --git a/src/controller/cluster.c b/src/controller/cluster.c index 58580ed..1fd1d7a 100644 --- a/src/controller/cluster.c +++ b/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" }; diff --git a/src/include/xce/types.h b/src/include/xce/types.h index d8840e2..9df50cc 100644 --- a/src/include/xce/types.h +++ b/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_RUN(j) ((j->state & JOB_STAT_RUN) != 0) -#define JOB_IS_EXIT(j) ((j->state & JOB_STAT_EXIT) != 0) -#define JOB_IS_DONE(j) ((j->state & JOB_STAT_DONE) != 0) +#define JOB_IS_PEND(state) (((state) & JOB_STAT_PEND) != 0) +#define JOB_IS_RUN(state) (((state) & JOB_STAT_RUN) != 0) +#define JOB_IS_EXIT(state) (((state) & JOB_STAT_EXIT) != 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, diff --git a/src/scheduler/include/plugin.h b/src/scheduler/include/plugin.h index b5e9c8f..449b36e 100644 --- a/src/scheduler/include/plugin.h +++ b/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*, - xce_conf4sched_t*); +typedef int (*plugin_update_config_hook_t)(void*, xce_sched_config_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*, diff --git a/src/scheduler/include/sched/types.h b/src/scheduler/include/sched/types.h index 5669de6..99af412 100644 --- a/src/scheduler/include/sched/types.h +++ b/src/scheduler/include/sched/types.h @@ -5,6 +5,7 @@ #include #include +#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*); diff --git a/src/scheduler/include/scheduler.h b/src/scheduler/include/scheduler.h index 4baca3e..b8adcb8 100644 --- a/src/scheduler/include/scheduler.h +++ b/src/scheduler/include/scheduler.h @@ -49,22 +49,22 @@ typedef struct { link))) typedef struct { - mylock_t lock; - int stop; - xcecli_ctx_t apictx; - xce_conf4sched_t* config; - kvmgr_t plugins; - dataq_t* pend_updates[2]; - kvmgr_t jobs; - kvmgr_t queues; - kvmgr_t workers; - kvmgr_t users; - kvmgr_t projects; - pthread_t jobpicker; - pthread_t allocator; - cbuf_t wbuf; /* action to trigger scheduling */ - cbuf_t jbuf; /* job picked for allocation */ - cbuf_t decisions; /* scheduler decisions */ + mylock_t lock; + int stop; + xcecli_ctx_t apictx; + xce_sched_config_t* config; + kvmgr_t plugins; + dataq_t* pend_updates[2]; + kvmgr_t jobs; + kvmgr_t queues; + kvmgr_t workers; + kvmgr_t users; + kvmgr_t projects; + pthread_t jobpicker; + pthread_t allocator; + cbuf_t wbuf; /* action to trigger scheduling */ + cbuf_t jbuf; /* job picked for allocation */ + cbuf_t decisions; /* scheduler decisions */ } xce_scheduler_t; extern int sched_init(xce_scheduler_t*, xce_client_config_t*, diff --git a/src/scheduler/src/handler.c b/src/scheduler/src/handler.c index aa8a9fa..3e9f016 100644 --- a/src/scheduler/src/handler.c +++ b/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]; - xce_sched_update_request_t *req = NULL; + char jobidString[128]; + 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; diff --git a/src/scheduler/src/plugin.c b/src/scheduler/src/plugin.c index 0e4fe2a..f662937 100644 --- a/src/scheduler/src/plugin.c +++ b/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, - xce_conf4sched_t* config) +int plugin_config_update(void* arg, xce_sched_config_t* old, + xce_sched_config_t* config) { xce_scheduler_t* sched = (xce_scheduler_t*)arg; xlist_t lst; diff --git a/src/scheduler/src/plugins/fcfs.c b/src/scheduler/src/plugins/fcfs.c index 63b62ae..026fbf8 100644 --- a/src/scheduler/src/plugins/fcfs.c +++ b/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, - xce_conf4sched_t* config) +static int fcfs_update_config(xce_sched_plugin_t* plugin, + xce_sched_config_t* old, + xce_sched_config_t* config) { fcfs_plugin_data_t* privdata = NULL; diff --git a/src/scheduler/src/plugins/simple_allocator.c b/src/scheduler/src/plugins/simple_allocator.c index dcf7ad0..25d3ef5 100644 --- a/src/scheduler/src/plugins/simple_allocator.c +++ b/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) - && ((state == JOB_STAT_EXIT) || (state == JOB_STAT_DONE))) { + if (JOB_IS_RUN(job->state) && JOB_IS_FINISH(state)) { slotsInUse--; } diff --git a/src/scheduler/src/schedobj.c b/src/scheduler/src/schedobj.c index ceec602..721b020 100644 --- a/src/scheduler/src/schedobj.c +++ b/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; + } +} diff --git a/src/scheduler/src/scheduler.c b/src/scheduler/src/scheduler.c index 11ed20d..0c448d5 100644 --- a/src/scheduler/src/scheduler.c +++ b/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