fix(scheduler): allocate every primary of a COMPOUND, not only the last (bug-1863)
Each primary's allocator (seqjob/parjob my_alloc) sets ctx->alloc to a
fresh job allocation covering only its own hosts, so a multi-primary
COMPOUND ("1*{A} + 2*{B}") dispatched only the last primary's hosts --
leaking the earlier allocations while every primary's hosts had been
debited in ctx.uncomitted.
sched_make_compound_alloc_ now folds each primary's allocation into one
job allocation via the new libxce merge_job_alloc(): a host placed by
several primaries becomes one entry with per-resource values added
(matched on resname, instid and reserve, so distinct instances are never
summed); new hosts and shared allocations are appended/merged the same
way. Partial-failure rollback remains bug-1766.
Tests: "1*{..} + 2*{..}" on 2x2 slots now allocates 3 slots across two
distinct hosts; "1*{..} + 1*{..}" yields one host entry with 2 slots
(both kept only the last primary before). Scheduler suite 87/87, lib 1/1.
Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
This commit is contained in:
parent
a6a0502d62
commit
aea3473823
4 changed files with 176 additions and 5 deletions
|
|
@ -14,5 +14,6 @@ extern xce_alloc_t** make_allocs(int, ...);
|
||||||
extern xce_host_alloc_t** make_host_allocs(int, ...);
|
extern xce_host_alloc_t** make_host_allocs(int, ...);
|
||||||
extern xce_job_account_t* make_job_account(xce_job_alloc_t*);
|
extern xce_job_account_t* make_job_account(xce_job_alloc_t*);
|
||||||
extern xce_job_alloc_t* dup_job_alloc(const xce_job_alloc_t*);
|
extern xce_job_alloc_t* dup_job_alloc(const xce_job_alloc_t*);
|
||||||
|
extern int merge_job_alloc(xce_job_alloc_t*, const xce_job_alloc_t*);
|
||||||
|
|
||||||
#endif
|
#endif
|
||||||
|
|
|
||||||
|
|
@ -323,3 +323,103 @@ xce_job_account_t* make_job_account(xce_job_alloc_t* jalloc)
|
||||||
|
|
||||||
return account;
|
return account;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/* merge_job_alloc() (bug-1863) -- add src's allocations into dst: a host
|
||||||
|
* already in dst gets src's per-resource values added to its matching
|
||||||
|
* entries (same resname, instid and reserve) or appended as new entries; a
|
||||||
|
* new host is appended; shared allocations merge the same way. Deep-copies
|
||||||
|
* from src (the caller still owns and frees it), so a host appears once in
|
||||||
|
* dst however many compound primaries placed tasks on it. Returns 0, or
|
||||||
|
* -ENOMEM with dst possibly partly merged (same OOM convention as
|
||||||
|
* dup_job_alloc above). */
|
||||||
|
static int same_alloc_slot(const xce_alloc_t* a, const xce_alloc_t* b)
|
||||||
|
{
|
||||||
|
return (strcmp(a->resname, b->resname) == 0) && (a->instid == b->instid)
|
||||||
|
&& (strcmp(a->reserve ? a->reserve : "",
|
||||||
|
b->reserve ? b->reserve : "") == 0);
|
||||||
|
}
|
||||||
|
|
||||||
|
static int merge_allocs(int* ndst, xce_alloc_t*** dst, int nsrc,
|
||||||
|
xce_alloc_t** src)
|
||||||
|
{
|
||||||
|
int i, j;
|
||||||
|
|
||||||
|
for (i = 0; i < nsrc; i++) {
|
||||||
|
xce_alloc_t** grown;
|
||||||
|
|
||||||
|
if (src[i] == NULL) {
|
||||||
|
continue;
|
||||||
|
}
|
||||||
|
for (j = 0; j < *ndst; j++) {
|
||||||
|
if (((*dst)[j] != NULL) && same_alloc_slot((*dst)[j], src[i])) {
|
||||||
|
(*dst)[j]->value += src[i]->value;
|
||||||
|
break;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
if (j < *ndst) {
|
||||||
|
continue;
|
||||||
|
}
|
||||||
|
|
||||||
|
grown = (xce_alloc_t**)realloc(*dst,
|
||||||
|
sizeof(*grown) * (size_t)(*ndst + 1));
|
||||||
|
if (grown == NULL) {
|
||||||
|
return -ENOMEM;
|
||||||
|
}
|
||||||
|
*dst = grown;
|
||||||
|
(*dst)[*ndst] = dup_alloc(src[i]);
|
||||||
|
if ((*dst)[*ndst] == NULL) {
|
||||||
|
return -ENOMEM;
|
||||||
|
}
|
||||||
|
(*ndst)++;
|
||||||
|
}
|
||||||
|
|
||||||
|
return 0;
|
||||||
|
}
|
||||||
|
|
||||||
|
int merge_job_alloc(xce_job_alloc_t* dst, const xce_job_alloc_t* src)
|
||||||
|
{
|
||||||
|
int i, j, rc;
|
||||||
|
|
||||||
|
if ((dst == NULL) || (src == NULL)) {
|
||||||
|
return -EINVAL;
|
||||||
|
}
|
||||||
|
|
||||||
|
for (i = 0; i < src->nhosts; i++) {
|
||||||
|
const xce_host_alloc_t* sh = src->allocs[i];
|
||||||
|
xce_host_alloc_t** grown;
|
||||||
|
|
||||||
|
if (sh == NULL) {
|
||||||
|
continue;
|
||||||
|
}
|
||||||
|
for (j = 0; j < dst->nhosts; j++) {
|
||||||
|
if ((dst->allocs[j] != NULL)
|
||||||
|
&& (strcmp(dst->allocs[j]->host, sh->host) == 0)) {
|
||||||
|
break;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
if (j < dst->nhosts) {
|
||||||
|
rc = merge_allocs(&(dst->allocs[j]->nallocs),
|
||||||
|
&(dst->allocs[j]->allocs), sh->nallocs,
|
||||||
|
sh->allocs);
|
||||||
|
if (rc < 0) {
|
||||||
|
return rc;
|
||||||
|
}
|
||||||
|
continue;
|
||||||
|
}
|
||||||
|
|
||||||
|
grown = (xce_host_alloc_t**)realloc(
|
||||||
|
dst->allocs, sizeof(*grown) * (size_t)(dst->nhosts + 1));
|
||||||
|
if (grown == NULL) {
|
||||||
|
return -ENOMEM;
|
||||||
|
}
|
||||||
|
dst->allocs = grown;
|
||||||
|
dst->allocs[dst->nhosts] = dup_host_alloc(sh);
|
||||||
|
if (dst->allocs[dst->nhosts] == NULL) {
|
||||||
|
return -ENOMEM;
|
||||||
|
}
|
||||||
|
dst->nhosts++;
|
||||||
|
}
|
||||||
|
|
||||||
|
return merge_allocs(&(dst->nres_shared), &(dst->allocs_shared),
|
||||||
|
src->nres_shared, src->allocs_shared);
|
||||||
|
}
|
||||||
|
|
|
||||||
|
|
@ -280,9 +280,10 @@ static int sched_make_compound_alloc_(xce_sched_core_t* core,
|
||||||
xce_sched_obj_t* obj,
|
xce_sched_obj_t* obj,
|
||||||
xce_compound_resreq_t* resreq)
|
xce_compound_resreq_t* resreq)
|
||||||
{
|
{
|
||||||
int i;
|
int i;
|
||||||
xce_sched_ctx_t ctx;
|
xce_sched_ctx_t ctx;
|
||||||
int rc = 0;
|
xce_job_alloc_t* merged = NULL;
|
||||||
|
int rc = 0;
|
||||||
|
|
||||||
sched_ctx_init(&ctx, core, obj);
|
sched_ctx_init(&ctx, core, obj);
|
||||||
|
|
||||||
|
|
@ -295,10 +296,27 @@ static int sched_make_compound_alloc_(xce_sched_core_t* core,
|
||||||
if (rc <= 0) {
|
if (rc <= 0) {
|
||||||
break;
|
break;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/* bug-1863: each primary's allocator sets a fresh ctx.alloc for its
|
||||||
|
* own hosts only -- fold it into the whole compound's allocation
|
||||||
|
* (every primary's hosts are already debited in ctx.uncomitted) */
|
||||||
|
if (ctx.alloc != NULL) {
|
||||||
|
if (merged == NULL) {
|
||||||
|
MOVE_PTR(merged, ctx.alloc);
|
||||||
|
} else {
|
||||||
|
int mrc = merge_job_alloc(merged, ctx.alloc);
|
||||||
|
|
||||||
|
free_JobAlloc_ptr(&(ctx.alloc));
|
||||||
|
if (mrc < 0) {
|
||||||
|
rc = mrc;
|
||||||
|
break;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
if (ctx.alloc != NULL) {
|
if (merged != NULL) {
|
||||||
MOVE_PTR(obj->alloc, ctx.alloc);
|
MOVE_PTR(obj->alloc, merged);
|
||||||
}
|
}
|
||||||
|
|
||||||
if (ctx.reason != NULL) {
|
if (ctx.reason != NULL) {
|
||||||
|
|
|
||||||
|
|
@ -797,6 +797,56 @@ static void test_alter_falls_back_to_alternative_that_fits(void)
|
||||||
"the job must be placed by the second alternative (1 slot)");
|
"the job must be placed by the second alternative (1 slot)");
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/*
|
||||||
|
* bug-1863: each primary of a COMPOUND replaced ctx->alloc, so the
|
||||||
|
* dispatched allocation kept only the LAST primary's hosts while every
|
||||||
|
* primary's hosts had been debited. The whole compound must be allocated.
|
||||||
|
*/
|
||||||
|
static void test_compound_allocation_includes_every_primary(void)
|
||||||
|
{
|
||||||
|
xce_scheduler_t* sched = start_sched();
|
||||||
|
xce_decision_t* decision;
|
||||||
|
int i, j;
|
||||||
|
|
||||||
|
e2e_push_queue_and_workers(sched, "cmp2_queue");
|
||||||
|
decision = e2e_dispatch_resreq(sched, JOB_ID + 5, "cmp2_queue",
|
||||||
|
"1*{rusage[mem=1]} + 2*{rusage[mem=1]}");
|
||||||
|
TEST_ASSERT_NOT_NULL(decision);
|
||||||
|
TEST_ASSERT_EQUAL_INT(XCE_DECISION_DISPATCH, decision->op);
|
||||||
|
TEST_ASSERT_EQUAL_FLOAT_MESSAGE(3.0f,
|
||||||
|
(float)e2e_alloc_total_slots(decision->u.alloc),
|
||||||
|
"1 + 2 tasks: the allocation must cover both primaries");
|
||||||
|
|
||||||
|
/* one entry per host */
|
||||||
|
for (i = 0; i < decision->u.alloc->nhosts; i++) {
|
||||||
|
for (j = i + 1; j < decision->u.alloc->nhosts; j++) {
|
||||||
|
TEST_ASSERT_TRUE_MESSAGE(
|
||||||
|
strcmp(decision->u.alloc->allocs[i]->host,
|
||||||
|
decision->u.alloc->allocs[j]->host) != 0,
|
||||||
|
"a host must appear once in a job allocation");
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/* bug-1863: two primaries landing on the same host merge into that host's
|
||||||
|
* single entry (slots summed), not two entries or only the last one. */
|
||||||
|
static void test_compound_primaries_on_same_host_merge(void)
|
||||||
|
{
|
||||||
|
xce_scheduler_t* sched = start_sched();
|
||||||
|
xce_decision_t* decision;
|
||||||
|
|
||||||
|
e2e_push_queue_and_workers(sched, "cmp3_queue");
|
||||||
|
decision = e2e_dispatch_resreq(sched, JOB_ID + 6, "cmp3_queue",
|
||||||
|
"1*{rusage[mem=1]} + 1*{rusage[mem=1]}");
|
||||||
|
TEST_ASSERT_NOT_NULL(decision);
|
||||||
|
TEST_ASSERT_EQUAL_INT(XCE_DECISION_DISPATCH, decision->op);
|
||||||
|
TEST_ASSERT_EQUAL_FLOAT(2.0f,
|
||||||
|
(float)e2e_alloc_total_slots(decision->u.alloc));
|
||||||
|
TEST_ASSERT_EQUAL_INT_MESSAGE(1, decision->u.alloc->nhosts,
|
||||||
|
"both 1-task primaries fit the first 2-slot host -- one merged "
|
||||||
|
"entry");
|
||||||
|
}
|
||||||
|
|
||||||
int main(void)
|
int main(void)
|
||||||
{
|
{
|
||||||
UNITY_BEGIN();
|
UNITY_BEGIN();
|
||||||
|
|
@ -806,5 +856,7 @@ int main(void)
|
||||||
RUN_TEST(test_alter_resreq_job_dispatches_via_real_allocator);
|
RUN_TEST(test_alter_resreq_job_dispatches_via_real_allocator);
|
||||||
RUN_TEST(test_compound_primary_needing_more_slots_than_exist_pends);
|
RUN_TEST(test_compound_primary_needing_more_slots_than_exist_pends);
|
||||||
RUN_TEST(test_alter_falls_back_to_alternative_that_fits);
|
RUN_TEST(test_alter_falls_back_to_alternative_that_fits);
|
||||||
|
RUN_TEST(test_compound_allocation_includes_every_primary);
|
||||||
|
RUN_TEST(test_compound_primaries_on_same_host_merge);
|
||||||
return UNITY_END();
|
return UNITY_END();
|
||||||
}
|
}
|
||||||
|
|
|
||||||
Loading…
Add table
Add a link
Reference in a new issue