lei.chen/myxce
1
0
Fork 1
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

fix(scheduler): host-scoped limit staged N times per dispatch (bug-1846 part 1)

Browse source 
match_job_host_limit_() (plugins/limit.c) runs on every per-host
allocation attempt via limit_prepare_allocation(), and unconditionally
xlist_store()d the matched vres into obj->hlimits[worker->idx]. That list
is only rebuilt by limit_job_election() on a configver bump, so a job that
kept pending on the same host gained one more copy per attempt.
sched_fill_limit_requests() then staged every copy into
ctx->alloc_req.host, and seqjob's make_host_alloc() built a host alloc
carrying all of them.

Found via temporary per-decision --sim-verify diagnostics (reverted, not
committed) against scheduler-replay-20260923-085356.replay: pc13v host
allocs held mem + slots + ~70 identical limit entries (72) instead of 3.
After the fix the replay side is always 3 entries, and the LIVE recording
itself still shows the inflation (72/81/96/114/138/177 entries, resetting
only on a config push) -- so this is a production bug, not sim-only.
make_job_account() creates one res_account per entry and
sched_update_res_accounts() charges each, so the host limit was charged
(and later released) N-fold.

Fix: hlimits_contains_() pointer-identity check (vres is interned per name
in privdata->instances); a matched vres is stored once per (job, worker).

Test: test_limit_host_dedup.c (sched_limit_host_dedup_test) -- five
plugin_prepare_allocation() attempts for one job on one worker must leave
hlimits[worker]->count == 1 and one staged copy of the vres in
alloc_req.host. RED before the fix (Expected 1 Was 2), GREEN after; full
scheduler suite 86/86 (asan).

Still open (bug-1846 part 2, logged in buglog): 9 live-recorded rounds end
after 11-12 decisions while replay processes the whole candidate list; not
root-caused, needs the live scheduler.log or a fresh capture.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
This commit is contained in:
Rongsong Shen 2026-09-24 04:21:27 +00:00
parent fcf119b4ac
commit 77f047f2e2
No known key found for this signature in database
GPG key ID: 5BCDAAE0123E9FE2
4 changed files with 319 additions and 5 deletions
Download patch file Download diff file Expand all files Collapse all files

27
src/scheduler/src/plugins/limit.c
View file

@ -381,6 +381,31 @@ static int limit_job_election(void* arg, xce_sched_job_t* job)
return rc;
}
/*
* hlimits_contains_ - bug-1846: match_job_host_limit_() runs on EVERY
* per-host allocation attempt (limit_prepare_allocation(), via
* plugin_prepare_allocation()), but obj->hlimits is only rebuilt when
* limit_job_election() sees a config-version bump. Storing the matched vres
* unconditionally therefore appended one more copy per attempt for a job
* that kept pending on the same host, and sched_fill_limit_requests()
* staged every copy into ctx->alloc_req.host -- a real captured replay
* showed a DISPATCH host alloc carrying 70 identical limit entries. A vres
* is interned per name in privdata->instances, so pointer identity is the
* right membership test.
*/
static int hlimits_contains_(xlist_t* hlimits, xce_vres_t* vres)
{
int i;
for (i = 0; i < hlimits->count; i++) {
if ((xce_vres_t*)xlist_get(hlimits, i) == vres) {
return 1;
}
}
return 0;
}
static int match_job_host_limit_(xce_sched_plugin_t* plugin,
xce_sched_ctx_t* ctx, xce_limit_t* limit)
{
@ -480,7 +505,7 @@ static int match_job_host_limit_(xce_sched_plugin_t* plugin,
}
rc = update_limit_res_value(&(worker->resvals), vres);
if (rc >= 0) {
if ((rc >= 0) && !hlimits_contains_(hlimits, vres)) {
rc = xlist_store(hlimits, vres);
}

24
src/scheduler/testsuite/meson.build
View file

@ -2207,6 +2207,30 @@ test('scheduler sim contention limit driver parity test',
sched_sim_contention_limit_driver_parity_test, suite: 'scheduler',
env: ['ASAN_OPTIONS=detect_leaks=0'])
# bug-1846: a host-scoped limit must be staged once per (job, worker) no
# matter how many allocation attempts the job makes there --
# match_job_host_limit_() used to append a fresh copy to obj->hlimits on
# every attempt, inflating the DISPATCH host allocation (72 vs 3 entries in
# a real captured replay). Links every scheduler source except main.c, same
# as the driver-parity fixture above.
sched_limit_host_dedup_test_srcs = ['test_limit_host_dedup.c']
foreach f : scheduler_srcs
if f != 'src/main.c'
sched_limit_host_dedup_test_srcs += ['../' + f]
endif
endforeach
sched_limit_host_dedup_test = executable(
'sched_limit_host_dedup_test',
sched_limit_host_dedup_test_srcs,
include_directories:[incdir],
dependencies: [scheduler_test_deps],
install: false)
test('scheduler limit host dedup test',
sched_limit_host_dedup_test, suite: 'scheduler',
env: ['ASAN_OPTIONS=detect_leaks=0'])
# scheduler-sim-live-allocation spec, task 5.2 (Req 2.1, 2.2, 2.3, 2.4): the
# feature's concrete, original acceptance test (bug-1826) -- a real global
# AND host-scoped limit configured in a real SnapshotEntry's Conf4Sched,

260
src/scheduler/testsuite/test_limit_host_dedup.c Normal file
View file

@ -0,0 +1,260 @@
/*
* test_limit_host_dedup.c - bug-1846 regression: a host-scoped limit must
* be staged ONCE per (job, worker), no matter how many allocation attempts
* that job makes on that worker.
*
* limit_prepare_allocation() -> match_job_host_limit_() (plugins/limit.c)
* runs on EVERY per-host allocation attempt (plugin_prepare_allocation(),
* called by resource.c's sched_try_alloc_on_host_()), and used to
* xlist_store() the matched vres into obj->hlimits[worker->idx]
* unconditionally. obj->hlimits is only rebuilt when limit_job_election()
* sees a config-version bump, so a job that pended on the same host round
* after round accumulated one extra copy of the limit per attempt.
* plugin_prepare_allocation() then staged every copy into
* ctx->alloc_req.host via sched_fill_limit_requests(), and seqjob's
* make_host_alloc() built a host allocation carrying all of them -- live
* captured in scheduler-replay-20260923-085356.replay as a DISPATCH whose
* pc13v host alloc held mem + slots + 70 identical limit entries (72)
* where the recording held 3.
*
* This fixture calls plugin_prepare_allocation() repeatedly for one job on
* one worker -- exactly what successive pending rounds do -- and requires
* the host-limit list and the staged host request to each hold the limit
* exactly once.
*/
#include <errno.h>
#include <stdint.h>
#include <stdlib.h>
#include <string.h>
#include <unity/unity.h>
#include "controller.capnp.h"
#include "plugin.h"
#include "resobj.h"
#include "sched/types.h"
#include "scheduler.h"
#include "xcelimit.h"
#include "xstring.h"
#define WORKER_NAME "hdedup_worker"
#define QUEUE_NAME "normal"
#define TEST_RESREQ ((char*)"select[type==any] rusage[mem=1]")
#define LIMIT_RESNAME "hdedup_host_res"
#define LIMIT_NAME "hdedup_host_limit"
#define JOB_ID ((uint64_t)518461ULL)
#define NATTEMPTS (5)
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));
}
/* same bare-resource shape test_sim_contention_limit_driver_parity.c uses:
* is_host_based_limit() resolves LIMIT_NAME as host-based from
* limit->val.percent != 0, not from any flag on this resource. */
static void register_plain_resource(var resmgr, const char* name)
{
xce_resource_t res;
memset(&res, 0, sizeof(res));
res.name = strdup(name);
TEST_ASSERT_NOT_NULL(makeResource(resmgr, &res));
}
static void add_worker_resval(xce_sched_worker_t* worker,
xce_sched_core_t* core, const char* resname,
double value)
{
xce_resource_t* r = NULL;
xce_val_t* val = NULL;
int rc;
rc = key_find(core->resmgr, (void*)resname, (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 = value;
zarray_put(&(worker->resvals), r->idx, val);
}
static xce_sched_worker_t* make_worker(xce_sched_core_t* core)
{
xce_sched_worker_t* w = (xce_sched_worker_t*)calloc(1, sizeof(*w));
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);
add_worker_resval(w, core, "slots", 4.0);
add_worker_resval(w, core, LIMIT_RESNAME, 1000.0);
return w;
}
static xce_sched_job_t* make_job(xce_scheduler_t* sched)
{
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_NAME;
submit.user = (char*)"hdedup_user";
submit.project_name = (char*)"default";
submit.resreq = TEST_RESREQ;
submit.mintasks = 1;
submit.maxtasks = 1;
memset(&j, 0, sizeof(j));
j.jobid = JOB_ID;
j.state = JOB_STAT_PEND;
j.sub_time = 1000ULL;
j.queue_time = 1000ULL;
j.submit = &submit;
job = make_sched_job(sched, &j);
TEST_ASSERT_NOT_NULL(job);
TEST_ASSERT_EQUAL_INT(XCE_RESREQ_PRIMARY, job->schedobj->tag);
rc = sched_add_job(&(sched->core), job);
TEST_ASSERT_TRUE(rc >= 0);
return job;
}
static int count_staged(const xvec_idval_t* v, int idx)
{
int i;
int n = 0;
for (i = 0; i < v->count; i++) {
int residx = -1;
xvec_get((xvec_t*)&(v->idxs), i, &residx);
if (residx == idx) {
n++;
}
}
return n;
}
static xce_scheduler_t g_sched;
static void test_repeated_host_attempts_stage_host_limit_once(void)
{
xce_limit_val_t limit_val = { .resname = (char*)LIMIT_RESNAME,
.percent = 1,
.val = 50.0 };
xce_limit_config_t limit_cfg = { .name = (char*)LIMIT_NAME,
.val = &limit_val };
xce_limit_config_t* limits[1] = { &limit_cfg };
xce_conf4sched_t config;
xce_sched_plugin_t* limit_plugin = NULL;
xce_sched_queue_t* queue;
xce_sched_worker_t* w;
xce_sched_job_t* job;
xce_sched_ctx_t ctx;
xlist_t* hlimits;
xce_vres_t* vres;
int vres_idx;
int attempt;
int rc;
memset(&config, 0, sizeof(config));
config.nlimits = 1;
config.limits = limits;
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);
register_plain_resource(g_sched.core.resmgr, LIMIT_RESNAME);
limit_plugin = limit_new_plugin();
TEST_ASSERT_NOT_NULL(limit_plugin);
rc = plugin_register(&g_sched, limit_plugin);
TEST_ASSERT_TRUE(rc >= 0);
rc = limit_seed_config(limit_plugin, &(g_sched.core), &config);
TEST_ASSERT_TRUE(rc >= 0);
queue = (xce_sched_queue_t*)calloc(1, sizeof(*queue));
TEST_ASSERT_NOT_NULL(queue);
queue->name = makeString(QUEUE_NAME);
rc = sched_add_queue(&(g_sched.core), queue);
TEST_ASSERT_TRUE(rc >= 0);
w = make_worker(&(g_sched.core));
rc = sched_add_worker(&(g_sched.core), w);
TEST_ASSERT_TRUE(rc >= 0);
TEST_ASSERT_TRUE(w->idx >= 0);
job = make_job(&g_sched);
/* One allocation attempt per simulated pending round, all on the same
* worker, with no config change in between (so limit_job_election()
* never resets obj->hlimits). */
for (attempt = 0; attempt < NATTEMPTS; attempt++) {
rc = sched_ctx_init(&ctx, &(g_sched.core), job->schedobj);
TEST_ASSERT_EQUAL_INT(0, rc);
ctx.worker = w;
ctx.nhosts = 1;
ctx.nslots = 1;
rc = plugin_prepare_allocation(&(g_sched.core), &ctx);
TEST_ASSERT_TRUE(rc >= 0);
hlimits = (xlist_t*)zarray_get(&(job->schedobj->hlimits), w->idx);
TEST_ASSERT_NOT_NULL_MESSAGE(
hlimits, "the host-scoped limit must match on this worker");
TEST_ASSERT_EQUAL_INT_MESSAGE(
1, hlimits->count,
"obj->hlimits[worker] must hold the matched limit exactly once, "
"however many attempts the job has made on this worker");
vres = (xce_vres_t*)xlist_get(hlimits, 0);
TEST_ASSERT_NOT_NULL(vres);
vres_idx = vres->vres->idx;
TEST_ASSERT_EQUAL_INT_MESSAGE(
1, count_staged(&(ctx.alloc_req.host), vres_idx),
"alloc_req.host must stage the host limit's vres exactly once "
"-- every extra copy becomes an extra entry in the DISPATCH's "
"host allocation");
sched_ctx_fini(&ctx);
}
printf("test_repeated_host_attempts_stage_host_limit_once: PASS\n");
}
int main(void)
{
UNITY_BEGIN();
RUN_TEST(test_repeated_host_attempts_stage_host_limit_once);
return UNITY_END();
}