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

feat(resreq-maxtasks-wiring): add task-count derivation helper

Browse source 
Adds apply_job_task_count() to overwrite a standalone primary resreq's
minTasks/maxTasks with a job's real requested task count, no-op for
COMPOUND/ALTER nested primaries, with a defensive min>max clamp.
Closes the gap that lets scheduler-allocation's multi-host allocator
ever be reached (bug-1770). Wiring at live ingest call sites follows
in subsequent tasks.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01XYdtrEWWZfKRRGED7raEYg
This commit is contained in:
Rongsong Shen 2026-09-06 03:26:32 +00:00
parent ecfb0a3c64
commit 799a71ecb1
No known key found for this signature in database
GPG key ID: 5BCDAAE0123E9FE2
10 changed files with 918 additions and 0 deletions
Download patch file Download diff file Expand all files Collapse all files

12
src/include/resreq.h
View file

@ -316,4 +316,16 @@ extern int calc_wanted_global_resvals(xvec_idval_t*, var, xlist_t*,
extern int calc_wanted_host_resvals(xvec_idval_t*, var, xlist_t*,
rusage_per_type_t, int, int);
/*
* Overwrite minTasks/maxTasks on resreq's own top-level primary with the
* job's real requested task count. No-op if resreq is NULL or if its
* top-level tag is not XCE_RESREQ_PRIMARY -- a COMPOUND or ALTER resreq's
* nested primaries are left exactly as parse_resreq_primary() set them.
* If mintasks > maxtasks, maxtasks is defensively raised to mintasks
* before assignment (mirrors the apiserver's own convert.go clamp
* direction). Idempotent; never allocates memory.
*/
extern void apply_job_task_count(xce_resreq_t* resreq, uint32_t mintasks,
uint32_t maxtasks);
#endif

1
src/lib/meson.build
View file

@ -160,6 +160,7 @@ libxce_test_src = ['testsuite/test.c',
'testsuite/test_xvec.c',
'testsuite/test_resreq.c',
'testsuite/test_resreq_fingerprint.c',
'testsuite/test_resreq_task_count.c',
'testsuite/test_resmgr_fixture.c',
'testsuite/test_select.c',
'testsuite/test_bitset.c',

34
src/lib/resreq.c
View file

@ -482,6 +482,40 @@ int parse_resreq(xce_resreq_t* resreq, char* data, var resmgr, char** errmsg)
return rc;
}
/*
* TaskCountWiring (resreq-maxtasks-wiring spec): overwrite a standalone
* primary resreq's minTasks/maxTasks with the job's real requested task
* count. A COMPOUND/ALTER resreq's nested primaries are left exactly as
* parse_resreq_primary() set them -- only the job's own top-level
* standalone primary is in scope (design.md's TaskCountWiring component;
* requirements 1.1, 1.2, 1.5, 3.3).
*
* Narrowing conversion note: mintasks/maxtasks are uint32_t, matching
* job_submit_data_t.mintasks/maxtasks (the caller's own source of truth);
* xce_primary_resreq_t.minTasks/maxTasks are plain int. This is a
* deliberate, one-time, acknowledged narrowing assignment -- safe for
* realistic task counts (well under INT_MAX) -- not a mismatch to "fix"
* by changing either type.
*/
void apply_job_task_count(xce_resreq_t* resreq, uint32_t mintasks,
uint32_t maxtasks)
{
if (resreq == NULL || resreq->tag != XCE_RESREQ_PRIMARY) {
return;
}
/* Defensive clamp: mirror convert.go's own clamp direction -- raise
* the max, never lower the min. The apiserver already guarantees
* mintasks <= maxtasks twice before a job reaches this layer; this is
* a redundant safety net, not a new job-rejecting error path. */
if (mintasks > maxtasks) {
maxtasks = mintasks;
}
resreq->u.primary->minTasks = (int)mintasks;
resreq->u.primary->maxTasks = (int)maxtasks;
}
void kill_primary_resreq(xce_primary_resreq_t* resreq)
{
int i;

2
src/lib/testsuite/test.c
View file

@ -23,6 +23,7 @@ extern void test_limit(void);
extern void test_xvec(void);
extern void test_resreq(void);
extern void test_resreq_fingerprint(void);
extern void test_resreq_task_count(void);
extern void test_select(void);
extern void test_bitset(void);
extern void test_kv(void);
@ -147,6 +148,7 @@ int main(int argc, char* argv[])
TEST(test_xvec);
TEST(test_resreq);
TEST(test_resreq_fingerprint);
TEST(test_resreq_task_count);
TEST(test_select);
TEST(test_bitset);
TEST(test_kv);

262
src/lib/testsuite/test_resreq_task_count.c Normal file
View file

@ -0,0 +1,262 @@
/* test_resreq_task_count.c — resreq-maxtasks-wiring spec, task 1
* (TaskCountWiring component).
*
* Unit tests for apply_job_task_count() (src/lib/resreq.c), the pure
* helper that overwrites a standalone primary resreq's minTasks/maxTasks
* with the job's real requested task count, per design.md's
* TaskCountWiring component. Covers the three cases the task requires:
* - a standalone primary resreq receives the applied task count
* (Requirements 1.1, 1.2)
* - a COMPOUND/ALTER resreq's nested primaries are left untouched
* (Requirement 3.3)
* - an invalid (minTasks > maxTasks) pair is defensively clamped by
* raising maxTasks to minTasks, mirroring convert.go's clamp
* direction (Requirement 1.5)
*/
#include <stdint.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unity/unity.h>
#include "resreq.h"
#include "utils.h"
/* ── standalone primary: overwritten with the requested task count ─────── */
void test_apply_job_task_count_standalone_primary_override(void)
{
char* resreqString = "rusage[mem=20]";
char* errmsg = NULL;
int rc = 0;
xce_resreq_t resreq;
memset(&resreq, 0x0, sizeof(resreq));
rc = parse_resreq(&resreq, resreqString, NULL, &errmsg);
TEST_ASSERT_TRUE(rc >= 0);
TEST_ASSERT_TRUE(resreq.tag == XCE_RESREQ_PRIMARY);
/* Sanity: parser's own hardcoded default before the helper ever runs. */
TEST_ASSERT_EQUAL_INT(1, resreq.u.primary->minTasks);
TEST_ASSERT_EQUAL_INT(1, resreq.u.primary->maxTasks);
apply_job_task_count(&resreq, 4, 6);
TEST_ASSERT_EQUAL_INT(4, resreq.u.primary->minTasks);
TEST_ASSERT_EQUAL_INT(6, resreq.u.primary->maxTasks);
/* Idempotency: calling again with identical inputs must not change
* the result (design.md's TaskCountWiring invariant). */
apply_job_task_count(&resreq, 4, 6);
TEST_ASSERT_EQUAL_INT(4, resreq.u.primary->minTasks);
TEST_ASSERT_EQUAL_INT(6, resreq.u.primary->maxTasks);
kill_resreq(&resreq);
FREEUP(errmsg);
}
/* A default/single-task request (Requirement 1.2) must still land on 1/1 --
* exercised explicitly rather than merely relying on the parser's own
* pre-helper default, since the helper is expected to be applied
* unconditionally at every live ingest call site regardless of the
* requested count. */
void test_apply_job_task_count_default_single_task(void)
{
char* resreqString = "rusage[mem=20]";
char* errmsg = NULL;
int rc = 0;
xce_resreq_t resreq;
memset(&resreq, 0x0, sizeof(resreq));
rc = parse_resreq(&resreq, resreqString, NULL, &errmsg);
TEST_ASSERT_TRUE(rc >= 0);
TEST_ASSERT_TRUE(resreq.tag == XCE_RESREQ_PRIMARY);
apply_job_task_count(&resreq, 1, 1);
TEST_ASSERT_EQUAL_INT(1, resreq.u.primary->minTasks);
TEST_ASSERT_EQUAL_INT(1, resreq.u.primary->maxTasks);
kill_resreq(&resreq);
FREEUP(errmsg);
}
/* ── COMPOUND/ALTER: nested primaries left exactly as parsed (no-op) ────── */
void test_apply_job_task_count_compound_noop(void)
{
char* resreqString = "2*{rusage[mem=100:duration=20:decay=1]} + "
"4*{rusage[mem=80:duration=30:decay=1]}";
char* errmsg = NULL;
int rc = 0;
xce_resreq_t resreq;
int count;
int before_min[8];
int before_max[8];
int i;
memset(&resreq, 0x0, sizeof(resreq));
rc = parse_resreq(&resreq, resreqString, NULL, &errmsg);
TEST_ASSERT_TRUE(rc >= 0);
TEST_ASSERT_TRUE(resreq.tag == XCE_RESREQ_COMPOUND);
count = resreq.u.compound->primaries.count;
TEST_ASSERT_TRUE(count > 0);
TEST_ASSERT_TRUE(count <= 8);
/* Snapshot whatever parse_resreq() actually produced for each nested
* primary -- the "N*{...}" compound syntax sets a nested primary's
* minTasks/maxTasks to N (not the bare-rusage default of 1), so the
* no-op guarantee is "unchanged from parse", not "equal to 1". */
for (i = 0; i < count; i++) {
xce_primary_resreq_t* p = (xce_primary_resreq_t*)xlist_get(
&(resreq.u.compound->primaries), i);
TEST_ASSERT_NOT_NULL(p);
before_min[i] = p->minTasks;
before_max[i] = p->maxTasks;
}
apply_job_task_count(&resreq, 4, 6);
/* No-op: tag unchanged, every nested primary left exactly as
* parse_resreq() set it. Only the job's own top-level standalone
* primary is in this helper's scope (Requirement 3.3). */
TEST_ASSERT_TRUE(resreq.tag == XCE_RESREQ_COMPOUND);
for (i = 0; i < count; i++) {
xce_primary_resreq_t* p = (xce_primary_resreq_t*)xlist_get(
&(resreq.u.compound->primaries), i);
TEST_ASSERT_NOT_NULL(p);
TEST_ASSERT_EQUAL_INT(before_min[i], p->minTasks);
TEST_ASSERT_EQUAL_INT(before_max[i], p->maxTasks);
}
kill_resreq(&resreq);
FREEUP(errmsg);
}
void test_apply_job_task_count_alter_noop(void)
{
char* resreqString
= "{2*{rusage[mem=100:duration=20:decay=1]} + "
"4*{rusage[mem=80:duration=30:decay=1]}} || "
"{3*{rusage[mem=150:duration=15:decay=1]} + "
"6*{rusage[mem=120:duration=20:decay=1]}} ";
char* errmsg = NULL;
int rc = 0;
xce_resreq_t resreq;
int ncompounds;
int before_min[8][8];
int before_max[8][8];
int nprimaries[8];
int i;
memset(&resreq, 0x0, sizeof(resreq));
rc = parse_resreq(&resreq, resreqString, NULL, &errmsg);
TEST_ASSERT_TRUE(rc >= 0);
TEST_ASSERT_TRUE(resreq.tag == XCE_RESREQ_ALTER);
ncompounds = resreq.u.alter->compounds.count;
TEST_ASSERT_TRUE(ncompounds > 0);
TEST_ASSERT_TRUE(ncompounds <= 8);
/* Snapshot whatever parse_resreq() actually produced for every nested
* primary across every compound branch -- the "N*{...}" compound
* syntax sets a nested primary's minTasks/maxTasks to N (not the
* bare-rusage default of 1), so the no-op guarantee is "unchanged
* from parse", not "equal to 1". */
for (i = 0; i < ncompounds; i++) {
xce_compound_resreq_t* c
= (xce_compound_resreq_t*)xlist_get(&(resreq.u.alter->compounds), i);
int j;
TEST_ASSERT_NOT_NULL(c);
nprimaries[i] = c->primaries.count;
TEST_ASSERT_TRUE(nprimaries[i] <= 8);
for (j = 0; j < nprimaries[i]; j++) {
xce_primary_resreq_t* p
= (xce_primary_resreq_t*)xlist_get(&(c->primaries), j);
TEST_ASSERT_NOT_NULL(p);
before_min[i][j] = p->minTasks;
before_max[i][j] = p->maxTasks;
}
}
apply_job_task_count(&resreq, 4, 6);
/* No-op: tag unchanged, every nested primary left exactly as
* parse_resreq() set it (Requirement 3.3). */
TEST_ASSERT_TRUE(resreq.tag == XCE_RESREQ_ALTER);
for (i = 0; i < ncompounds; i++) {
xce_compound_resreq_t* c
= (xce_compound_resreq_t*)xlist_get(&(resreq.u.alter->compounds), i);
int j;
TEST_ASSERT_NOT_NULL(c);
for (j = 0; j < nprimaries[i]; j++) {
xce_primary_resreq_t* p
= (xce_primary_resreq_t*)xlist_get(&(c->primaries), j);
TEST_ASSERT_NOT_NULL(p);
TEST_ASSERT_EQUAL_INT(before_min[i][j], p->minTasks);
TEST_ASSERT_EQUAL_INT(before_max[i][j], p->maxTasks);
}
}
kill_resreq(&resreq);
FREEUP(errmsg);
}
/* ── defensive clamp: minTasks > maxTasks must not reach dispatch ───────── */
void test_apply_job_task_count_clamps_min_exceeds_max(void)
{
char* resreqString = "rusage[mem=20]";
char* errmsg = NULL;
int rc = 0;
xce_resreq_t resreq;
memset(&resreq, 0x0, sizeof(resreq));
rc = parse_resreq(&resreq, resreqString, NULL, &errmsg);
TEST_ASSERT_TRUE(rc >= 0);
TEST_ASSERT_TRUE(resreq.tag == XCE_RESREQ_PRIMARY);
/* mintasks (6) > maxtasks (4): defensively raise maxtasks to
* mintasks, mirroring convert.go's own clamp direction exactly. */
apply_job_task_count(&resreq, 6, 4);
TEST_ASSERT_EQUAL_INT(6, resreq.u.primary->minTasks);
TEST_ASSERT_EQUAL_INT(6, resreq.u.primary->maxTasks);
kill_resreq(&resreq);
FREEUP(errmsg);
}
/* ── defensive: NULL resreq must be a safe no-op, never a crash ─────────── */
void test_apply_job_task_count_null_resreq_is_noop(void)
{
apply_job_task_count(NULL, 4, 6);
}
void test_resreq_task_count(void)
{
printf("\n>> resreq task count test\n\n");
RUN_TEST(test_apply_job_task_count_standalone_primary_override);
RUN_TEST(test_apply_job_task_count_default_single_task);
RUN_TEST(test_apply_job_task_count_compound_noop);
RUN_TEST(test_apply_job_task_count_alter_noop);
RUN_TEST(test_apply_job_task_count_clamps_min_exceeds_max);
RUN_TEST(test_apply_job_task_count_null_resreq_is_noop);
}