From 5560ae91da63b564cec70ccd665a74d20c190bc8 Mon Sep 17 00:00:00 2001 From: Rongsong Shen Date: Tue, 23 Dec 2025 15:29:26 +0800 Subject: [PATCH] pass pend reason back to controller --- src/include/ordq.h | 7 +- src/include/preason.h | 7 +- src/lib/ordq.c | 51 +++++-- src/lib/preason.c | 21 +++ src/lib/testsuite/test_ordq.c | 50 ++++-- src/scheduler/include/decision.h | 6 +- src/scheduler/include/plugin.h | 4 +- src/scheduler/include/scheduler.h | 1 + src/scheduler/meson.build | 4 +- src/scheduler/src/decision.c | 7 +- src/scheduler/src/main.c | 10 ++ src/scheduler/src/plugin.c | 32 ++++ src/scheduler/src/plugin_default_allocator.c | 41 ----- src/scheduler/src/plugins/default_allocator.c | 67 ++++++++ .../src/{plugin_fcfs.c => plugins/fcfs.c} | 143 +++++++++++++----- src/scheduler/src/scheduler.c | 21 ++- 16 files changed, 349 insertions(+), 123 deletions(-) delete mode 100644 src/scheduler/src/plugin_default_allocator.c create mode 100644 src/scheduler/src/plugins/default_allocator.c rename src/scheduler/src/{plugin_fcfs.c => plugins/fcfs.c} (63%) diff --git a/src/include/ordq.h b/src/include/ordq.h index 0933636..6bb3d9c 100644 --- a/src/include/ordq.h +++ b/src/include/ordq.h @@ -14,7 +14,7 @@ typedef struct _ordq_item { typedef int (*ordq_cmp_fn_t)(void*, void*, void*); typedef void (*ordq_data_killer_t)(void*); -typedef int (*ordq_skip_fn_t)(void*, va_list); +typedef int (*ordq_match_fn_t)(void*, va_list); typedef struct { mylock_t lock; @@ -27,8 +27,9 @@ typedef struct { extern int ordq_init(ordq_t*, ordq_data_killer_t, ordq_cmp_fn_t); extern void ordq_fini(ordq_t* q); extern int ordq_put(ordq_t*, void*); -extern ordq_item_t* ordq_get(ordq_t*, ordq_skip_fn_t, ...); -extern ordq_item_t* ordq_peek(ordq_t*, ordq_skip_fn_t, ...); +extern int ordq_del(ordq_t*, ordq_match_fn_t, ...); +extern ordq_item_t* ordq_get(ordq_t*, ordq_match_fn_t, ...); +extern ordq_item_t* ordq_peek(ordq_t*, ordq_match_fn_t, ...); extern void ordq_set_ctx(ordq_t*, void*); #endif diff --git a/src/include/preason.h b/src/include/preason.h index 296ea04..fc138d2 100644 --- a/src/include/preason.h +++ b/src/include/preason.h @@ -4,7 +4,10 @@ #include "xce/types.h" -extern void erase_job_preason(xce_job_preason_t*); -extern int add_job_preason(xce_job_preason_t*, int, char*, char*); +#define PREASON_BAD_RESREQ (1) + +extern void erase_job_preason(xce_job_preason_t*); +extern int add_job_preason(xce_job_preason_t*, int, char*, char*); +extern xce_pendreason_t* make_pendreason(uint64_t); #endif diff --git a/src/lib/ordq.c b/src/lib/ordq.c index 2dc4868..28702d4 100644 --- a/src/lib/ordq.c +++ b/src/lib/ordq.c @@ -79,7 +79,7 @@ void ordq_fini(ordq_t* q) mylock_fini(&(q->lock)); } -static ordq_item_t* ordq_get_(ordq_t* q, ordq_skip_fn_t skipfn, va_list ap) +static ordq_item_t* ordq_get_(ordq_t* q, ordq_match_fn_t matchfn, va_list ap) { ordq_item_t** p = NULL; ordq_item_t* it = NULL; @@ -90,15 +90,15 @@ static ordq_item_t* ordq_get_(ordq_t* q, ordq_skip_fn_t skipfn, va_list ap) va_list ap1; int rc = 0; - if (skipfn == NULL) { + if (matchfn == NULL) { break; } va_copy(ap1, ap); - rc = skipfn((*p)->data, ap1); + rc = matchfn((*p)->data, ap1); va_end(ap1); - if (!rc) { + if (rc) { break; } @@ -114,41 +114,68 @@ static ordq_item_t* ordq_get_(ordq_t* q, ordq_skip_fn_t skipfn, va_list ap) return it; } -ordq_item_t* ordq_get(ordq_t* q, ordq_skip_fn_t skipfn, ...) +ordq_item_t* ordq_get(ordq_t* q, ordq_match_fn_t matchfn, ...) { ordq_item_t* it; va_list ap; - va_start(ap, skipfn); + va_start(ap, matchfn); mylock_lock(&(q->lock)); - it = ordq_get_(q, skipfn, ap); + it = ordq_get_(q, matchfn, ap); mylock_unlock(&(q->lock)); va_end(ap); return it; } -ordq_item_t* ordq_peek(ordq_t* q, ordq_skip_fn_t skipfn, ...) +int ordq_del(ordq_t* q, ordq_match_fn_t matchfn, ...) +{ + ordq_item_t* it = NULL; + int count = 0; + va_list ap, ap1; + + va_start(ap, matchfn); + + va_copy(ap1, ap); + + mylock_lock(&(q->lock)); + while ((it = ordq_get_(q, matchfn, ap1)) != NULL) { + if (q->killer != NULL) { + q->killer(it->data); + } + free(it); + count++; + va_copy(ap1, ap); + } + mylock_unlock(&(q->lock)); + + va_end(ap1); + va_end(ap); + + return count; +} + +ordq_item_t* ordq_peek(ordq_t* q, ordq_match_fn_t matchfn, ...) { ordq_item_t* it = NULL; va_list ap; - va_start(ap, skipfn); + va_start(ap, matchfn); mylock_lock(&(q->lock)); it = q->head; while (it != NULL) { int rc = 0; va_list ap1; - if (skipfn == NULL) { + if (matchfn == NULL) { break; } va_copy(ap1, ap); - rc = skipfn(it->data, ap1); + rc = matchfn(it->data, ap1); va_end(ap1); - if (!rc) { + if (rc) { break; } diff --git a/src/lib/preason.c b/src/lib/preason.c index 91f72cf..11411af 100644 --- a/src/lib/preason.c +++ b/src/lib/preason.c @@ -1,4 +1,5 @@ #include +#include #include "preason.h" #include "utils.h" @@ -149,3 +150,23 @@ int add_job_preason(xce_job_preason_t* reasons, int code, char* host, char* res) return rc; } + +xce_pendreason_t *make_pendreason(uint64_t jobid) { + xce_pendreason_t *reason = NULL; + xce_job_preason_t *preason = NULL; + + preason = (xce_job_preason_t *)calloc(1, sizeof(*preason)); + reason = (xce_pendreason_t*)calloc(1, sizeof(*reason)); + + if ((preason == NULL) || (reason == NULL)) { + FREEUP(reason); + FREEUP(preason); + + return NULL; + } + + reason->jobid = jobid; + reason->reasons = preason; + + return reason; +} diff --git a/src/lib/testsuite/test_ordq.c b/src/lib/testsuite/test_ordq.c index 1dbb8c0..17c101d 100644 --- a/src/lib/testsuite/test_ordq.c +++ b/src/lib/testsuite/test_ordq.c @@ -47,14 +47,14 @@ static void test_ordq_get(void) return; } -static int test_skip(void* data, va_list ap) +static int test_match(void* data, va_list ap) { int val = (int)(long)data; - return (val % 2 != 0); + return (val % 2 == 0); } -static void test_ordq_get_skip(void) +static void test_ordq_get_match(void) { int rc; ordq_t q; @@ -73,7 +73,7 @@ static void test_ordq_get_skip(void) j = 0; nc = 0; for (i = 0; i < 5; i++) { - it = ordq_get(&q, test_skip); + it = ordq_get(&q, test_match); if (it == NULL) { break; @@ -125,14 +125,14 @@ static void test_ordq_peek(void) return; } -static int test_skip2(void* data, va_list ap) +static int test_match2(void* data, va_list ap) { int val = (int)(long)data; - return (val % 3 != 0); + return (val % 3 == 0); } -static void test_ordq_peek_skip(void) +static void test_ordq_peek_match(void) { int rc; ordq_t q; @@ -150,7 +150,7 @@ static void test_ordq_peek_skip(void) for (i = 0; i < 8; i++) { it = ordq_peek( - &q, test_skip2); /* ordq_peek() will not remove item from q */ + &q, test_match2); /* ordq_peek() will not remove item from q */ TEST_ASSERT_TRUE((it != NULL) && ((int)(long)it->data == val[4])); } @@ -160,12 +160,42 @@ static void test_ordq_peek_skip(void) return; } +static void test_ordq_del(void) +{ + int rc; + ordq_t q; + ordq_item_t* it; + int keys[8] = { 9, 8, 1, 4, 7, 2, 6, 2 }; + int i; + int count; + + rc = ordq_init(&q, NULL, (ordq_cmp_fn_t)long_cmp); + TEST_ASSERT_TRUE(rc >= 0); + + for (i = 0; i < 8; i++) { + ordq_put(&q, (void*)(long)(keys[i])); + } + + count = ordq_del(&q, test_match2); + TEST_ASSERT_TRUE(count == 2); + + while ((it = ordq_get(&q, NULL)) != NULL) { + TEST_ASSERT_TRUE((int)(long)it->data % 3 != 0); + free(it); + } + + ordq_fini(&q); + + return; +} + void test_ordq(void) { printf("\n>> ordq test\n\n"); RUN_TEST(test_ordq_get); RUN_TEST(test_ordq_peek); - RUN_TEST(test_ordq_get_skip); - RUN_TEST(test_ordq_peek_skip); + RUN_TEST(test_ordq_get_match); + RUN_TEST(test_ordq_peek_match); + RUN_TEST(test_ordq_del); } diff --git a/src/scheduler/include/decision.h b/src/scheduler/include/decision.h index b247fa4..74fecbb 100644 --- a/src/scheduler/include/decision.h +++ b/src/scheduler/include/decision.h @@ -14,13 +14,13 @@ typedef enum { typedef struct { xce_decision_op_t op; - uint64_t jobid; union { - xce_job_alloc_t* alloc; + xce_job_alloc_t* alloc; + xce_pendreason_t* reason; } u; } xce_decision_t; -extern xce_decision_t* make_pend_decision(uint64_t); +extern xce_decision_t* make_pend_decision(xce_pendreason_t*); extern xce_decision_t* make_dispatch_decision(xce_job_alloc_t*); #endif diff --git a/src/scheduler/include/plugin.h b/src/scheduler/include/plugin.h index 7ae3b97..575b489 100644 --- a/src/scheduler/include/plugin.h +++ b/src/scheduler/include/plugin.h @@ -29,7 +29,8 @@ 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); +typedef xce_job_alloc_t* (*plugin_job_allocate_fn_t)(void*, int64_t, + xce_pendreason_t**); typedef struct { plugin_update_config_hook_t config_hook; @@ -60,6 +61,7 @@ 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 plugin_commit_decision(void*, xce_decision_t*); extern int register_fcfs(void*); extern int register_default_allocator(void*); diff --git a/src/scheduler/include/scheduler.h b/src/scheduler/include/scheduler.h index 7bacc87..4cf7b9a 100644 --- a/src/scheduler/include/scheduler.h +++ b/src/scheduler/include/scheduler.h @@ -81,5 +81,6 @@ 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*); extern int sched_commit_decision(xce_scheduler_t*, xce_decision_t*); +extern int sched_publish_pend_reason(xce_scheduler_t*, xce_preason_set_t*); #endif diff --git a/src/scheduler/meson.build b/src/scheduler/meson.build index 0b84644..7cb0f48 100644 --- a/src/scheduler/meson.build +++ b/src/scheduler/meson.build @@ -5,8 +5,8 @@ scheduler_srcs = [ 'src/decision.c', 'src/schedobj.c', 'src/plugin.c', - 'src/plugin_fcfs.c', - 'src/plugin_default_allocator.c', + 'src/plugins/fcfs.c', + 'src/plugins/default_allocator.c', ] incdir = include_directories('src', diff --git a/src/scheduler/src/decision.c b/src/scheduler/src/decision.c index 46deaba..786e3af 100644 --- a/src/scheduler/src/decision.c +++ b/src/scheduler/src/decision.c @@ -3,7 +3,7 @@ #include "decision.h" -xce_decision_t* make_pend_decision(uint64_t jobid) +xce_decision_t* make_pend_decision(xce_pendreason_t* reason) { xce_decision_t* decision; @@ -12,8 +12,8 @@ xce_decision_t* make_pend_decision(uint64_t jobid) return NULL; } - decision->op = XCE_DECISION_PEND; - decision->jobid = jobid; + decision->op = XCE_DECISION_PEND; + decision->u.reason = reason; return decision; } @@ -28,7 +28,6 @@ xce_decision_t* make_dispatch_decision(xce_job_alloc_t* alloc) decision->op = XCE_DECISION_DISPATCH; decision->u.alloc = alloc; - decision->jobid = alloc->jobid; return decision; } diff --git a/src/scheduler/src/main.c b/src/scheduler/src/main.c index 459168f..1d0b9bb 100644 --- a/src/scheduler/src/main.c +++ b/src/scheduler/src/main.c @@ -123,6 +123,12 @@ int main(int argc, char* argv[]) while (1) { xce_decision_t* decision = NULL; + int rc = 0; + xce_pendreason_t *preasons[1]; + xce_preason_set_t reasons = { + .nreasons = 1, + .reasons = preasons + }; cbuf_get(&(sched.decisions), &decision); @@ -141,10 +147,14 @@ int main(int argc, char* argv[]) break; case XCE_DECISION_PEND: /* set pend reason for jobs */ + preasons[ 0] = decision->u.reason; + rc = sched_publish_pend_reason(&sched, &reasons); break; default: break; } + + TRACE("%s rc %d", __func__, rc); } sched_end_session(&sched); diff --git a/src/scheduler/src/plugin.c b/src/scheduler/src/plugin.c index 465e162..5bad0d5 100644 --- a/src/scheduler/src/plugin.c +++ b/src/scheduler/src/plugin.c @@ -252,3 +252,35 @@ int plugin_end_session(void* arg) return rc; } + +static int plugin_commit_decision_(void* arg1, void* arg2) { + xce_sched_plugin_t *plugin = (xce_sched_plugin_t *)arg2; + + if (plugin == NULL) { + return 0; + } + + if (plugin->hooks.commit_hook != NULL) { + return plugin->hooks.commit_hook(plugin, arg1); + } + + return 0; +} + +int plugin_commit_decision(void* arg1, xce_decision_t* decision) +{ + xce_scheduler_t* sched = (xce_scheduler_t*)arg1; + xlist_t lst; + int rc; + + if (sched == NULL) { + return -EINVAL; + } + + xlist_init(&lst, 3); + xlist_store(&lst, plugin_commit_decision_); + rc = kvmgr_iter(&(sched->plugins), NULL, &lst, decision); + xlist_fini(&lst); + + return rc; +} diff --git a/src/scheduler/src/plugin_default_allocator.c b/src/scheduler/src/plugin_default_allocator.c deleted file mode 100644 index 495635d..0000000 --- a/src/scheduler/src/plugin_default_allocator.c +++ /dev/null @@ -1,41 +0,0 @@ -#include -#include -#include -#include -#include - -#include "plugin.h" -#include "scheduler.h" -#include "utils.h" - -static int default_allocator_init(void* arg) { return 1; } - -static void default_allocator_fini(void* arg) { return; } - -static int my_update_job(xce_sched_plugin_t *plugin, xce_sched_job_t *j) { - /* FIXME: we need to add logic to check whether the job should be handled by me */ - j->allocator = plugin; - return 1; -} - -static xce_job_alloc_t* my_alloc(xce_sched_plugin_t* plugin, int64_t jobid) -{ - return NULL; -} - -static xce_sched_plugin_t default_allocator_plugin = { - .name = "xce_sched_default_allocator", - .hooks = { .config_hook = NULL, .job_hook = (plugin_update_job_hook_t)my_update_job }, - .init_fn = (plugin_init_fn_t)default_allocator_init, - .fini_fn = (plugin_fini_fn_t)default_allocator_fini, - .begin_session_fn = NULL, - .end_session_fn = NULL, - .job_elect_fn = NULL, - .job_alloc_fn = (plugin_job_allocate_fn_t)my_alloc, - .data = NULL -}; - -int register_default_allocator(void* sched) -{ - return plugin_register(sched, &default_allocator_plugin); -} diff --git a/src/scheduler/src/plugins/default_allocator.c b/src/scheduler/src/plugins/default_allocator.c new file mode 100644 index 0000000..ea4f2dc --- /dev/null +++ b/src/scheduler/src/plugins/default_allocator.c @@ -0,0 +1,67 @@ +#include +#include +#include +#include +#include + +#include "decision.h" +#include "plugin.h" +#include "preason.h" +#include "scheduler.h" +#include "utils.h" +#include "xce/types.h" + +static int default_allocator_init(xce_sched_plugin_t* plugin) { return 1; } + +static void default_allocator_fini(xce_sched_plugin_t* plugin) { + if (plugin == NULL) { + return; + } + FREEUP(plugin); +} + +static int my_update_job(xce_sched_plugin_t* plugin, xce_sched_job_t* j) +{ + /* FIXME: we need to add logic to check whether the job should be handled by + * me */ + j->allocator = plugin; + return 1; +} + +static xce_job_alloc_t* my_alloc(xce_sched_plugin_t* plugin, int64_t jobid, + xce_pendreason_t** preason) +{ + xce_pendreason_t* reason = make_pendreason(jobid); + + add_job_preason(reason->reasons, PREASON_BAD_RESREQ, NULL, NULL); + + *preason = reason; + return NULL; +} + +int register_default_allocator(void* sched) +{ + static xce_sched_plugin_t default_allocator_plugin = { + .name = "xce_sched_default_allocator", + .hooks = { .config_hook = NULL, + .job_hook = (plugin_update_job_hook_t)my_update_job }, + .init_fn = (plugin_init_fn_t)default_allocator_init, + .fini_fn = (plugin_fini_fn_t)default_allocator_fini, + .begin_session_fn = NULL, + .end_session_fn = NULL, + .job_elect_fn = NULL, + .job_alloc_fn = (plugin_job_allocate_fn_t)my_alloc, + .data = NULL + }; + + xce_sched_plugin_t *plugin = NULL; + + plugin= (xce_sched_plugin_t *)calloc(1, sizeof(*plugin)); + if (plugin == NULL) { + return -ENOMEM; + } + + memcpy(plugin, &default_allocator_plugin, sizeof(*plugin)); + + return plugin_register(sched, plugin); +} diff --git a/src/scheduler/src/plugin_fcfs.c b/src/scheduler/src/plugins/fcfs.c similarity index 63% rename from src/scheduler/src/plugin_fcfs.c rename to src/scheduler/src/plugins/fcfs.c index f24a0e0..8cf314e 100644 --- a/src/scheduler/src/plugin_fcfs.c +++ b/src/scheduler/src/plugins/fcfs.c @@ -86,9 +86,9 @@ static int fcfs_priv_data_init(fcfs_plugin_data_t* data) return 0; } -static int fcfs_update_config(void* arg, xce_conf4sched_t* config) +static int fcfs_update_config(xce_sched_plugin_t* plugin, + xce_conf4sched_t* config) { - xce_sched_plugin_t* plugin = (xce_sched_plugin_t*)arg; fcfs_plugin_data_t* privdata = NULL; TRACE("%s ...", __func__); @@ -100,29 +100,39 @@ static int fcfs_update_config(void* arg, xce_conf4sched_t* config) return 0; } -static int skip_not_matched_queue(void* arg1, va_list ap) +static int queue_match(void* arg1, va_list ap) { fcfs_queue_t* q = (fcfs_queue_t*)arg1; char* name = va_arg(ap, char*); if ((name == NULL) || (q == NULL)) { - return 1; + return 0; } - return (strcmp(name, q->queue->name) != 0); + return (strcmp(name, q->queue->name) == 0); } -static int fcfs_update_queue(void* arg, xce_sched_queue_t* q) +static int job_match(void* arg1, va_list ap) +{ + fcfs_job_t* j = (fcfs_job_t*)arg1; + uint64_t jobid = va_arg(ap, uint64_t); + + if (j == NULL) { + return 0; + } + + return (j->job->jobid == jobid); +} + +static int fcfs_update_queue(xce_sched_plugin_t* plugin, xce_sched_queue_t* q) { - 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; fcfs_queue_t* queue = NULL; TRACE("%s queue %s...", __func__, q->name); - it = ordq_get(&(privdata->queues), (ordq_skip_fn_t)skip_not_matched_queue, - q->name); + it = ordq_get(&(privdata->queues), (ordq_match_fn_t)queue_match, q->name); if (it == NULL) { queue = (fcfs_queue_t*)calloc(1, sizeof(*queue)); if (queue == NULL) { @@ -146,8 +156,7 @@ static fcfs_queue_t* fcfs_get_queue(fcfs_plugin_data_t* data, char* name) { ordq_item_t* it = NULL; - it = ordq_peek(&(data->queues), (ordq_skip_fn_t)skip_not_matched_queue, - name); + it = ordq_peek(&(data->queues), (ordq_match_fn_t)queue_match, name); if (it == NULL) { return NULL; } @@ -155,9 +164,8 @@ static fcfs_queue_t* fcfs_get_queue(fcfs_plugin_data_t* data, char* name) return (fcfs_queue_t*)it->data; } -static int fcfs_update_job(void* arg, xce_sched_job_t* j) +static int fcfs_update_job(xce_sched_plugin_t* plugin, xce_sched_job_t* j) { - xce_sched_plugin_t* plugin = (xce_sched_plugin_t*)arg; fcfs_plugin_data_t* privdata = (fcfs_plugin_data_t*)plugin->data; fcfs_queue_t* q = NULL; fcfs_job_t* job = NULL; @@ -198,6 +206,7 @@ static void* fcfs_elect_job(xce_sched_plugin_t* plugin) ordq_t* queues = &(privdata->queues); ordq_item_t* it = NULL; int go = 0; + uint64_t jobid = 0; pthread_testcancel(); @@ -222,7 +231,7 @@ static void* fcfs_elect_job(xce_sched_plugin_t* plugin) pthread_testcancel(); if (sched_job_is_ready(j->job)) { - uint64_t jobid = j->job->jobid; + jobid = j->job->jobid; TRACE("%s jobid %ld", __func__, jobid); cbuf_put(&(sched->jbuf), &jobid); @@ -238,15 +247,18 @@ static void* fcfs_elect_job(xce_sched_plugin_t* plugin) } } + /* end of scheduling session */ + jobid = 0; + cbuf_put(&(sched->jbuf), &jobid); + sleep(1); } return NULL; } -static int fcfs_end_session(void* arg) +static int fcfs_end_session(xce_sched_plugin_t* plugin) { - xce_sched_plugin_t* plugin = (xce_sched_plugin_t*)arg; fcfs_plugin_data_t* privdata = (fcfs_plugin_data_t*)plugin->data; privdata->new_session = 1; @@ -254,9 +266,8 @@ static int fcfs_end_session(void* arg) return 0; } -static int fcfs_init(void* arg) +static int fcfs_init(xce_sched_plugin_t* plugin) { - 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)); @@ -270,32 +281,86 @@ static int fcfs_init(void* arg) return 1; } -static void fcfs_fini(void* arg) +static void fcfs_fini(xce_sched_plugin_t* plugin) { - xce_sched_plugin_t* plugin = (xce_sched_plugin_t*)arg; - fcfs_plugin_data_t* privdata = (fcfs_plugin_data_t*)plugin->data; + fcfs_plugin_data_t* privdata = NULL; - if (privdata == NULL) { + if (plugin == NULL) { return; } - ordq_fini(&(privdata->queues)); - free(privdata); + privdata = (fcfs_plugin_data_t*)plugin->data; + if (privdata != NULL) { + ordq_fini(&(privdata->queues)); + free(privdata); + } + + FREEUP(plugin); } -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 }, - .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 -}; +static int fcfs_del_job(xce_sched_plugin_t* plugin, uint64_t jobid) +{ + fcfs_plugin_data_t* privdata = (fcfs_plugin_data_t*)plugin->data; + xce_sched_job_t* job = NULL; + fcfs_queue_t* q = NULL; -int register_fcfs(void* sched) { return plugin_register(sched, &fcfs_plugin); } + if (privdata == NULL) { + return 0; + } + + job = sched_get_job(plugin->sched, jobid); + if (job == NULL) { + return 0; + } + + q = fcfs_get_queue(privdata, job->queue->name); + if (q == NULL) { + return 0; + } + + return ordq_del(&(q->pendJobs), (ordq_match_fn_t)job_match, jobid); +} + +static int fcfs_commit(xce_sched_plugin_t* plugin, xce_decision_t* decision) +{ + int rc = 0; + switch (decision->op) { + case XCE_DECISION_DISPATCH: + rc = fcfs_del_job(plugin, decision->u.alloc->jobid); + break; + default: + break; + } + return rc; +} + +int register_fcfs(void* sched) +{ + static xce_sched_plugin_t fcfs_plugin = { + .name = "xce_sched_fcfs", + .hooks + = { .config_hook = (plugin_update_config_hook_t)fcfs_update_config, + .queue_hook = (plugin_update_queue_hook_t)fcfs_update_queue, + .job_hook = (plugin_update_job_hook_t)fcfs_update_job, + .commit_hook = (plugin_commit_decision_hook_t)fcfs_commit, + .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 + }; + + xce_sched_plugin_t* plugin + = (xce_sched_plugin_t*)calloc(1, sizeof(*plugin)); + + if (plugin == NULL) { + return -ENOMEM; + } + + memcpy(plugin, &fcfs_plugin, sizeof(*plugin)); + + return plugin_register(sched, plugin); +} diff --git a/src/scheduler/src/scheduler.c b/src/scheduler/src/scheduler.c index 6c27ec6..ff46a6b 100644 --- a/src/scheduler/src/scheduler.c +++ b/src/scheduler/src/scheduler.c @@ -33,6 +33,7 @@ static xce_decision_t* calc_decision(xce_scheduler_t* sched, uint64_t jobid) xce_sched_job_t* job = sched_get_job(sched, jobid); xce_sched_plugin_t* plugin = NULL; xce_job_alloc_t* alloc = NULL; + xce_pendreason_t* reason = NULL; if (job == NULL) { ERROR("%s job %lu can not be found", __func__, jobid); @@ -51,15 +52,16 @@ static xce_decision_t* calc_decision(xce_scheduler_t* sched, uint64_t jobid) return NULL; } - alloc = plugin->job_alloc_fn(job->allocator, jobid); + alloc = plugin->job_alloc_fn(job->allocator, jobid, &reason); if (alloc == NULL) { - return make_pend_decision(jobid); + return make_pend_decision(reason); } return make_dispatch_decision(alloc); } -static int commit_decision(xce_scheduler_t *sched, xce_decision_t *decision) { +static int commit_decision(xce_scheduler_t* sched, xce_decision_t* decision) +{ return sched_commit_decision(sched, decision); } @@ -86,7 +88,7 @@ static void* sched_allocator(void* arg) decision = calc_decision(sched, jobid); commit_decision(sched, decision); - + cbuf_put(&(sched->decisions), &decision); } @@ -401,6 +403,13 @@ int sched_job_is_ready(xce_sched_job_t* j) { return 1; } int sched_queue_is_ready(xce_sched_queue_t* q) { return 1; } -int sched_commit_decision(xce_scheduler_t *sched, xce_decision_t *decision) { - return 0; +int sched_commit_decision(xce_scheduler_t* sched, xce_decision_t* decision) +{ + return plugin_commit_decision(sched, decision); +} + +int sched_publish_pend_reason(xce_scheduler_t* sched, + xce_preason_set_t* reasons) +{ + return xcecli_set_pend_reason(sched->apictx, reasons); }