lei.chen/myxce
1
0
Fork 1
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions
myxce/src/scheduler/testsuite/test_queue_admin_control.c
Rongsong Shen f51a19d3d5
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>
2026-10-10 08:37:11 +00:00

290 lines
9.6 KiB
C
Raw Blame History

/*
* 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();
}
Reference in a new issue View git blame Copy permalink