fix(scheduler): hold jobs of inactive queues at dispatch (bug-1938)
sched_queue_is_ready() in core.c was a `return 1` stub, so the picker (builtin_fcfs and the legacy fcfs) took jobs from every queue and admin queue inactivation never reached dispatch. It now returns false for a queue with QUEUE_STAT_INACT_ADMIN or QUEUE_STAT_INACT_WINDOW set, so an inactive queue's jobs stay pending. QUEUE_STAT_CLOSED is deliberately not a gate (LSF semantics: closing only blocks new submissions; already-pending jobs still dispatch). The status reaches the core through the same queue add/update paths in both drivers, so live and sim behave identically. test_queue_admin_control.c covers open, closed and inactive queues on both fixture shapes. Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
This commit is contained in:
parent
2fed7a6e5f
commit
f51a19d3d5
GPG key ID:
80533536B0EB651C
4 changed files with 337 additions and 8 deletions
|
|
@ -645,5 +645,23 @@ xce_sched_project_t* sched_get_project(xce_sched_core_t* core, char* name)
|
|||
/* FIXME this is just stub, we need to add more */
|
||||
int sched_job_is_ready(xce_sched_job_t* j) { return 1; }
|
||||
|
||||
/* FIXME this is just stub, we need to add more */
|
||||
int sched_queue_is_ready(xce_sched_queue_t* q) { return 1; }
|
||||
/*
|
||||
* sched_queue_is_ready - may the picker take jobs from this queue? (bug-1938)
|
||||
*
|
||||
* LSF semantics: an inactive queue (inactivated by an admin, or outside its
|
||||
* dispatch/run window) dispatches nothing, so its pending jobs stay pending.
|
||||
* QUEUE_STAT_CLOSED does not gate dispatch: closing a queue only stops new
|
||||
* submissions (refused by the controller), and jobs already pending in it
|
||||
* still run. The status is the controller's, copied in by every queue
|
||||
* add/update path (core_build_queue, core_upsert_queue, make_sched_queue),
|
||||
* so the live and simulation drivers read the same bits.
|
||||
*/
|
||||
int sched_queue_is_ready(xce_sched_queue_t* q)
|
||||
{
|
||||
if (q == NULL) {
|
||||
return 0;
|
||||
}
|
||||
|
||||
return (q->status & (QUEUE_STAT_INACT_ADMIN | QUEUE_STAT_INACT_WINDOW))
|
||||
== 0;
|
||||
}
|
||||
|
|
|
|||
|
|
@ -827,12 +827,11 @@ int builtin_fcfs_begin_session(void* picker)
|
|||
* and at exit -- both real, safe, per-call cancellation points (mirroring
|
||||
* sched_job_picker()'s own pthread_testcancel() immediately before every
|
||||
* next_job() call, which remains in place too -- belt-and-suspenders, not
|
||||
* redundant dead code). This does not weaken cancel-safety in practice
|
||||
* today: sched_job_is_ready()/sched_queue_is_ready() are unconditional
|
||||
* `return 1` stubs (scheduler.c), so this walk always returns on the very
|
||||
* first candidate examined -- fcfs_elect_job()'s own per-iteration
|
||||
* testcancel calls are equally never exercised mid-walk today, for the
|
||||
* same reason.
|
||||
* redundant dead code). This does not weaken cancel-safety in practice:
|
||||
* sched_job_is_ready() is an unconditional `return 1` stub (core.c), and
|
||||
* sched_queue_is_ready() only skips inactive queues (bug-1938), so a walk
|
||||
* skips at most every queue once, a bounded, non-blocking scan, before it
|
||||
* returns on the first job of the first ready queue.
|
||||
*/
|
||||
xce_sched_job_t* builtin_fcfs_next_job(void* picker)
|
||||
{
|
||||
|
|
|
|||
|
|
@ -3048,3 +3048,25 @@ test('scheduler trace equivalence test',
|
|||
sched_trace_equivalence_test, suite: 'scheduler',
|
||||
env: ['ASAN_OPTIONS=detect_leaks=0'],
|
||||
depends: scheduler)
|
||||
|
||||
# bug-1938: admin queue control reaches dispatch (an inactive queue holds its
|
||||
# pending jobs, a closed queue still dispatches them) on both the
|
||||
# normal-mode and sim-mode fixture shapes. Real-scheduler_srcs link
|
||||
# technique of sched_sim_contention_limit_driver_parity_test.
|
||||
sched_queue_admin_control_test_srcs = ['test_queue_admin_control.c']
|
||||
foreach f : scheduler_srcs
|
||||
if f != 'src/main.c'
|
||||
sched_queue_admin_control_test_srcs += ['../' + f]
|
||||
endif
|
||||
endforeach
|
||||
|
||||
sched_queue_admin_control_test = executable(
|
||||
'sched_queue_admin_control_test',
|
||||
sched_queue_admin_control_test_srcs,
|
||||
include_directories:[incdir],
|
||||
dependencies: [scheduler_test_deps],
|
||||
install: false)
|
||||
|
||||
test('scheduler queue admin control test',
|
||||
sched_queue_admin_control_test, suite: 'scheduler',
|
||||
env: ['ASAN_OPTIONS=detect_leaks=0'])
|
||||
|
|
|
|||
290
src/scheduler/testsuite/test_queue_admin_control.c
Normal file
290
src/scheduler/testsuite/test_queue_admin_control.c
Normal file
|
|
@ -0,0 +1,290 @@
|
|||
/*
|
||||
* test_queue_admin_control.c - bug-1938: admin queue control must reach
|
||||
* dispatch, identically on the normal-mode and sim-mode drivers.
|
||||
*
|
||||
* LSF semantics, which the controller's queue control (controller/queue.c)
|
||||
* follows:
|
||||
* - QUEUE_STAT_INACT_ADMIN ("inactive"): the queue dispatches nothing;
|
||||
* its pending jobs stay pending.
|
||||
* - QUEUE_STAT_CLOSED ("closed"): the queue accepts no new submissions,
|
||||
* but jobs already pending in it are still dispatched.
|
||||
*
|
||||
* The fixture has three queues (open, closed, inactive), one pending job in
|
||||
* each and a worker big enough for all three, so capacity never decides
|
||||
* the outcome: only the queue state does. Both drivers' fixtures are built
|
||||
* the way test_sim_contention_limit_driver_parity.c builds them (real
|
||||
* scheduler_srcs, real picker/allocator threads, the sim side bootstrapped
|
||||
* through sim_loop.c's own sim_bootstrap_* entry points), `static` fixtures
|
||||
* and exactly one RUN_TEST() (bug-1598/1632/1763 fixture-safety convention).
|
||||
*/
|
||||
#include <errno.h>
|
||||
#include <pthread.h>
|
||||
#include <stdint.h>
|
||||
#include <stdlib.h>
|
||||
#include <string.h>
|
||||
|
||||
#include <unity/unity.h>
|
||||
|
||||
#include "controller.capnp.h"
|
||||
#include "decision.h"
|
||||
#include "plugin.h"
|
||||
#include "preason.h"
|
||||
#include "resobj.h"
|
||||
#include "sched/types.h"
|
||||
#include "scheduler.h"
|
||||
#include "sim.h"
|
||||
#include "xstring.h"
|
||||
|
||||
extern void sim_bootstrap_picker(xce_sched_sim_driver_t* sim);
|
||||
extern void sim_bootstrap_seqjob_allocator(xce_sched_sim_driver_t* sim);
|
||||
extern void sim_bootstrap_parjob_allocator(xce_sched_sim_driver_t* sim);
|
||||
|
||||
#define WORKER_NAME "qctl_worker"
|
||||
#define QUEUE_OPEN_N "qctl_open"
|
||||
#define QUEUE_CLOSED_N "qctl_closed"
|
||||
#define QUEUE_INACT_N "qctl_inactive"
|
||||
#define TEST_RESREQ ((char*)"select[type==any] rusage[mem=1]")
|
||||
|
||||
#define JOB_OPEN_ID ((uint64_t)193801ULL)
|
||||
#define JOB_CLOSED_ID ((uint64_t)193802ULL)
|
||||
#define JOB_INACT_ID ((uint64_t)193803ULL)
|
||||
|
||||
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 xce_sched_worker_t* make_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->status = 0;
|
||||
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);
|
||||
TEST_ASSERT_NOT_NULL(r);
|
||||
|
||||
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);
|
||||
|
||||
return w;
|
||||
}
|
||||
|
||||
static void add_queue(xce_sched_core_t* core, xce_sched_plugin_t* picker,
|
||||
const char* name, int status)
|
||||
{
|
||||
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->status = status;
|
||||
rc = sched_add_queue(core, q);
|
||||
TEST_ASSERT_TRUE(rc >= 0);
|
||||
picker->picker_iface->on_queue_update(picker, q);
|
||||
}
|
||||
|
||||
static void push_job(xce_sched_core_t* core, xce_scheduler_t* job_ctor_wrapper,
|
||||
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*)"qctl_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(job_ctor_wrapper, &j);
|
||||
TEST_ASSERT_NOT_NULL(job);
|
||||
TEST_ASSERT_EQUAL_INT(XCE_RESREQ_PRIMARY, job->schedobj->tag);
|
||||
job->schedobj->u.primary->resreq->minTasks = 1;
|
||||
job->schedobj->u.primary->resreq->maxTasks = 1;
|
||||
|
||||
rc = sched_add_job(core, job);
|
||||
TEST_ASSERT_TRUE(rc >= 0);
|
||||
picker->picker_iface->on_job_push(picker, job);
|
||||
}
|
||||
|
||||
/* run_round - one full picker/allocator session; returns how many decisions
|
||||
* the session produced before its end-of-session NULL marker. Draining up
|
||||
* to the marker (instead of expecting a fixed count) keeps a wrong count a
|
||||
* plain assertion failure rather than a hang. */
|
||||
static int run_round(xce_sched_core_t* core)
|
||||
{
|
||||
pthread_t picker_tid;
|
||||
pthread_t allocator_tid;
|
||||
xce_decision_t* d;
|
||||
int n = 0;
|
||||
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);
|
||||
|
||||
for (;;) {
|
||||
d = NULL;
|
||||
cbuf_get(&(core->decisions), &d);
|
||||
if (d == NULL) {
|
||||
break;
|
||||
}
|
||||
n++;
|
||||
}
|
||||
|
||||
rc = xce_core_end_session(core);
|
||||
TEST_ASSERT_EQUAL_INT(0, rc);
|
||||
|
||||
return n;
|
||||
}
|
||||
|
||||
static void populate(xce_sched_core_t* core, xce_scheduler_t* job_wrapper,
|
||||
xce_sched_plugin_t* picker)
|
||||
{
|
||||
int rc;
|
||||
|
||||
add_queue(core, picker, QUEUE_OPEN_N, QUEUE_STAT_OPEN);
|
||||
add_queue(core, picker, QUEUE_CLOSED_N, QUEUE_STAT_CLOSED);
|
||||
add_queue(core, picker, QUEUE_INACT_N, QUEUE_STAT_INACT_ADMIN);
|
||||
|
||||
rc = sched_add_worker(core, make_worker(core, 8.0));
|
||||
TEST_ASSERT_TRUE(rc >= 0);
|
||||
|
||||
push_job(core, job_wrapper, picker, JOB_OPEN_ID, QUEUE_OPEN_N, 1000ULL);
|
||||
push_job(core, job_wrapper, picker, JOB_CLOSED_ID, QUEUE_CLOSED_N,
|
||||
1001ULL);
|
||||
push_job(core, job_wrapper, picker, JOB_INACT_ID, QUEUE_INACT_N, 1002ULL);
|
||||
}
|
||||
|
||||
static void assert_outcome(xce_sched_core_t* core, int ndecisions,
|
||||
const char* mode)
|
||||
{
|
||||
char msg[160];
|
||||
|
||||
snprintf(msg, sizeof(msg),
|
||||
"%s: only the open and closed queues' jobs may be decided "
|
||||
"(an inactive queue dispatches nothing)",
|
||||
mode);
|
||||
TEST_ASSERT_EQUAL_INT_MESSAGE(2, ndecisions, msg);
|
||||
|
||||
snprintf(msg, sizeof(msg), "%s: open queue job must dispatch", mode);
|
||||
TEST_ASSERT_EQUAL_UINT_MESSAGE(JOB_STAT_RUN,
|
||||
sched_get_job(core, JOB_OPEN_ID)->state,
|
||||
msg);
|
||||
snprintf(msg, sizeof(msg),
|
||||
"%s: closed queue job already pending must still dispatch",
|
||||
mode);
|
||||
TEST_ASSERT_EQUAL_UINT_MESSAGE(JOB_STAT_RUN,
|
||||
sched_get_job(core, JOB_CLOSED_ID)->state,
|
||||
msg);
|
||||
snprintf(msg, sizeof(msg), "%s: inactive queue job must stay pending",
|
||||
mode);
|
||||
TEST_ASSERT_EQUAL_UINT_MESSAGE(JOB_STAT_PEND,
|
||||
sched_get_job(core, JOB_INACT_ID)->state,
|
||||
msg);
|
||||
}
|
||||
|
||||
static xce_scheduler_t g_normal;
|
||||
static xce_sched_sim_driver_t g_sim;
|
||||
static xce_scheduler_t g_sim_job_wrapper;
|
||||
|
||||
static void test_inactive_queue_holds_jobs_closed_queue_dispatches(void)
|
||||
{
|
||||
xce_sched_plugin_t* fcfs_normal = NULL;
|
||||
xce_sched_plugin_t* fcfs_sim = NULL;
|
||||
int n_normal;
|
||||
int n_sim;
|
||||
int rc;
|
||||
|
||||
/* normal-mode-shaped fixture */
|
||||
memset(&g_normal, 0, sizeof(g_normal));
|
||||
rc = xce_sched_core_init(&(g_normal.core));
|
||||
TEST_ASSERT_EQUAL_INT(0, rc);
|
||||
register_slots_resource(g_normal.core.resmgr);
|
||||
|
||||
fcfs_normal = builtin_fcfs_new_plugin();
|
||||
TEST_ASSERT_NOT_NULL(fcfs_normal);
|
||||
rc = plugin_register(&g_normal, fcfs_normal);
|
||||
TEST_ASSERT_TRUE(rc >= 0);
|
||||
g_normal.core.job_pick_plugin = (char*)"builtin_fcfs";
|
||||
rc = register_seqjob_allocator(&g_normal);
|
||||
TEST_ASSERT_TRUE(rc >= 0);
|
||||
rc = register_parjob_allocator(&g_normal);
|
||||
TEST_ASSERT_TRUE(rc >= 0);
|
||||
|
||||
populate(&(g_normal.core), &g_normal, fcfs_normal);
|
||||
n_normal = run_round(&(g_normal.core));
|
||||
|
||||
/* sim-mode-shaped fixture */
|
||||
memset(&g_sim, 0, sizeof(g_sim));
|
||||
rc = xce_sched_core_init(&(g_sim.core));
|
||||
TEST_ASSERT_EQUAL_INT(0, rc);
|
||||
register_slots_resource(g_sim.core.resmgr);
|
||||
|
||||
sim_bootstrap_picker(&g_sim);
|
||||
rc = key_find(g_sim.core.plugins, "builtin_fcfs", (void**)&fcfs_sim);
|
||||
TEST_ASSERT_TRUE(rc >= 0);
|
||||
TEST_ASSERT_NOT_NULL(fcfs_sim);
|
||||
sim_bootstrap_seqjob_allocator(&g_sim);
|
||||
sim_bootstrap_parjob_allocator(&g_sim);
|
||||
|
||||
memset(&g_sim_job_wrapper, 0, sizeof(g_sim_job_wrapper));
|
||||
rc = xce_sched_core_init(&(g_sim_job_wrapper.core));
|
||||
TEST_ASSERT_EQUAL_INT(0, rc);
|
||||
g_sim_job_wrapper.core.resmgr = g_sim.core.resmgr;
|
||||
g_sim_job_wrapper.core.workers = g_sim.core.workers;
|
||||
g_sim_job_wrapper.core.queues = g_sim.core.queues;
|
||||
|
||||
populate(&(g_sim.core), &g_sim_job_wrapper, fcfs_sim);
|
||||
n_sim = run_round(&(g_sim.core));
|
||||
|
||||
assert_outcome(&(g_normal.core), n_normal, "normal mode");
|
||||
assert_outcome(&(g_sim.core), n_sim, "sim mode");
|
||||
}
|
||||
|
||||
int main(void)
|
||||
{
|
||||
UNITY_BEGIN();
|
||||
RUN_TEST(test_inactive_queue_holds_jobs_closed_queue_dispatches);
|
||||
return UNITY_END();
|
||||
}
|
||||
Loading…
Add table
Add a link
Reference in a new issue