#define _XOPEN_SOURCE 700 #include #include #include #include #include #include "kvmgr.h" #include "resreq.h" #include "rusage.h" #include "section.h" #include "utils.h" #include "xce/types.h" #include "xce_resreq.h" #include "xce_select_peg.h" #include "xlist.h" #include "xvec.h" /* ── resreq subsystem lifecycle ───────────────────────────────────────────── */ void resreq_init(int optimize_level) { sel_optimize_set(optimize_level); sel_node_table_init(); } void resreq_fini(void) { sel_node_table_fini(); } /* ── type helpers ─────────────────────────────────────────────────────────── */ res_type_t valtype_to_res_type(int v) { switch (v) { case XCE_VAL_NUMERIC: return RES_TYPE_NUMERIC; case XCE_VAL_STRING: return RES_TYPE_STRING; case XCE_VAL_BOOLEAN: return RES_TYPE_BOOL; default: ERROR("valtype_to_res_type: unknown valtype %d", v); abort(); } } static int get_parse_errmsg(char** errmsg, const char* data, int pos) { size_t bufsize = 0; FILE* mem_stream = NULL; int i; if ((errmsg == NULL) || (data == NULL) || (pos < 0)) { return -EINVAL; } mem_stream = open_memstream(errmsg, &bufsize); if (mem_stream == NULL) { return -errno; } fprintf(mem_stream, "%s\n", data); for (i = 0; i < pos - 1; i++) { fprintf(mem_stream, " "); } fprintf(mem_stream, "^Error\n"); fflush(mem_stream); fclose(mem_stream); return 1; } void kill_resreq_data(xce_resreq_data_t* data) { if (data == NULL) { return; } FREEUP(data->selectString); FREEUP(data->rusageString); FREEUP(data->spanString); FREEUP(data->sameString); FREEUP(data->affinityString); } void kill_section_ctx(section_context_t* ctx) { if (ctx == NULL) { return; } kill_resreq_data(&(ctx->data)); } int parse_section(xce_resreq_data_t* resreq, char* data, char** errmsg) { section_context_t ctx = { 0 }; xce_section_context_t* p = NULL; int rc = 1; if ((resreq == NULL) || (data == NULL) || (errmsg == NULL)) { return -EINVAL; } *errmsg = NULL; ctx.input = data; memset(resreq, 0x0, sizeof(*resreq)); p = xce_section_create(&ctx); if (p == NULL) { return -errno; } xce_section_parse(p, NULL); if (ctx.has_error) { get_parse_errmsg(errmsg, ctx.input, ctx.err_pos); rc = -EINVAL; } else { MOVE_PTR(resreq->selectString, ctx.data.selectString); MOVE_PTR(resreq->rusageString, ctx.data.rusageString); MOVE_PTR(resreq->spanString, ctx.data.spanString); MOVE_PTR(resreq->sameString, ctx.data.sameString); MOVE_PTR(resreq->affinityString, ctx.data.affinityString); } kill_section_ctx(&ctx); xce_section_destroy(p); return rc; } void kill_rusage_value(rusage_value_t* val) { if (val == NULL) { return; } FREEUP(val->res); } void kill_rusage_ctx(rusage_context_t* ctx) { if (ctx == NULL) { return; } return; } int parse_rusage(xlist_t* rusages, char* data, char** errmsg) { rusage_context_t ctx = { 0 }; xce_rusage_context_t* p = NULL; int rc = 1; if ((rusages == NULL) || (data == NULL) || (errmsg == NULL)) { return -EINVAL; } *errmsg = NULL; ctx.input = data; ctx.rusages = rusages; p = xce_rusage_create(&ctx); if (p == NULL) { return -errno; } xce_rusage_parse(p, NULL); if (ctx.has_error) { get_parse_errmsg(errmsg, ctx.input, ctx.err_pos); rc = -EINVAL; /* FIXME: need to freeup partial data in rusages */ } kill_rusage_ctx(&ctx); xce_rusage_destroy(p); return rc; } /* Depth pre-scan: count '!' and '(' as depth increments. * Returns 0 if depth is within limit, -1 if exceeded. */ static int check_select_depth(const char* input, int max_depth) { int depth = 0; const char* p; for (p = input; *p; p++) { if (*p == '!' || *p == '(') { depth++; if (depth > max_depth) return -1; } else if (*p == ')') { if (depth > 0) depth--; } } return 0; } static int parse_select(xce_primary_resreq_t* primary, char* selectString, var resmgr, char** errmsg) { select_context_t ctx; xce_select_context_t* p = NULL; int rc = 1; /* Without a resource manager, resource names cannot be resolved. * Leave primary->select = NULL and succeed — callers that need * select evaluation must supply a valid resmgr. */ if (resmgr == NULL) { return 1; } memset(&ctx, 0, sizeof(ctx)); /* Length guard */ if (strlen(selectString) > SELECT_MAX_INPUT_LEN) { *errmsg = strdup("select expression exceeds maximum length"); return -EINVAL; } /* Depth guard */ if (check_select_depth(selectString, SELECT_MAX_DEPTH) < 0) { *errmsg = strdup("select expression exceeds maximum nesting depth"); return -EINVAL; } ctx.resmgr = resmgr; ctx.input = selectString; p = xce_select_create(&ctx); if (p == NULL) return -errno; /* LC_NUMERIC guard: strtod/snprintf in PEG actions must use '.' as the * decimal separator regardless of the calling thread's locale setting. */ locale_t cloc = newlocale(LC_NUMERIC_MASK, "C", (locale_t)0); if (cloc == (locale_t)0) { xce_select_destroy(p); if (sel_get_optimize_level() > 0) { ERROR("newlocale(LC_NUMERIC_MASK,C) failed: %s — scheduler cannot " "safely parse numeric resreq values", strerror(errno)); abort(); } ERROR("newlocale(LC_NUMERIC_MASK,C) failed: %s — rejecting resreq parse", strerror(errno)); *errmsg = strdup("locale initialization failed"); return -EINVAL; } locale_t old = uselocale(cloc); xce_select_parse(p, NULL); uselocale(old); freelocale(cloc); if (ctx.has_error || ctx.root == NULL) { if (ctx.errmsg[0] != '\0') *errmsg = strdup(ctx.errmsg); else get_parse_errmsg(errmsg, selectString, ctx.err_pos); rc = -EINVAL; goto cleanup_parser; } /* Phase 4c: flatten AND/OR right-associative chains to left-associative */ { int flat_changed; do { flat_changed = 0; ctx.root = sel_flatten_pass(ctx.root, &flat_changed); if (ctx.root == NULL) { rc = -ENOMEM; goto cleanup_parser; } } while (flat_changed); } /* Phase 4d: const-fold (single pass — does not create new flatten opportunities) */ ctx.root = sel_const_fold_pass(ctx.root); if (ctx.root == NULL) { rc = -ENOMEM; goto cleanup_parser; } { xce_select_t* sel = calloc(1, sizeof(*sel)); if (sel == NULL) { if (sel_get_optimize_level() == 0) sel_free_tree(ctx.root); rc = -ENOMEM; } else { sel->root = ctx.root; primary->select = sel; } } cleanup_parser: xce_select_destroy(p); return rc; } /* Req 7.1 (scheduler-allocation spec): stable resource-usage fingerprint * over a primary resreq's parsed `rusages` list. * * Style: FNV-1a 64-bit, consistent with this spec's cache-key hash * (requirements.md "New types (Req 7)", cache_key_hash_ -- FNV-1a 32-bit * over xce_cache_key_t, a later task's own scope). No existing hash-update * helper for this purpose exists anywhere in this codebase; this is new, * self-contained, file-local code. * * Design choice -- ORDER-DEPENDENT: each rusage_value_t is folded into the * hash in xlist storage order (the order its `rusage[...]` clause was * parsed in), rather than sorted/combined commutatively. This is simpler * (no auxiliary sort, no need for a commutative combine step) and is * sufficient for Req 7.1's actual requirement: parsing the *same* resreq * string twice always produces the same xlist order, so the fingerprint * is stable for identical input, which is all the unsatisfiable-resreq * cache needs (a false miss on a reordered-but-equivalent resreq costs one * redundant, still-correct allocation attempt; it can never cause a false * hit). * * Field selection -- deliberately per-field, NOT a raw memcmp/byte-hash of * `sizeof(rusage_value_t)`: `rusage_value_t.res` is a `char*` (owned, * strdup'd per parse -- see kill_rusage_value()'s plain `FREEUP(val->res)`, * not an xstring-interned/refcounted pointer). Hashing the raw struct * bytes would hash the POINTER VALUE (heap address), which differs across * separate parses of identical content and would make the "fingerprint" * unstable by construction. Each field is hashed individually: the * resource name's string CONTENT (not its pointer), then the raw bytes of * `value`/`per_flag`/`duration`/`decay_flag` (safe to hash as raw bytes -- * none of those are pointers). */ static uint64_t hash_rusages(xlist_t* rusages) { uint64_t h = 14695981039346656037ULL; /* FNV-1a 64-bit offset basis */ int i; if (rusages == NULL) { return h; } for (i = 0; i < rusages->count; i++) { rusage_value_t* v = (rusage_value_t*)xlist_get(rusages, i); const unsigned char* p; size_t n; size_t j; if (v == NULL) { continue; } /* Resource name: hash the string content, not the pointer. */ p = (v->res != NULL) ? (const unsigned char*)v->res : (const unsigned char*)""; n = (v->res != NULL) ? strlen(v->res) : 0; for (j = 0; j < n; j++) { h = (h ^ (uint64_t)p[j]) * 1099511628211ULL; } /* Field separator -- keeps a name's trailing bytes from aliasing * with the numeric fields that follow (e.g. "ab"+1 vs "a"+"b1"). */ h = (h ^ 0xFFu) * 1099511628211ULL; p = (const unsigned char*)&v->value; for (j = 0; j < sizeof(v->value); j++) { h = (h ^ (uint64_t)p[j]) * 1099511628211ULL; } p = (const unsigned char*)&v->per_flag; for (j = 0; j < sizeof(v->per_flag); j++) { h = (h ^ (uint64_t)p[j]) * 1099511628211ULL; } p = (const unsigned char*)&v->duration; for (j = 0; j < sizeof(v->duration); j++) { h = (h ^ (uint64_t)p[j]) * 1099511628211ULL; } p = (const unsigned char*)&v->decay_flag; for (j = 0; j < sizeof(v->decay_flag); j++) { h = (h ^ (uint64_t)p[j]) * 1099511628211ULL; } } return h; } int parse_resreq_primary(xce_primary_resreq_t* primary, char* resreqString, var resmgr, char** errmsg) { int rc; xce_resreq_data_t data; if ((primary == NULL) || (resreqString == NULL) || (errmsg == NULL)) { return -EINVAL; } memset(primary, 0x0, sizeof(*primary)); rc = parse_section(&data, (char*)resreqString, errmsg); if (rc < 0) { return rc; } /* Parse select clause BEFORE kill_resreq_data (data.selectString freed there) */ if (data.selectString != NULL && data.selectString[0] != '\0') { rc = parse_select(primary, data.selectString, resmgr, errmsg); if (rc < 0) { kill_resreq_data(&data); return rc; } } xlist_init(&(primary->rusages), 8); if ((data.rusageString != NULL) && (strlen(data.rusageString) != 0)) { rc = parse_rusage(&(primary->rusages), data.rusageString, errmsg); if (rc < 0) { kill_resreq_data(&data); return rc; } } /* FIXME: add others later */ kill_resreq_data(&data); primary->minTasks = 1; primary->maxTasks = 1; /* Req 7.1: compute once, at parse time -- stable for the life of this * resreq, reused by every later cache lookup instead of recomputing. */ primary->rusage_hash = hash_rusages(&(primary->rusages)); return rc; } void kill_resreq_ctx(resreq_context_t* ctx) { if (ctx == NULL) { return; } } int parse_resreq(xce_resreq_t* resreq, char* data, var resmgr, char** errmsg) { resreq_context_t ctx = { 0 }; xce_resreq_context_t* p = NULL; int rc = 1; if ((resreq == NULL) || (data == NULL) || (errmsg == NULL)) { return -EINVAL; } *errmsg = NULL; memset(resreq, 0x0, sizeof(*resreq)); ctx.resmgr = resmgr; ctx.input = data; ctx.resreq = resreq; p = xce_resreq_create(&ctx); if (p == NULL) { return -errno; } xce_resreq_parse(p, NULL); if (ctx.has_error) { get_parse_errmsg(errmsg, ctx.input, ctx.err_pos); rc = -EINVAL; /* FIXME: need to freeup partial data in resreqs */ } kill_resreq_ctx(&ctx); xce_resreq_destroy(p); return rc; } /* * TaskCountWiring (resreq-maxtasks-wiring spec): overwrite a standalone * primary resreq's minTasks/maxTasks with the job's real requested task * count. A COMPOUND/ALTER resreq's nested primaries are left exactly as * parse_resreq_primary() set them -- only the job's own top-level * standalone primary is in scope (design.md's TaskCountWiring component; * requirements 1.1, 1.2, 1.5, 3.3). * * Narrowing conversion note: mintasks/maxtasks are uint32_t, matching * job_submit_data_t.mintasks/maxtasks (the caller's own source of truth); * xce_primary_resreq_t.minTasks/maxTasks are plain int. This is a * deliberate, one-time, acknowledged narrowing assignment -- safe for * realistic task counts (well under INT_MAX) -- not a mismatch to "fix" * by changing either type. */ void apply_job_task_count(xce_resreq_t* resreq, uint32_t mintasks, uint32_t maxtasks) { if (resreq == NULL || resreq->tag != XCE_RESREQ_PRIMARY) { return; } /* Defensive clamp: mirror convert.go's own clamp direction -- raise * the max, never lower the min. The apiserver already guarantees * mintasks <= maxtasks twice before a job reaches this layer; this is * a redundant safety net, not a new job-rejecting error path. */ if (mintasks > maxtasks) { maxtasks = mintasks; } resreq->u.primary->minTasks = (int)mintasks; resreq->u.primary->maxTasks = (int)maxtasks; } void kill_primary_resreq(xce_primary_resreq_t* resreq) { int i; if (resreq == NULL) { return; } if (resreq->select != NULL) { xce_select_node_t* root = resreq->select->root; resreq->select->root = NULL; if (sel_get_optimize_level() == 0 && root != NULL) sel_free_tree(root); FREEUP(resreq->select); } resreq->select = NULL; for (i = 0; i < resreq->rusages.count; i++) { rusage_value_t* v = (rusage_value_t*)xlist_get(&(resreq->rusages), i); if (v == NULL) { continue; } kill_rusage_value(v); FREEUP(v); } xlist_fini(&(resreq->rusages)); } void kill_compound_resreq(xce_compound_resreq_t* resreq) { int i; if (resreq == NULL) { return; } for (i = 0; i < resreq->primaries.count; i++) { xce_primary_resreq_t* primary = (xce_primary_resreq_t*)xlist_get(&(resreq->primaries), i); if (primary == NULL) { continue; } kill_primary_resreq(primary); FREEUP(primary); } xlist_fini(&(resreq->primaries)); } void kill_alter_resreq(xce_alter_resreq_t* resreq) { int i; if (resreq == NULL) { return; } for (i = 0; i < resreq->compounds.count; i++) { xce_compound_resreq_t* c = (xce_compound_resreq_t*)xlist_get(&(resreq->compounds), i); if (c == NULL) { continue; } kill_compound_resreq(c); FREEUP(c); } xlist_fini(&(resreq->compounds)); } void kill_resreq(xce_resreq_t* resreq) { if (resreq == NULL) { return; } switch (resreq->tag) { case XCE_RESREQ_PRIMARY: kill_primary_resreq(resreq->u.primary); FREEUP(resreq->u.primary); break; case XCE_RESREQ_COMPOUND: kill_compound_resreq(resreq->u.compound); FREEUP(resreq->u.compound); break; case XCE_RESREQ_ALTER: kill_alter_resreq(resreq->u.alter); FREEUP(resreq->u.alter); break; default: break; } } static int print_rusages(FILE* fp, xlist_t* rusages, int level) { int i; fprintf(fp, "%*sRusages:\n", (int)(2 * level), ""); for (i = 0; i < rusages->count; i++) { rusage_value_t* v = (rusage_value_t*)xlist_get(rusages, i); if (v == NULL) { continue; } fprintf(fp, "%*s res %s value %f per %d duration %f decay %d\n", (int)(2 * (level + 1)), "", v->res, v->value, v->per_flag, v->duration, v->decay_flag); } return 0; } static int print_primary_resreq(FILE* fp, xce_primary_resreq_t* resreq, int level) { fprintf(fp, "%*sPrimary:\n", (int)(2 * level), ""); fprintf(fp, "%*sTasks: min = %d, max = %d\n", (int)(2 * (level + 1)), "", resreq->minTasks, resreq->maxTasks); #if 0 fprintf(fp, "%*Select: %s\n", (int)(2*(level +1)),"", "Add later"); #endif print_rusages(fp, &(resreq->rusages), level + 1); return 0; } static int print_compound_resreq(FILE* fp, xce_compound_resreq_t* resreq, int level) { int i; fprintf(fp, "%*sCompound:\n", (int)(2 * level), ""); for (i = 0; i < resreq->primaries.count; i++) { xce_primary_resreq_t* primary = (xce_primary_resreq_t*)xlist_get(&(resreq->primaries), i); if (primary == NULL) { continue; } print_primary_resreq(fp, primary, level + 1); } return 0; } static int print_alter_resreq(FILE* fp, xce_alter_resreq_t* resreq, int level) { int i; fprintf(fp, "%*sAlternative:\n", (int)(2 * level), ""); for (i = 0; i < resreq->compounds.count; i++) { xce_compound_resreq_t* c = (xce_compound_resreq_t*)xlist_get(&(resreq->compounds), i); if (c == NULL) { continue; } print_compound_resreq(fp, c, level + 1); } return 0; } static int print_resreq(FILE* fp, xce_resreq_t* resreq, int level) { if ((fp == NULL) || (resreq == NULL) || (level < 0)) { return -EINVAL; } switch (resreq->tag) { case XCE_RESREQ_PRIMARY: return print_primary_resreq(fp, resreq->u.primary, level); case XCE_RESREQ_COMPOUND: return print_compound_resreq(fp, resreq->u.compound, level); case XCE_RESREQ_ALTER: return print_alter_resreq(fp, resreq->u.alter, level); default: break; } return -EINVAL; } int format_resreq(char** buf, xce_resreq_t* resreq) { size_t bufsize = 0; FILE* fp = NULL; int rc = 0; if ((buf == NULL) || (resreq == NULL)) { return -EINVAL; } fp = open_memstream(buf, &bufsize); if (fp == NULL) { return -errno; } rc = print_resreq(fp, resreq, 0); fflush(fp); fclose(fp); return rc; } /* * calc_wanted_job_total_resvals_ - shared body for calc_wanted_global_resvals() * and calc_wanted_host_resval_job_totals(): computes each matching rusage * entry's JOB-WIDE TOTAL want (PER_TASK: value*nslots, PER_JOB: value, * PER_HOST: value*nhosts -- nslots/nhosts are the whole job's totals, not * any one host's share). `match_global` selects which resources are * "matching": nonzero processes only XCE_RESF_GLOBAL resources (the * original calc_wanted_global_resvals() behavior), zero processes only * non-GLOBAL (per-host) ones instead. */ static int calc_wanted_job_total_resvals_(xvec_idval_t* resvals, var resmgr, xlist_t* rusages, rusage_per_type_t perflag, int nhosts, int nslots, int match_global) { int i; for (i = 0; i < rusages->count; i++) { rusage_value_t* v = (rusage_value_t*)xlist_get(rusages, i); xce_resource_t* r = NULL; float val; rusage_per_type_t perflag1; int rc = 0; if (v == NULL) { continue; } lock(resmgr, RTTI_LOCK_READ); rc = key_find(resmgr, (void*)v->res, (void**)&r); unlock(resmgr); if (rc <= 0) { continue; } if (r == NULL) { continue; } if (match_global) { if ((r->flags & XCE_RESF_GLOBAL) == 0) { continue; } } else { if ((r->flags & XCE_RESF_GLOBAL) != 0) { continue; } } perflag1 = v->per_flag; if (perflag1 == PER_DEFAULT) { perflag1 = perflag; } switch (perflag1) { case PER_TASK: val = v->value * nslots; break; case PER_JOB: val = v->value; break; case PER_HOST: val = v->value * nhosts; break; default: /* something wrong */ val = 0.0; break; } xvec_idval_store(resvals, r->idx, val); } return 0; } int calc_wanted_global_resvals(xvec_idval_t* resvals, var resmgr, xlist_t* rusages, rusage_per_type_t perflag, int nhosts, int nslots) { return calc_wanted_job_total_resvals_(resvals, resmgr, rusages, perflag, nhosts, nslots, 1); } /* * calc_wanted_host_resval_job_totals - bug-1790: like * calc_wanted_global_resvals(), but for resources NOT flagged * XCE_RESF_GLOBAL (per-host resources) -- computes each such resource's * JOB-WIDE TOTAL want, not any one host's own share (calc_wanted_host_ * resvals(), below, computes that instead, for the real per-host capacity * check). Exists so a GLOBAL limit built on top of a per-host resource * (e.g. a future absolute global mem cap, "slots" already being the one * bug-1789 fixed) can read a real aggregate instead of an always-missing * value. * * seqjob_prepare_allocation()/parjob_prepare_allocation()'s global-phase * branch (ctx->worker == NULL) calls this into ctx->alloc_req.host * immediately alongside its existing calc_wanted_global_resvals() call. * This is safe despite .host normally being per-host/per-bucket-varying * data (see parjob_allocator.c's own per-bucket reset): plugin_prepare_ * allocation() (plugin.c) only resets .host at the START of a PER-HOST * call (ctx->worker != NULL), never during the global phase, so the value * written here survives long enough for sched_fill_limit_requests() * (resource.c) to read it -- called later in the SAME global-phase * invocation, for obj->glimits -- before any later per-host call wipes * and correctly repopulates .host with that specific host's own share via * calc_wanted_host_resvals(). */ int calc_wanted_host_resval_job_totals(xvec_idval_t* resvals, var resmgr, xlist_t* rusages, rusage_per_type_t perflag, int nhosts, int nslots) { return calc_wanted_job_total_resvals_(resvals, resmgr, rusages, perflag, nhosts, nslots, 0); } int calc_wanted_host_resvals(xvec_idval_t* resvals, var resmgr, xlist_t* rusages, rusage_per_type_t perflag, int nhosts, int nslots) { int i; for (i = 0; i < rusages->count; i++) { rusage_value_t* v = (rusage_value_t*)xlist_get(rusages, i); xce_resource_t* r = NULL; float val; rusage_per_type_t perflag1; int rc = 0; if (v == NULL) { continue; } lock(resmgr, RTTI_LOCK_READ); rc = key_find(resmgr, (void*)v->res, (void**)&r); unlock(resmgr); if (rc <= 0) { continue; } if (r == NULL) { continue; } if ((r->flags & XCE_RESF_GLOBAL) != 0) { continue; } perflag1 = v->per_flag; if (perflag1 == PER_DEFAULT) { perflag1 = perflag; } if (perflag1 == PER_JOB) { /* FIXME: PER_JOB should be global? */ continue; } switch (perflag1) { case PER_TASK: val = v->value * nslots; break; case PER_HOST: val = v->value; break; default: /* something wrong */ val = 0.0; break; } xvec_idval_store(resvals, r->idx, val); } return 0; }