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

fix(worker): real maxswap/maxtmp detection; feat(scheduler): annotate multi-slot assigned jobs

Browse source 
bug-1554: init_host_attrs() replaces hardcoded placeholder stubs with real
detection -- maxswap from sysinfo()'s totalswap (reusing the same syscall
snapshot already taken for maxmem), maxtmp from a fresh statvfs("/tmp")
total-capacity measurement. No synthetic fallback on detection failure, so
a system with swap genuinely disabled correctly reports 0.

bug-1776: the scheduler simulator's "info worker" assigned-jobs listing now
annotates a job consuming more than one "slots" unit (e.g. "jobid*2"),
using per-host allocation data that was already being walked but discarded.
Falls back to the pre-existing per-job listing when jobs on the same worker
have differing slot counts, since a single suffix on a compressed range
would be ambiguous.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01Hb6Edtj4KbS8CrgEWYdA2E
This commit is contained in:
Rongsong Shen 2026-09-07 04:39:22 +00:00
parent a6b735b6f2
commit a35f597e06
No known key found for this signature in database
GPG key ID: 5BCDAAE0123E9FE2
9 changed files with 660 additions and 28 deletions
Download patch file Download diff file Expand all files Collapse all files

81
src/scheduler/src/repl_info.c
View file

@ -445,6 +445,42 @@ static bool repl_append_u64(uint64_t** arr, int* n, uint64_t val)
return true;
}
/* repl_append_int - int-array counterpart to repl_append_u64() above, same
* exact-size realloc idiom and same degrade-gracefully-on-OOM contract. */
static bool repl_append_int(int** arr, int* n, int val)
{
int* grown = (int*)realloc(*arr, (size_t)(*n + 1) * sizeof(**arr));
if (grown == NULL) {
return false;
}
grown[*n] = val;
*arr = grown;
(*n)++;
return true;
}
/* repl_worker_alloc_nslots - how many "slots" this per-host allocation
* consumed, so the assigned-jobs listing can annotate a multi-slot job
* (e.g. "jobid*2") instead of implying one slot per job unconditionally.
* Mirrors the controller's own get_nslots_in_alloc() (job.c) -- same
* resname scan, same "no explicit slots entry -> 1" default -- duplicated
* here rather than shared since that helper is file-static to job.c and
* this is a read-only REPL display concern, not scheduling logic. */
static int repl_worker_alloc_nslots(xce_host_alloc_t* ha)
{
int i;
for (i = 0; i < ha->nallocs; i++) {
if ((ha->allocs[i] != NULL) && (ha->allocs[i]->resname != NULL)
&& (strcmp(ha->allocs[i]->resname, "slots") == 0)) {
return (int)ha->allocs[i]->value;
}
}
return 1;
}
/*
* repl_info_worker_one - AC 22's "info worker <name>": "Dump one worker:
* state, resources, assigned jobs." `w->resvals` holds the resource's
@ -667,11 +703,21 @@ static void repl_info_worker_one(xce_sched_sim_driver_t* sim,
* jobid-range-compression helper (repl_fmt_jobid_range(), repl_fmt.c,
* task 1.2) repl_break.c's JOB_PUSH event echo and the Round Rollup's
* pend-jobid display already use, instead of printing one "assigned
* job: jobid=..." line per matching job.
* job: jobid=..." line per matching job. Each job's per-host "slots"
* allocation count (assigned_slots[], repl_worker_alloc_nslots()) rides
* alongside so a multi-slot job is annotated "*N" -- but ONLY when
* every assigned job on this worker consumed the SAME slot count: a
* single trailing "*N" on a compressed multi-job range is unambiguous
* only in that uniform case, so a non-uniform mix falls back to the
* uncompressed per-job listing (same fallback tier a too-small
* compression buffer already uses), each line annotated with its own
* count.
*/
{
uint64_t* assigned_jobids = NULL;
int* assigned_slots = NULL;
int n_assigned = 0;
int n_slots = 0;
job = NULL;
rc = iter_make(&it, sim->core.jobs);
@ -691,8 +737,11 @@ static void repl_info_worker_one(xce_sched_sim_driver_t* sim,
if ((ha != NULL) && (ha->host != NULL)
&& (strcmp(ha->host, w->name) == 0)) {
(void)repl_append_u64(&assigned_jobids, &n_assigned,
job->jobid);
if (repl_append_u64(&assigned_jobids, &n_assigned,
job->jobid)) {
(void)repl_append_int(&assigned_slots, &n_slots,
repl_worker_alloc_nslots(ha));
}
break;
}
}
@ -707,9 +756,20 @@ static void repl_info_worker_one(xce_sched_sim_driver_t* sim,
char jobids_buf[1024];
int jr_rc = repl_fmt_jobid_range(jobids_buf, sizeof(jobids_buf),
assigned_jobids, n_assigned);
bool uniform_slots = (n_slots == n_assigned);
if (jr_rc >= 0) {
fprintf(stdout, "[sim] assigned jobs: %s\n", jobids_buf);
for (i = 1; uniform_slots && (i < n_slots); i++) {
uniform_slots = (assigned_slots[i] == assigned_slots[0]);
}
if ((jr_rc >= 0) && uniform_slots) {
if ((n_slots > 0) && (assigned_slots[0] > 1)) {
fprintf(stdout, "[sim] assigned jobs: %s*%d\n",
jobids_buf, assigned_slots[0]);
} else {
fprintf(stdout, "[sim] assigned jobs: %s\n",
jobids_buf);
}
} else {
/* bug-1591: fmt_jobid() renders the raw idgen()-packed
* jobid in the same human-readable form already used by
@ -718,15 +778,22 @@ static void repl_info_worker_one(xce_sched_sim_driver_t* sim,
* waldump, and job.c's log lines since bug-1588). */
for (i = 0; i < n_assigned; i++) {
char jobidStr[64];
int slots = (i < n_slots) ? assigned_slots[i] : 1;
fmt_jobid(jobidStr, sizeof(jobidStr), assigned_jobids[i]);
fprintf(stdout, "[sim] assigned job: jobid=%s\n",
jobidStr);
if (slots > 1) {
fprintf(stdout, "[sim] assigned job: jobid=%s*%d\n",
jobidStr, slots);
} else {
fprintf(stdout, "[sim] assigned job: jobid=%s\n",
jobidStr);
}
}
}
}
free(assigned_jobids);
free(assigned_slots);
}
}

241
src/scheduler/testsuite/test_repl_resource_table.c
View file

@ -1056,6 +1056,245 @@ test_info_worker_assigned_job_listing_compresses_contiguous_array_indices(
"array_indices: PASS\n");
}
/*
* Improvement request (simulator REPL user feedback): a job consuming more
* than one "slots" unit on a worker should be visibly distinguishable from
* a single-slot job in the assigned-jobs listing (e.g. "jobid*2"), instead
* of every assigned job looking identical regardless of how much of the
* worker it actually occupies.
*/
static void
test_info_worker_assigned_job_listing_appends_slot_count_when_uniform(void)
{
xce_sched_sim_driver_t sim;
xce_core_event_t events[2];
xce_queue_t wire_queue;
xce_worker_t wire_worker;
xce_worker_t* worker_ptrs[1];
xce_job_t* job;
xce_job_t* all_jobs[1];
uint64_t jobid;
xce_sched_job_t* sjob;
xce_alloc_t slots_alloc;
xce_alloc_t* slots_alloc_ptrs[1];
xce_host_alloc_t ha_host;
xce_host_alloc_t* ha_host_ptrs[1];
xce_job_alloc_t alloc;
char* captured;
char expected_range[256];
char expected_line[300];
int rc;
memset(&sim, 0, sizeof(sim));
rc = xce_sched_core_init(&(sim.core));
TEST_ASSERT_EQUAL_INT(0, rc);
memset(&wire_queue, 0, sizeof(wire_queue));
wire_queue.queue = "normal";
wire_queue.priority = 5;
wire_queue.status = QUEUE_STAT_OPEN;
jobid = mk_plain_jobid_bpi(4, 2, 1700000010, 1);
job = make_job(jobid, JOB_STAT_RUN, "normal", "alice", "proj1");
all_jobs[0] = job;
memset(events, 0, sizeof(events));
events[0].type = XCE_CORE_EVENT_QUEUE_UPDATE;
events[0].u.queue_update.queue = &wire_queue;
events[1].type = XCE_CORE_EVENT_JOBS_PUSH;
events[1].u.jobs_push.njobs = 1;
events[1].u.jobs_push.jobs = all_jobs;
rc = xce_core_push_events(&(sim.core), events, 2);
TEST_ASSERT_EQUAL_INT(0, rc);
memset(&wire_worker, 0, sizeof(wire_worker));
wire_worker.name = "hostA";
wire_worker.status = 0;
worker_ptrs[0] = &wire_worker;
memset(events, 0, sizeof(events));
events[0].type = XCE_CORE_EVENT_WORKERS_PUSH;
events[0].u.workers_push.nworkers = 1;
events[0].u.workers_push.workers = worker_ptrs;
rc = xce_core_push_events(&(sim.core), events, 1);
TEST_ASSERT_EQUAL_INT(0, rc);
memset(&slots_alloc, 0, sizeof(slots_alloc));
slots_alloc.resname = "slots";
slots_alloc.value = 2.0;
slots_alloc_ptrs[0] = &slots_alloc;
memset(&ha_host, 0, sizeof(ha_host));
ha_host.host = "hostA";
ha_host.nallocs = 1;
ha_host.allocs = slots_alloc_ptrs;
ha_host_ptrs[0] = &ha_host;
memset(&alloc, 0, sizeof(alloc));
alloc.nhosts = 1;
alloc.allocs = ha_host_ptrs;
sjob = sched_get_job(&(sim.core), jobid);
TEST_ASSERT_NOT_NULL(sjob);
sjob->allocs = &alloc;
rc = repl_fmt_jobid_range(expected_range, sizeof(expected_range), &jobid,
1);
TEST_ASSERT_TRUE(rc >= 0);
snprintf(expected_line, sizeof(expected_line), "assigned jobs: %s*2\n",
expected_range);
captured = capture_dispatch(&sim, "info worker hostA");
TEST_ASSERT_NOT_NULL_MESSAGE(
strstr(captured, expected_line),
"a job consuming 2 'slots' on this worker must be annotated "
"'*2' in the assigned-jobs listing");
free(captured);
printf("test_info_worker_assigned_job_listing_appends_slot_count_when_"
"uniform: PASS\n");
}
/*
* Two jobs on the SAME worker consuming DIFFERENT slot counts cannot be
* annotated with a single trailing "*N" on a compressed range (which count
* would it even mean?) -- must fall back to the pre-existing uncompressed
* per-job listing, each line annotated with its own count.
*/
static void
test_info_worker_assigned_job_listing_falls_back_per_job_when_slots_differ(
void)
{
xce_sched_sim_driver_t sim;
xce_core_event_t events[2];
xce_queue_t wire_queue;
xce_worker_t wire_worker;
xce_worker_t* worker_ptrs[1];
xce_job_t* job_a;
xce_job_t* job_b;
xce_job_t* all_jobs[2];
uint64_t jobid_a;
uint64_t jobid_b;
xce_sched_job_t* sjob;
xce_alloc_t slots_alloc_a;
xce_alloc_t* slots_alloc_a_ptrs[1];
xce_alloc_t slots_alloc_b;
xce_alloc_t* slots_alloc_b_ptrs[1];
xce_host_alloc_t ha_a;
xce_host_alloc_t* ha_a_ptrs[1];
xce_job_alloc_t alloc_a;
xce_host_alloc_t ha_b;
xce_host_alloc_t* ha_b_ptrs[1];
xce_job_alloc_t alloc_b;
char* captured;
char jobidStr_a[64];
char jobidStr_b[64];
char expected_a[96];
char expected_b[96];
int rc;
memset(&sim, 0, sizeof(sim));
rc = xce_sched_core_init(&(sim.core));
TEST_ASSERT_EQUAL_INT(0, rc);
memset(&wire_queue, 0, sizeof(wire_queue));
wire_queue.queue = "normal";
wire_queue.priority = 5;
wire_queue.status = QUEUE_STAT_OPEN;
jobid_a = mk_plain_jobid_bpi(5, 2, 1700000020, 1);
jobid_b = mk_plain_jobid_bpi(5, 2, 1700000021, 2);
job_a = make_job(jobid_a, JOB_STAT_RUN, "normal", "alice", "proj1");
job_b = make_job(jobid_b, JOB_STAT_RUN, "normal", "alice", "proj1");
all_jobs[0] = job_a;
all_jobs[1] = job_b;
memset(events, 0, sizeof(events));
events[0].type = XCE_CORE_EVENT_QUEUE_UPDATE;
events[0].u.queue_update.queue = &wire_queue;
events[1].type = XCE_CORE_EVENT_JOBS_PUSH;
events[1].u.jobs_push.njobs = 2;
events[1].u.jobs_push.jobs = all_jobs;
rc = xce_core_push_events(&(sim.core), events, 2);
TEST_ASSERT_EQUAL_INT(0, rc);
memset(&wire_worker, 0, sizeof(wire_worker));
wire_worker.name = "hostA";
wire_worker.status = 0;
worker_ptrs[0] = &wire_worker;
memset(events, 0, sizeof(events));
events[0].type = XCE_CORE_EVENT_WORKERS_PUSH;
events[0].u.workers_push.nworkers = 1;
events[0].u.workers_push.workers = worker_ptrs;
rc = xce_core_push_events(&(sim.core), events, 1);
TEST_ASSERT_EQUAL_INT(0, rc);
memset(&slots_alloc_a, 0, sizeof(slots_alloc_a));
slots_alloc_a.resname = "slots";
slots_alloc_a.value = 2.0;
slots_alloc_a_ptrs[0] = &slots_alloc_a;
memset(&ha_a, 0, sizeof(ha_a));
ha_a.host = "hostA";
ha_a.nallocs = 1;
ha_a.allocs = slots_alloc_a_ptrs;
ha_a_ptrs[0] = &ha_a;
memset(&alloc_a, 0, sizeof(alloc_a));
alloc_a.nhosts = 1;
alloc_a.allocs = ha_a_ptrs;
memset(&slots_alloc_b, 0, sizeof(slots_alloc_b));
slots_alloc_b.resname = "slots";
slots_alloc_b.value = 4.0;
slots_alloc_b_ptrs[0] = &slots_alloc_b;
memset(&ha_b, 0, sizeof(ha_b));
ha_b.host = "hostA";
ha_b.nallocs = 1;
ha_b.allocs = slots_alloc_b_ptrs;
ha_b_ptrs[0] = &ha_b;
memset(&alloc_b, 0, sizeof(alloc_b));
alloc_b.nhosts = 1;
alloc_b.allocs = ha_b_ptrs;
sjob = sched_get_job(&(sim.core), jobid_a);
TEST_ASSERT_NOT_NULL(sjob);
sjob->allocs = &alloc_a;
sjob = sched_get_job(&(sim.core), jobid_b);
TEST_ASSERT_NOT_NULL(sjob);
sjob->allocs = &alloc_b;
fmt_jobid(jobidStr_a, sizeof(jobidStr_a), jobid_a);
fmt_jobid(jobidStr_b, sizeof(jobidStr_b), jobid_b);
snprintf(expected_a, sizeof(expected_a), "assigned job: jobid=%s*2\n",
jobidStr_a);
snprintf(expected_b, sizeof(expected_b), "assigned job: jobid=%s*4\n",
jobidStr_b);
captured = capture_dispatch(&sim, "info worker hostA");
TEST_ASSERT_NOT_NULL_MESSAGE(
strstr(captured, expected_a),
"differing per-job slot counts on the same worker must fall back "
"to the uncompressed per-job listing, annotated with this job's "
"own slot count");
TEST_ASSERT_NOT_NULL_MESSAGE(
strstr(captured, expected_b),
"differing per-job slot counts on the same worker must fall back "
"to the uncompressed per-job listing, annotated with this job's "
"own slot count");
free(captured);
printf("test_info_worker_assigned_job_listing_falls_back_per_job_when_"
"slots_differ: PASS\n");
}
/*
* AC 6.6, `info resource all`: the same three-section grouping as `info
* worker`'s own version above, applied to the cluster-wide GLOBAL/SHARED
@ -1366,6 +1605,8 @@ int main(void)
RUN_TEST(test_info_worker_units_mode_both_renders_raw_and_human_without_mutating_stored_value);
RUN_TEST(test_info_worker_units_mode_converts_swp_megabytes_before_human_scaling);
RUN_TEST(test_info_worker_assigned_job_listing_compresses_contiguous_array_indices);
RUN_TEST(test_info_worker_assigned_job_listing_appends_slot_count_when_uniform);
RUN_TEST(test_info_worker_assigned_job_listing_falls_back_per_job_when_slots_differ);
RUN_TEST(test_info_resource_groups_resources_into_labeled_table_sections_in_order);
RUN_TEST(test_info_resource_load_metric_whole_number_omits_trailing_zeros);
RUN_TEST(test_info_resource_units_mode_both_renders_raw_and_human_for_bytes_resource);