myxce/src/scheduler/src/resource.c
Rongsong Shen c1cd8f917d
fix(scheduler): free registry entries after their tables at core teardown; fix dup_val and the unmeet re-init leak (bug-1884, part 5)
A failed key_add() runs the table's datakiller on the rejected value, and
callers keep using it (JOBS_PUSH merges a duplicate job into the existing
one; sched_init_queues() re-adds every queue on each config apply), so the
job, queue, worker and resource datakillers must stay shallow. Part 4 made
kill_sched_job() free the whole job; that is reverted. Instead
xce_sched_core_fini() collects each table's entries, destroys the table,
then frees each entry with a full destructor (core.c). The resource table
is released the same way (a reference per resobj).

Also:
- dup_val() tested which == XCE_VAL_STRING (0, the resource value type,
  equal to Value_u_ivalue32): real string values were shallow-copied and a
  nonzero int32 went to makeString() as a pointer. It now tests
  Value_u_svalue and strdup()s, as free_Value() expects.
- The shared/host satisfy checks re-initialize the unmeet vector they are
  given; the allocators passed the ctx's live one, dropping its buffers on
  every candidate host. The callers finalize it first.
- add_vres() registers a resobj via makeResource(), like every other
  resource, instead of a bare calloc(); test_limit.c drops it with
  dereference().
- plugin_register_core() rejects a duplicate name before init_fn/key_add,
  since a failed add now finalizes (frees) the plugin and callers free it
  on failure.
- sharedInUse gets kill_sched_resval, as put_resvals() fills it.

Exit-time leak on the reproduction replay 7,469 -> 928 bytes.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
2026-09-29 09:28:08 +00:00

1067 lines
36 KiB
C

#include <errno.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include "kvmgr.h"
#include "plugin.h"
#include "preason.h"
#include "resreq.h"
#include "sched/types.h"
#include "scheduler.h"
#include "utils.h"
#include "xce/types.h"
#include "xcealloc.h"
#include "xcelimit.h"
#include "zarray.h"
int get_resvals(zarray_t* array, int nres, int* idxs, float* vals)
{
xce_val_t* xvals[nres + 1];
int rc = 0;
int i;
if ((array == NULL) || (nres < 0) || (idxs == NULL) || (vals == NULL)) {
return -EINVAL;
}
if (nres == 0) {
return 0;
}
memset(vals, 0x0, sizeof(*vals) * nres);
rc = zarray_get_ex(array, nres, idxs, (void**)xvals);
if (rc < 0) {
return rc;
}
for (i = 0; i < nres; i++) {
if (xvals[i] == NULL) {
continue;
}
if (xvals[i]->which != Value_u_fvalue) {
continue;
}
vals[i] = xvals[i]->u.fvalue;
}
return nres;
}
int put_resvals(zarray_t* array, int nres, int* idxs, float* vals)
{
xce_val_t* xvals[nres + 1];
int rc, i;
if ((array == NULL) || (nres < 0) || (idxs == NULL) || (vals == NULL)) {
return -EINVAL;
}
if (nres == 0) {
return 0;
}
rc = zarray_get_ex(array, nres, idxs, (void**)xvals);
if (rc < 0) {
return rc;
}
/* bug-1569 thread B (3rd stacked cause): idx 0 is a legitimate resource
* index (e.g. "slots", almost always the FIRST resource registered --
* confirmed live, resmgr assigns it idx 0), not a sentinel -- get_resvals()
* above has no such guard, and zarray_t is a plain 0-based index array
* throughout this codebase (core_events.c's zarray_put(&worker->resvals,
* r->idx, ...), same convention). `<= 0` skipped idx 0 here, leaving
* xvals[0] NULL for the "create a zero-value entry if missing" step
* below -- live-confirmed as a reproducible SIGSEGV (gdb: `xvals[i]->u.
* fvalue = vals[i]` on a NULL xvals[0]) the moment a job's resource
* charge-on-dispatch first reached this function, since no job had ever
* been genuinely dispatched before this bug's other two stacked causes
* (scheduler.c's sched_init_queues() picker-registration gap;
* core_events.c's core_build_job() never constructing job->schedobj)
* were fixed. */
for (i = 0; i < nres; i++) {
if (idxs[i] < 0) {
continue;
}
if (xvals[i] == NULL) {
xce_val_t* val = (xce_val_t*)calloc(1, sizeof(*val));
if (val == NULL) {
continue;
}
val->which = Value_u_fvalue;
val->u.fvalue = 0.0;
zarray_put(array, idxs[i], val);
xvals[i] = val;
}
}
for (i = 0; i < nres; i++) {
if (xvals[i] == NULL) {
/* idxs[i] < 0 (skipped above) or the calloc() above failed --
* either way there is nowhere to write this value. */
continue;
}
xvals[i]->u.fvalue = vals[i];
}
return nres;
}
static int unsatisfy_(xce_sched_core_t* core, int nres, int* idxs, float* total,
float* inuse, float* uncomitted, float* wanted,
int* unmeet_idxs, float* unmeet_vals)
{
int i, j;
int count;
count = 0;
for (i = 0; i < nres; i++) {
if (total[i] < inuse[i] + uncomitted[i] + wanted[i]) {
TRACE("%s resource %d total %f inuse %f uncomit %f wanted %f",
__func__, idxs[i], total[i], inuse[i], uncomitted[i],
wanted[i]);
count++;
}
}
if (count == 0) {
return 0;
}
j = 0;
for (i = 0; i < nres; i++) {
if (total[i] < inuse[i] + uncomitted[i] + wanted[i]) {
xce_resource_t* r = NULL;
float need = inuse[i] + uncomitted[i] + wanted[i] - total[i];
vget(core->resmgr, idxs[i], (void**)&r);
TRACE("%s unmet resource %d (%s) shortfall %f", __func__, idxs[i],
r ? r->name : "?", need);
unmeet_idxs[j] = idxs[i];
unmeet_vals[j] = need;
j++;
}
}
return count;
}
static host_eval_ctx_t* build_host_eval_ctx(const xce_sched_worker_t* w,
int num_res)
{
host_eval_ctx_t* ctx = calloc(1, sizeof(*ctx));
if (ctx == NULL)
return NULL;
if (num_res <= 0) {
/* valid empty context — numvals/strvals remain NULL, bounds checks
* protect access */
ctx->num_res = 0;
return ctx;
}
double* numvals = calloc(num_res, sizeof(double));
const char** strvals = calloc(num_res, sizeof(char*));
if (numvals == NULL || strvals == NULL) {
free(numvals);
free(strvals);
free(ctx);
return NULL;
}
for (int i = 0; i < num_res; i++) {
xce_val_t* v = (xce_val_t*)zarray_get((zarray_t*)&w->resvals, i);
if (v == NULL)
continue;
switch (v->which) {
case Value_u_fvalue:
numvals[i] = (double)v->u.fvalue;
break;
case Value_u_bvalue:
numvals[i] = v->u.bvalue ? 1.0 : 0.0;
break;
case Value_u_svalue:
strvals[i] = v->u.svalue;
break;
}
}
ctx->num_res = num_res;
ctx->numvals = numvals;
ctx->strvals = strvals;
return ctx;
}
static void free_host_eval_ctx(host_eval_ctx_t* ctx)
{
if (ctx == NULL)
return;
free((void*)ctx->numvals);
free((void*)ctx->strvals);
free(ctx);
}
void sched_worker_release_host_ctx(xce_sched_worker_t* w)
{
if (w == NULL)
return;
free_host_eval_ctx(w->host_eval_ctx);
w->host_eval_ctx = NULL;
}
int sched_meet_host_selection(xce_sched_core_t* core, xce_sched_worker_t* w,
xce_select_t* s)
{
if (s == NULL || s->root == NULL)
return 1;
/* Build host context lazily — reused for all jobs against this host. */
if (w->host_eval_ctx == NULL) {
int num_res = (int)w->resvals.caps;
w->host_eval_ctx = build_host_eval_ctx(w, num_res);
if (w->host_eval_ctx == NULL) {
/* OOM: conservatively pass host */
return 1;
}
}
/* Fast path: use session-level memoisation context when available. */
if (core->sel_eval_ctx != NULL) {
/* Reset memoisation bitmap only when moving to a different host. */
if (core->sel_eval_ctx->host != w->host_eval_ctx)
select_eval_ctx_reset(core->sel_eval_ctx, w->host_eval_ctx);
int rc = eval_select(s->root, core->sel_eval_ctx);
return (rc > 0) ? 1 : 0;
}
/* Fallback (no session context — e.g. unit tests or OOM at session start).
*/
int rc = eval_select_simple(s->root, w->host_eval_ctx);
return (rc > 0) ? 1 : 0;
}
static int sched_check_shared_satisfy_(xce_sched_core_t* core, xvec_t* total,
xvec_t* inuse, xvec_t* uncomitted,
xvec_idval_t* wanted,
xvec_idval_t* unmeet)
{
int nres = wanted->count;
int rc;
rc = xvec_init(total, nres, sizeof(float));
if (rc < 0) {
return rc;
}
rc = xvec_init(inuse, nres, sizeof(float));
if (rc < 0) {
return rc;
}
rc = xvec_idval_init(unmeet, nres);
if (rc < 0) {
return rc;
}
rc = get_resvals(&(core->sharedInst), nres, (int*)wanted->idxs.data,
(float*)total->data);
if (rc < 0) {
return rc;
}
rc = get_resvals(&(core->sharedInUse), nres, (int*)wanted->idxs.data,
(float*)inuse->data);
if (rc < 0) {
return rc;
}
rc = unsatisfy_(core, nres, (int*)wanted->idxs.data, (float*)total->data,
(float*)inuse->data, (float*)uncomitted->data,
(float*)wanted->vals.data, (int*)unmeet->idxs.data,
(float*)unmeet->vals.data);
if (rc < 0) {
return rc;
}
if (rc > 0) {
unmeet->count = rc;
return 0;
}
return 1;
}
int sched_check_shared_satisfy(xce_sched_core_t* core, xvec_t* uncomitted,
xvec_idval_t* wanted, xvec_idval_t* unmeet)
{
xvec_t total;
xvec_t inuse;
int rc;
if (wanted->count == 0) {
return 1;
}
rc = sched_check_shared_satisfy_(core, &total, &inuse, uncomitted, wanted,
unmeet);
if (rc < 0) {
xvec_idval_fini(unmeet);
}
xvec_fini(&total);
xvec_fini(&inuse);
return rc;
}
static int sched_check_host_satisfy_(xce_sched_core_t* core,
xce_sched_worker_t* w, xvec_t* total,
xvec_t* inuse, xvec_t* uncomitted,
xvec_idval_t* wanted, xvec_idval_t* unmeet)
{
int nres = wanted->count;
int rc = 0;
rc = xvec_init(total, nres, sizeof(float));
if (rc < 0) {
return rc;
}
rc = xvec_init(inuse, nres, sizeof(float));
if (rc < 0) {
return rc;
}
rc = xvec_idval_init(unmeet, nres);
if (rc < 0) {
return rc;
}
rc = get_resvals(&(w->resvals), nres, (int*)wanted->idxs.data,
(float*)total->data);
if (rc < 0) {
return rc;
}
rc = get_resvals(&(w->resInUse), nres, (int*)wanted->idxs.data,
(float*)inuse->data);
if (rc < 0) {
return rc;
}
rc = unsatisfy_(core, nres, (int*)wanted->idxs.data, (float*)total->data,
(float*)inuse->data, (float*)uncomitted->data,
(float*)wanted->vals.data, (int*)unmeet->idxs.data,
(float*)unmeet->vals.data);
if (rc < 0) {
return rc;
}
if (rc > 0) {
unmeet->count = rc;
return 0;
}
return 1;
}
int sched_check_host_satisfy(xce_sched_core_t* core, xce_sched_worker_t* w,
xvec_t* uncomitted, xvec_idval_t* wanted,
xvec_idval_t* unmeet)
{
xvec_t total;
xvec_t inuse;
int rc;
if (wanted->count == 0) {
return 1;
}
rc = sched_check_host_satisfy_(core, w, &total, &inuse, uncomitted, wanted,
unmeet);
if (rc < 0) {
xvec_idval_fini(unmeet);
}
xvec_fini(&total);
xvec_fini(&inuse);
return rc;
}
static int sched_update_res_accounts(xce_sched_core_t* core, zarray_t* array,
int global, int naccounts,
xce_res_account_t** accounts, int charge)
{
int i;
int idxs[naccounts + 1];
float vals[naccounts + 1];
if (naccounts < 0) {
return -EINVAL;
}
if (naccounts == 0) {
return 0;
}
for (i = 0; i < naccounts; i++) {
xce_res_account_t* acct = accounts[i];
if (acct->residx < 0) {
xce_resource_t* r = NULL;
int rc = 0;
lock(core->resmgr, RTTI_LOCK_READ);
rc = key_find(core->resmgr, (void*)acct->resname, (void**)&r);
unlock(core->resmgr);
if (rc < 0) {
return rc;
}
/* It's possible that r return NULL if acct->resname is a
* virtual resource of limit */
if (r == NULL) {
int rc;
rc = add_vres(&r, core->resmgr, acct->resname, global);
if (rc < 0) {
return rc;
}
lock(core->resmgr, RTTI_LOCK_READ);
rc = key_find(core->resmgr, (void*)acct->resname,
(void**)&r);
unlock(core->resmgr);
if (rc < 0) {
return rc;
}
}
if (r != NULL) {
acct->residx = r->idx;
}
}
}
for (i = 0; i < naccounts; i++) {
idxs[i] = accounts[i]->residx;
}
get_resvals(array, naccounts, idxs, vals);
for (i = 0; i < naccounts; i++) {
float val = accounts[i]->used + accounts[i]->reserved;
TRACE("%s resource %d old %f delta %f charge %d", __func__, idxs[i],
vals[i], val, charge);
if (charge) {
vals[i] += val;
} else {
vals[i] -= val;
}
}
put_resvals(array, naccounts, idxs, vals);
return 0;
}
/*
* sched_try_alloc_on_host_ - the "try_alloc(worker, plan)" leaf of the
* Composition Algorithm (requirements.md, scheduler-allocation spec):
* select filter, then per-host prepare (plugin_prepare_allocation() with
* ctx->worker set, which is what actually populates ctx->alloc_req.host
* for THIS candidate -- xce_host_iter_t's own next() never does this; it
* only pre-filters against whatever is already staged), then the real
* resource-satisfaction check, then commit. Shared by every allocator
* plugin's bucket loop (seqjob_allocator.c's single-bucket case,
* parjob_allocator.c's multi-bucket case) -- exported (was `static`, and
* wrapped by the now-deleted variadic sched_try_alloc_on_hosts(), whose
* sole caller was seqjob_allocator.c's own pre-rewrap xlist_find() call
* site) so both plugins can call it directly with real parameters instead
* of each carrying its own private copy of this logic. Returns >0 on
* success (host claimed, ctx->alloc_req.host/ctx->worker hold the
* committed want), 0 if this host does not work (a pend reason is
* recorded), <0 on infrastructure error.
*/
int sched_try_alloc_on_host_(xce_sched_worker_t* w, xce_sched_core_t* core,
xce_sched_ctx_t* ctx)
{
xce_sched_obj_t* schedobj = ctx->schedobj;
xce_primary_resreq_t* primary = ctx->primary;
int rc;
xvec_t* uncomit = NULL;
/* scheduler-sim-live-allocation spec, task 2.1 (Core-Safe Allocation
* Chain): sched_try_alloc_on_host_() itself is now core-typed, so its
* calls into sched_meet_host_selection()/plugin_prepare_allocation()/
* sched_check_host_satisfy() (all already core-typed, tasks 1.2/1.3)
* pass `core` straight through -- no more &(sched->core) wrap. */
rc = sched_meet_host_selection(core, w, primary->select);
if (rc == 0) {
if (ctx->reason == NULL) {
ctx->reason = make_pendreason(0);
}
if (ctx->reason == NULL) {
return -errno;
}
add_job_preason(ctx->reason->reasons, PREASON_MATCH_FAIL, w->name,
NULL);
return 0;
}
rc = sched_ctx_get_uncomit_host(&uncomit, ctx, w->idx);
if (rc < 0) {
return rc;
}
ctx->worker = w;
rc = plugin_prepare_allocation(core, ctx);
if (rc < 0) {
ctx->worker = NULL;
return rc;
}
/* bug-1884: the check (re)initializes unmeet itself */
xvec_idval_fini(&(ctx->unmeet));
rc = sched_check_host_satisfy(core, w, uncomit,
&(ctx->alloc_req.host), &(ctx->unmeet));
if (rc < 0) {
/* FIXME: */
return rc;
}
if (rc == 0) {
int i;
if (ctx)
for (i = 0; i < ctx->unmeet.count; i++) {
xce_resource_t* r = NULL;
rc = vget(core->resmgr, ((int*)ctx->unmeet.idxs.data)[i],
(void**)&r);
if (rc < 0) {
continue;
}
add_job_preason(ctx->reason->reasons, PREASON_NO_RESOURCE,
w->name, r->name);
}
return 0;
}
rc = xvec_idval(uncomit, &(ctx->alloc_req.host), XVEC_ADD_FLOAT);
if (rc < 0) {
return rc;
}
/* host is OK */
return 1;
}
/*
* sched_ctx_uncommit_host - reverses exactly one provisional per-host
* resource debit a prior, successful sched_try_alloc_on_host_() call made
* for host `idx`, undoing the XVEC_ADD_FLOAT commit
* (`xvec_idval(uncomit, &(ctx->alloc_req.host), XVEC_ADD_FLOAT)`, above)
* with the exact inverse operation against the SAME per-host uncommitted
* vector (`ctx->uncomitted.hosts[idx]`, via sched_ctx_get_uncomit_host() --
* the same get-or-create accessor the commit path itself uses, so a
* reversal always lands on the identical xvec_t instance the debit did).
*
* `committed` must be the exact xvec_idval_t snapshot the earlier commit
* added (e.g. the `claimed[]` entry's saved `alloc_req.host`, per
* requirements.md's "Rollback scope" -- ctx->alloc_req is reset and
* repopulated for every subsequent host tried, so callers must retain
* their own copy rather than reading ctx->alloc_req.host after the fact).
*
* Scoped entirely to ctx->uncomitted (per-job-allocation-attempt scratch
* state) -- never touches the real, session-wide resInUse/sharedInUse
* books (see resource-accounting.md / requirements.md's "Rollback scope").
*
* Returns the same contract as xvec_idval(): >=0 (count of entries
* applied) on success, negative errno on failure (e.g. -EINVAL if ctx/idx
* are invalid, propagated from sched_ctx_get_uncomit_host()).
*/
int sched_ctx_uncommit_host(xce_sched_ctx_t* ctx, int idx,
xvec_idval_t* committed)
{
xvec_t* uncomit = NULL;
int rc;
rc = sched_ctx_get_uncomit_host(&uncomit, ctx, idx);
if (rc < 0) {
return rc;
}
return xvec_idval(uncomit, committed, XVEC_SUB_FLOAT);
}
/*
* bug-1799 fix (2026-09-11): `w->resInUse` is a per-round "reset, then
* fully re-derive from every currently-RUN job" quantity -- core_session.c's
* core_reset_workers() (the reset half) deliberately SKIPS zeroing a
* worker's resInUse while it is IS_WORKER_UNAVAIL/IS_WORKER_RECONCILING
* (bug-1751, worker-rejoin-resync task 5), so its last known-good in-flight
* usage survives until the worker's real state is confirmed. sched_account_
* jobs() (scheduler.c) is the re-derive half -- called once per round via
* this function (sched_charge_account()) -- and had NO matching skip: it
* unconditionally re-added every RUN job's charge on top of whatever the
* (possibly not-reset) resInUse already held, so a worker that stayed
* RECONCILING/UNAVAIL across several rounds had its resInUse grow by one
* extra full charge PER ROUND instead of being cleanly re-summed --
* eventually exceeding the resource's real capacity (live-observed: a
* 22-slot worker showing used=26.00). Skipping the per-host charge for
* exactly the same two status bits core_reset_workers() already skips the
* reset for restores the "resum only what reset also touched" symmetry:
* while a worker is UNAVAIL/RECONCILING, its resInUse is simply left
* FROZEN at its last value (neither reset nor re-summed) until the status
* bit clears, at which point normal reset+resum resumes and self-corrects
* on the very next round. Scoped to the per-HOST loop only -- core->
* sharedInUse's own reset (core_reset_shared_resval(), core_session.c) is
* unconditional (no per-worker gating exists for it), so the shared-
* resource charge two lines below is intentionally left untouched.
*/
static int sched_update_account(xce_sched_core_t* core,
xce_job_account_t* account, int charge)
{
int i;
sched_update_res_accounts(core, &(core->sharedInUse), 1,
account->nres_shared, account->accounts_shared,
charge);
for (i = 0; i < account->nhosts; i++) {
xce_host_account_t* haccount = account->accounts[i];
xce_sched_worker_t* w = NULL;
if (haccount == NULL) {
continue;
}
w = sched_get_worker(core, haccount->host);
if (w == NULL) {
continue;
}
if (IS_WORKER_UNAVAIL(w->status) || IS_WORKER_RECONCILING(w->status)) {
continue;
}
sched_update_res_accounts(core, &(w->resInUse), 0, haccount->naccount,
haccount->accounts, charge);
}
return 0;
}
int sched_charge_account(xce_sched_core_t* core, xce_job_account_t* acct)
{
return sched_update_account(core, acct, 1);
}
int sched_uncharge_account(xce_sched_core_t* core, xce_job_account_t* acct)
{
return sched_update_account(core, acct, 0);
}
int sched_commit_job_alloc(xce_sched_core_t* core, xce_sched_job_t* job)
{
if (job->accounts == NULL) {
return 0;
}
return sched_charge_account(core, job->accounts);
}
/*
* find_host_account_by_resname - the xce_res_account_t within `hacct`
* matching `resname` (e.g. "mem"), or NULL if no such entry exists yet.
*/
static xce_res_account_t*
find_host_account_by_resname(xce_host_account_t* hacct, const char* resname)
{
int i;
if ((hacct == NULL) || (resname == NULL)) {
return NULL;
}
for (i = 0; i < hacct->naccount; i++) {
xce_res_account_t* a = hacct->accounts[i];
if ((a != NULL) && (a->resname != NULL)
&& (strcmp(a->resname, resname) == 0)) {
return a;
}
}
return NULL;
}
/*
* ensure_host_account_by_resname - find_host_account_by_resname(), or
* (resource-accounting.md correction) CREATE a fresh reserved=0 entry for
* `resname` if the job never requested it via rusage[].
*
* A job with no rusage[mem=...] clause genuinely reserves nothing -- but
* this codebase's admission check (sched_check_host_satisfy_(), above)
* never reads a live OS "how much is actually free" figure: it compares
* the worker's self-reported STATIC total (w->resvals) against this
* scheduler's OWN running tally (w->resInUse), built entirely by summing
* every job's xce_res_account_t.used+reserved. A reservation of 0 does NOT
* mean nothing needs charging -- charged = max(reserved, actual) = actual
* when reserved is 0, same formula, and that actual usage is real memory
* no other job can also use. Skipping it (as an earlier version of this
* function did) let an unreserved job's genuine consumption stay
* permanently invisible to w->resInUse -- the scheduler would believe
* memory was free that a live, unreserved job was actually sitting on, a
* real over-subscription risk, not merely a cosmetic gap.
*
* The new entry's residx is resolved the same way
* sched_update_res_accounts() resolves a fresh account's residx above --
* key_find() against core->resmgr, same lock discipline. Returns NULL
* (nothing to charge, not an error) if `resname` isn't even a registered
* resource in this cluster's config; the caller must treat that as a
* no-op, matching find_host_account_by_resname()'s own contract.
*/
static xce_res_account_t*
ensure_host_account_by_resname(xce_sched_core_t* core,
xce_host_account_t* hacct, const char* resname)
{
xce_res_account_t* entry;
xce_res_account_t** grown;
xce_resource_t* r = NULL;
int rc = 0;
entry = find_host_account_by_resname(hacct, resname);
if (entry != NULL) {
return entry;
}
lock(core->resmgr, RTTI_LOCK_READ);
rc = key_find(core->resmgr, (void*)resname, (void**)&r);
unlock(core->resmgr);
if ((rc < 0) || (r == NULL)) {
return NULL;
}
entry = (xce_res_account_t*)calloc(1, sizeof(*entry));
if (entry == NULL) {
return NULL;
}
entry->resname = strdup(resname);
entry->residx = r->idx;
entry->used = 0.0;
entry->reserved = 0.0;
entry->allocated = 0.0;
grown = (xce_res_account_t**)realloc(
hacct->accounts, (size_t)(hacct->naccount + 1) * sizeof(*grown));
if (grown == NULL) {
FREEUP(entry->resname);
FREEUP(entry);
return NULL;
}
hacct->accounts = grown;
hacct->accounts[hacct->naccount] = entry;
hacct->naccount++;
return entry;
}
/*
* sched_reconcile_host_resource_usage - resource-accounting.md's
* reservation-vs-actual-usage policy for one REAL (gauge-style, e.g. "mem")
* per-host resource: updates `acct_entry->used` to the OVERAGE beyond its
* reservation (max(0, actual_usage - reserved)) and applies just that
* change as a DELTA to `w`'s resInUse, so the pool ends up charged
* `used + reserved` == `max(reserved, actual)` for this resource -- per
* sched_update_res_accounts()'s own existing charge formula above, which
* this function does not duplicate or bypass, only feeds correctly.
*
* Deliberately NOT a full uncharge-then-recharge of the whole
* xce_job_account_t (sched_uncharge_account()+sched_charge_account()) --
* that would re-touch every OTHER resource charged in the same account
* (e.g. "slots") for no reason, since only THIS resource's usage changed.
* A single get_resvals()/put_resvals() delta on the one affected residx is
* the minimal correct update.
*
* A no-op (returns 0, `used` unchanged) when the computed overage is
* unchanged from what's already charged -- the common case, since most
* rusage ticks report usage still below the reservation.
*
* bug-1814 follow-up fix (2026-09-15): must skip entirely while the worker
* is IS_WORKER_UNAVAIL/IS_WORKER_RECONCILING, mirroring sched_update_
* account()'s own bug-1801 guard above -- this is a THIRD independent
* w->resInUse-mutating path (alongside charge/uncharge) that bug-1801 did
* not cover. Without this, a RECONCILING worker's resInUse -- deliberately
* left FROZEN by core_reset_workers() (bug-1751) and now also by sched_
* update_account() (bug-1801), so a clean per-round rebuild can restore it
* once the worker's real state is confirmed -- keeps getting mutated by
* every rusage tick that arrives during the same window regardless. Live-
* confirmed root cause of a job's "mem used" figure that grows and never
* self-corrects until the job finishes: a JOBS_PUSH resync merge occurring
* in the same RECONCILING window can replace a job's xce_res_account_t
* with a freshly-built one (`.used=0.0`, core_events.c's JOBS_PUSH RUN-
* merge branch) without uncharging the old one first (safe ONLY because
* the next clean reset+resum round is supposed to correct it -- exactly
* the round that stays skipped while RECONCILING). The very next rusage
* tick then computes its delta against that fresh 0.0 baseline and adds
* the full usage on top of whatever is already frozen in the pool, instead
* of leaving the pool alone until the worker's status clears and a normal
* rebuild can reconcile everything at once.
*/
int sched_reconcile_host_resource_usage(xce_sched_core_t* core,
xce_sched_worker_t* w,
xce_res_account_t* acct_entry,
double actual_usage)
{
double new_used;
double delta;
int idx[1];
float val[1];
if ((core == NULL) || (w == NULL) || (acct_entry == NULL)) {
return -EINVAL;
}
if (IS_WORKER_UNAVAIL(w->status) || IS_WORKER_RECONCILING(w->status)) {
return 0;
}
new_used = actual_usage - acct_entry->reserved;
if (new_used < 0.0) {
new_used = 0.0;
}
delta = new_used - acct_entry->used;
if (delta == 0.0) {
return 0;
}
idx[0] = acct_entry->residx;
get_resvals(&(w->resInUse), 1, idx, val);
val[0] += (float)delta;
put_resvals(&(w->resInUse), 1, idx, val);
acct_entry->used = new_used;
return 0;
}
/*
* sched_reconcile_job_resource_usage - resource-accounting.md's policy,
* applied to a job's per-host account from a raw rusage report
* (xce_job_rusages_t, the worker's own wire shape forwarded verbatim to the
* scheduler by sched_job_rusage_update_fun(), controller/sched.c).
*
* Scoped to SINGLE-HOST jobs only (job->accounts->nhosts == 1): the wire
* report carries no host field at all (update_job_rusages(), controller/
* job.c, derives the reporting host locally from the endpoint the request
* arrived on and never attaches it to the forwarded xce_job_rusages_t), so
* there is no way to attribute a report to one specific host of a
* multi-host job today. That is a real, separate gap in the wire protocol
* (a host-attribution gap, not a reservation-accounting gap) -- out of
* this function's scope. A multi-host job's memory stays charged at its
* flat per-host reservation, exactly as it always has; nothing here makes
* that worse.
*
* Only resources this project can reconcile against a live OS measurement
* go through this path -- currently just "mem" (SCHED_MEM_RESOURCE_NAME
* below), matched against the worker's real wire resname for it
* (SCHED_RUSAGE_MEM_RESNAME) -- same worker-resname-vs-scheduler/
* controller-resource-name split bug-1735 already established for the
* controller side, mapped explicitly here rather than unified across the
* codebase (out of this fix's scope too).
*
* A job that never requested rusage[mem=...] gets reserved=0, NOT skipped
* entirely: ensure_host_account_by_resname() creates the missing account
* entry on first use rather than treating "no reservation" as "nothing to
* charge" -- see that function's own doc comment for why this codebase's
* admission check specifically requires it (no live OS "available" read
* anywhere in the path; every job's real usage must be reflected in
* w->resInUse, reserved or not). The only genuine no-op is when "mem"
* isn't even a registered resource in this cluster's config at all.
*/
#define SCHED_RUSAGE_MEM_RESNAME "memory_rss"
#define SCHED_MEM_RESOURCE_NAME "mem"
int sched_reconcile_job_resource_usage(xce_sched_core_t* core,
xce_job_account_t* jacct,
xce_job_rusages_t* rusages)
{
xce_host_account_t* hacct;
xce_res_account_t* mem_acct;
xce_sched_worker_t* w;
int i;
if ((core == NULL) || (jacct == NULL) || (rusages == NULL)) {
return -EINVAL;
}
if (jacct->nhosts != 1) {
return 0;
}
hacct = jacct->accounts[0];
if (hacct == NULL) {
return 0;
}
mem_acct = ensure_host_account_by_resname(core, hacct,
SCHED_MEM_RESOURCE_NAME);
if (mem_acct == NULL) {
return 0;
}
w = sched_get_worker(core, hacct->host);
if (w == NULL) {
return -ENOENT;
}
for (i = 0; i < rusages->count; i++) {
xce_rusage_t* ru = rusages->usages[i];
if ((ru == NULL) || (ru->resname == NULL)
|| (strcmp(ru->resname, SCHED_RUSAGE_MEM_RESNAME) != 0)) {
continue;
}
return sched_reconcile_host_resource_usage(core, w, mem_acct,
(double)ru->value);
}
return 0;
}
static float get_resval_by_id(xvec_idval_t* vec, int idx)
{
int i;
for (i = 0; i < vec->count; i++) {
int residx;
xvec_get(&(vec->idxs), i, &residx);
if (residx == idx) {
float resval;
xvec_get(&(vec->vals), i, &resval);
return resval;
}
}
return 0.0;
}
/*
* sched_fill_limit_requests - for each matched limit, compute what THIS
* job would consume against it and store it into the limit's own virtual
* resource entry, ready for the generic satisfy-check machinery to compare
* against the limit's cap like any other resource.
*
* A limit's own scope (global vs host -- which of alloc_req->shared/.host
* its OWN virtual resource entry, vres->vres, gets written into) is
* independent of its UNDERLYING resource's (vres->res) own nature: "slots"
* is a per-host resource (not XCE_RESF_GLOBAL) even when the limit built on
* top of it is global (e.g. limits.yaml's ug1_normal_slots, summing slots
* across the whole cluster) -- bug-1789/1790. Both are derivable from each
* resource's own XCE_RESF_GLOBAL flag: `vres->res`'s flag is the SAME one
* calc_wanted_global_resvals()/calc_wanted_host_resvals() (resreq.c)
* dispatch on to decide where THEY write a resource's want; `vres->vres`'s
* flag was set by instantiate_limit()->add_vres() from the exact same
* is_host_based_limit() call that decided whether this limit landed in
* obj->glimits or obj->hlimits in the first place.
*
* Three of the four (scope, underlying-resource) combinations read the
* underlying resource's want straight out of alloc_req, already populated
* by whichever allocator plugin ran (calc_wanted_global_resvals()/
* calc_wanted_host_resvals()). The fourth -- a GLOBAL limit on a per-host
* underlying resource -- is different in kind, not just in which vector to
* read: alloc_req->host only ever holds per-host/per-bucket-specific data
* for the REAL per-host capacity check (a different question, with a
* different answer, computed fresh per candidate host), never a job-WIDE
* total -- EXCEPT for "slots", which seqjob_prepare_allocation()/
* parjob_prepare_allocation() both unconditionally stage into alloc_req
* ->host regardless of phase (their own "add slots" step, outside the
* normal rusages-driven computation) -- so for "slots" specifically,
* alloc_req->host during the global phase already holds exactly the
* job-wide total this combination needs, the same way it would for the
* general per-host-phase case. Reading alloc_req->host FIRST therefore
* naturally gets "slots" right with no name-based special case at all
* (bug-1790 refactor regression, live-confirmed against gitea#46's
* original ug1_slots/ug1_normal_slots/limit1 scenario: an earlier version
* of this fix instead special-cased strcmp(vres->res->name, "slots") to
* fall back to `nslots` directly -- functionally equivalent, since
* alloc_req->host already holds exactly that value for "slots", but a
* needless string-name special case where reading the data that's already
* there does the same job). Any OTHER per-host resource genuinely has
* nothing staged in alloc_req->host during the global phase (calc_wanted_
* host_resvals(), the thing that WOULD stage it, only ever runs in the
* per-host branch), so the read comes back 0 (get_resval_by_id()'s "not
* found" value -- indistinguishable from a genuine want of 0, but "slots"
* is never legitimately 0 here: seqjob only ever runs for maxTasks==1,
* parjob only for maxTasks>1, so nslots is always >=1) and `rusages`/
* `nhosts`/`nslots` (the job's own declared resource requests, and its
* whole-job totals -- not any one host's share) are used to compute the
* aggregate directly, on demand, via calc_wanted_job_total_resval()
* (resreq.c) instead -- deliberately kept as this function's own
* responsibility rather than the allocator plugins': what a LIMIT needs to
* see is this function's concern, not seqjob/parjob's, which should only
* ever compute what the job itself wants for capacity-satisfy purposes.
*/
int sched_fill_limit_requests(xce_sched_alloc_req_t* alloc_req,
xlist_t* limits, xlist_t* rusages, int nhosts,
int nslots)
{
int i;
for (i = 0; i < limits->count; i++) {
xce_vres_t* vres = (xce_vres_t*)xlist_get(limits, i);
xvec_idval_t* dst;
int src_is_global;
int dst_is_global;
float val = 0.0;
if (vres == NULL) {
continue;
}
src_is_global = (vres->res->flags & XCE_RESF_GLOBAL) != 0;
dst_is_global = (vres->vres->flags & XCE_RESF_GLOBAL) != 0;
dst = dst_is_global ? &(alloc_req->shared) : &(alloc_req->host);
if (dst_is_global && !src_is_global) {
val = get_resval_by_id(&(alloc_req->host), vres->res->idx);
if (val == 0.0) {
calc_wanted_job_total_resval(rusages, PER_TASK, nhosts,
nslots, vres->res->name, &val);
}
} else {
xvec_idval_t* src
= src_is_global ? &(alloc_req->shared) : &(alloc_req->host);
val = get_resval_by_id(src, vres->res->idx);
}
xvec_idval_store(dst, vres->vres->idx, val);
}
return 0;
}