From c478ea8516960093a36f616f7018e031a1111671 Mon Sep 17 00:00:00 2001 From: Rongsong Shen Date: Tue, 16 Dec 2025 16:08:20 +0800 Subject: [PATCH] scheduler framework --- src/scheduler/include/scheduler.h | 10 ++- src/scheduler/src/main.c | 28 ++++-- src/scheduler/src/scheduler.c | 140 ++++++++++++++++++++++-------- 3 files changed, 137 insertions(+), 41 deletions(-) diff --git a/src/scheduler/include/scheduler.h b/src/scheduler/include/scheduler.h index c6877198..377887bf 100644 --- a/src/scheduler/include/scheduler.h +++ b/src/scheduler/include/scheduler.h @@ -2,6 +2,8 @@ #define _SCHEDULER_H_ 1 +#include + #include "client.h" #include "xce/types.h" #include "xlist.h" @@ -21,13 +23,17 @@ typedef struct { typedef struct { xcecli_ctx_t apictx; - xlist_t decisions; + 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*, 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*); +extern void sched_go(xce_scheduler_t*); #endif diff --git a/src/scheduler/src/main.c b/src/scheduler/src/main.c index 1130d851..d1342a22 100644 --- a/src/scheduler/src/main.c +++ b/src/scheduler/src/main.c @@ -70,7 +70,6 @@ int main(int argc, char* argv[]) int logcount = 10; xce_client_config_t config; xce_scheduler_t sched; - int njobs = 0; memset(&arguments, 0x0, sizeof(arguments)); @@ -94,13 +93,32 @@ int main(int argc, char* argv[]) while (1) { sched_update(&sched); - njobs = sched_go(&sched); + sched_go(&sched); - if (njobs > 0) { - /* flush decision list to controller */ + while(1) { + xce_decision_t *decision = NULL; + + cbuf_get(&(sched.decisions), &decision); + + if (decision == NULL) { + /* start new session */ + break; + } + + switch(decision->op) { + case XCE_DECISION_NONE: + break; + case XCE_DECISION_DISPATCH: + /* tell controller to dispatch job */ + break; + case XCE_DECISION_PEND: + /* set pend reason for jobs */ + break; + default: + break; + } } - /* handle scheduler core outputs */ sleep(1); } diff --git a/src/scheduler/src/scheduler.c b/src/scheduler/src/scheduler.c index 6ee42cba..c3fe277f 100644 --- a/src/scheduler/src/scheduler.c +++ b/src/scheduler/src/scheduler.c @@ -1,4 +1,5 @@ #include +#include #include #include #include @@ -6,11 +7,78 @@ #include "client.h" #include "scheduler.h" +#include "utils.h" + +static uint64_t pick_job(xce_scheduler_t* sched) +{ + uint64_t jobid = 0; + + return jobid; +} + +static void* sched_job_picker(void* arg) +{ + xce_scheduler_t* sched = (xce_scheduler_t*)arg; + + 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); + } + + return NULL; +} + +static xce_decision_t *calc_decision(xce_scheduler_t *sched, uint64_t jobid) { + return NULL; +} + +static void* sched_allocator(void* arg) +{ + xce_scheduler_t* sched = (xce_scheduler_t*)arg; + + xce_use(sched); + while (1) { + uint64_t jobid; + xce_decision_t* decision = NULL; + + pthread_testcancel(); + + cbuf_get(&(sched->jbuf), &jobid); + if (jobid == 0) { + /* end of scheduling session */ + cbuf_put(&(sched->decisions), NULL); + continue; + } + + /* calculate decision for jobid */ + decision = calc_decision(sched, jobid); + + cbuf_put(&(sched->decisions), &decision); + } + + return NULL; +} int sched_init(xce_scheduler_t* sched, xce_client_config_t* config, xce_request_handler_t handler) { - int rc = 0; + int rc = 0; + int ndecisions = 5; /* FIXME: make this configurable */ + int njobs = 1; /* FIXME: make this configurable */ if ((sched == NULL) || (config == NULL) || (handler == NULL)) { return -EINVAL; @@ -18,8 +86,9 @@ int sched_init(xce_scheduler_t* sched, xce_client_config_t* config, memset(sched, 0x0, sizeof(*sched)); - /* FIXME: configurable delta for xlist_init()? */ - xlist_init(&(sched->decisions), 32); + cbuf_init(&(sched->wbuf), 1, sizeof(int)); + cbuf_init(&(sched->jbuf), njobs, sizeof(uint64_t)); + cbuf_init(&(sched->decisions), ndecisions, sizeof(xce_decision_t*)); rc = xcecli_ctx_init(&(sched->apictx), config); if (rc < 0) { @@ -33,16 +102,43 @@ int sched_init(xce_scheduler_t* sched, xce_client_config_t* config, return rc; } - return xcecli_join(sched->apictx, HostRole_scheduler, NULL); + rc = xcecli_join(sched->apictx, HostRole_scheduler, NULL); + if (rc < 0) { + return rc; + } + + rc = pthread_create(&(sched->jobpicker), NULL, &sched_job_picker, sched); + if (rc < 0) { + return rc; + } + + rc = pthread_create(&(sched->allocator), NULL, &sched_allocator, sched); + + return rc; } void sched_fini(xce_scheduler_t* sched) { + void* res; + + if (sched == NULL) { + return; + } + + pthread_cancel(sched->jobpicker); + pthread_join(sched->jobpicker, &res); + + pthread_cancel(sched->allocator); + pthread_join(sched->allocator, &res); + xcecli_wakeup(sched->apictx, (int)-EINTR); xcecli_stop(sched->apictx); xcecli_ctx_fini(sched->apictx); - xlist_fini(&(sched->decisions)); + cbuf_fini(&(sched->decisions)); + cbuf_fini(&(sched->jbuf)); + cbuf_fini(&(sched->wbuf)); + return; } @@ -67,35 +163,11 @@ static int sched_update_workers(xce_scheduler_t* sched, int nworkers, 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) +void sched_go(xce_scheduler_t* sched) { - *decision = NULL; + int go = 1; - 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; + cbuf_put(&(sched->wbuf), &go); + + return; }