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

fix(controller,apiserver,xcecli): expose job CPU peak and efficiency to the apiserver and xcecli (bug-1736)

Browse source 
The controller computed run usage (CPU time, memory, CPU peak and
efficiency, LSF bjobs -l definitions) but the capnp Job struct had no
field for it, so the job pushed to the apiserver never carried it, and
snapshots (the same struct) dropped it: a job restored from a checkpoint
lost its accumulated usage.

- controller.capnp: Job.runUsage @13 (JobRunusage); old entries decode
  with no run usage.
- apiserver: map run_usage into JobInfo; max_mem, avg_mem, used_mem,
  cpu_time, peak_cpu, peak_cpu_time, peak_cpu_duration, mem_efficiency,
  avg_cpu_efficiency, peak_cpu_efficiency become queryable columns.
- xcecli: -o fields MAX_MEM, AVG_MEM, MEM_EFFICIENCY, CPU_PEAK,
  CPU_PEAK_DURATION, CPU_EFFICIENCY, CPU_PEAK_EFFICIENCY (CPU_TIME now
  renders too), and a MEMORY USAGE / CPU USAGE block in jobs list -l.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
This commit is contained in:
Rongsong Shen 2026-10-06 13:54:43 +00:00
parent e2472aa73c
commit 1cedd7f791
Signed by untrusted user: rshen
GPG key ID: 80533536B0EB651C
9 changed files with 466 additions and 56 deletions
Download patch file Download diff file Expand all files Collapse all files

123
src/controller/testsuite/test_persist_snapshot_entities.c
View file

@ -739,6 +739,128 @@ static void test_snapent_checkpoint_exit_info_roundtrip(void)
snapent_teardown(src);
}
/* ------------------------------------------------------------------ */
/* bug-1736: job.run_usage must round-trip through a checkpoint, every */
/* field. The capnp Job struct had no run-usage field, so a job */
/* restored from a snapshot lost its accumulated CPU/memory usage and */
/* the CPU peak/efficiency values (the WAL_JOB_RUSAGE facts behind the */
/* checkpoint are trimmed), and the same struct is what the apiserver */
/* receives, so they never reached it either. */
/* ------------------------------------------------------------------ */
static xce_job_runusage_t* snapent_make_runusage(void)
{
xce_job_runusage_t* ru = calloc(1, sizeof(*ru));
TEST_ASSERT_NOT_NULL(ru);
ru->max_mem = 512.5;
ru->avg_mem = 300.25;
ru->used_mem = 410.0;
ru->cpu_time = 1234;
ru->peak_cpu_time = 4;
ru->peak_cpu_duration = 30;
ru->mem_efficiency = 80.5f;
ru->avg_cpu_efficiency = 62.25f;
ru->peak_cpu_efficiency = 95.0f;
ru->peak_cpu = 3.75f;
ru->num_hosts = 2;
ru->host_runusage = calloc(2, sizeof(xce_host_runusage_t*));
TEST_ASSERT_NOT_NULL(ru->host_runusage);
for (int h = 0; h < 2; h++) {
xce_host_runusage_t* hr = calloc(1, sizeof(*hr));
TEST_ASSERT_NOT_NULL(hr);
hr->host_name = strdup(h == 0 ? "hostA" : "hostB");
hr->cpu_time = 600 + h;
hr->mem_usage = 200.5 + h;
hr->mem_swap = 10.0 + h;
hr->tasks_current = 3 + h;
hr->sample_time = 1790000000 + h;
hr->cpu_rate = 1.5f + (float)h;
ru->host_runusage[h] = hr;
}
return ru;
}
static void snapent_assert_runusage(const xce_job_runusage_t* ru)
{
TEST_ASSERT_NOT_NULL_MESSAGE(ru,
"bug-1736: a checkpointed job must keep its run_usage");
TEST_ASSERT_EQUAL_FLOAT((float)(512.5), (float)(ru->max_mem));
TEST_ASSERT_EQUAL_FLOAT((float)(300.25), (float)(ru->avg_mem));
TEST_ASSERT_EQUAL_FLOAT((float)(410.0), (float)(ru->used_mem));
TEST_ASSERT_EQUAL_INT(1234, ru->cpu_time);
TEST_ASSERT_EQUAL_INT(4, ru->peak_cpu_time);
TEST_ASSERT_EQUAL_INT(30, ru->peak_cpu_duration);
TEST_ASSERT_EQUAL_FLOAT(80.5f, ru->mem_efficiency);
TEST_ASSERT_EQUAL_FLOAT(62.25f, ru->avg_cpu_efficiency);
TEST_ASSERT_EQUAL_FLOAT(95.0f, ru->peak_cpu_efficiency);
TEST_ASSERT_EQUAL_FLOAT(3.75f, ru->peak_cpu);
TEST_ASSERT_EQUAL_INT(2, ru->num_hosts);
for (int h = 0; h < 2; h++) {
const xce_host_runusage_t* hr = ru->host_runusage[h];
TEST_ASSERT_NOT_NULL(hr);
TEST_ASSERT_EQUAL_STRING(h == 0 ? "hostA" : "hostB", hr->host_name);
TEST_ASSERT_EQUAL_INT(600 + h, hr->cpu_time);
TEST_ASSERT_EQUAL_FLOAT((float)(200.5 + h), (float)(hr->mem_usage));
TEST_ASSERT_EQUAL_FLOAT((float)(10.0 + h), (float)(hr->mem_swap));
TEST_ASSERT_EQUAL_INT(3 + h, hr->tasks_current);
TEST_ASSERT_EQUAL_INT64(1790000000 + h, hr->sample_time);
TEST_ASSERT_EQUAL_FLOAT(1.5f + (float)h, hr->cpu_rate);
}
}
static void test_snapent_checkpoint_run_usage_roundtrip(void)
{
const int n_jobs = 2;
snapent_fixture_t* src = snapent_make(n_jobs, 0);
TEST_ASSERT_NOT_NULL(src);
/* job 0: full run usage; job 1: none reported yet (NULL) */
job_t* j = snapent_find_job(src->c, snapent_jobid(0));
TEST_ASSERT_NOT_NULL(j);
lock(j, RTTI_LOCK_WRITE);
j->job.run_usage = snapent_make_runusage();
unlock(j);
TEST_ASSERT_EQUAL_INT(0, persist_checkpoint(&src->c->persist));
uint64_t lsn = src->c->persist.checkpoint_lsn;
char path[4096];
snprintf(path, sizeof(path), "%s/snap-%016" PRIx64 ".capnp",
src->c->persist.snap_path, lsn);
/* 1. the checkpoint's own job frames carry it */
xce_job_t decoded[8];
int count = snapent_read_checkpoint_jobs(path, decoded, 8);
TEST_ASSERT_EQUAL_INT(n_jobs, count);
xce_job_t* d0 = snapent_find_decoded_job(decoded, count, snapent_jobid(0));
xce_job_t* d1 = snapent_find_decoded_job(decoded, count, snapent_jobid(1));
TEST_ASSERT_NOT_NULL(d0);
TEST_ASSERT_NOT_NULL(d1);
snapent_assert_runusage(d0->run_usage);
TEST_ASSERT_NULL_MESSAGE(d1->run_usage,
"a job with no run usage must come back with none");
snapent_free_decoded_jobs(decoded, count);
/* 2. a real persist_load_snapshot restore into an empty member */
snapent_fixture_t* dst = snapent_make(0, 0);
TEST_ASSERT_NOT_NULL(dst);
bool entities_restored = false;
TEST_ASSERT_EQUAL_INT(0, persist_load_snapshot(&dst->c->persist, path, lsn,
&entities_restored));
TEST_ASSERT_TRUE(entities_restored);
job_t* r0 = snapent_find_job(dst->c, snapent_jobid(0));
job_t* r1 = snapent_find_job(dst->c, snapent_jobid(1));
TEST_ASSERT_NOT_NULL(r0);
TEST_ASSERT_NOT_NULL(r1);
snapent_assert_runusage(r0->job.run_usage);
TEST_ASSERT_NULL(r1->job.run_usage);
snapent_teardown(dst);
snapent_teardown(src);
}
/* ------------------------------------------------------------------ */
/* Test 3 (task 2.1, RED): concurrent per-entity mutation during the */
/* checkpoint walk must never be captured as a torn mix of fields */
@ -2299,6 +2421,7 @@ void test_persist_snapshot_entities(void)
RUN_TEST(test_snapent_fixture_seeds_jobs_and_queues);
RUN_TEST(test_snapent_checkpoint_job_content_roundtrip);
RUN_TEST(test_snapent_checkpoint_exit_info_roundtrip);
RUN_TEST(test_snapent_checkpoint_run_usage_roundtrip);
RUN_TEST(test_snapent_checkpoint_concurrent_mutation_not_torn);
RUN_TEST(test_snapent_checkpoint_write_failure_leaves_no_final_file);
RUN_TEST(test_snapent_checkpoint_mid_encode_failure_discards_whole_file);