diff --git a/src/include/client.h b/src/include/client.h index cfbb6857..5b301a32 100644 --- a/src/include/client.h +++ b/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) diff --git a/src/scheduler/include/scheduler.h b/src/scheduler/include/scheduler.h index f7a93a96..40611fe4 100644 --- a/src/scheduler/include/scheduler.h +++ b/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 diff --git a/src/scheduler/meson.build b/src/scheduler/meson.build index 76b5e64f..4c4292f4 100644 --- a/src/scheduler/meson.build +++ b/src/scheduler/meson.build @@ -1,6 +1,7 @@ scheduler_srcs = [ 'src/main.c', 'src/handler.c', + 'src/scheduler.c', ] incdir = include_directories('src', diff --git a/src/scheduler/src/handler.c b/src/scheduler/src/handler.c index 85ca075e..b6525988 100644 --- a/src/scheduler/src/handler.c +++ b/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; diff --git a/src/scheduler/src/main.c b/src/scheduler/src/main.c index 247683e4..1130d851 100644 --- a/src/scheduler/src/main.c +++ b/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"); diff --git a/src/scheduler/src/scheduler.c b/src/scheduler/src/scheduler.c new file mode 100644 index 00000000..6ee42cba --- /dev/null +++ b/src/scheduler/src/scheduler.c @@ -0,0 +1,101 @@ +#include +#include +#include +#include +#include + +#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; +}