diff --git a/src/scheduler/include/decision.h b/src/scheduler/include/decision.h new file mode 100644 index 00000000..cd06e749 --- /dev/null +++ b/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 diff --git a/src/scheduler/include/scheduler.h b/src/scheduler/include/scheduler.h index f8187ad4..0ecdae65 100644 --- a/src/scheduler/include/scheduler.h +++ b/src/scheduler/include/scheduler.h @@ -4,26 +4,46 @@ #include +#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_DECISION_DISPATCH, - XCE_DECISION_PEND -} xce_decision_op_t; + XCE_SCHED_OP_NONE = 0, + XCE_SCHED_OP_UPDATE_CONFIG, + XCE_SCHED_OP_UPDATE_JOB, + 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 */ diff --git a/src/scheduler/meson.build b/src/scheduler/meson.build index 4c4292f4..33ecc930 100644 --- a/src/scheduler/meson.build +++ b/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', diff --git a/src/scheduler/src/decision.c b/src/scheduler/src/decision.c new file mode 100644 index 00000000..cb49eead --- /dev/null +++ b/src/scheduler/src/decision.c @@ -0,0 +1,5 @@ +#include +#include + +#include "decision.h" + diff --git a/src/scheduler/src/handler.c b/src/scheduler/src/handler.c index b6525988..d6c94c60 100644 --- a/src/scheduler/src/handler.c +++ b/src/scheduler/src/handler.c @@ -4,16 +4,37 @@ #include #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; diff --git a/src/scheduler/src/main.c b/src/scheduler/src/main.c index fb35e477..e6f63b5c 100644 --- a/src/scheduler/src/main.c +++ b/src/scheduler/src/main.c @@ -1,4 +1,5 @@ #include +#include #include #include #include @@ -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); } diff --git a/src/scheduler/src/scheduler.c b/src/scheduler/src/scheduler.c index de2b26c9..09dba3ee 100644 --- a/src/scheduler/src/scheduler.c +++ b/src/scheduler/src/scheduler.c @@ -6,6 +6,7 @@ #include #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) {