diff --git a/src/scheduler/include/plugin.h b/src/scheduler/include/plugin.h index 60a0699..cea6c5c 100644 --- a/src/scheduler/include/plugin.h +++ b/src/scheduler/include/plugin.h @@ -21,8 +21,11 @@ typedef int (*plugin_init_fn_t)(void*); typedef void (*plugin_fini_fn_t)(void*); /* plugin function to elect job to be scheduler */ -typedef int64_t (*plugin_job_election_fn_t)(void*); +typedef void* (*plugin_job_election_fn_t)(void*); +typedef int (*plugin_begin_session_fn_t)(void*); + +typedef int (*plugin_end_session_fn_t)(void*); /* allocation function of job */ typedef xce_job_alloc_t* (*plugin_job_allocate_fn_t)(void*, int64_t*); @@ -34,14 +37,16 @@ typedef struct { } plugin_hooks_t; typedef struct { - char* name; - void* sched; - plugin_init_fn_t init_fn; - plugin_fini_fn_t fini_fn; - plugin_job_election_fn_t job_elect_fn; - plugin_job_allocate_fn_t job_alloc_fn; - plugin_hooks_t hooks; - plugin_data_t data; + char* name; + void* sched; + plugin_init_fn_t init_fn; + plugin_fini_fn_t fini_fn; + plugin_begin_session_fn_t begin_session_fn; + plugin_end_session_fn_t end_session_fn; + plugin_job_election_fn_t job_elect_fn; + plugin_job_allocate_fn_t job_alloc_fn; + plugin_hooks_t hooks; + plugin_data_t data; } xce_sched_plugin_t; extern int plugin_register(void*, void*); @@ -50,6 +55,8 @@ extern int plugin_config_update(void*, xce_conf4sched_t*); extern int plugin_queue_update(void*, xce_sched_queue_t*); extern int plugin_job_update(void*, xce_sched_job_t*); extern int plugin_worker_update(void*, xce_sched_worker_t*); +extern int plugin_begin_session(void*); +extern int plugin_end_session(void*); extern int register_fcfs(void*); diff --git a/src/scheduler/include/scheduler.h b/src/scheduler/include/scheduler.h index acd3d5d..ea89ad2 100644 --- a/src/scheduler/include/scheduler.h +++ b/src/scheduler/include/scheduler.h @@ -66,6 +66,8 @@ extern int sched_init(xce_scheduler_t*, xce_client_config_t*, extern void sched_fini(xce_scheduler_t*); extern int sched_update(xce_scheduler_t*); extern void sched_go(xce_scheduler_t*); +extern void sched_begin_session(xce_scheduler_t*); +extern void sched_end_session(xce_scheduler_t*); extern xce_sched_job_t* sched_get_job(xce_scheduler_t*, uint64_t); extern xce_sched_queue_t* sched_get_queue(xce_scheduler_t*, char*); extern xce_sched_worker_t* sched_get_worker(xce_scheduler_t*, char*); @@ -77,5 +79,6 @@ extern int sched_add_user(xce_scheduler_t*, xce_sched_user_t*); extern int sched_add_worker(xce_scheduler_t*, xce_sched_worker_t*); extern int sched_add_project(xce_scheduler_t*, xce_sched_project_t*); extern int sched_job_is_ready(xce_sched_job_t*); +extern int sched_queue_is_ready(xce_sched_queue_t*); #endif diff --git a/src/scheduler/src/main.c b/src/scheduler/src/main.c index a494df0..459168f 100644 --- a/src/scheduler/src/main.c +++ b/src/scheduler/src/main.c @@ -71,7 +71,7 @@ static void stop_scheduler(xce_scheduler_t* sched) INFO("stop to scheduling jobs..."); sched->stop = 1; - cbuf_put(&(sched->decisions), &decision); + cbuf_tryput(&(sched->decisions), &decision); } static void sig_handler(int signo) @@ -118,6 +118,7 @@ int main(int argc, char* argv[]) while (1) { sched_update(&sched); + sched_begin_session(&sched); sched_go(&sched); while (1) { @@ -146,6 +147,8 @@ int main(int argc, char* argv[]) } } + sched_end_session(&sched); + if (sched.stop) { break; } diff --git a/src/scheduler/src/plugin.c b/src/scheduler/src/plugin.c index e75aedd..465e162 100644 --- a/src/scheduler/src/plugin.c +++ b/src/scheduler/src/plugin.c @@ -186,3 +186,69 @@ int plugin_worker_update(void* arg, xce_sched_worker_t* worker) return rc; } + +static int plugin_begin_session_(void* arg1, void* arg2) +{ + xce_sched_plugin_t* plugin = (xce_sched_plugin_t*)arg2; + + if (plugin == NULL) { + return 0; + } + + if (plugin->begin_session_fn != NULL) { + return plugin->begin_session_fn(plugin); + } + + return 0; +} + +int plugin_begin_session(void* arg) +{ + xce_scheduler_t* sched = (xce_scheduler_t*)arg; + xlist_t lst; + int rc; + + if (sched == NULL) { + return -EINVAL; + } + + xlist_init(&lst, 3); + xlist_store(&lst, plugin_begin_session_); + rc = kvmgr_iter(&(sched->plugins), NULL, &lst, NULL); + xlist_fini(&lst); + + return rc; +} + +static int plugin_end_session_(void* arg1, void* arg2) +{ + xce_sched_plugin_t* plugin = (xce_sched_plugin_t*)arg2; + + if (plugin == NULL) { + return 0; + } + + if (plugin->end_session_fn != NULL) { + return plugin->end_session_fn(plugin); + } + + return 0; +} + +int plugin_end_session(void* arg) +{ + xce_scheduler_t* sched = (xce_scheduler_t*)arg; + xlist_t lst; + int rc; + + if (sched == NULL) { + return -EINVAL; + } + + xlist_init(&lst, 3); + xlist_store(&lst, plugin_end_session_); + rc = kvmgr_iter(&(sched->plugins), NULL, &lst, NULL); + xlist_fini(&lst); + + return rc; +} diff --git a/src/scheduler/src/plugin_fcfs.c b/src/scheduler/src/plugin_fcfs.c index e26138a..f24a0e0 100644 --- a/src/scheduler/src/plugin_fcfs.c +++ b/src/scheduler/src/plugin_fcfs.c @@ -20,6 +20,7 @@ typedef struct { typedef struct { ordq_t queues; + int new_session; } fcfs_plugin_data_t; static void fcfs_job_killer(void* arg) @@ -81,6 +82,7 @@ static int fcfs_priv_data_init(fcfs_plugin_data_t* data) { ordq_init(&(data->queues), (ordq_data_killer_t)fcfs_queue_killer, (ordq_cmp_fn_t)fcfs_queue_cmp); + data->new_session = 0; return 0; } @@ -92,15 +94,8 @@ static int fcfs_update_config(void* arg, xce_conf4sched_t* config) TRACE("%s ...", __func__); privdata = (fcfs_plugin_data_t*)plugin->data; - if (privdata == NULL) { - privdata = (fcfs_plugin_data_t*)calloc(1, sizeof(*privdata)); - if (privdata == NULL) { - return -ENOMEM; - } - fcfs_priv_data_init(privdata); - plugin->data = privdata; - } + xce_use(privdata); return 0; } @@ -194,73 +189,113 @@ static int fcfs_update_job(void* arg, xce_sched_job_t* j) return 0; } -static int skip_no_pend_job_queue(void* arg1, va_list ap) +static void* fcfs_elect_job(xce_sched_plugin_t* plugin) { - fcfs_queue_t* q = (fcfs_queue_t*)arg1; + xce_scheduler_t* sched = plugin->sched; + fcfs_plugin_data_t* privdata = (fcfs_plugin_data_t*)plugin->data; - if (q == NULL) { - return 1; + while (1) { + ordq_t* queues = &(privdata->queues); + ordq_item_t* it = NULL; + int go = 0; + + pthread_testcancel(); + + cbuf_get(&(sched->wbuf), &go); + privdata->new_session = 0; + + for (it = queues->head; it != NULL; it = it->next) { + fcfs_queue_t* q = (fcfs_queue_t*)it->data; + ordq_t* pendJobs = NULL; + ordq_item_t* it2 = NULL; + + pthread_testcancel(); + + if (!sched_queue_is_ready(q->queue)) { + continue; + } + + pendJobs = &(q->pendJobs); + for (it2 = pendJobs->head; it2 != NULL; it2 = it2->next) { + fcfs_job_t* j = (fcfs_job_t*)it2->data; + + pthread_testcancel(); + + if (sched_job_is_ready(j->job)) { + uint64_t jobid = j->job->jobid; + + TRACE("%s jobid %ld", __func__, jobid); + cbuf_put(&(sched->jbuf), &jobid); + } + + if (privdata->new_session) { + break; + } + } + + if (privdata->new_session) { + break; + } + } + + sleep(1); } - return (ordq_peek(&(q->pendJobs), NULL) == NULL); + return NULL; } -static int skip_not_ready_job(void* arg1, va_list ap) -{ - fcfs_job_t* job = (fcfs_job_t*)arg1; - - return !sched_job_is_ready(job->job); -} - -static int64_t fcfs_elect_job(void* arg) +static int fcfs_end_session(void* arg) { xce_sched_plugin_t* plugin = (xce_sched_plugin_t*)arg; fcfs_plugin_data_t* privdata = (fcfs_plugin_data_t*)plugin->data; - ordq_item_t* it = NULL; - fcfs_queue_t* q = NULL; - fcfs_job_t* job = NULL; - uint64_t jobid; - TRACE("%s ...", __func__); + privdata->new_session = 1; - it = ordq_peek(&(privdata->queues), (ordq_skip_fn_t)skip_no_pend_job_queue); - if (it == NULL) { - TRACE("%s no job to be elected", __func__); + return 0; +} - return 0; +static int fcfs_init(void* arg) +{ + xce_sched_plugin_t* plugin = (xce_sched_plugin_t*)arg; + fcfs_plugin_data_t* privdata = NULL; + + privdata = (fcfs_plugin_data_t*)calloc(1, sizeof(*privdata)); + if (privdata == NULL) { + return -ENOMEM; } - q = (fcfs_queue_t*)it->data; - it = ordq_get(&(q->pendJobs), skip_not_ready_job); - if (it == NULL) { - TRACE("%s no job to be elected", __func__); + fcfs_priv_data_init(privdata); + plugin->data = privdata; - return 0; + return 1; +} + +static void fcfs_fini(void* arg) +{ + xce_sched_plugin_t* plugin = (xce_sched_plugin_t*)arg; + fcfs_plugin_data_t* privdata = (fcfs_plugin_data_t*)plugin->data; + + if (privdata == NULL) { + return; } - job = (fcfs_job_t*)it->data; - jobid = job->job->jobid; - free(job); - free(it); - - if (1) { - char idstring[128]; - - fmt_jobid(idstring, sizeof(idstring) - 1, jobid); - TRACE("%s job %s has been elected", __func__, idstring); - } - return jobid; + ordq_fini(&(privdata->queues)); + free(privdata); } static xce_sched_plugin_t fcfs_plugin = { - .name = "xce_sched_fcfs", - .hooks = { .config_hook = fcfs_update_config, - .queue_hook = fcfs_update_queue, - .job_hook = fcfs_update_job, - .worker_hook = NULL }, - .job_elect_fn = fcfs_elect_job, - .job_alloc_fn = NULL, - .data = NULL + .name = "xce_sched_fcfs", + .hooks = { .config_hook = fcfs_update_config, + .queue_hook = fcfs_update_queue, + .job_hook = fcfs_update_job, + .worker_hook = NULL }, + .init_fn = (plugin_init_fn_t)fcfs_init, + .fini_fn = (plugin_fini_fn_t)fcfs_fini, + .begin_session_fn = NULL, + .end_session_fn = (plugin_end_session_fn_t)fcfs_end_session, + .job_elect_fn = (plugin_job_election_fn_t)fcfs_elect_job, + .job_alloc_fn = NULL, + .data = NULL }; int register_fcfs(void* sched) { return plugin_register(sched, &fcfs_plugin); } diff --git a/src/scheduler/src/scheduler.c b/src/scheduler/src/scheduler.c index 7796e41..33aa3f6 100644 --- a/src/scheduler/src/scheduler.c +++ b/src/scheduler/src/scheduler.c @@ -13,54 +13,18 @@ #include "scheduler.h" #include "utils.h" -static uint64_t pick_job(xce_scheduler_t* sched) -{ - static xce_sched_plugin_t* job_picker = NULL; - uint64_t jobid = 0; - - if (job_picker == NULL) { - char* job_picker_plugin = "xce_sched_fcfs"; - - /* set job_pick_plugin according configuration */ - job_picker - = kvmgr_lock(&(sched->plugins), job_picker_plugin, OBJREF_NOP); - if ((job_picker == NULL) || (job_picker->job_elect_fn == NULL)) { - /* something wrong */ - return 0; - } - } - - if (job_picker->job_elect_fn != NULL) { - jobid = job_picker->job_elect_fn(job_picker); - } - - return jobid; -} - static void* sched_job_picker(void* arg) { - xce_scheduler_t* sched = (xce_scheduler_t*)arg; + xce_sched_plugin_t* job_picker = NULL; + xce_scheduler_t* sched = (xce_scheduler_t*)arg; + char* job_pick_plugin = "xce_sched_fcfs"; - 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); + job_picker = kvmgr_lock(&(sched->plugins), job_pick_plugin, OBJREF_NOP); + if ((job_picker == NULL) || (job_picker->job_elect_fn == NULL)) { + return NULL; } - return NULL; + return job_picker->job_elect_fn(job_picker); } static xce_decision_t* calc_decision(xce_scheduler_t* sched, uint64_t jobid) @@ -335,6 +299,20 @@ void sched_go(xce_scheduler_t* sched) return; } +void sched_begin_session(xce_scheduler_t* sched) +{ + plugin_begin_session(sched); + + return; +} + +void sched_end_session(xce_scheduler_t* sched) +{ + plugin_end_session(sched); + + return; +} + int sched_add_job(xce_scheduler_t* sched, xce_sched_job_t* j) { return kvmgr_add(&(sched->jobs), (void*)j->jobid, j); @@ -386,3 +364,5 @@ xce_sched_project_t* sched_get_project(xce_scheduler_t* sched, char* name) } int sched_job_is_ready(xce_sched_job_t* j) { return 1; } + +int sched_queue_is_ready(xce_sched_queue_t* q) { return 1; }