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

fix(scheduler): place a task-range job with the largest count that fits (bug-1931)

Browse source 
The parjob plan generator split only maxTasks, so a job submitted with a
task range (minTasks < maxTasks) pended until all maxTasks tasks fit,
even when minTasks did (a tasks 2..5 job pended on a host with 2 free
slots).

The runner now handles a task range in sched_alloc_range_(): it probes
minTasks first and pends after that one search when it does not fit;
otherwise it places maxTasks when that fits, or binary-searches for the
largest count that fits and places it. Fitting is monotone in the count
for balanced plans, so the search is exact for today's plan shapes.
ctx->ntasks carries the count; parjob's job-level pass scales per-task
shared wants and global limits to it, and the generator gets the request
with its task count fixed to it. Fixed counts keep the old path.

The decision trace shows each count tried (task count N (minimum) /
(maximum), placing N tasks) and releases a fitting probe's hosts.
Test fixtures that used minTasks = 1 as filler for an N-task job are now
fixed-count.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
This commit is contained in:
Rongsong Shen 2026-10-09 08:51:44 +00:00
parent 7cdf25e001
commit e3b87d28f3
Signed by untrusted user: rshen
GPG key ID: 80533536B0EB651C
19 changed files with 726 additions and 32 deletions
Download patch file Download diff file Expand all files Collapse all files

5
src/scheduler/include/sched/trace.h
View file

@ -189,6 +189,7 @@ typedef enum {
SCHED_TRACE_E_ROLLBACK, /* released hosts, merged (5.8) */
SCHED_TRACE_E_CHECK_ROW, /* cluster-wide check row (4.1, 4.2) */
SCHED_TRACE_E_HOST_ROW, /* unmet row under a host (5.4, 5.5) */
SCHED_TRACE_E_TASK_COUNT, /* task-range count tried (bug-1931) */
SCHED_TRACE_E_NTYPES
} sched_trace_entry_type_t;
@ -292,7 +293,7 @@ typedef struct sched_trace_entry_s {
int rc; /* > 0 satisfied, 0 not, < 0 error */
int not_tried_after; /* members after this one, not tried */
} resreq_end;
struct { /* ALT_SKIPPED, ALT_RESULT (i == 0: none chosen) */
struct { /* ALT_SKIPPED, ALT_RESULT (i == 0: none chosen), TASK_COUNT */
int i;
int n;
} alt;
@ -626,6 +627,8 @@ extern void sched_trace_check_begin(xce_sched_trace_rec_t* rec,
const xce_sched_obj_t* obj);
extern void sched_trace_check_end(xce_sched_trace_rec_t* rec, int rc,
bool plans_follow);
/* bug-1931: count n next; kind 1 minimum, 2 maximum, 3 placing, else 0 */
extern void sched_trace_task_count(xce_sched_trace_rec_t* rec, int n, int kind);
/* No plan generated for the current requirement, and why (4.7). Counts
* the requirement under "no plan generated" (9.3), once: a search end

4
src/scheduler/include/sched/types.h
View file

@ -189,6 +189,10 @@ typedef struct {
xce_primary_resreq_t* primary;
int nhosts;
int nslots;
/* bug-1931: the task count a task-range attempt (minTasks < maxTasks)
* is trying, set by the runner before the job-level prepare pass; 0
* (sched_ctx_init()/sched_ctx_set_primary()) means primary->maxTasks. */
int ntasks;
struct {
xvec_t* shared;
zarray_t hosts;

153
src/scheduler/src/alloc_runner.c
View file

@ -66,9 +66,12 @@ static int sched_alloc_ensure_reason_(xce_sched_ctx_t* ctx)
* and returns 0 (Req 2.2). vget() returns 0 on success, so only a negative
* result skips an entry (bug-1893).
*
* Any failure returns its negative error code (Req 2.4). */
* Any failure returns its negative error code (Req 2.4).
*
* bug-1931: a task-range attempt checks each count with debit == 0 and
* debits the placed count itself (sched_alloc_count_()). */
static int sched_alloc_prepare_shared_(xce_sched_core_t* core,
xce_sched_ctx_t* ctx)
xce_sched_ctx_t* ctx, int debit)
{
xvec_t* uncomit = NULL;
int rc;
@ -106,6 +109,10 @@ static int sched_alloc_prepare_shared_(xce_sched_core_t* core,
return 0;
}
if (!debit) {
return 1;
}
rc = xvec_idval(uncomit, &(ctx->alloc_req.shared), XVEC_ADD_FLOAT);
if (rc < 0) {
return rc;
@ -465,21 +472,37 @@ static int sched_finalize_job_alloc_(xce_scheduler_t* sched,
*
* Satisfied: the result built from the plan's claims, which the
* assembly consumes (1, or -ENOMEM). No plan satisfied: 0. Claims-buffer
* growth or snapshot failure: that error. No snapshot is freed here. */
* growth or snapshot failure: that error. No snapshot is freed here.
*
* bug-1931: `ntasks` > 0 splits that many tasks (the generator gets the
* request with its task count fixed to it); 0 splits maxTasks. Without
* `commit` a satisfied plan is a probe: its claims are released and
* nothing is built (1). `end_search` stores the search end in the trace;
* a task-range attempt stores one for all its counts. */
static int sched_alloc_search_(const xce_sched_allocator_iface_t* iface,
xce_scheduler_t* sched, xce_sched_core_t* core,
xce_sched_ctx_t* ctx)
xce_sched_ctx_t* ctx, int ntasks, int commit,
int end_search)
{
xce_alloc_plan_iter_t iter;
xce_alloc_plan_t* plan;
xce_primary_resreq_t sized;
sched_alloc_claim_t* claims = NULL;
int cap = 0;
int n = 0;
int rc = 0;
int k;
sized = *(ctx->primary);
if (ntasks > 0) {
sized.minTasks = ntasks;
sized.maxTasks = ntasks;
}
/* A generator that cannot start is not a resource verdict: the attempt
* pends, with nothing for plan_fini to release (Req 3.4). */
if (iface->plan_init(&iter, ctx->primary, core) != 0) {
if (iface->plan_init(&iter, (ntasks > 0) ? &sized : ctx->primary, core)
!= 0) {
/* Decision trace (4.7): counted as "no plan generated" here; no
* search end follows */
SCHED_TRACE(core,
@ -510,19 +533,125 @@ static int sched_alloc_search_(const xce_sched_allocator_iface_t* iface,
}
}
if (rc > 0) {
if ((rc > 0) && commit) {
rc = sched_finalize_job_alloc_(sched, ctx, claims, n);
/* Decision trace (4.5) */
SCHED_TRACE(
core, (rc > 0) ? sched_trace_commit(core->trace_rec)
: sched_trace_finalize_failed(core->trace_rec, rc));
} else if (rc > 0) {
/* bug-1931: a probe that fits gives its hosts back */
SCHED_TRACE(core, for (k = 0; k < n; k++) {
sched_trace_rollback_host(core->trace_rec, claims[k].w->name);
});
sched_rollback_claims_(ctx, claims, n);
}
iface->plan_fini(&iter);
free(claims);
/* Decision trace (4.6, 4.7, 4.9): the requirement's plan counts */
SCHED_TRACE(core, sched_trace_search_end(core->trace_rec, rc));
if (end_search) {
SCHED_TRACE(core, sched_trace_search_end(core->trace_rec, rc));
}
return rc;
}
/* bug-1931: one task count of a task-range attempt -- the job-level
* prepare pass for `ntasks` (per-task shared wants and global limits scale
* with it), the cluster-wide check and the plan search (*searched set when
* it ran). `commit` places the count and debits its shared wants;
* otherwise the count is only probed and nothing stays debited. 1 fits,
* 0 does not, < 0 error. */
static int sched_alloc_count_(const xce_sched_allocator_iface_t* iface,
xce_sched_plugin_t* plugin, xce_sched_ctx_t* ctx,
int ntasks, int kind, int commit, int* searched)
{
xce_sched_core_t* core = plugin->core;
xvec_t* uncomit = NULL;
int rc;
ctx->ntasks = ntasks;
ctx->worker = NULL;
rc = plugin_prepare_allocation(core, ctx);
if (rc < 0) {
return rc;
}
SCHED_TRACE(core, sched_trace_task_count(core->trace_rec, ntasks, kind));
SCHED_TRACE(core, sched_trace_check_begin(core->trace_rec, ctx->schedobj));
rc = sched_alloc_prepare_shared_(core, ctx, 0);
SCHED_TRACE(core, sched_trace_check_end(core->trace_rec, rc, rc > 0));
if (rc <= 0) {
return rc;
}
/* as the fixed-count path, the placed count's shared wants are counted
* before its search, and given back when it places nothing */
if (commit) {
rc = sched_ctx_get_uncomit_shared(&uncomit, ctx);
if (rc >= 0) {
rc = xvec_idval(uncomit, &(ctx->alloc_req.shared), XVEC_ADD_FLOAT);
}
if (rc < 0) {
return rc;
}
}
*searched = 1;
rc = sched_alloc_search_(iface, (xce_scheduler_t*)plugin->sched, core, ctx,
ntasks, commit, 0);
if (commit && (rc <= 0)) {
(void)xvec_idval(uncomit, &(ctx->alloc_req.shared), XVEC_SUB_FLOAT);
}
return rc;
}
/* bug-1931: a task range (minTasks < maxTasks) places the largest count
* that fits. Fitting is monotone in the count: each part of the balanced
* split of N-1 tasks into k parts is no larger than N's, a host that fits
* a part fits any smaller part (per-host requests are per task), the
* first-fit, largest-part-first claim is complete for balanced plans, and
* per-task shared wants only grow with N. So minTasks is tried first and,
* when it does not fit, the attempt pends after that one search; then
* maxTasks, placed when it fits; else a binary search between them, and
* the largest fitting count is placed. One search end covers the plan
* searches of the attempt; like a fixed count, none when no search ran. */
static int sched_alloc_range_(const xce_sched_allocator_iface_t* iface,
xce_sched_plugin_t* plugin, xce_sched_ctx_t* ctx)
{
xce_sched_core_t* core = plugin->core;
int fit = ctx->primary->minTasks;
int miss = ctx->primary->maxTasks;
int searched = 0;
int rc;
rc = sched_alloc_count_(iface, plugin, ctx, fit, 1, 0, &searched);
if (rc > 0) {
rc = sched_alloc_count_(iface, plugin, ctx, miss, 2, 1, &searched);
while (rc == 0) {
int mid = fit + (miss - fit) / 2;
if (mid == fit) {
rc = sched_alloc_count_(iface, plugin, ctx, fit, 3, 1,
&searched);
break;
}
rc = sched_alloc_count_(iface, plugin, ctx, mid, 0, 0, &searched);
if (rc > 0) {
fit = mid;
rc = 0;
} else if (rc == 0) {
miss = mid;
}
}
}
if (searched) {
SCHED_TRACE(core, sched_trace_search_end(core->trace_rec, rc));
}
return rc;
}
@ -557,11 +686,17 @@ int sched_allocator_run(void* plugin_arg, xce_sched_ctx_t* ctx)
return rc;
}
/* bug-1931: a task range searches its counts (sched_alloc_range_()) */
if ((ctx->primary->minTasks > 0)
&& (ctx->primary->minTasks < ctx->primary->maxTasks)) {
return sched_alloc_range_(iface, plugin, ctx);
}
/* Decision trace (4.1, 4.2, 4.8): the cluster-wide check's scope; its
* rows come from resource.c, its end says whether plans follow */
SCHED_TRACE(plugin->core, sched_trace_check_begin(plugin->core->trace_rec,
ctx->schedobj));
rc = sched_alloc_prepare_shared_(plugin->core, ctx);
rc = sched_alloc_prepare_shared_(plugin->core, ctx, 1);
SCHED_TRACE(plugin->core,
sched_trace_check_end(plugin->core->trace_rec, rc, rc > 0));
if (rc <= 0) {
@ -569,5 +704,5 @@ int sched_allocator_run(void* plugin_arg, xce_sched_ctx_t* ctx)
}
return sched_alloc_search_(iface, (xce_scheduler_t*)plugin->sched,
plugin->core, ctx);
plugin->core, ctx, 0, 1, 1);
}

13
src/scheduler/src/plugins/parjob_allocator.c
View file

@ -65,7 +65,8 @@ static int skip_bad_worker_(xce_sched_worker_t* w)
* before sched_allocator_run()'s plan/bucket loop ever runs -- ctx->nhosts/
* ctx->nslots have not been set by that loop yet at this point, so this
* branch computes its own job-wide scale (1 host, the job's full
* maxTasks) rather than reading them, generalizing seqjob's own
* maxTasks, or for a task range the count ctx->ntasks the runner is
* trying -- bug-1931) rather than reading them, generalizing seqjob's own
* self-contained "1 task total" literal to "maxTasks tasks total." This
* keeps the shared check independent of whichever plan/bucket-shape is
* later tried, per Req 4 AC 7 ("computed once per primary resreq...
@ -114,8 +115,11 @@ static int parjob_prepare_allocation(void* arg, xce_sched_ctx_t* ctx)
}
if (ctx->worker == NULL) {
/* bug-1931: a task-range attempt scales the job-level wants (per
* task shared resources, a global slots limit) to the count it is
* trying, set by the runner in ctx->ntasks */
ctx->nhosts = 1;
ctx->nslots = primary->maxTasks;
ctx->nslots = (ctx->ntasks > 0) ? ctx->ntasks : primary->maxTasks;
calc_wanted_global_resvals(&(ctx->alloc_req.shared), core->resmgr,
&(primary->rusages), PER_TASK, ctx->nhosts,
@ -163,7 +167,10 @@ static int parjob_prepare_allocation(void* arg, xce_sched_ctx_t* ctx)
* but never dereferenced -- plan SHAPE and ORDER never inspect
* core->workers or any other live worker/resource state. Only
* try_alloc_on_host()/host_iter.c (task 4.2's job) ever look at live
* worker state, to decide whether a generated plan succeeds.
* worker state, to decide whether a generated plan succeeds. For a task
* range (minTasks < maxTasks) the runner passes the request with its task
* count fixed to the count it is trying (bug-1931, alloc_runner.c), so
* "maxTasks" here is always the number of tasks to split.
*
* Partition rule for a given host-count k (requirements.md, "Data
* Structure Changes Summary", v1 default): the canonical BALANCED split

7
src/scheduler/src/repl_trace_render.c
View file

@ -602,6 +602,13 @@ static void put_entry(FILE* f, const sched_trace_entry_t* e,
/* stored one level below its host line (task 2.2) */
put_row(f, &e->u.row, false, units);
break;
case SCHED_TRACE_E_TASK_COUNT: /* bug-1931 */
fprintf(f, "%s %d%s", (e->u.alt.n == 3) ? "placing" : "task count",
e->u.alt.i, (e->u.alt.n == 1) ? " (minimum)"
: (e->u.alt.n == 2) ? " (maximum)"
: (e->u.alt.n == 3) ? " tasks"
: "");
break;
case SCHED_TRACE_E_NTYPES:
default:
fprintf(f, "unknown trace entry %d", (int)e->type);

11
src/scheduler/src/sched_trace.c
View file

@ -574,6 +574,17 @@ void sched_trace_check_end(xce_sched_trace_rec_t* rec, int rc,
e->u.check_end.plans_follow = plans_follow;
}
void sched_trace_task_count(xce_sched_trace_rec_t* rec, int n, int kind)
{
sched_trace_entry_t* e = sti_entry_add(rec, SCHED_TRACE_E_TASK_COUNT);
if (e == NULL) {
return;
}
e->u.alt.i = n;
e->u.alt.n = kind;
}
/* The NO_PLAN entry, or NULL when it is not stored (over the line bound,
* or the record is incomplete). */
static sched_trace_entry_t* no_plan_add(xce_sched_trace_rec_t* rec,

2
src/scheduler/src/schedobj.c
View file

@ -553,6 +553,8 @@ void sched_ctx_set_primary(xce_sched_ctx_t* ctx, xce_primary_resreq_t* primary)
* below compute a per-host want instead of this primary's job-level
* (shared) want */
ctx->worker = NULL;
/* bug-1931: a previous primary's task count is not this one's */
ctx->ntasks = 0;
/* scheduler-sim-live-allocation spec, task 2.1 (Core-Safe Allocation
* Chain): ctx->sched is now honestly typed xce_sched_core_t* (set by

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

@ -1869,6 +1869,15 @@ sched_parjob_combinatorial_test = executable(
dependencies: [scheduler_test_deps],
install: false)
# bug-1931: a parallel job's task range (minTasks..maxTasks) through the
# real runner and parjob -- minimum first, then the largest count that fits.
sched_parjob_task_range_test = executable(
'sched_parjob_task_range_test',
['test_parjob_task_range.c'],
include_directories:[incdir],
dependencies: [scheduler_test_deps],
install: false)
# Register as meson tests so `meson test --suite scheduler` runs them all.
# detect_leaks=0 matches the suite convention (fixtures are not torn down).
test('scheduler test', scheduler_test, suite: 'scheduler',
@ -2121,6 +2130,9 @@ test('scheduler seqjob rewrap equivalence test',
test('scheduler parjob combinatorial test',
sched_parjob_combinatorial_test, suite: 'scheduler',
env: ['ASAN_OPTIONS=detect_leaks=0'])
test('scheduler parjob task range test',
sched_parjob_task_range_test, suite: 'scheduler',
env: ['ASAN_OPTIONS=detect_leaks=0'])
test('scheduler cache dispatch test',
sched_cache_dispatch_test, suite: 'scheduler',
env: ['ASAN_OPTIONS=detect_leaks=0'])

2
src/scheduler/testsuite/test_alloc_equivalence_parjob.c
View file

@ -250,7 +250,7 @@ static void eq_primary_init(eq_primary_t* p, int max_tasks, float mem,
float lic, const eq_fixture_t* f, int rack)
{
memset(p, 0, sizeof(*p));
p->req.minTasks = (max_tasks > 0) ? 1 : 0;
p->req.minTasks = max_tasks; /* a fixed count (bug-1931) */
p->req.maxTasks = max_tasks;
xlist_init(&(p->req.rusages), 4);

2
src/scheduler/testsuite/test_alloc_runner.c
View file

@ -537,7 +537,7 @@ static void test_helpers_empty_shared_is_met(void)
TEST_ASSERT_EQUAL_INT(0, sched_alloc_ensure_reason_(&(f.ctx)));
TEST_ASSERT_NOT_NULL(f.ctx.reason);
TEST_ASSERT_EQUAL_INT(
1, sched_alloc_prepare_shared_(&(f.sched.core), &(f.ctx)));
1, sched_alloc_prepare_shared_(&(f.sched.core), &(f.ctx), 1));
rt_assert_no_shared_uncomit(&f);
rt_fixture_fini(&f);
}

2
src/scheduler/testsuite/test_alloc_runner_result.c
View file

@ -334,7 +334,7 @@ static void rr_fixture_init(rr_fixture_t* f)
f->plugin.core = &(f->sched.core);
f->plugin.allocator_iface = &g_iface;
f->req.minTasks = 1;
f->req.minTasks = 4; /* a fixed count (bug-1931) */
f->req.maxTasks = 4;
f->primary_obj.resreq = &(f->req);
f->obj.tag = XCE_RESREQ_PRIMARY;

2
src/scheduler/testsuite/test_alloc_runner_search.c
View file

@ -328,7 +328,7 @@ static xce_sched_worker_t* sr_add_host(sr_fixture_t* f, const char* name,
* after every sr_add_host(). */
static void sr_ctx_begin(sr_fixture_t* f, int max_tasks)
{
f->req.minTasks = 1;
f->req.minTasks = max_tasks; /* a fixed count (bug-1931) */
f->req.maxTasks = max_tasks;
f->primary_obj.resreq = &(f->req);
f->obj.tag = XCE_RESREQ_PRIMARY;

12
src/scheduler/testsuite/test_parjob_combinatorial.c
View file

@ -711,7 +711,7 @@ static void test_multi_bucket_plan_succeeds_through_real_runner(void)
= add_worker_with_slots(&(sched.core), "host-b", slots_res->idx, 1.0f);
memset(&primary, 0, sizeof(primary));
primary.minTasks = 1;
primary.minTasks = 3; /* a fixed count (bug-1931) */
primary.maxTasks = 3;
primary.select = NULL;
@ -776,7 +776,7 @@ static void test_allocator_reports_enomem_when_allocs_cannot_be_built(void)
add_worker_with_slots(&(sched.core), "host-b", slots_res->idx, 1.0f);
memset(&primary, 0, sizeof(primary));
primary.minTasks = 1;
primary.minTasks = 3; /* a fixed count (bug-1931) */
primary.maxTasks = 3;
primary.select = NULL;
@ -862,7 +862,7 @@ static void test_descending_bucket_order_succeeds_where_ascending_fails(void)
put_resvals(&(host_second->resvals), 1, idx, val);
memset(&primary, 0, sizeof(primary));
primary.minTasks = 1;
primary.minTasks = 3; /* a fixed count (bug-1931) */
primary.maxTasks = 3;
primary.select = NULL;
@ -992,7 +992,7 @@ test_mid_plan_bucket_failure_rolls_back_before_later_plan_succeeds(void)
h4 = add_worker_with_slots(&(sched.core), "h4", slots_res->idx, 1.0f);
memset(&primary, 0, sizeof(primary));
primary.minTasks = 1;
primary.minTasks = 4; /* a fixed count (bug-1931) */
primary.maxTasks = 4;
primary.select = NULL;
@ -1066,7 +1066,7 @@ static void test_returns_zero_when_every_plan_is_exhausted(void)
add_worker_with_slots(&(sched.core), "lonely-host", slots_res->idx, 1.0f);
memset(&primary, 0, sizeof(primary));
primary.minTasks = 1;
primary.minTasks = 3; /* a fixed count (bug-1931) */
primary.maxTasks = 3;
primary.select = NULL;
@ -1160,7 +1160,7 @@ static void test_shared_resource_debited_once_regardless_of_host_count(void)
}
memset(&primary, 0, sizeof(primary));
primary.minTasks = 1;
primary.minTasks = 4; /* a fixed count (bug-1931) */
primary.maxTasks = 4;
primary.select = NULL;

459
src/scheduler/testsuite/test_parjob_task_range.c Normal file
View file

@ -0,0 +1,459 @@
/*
* test_parjob_task_range.c - bug-1931: a parallel job's task range
* (minTasks..maxTasks, "-n min,max") through the real runner
* (sched_allocator_run(), alloc_runner.c) and the real parjob plan
* generator and prepare hook (plugins/parjob_allocator.c).
*
* Before bug-1931 the generator split maxTasks only, so a job whose
* minimum fit but whose maximum did not pended. The runner now checks
* minTasks first (pend after that one search when it does not fit), then
* searches upward for the largest task count whose plans fit, and places
* that. Plans stay today's balanced partitions: the expected counts below
* are the largest N whose BALANCED plan fits (capacity-driven shapes such
* as [4,1] belong to the parjob-placement spec).
*
* Same white-box stack and stubs as test_parjob_combinatorial.c; the
* plugin_prepare_allocation() stub also records the task counts it sees.
*/
#include <errno.h>
#include <pthread.h>
#include <rtti.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>
#include <unity/unity.h>
#include "preason.h"
#include "resobj.h"
#include "resreq.h"
#include "sched/alloc_iter.h"
#include "scheduler.h"
#include "xcealloc.h"
#include "../src/core.c"
#include "../src/alloc_runner.c"
#include "../src/host_iter.c"
#include "../src/plugins/parjob_allocator.c"
#include "../src/resource.c"
/* Hooked sources compile SCHED_TRACE bodies (scheduler-sim-decision-trace). */
#include "../src/sched_trace.c"
#include "../src/sched_trace_scan.c"
#include "../src/sched_trace_capture.c"
#include "../src/sched_trace_outcome.c"
#include "../src/schedobj.c"
/* ---------------------------------------------------------------------- */
/* Link-satisfying stubs (test_parjob_combinatorial.c convention) */
/* ---------------------------------------------------------------------- */
int plugin_register(void* s, void* p) { return (void)s, (void)p, 0; }
void plugin_unregister(void* arg) { (void)arg; }
int plugin_job_update(void* s, xce_sched_job_t* j)
{
return (void)s, (void)j, 0;
}
int plugin_queue_update(void* s, xce_sched_queue_t* q)
{
return (void)s, (void)q, 0;
}
int plugin_job_rusage_update(void* s, xce_sched_job_t* j, xce_job_rusages_t* r)
{
return (void)s, (void)j, (void)r, 0;
}
xce_alloc_t** sched_make_allocs(xce_scheduler_t* sched, char* reserve, int nres,
int* idxs, float* vals, int* duration)
{
xce_alloc_t** allocs;
int i;
(void)sched, (void)reserve, (void)idxs, (void)vals, (void)duration;
if (nres <= 0) {
return NULL;
}
allocs = (xce_alloc_t**)calloc(nres, sizeof(*allocs));
TEST_ASSERT_NOT_NULL(allocs);
for (i = 0; i < nres; i++) {
allocs[i] = (xce_alloc_t*)calloc(1, sizeof(xce_alloc_t));
}
return allocs;
}
/* Task counts seen by the prepare passes: the job-level pass's count
* (parjob sets ctx->nslots to it) and the largest per-host bucket. */
static int g_max_job_level_nslots = 0;
static int g_max_host_nslots = 0;
static int g_job_level_calls = 0;
/* plugin.c's fan-out, parjob's hook only; resets as plugin.c (bug-1883). */
int plugin_prepare_allocation(xce_sched_core_t* core, xce_sched_ctx_t* ctx)
{
xce_sched_plugin_t fake_plugin;
int rc;
if (ctx->worker == NULL) {
xvec_idval_fini(&(ctx->alloc_req.shared));
xvec_idval_init(&(ctx->alloc_req.shared), 8);
xvec_idval_fini(&(ctx->alloc_req.host));
xvec_idval_init(&(ctx->alloc_req.host), 8);
} else {
xvec_idval_fini(&(ctx->alloc_req.host));
xvec_idval_init(&(ctx->alloc_req.host), 8);
}
memset(&fake_plugin, 0, sizeof(fake_plugin));
fake_plugin.sched = core;
fake_plugin.core = core;
rc = parjob_prepare_allocation(&fake_plugin, ctx);
if (ctx->worker == NULL) {
g_job_level_calls++;
if (ctx->nslots > g_max_job_level_nslots) {
g_max_job_level_nslots = ctx->nslots;
}
} else if (ctx->nslots > g_max_host_nslots) {
g_max_host_nslots = ctx->nslots;
}
return rc;
}
/* ---------------------------------------------------------------------- */
/* Fixture */
/* ---------------------------------------------------------------------- */
typedef struct {
xce_scheduler_t sched;
xce_resource_t* slots;
xce_sched_obj_t obj;
xce_primary_schedobj_t primary_obj;
xce_primary_resreq_t primary;
xce_sched_ctx_t ctx;
xce_sched_worker_t* hosts[4];
int nhosts;
} range_fixture_t;
static void fixture_init(range_fixture_t* f, int min_tasks, int max_tasks)
{
xce_resource_t res;
memset(f, 0, sizeof(*f));
TEST_ASSERT_EQUAL_INT(0, xce_sched_core_init(&(f->sched.core)));
memset(&res, 0, sizeof(res));
res.name = strdup("slots");
f->slots = makeResource(f->sched.core.resmgr, &res);
TEST_ASSERT_NOT_NULL(f->slots);
f->primary.minTasks = min_tasks;
f->primary.maxTasks = max_tasks;
xlist_init(&(f->primary.rusages), 4);
f->primary_obj.resreq = &(f->primary);
f->obj.tag = XCE_RESREQ_PRIMARY;
f->obj.u.primary = &(f->primary_obj);
}
/* A real sched_add_worker() host with `free_slots` free "slots". */
static xce_sched_worker_t* fixture_add_host(range_fixture_t* f,
const char* name, float free_slots)
{
xce_sched_worker_t* w = (xce_sched_worker_t*)calloc(1, sizeof(*w));
int idx[1];
float val[1];
TEST_ASSERT_NOT_NULL(w);
w->name = strdup(name);
TEST_ASSERT_EQUAL_INT(0, zarray_init(&(w->resvals), 8, NULL));
TEST_ASSERT_EQUAL_INT(0, zarray_init(&(w->resInUse), 8, NULL));
idx[0] = f->slots->idx;
val[0] = free_slots;
put_resvals(&(w->resvals), 1, idx, val);
val[0] = 0.0f;
put_resvals(&(w->resInUse), 1, idx, val);
TEST_ASSERT_TRUE(sched_add_worker(&(f->sched.core), w) >= 0);
f->hosts[f->nhosts++] = w;
return w;
}
/* The production sequence of one attempt (sched_make_primary_
* allocation()): ctx init, set_primary (the job-level prepare pass),
* then parjob's entry point, the framework runner. The recorded task
* counts are the runner's: set_primary's own pass (at maxTasks) is
* counted only for a fixed count, where the runner reuses it. */
static int fixture_run(range_fixture_t* f)
{
xce_sched_plugin_t plugin;
g_max_job_level_nslots = 0;
g_max_host_nslots = 0;
g_job_level_calls = 0;
TEST_ASSERT_EQUAL_INT(
0, sched_ctx_init(&(f->ctx), &(f->sched.core), &(f->obj)));
sched_ctx_set_primary(&(f->ctx), &(f->primary));
if (f->primary.minTasks < f->primary.maxTasks) {
g_max_job_level_nslots = 0;
g_job_level_calls = 0;
}
memset(&plugin, 0, sizeof(plugin));
plugin.sched = &(f->sched);
plugin.core = &(f->sched.core);
plugin.allocator_iface = &parjob_alloc_iface;
return sched_allocator_run(&plugin, &(f->ctx));
}
static void fixture_fini(range_fixture_t* f)
{
sched_ctx_fini(&(f->ctx));
xlist_fini(&(f->primary.rusages));
}
/* Slots the attempt debited on `w` (ctx->uncomitted), 0 when none. */
static float host_debit(range_fixture_t* f, xce_sched_worker_t* w)
{
xvec_t* v = NULL;
if ((sched_ctx_get_uncomit_host(&v, &(f->ctx), w->idx) != 1)
|| (v == NULL) || ((int)v->count <= f->slots->idx)) {
return 0.0f;
}
return ((float*)v->data)[f->slots->idx];
}
/* Total tasks placed: the slots debited across every host. */
static float placed_tasks(range_fixture_t* f)
{
float total = 0.0f;
int i;
for (i = 0; i < f->nhosts; i++) {
total += host_debit(f, f->hosts[i]);
}
return total;
}
/* ---------------------------------------------------------------------- */
/* Tests */
/* ---------------------------------------------------------------------- */
/* The reported case: tasks 2..5, one host with 2 free slots (pc13v with
* 22 total, 20 in use). The job starts with 2 tasks, not pends. */
static void test_range_places_minimum_when_only_minimum_fits(void)
{
range_fixture_t f;
fixture_init(&f, 2, 5);
fixture_add_host(&f, "pc13v", 2.0f);
TEST_ASSERT_EQUAL_INT_MESSAGE(1, fixture_run(&f),
"a 2..5 job must start when 2 slots are free");
TEST_ASSERT_NOT_NULL(f.ctx.alloc);
TEST_ASSERT_EQUAL_INT(1, f.ctx.alloc->nhosts);
TEST_ASSERT_EQUAL_FLOAT(2.0f, host_debit(&f, f.hosts[0]));
fixture_fini(&f);
}
/* Min-first: when minTasks does not fit, no larger count is tried -- no
* job-level pass and no host check above minTasks -- and the job pends
* with the host's shortfall recorded. */
static void test_range_minimum_unfit_pends_without_trying_larger_counts(void)
{
range_fixture_t f;
fixture_init(&f, 2, 5);
fixture_add_host(&f, "pc13v", 1.0f);
TEST_ASSERT_EQUAL_INT(0, fixture_run(&f));
TEST_ASSERT_NULL(f.ctx.alloc);
TEST_ASSERT_EQUAL_INT_MESSAGE(2, g_max_job_level_nslots,
"only the minimum's job-level pass may run");
TEST_ASSERT_TRUE_MESSAGE(g_max_host_nslots <= 2,
"no host may be checked for more than minTasks");
TEST_ASSERT_NOT_NULL(f.ctx.reason);
TEST_ASSERT_TRUE(f.ctx.reason->reasons->nres_reasons > 0);
TEST_ASSERT_EQUAL_FLOAT(0.0f, placed_tasks(&f));
fixture_fini(&f);
}
/* Largest fitting count: hosts with 3, 3 and 1 free, tasks 2..9. 9, 8
* and 7 have no balanced plan that fits ([3,2,2] needs 2 slots on the
* third host); 6 fits as [3,3]. */
static void test_range_places_largest_count_whose_balanced_plan_fits(void)
{
range_fixture_t f;
fixture_init(&f, 2, 9);
fixture_add_host(&f, "host-a", 3.0f);
fixture_add_host(&f, "host-b", 3.0f);
fixture_add_host(&f, "host-c", 1.0f);
TEST_ASSERT_EQUAL_INT(1, fixture_run(&f));
TEST_ASSERT_NOT_NULL(f.ctx.alloc);
TEST_ASSERT_EQUAL_INT(2, f.ctx.alloc->nhosts);
TEST_ASSERT_EQUAL_FLOAT(6.0f, placed_tasks(&f));
TEST_ASSERT_EQUAL_FLOAT(3.0f, host_debit(&f, f.hosts[0]));
TEST_ASSERT_EQUAL_FLOAT(3.0f, host_debit(&f, f.hosts[1]));
TEST_ASSERT_EQUAL_FLOAT(0.0f, host_debit(&f, f.hosts[2]));
fixture_fini(&f);
}
/* The maximum is placed whenever it fits. */
static void test_range_places_maximum_when_it_fits(void)
{
range_fixture_t f;
fixture_init(&f, 2, 8);
fixture_add_host(&f, "host-a", 4.0f);
fixture_add_host(&f, "host-b", 4.0f);
TEST_ASSERT_EQUAL_INT(1, fixture_run(&f));
TEST_ASSERT_EQUAL_FLOAT(8.0f, placed_tasks(&f));
TEST_ASSERT_EQUAL_FLOAT(4.0f, host_debit(&f, f.hosts[0]));
TEST_ASSERT_EQUAL_FLOAT(4.0f, host_debit(&f, f.hosts[1]));
fixture_fini(&f);
}
/* minTasks == 1: the 1-task count still asks the host for its slot (the
* per-host pass keys "is this a sequential job" off the job's maxTasks,
* not the count being placed). */
static void test_range_with_minimum_one_places_one_task(void)
{
range_fixture_t f;
fixture_init(&f, 1, 3);
fixture_add_host(&f, "host-a", 1.0f);
TEST_ASSERT_EQUAL_INT(1, fixture_run(&f));
TEST_ASSERT_EQUAL_FLOAT(1.0f, host_debit(&f, f.hosts[0]));
fixture_fini(&f);
}
/* A per-task shared (cluster-wide) resource scales with the count being
* placed (finding 4a): 4 licenses, 1 per task, tasks 2..8 on a 10-slot
* host -> 4 tasks, and exactly 4 licenses debited. */
static void test_range_shared_per_task_resource_scales_with_placed_count(void)
{
range_fixture_t f;
xce_resource_t res;
xce_resource_t* lic;
rusage_value_t lic_rusage;
xvec_t* shared = NULL;
int idx[1];
float val[1];
fixture_init(&f, 2, 8);
fixture_add_host(&f, "host-a", 10.0f);
memset(&res, 0, sizeof(res));
res.name = strdup("licenses");
res.flags = XCE_RESF_GLOBAL;
lic = makeResource(f.sched.core.resmgr, &res);
TEST_ASSERT_NOT_NULL(lic);
idx[0] = lic->idx;
val[0] = 4.0f;
put_resvals(&(f.sched.core.sharedInst), 1, idx, val);
memset(&lic_rusage, 0, sizeof(lic_rusage));
lic_rusage.res = strdup("licenses");
lic_rusage.value = 1.0f;
lic_rusage.per_flag = PER_TASK;
xlist_store(&(f.primary.rusages), &lic_rusage);
TEST_ASSERT_EQUAL_INT_MESSAGE(1, fixture_run(&f),
"4 licenses hold 4 of the 2..8 tasks");
TEST_ASSERT_EQUAL_FLOAT(4.0f, placed_tasks(&f));
TEST_ASSERT_EQUAL_INT(1, sched_ctx_get_uncomit_shared(&shared, &(f.ctx)));
TEST_ASSERT_NOT_NULL(shared);
TEST_ASSERT_EQUAL_FLOAT(4.0f, ((float*)shared->data)[lic->idx]);
fixture_fini(&f);
FREEUP(lic_rusage.res);
}
/* The maximum passes the cluster-wide check (100 licenses) but no plan
* places it on a 3-slot host: its shared debit is given back, and only
* the placed count's (3) stays. */
static void test_range_unplaced_maximum_leaves_no_shared_debit(void)
{
range_fixture_t f;
xce_resource_t res;
xce_resource_t* lic;
rusage_value_t lic_rusage;
xvec_t* shared = NULL;
int idx[1];
float val[1];
fixture_init(&f, 2, 5);
fixture_add_host(&f, "host-a", 3.0f);
memset(&res, 0, sizeof(res));
res.name = strdup("licenses");
res.flags = XCE_RESF_GLOBAL;
lic = makeResource(f.sched.core.resmgr, &res);
TEST_ASSERT_NOT_NULL(lic);
idx[0] = lic->idx;
val[0] = 100.0f;
put_resvals(&(f.sched.core.sharedInst), 1, idx, val);
memset(&lic_rusage, 0, sizeof(lic_rusage));
lic_rusage.res = strdup("licenses");
lic_rusage.value = 1.0f;
lic_rusage.per_flag = PER_TASK;
xlist_store(&(f.primary.rusages), &lic_rusage);
TEST_ASSERT_EQUAL_INT(1, fixture_run(&f));
TEST_ASSERT_EQUAL_FLOAT(3.0f, placed_tasks(&f));
TEST_ASSERT_EQUAL_INT(1, sched_ctx_get_uncomit_shared(&shared, &(f.ctx)));
TEST_ASSERT_NOT_NULL(shared);
TEST_ASSERT_EQUAL_FLOAT(3.0f, ((float*)shared->data)[lic->idx]);
fixture_fini(&f);
FREEUP(lic_rusage.res);
}
/* A fixed task count (min == max) keeps today's path: one job-level pass
* at maxTasks and today's balanced plans. */
static void test_fixed_count_keeps_single_job_level_pass(void)
{
range_fixture_t f;
fixture_init(&f, 4, 4);
fixture_add_host(&f, "host-a", 2.0f);
fixture_add_host(&f, "host-b", 2.0f);
TEST_ASSERT_EQUAL_INT(1, fixture_run(&f));
TEST_ASSERT_EQUAL_INT(1, g_job_level_calls);
TEST_ASSERT_EQUAL_FLOAT(2.0f, host_debit(&f, f.hosts[0]));
TEST_ASSERT_EQUAL_FLOAT(2.0f, host_debit(&f, f.hosts[1]));
fixture_fini(&f);
}
void setUp(void) { return; }
void tearDown(void) { return; }
int main(void)
{
UNITY_BEGIN();
RUN_TEST(test_range_places_minimum_when_only_minimum_fits);
RUN_TEST(test_range_minimum_unfit_pends_without_trying_larger_counts);
RUN_TEST(test_range_places_largest_count_whose_balanced_plan_fits);
RUN_TEST(test_range_places_maximum_when_it_fits);
RUN_TEST(test_range_with_minimum_one_places_one_task);
RUN_TEST(test_range_shared_per_task_resource_scales_with_placed_count);
RUN_TEST(test_range_unplaced_maximum_leaves_no_shared_debit);
RUN_TEST(test_fixed_count_keeps_single_job_level_pass);
return UNITY_END();
}

21
src/scheduler/testsuite/test_repl_trace_e2e.c
View file

@ -288,26 +288,35 @@ static void test_step_mode_full_all_jobs_and_file_copy(void)
"reported with each host's verdict\n",
"\n requirement single tasks 2..4 requests mem=1024.000000 "
"(1.00 KB)\n",
/* bug-1931: the task range checks its minimum first (a probe
* that gives its host back), then places its maximum */
"\n task count 2 (minimum)\n",
"\n cluster-wide limit cluster_slots on slots: value 6 usage 0"
" reserved 0 requested 2 passed\n",
"\n plan 1 hosts 1 tasks per position [2]\n",
"\n plan 1 succeeded\n"
" released hosts: w01\n"
" task count 4 (maximum)\n",
"\n cluster-wide limit cluster_slots on slots: value 6 usage 0"
" reserved 0 requested 4 passed\n",
"\n cluster-wide check passed\n",
"\n plan 1 hosts 1 tasks per position [4]\n",
"\n plan 2 hosts 1 tasks per position [4]\n",
"\n position 1 (4 tasks)\n",
"\n host w01 rejected (per-host check): limit exceeded\n"
" limit host_slots (h=w01) on slots: value 3 usage 0 "
"reserved 0 requested 4\n",
"\n host w02 rejected (per-host check): limit exceeded\n",
"\n position 1 not filled: no host accepted\n",
"\n plan 1 failed: position 1 could not be filled\n",
"\n plan 2 hosts 2 tasks per position [2, 2]\n",
"\n plan 2 failed: position 1 could not be filled\n",
"\n plan 3 hosts 2 tasks per position [2, 2]\n",
"\n host w01 accepted\n",
"\n position 1 filled by w01\n",
"\n host w01 skipped: already claimed by an earlier "
"position\n",
"\n position 2 filled by w02\n",
"\n plan 2 succeeded\n",
"\n plan 2 committed; no further plans tried\n",
"\n plans: 2 generated, 2 tried, 0 discarded, 0 not tried\n",
"\n plan 3 succeeded\n",
"\n plan 3 committed; no further plans tried\n",
"\n plans: 3 generated, 3 tried, 0 discarded, 0 not tried\n",
"\n single requirement satisfied\n",
"\n outcome: DISPATCH\n",
"\n host w01: mem=2048.000000 (2.00 KB) slots=2 limit charges: "

2
src/scheduler/testsuite/test_trace_bounds.c
View file

@ -193,7 +193,7 @@ static void start_sched(void)
j->submit->user = (char*)"bounds_user";
j->submit->project_name = (char*)"bounds_proj";
j->submit->resreq = (char*)"rusage[mem=1]";
j->submit->mintasks = 1;
j->submit->mintasks = MAX_TASKS; /* a fixed count (bug-1931) */
j->submit->maxtasks = MAX_TASKS;
jobs[0] = j;
memset(&ev, 0, sizeof(ev));

2
src/scheduler/testsuite/test_trace_hooks_runner.c
View file

@ -412,7 +412,7 @@ static void rt_init(rt_fixture_t* f, const rt_scenario_t* sc, bool traced)
rt_add_host(f, &(sc->hosts[i]));
}
f->req.minTasks = 1;
f->req.minTasks = sc->max_tasks; /* a fixed count (bug-1931) */
f->req.maxTasks = sc->max_tasks;
if (sc->select_rack > 0) {
/* select `rack == select_rack` */

3
src/scheduler/testsuite/test_trace_model.c
View file

@ -1109,6 +1109,7 @@ static int emit_every_kind(xce_sched_trace_rec_t* rec)
sched_trace_check_begin(rec, NULL);
sched_trace_row(rec, &row); /* 4 */
sched_trace_check_end(rec, 1, true); /* 5 */
sched_trace_task_count(rec, 2, 1); /* 5a, bug-1931 */
sched_trace_no_plan(rec, XCE_TRACE_NOPLAN_DECLINED); /* 6 */
sched_trace_plan_discarded(rec, &plan); /* 7 */
sched_trace_plan_not_tried(rec, &plan, -12); /* 8 */
@ -1131,7 +1132,7 @@ static int emit_every_kind(xce_sched_trace_rec_t* rec)
sched_trace_resreq_end(rec, 0, 0); /* 22 */
sched_trace_path_pop(rec);
sched_trace_alt_result(rec, 0, 2); /* 23 */
return 23;
return 24;
}
static void test_every_entry_kind_is_one_entry(void)

44
src/scheduler/testsuite/test_trace_render.c
View file

@ -1066,6 +1066,49 @@ static void test_known_unsatisfiable_names_source_job(void)
sched_trace_rec_free(rec);
}
/* bug-1931: a task range names each count it tries -- the minimum, the
* maximum, a count between them, and the count it places. */
static void test_task_range_counts(void)
{
static const char* const rn[] = { "slots" };
static const float rv[] = { 1.0f };
xce_sched_trace_rec_t* rec;
fx_primary_t p;
char* out;
rec = new_rec(XCE_TRACE_LEVEL_FULL, XCE_TRACE_LINE_BOUND_DEFAULT,
XCE_TRACE_REC_DECISION);
TEST_ASSERT_NOT_NULL(rec);
set_selection(rec);
set_glimits(rec, 0, NULL, NULL);
make_primary(&p, 2, 9, 1, rn, rv, NULL);
sched_trace_path_push(rec, XCE_TRACE_PATH_SINGLE, 1, 1);
sched_trace_resreq_begin(rec, &p.p, 0);
sched_trace_task_count(rec, 2, 1);
sched_trace_task_count(rec, 9, 2);
sched_trace_task_count(rec, 5, 0);
sched_trace_task_count(rec, 5, 3);
sched_trace_resreq_end(rec, 0, 0);
sched_trace_path_pop(rec);
sched_trace_rec_finish(rec, XCE_TRACE_REC_PEND_NO_REASON, 0);
out = render(rec, XCE_REPL_UNITS_RAW);
assert_golden(
"[trace] job 1-2-1000-42 session 7 position 3 level full\n"
SEL_LINE NO_LIMITS_LINE
" requirement single tasks 2..9 requests slots=1\n"
" task count 2 (minimum)\n"
" task count 9 (maximum)\n"
" task count 5\n"
" placing 5 tasks\n"
" single requirement not satisfied\n"
" no decision: job pended without any pend reason\n",
out);
free(out);
sched_trace_rec_free(rec);
}
static void test_reconstructed_record(void)
{
xce_sched_trace_rec_t* rec;
@ -1364,6 +1407,7 @@ int main(void)
RUN_TEST(test_no_decision_internal_error_keeps_hierarchy);
RUN_TEST(test_no_decision_pend_without_reason_keeps_hierarchy);
RUN_TEST(test_known_unsatisfiable_names_source_job);
RUN_TEST(test_task_range_counts);
RUN_TEST(test_reconstructed_record);
RUN_TEST(test_selection_and_limits_unavailable);
RUN_TEST(test_entry_kinds_and_edge_cases);