diff --git a/src/scheduler/src/core_events.c b/src/scheduler/src/core_events.c index ab958d89..6eebaf08 100644 --- a/src/scheduler/src/core_events.c +++ b/src/scheduler/src/core_events.c @@ -1363,9 +1363,12 @@ int xce_core_push_events(xce_sched_core_t* core, xce_core_event_t* events, xce_sched_queue_t* q = sched_get_queue(core, updated->submit->queue); + /* bug-1944: only the queue changes; job->priority is + * the job's own (0 at ingest, core_build_job), so a + * moved job does not overtake the target queue's jobs. + * Queue priority already orders queues in the picker. */ if (q != NULL) { - job->queue = q; - job->priority = q->priority; + job->queue = q; } } diff --git a/src/scheduler/src/schedobj.c b/src/scheduler/src/schedobj.c index 232947b2..6b05a2fe 100644 --- a/src/scheduler/src/schedobj.c +++ b/src/scheduler/src/schedobj.c @@ -673,15 +673,15 @@ int sched_update_existing_job(xce_scheduler_t* sched, xce_job_t* updated) /* mutable submit-derived fields */ if (updated->submit != NULL) { - /* queue — re-resolve the queue pointer by name. xce_job_t carries no - * per-job priority, so job priority follows the (possibly new) queue, - * consistent with the queue-derived priority model. */ + /* queue — re-resolve the queue pointer by name. job->priority is + * the job's own (0 at make_sched_job) and stays as is: copying the + * queue's priority onto a moved job made it overtake the target + * queue's earlier jobs (bug-1944). */ if (updated->submit->queue != NULL) { xce_sched_queue_t* q = sched_get_queue(&(sched->core), updated->submit->queue); if (q != NULL) { - job->queue = q; - job->priority = q->priority; + job->queue = q; } } diff --git a/src/scheduler/testsuite/meson.build b/src/scheduler/testsuite/meson.build index a9aa7d4b..10cfaf6f 100644 --- a/src/scheduler/testsuite/meson.build +++ b/src/scheduler/testsuite/meson.build @@ -3113,3 +3113,24 @@ sched_sim_group_limit_parity_test = executable( test('scheduler sim group limit parity test', sched_sim_group_limit_parity_test, suite: 'scheduler', env: ['ASAN_OPTIONS=detect_leaks=0']) + +# bug-1944: a job moved to another queue (core JOB_UPDATE event) keeps its +# own priority and does not overtake jobs already pending in the target +# queue. Same real-scheduler_srcs link technique as above. +sched_job_move_priority_test_srcs = ['test_job_move_priority.c'] +foreach f : scheduler_srcs + if f != 'src/main.c' + sched_job_move_priority_test_srcs += ['../' + f] + endif +endforeach + +sched_job_move_priority_test = executable( + 'sched_job_move_priority_test', + sched_job_move_priority_test_srcs, + include_directories:[incdir], + dependencies: [scheduler_test_deps], + install: false) + +test('scheduler job move priority test', + sched_job_move_priority_test, suite: 'scheduler', + env: ['ASAN_OPTIONS=detect_leaks=0']) diff --git a/src/scheduler/testsuite/test_job_move_priority.c b/src/scheduler/testsuite/test_job_move_priority.c new file mode 100644 index 00000000..8a891d9b --- /dev/null +++ b/src/scheduler/testsuite/test_job_move_priority.c @@ -0,0 +1,227 @@ +/* + * test_job_move_priority.c - bug-1944: moving a pending job to another + * queue must not change the job's own priority, so the moved job does not + * overtake jobs submitted earlier to the target queue. + * + * Fixture: queue qa (priority 0) and queue qb (priority 10); jobs B1 and B2 + * pending in qb (queue_time 1000, 1001), job M pending in qa (1002). M is + * moved to qb through the core's JOB_UPDATE event (core_events.c), the path + * both drivers use for a job-fields update. One worker slot, so only the + * first job the picker offers runs. FCFS orders a queue's jobs by job + * priority, then queue time: B1 must run and M must stay pending with its + * submit-time priority (0). Built like test_sim_contention_limit_driver_ + * parity.c's normal-mode fixture (real scheduler_srcs, real picker and + * allocator threads, `static` fixture, exactly one RUN_TEST()). + */ +#include +#include +#include +#include +#include + +#include + +#include "controller.capnp.h" +#include "decision.h" +#include "plugin.h" +#include "preason.h" +#include "resobj.h" +#include "sched/types.h" +#include "scheduler.h" +#include "xstring.h" + +#define WORKER_NAME "jmove_worker" +#define QUEUE_A "jmove_qa" +#define QUEUE_B "jmove_qb" +#define TEST_RESREQ ((char*)"select[type==any] rusage[mem=1]") + +#define JOB_B1_ID ((uint64_t)194401ULL) +#define JOB_B2_ID ((uint64_t)194402ULL) +#define JOB_M_ID ((uint64_t)194403ULL) + +void setUp(void) { return; } +void tearDown(void) { return; } + +static void register_slots_resource(var resmgr) +{ + xce_resource_t res; + + memset(&res, 0, sizeof(res)); + res.name = strdup("slots"); + res.valtype = XCE_VAL_NUMERIC; + res.ordtype = XCE_ORD_DECR; + res.flags = XCE_RESF_LIMIT | XCE_RESF_CONSUMABLE | XCE_RESF_RELEASE + | XCE_RESF_DEFAULT_HOST; + + TEST_ASSERT_NOT_NULL(makeResource(resmgr, &res)); +} + +static void add_worker(xce_sched_core_t* core, double slots) +{ + xce_sched_worker_t* w = (xce_sched_worker_t*)calloc(1, sizeof(*w)); + xce_resource_t* r = NULL; + xce_val_t* val = NULL; + int rc; + + TEST_ASSERT_NOT_NULL(w); + w->name = strdup(WORKER_NAME); + w->idx = -1; + zarray_init(&(w->resvals), 128, (zarray_item_killer_t)kill_sched_resval); + zarray_init(&(w->resInUse), 128, (zarray_item_killer_t)kill_sched_resval); + + rc = key_find(core->resmgr, (void*)"slots", (void**)&r); + TEST_ASSERT_TRUE(rc >= 0); + val = (xce_val_t*)calloc(1, sizeof(*val)); + TEST_ASSERT_NOT_NULL(val); + val->which = Value_u_fvalue; + val->u.fvalue = slots; + zarray_put(&(w->resvals), r->idx, val); + + rc = sched_add_worker(core, w); + TEST_ASSERT_TRUE(rc >= 0); +} + +static void add_queue(xce_sched_core_t* core, xce_sched_plugin_t* picker, + const char* name, int priority) +{ + xce_sched_queue_t* q = (xce_sched_queue_t*)calloc(1, sizeof(*q)); + int rc; + + TEST_ASSERT_NOT_NULL(q); + q->name = makeString((char*)name); + q->priority = priority; + rc = sched_add_queue(core, q); + TEST_ASSERT_TRUE(rc >= 0); + picker->picker_iface->on_queue_update(picker, q); +} + +static void push_job(xce_scheduler_t* sched, xce_sched_plugin_t* picker, + uint64_t jobid, const char* queue, uint64_t queue_time) +{ + job_submit_data_t submit; + xce_job_t j; + xce_sched_job_t* job; + int rc; + + memset(&submit, 0, sizeof(submit)); + submit.queue = (char*)queue; + submit.user = (char*)"jmove_user"; + submit.project_name = (char*)"default"; + submit.resreq = TEST_RESREQ; + submit.mintasks = 1; + submit.maxtasks = 1; + + memset(&j, 0, sizeof(j)); + j.jobid = jobid; + j.state = JOB_STAT_PEND; + j.sub_time = queue_time; + j.queue_time = queue_time; + j.submit = &submit; + + job = make_sched_job(sched, &j); + TEST_ASSERT_NOT_NULL(job); + job->schedobj->u.primary->resreq->minTasks = 1; + job->schedobj->u.primary->resreq->maxTasks = 1; + rc = sched_add_job(&(sched->core), job); + TEST_ASSERT_TRUE(rc >= 0); + picker->picker_iface->on_job_push(picker, job); +} + +static void run_round(xce_sched_core_t* core) +{ + pthread_t picker_tid; + pthread_t allocator_tid; + xce_decision_t* d; + int rc; + + rc = pthread_create(&picker_tid, NULL, sched_job_picker, core); + TEST_ASSERT_EQUAL_INT(0, rc); + rc = pthread_create(&allocator_tid, NULL, sched_allocator, core); + TEST_ASSERT_EQUAL_INT(0, rc); + + rc = xce_core_go(core); + TEST_ASSERT_EQUAL_INT(0, rc); + + do { + d = NULL; + cbuf_get(&(core->decisions), &d); + } while (d != NULL); + + rc = xce_core_end_session(core); + TEST_ASSERT_EQUAL_INT(0, rc); +} + +static xce_scheduler_t g_sched; + +static void test_moved_job_keeps_its_priority_and_place(void) +{ + xce_sched_plugin_t* fcfs = NULL; + job_submit_data_t move_submit; + xce_job_t move; + xce_core_event_t ev; + xce_sched_job_t* m; + int rc; + + memset(&g_sched, 0, sizeof(g_sched)); + rc = xce_sched_core_init(&(g_sched.core)); + TEST_ASSERT_EQUAL_INT(0, rc); + register_slots_resource(g_sched.core.resmgr); + + fcfs = builtin_fcfs_new_plugin(); + TEST_ASSERT_NOT_NULL(fcfs); + rc = plugin_register(&g_sched, fcfs); + TEST_ASSERT_TRUE(rc >= 0); + g_sched.core.job_pick_plugin = (char*)"builtin_fcfs"; + rc = register_seqjob_allocator(&g_sched); + TEST_ASSERT_TRUE(rc >= 0); + rc = register_parjob_allocator(&g_sched); + TEST_ASSERT_TRUE(rc >= 0); + + add_queue(&(g_sched.core), fcfs, QUEUE_A, 0); + add_queue(&(g_sched.core), fcfs, QUEUE_B, 10); + add_worker(&(g_sched.core), 1.0); + + push_job(&g_sched, fcfs, JOB_B1_ID, QUEUE_B, 1000ULL); + push_job(&g_sched, fcfs, JOB_B2_ID, QUEUE_B, 1001ULL); + push_job(&g_sched, fcfs, JOB_M_ID, QUEUE_A, 1002ULL); + + m = sched_get_job(&(g_sched.core), JOB_M_ID); + TEST_ASSERT_NOT_NULL(m); + TEST_ASSERT_EQUAL_INT(0, m->priority); + + /* move M from qa to qb */ + memset(&move_submit, 0, sizeof(move_submit)); + move_submit.queue = (char*)QUEUE_B; + memset(&move, 0, sizeof(move)); + move.jobid = JOB_M_ID; + move.state = JOB_STAT_PEND; + move.submit = &move_submit; + + memset(&ev, 0, sizeof(ev)); + ev.type = XCE_CORE_EVENT_JOB_UPDATE; + ev.u.job_update.job = &move; + rc = xce_core_push_events(&(g_sched.core), &ev, 1); + TEST_ASSERT_TRUE(rc >= 0); + + TEST_ASSERT_EQUAL_STRING_MESSAGE(QUEUE_B, m->queue->name, + "the move must reach the job"); + TEST_ASSERT_EQUAL_INT_MESSAGE( + 0, m->priority, + "moving a job must not change its own priority (bug-1944)"); + + run_round(&(g_sched.core)); + + TEST_ASSERT_EQUAL_UINT_MESSAGE( + JOB_STAT_RUN, sched_get_job(&(g_sched.core), JOB_B1_ID)->state, + "B1, first in qb, must take the only slot"); + TEST_ASSERT_EQUAL_UINT_MESSAGE( + JOB_STAT_PEND, m->state, + "the moved job must not overtake jobs already pending in qb"); +} + +int main(void) +{ + UNITY_BEGIN(); + RUN_TEST(test_moved_job_keeps_its_priority_and_place); + return UNITY_END(); +} diff --git a/src/scheduler/testsuite/test_job_update.c b/src/scheduler/testsuite/test_job_update.c index c262c2d2..4d9c40c0 100644 --- a/src/scheduler/testsuite/test_job_update.c +++ b/src/scheduler/testsuite/test_job_update.c @@ -212,7 +212,9 @@ test_update_existing_job_mutates_in_place(void) /* mutable fields updated */ TEST_ASSERT_EQUAL_UINT(JOB_STAT_RUN, job->state); TEST_ASSERT_EQUAL_PTR(g_test_queues[0], job->queue); - TEST_ASSERT_EQUAL_INT(5, job->priority); + /* bug-1944: the job keeps its own priority; the queue's (5) is not + * copied onto it */ + TEST_ASSERT_EQUAL_INT(1, job->priority); /* live runtime state preserved (object identity, not a fresh calloc) */ TEST_ASSERT_EQUAL_PTR(schedobj, job->schedobj);