diff --git a/src/include/xcealloc.h b/src/include/xcealloc.h index 98111c1f..be370b2f 100644 --- a/src/include/xcealloc.h +++ b/src/include/xcealloc.h @@ -14,5 +14,6 @@ extern xce_alloc_t** make_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_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 diff --git a/src/lib/xcealloc.c b/src/lib/xcealloc.c index e02f19b3..2de73373 100644 --- a/src/lib/xcealloc.c +++ b/src/lib/xcealloc.c @@ -323,3 +323,103 @@ xce_job_account_t* make_job_account(xce_job_alloc_t* jalloc) 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); +} diff --git a/src/scheduler/src/scheduler.c b/src/scheduler/src/scheduler.c index a6d059ce..59d30aeb 100644 --- a/src/scheduler/src/scheduler.c +++ b/src/scheduler/src/scheduler.c @@ -280,9 +280,10 @@ static int sched_make_compound_alloc_(xce_sched_core_t* core, xce_sched_obj_t* obj, xce_compound_resreq_t* resreq) { - int i; - xce_sched_ctx_t ctx; - int rc = 0; + int i; + xce_sched_ctx_t ctx; + xce_job_alloc_t* merged = NULL; + int rc = 0; sched_ctx_init(&ctx, core, obj); @@ -295,10 +296,27 @@ static int sched_make_compound_alloc_(xce_sched_core_t* core, if (rc <= 0) { 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) { - MOVE_PTR(obj->alloc, ctx.alloc); + if (merged != NULL) { + MOVE_PTR(obj->alloc, merged); } if (ctx.reason != NULL) { diff --git a/src/scheduler/testsuite/test_parjob_e2e_multislot.c b/src/scheduler/testsuite/test_parjob_e2e_multislot.c index 4df5839e..62dac25a 100644 --- a/src/scheduler/testsuite/test_parjob_e2e_multislot.c +++ b/src/scheduler/testsuite/test_parjob_e2e_multislot.c @@ -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)"); } +/* + * 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) { UNITY_BEGIN(); @@ -806,5 +856,7 @@ int main(void) RUN_TEST(test_alter_resreq_job_dispatches_via_real_allocator); RUN_TEST(test_compound_primary_needing_more_slots_than_exist_pends); 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(); }