diff --git a/.kiro/specs/scheduler-core/tasks.md b/.kiro/specs/scheduler-core/tasks.md index 6c996377..103cb1fa 100644 --- a/.kiro/specs/scheduler-core/tasks.md +++ b/.kiro/specs/scheduler-core/tasks.md @@ -32,7 +32,7 @@ - _Requirements: 3.1, 3.3_ - _Depends: 1.1, 1.2_ -- [ ] 2.2 Implement builtin_fcfs state-change hooks +- [x] 2.2 Implement builtin_fcfs state-change hooks - `on_queue_update`: upsert a queue into the picker's ordered queue list, keyed by descending priority - `on_job_push`: add a newly-pending job into its queue's candidate set, ordered by priority desc then submission/queue time asc - `on_job_state_change`: remove a job from candidates when it leaves pending state, re-add it when it returns to pending diff --git a/src/scheduler/src/plugins/builtin_fcfs.c b/src/scheduler/src/plugins/builtin_fcfs.c index 10d7fcd4..6e7a0c6b 100644 --- a/src/scheduler/src/plugins/builtin_fcfs.c +++ b/src/scheduler/src/plugins/builtin_fcfs.c @@ -44,6 +44,7 @@ #include #include #include +#include #include "ordq.h" #include "plugin.h" @@ -238,3 +239,312 @@ void builtin_fcfs_fini(void* picker) plugin->data = NULL; } + +/* --------------------------------------------------------------------- + * Task 2.2: state-change hooks (Req 3 AC 2's on_queue_update/on_job_push/ + * on_job_state_change/on_job_update rows; Req 1 AC 2's four state-change + * hook signatures). These are the only hooks that ever populate or mutate + * priv->queues / a queue's pendJobs -- next_job() (task 2.3) only reads + * them. + * ------------------------------------------------------------------- */ + +/* ordq_match_fn_t for fcfs_plugin_data_t.queues: matches by queue name. + * Mirrors plugins/fcfs.c's queue_match exactly (same lookup key, same ordq + * varargs convention). */ +static int queue_match(void* arg1, va_list ap) +{ + fcfs_queue_t* q = (fcfs_queue_t*)arg1; + char* name = va_arg(ap, char*); + + if ((q == NULL) || (name == NULL)) { + return 0; + } + + return (strcmp(name, q->queue->name) == 0); +} + +/* ordq_match_fn_t for a queue's pendJobs: matches by jobid. Mirrors + * plugins/fcfs.c's job_match exactly. */ +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); +} + +/* Looks up (without removing) the fcfs_queue_t already registered for + * `name` in priv->queues -- ordq_peek leaves the ordq_t untouched, so + * callers that only need to read/insert into that queue's own pendJobs + * (on_job_push, on_job_state_change, on_job_update) don't have to pay for + * a remove+reinsert of the queue entry itself. Mirrors plugins/fcfs.c's + * fcfs_get_queue. */ +static fcfs_queue_t* fcfs_find_queue(fcfs_plugin_data_t* priv, char* name) +{ + ordq_item_t* it + = ordq_peek(&(priv->queues), (ordq_match_fn_t)queue_match, name); + + if (it == NULL) { + return NULL; + } + + return (fcfs_queue_t*)it->data; +} + +/* Removes any fcfs_job_t for `jobid` from whichever queue's pendJobs + * currently holds it. A job's fcfs_job_t wrapper lives only inside its + * current queue's own ordq_t -- there is no queue back-pointer on + * fcfs_job_t (task 2.1's struct, unchanged here) -- so on_job_update (the + * only caller that must handle a job moving to a *different* queue) walks + * every registered queue rather than assuming job->queue already names the + * queue the stale entry is sitting in. jobid is globally unique and a job + * is never a pending candidate in more than one queue at once, so at most + * one queue's ordq_del ever actually removes anything. Returns the total + * removed (0 or 1). */ +static int fcfs_remove_job_everywhere(fcfs_plugin_data_t* priv, uint64_t jobid) +{ + ordq_item_t* it = NULL; + int removed = 0; + + for (it = priv->queues.head; it != NULL; it = it->next) { + fcfs_queue_t* fq = (fcfs_queue_t*)it->data; + + removed += ordq_del(&(fq->pendJobs), (ordq_match_fn_t)job_match, jobid); + } + + return removed; +} + +/* + * builtin_fcfs_on_queue_update - xce_job_picker_iface_t.on_queue_update + * (Req 3 AC 2 "on_queue_update" row, Req 1 AC 2 QUEUE_UPDATE event). + * + * Upserts `queue` into priv->queues, keyed by queue->name: + * - unseen name: allocates a new fcfs_queue_t pointing at `queue`, with + * its own empty ordq_t pendJobs (priority desc, queue_time asc -- + * fcfs_job_cmp), then inserts it. + * - known name: reuses the existing fcfs_queue_t (and its pendJobs + * as-is -- Req 3 AC 2 only asks this hook to update ordering, not + * touch candidates), repoints it at the (possibly new) `queue` + * pointer, and reinserts it. + * Either way the queue ends up back in priv->queues via ordq_put, whose + * linear insert re-sorts on fcfs_queue_cmp (descending queue->priority) -- + * this is how "update its priority and reorder" (Req 3 AC 2) is satisfied: + * the fresh insert reads whatever priority `queue` carries right now (the + * core mutates queues in place -- sched_update_existing_queue -- so by the + * time this hook runs queue->priority already holds the new value). + */ +int builtin_fcfs_on_queue_update(void* picker, xce_sched_queue_t* queue) +{ + xce_sched_plugin_t* plugin = (xce_sched_plugin_t*)picker; + fcfs_plugin_data_t* priv = NULL; + ordq_item_t* it = NULL; + fcfs_queue_t* fq = NULL; + int rc = 0; + + if ((plugin == NULL) || (queue == NULL) || (queue->name == NULL)) { + return -EINVAL; + } + + priv = (fcfs_plugin_data_t*)plugin->data; + if (priv == NULL) { + return -EINVAL; + } + + it = ordq_get(&(priv->queues), (ordq_match_fn_t)queue_match, queue->name); + if (it == NULL) { + fq = (fcfs_queue_t*)calloc(1, sizeof(*fq)); + if (fq == NULL) { + return -ENOMEM; + } + + rc = ordq_init(&(fq->pendJobs), (ordq_data_killer_t)fcfs_job_killer, + (ordq_cmp_fn_t)fcfs_job_cmp); + if (rc < 0) { + FREEUP(fq); + return rc; + } + } else { + fq = (fcfs_queue_t*)it->data; + FREEUP(it); + } + + fq->queue = queue; + + return ordq_put(&(priv->queues), (void*)fq); +} + +/* + * builtin_fcfs_on_job_push - xce_job_picker_iface_t.on_job_push (Req 3 AC 2 + * "on_job_push" row, Req 1 AC 2 JOBS_PUSH event). + * + * If `job` is pending (JOB_IS_PEND(job->state)), wraps it in a new + * fcfs_job_t and inserts it into its queue's pendJobs (found via + * job->queue->name in priv->queues), ordered by priority desc, queue_time + * asc (fcfs_job_cmp). Non-pending jobs are not candidates and are silently + * skipped -- Req 3 AC 2's row is conditioned on JOB_STAT_PEND. Skips (does + * not double-insert) a job whose jobid is already present in the target + * queue's pendJobs, so a duplicate JOBS_PUSH delivery for the same job + * cannot create two candidates for it. + */ +int builtin_fcfs_on_job_push(void* picker, xce_sched_job_t* job) +{ + xce_sched_plugin_t* plugin = (xce_sched_plugin_t*)picker; + fcfs_plugin_data_t* priv = NULL; + fcfs_queue_t* fq = NULL; + fcfs_job_t* fj = NULL; + + if ((plugin == NULL) || (job == NULL) || (job->queue == NULL) + || (job->queue->name == NULL)) { + return -EINVAL; + } + + priv = (fcfs_plugin_data_t*)plugin->data; + if (priv == NULL) { + return -EINVAL; + } + + if (!JOB_IS_PEND(job->state)) { + return 0; + } + + fq = fcfs_find_queue(priv, job->queue->name); + if (fq == NULL) { + return -EINVAL; + } + + if (ordq_peek(&(fq->pendJobs), (ordq_match_fn_t)job_match, job->jobid) + != NULL) { + return 0; /* already a candidate -- see header comment */ + } + + fj = (fcfs_job_t*)calloc(1, sizeof(*fj)); + if (fj == NULL) { + return -ENOMEM; + } + + fj->job = job; + + return ordq_put(&(fq->pendJobs), (void*)fj); +} + +/* + * builtin_fcfs_on_job_state_change - xce_job_picker_iface_t. + * on_job_state_change (Req 3 AC 2 "on_job_state_change" row, Req 1 AC 2 + * JOB_STATE_CHANGE event). + * + * `job->queue` does not change as part of a state transition, so unlike + * on_job_update below this only ever needs to touch job->queue's own + * pendJobs, found once via job->queue->name: + * - `state` no longer pending: removes job's fcfs_job_t by jobid (a + * no-op, via ordq_del's "not found" return, if it wasn't a candidate + * already -- e.g. two non-pending transitions in a row). + * - `state` is pending again: re-adds job's fcfs_job_t, unless it is + * already present (idempotent for repeated pending notifications, same + * guard as on_job_push). + */ +int builtin_fcfs_on_job_state_change(void* picker, xce_sched_job_t* job, + uint32_t state) +{ + xce_sched_plugin_t* plugin = (xce_sched_plugin_t*)picker; + fcfs_plugin_data_t* priv = NULL; + fcfs_queue_t* fq = NULL; + + if ((plugin == NULL) || (job == NULL) || (job->queue == NULL) + || (job->queue->name == NULL)) { + return -EINVAL; + } + + priv = (fcfs_plugin_data_t*)plugin->data; + if (priv == NULL) { + return -EINVAL; + } + + fq = fcfs_find_queue(priv, job->queue->name); + if (fq == NULL) { + return -EINVAL; + } + + if (!JOB_IS_PEND(state)) { + ordq_del(&(fq->pendJobs), (ordq_match_fn_t)job_match, job->jobid); + return 0; + } + + if (ordq_peek(&(fq->pendJobs), (ordq_match_fn_t)job_match, job->jobid) + == NULL) { + fcfs_job_t* fj = (fcfs_job_t*)calloc(1, sizeof(*fj)); + + if (fj == NULL) { + return -ENOMEM; + } + + fj->job = job; + + return ordq_put(&(fq->pendJobs), (void*)fj); + } + + return 0; +} + +/* + * builtin_fcfs_on_job_update - xce_job_picker_iface_t.on_job_update (Req 3 + * AC 2 "on_job_update" row, Req 1 AC 2 JOB_UPDATE event). + * + * By the time this hook runs, `job` has already been mutated in place by + * the core (sched_update_existing_job-style field-copy, same convention as + * on_queue_update's `queue` argument) -- there is no "old priority"/"old + * queue" left to diff against, so this unconditionally repositions: + * 1. Removes job's fcfs_job_t from wherever it currently sits (any + * queue -- see fcfs_remove_job_everywhere; this is what makes a + * queue reassignment correct, not just a same-queue priority change). + * 2. If `job` is still pending, re-inserts a fresh fcfs_job_t into its + * *current* job->queue's pendJobs, ordered by fcfs_job_cmp -- this is + * the "correct position per ordering" Req 3 AC 2 asks for. + * A non-pending job is left removed and not reinserted -- pendJobs holds + * only pending candidates (same invariant on_job_push/on_job_state_change + * maintain). + */ +int builtin_fcfs_on_job_update(void* picker, xce_sched_job_t* job) +{ + xce_sched_plugin_t* plugin = (xce_sched_plugin_t*)picker; + fcfs_plugin_data_t* priv = NULL; + fcfs_queue_t* fq = NULL; + fcfs_job_t* fj = NULL; + + if ((plugin == NULL) || (job == NULL)) { + return -EINVAL; + } + + priv = (fcfs_plugin_data_t*)plugin->data; + if (priv == NULL) { + return -EINVAL; + } + + fcfs_remove_job_everywhere(priv, job->jobid); + + if (!JOB_IS_PEND(job->state)) { + return 0; + } + + if ((job->queue == NULL) || (job->queue->name == NULL)) { + return -EINVAL; + } + + fq = fcfs_find_queue(priv, job->queue->name); + if (fq == NULL) { + return -EINVAL; + } + + fj = (fcfs_job_t*)calloc(1, sizeof(*fj)); + if (fj == NULL) { + return -ENOMEM; + } + + fj->job = job; + + return ordq_put(&(fq->pendJobs), (void*)fj); +} diff --git a/src/scheduler/testsuite/test_picker_fcfs.c b/src/scheduler/testsuite/test_picker_fcfs.c index 7fa70663..aafdd76e 100644 --- a/src/scheduler/testsuite/test_picker_fcfs.c +++ b/src/scheduler/testsuite/test_picker_fcfs.c @@ -209,6 +209,527 @@ static void test_builtin_fcfs_fini_releases_populated_queues_and_jobs(void) TEST_ASSERT_NULL(plugin.data); } +/* ---------------------------------------------------------------------- */ +/* Task 2.2: state-change hooks (Req 3 AC 2's on_queue_update/on_job_push/ + * on_job_state_change/on_job_update rows). Every test below drives the + * hooks directly against a real init()'d picker and then inspects the + * underlying ordq_t structures (priv->queues, and each fcfs_queue_t's + * pendJobs) by walking ->head chains -- next_job() doesn't exist yet + * (task 2.3), so this is the only way to observe candidate-set order this + * task can prove. + * RED: before task 2.2, builtin_fcfs_on_queue_update/on_job_push/ + * on_job_state_change/on_job_update do not exist -- this file + * fails to compile (implicit declaration / undefined symbols). + * GREEN: the hooks upsert/insert/remove/reposition exactly as Req 3 AC 2 + * specifies, verified by walking the resulting ordq_t chains. + * ---------------------------------------------------------------------- */ + +/* Walks an ordq_t of fcfs_queue_t entries and collects queue names in list + * order, so a test can assert on priv->queues' actual ordering (descending + * queue->priority, fcfs_queue_cmp) without depending on next_job(). */ +static size_t collect_queue_names(ordq_t* q, const char** out, size_t max) +{ + ordq_item_t* it = q->head; + size_t n = 0; + + while ((it != NULL) && (n < max)) { + fcfs_queue_t* fq = (fcfs_queue_t*)it->data; + + out[n++] = fq->queue->name; + it = it->next; + } + + return n; +} + +/* Walks an ordq_t of fcfs_job_t entries and collects jobids in list order, + * so a test can assert on a queue's pendJobs actual ordering (priority + * desc, queue_time asc, fcfs_job_cmp). */ +static size_t collect_jobids(ordq_t* q, uint64_t* out, size_t max) +{ + ordq_item_t* it = q->head; + size_t n = 0; + + while ((it != NULL) && (n < max)) { + fcfs_job_t* fj = (fcfs_job_t*)it->data; + + out[n++] = fj->job->jobid; + it = it->next; + } + + return n; +} + +/* Req 3 AC 2 "on_queue_update" row, unseen-name branch: allocates a new + * fcfs_queue_t pointing at `queue`, with an empty pendJobs initialized on + * fcfs_job_cmp/fcfs_job_killer -- ready for a later on_job_push, task 2.1's + * own comparators reused rather than re-implemented. */ +static void test_on_queue_update_inserts_new_queue_with_empty_pendjobs(void) +{ + xce_sched_plugin_t plugin; + fcfs_plugin_data_t* priv; + xce_sched_queue_t q1 = { .name = "q1", .priority = 5 }; + fcfs_queue_t* fq; + + fixture_init(&plugin); + TEST_ASSERT_EQUAL_INT(0, builtin_fcfs_init(&plugin, NULL)); + priv = (fcfs_plugin_data_t*)plugin.data; + + TEST_ASSERT_EQUAL_INT(0, builtin_fcfs_on_queue_update(&plugin, &q1)); + + TEST_ASSERT_NOT_NULL(priv->queues.head); + fq = (fcfs_queue_t*)priv->queues.head->data; + TEST_ASSERT_EQUAL_PTR(&q1, fq->queue); + TEST_ASSERT_NULL(fq->pendJobs.head); + TEST_ASSERT_EQUAL_PTR(fcfs_job_cmp, fq->pendJobs.cmp); + TEST_ASSERT_EQUAL_PTR(fcfs_job_killer, fq->pendJobs.killer); + + builtin_fcfs_fini(&plugin); +} + +/* Req 3 AC 2 "on_queue_update" row + Req 3 AC 4 "inter-queue by descending + * queue->priority": three queues pushed out of priority order must come + * back out of priv->queues sorted highest-priority-first. */ +static void test_on_queue_update_orders_queues_by_priority_desc(void) +{ + xce_sched_plugin_t plugin; + xce_sched_queue_t qlow = { .name = "low", .priority = 1 }; + xce_sched_queue_t qhigh = { .name = "high", .priority = 9 }; + xce_sched_queue_t qmid = { .name = "mid", .priority = 5 }; + const char* names[3]; + + fixture_init(&plugin); + TEST_ASSERT_EQUAL_INT(0, builtin_fcfs_init(&plugin, NULL)); + + TEST_ASSERT_EQUAL_INT(0, builtin_fcfs_on_queue_update(&plugin, &qlow)); + TEST_ASSERT_EQUAL_INT(0, builtin_fcfs_on_queue_update(&plugin, &qhigh)); + TEST_ASSERT_EQUAL_INT(0, builtin_fcfs_on_queue_update(&plugin, &qmid)); + + TEST_ASSERT_EQUAL_UINT( + 3, collect_queue_names(&(((fcfs_plugin_data_t*)plugin.data)->queues), + names, 3)); + TEST_ASSERT_EQUAL_STRING("high", names[0]); + TEST_ASSERT_EQUAL_STRING("mid", names[1]); + TEST_ASSERT_EQUAL_STRING("low", names[2]); + + builtin_fcfs_fini(&plugin); +} + +/* Req 3 AC 2 "on_queue_update" row, known-name branch: "If queue exists, + * update its priority and reorder." Raising q1's priority above q2's (by + * mutating the same xce_sched_queue_t in place, the way + * sched_update_existing_queue does) and re-calling on_queue_update must + * move q1 ahead of q2 -- without allocating a second fcfs_queue_t for + * "q1" (checked by identity: the head entry's pointer is still &q1). */ +static void +test_on_queue_update_reorders_existing_queue_on_priority_change(void) +{ + xce_sched_plugin_t plugin; + fcfs_plugin_data_t* priv; + xce_sched_queue_t q1 = { .name = "q1", .priority = 1 }; + xce_sched_queue_t q2 = { .name = "q2", .priority = 5 }; + const char* names[2]; + + fixture_init(&plugin); + TEST_ASSERT_EQUAL_INT(0, builtin_fcfs_init(&plugin, NULL)); + priv = (fcfs_plugin_data_t*)plugin.data; + + TEST_ASSERT_EQUAL_INT(0, builtin_fcfs_on_queue_update(&plugin, &q1)); + TEST_ASSERT_EQUAL_INT(0, builtin_fcfs_on_queue_update(&plugin, &q2)); + TEST_ASSERT_EQUAL_UINT(2, collect_queue_names(&(priv->queues), names, 2)); + TEST_ASSERT_EQUAL_STRING("q2", names[0]); /* q2 (5) ahead of q1 (1) */ + + q1.priority = 9; /* in-place mutation, mirrors the live core's update */ + TEST_ASSERT_EQUAL_INT(0, builtin_fcfs_on_queue_update(&plugin, &q1)); + + TEST_ASSERT_EQUAL_UINT(2, collect_queue_names(&(priv->queues), names, 2)); + TEST_ASSERT_EQUAL_STRING("q1", names[0]); /* now q1 (9) ahead of q2 (5) */ + TEST_ASSERT_EQUAL_STRING("q2", names[1]); + + builtin_fcfs_fini(&plugin); +} + +/* Defensive-argument contract: NULL picker, NULL queue, an unnamed queue, + * and a picker that was never init()'d must all report -EINVAL rather than + * crash -- same convention as builtin_fcfs_init's own NULL check. */ +static void test_on_queue_update_rejects_invalid_args(void) +{ + xce_sched_plugin_t plugin; + xce_sched_queue_t q = { .name = "q1", .priority = 1 }; + xce_sched_queue_t noname = { .name = NULL, .priority = 1 }; + + fixture_init(&plugin); /* plugin.data == NULL, never init()'d */ + + TEST_ASSERT_EQUAL_INT(-EINVAL, builtin_fcfs_on_queue_update(NULL, &q)); + TEST_ASSERT_EQUAL_INT(-EINVAL, builtin_fcfs_on_queue_update(&plugin, NULL)); + TEST_ASSERT_EQUAL_INT(-EINVAL, + builtin_fcfs_on_queue_update(&plugin, &noname)); + TEST_ASSERT_EQUAL_INT(-EINVAL, builtin_fcfs_on_queue_update(&plugin, &q)); +} + +/* Req 3 AC 2 "on_job_push" row: a pending job is wrapped and inserted into + * its own queue's pendJobs (found via job->queue->name). */ +static void test_on_job_push_adds_pending_job_to_its_queue(void) +{ + xce_sched_plugin_t plugin; + fcfs_plugin_data_t* priv; + xce_sched_queue_t q1 = { .name = "q1", .priority = 1 }; + xce_sched_job_t j1 = { .jobid = 42, + .priority = 3, + .queue_time = 100, + .state = JOB_STAT_PEND, + .queue = &q1 }; + fcfs_queue_t* fq; + uint64_t jobids[1]; + + fixture_init(&plugin); + TEST_ASSERT_EQUAL_INT(0, builtin_fcfs_init(&plugin, NULL)); + priv = (fcfs_plugin_data_t*)plugin.data; + TEST_ASSERT_EQUAL_INT(0, builtin_fcfs_on_queue_update(&plugin, &q1)); + + TEST_ASSERT_EQUAL_INT(0, builtin_fcfs_on_job_push(&plugin, &j1)); + + fq = (fcfs_queue_t*)priv->queues.head->data; + TEST_ASSERT_EQUAL_UINT(1, collect_jobids(&(fq->pendJobs), jobids, 1)); + TEST_ASSERT_EQUAL_UINT64(42, jobids[0]); + + builtin_fcfs_fini(&plugin); +} + +/* Req 3 AC 2 "on_job_push" row + Req 3 AC 4 "intra-queue by priority desc, + * queue_time asc": three pending jobs pushed out of order come back sorted + * by priority desc, then (for the priority tie) queue_time asc. */ +static void test_on_job_push_orders_jobs_priority_desc_then_queue_time_asc(void) +{ + xce_sched_plugin_t plugin; + fcfs_plugin_data_t* priv; + xce_sched_queue_t q1 = { .name = "q1", .priority = 1 }; + /* tie on priority (5): j_tie_late (queue_time 200) must sort AFTER + * j_tie_early (queue_time 100) despite being pushed first. */ + xce_sched_job_t j_tie_late = { .jobid = 1, + .priority = 5, + .queue_time = 200, + .state = JOB_STAT_PEND, + .queue = &q1 }; + xce_sched_job_t j_low = { .jobid = 2, + .priority = 1, + .queue_time = 50, + .state = JOB_STAT_PEND, + .queue = &q1 }; + xce_sched_job_t j_tie_early = { .jobid = 3, + .priority = 5, + .queue_time = 100, + .state = JOB_STAT_PEND, + .queue = &q1 }; + fcfs_queue_t* fq; + uint64_t jobids[3]; + + fixture_init(&plugin); + TEST_ASSERT_EQUAL_INT(0, builtin_fcfs_init(&plugin, NULL)); + priv = (fcfs_plugin_data_t*)plugin.data; + TEST_ASSERT_EQUAL_INT(0, builtin_fcfs_on_queue_update(&plugin, &q1)); + + TEST_ASSERT_EQUAL_INT(0, builtin_fcfs_on_job_push(&plugin, &j_tie_late)); + TEST_ASSERT_EQUAL_INT(0, builtin_fcfs_on_job_push(&plugin, &j_low)); + TEST_ASSERT_EQUAL_INT(0, builtin_fcfs_on_job_push(&plugin, &j_tie_early)); + + fq = (fcfs_queue_t*)priv->queues.head->data; + TEST_ASSERT_EQUAL_UINT(3, collect_jobids(&(fq->pendJobs), jobids, 3)); + TEST_ASSERT_EQUAL_UINT64(3, jobids[0]); /* j_tie_early: prio 5, t=100 */ + TEST_ASSERT_EQUAL_UINT64(1, jobids[1]); /* j_tie_late: prio 5, t=200 */ + TEST_ASSERT_EQUAL_UINT64(2, jobids[2]); /* j_low: prio 1 */ + + builtin_fcfs_fini(&plugin); +} + +/* Req 3 AC 2 "on_job_push" row's JOB_STAT_PEND guard: a non-pending job + * (state 0, no JOB_STAT_PEND bit) must not be added to any candidate set. */ +static void test_on_job_push_skips_non_pending_job(void) +{ + xce_sched_plugin_t plugin; + fcfs_plugin_data_t* priv; + xce_sched_queue_t q1 = { .name = "q1", .priority = 1 }; + xce_sched_job_t j1 = { .jobid = 1, + .priority = 3, + .queue_time = 100, + .state = 0, /* not pending */ + .queue = &q1 }; + fcfs_queue_t* fq; + + fixture_init(&plugin); + TEST_ASSERT_EQUAL_INT(0, builtin_fcfs_init(&plugin, NULL)); + priv = (fcfs_plugin_data_t*)plugin.data; + TEST_ASSERT_EQUAL_INT(0, builtin_fcfs_on_queue_update(&plugin, &q1)); + + TEST_ASSERT_EQUAL_INT(0, builtin_fcfs_on_job_push(&plugin, &j1)); + + fq = (fcfs_queue_t*)priv->queues.head->data; + TEST_ASSERT_NULL(fq->pendJobs.head); + + builtin_fcfs_fini(&plugin); +} + +/* A pending job whose queue was never registered via on_queue_update has + * nowhere to go -- -EINVAL, not a crash or a silent drop into the wrong + * queue. */ +static void test_on_job_push_rejects_unknown_queue(void) +{ + xce_sched_plugin_t plugin; + xce_sched_queue_t unregistered = { .name = "ghost", .priority = 1 }; + xce_sched_job_t j1 = { .jobid = 1, + .priority = 3, + .queue_time = 100, + .state = JOB_STAT_PEND, + .queue = &unregistered }; + + fixture_init(&plugin); + TEST_ASSERT_EQUAL_INT(0, builtin_fcfs_init(&plugin, NULL)); + + TEST_ASSERT_EQUAL_INT(-EINVAL, builtin_fcfs_on_job_push(&plugin, &j1)); + + builtin_fcfs_fini(&plugin); +} + +/* A duplicate JOBS_PUSH delivery for the same still-pending job must not + * create a second candidate entry. */ +static void test_on_job_push_ignores_duplicate_push_of_same_job(void) +{ + xce_sched_plugin_t plugin; + fcfs_plugin_data_t* priv; + xce_sched_queue_t q1 = { .name = "q1", .priority = 1 }; + xce_sched_job_t j1 = { .jobid = 7, + .priority = 3, + .queue_time = 100, + .state = JOB_STAT_PEND, + .queue = &q1 }; + fcfs_queue_t* fq; + uint64_t jobids[2]; + + fixture_init(&plugin); + TEST_ASSERT_EQUAL_INT(0, builtin_fcfs_init(&plugin, NULL)); + priv = (fcfs_plugin_data_t*)plugin.data; + TEST_ASSERT_EQUAL_INT(0, builtin_fcfs_on_queue_update(&plugin, &q1)); + + TEST_ASSERT_EQUAL_INT(0, builtin_fcfs_on_job_push(&plugin, &j1)); + TEST_ASSERT_EQUAL_INT(0, builtin_fcfs_on_job_push(&plugin, &j1)); + + fq = (fcfs_queue_t*)priv->queues.head->data; + TEST_ASSERT_EQUAL_UINT(1, collect_jobids(&(fq->pendJobs), jobids, 2)); + + builtin_fcfs_fini(&plugin); +} + +/* Req 3 AC 2 "on_job_state_change" row: a job leaving pending state + * (EXIT/FAIL/DONE) is removed from its queue's candidate set. */ +static void test_on_job_state_change_removes_job_when_leaving_pending(void) +{ + xce_sched_plugin_t plugin; + fcfs_plugin_data_t* priv; + xce_sched_queue_t q1 = { .name = "q1", .priority = 1 }; + xce_sched_job_t j1 = { .jobid = 1, + .priority = 3, + .queue_time = 100, + .state = JOB_STAT_PEND, + .queue = &q1 }; + fcfs_queue_t* fq; + + fixture_init(&plugin); + TEST_ASSERT_EQUAL_INT(0, builtin_fcfs_init(&plugin, NULL)); + priv = (fcfs_plugin_data_t*)plugin.data; + TEST_ASSERT_EQUAL_INT(0, builtin_fcfs_on_queue_update(&plugin, &q1)); + TEST_ASSERT_EQUAL_INT(0, builtin_fcfs_on_job_push(&plugin, &j1)); + + j1.state = JOB_STAT_DONE; + TEST_ASSERT_EQUAL_INT( + 0, builtin_fcfs_on_job_state_change(&plugin, &j1, JOB_STAT_DONE)); + + fq = (fcfs_queue_t*)priv->queues.head->data; + TEST_ASSERT_NULL(fq->pendJobs.head); + + builtin_fcfs_fini(&plugin); +} + +/* Req 3 AC 2 "on_job_state_change" row: a job returning to PEND is re-added + * to its queue's candidate set, in its correctly ordered position among + * jobs already there. */ +static void +test_on_job_state_change_readds_job_in_order_when_returning_to_pending(void) +{ + xce_sched_plugin_t plugin; + fcfs_plugin_data_t* priv; + xce_sched_queue_t q1 = { .name = "q1", .priority = 1 }; + xce_sched_job_t jlow = { .jobid = 1, + .priority = 1, + .queue_time = 100, + .state = JOB_STAT_PEND, + .queue = &q1 }; + xce_sched_job_t jresurrected = { .jobid = 2, + .priority = 9, + .queue_time = 50, + .state = JOB_STAT_DONE, + .queue = &q1 }; + fcfs_queue_t* fq; + uint64_t jobids[2]; + + fixture_init(&plugin); + TEST_ASSERT_EQUAL_INT(0, builtin_fcfs_init(&plugin, NULL)); + priv = (fcfs_plugin_data_t*)plugin.data; + TEST_ASSERT_EQUAL_INT(0, builtin_fcfs_on_queue_update(&plugin, &q1)); + TEST_ASSERT_EQUAL_INT(0, builtin_fcfs_on_job_push(&plugin, &jlow)); + /* jresurrected starts DONE -- never pushed, not a candidate yet. */ + + jresurrected.state = JOB_STAT_PEND; + TEST_ASSERT_EQUAL_INT(0, builtin_fcfs_on_job_state_change( + &plugin, &jresurrected, JOB_STAT_PEND)); + + fq = (fcfs_queue_t*)priv->queues.head->data; + TEST_ASSERT_EQUAL_UINT(2, collect_jobids(&(fq->pendJobs), jobids, 2)); + TEST_ASSERT_EQUAL_UINT64(2, jobids[0]); /* higher priority first */ + TEST_ASSERT_EQUAL_UINT64(1, jobids[1]); + + builtin_fcfs_fini(&plugin); +} + +/* Idempotency: on_job_state_change(..., PEND) called twice in a row for a + * job that is already a candidate must not create a duplicate entry. */ +static void test_on_job_state_change_is_idempotent_when_already_pending(void) +{ + xce_sched_plugin_t plugin; + fcfs_plugin_data_t* priv; + xce_sched_queue_t q1 = { .name = "q1", .priority = 1 }; + xce_sched_job_t j1 = { .jobid = 1, + .priority = 3, + .queue_time = 100, + .state = JOB_STAT_PEND, + .queue = &q1 }; + fcfs_queue_t* fq; + uint64_t jobids[2]; + + fixture_init(&plugin); + TEST_ASSERT_EQUAL_INT(0, builtin_fcfs_init(&plugin, NULL)); + priv = (fcfs_plugin_data_t*)plugin.data; + TEST_ASSERT_EQUAL_INT(0, builtin_fcfs_on_queue_update(&plugin, &q1)); + TEST_ASSERT_EQUAL_INT(0, builtin_fcfs_on_job_push(&plugin, &j1)); + + TEST_ASSERT_EQUAL_INT( + 0, builtin_fcfs_on_job_state_change(&plugin, &j1, JOB_STAT_PEND)); + + fq = (fcfs_queue_t*)priv->queues.head->data; + TEST_ASSERT_EQUAL_UINT(1, collect_jobids(&(fq->pendJobs), jobids, 2)); + + builtin_fcfs_fini(&plugin); +} + +/* Req 3 AC 2 "on_job_update" row, priority-change case: raising j1's + * priority above j2's (mutated in place, mirroring the live core) and + * calling on_job_update must move j1 ahead of j2 within their shared + * queue. */ +static void +test_on_job_update_repositions_job_within_same_queue_on_priority_change(void) +{ + xce_sched_plugin_t plugin; + fcfs_plugin_data_t* priv; + xce_sched_queue_t q1 = { .name = "q1", .priority = 1 }; + xce_sched_job_t j1 = { .jobid = 1, + .priority = 1, + .queue_time = 100, + .state = JOB_STAT_PEND, + .queue = &q1 }; + xce_sched_job_t j2 = { .jobid = 2, + .priority = 5, + .queue_time = 100, + .state = JOB_STAT_PEND, + .queue = &q1 }; + fcfs_queue_t* fq; + uint64_t jobids[2]; + + fixture_init(&plugin); + TEST_ASSERT_EQUAL_INT(0, builtin_fcfs_init(&plugin, NULL)); + priv = (fcfs_plugin_data_t*)plugin.data; + TEST_ASSERT_EQUAL_INT(0, builtin_fcfs_on_queue_update(&plugin, &q1)); + TEST_ASSERT_EQUAL_INT(0, builtin_fcfs_on_job_push(&plugin, &j1)); + TEST_ASSERT_EQUAL_INT(0, builtin_fcfs_on_job_push(&plugin, &j2)); + + fq = (fcfs_queue_t*)priv->queues.head->data; + TEST_ASSERT_EQUAL_UINT(2, collect_jobids(&(fq->pendJobs), jobids, 2)); + TEST_ASSERT_EQUAL_UINT64(2, jobids[0]); /* j2 (5) ahead of j1 (1) */ + + j1.priority = 9; /* in-place mutation, mirrors sched_update_existing_job */ + TEST_ASSERT_EQUAL_INT(0, builtin_fcfs_on_job_update(&plugin, &j1)); + + TEST_ASSERT_EQUAL_UINT(2, collect_jobids(&(fq->pendJobs), jobids, 2)); + TEST_ASSERT_EQUAL_UINT64(1, jobids[0]); /* now j1 (9) ahead of j2 (5) */ + TEST_ASSERT_EQUAL_UINT64(2, jobids[1]); + + builtin_fcfs_fini(&plugin); +} + +/* Req 3 AC 2 "on_job_update" row, queue-reassignment case: moving a job's + * `queue` field to a different registered queue (mutated in place) and + * calling on_job_update must remove it from the old queue's pendJobs and + * insert it into the new queue's pendJobs. */ +static void +test_on_job_update_moves_job_to_new_queue_on_queue_reassignment(void) +{ + xce_sched_plugin_t plugin; + fcfs_plugin_data_t* priv; + xce_sched_queue_t qa = { .name = "qa", .priority = 1 }; + xce_sched_queue_t qb = { .name = "qb", .priority = 1 }; + xce_sched_job_t j1 = { .jobid = 1, + .priority = 3, + .queue_time = 100, + .state = JOB_STAT_PEND, + .queue = &qa }; + fcfs_queue_t* fqa; + fcfs_queue_t* fqb; + uint64_t jobids[1]; + + fixture_init(&plugin); + TEST_ASSERT_EQUAL_INT(0, builtin_fcfs_init(&plugin, NULL)); + priv = (fcfs_plugin_data_t*)plugin.data; + TEST_ASSERT_EQUAL_INT(0, builtin_fcfs_on_queue_update(&plugin, &qa)); + TEST_ASSERT_EQUAL_INT(0, builtin_fcfs_on_queue_update(&plugin, &qb)); + TEST_ASSERT_EQUAL_INT(0, builtin_fcfs_on_job_push(&plugin, &j1)); + + j1.queue = &qb; /* in-place reassignment */ + TEST_ASSERT_EQUAL_INT(0, builtin_fcfs_on_job_update(&plugin, &j1)); + + fqa = fcfs_find_queue(priv, "qa"); + fqb = fcfs_find_queue(priv, "qb"); + TEST_ASSERT_NULL(fqa->pendJobs.head); /* gone from the old queue */ + TEST_ASSERT_EQUAL_UINT(1, collect_jobids(&(fqb->pendJobs), jobids, 1)); + TEST_ASSERT_EQUAL_UINT64(1, jobids[0]); /* present in the new queue */ + + builtin_fcfs_fini(&plugin); +} + +/* on_job_update on a job that is not (or no longer) pending must not add + * it to any queue's candidate set -- pendJobs holds only pending jobs. */ +static void test_on_job_update_is_noop_for_non_pending_job(void) +{ + xce_sched_plugin_t plugin; + fcfs_plugin_data_t* priv; + xce_sched_queue_t q1 = { .name = "q1", .priority = 1 }; + xce_sched_job_t j1 = { .jobid = 1, + .priority = 3, + .queue_time = 100, + .state = JOB_STAT_RUN, /* not pending */ + .queue = &q1 }; + fcfs_queue_t* fq; + + fixture_init(&plugin); + TEST_ASSERT_EQUAL_INT(0, builtin_fcfs_init(&plugin, NULL)); + priv = (fcfs_plugin_data_t*)plugin.data; + TEST_ASSERT_EQUAL_INT(0, builtin_fcfs_on_queue_update(&plugin, &q1)); + + TEST_ASSERT_EQUAL_INT(0, builtin_fcfs_on_job_update(&plugin, &j1)); + + fq = (fcfs_queue_t*)priv->queues.head->data; + TEST_ASSERT_NULL(fq->pendJobs.head); + + builtin_fcfs_fini(&plugin); +} + int main(void) { UNITY_BEGIN(); @@ -218,5 +739,27 @@ int main(void) RUN_TEST(test_builtin_fcfs_fini_releases_empty_state); RUN_TEST(test_builtin_fcfs_fini_handles_null_and_unset_picker); RUN_TEST(test_builtin_fcfs_fini_releases_populated_queues_and_jobs); + + RUN_TEST(test_on_queue_update_inserts_new_queue_with_empty_pendjobs); + RUN_TEST(test_on_queue_update_orders_queues_by_priority_desc); + RUN_TEST(test_on_queue_update_reorders_existing_queue_on_priority_change); + RUN_TEST(test_on_queue_update_rejects_invalid_args); + + RUN_TEST(test_on_job_push_adds_pending_job_to_its_queue); + RUN_TEST(test_on_job_push_orders_jobs_priority_desc_then_queue_time_asc); + RUN_TEST(test_on_job_push_skips_non_pending_job); + RUN_TEST(test_on_job_push_rejects_unknown_queue); + RUN_TEST(test_on_job_push_ignores_duplicate_push_of_same_job); + + RUN_TEST(test_on_job_state_change_removes_job_when_leaving_pending); + RUN_TEST( + test_on_job_state_change_readds_job_in_order_when_returning_to_pending); + RUN_TEST(test_on_job_state_change_is_idempotent_when_already_pending); + + RUN_TEST( + test_on_job_update_repositions_job_within_same_queue_on_priority_change); + RUN_TEST(test_on_job_update_moves_job_to_new_queue_on_queue_reassignment); + RUN_TEST(test_on_job_update_is_noop_for_non_pending_job); + return UNITY_END(); }