diff --git a/src/controller/include/cluster.h b/src/controller/include/cluster.h index 9ca999e2..164b3d23 100644 --- a/src/controller/include/cluster.h +++ b/src/controller/include/cluster.h @@ -345,6 +345,8 @@ extern int update_job_rusages(cluster_t*, endpoint_t*, xce_request_t*); extern int update_host_runusage(xce_job_runusage_t**, char*, int, xce_rusage_t**); extern void aggregate_job_runusage(xce_job_runusage_t*); +extern void update_job_cpu_stats(xce_job_runusage_t*, const char*, int, + int64_t, int64_t, int, double); extern int update_pids(job_pidinfo_t**, int, int*); extern int update_pgids(job_pidinfo_t**, int, int*); extern int get_ep_role(endpoint_t*); diff --git a/src/controller/job.c b/src/controller/job.c index 9753d47c..eb1bd6f5 100644 --- a/src/controller/job.c +++ b/src/controller/job.c @@ -13,6 +13,7 @@ #include "persist.h" #include "preason.h" #include "request.h" +#include "resreq.h" #include "typemap.h" /* xce_job_state_entry_t -- worker-rejoin-resync task 2.2 */ #include "xce/cluster_def.h" #include "xce/errnox.h" @@ -1064,6 +1065,7 @@ void merge_job_runusage(xce_job_runusage_t** dst, xce_job_runusage_t* src) d->mem_efficiency = src->mem_efficiency; d->avg_cpu_efficiency = src->avg_cpu_efficiency; d->peak_cpu_efficiency = src->peak_cpu_efficiency; + d->peak_cpu = src->peak_cpu; for (i = 0; i < src->num_hosts; i++) { xce_host_runusage_t* sh = src->host_runusage[i]; @@ -1089,6 +1091,8 @@ void merge_job_runusage(xce_job_runusage_t** dst, xce_job_runusage_t* src) d->host_runusage[matched]->mem_usage = sh->mem_usage; d->host_runusage[matched]->mem_swap = sh->mem_swap; d->host_runusage[matched]->tasks_current = sh->tasks_current; + d->host_runusage[matched]->sample_time = sh->sample_time; + d->host_runusage[matched]->cpu_rate = sh->cpu_rate; } else { /* Genuinely new host: append, transferring ownership of sh's * own plain-malloc'd host_name directly -- never xstring. */ @@ -3751,12 +3755,74 @@ int do_jobcontrol_op(cluster_t* c, endpoint_t* ep, xce_request_t* request) return rc; } +/* bug-1736: `host`'s cumulative cpu_time before its next report is + * applied, or -1 when it has not reported yet. */ +static int host_prev_cpu_time(xce_job_runusage_t* runusage, const char* host) +{ + int i; + + if ((runusage == NULL) || (host == NULL)) { + return -1; + } + for (i = 0; i < runusage->num_hosts; i++) { + xce_host_runusage_t* h = runusage->host_runusage[i]; + + if ((h != NULL) && (h->host_name != NULL) + && (strcmp(h->host_name, host) == 0)) { + return h->cpu_time; + } + } + return -1; +} + +/* bug-1736: the job's allocated slots, summed over its hosts (0 when it has + * no allocation). */ +static int job_alloc_nslots(job_t* j) +{ + int i, n = 0; + + if (j->job.allocs == NULL) { + return 0; + } + for (i = 0; i < j->job.allocs->nhosts; i++) { + n += get_nslots_in_alloc(j->job.allocs->allocs[i]); + } + return n; +} + +/* bug-1736: the job-wide rusage[mem] request in MB (the same total the + * scheduler reserves: per task by default), 0 when the job requests none + * or its resource request is not a single primary one. */ +static double job_requested_mem_mb(cluster_t* c, job_t* j) +{ + xce_resreq_t resreq; + char* errmsg = NULL; + float mem = 0.0f; + + if ((j->job.submit == NULL) || NULLSTR(j->job.submit->resreq) + || (j->job.allocs == NULL)) { + return 0.0; + } + if (parse_resreq(&resreq, j->job.submit->resreq, c->resmgr, &errmsg) < 0) { + FREEUP(errmsg); + return 0.0; + } + if ((resreq.tag == XCE_RESREQ_PRIMARY) && (resreq.u.primary != NULL)) { + (void)calc_wanted_job_total_resval(&(resreq.u.primary->rusages), + PER_TASK, j->job.allocs->nhosts, + job_alloc_nslots(j), "mem", &mem); + } + kill_resreq(&resreq); + return (double)mem; +} + int update_job_rusages(cluster_t* c, endpoint_t* ep, xce_request_t* request) { xce_job_rusages_t* rusages = request->body.updatemsg->data.jobrusages; job_t* j = NULL; worker_t* w = NULL; char* host = NULL; + int prev_cpu; TRACE("%s trying to update job(%ld) rusage", __func__, rusages->jobid); @@ -3795,6 +3861,7 @@ int update_job_rusages(cluster_t* c, endpoint_t* ep, xce_request_t* request) lock(j, RTTI_LOCK_WRITE); xce_run_hooks(&(c->hooksys), XCE_CONTROLLER_HOOK_JOB, HOOK_BEFORE_EVENT, HOOK_DETAIL_JOB_UPDATE, c, j); + prev_cpu = host_prev_cpu_time(j->job.run_usage, host); update_host_runusage(&(j->job.run_usage), host, rusages->count, rusages->usages); @@ -3803,6 +3870,11 @@ int update_job_rusages(cluster_t* c, endpoint_t* ep, xce_request_t* request) * standing "update other fields of job rusage later" FIXME. */ aggregate_job_runusage(j->job.run_usage); + /* bug-1736: CPU peak and efficiency fields (LSF definitions) */ + update_job_cpu_stats(j->job.run_usage, host, prev_cpu, (int64_t)time(NULL), + (int64_t)j->job.start_time, job_alloc_nslots(j), + job_requested_mem_mb(c, j)); + update_pids(&(j->job.pidinfo), rusages->npids, rusages->pids); update_pgids(&(j->job.pidinfo), rusages->npgids, rusages->pgids); diff --git a/src/controller/misc.c b/src/controller/misc.c index c915e474..5d69dd50 100644 --- a/src/controller/misc.c +++ b/src/controller/misc.c @@ -158,35 +158,15 @@ int update_host_runusage(xce_job_runusage_t** runusage, char* host, int count, return 0; } -/* bug-1735/bug-1736: derives xce_job_runusage_t's own job-level max_mem/ - * avg_mem/used_mem/cpu_time/peak_cpu_time from the per-host - * host_runusage[] entries update_host_runusage() populates -- replaces - * update_job_rusages()'s (job.c) standing "update other fields of job - * rusage later" FIXME. +/* bug-1735: derives xce_job_runusage_t's own job-level max_mem/avg_mem/ + * used_mem/cpu_time from the per-host host_runusage[] entries + * update_host_runusage() populates -- replaces update_job_rusages()'s + * (job.c) standing "update other fields of job rusage later" FIXME. * * used_mem is the SUM across hosts (a multi-host job's total memory * footprint); max_mem/avg_mem are the max/average across hosts; cpu_time is - * the SUM across hosts. peak_cpu_time is the running MAX of that per-report - * cpu_time sum across every report this job has ever received (a job's own - * cpu_time can only trend upward within one report but this is called - * repeatedly over the job's life, and a merge/replay/WAL-catchup path could - * in principle deliver reports out of order -- tracking the max explicitly, - * not just trusting "latest == largest", is what makes it a genuine peak - * rather than a duplicate of cpu_time). - * - * bug-1736's OTHER two fields, peak_cpu_duration and the three efficiency - * fields (mem_efficiency/avg_cpu_efficiency/peak_cpu_efficiency), are - * DELIBERATELY left untouched here -- a documented scope decision, not an - * oversight (steering persistence.md's field-completeness rule explicitly - * allows this when documented). Unlike peak_cpu_time, these have no - * unambiguous definition to implement without inventing a formula: the - * efficiency fields need a "used vs. requested" ratio against a job's - * resource request, which is a resreq expression tree (parse_resreq()), - * not a plain core/slot count -- deriving a well-defined denominator from - * that is a real design task, not a two-line fix. peak_cpu_duration's own - * semantics (wall-clock elapsed to the peak? duration the peak was - * sustained?) were never specified anywhere. Both need a product decision - * before a formula can be implemented correctly. */ + * the SUM across hosts. The CPU peak and efficiency fields are + * update_job_cpu_stats()'s (bug-1736). */ void aggregate_job_runusage(xce_job_runusage_t* runusage) { int i, valid = 0; @@ -223,9 +203,90 @@ void aggregate_job_runusage(xce_job_runusage_t* runusage) runusage->max_mem = mem_max; runusage->avg_mem = mem_sum / valid; runusage->cpu_time = cpu_sum; +} - if (cpu_sum > runusage->peak_cpu_time) { - runusage->peak_cpu_time = cpu_sum; +/* bug-1736: the LSF bjobs -l CPU peak and efficiency fields, updated after + * `host`'s report has been applied (update_host_runusage() + + * aggregate_job_runusage()). prev_cpu is that host's cumulative cpu_time + * before the report (< 0: its first report); now/start_time are epoch + * seconds; nslots is the job's allocated slots; req_mem_mb its job-wide + * rusage[mem] request (0: none). + * + * - host cpu_rate: CPU-seconds gained / wall seconds since the host's + * previous report (cores busy); job cores = the sum over its hosts. + * - peak_cpu: highest job cores seen; peak_cpu_time is it rounded (the + * wire field is an integer); peak_cpu_duration the interval that set it. + * - avg_cpu_efficiency = cpu_time / elapsed / nslots, peak_cpu_efficiency + * = peak_cpu / nslots, mem_efficiency = highest used_mem / req_mem_mb, + * all in percent; 0 when the denominator is unknown. + * + * A report no later than the host's previous one (replay, out of order) or + * whose CPU counter went backwards leaves the rates alone. */ +void update_job_cpu_stats(xce_job_runusage_t* runusage, const char* host, + int prev_cpu, int64_t now, int64_t start_time, + int nslots, double req_mem_mb) +{ + xce_host_runusage_t* h = NULL; + double cores = 0.0; + int64_t interval = 0; + int i; + + if ((runusage == NULL) || (runusage->host_runusage == NULL) + || (host == NULL)) { + return; + } + + for (i = 0; i < runusage->num_hosts; i++) { + xce_host_runusage_t* e = runusage->host_runusage[i]; + + if ((e != NULL) && (e->host_name != NULL) + && (strcmp(e->host_name, host) == 0)) { + h = e; + break; + } + } + if (h == NULL) { + return; + } + + if (now > h->sample_time) { + if ((prev_cpu >= 0) && (h->sample_time > 0) + && (h->cpu_time >= prev_cpu)) { + interval = now - h->sample_time; + h->cpu_rate = (float)((double)(h->cpu_time - prev_cpu) + / (double)interval); + } + h->sample_time = now; + } + + for (i = 0; i < runusage->num_hosts; i++) { + if (runusage->host_runusage[i] != NULL) { + cores += runusage->host_runusage[i]->cpu_rate; + } + } + + if ((interval > 0) && (cores > runusage->peak_cpu)) { + runusage->peak_cpu = (float)cores; + runusage->peak_cpu_time = (int)(cores + 0.5); /* cores >= 0 */ + runusage->peak_cpu_duration = (int)interval; + } + + if (nslots > 0) { + runusage->peak_cpu_efficiency + = (float)(100.0 * runusage->peak_cpu / nslots); + if (now > start_time) { + runusage->avg_cpu_efficiency + = (float)(100.0 * runusage->cpu_time + / (double)(now - start_time) / nslots); + } + } + + if (req_mem_mb > 0.0) { + float eff = (float)(100.0 * runusage->used_mem / req_mem_mb); + + if (eff > runusage->mem_efficiency) { + runusage->mem_efficiency = eff; + } } } diff --git a/src/controller/testsuite/test_commit_barrier_reply.c b/src/controller/testsuite/test_commit_barrier_reply.c index 29c96e14..3659f721 100644 --- a/src/controller/testsuite/test_commit_barrier_reply.c +++ b/src/controller/testsuite/test_commit_barrier_reply.c @@ -198,6 +198,11 @@ int update_pids(job_pidinfo_t **i, int n, int *p) int update_pgids(job_pidinfo_t **i, int n, int *p) { (void)i; (void)n; (void)p; return 0; } void aggregate_job_runusage(xce_job_runusage_t *r) { (void)r; } +void update_job_cpu_stats(xce_job_runusage_t *r, const char *h, int p, + int64_t n, int64_t st, int ns, double m) +{ + (void)r; (void)h; (void)p; (void)n; (void)st; (void)ns; (void)m; +} int on_scheduler_join(cluster_t *c) { (void)c; return 0; } int add_apiserver(cluster_t *c, char *n, endpoint_t *e) { (void)c; (void)n; (void)e; return 0; } diff --git a/src/controller/testsuite/test_host_runusage.c b/src/controller/testsuite/test_host_runusage.c index 4398028e..019dbdac 100644 --- a/src/controller/testsuite/test_host_runusage.c +++ b/src/controller/testsuite/test_host_runusage.c @@ -273,33 +273,143 @@ static void test_aggregate_job_runusage_multi_host_sum_max_avg(void) free_runusage(r); } -static void test_aggregate_job_runusage_peak_cpu_time_is_running_max(void) +/* bug-1736: peak_cpu_time is now the LSF "CPU PEAK" (peak cores, set by + * update_job_cpu_stats()); aggregate_job_runusage() no longer touches it. */ +static void test_aggregate_job_runusage_leaves_peak_cpu_alone(void) { int cpu1[] = { 50 }; double mem1[] = { 100.0 }; xce_job_runusage_t* r = make_runusage_with_hosts(1, cpu1, mem1); + r->peak_cpu_time = 3; aggregate_job_runusage(r); - TEST_ASSERT_EQUAL_INT_MESSAGE(50, r->peak_cpu_time, - "peak_cpu_time tracks the first report"); + TEST_ASSERT_EQUAL_INT(50, r->cpu_time); + TEST_ASSERT_EQUAL_INT_MESSAGE(3, r->peak_cpu_time, + "cumulative cpu_time is not a peak"); - /* Second report: a LOWER cpu_time (e.g. a merge/replay delivered an - * older report after a newer one) -- peak must NOT regress. */ - r->host_runusage[0]->cpu_time = 20; - aggregate_job_runusage(r); - TEST_ASSERT_EQUAL_INT_MESSAGE( - 50, r->peak_cpu_time, - "peak_cpu_time (bug-1736) is the running MAX across every report " - "this job has ever received, not just the latest one -- a lower " - "subsequent report must not regress it"); - TEST_ASSERT_EQUAL_INT_MESSAGE( - 20, r->cpu_time, "cpu_time itself IS the latest report, unlike peak"); + free_runusage(r); +} - /* Third report: a NEW high -- peak must advance. */ - r->host_runusage[0]->cpu_time = 90; +/* ------------------------------------------------------------------ */ +/* Tests -- update_job_cpu_stats() (bug-1736, LSF definitions) */ +/* ------------------------------------------------------------------ */ + +#define ASSERT_FLOAT_NEAR(expected, actual) \ + TEST_ASSERT_TRUE_MESSAGE(fabs((double)(expected) - (double)(actual)) \ + < 1e-3, \ + #actual " != " #expected) + +/* One report from `host`: set its cumulative cpu (and mem), re-aggregate + * and update the stats, as update_job_rusages() does. prev_cpu < 0 means + * the host had no earlier report. */ +static void report(xce_job_runusage_t* r, int hostidx, int cpu, double mem, + int64_t now, int64_t start, int nslots, double req_mem) +{ + xce_host_runusage_t* h = r->host_runusage[hostidx]; + int prev_cpu = (h->sample_time > 0) ? h->cpu_time : -1; + + h->cpu_time = cpu; + h->mem_usage = mem; aggregate_job_runusage(r); - TEST_ASSERT_EQUAL_INT_MESSAGE(90, r->peak_cpu_time, - "peak_cpu_time advances on a new high"); + update_job_cpu_stats(r, h->host_name, prev_cpu, now, start, nslots, + req_mem); +} + +static xce_job_runusage_t* make_named_hosts(int n) +{ + static const char* names[] = { "ha", "hb", "hc" }; + xce_job_runusage_t* r = (xce_job_runusage_t*)calloc(1, sizeof(*r)); + int i; + + r->num_hosts = n; + r->host_runusage = (xce_host_runusage_t**)calloc(n, sizeof(*r->host_runusage)); + for (i = 0; i < n; i++) { + r->host_runusage[i] = (xce_host_runusage_t*)calloc(1, sizeof(xce_host_runusage_t)); + r->host_runusage[i]->host_name = strdup(names[i]); + } + return r; +} + +static void test_cpu_stats_rate_peak_and_efficiency(void) +{ + xce_job_runusage_t* r = make_named_hosts(1); + + /* first report: no interval yet, nothing to rate */ + report(r, 0, 0, 100.0, 100, 100, 2, 0.0); + ASSERT_FLOAT_NEAR(0.0, r->peak_cpu); + TEST_ASSERT_EQUAL_INT(100, (int)r->host_runusage[0]->sample_time); + + /* +30 CPU-seconds in 30 s: 1 core busy */ + report(r, 0, 30, 100.0, 130, 100, 2, 0.0); + ASSERT_FLOAT_NEAR(1.0, r->host_runusage[0]->cpu_rate); + ASSERT_FLOAT_NEAR(1.0, r->peak_cpu); + TEST_ASSERT_EQUAL_INT(1, r->peak_cpu_time); + TEST_ASSERT_EQUAL_INT(30, r->peak_cpu_duration); + ASSERT_FLOAT_NEAR(50.0, r->peak_cpu_efficiency); /* 1 core of 2 slots */ + ASSERT_FLOAT_NEAR(50.0, r->avg_cpu_efficiency); /* 30 s / 30 s / 2 */ + + /* +30 in 60 s: 0.5 core -- the peak and its duration stay */ + report(r, 0, 60, 100.0, 190, 100, 2, 0.0); + ASSERT_FLOAT_NEAR(0.5, r->host_runusage[0]->cpu_rate); + ASSERT_FLOAT_NEAR(1.0, r->peak_cpu); + TEST_ASSERT_EQUAL_INT(30, r->peak_cpu_duration); + ASSERT_FLOAT_NEAR(100.0 * 60 / 90 / 2, r->avg_cpu_efficiency); + + free_runusage(r); +} + +static void test_cpu_stats_sum_host_rates(void) +{ + xce_job_runusage_t* r = make_named_hosts(2); + + report(r, 0, 0, 10.0, 100, 100, 4, 0.0); + report(r, 1, 0, 10.0, 100, 100, 4, 0.0); + report(r, 0, 20, 10.0, 120, 100, 4, 0.0); /* host a: 1.0 core */ + report(r, 1, 30, 10.0, 160, 100, 4, 0.0); /* host b: 0.5 core */ + ASSERT_FLOAT_NEAR(1.5, r->peak_cpu); + TEST_ASSERT_EQUAL_INT(2, r->peak_cpu_time); /* rounded */ + TEST_ASSERT_EQUAL_INT(60, r->peak_cpu_duration); + ASSERT_FLOAT_NEAR(37.5, r->peak_cpu_efficiency); /* 1.5 / 4 */ + + free_runusage(r); +} + +static void test_cpu_stats_mem_efficiency_is_peak(void) +{ + xce_job_runusage_t* r = make_named_hosts(1); + + report(r, 0, 0, 256.0, 100, 100, 1, 512.0); + ASSERT_FLOAT_NEAR(50.0, r->mem_efficiency); + report(r, 0, 10, 128.0, 110, 100, 1, 512.0); + ASSERT_FLOAT_NEAR(50.0, r->mem_efficiency); /* peak, not latest */ + report(r, 0, 20, 384.0, 120, 100, 1, 512.0); + ASSERT_FLOAT_NEAR(75.0, r->mem_efficiency); + + free_runusage(r); +} + +static void test_cpu_stats_guards(void) +{ + xce_job_runusage_t* r = make_named_hosts(1); + + report(r, 0, 0, 64.0, 100, 100, 0, 0.0); + report(r, 0, 30, 64.0, 130, 100, 0, 0.0); + /* no slots, no memory request: no efficiencies */ + ASSERT_FLOAT_NEAR(0.0, r->peak_cpu_efficiency); + ASSERT_FLOAT_NEAR(0.0, r->avg_cpu_efficiency); + ASSERT_FLOAT_NEAR(0.0, r->mem_efficiency); + ASSERT_FLOAT_NEAR(1.0, r->peak_cpu); + + /* a replayed/out-of-order report (no later than the last one) and a + * CPU counter that went backwards leave the rate alone */ + report(r, 0, 90, 64.0, 130, 100, 0, 0.0); + ASSERT_FLOAT_NEAR(1.0, r->host_runusage[0]->cpu_rate); + report(r, 0, 10, 64.0, 140, 100, 0, 0.0); + ASSERT_FLOAT_NEAR(1.0, r->host_runusage[0]->cpu_rate); + ASSERT_FLOAT_NEAR(1.0, r->peak_cpu); + + update_job_cpu_stats(NULL, "ha", 0, 1, 1, 1, 1.0); /* must not crash */ + update_job_cpu_stats(r, "nosuch", 0, 200, 100, 1, 1.0); free_runusage(r); } @@ -333,7 +443,11 @@ int main(void) RUN_TEST(test_update_host_runusage_parses_swap_and_task_count); RUN_TEST(test_aggregate_job_runusage_single_host); RUN_TEST(test_aggregate_job_runusage_multi_host_sum_max_avg); - RUN_TEST(test_aggregate_job_runusage_peak_cpu_time_is_running_max); + RUN_TEST(test_aggregate_job_runusage_leaves_peak_cpu_alone); + RUN_TEST(test_cpu_stats_rate_peak_and_efficiency); + RUN_TEST(test_cpu_stats_sum_host_rates); + RUN_TEST(test_cpu_stats_mem_efficiency_is_peak); + RUN_TEST(test_cpu_stats_guards); RUN_TEST(test_aggregate_job_runusage_null_and_empty_are_safe); return UNITY_END(); diff --git a/src/controller/testsuite/test_wal_job_rusage_fact.c b/src/controller/testsuite/test_wal_job_rusage_fact.c index dc83187c..f4e7a293 100644 --- a/src/controller/testsuite/test_wal_job_rusage_fact.c +++ b/src/controller/testsuite/test_wal_job_rusage_fact.c @@ -188,10 +188,12 @@ static job_t* wru_seed_job(uint64_t jobid, uint32_t state) /* ── Real fixture data: 2 hosts' worth of accumulated CPU/mem usage ──────── */ static xce_host_runusage_t g_wru_host_a = { - .host_name = (char*)"wru-host-a", .cpu_time = 1200, .mem_usage = 512.5 + .host_name = (char*)"wru-host-a", .cpu_time = 1200, .mem_usage = 512.5, + .sample_time = 1790000000, .cpu_rate = 1.25f /* bug-1736 */ }; static xce_host_runusage_t g_wru_host_b = { - .host_name = (char*)"wru-host-b", .cpu_time = 900, .mem_usage = 256.25 + .host_name = (char*)"wru-host-b", .cpu_time = 900, .mem_usage = 256.25, + .sample_time = 1790000030, .cpu_rate = 0.5f }; static xce_host_runusage_t* g_wru_host_ptrs[2]; @@ -209,6 +211,7 @@ static xce_job_runusage_t wru_make_test_runusage(void) r.mem_efficiency = 0.75f; r.avg_cpu_efficiency = 0.6f; r.peak_cpu_efficiency = 0.9f; + r.peak_cpu = 1.75f; /* bug-1736 */ r.num_hosts = 2; g_wru_host_ptrs[0] = &g_wru_host_a; @@ -232,14 +235,19 @@ static void wru_assert_runusage_matches(xce_job_runusage_t* got) TEST_ASSERT_EQUAL_FLOAT(0.75f, got->mem_efficiency); TEST_ASSERT_EQUAL_FLOAT(0.6f, got->avg_cpu_efficiency); TEST_ASSERT_EQUAL_FLOAT(0.9f, got->peak_cpu_efficiency); + TEST_ASSERT_EQUAL_FLOAT(1.75f, got->peak_cpu); /* bug-1736 */ TEST_ASSERT_EQUAL_INT(2, got->num_hosts); TEST_ASSERT_NOT_NULL(got->host_runusage); TEST_ASSERT_EQUAL_STRING("wru-host-a", got->host_runusage[0]->host_name); TEST_ASSERT_EQUAL_INT(1200, got->host_runusage[0]->cpu_time); TEST_ASSERT_EQUAL_FLOAT(512.5, got->host_runusage[0]->mem_usage); + TEST_ASSERT_TRUE(got->host_runusage[0]->sample_time == 1790000000); + TEST_ASSERT_EQUAL_FLOAT(1.25f, got->host_runusage[0]->cpu_rate); TEST_ASSERT_EQUAL_STRING("wru-host-b", got->host_runusage[1]->host_name); TEST_ASSERT_EQUAL_INT(900, got->host_runusage[1]->cpu_time); TEST_ASSERT_EQUAL_FLOAT(256.25, got->host_runusage[1]->mem_usage); + TEST_ASSERT_TRUE(got->host_runusage[1]->sample_time == 1790000030); + TEST_ASSERT_EQUAL_FLOAT(0.5f, got->host_runusage[1]->cpu_rate); } /* Build a wal_jobstate_t header + trailing Cap'n Proto JobRunusage bytes -- @@ -449,7 +457,7 @@ test_wru_apply_committed_entry_merges_without_freeing_live_host(void) * this job would carry. */ xce_host_runusage_t update_host = { .host_name = (char*)"wru-live-host", .cpu_time = 999, - .mem_usage = 888.0 + .mem_usage = 888.0, .sample_time = 1790000100, .cpu_rate = 2.5f }; xce_host_runusage_t new_host = { .host_name = (char*)"wru-new-host", .cpu_time = 42, .mem_usage = 84.0 @@ -460,6 +468,7 @@ test_wru_apply_committed_entry_merges_without_freeing_live_host(void) memset(&r, 0, sizeof(r)); r.max_mem = 500.0; r.cpu_time = 250; + r.peak_cpu = 3.25f; /* bug-1736 */ r.num_hosts = 2; r.host_runusage = host_ptrs; @@ -478,6 +487,7 @@ test_wru_apply_committed_entry_merges_without_freeing_live_host(void) TEST_ASSERT_NOT_NULL(got); TEST_ASSERT_EQUAL_FLOAT(500.0, got->max_mem); TEST_ASSERT_EQUAL_INT(250, got->cpu_time); + TEST_ASSERT_EQUAL_FLOAT(3.25f, got->peak_cpu); /* bug-1736 */ TEST_ASSERT_EQUAL_INT(2, got->num_hosts); int live_idx = -1, found_new = 0; @@ -494,6 +504,9 @@ test_wru_apply_committed_entry_merges_without_freeing_live_host(void) "malloc'd pointer)"); TEST_ASSERT_EQUAL_INT(999, h->cpu_time); TEST_ASSERT_EQUAL_FLOAT(888.0, h->mem_usage); + /* bug-1736: a merged existing host keeps the rate baseline */ + TEST_ASSERT_TRUE(h->sample_time == 1790000100); + TEST_ASSERT_EQUAL_FLOAT(2.5f, h->cpu_rate); } else if (h->host_name != NULL && strcmp(h->host_name, "wru-new-host") == 0) { found_new = 1; diff --git a/src/include/xce/types.h b/src/include/xce/types.h index 6d933588..dbf7071e 100644 --- a/src/include/xce/types.h +++ b/src/include/xce/types.h @@ -451,6 +451,10 @@ typedef struct { double mem_usage; double mem_swap; int tasks_current; + /* bug-1736: this host's last report (epoch seconds) and its CPU rate + * over the interval that report closed (cores busy) */ + int64_t sample_time; + float cpu_rate; } xce_host_runusage_t; typedef struct { @@ -465,6 +469,7 @@ typedef struct { float peak_cpu_efficiency; int num_hosts; xce_host_runusage_t** host_runusage; + float peak_cpu; /* bug-1736: peak cores, unrounded */ } xce_job_runusage_t; typedef struct { diff --git a/src/schema/controller.capnp b/src/schema/controller.capnp index 6b3f3264..8a12e3a8 100644 --- a/src/schema/controller.capnp +++ b/src/schema/controller.capnp @@ -286,6 +286,8 @@ struct HostRunusage $C.mapname("xce_host_runusage_t"){ memUsage @2: Float64 $C.mapname("mem_usage"); memSwap @3: Float64 $C.mapname("mem_swap"); tasksCurrent @4: Int32 $C.mapname("tasks_current"); + sampleTime @5: Int64 $C.mapname("sample_time"); + cpuRate @6: Float32 $C.mapname("cpu_rate"); } struct JobRunusage $C.mapname("xce_job_runusage_t"){ @@ -299,6 +301,7 @@ struct JobRunusage $C.mapname("xce_job_runusage_t"){ avgCpuEfficiency @7: Float32 $C.mapname("avg_cpu_efficiency"); peakCpuEfficiency @8: Float32 $C.mapname("peak_cpu_efficiency"); hostRunusage @9: List(HostRunusage) $C.mapname("host_runusage") $C.maplistcount("num_hosts"); + peakCpu @10: Float32 $C.mapname("peak_cpu"); } struct JobFail $C.mapname("job_failinfo_t"){