feat(controller): compute the LSF CPU peak and efficiency fields (bug-1736)

update_job_cpu_stats() (misc.c), run after each usage report:
- each host's cpu_rate = CPU-seconds gained / wall seconds since its
  previous report; the job's cores = the sum over hosts
- peak_cpu = highest job cores seen (peak_cpu_time = it rounded, the
  wire field is an integer); peak_cpu_duration = the interval that set it
- avg_cpu_efficiency = cpu_time / elapsed / slots, peak_cpu_efficiency =
  peak_cpu / slots, mem_efficiency = highest used_mem / rusage[mem]
  request (job-wide total, per task by default), all in percent
aggregate_job_runusage() no longer treats cumulative cpu_time as a peak.

New fields host sample_time/cpu_rate and job peak_cpu are appended to
the capnp structs (old WAL entries still decode) and copied by
merge_job_runusage(), so a promoted standby keeps the rate baselines.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
This commit is contained in:
Rongsong Shen 2026-10-02 04:14:35 +00:00
parent 77b5102b3e
commit 6df644e6d5
Signed by untrusted user: rshen
GPG key ID: 80533536B0EB651C
8 changed files with 325 additions and 50 deletions

View file

@ -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);