merge: sync Rongsong's scheduler-framework (c5b172c7..3442a57d, 781 commits)
Merge origin/scheduler-framework into merge-sf-20260903 (97 commits on ours: bug-1490/1494/1498/1509/1513/1521/1522 fixes, PEND-reason P1, scheduler-agent Phase 2, bug-924 owner-pid shim). Bookkeeping excluded per policy: .kiro/, .wolf/, CHANGES.md, reports/ (92 files dropped); .claude/ + CLAUDE.md keep ours. Key theirs-side changes absorbed: - bug-1938/1960: QUEUE_STAT_CLOSED no longer gates dispatch (closed queue refuses new submissions only; pending jobs still run) -- supersedes our QUEUE_CLOSED preason on the dispatch path; sched_queue_status_reason drops that branch - bug-1846: config push now stashes and applies between scheduling rounds (sched_stash_config/sched_apply_pending_config), applied at the same main-thread round boundary that preserves bug-1513 serialization - bug-1760: sched_check_pending_rejoin exposed + xcecli_is_connection_broken public predicate - bug-1936: execuser/exechome/execcwd/exdata carried on submission (superset of our bug-1500 execuser fallback; our job-typed resolveExecUser removed as dead code) - bug-1945~1962: apiserver auth hardening (HS256 pinning, default- secret rejection, logout revocation, bcrypt, RBAC) -- our API-key branch folded into controllers.ParseTokenClaims so both coexist - bug-1736: rawRunUsage job run-usage fields in the Go model - xce_test relocated src/controller/testsuite/test.c -> src/harness/test.c; controller suite split into controller_test.c + fixtures; their clitest fixtures spawn xce_test mid-suite under `meson test -j` Ours preserved across the merge: - PEND-reason P1 (codes 1-9, preason.h, job_preason machinery) -- gate re-ported into their relocated allocator (sched_election.c) after plugin_job_election, with attach on the allocator thread only (bug-1521 race discipline) - PEND-reason P0 main_reason semantics RECONCILED to theirs: the merge initially re-carried our first-detail-add-promotes-main_reason fallback in add_job_preason (4e2ac86), but their whole trace/verify stack treats "detail-only => main_reason 0" as a contract -- sched_trace_capture counts main_reason as its own occurrence ("Expected 2 Was 3" in test_trace_capture), the outcome renderer prints "main reason: none" for detail-only PENDs (test_repl_trace_e2e), and the committed .replay/.sim_verify baselines encode main_reason=0 (test_repl_trace_baseline byte compare, test_trace_equivalence off-vs-traced). The fallback is dropped from the shared lib; the user-visible "overall reason" survives as an API-layer view: rawJobToModel (convert_pure.go) promotes the first detail row to main_reason_name when main_reason is 0 (main_reason VALUE marshals unchanged). add_job_preason keeps only the NULL-reasons guard - bug-1498: sched_job_is_ready rejects non-PEND/suspended jobs - bug-1490/1781: xce_jobctrl_t.action as capnp union discriminant (their jobCtrlUnionDiscriminant, same semantics) - xcecli typed PendingReason field (53c00834), bug-1522 reconnect resend, xcecli_force_reconnect + their is_connection_broken both exported in client.h/libclient - t1/t2/t3 test-node hostconfig in src/configs/hosts.yaml (bug-1494), /xce/v1/events + /xce/v1/apikeys whitelist routes (agent Phase 2) - lib ASAN_OPTIONS=detect_leaks=0 test env + their oom-test block - drop-audit: our bug-1504 fix covered by their bug-1680 (no action) bug-924 follow-up (replaces the67f3f6adscratch-dir wipe): the take-theirs audit's one "drop" was re-evaluated and REVERTED instead of re-applied -- a global `find /tmp -name 'xce_*' -exec rm -rf` at suite start/exit races the LIVE /tmp/xce_* fixtures of sibling suites under `meson test -j` (harness_unit_test's clitest fixtures spawn xce_test as a child mid-suite; the wipe destroyed the parent's fixtures -> test_test_cli failures). Coverage instead: - mkdtemp_clean shim gains mkdtemp_clean_mark/sweep_since (testsuite_support/mkdtemp_clean.[ch]) - new case_scratch_sweep.[ch]: aggregate runners (controller_test.c, harness_test.c -- hundreds of cases per process) call case_scratch_sweep_begin/_end_ok from setUp/tearDown; a PASSED case's dirs are swept immediately in tearDown (after engine reaping), FAILED cases keep their dirs for post-mortem; XCE_MKDTEMP_KEEP_ALL=1 disables. One full controller_unit_test run was observed to leak 12G+ to ENOSPC without this, cascading into unrelated "Expected Non-NULL" failures - harness_unit_test_deps links the shim library as well Test reconciliations on adopt: - core-ready-gates: sched_queue_is_ready now gates only INACT_ADMIN|INACT_WINDOW (bug-1938/1960); CLOSED queues elect -- covered end-to-end by test_queue_admin_control.c - test_worker_resync: the 4 restart-replay RUN assertions predate the bug-1509 B3 marking (8858afed) landing on the same lineage; replayed RUN jobs now assert RUN|UNKWN|TRANSIENT (TWR_ASSERT_REPLAYED_RUN); terminal DONE/EXIT assertions unchanged - convert_pure.go: our job-typed resolveExecUser deleted (superseded by their bug-1780/1936 two-arg version) - test_preason: the 3 P0-promotion assertions reverted to the detail-keeps-main-0 contract; new tests cover limit-code detail rows and explicit-main-wins-over-later-details; apiserver convert_test.go gains the detail-only name-promotion cases Co-Authored-By: Claude Code <noreply@anthropic.com>
This commit is contained in:
lei.chen
commit
0c8607345e
663 changed files with 139848 additions and 13614 deletions
|
|
@ -2,24 +2,16 @@ package backend
|
|||
|
||||
import (
|
||||
"apiserver/models"
|
||||
"apiserver/utils"
|
||||
"encoding/json"
|
||||
"time"
|
||||
"errors"
|
||||
"fmt"
|
||||
"regexp"
|
||||
"strconv"
|
||||
"strings"
|
||||
"time"
|
||||
)
|
||||
|
||||
// resolveExecUser returns the user a job should run as: an explicit exec_user
|
||||
// wins; otherwise the job runs as the submitting user. bug-1500: the submit
|
||||
// conversion previously dropped this entirely, so the worker received a NULL
|
||||
// execuser and ran every job as "nobody". Plain Go so it is unit-testable
|
||||
// without cgo — keep it OUT of the cgo-only convert.go.
|
||||
func resolveExecUser(job *models.JobInfo) string {
|
||||
execUser := job.ExecUser
|
||||
if execUser == "" {
|
||||
execUser = job.SubmitUser
|
||||
}
|
||||
return execUser
|
||||
}
|
||||
|
||||
// jobCtrlDiscriminant maps a models.JOB_* action to the capnp JobControl union
|
||||
// discriminant carried in xce_jobctrl_t.action: JobControl_u_siginfo (0) for
|
||||
// signal-based actions (kill/stop/resume), JobControl_u_ckptoption (1) for
|
||||
|
|
@ -128,12 +120,255 @@ func FormatJobID(uniqueJobID int64) string {
|
|||
clusterID, apiserverID, timestamp, sequence)
|
||||
}
|
||||
|
||||
// humanJobIDPattern matches a plain human-readable job ID, "{cluster}-
|
||||
// {apiserver}-{timestamp}-{sequence}" (no array-index brackets).
|
||||
var humanJobIDPattern = regexp.MustCompile(`^(\d+)-(\d+)-(\d+)-(\d+)$`)
|
||||
|
||||
// ParseHumanJobID parses a plain human-readable job ID (as produced by
|
||||
// FormatJobID, brackets already stripped by the caller) back into the
|
||||
// packed numeric job ID the job_id/unique_job_id DB columns store. Returns
|
||||
// ok=false if s doesn't match that shape (e.g. it's already numeric).
|
||||
//
|
||||
// Deliberately does not itself handle an "[idx]" array-child suffix:
|
||||
// controllers/jobs.go's parseArrayName already splits that off (for the
|
||||
// API's own array-index filter syntax) before this ever runs, leaving a
|
||||
// plain quad whose 4th number is the packed ID's low sequence bits with
|
||||
// tag=0 -- true for a real non-array job's own ID, and also true for an
|
||||
// array child's dash-quad prefix (FormatJobID's arrayBase is those same
|
||||
// bits already shifted into position with tag/idx zeroed, i.e. exactly
|
||||
// ExtractArrayBaseID(original)).
|
||||
func ParseHumanJobID(s string) (uint64, bool) {
|
||||
m := humanJobIDPattern.FindStringSubmatch(s)
|
||||
if m == nil {
|
||||
return 0, false
|
||||
}
|
||||
clusterID, _ := strconv.ParseUint(m[1], 10, 64)
|
||||
apiserverID, _ := strconv.ParseUint(m[2], 10, 64)
|
||||
timestamp, _ := strconv.ParseUint(m[3], 10, 64)
|
||||
sequence, _ := strconv.ParseUint(m[4], 10, 64)
|
||||
return (clusterID << clusterIDShift) |
|
||||
(apiserverID << apiserverIDShift) |
|
||||
(timestamp << timestampShift) |
|
||||
sequence, true
|
||||
}
|
||||
|
||||
// Errors returned by ParseHumanJobIDStrict. Callers map them to their own
|
||||
// messages: ErrJobIDSyntax is the written-form syntax rule, ErrJobIDWidth the
|
||||
// field-width rule (job-array-one-based-index 5.14).
|
||||
var (
|
||||
ErrJobIDSyntax = errors.New("job ID is not of the form cluster-apiserver-timestamp-sequence")
|
||||
ErrJobIDWidth = errors.New("job ID field out of range")
|
||||
)
|
||||
|
||||
// ParseHumanJobIDStrict parses a written job identity "c-a-ts-seq" (decimal
|
||||
// digits only, leading zeros allowed, no sign, no whitespace, no "[...]")
|
||||
// into its packed, untagged ID. Unlike ParseHumanJobID it checks every field
|
||||
// against its bit width -- cluster <= 63, apiserver <= 15, timestamp < 2^32,
|
||||
// sequence < 2^20 -- so an oversized field is rejected (ErrJobIDWidth)
|
||||
// instead of spilling into a neighbouring field. A field too large even for
|
||||
// uint64 is also ErrJobIDWidth. Anything not of that shape is ErrJobIDSyntax.
|
||||
func ParseHumanJobIDStrict(s string) (uint64, error) {
|
||||
m := humanJobIDPattern.FindStringSubmatch(s)
|
||||
if m == nil {
|
||||
return 0, ErrJobIDSyntax
|
||||
}
|
||||
widths := [4]int{nClusterIDBits, nApiserverIDBits, nTimestampBits, nSequenceBits}
|
||||
var f [4]uint64
|
||||
for i, w := range widths {
|
||||
v, err := strconv.ParseUint(m[i+1], 10, 64)
|
||||
if err != nil {
|
||||
if errors.Is(err, strconv.ErrRange) {
|
||||
return 0, ErrJobIDWidth
|
||||
}
|
||||
return 0, ErrJobIDSyntax
|
||||
}
|
||||
if v >= uint64(1)<<w {
|
||||
return 0, ErrJobIDWidth
|
||||
}
|
||||
f[i] = v
|
||||
}
|
||||
return (f[0] << clusterIDShift) |
|
||||
(f[1] << apiserverIDShift) |
|
||||
(f[2] << timestampShift) |
|
||||
f[3], nil
|
||||
}
|
||||
|
||||
// arrayIdxMask covers the 15-bit array index field (bits 0-14).
|
||||
const arrayIdxMask = uint64(1)<<nArrayIdxBits - 1
|
||||
|
||||
// IsArrayCapableKey reports whether key can name an array (an "<array-id>"
|
||||
// in "<array-id>[k]"): tag 0 and the low 15 (index) bits clear, i.e. the
|
||||
// sequence is a multiple of 32768 (job-array-one-based-index 5.8).
|
||||
func IsArrayCapableKey(key uint64) bool {
|
||||
return extractIDTag(key) == 0 && key&arrayIdxMask == 0
|
||||
}
|
||||
|
||||
// MakeElementID composes the ID of element k of the array identified by key:
|
||||
// key's cluster, apiserver, timestamp and sequence-high bits, tag
|
||||
// TAG_JOB_ARRAY and index k (masked to 15 bits). key's own tag and index bits
|
||||
// are ignored. Mirrors jobsel_element_id() in src/lib/utils.c.
|
||||
func MakeElementID(key uint64, k int) uint64 {
|
||||
base := key &^ (uint64(1<<nTagBits-1)<<tagShift | arrayIdxMask)
|
||||
return uint64(tagJobArray)<<tagShift | base | uint64(k)&arrayIdxMask
|
||||
}
|
||||
|
||||
// WorkerDefaultOutputPath returns the output file path the worker uses by default
|
||||
// for the given job ID and task index, matching uv_proc.c make_job_output_path().
|
||||
func WorkerDefaultOutputPath(uniqueJobID int64, taskIdx int) string {
|
||||
return fmt.Sprintf("/tmp/job_%s_task%d.out", FormatJobID(uniqueJobID), taskIdx)
|
||||
}
|
||||
|
||||
// jobCtrlUnionDiscriminant returns the value xce_jobctrl_t.action must carry
|
||||
// for the given job_ctrl_type_t action (models.JOB_STOP/RESUME/KILL/CHKPNT):
|
||||
// the capnp JobControl_u_which union discriminant (JobControl_u_siginfo=0,
|
||||
// JobControl_u_ckptoption=1 -- src/lib/controller.capnp.h), NOT the action
|
||||
// code itself. xce_jobctrl_req() (src/lib/request.c) switches on this field
|
||||
// and rejects anything else with -EINVAL; ControlJob (clib.go) previously
|
||||
// stored the raw action code there instead, so kill(3)/resume(2) always hit
|
||||
// that -EINVAL default case (gitea issue #45) while stop(1) happened to
|
||||
// collide with ckptoption(1) and read through the wrong union member.
|
||||
func jobCtrlUnionDiscriminant(action int) int {
|
||||
if action == models.JOB_CHKPNT {
|
||||
return 1 // JobControl_u_ckptoption
|
||||
}
|
||||
return 0 // JobControl_u_siginfo (stop/resume/kill all signal the job)
|
||||
}
|
||||
|
||||
// jobCtrlArgs is what ControlJob (clib.go) puts into xce_jobctrl_t's union
|
||||
// branch: Signo/Force fill xce_signal_t (stop/resume/kill), CkptKill/Method
|
||||
// fill xce_ckpt_options_t (checkpoint).
|
||||
type jobCtrlArgs struct {
|
||||
Signo int
|
||||
Force int
|
||||
Grace int // seconds; > 0: graceful kill/stop sequence (bug-1957)
|
||||
CkptKill int
|
||||
Method string
|
||||
}
|
||||
|
||||
// jobCtrlParams maps a job action and ControlJob's content argument to the
|
||||
// values sent to the controller. content is either the JSON-encoded
|
||||
// models.JobControlEvent the job control handlers pass (controllers/jobs.go)
|
||||
// or, for older callers, a bare signal name or number; empty means the
|
||||
// action's default signal.
|
||||
//
|
||||
// bug-1940: content used to be looked up only as a signal name, so the JSON
|
||||
// event always missed and every kill sent TERM with force 0, and checkpoint
|
||||
// kill/method were never set. An unknown signal is now an error instead of a
|
||||
// silent fallback to the default.
|
||||
func jobCtrlParams(action int, content string) (jobCtrlArgs, error) {
|
||||
var (
|
||||
args jobCtrlArgs
|
||||
signo int
|
||||
forceKill bool
|
||||
)
|
||||
|
||||
content = strings.TrimSpace(content)
|
||||
if strings.HasPrefix(content, "{") {
|
||||
var evt models.JobControlEvent
|
||||
|
||||
if err := json.Unmarshal([]byte(content), &evt); err != nil {
|
||||
return args, fmt.Errorf("invalid job control content: %w", err)
|
||||
}
|
||||
if evt.Signal != 0 && !isKnownSignal(evt.Signal) {
|
||||
return args, fmt.Errorf("unsupported signal: %d", evt.Signal)
|
||||
}
|
||||
signo, forceKill, args.Method = evt.Signal, evt.ForceKill, evt.Method
|
||||
} else if content != "" {
|
||||
n, err := parseSignal(content)
|
||||
if err != nil {
|
||||
return args, err
|
||||
}
|
||||
signo = n
|
||||
}
|
||||
|
||||
switch action {
|
||||
case models.JOB_STOP, models.JOB_RESUME, models.JOB_KILL:
|
||||
if signo == 0 {
|
||||
signo = defaultActionSignal(action)
|
||||
// bug-1957: a plain kill or stop is graceful, as LSF's
|
||||
// TERMINATE and SUSPEND actions; a force kill is immediate
|
||||
if action == models.JOB_STOP || (action == models.JOB_KILL && !forceKill) {
|
||||
args.Grace = jobTerminateInterval()
|
||||
}
|
||||
}
|
||||
args.Signo = signo
|
||||
if action == models.JOB_KILL && forceKill {
|
||||
args.Force = 1
|
||||
}
|
||||
case models.JOB_CHKPNT:
|
||||
if forceKill {
|
||||
args.CkptKill = 1
|
||||
}
|
||||
}
|
||||
return args, nil
|
||||
}
|
||||
|
||||
// defaultActionSignal is the signal a stop/resume/kill sends when the caller
|
||||
// gave none.
|
||||
//
|
||||
// bug-1957: kill and stop name the sequence's last signal (SIGKILL,
|
||||
// SIGSTOP); with a grace period the worker sends SIGINT/SIGTERM or SIGTSTP
|
||||
// first, so the job can clean up or prepare to stop.
|
||||
func defaultActionSignal(action int) int {
|
||||
switch action {
|
||||
case models.JOB_STOP:
|
||||
return models.JobSignalMapTable["STOP"] // 19
|
||||
case models.JOB_RESUME:
|
||||
return models.JobSignalMapTable["CONT"] // 18
|
||||
}
|
||||
return models.JobSignalMapTable["KILL"] // 9
|
||||
}
|
||||
|
||||
// defaultJobTerminateInterval is LSF's JOB_TERMINATE_INTERVAL default.
|
||||
const defaultJobTerminateInterval = 10
|
||||
|
||||
// jobTerminateInterval is job.terminate_interval, or the default when it is
|
||||
// unset or 0.
|
||||
func jobTerminateInterval() int {
|
||||
if utils.JobYamlConfig != nil && utils.JobYamlConfig.TerminateInterval > 0 {
|
||||
return utils.JobYamlConfig.TerminateInterval
|
||||
}
|
||||
return defaultJobTerminateInterval
|
||||
}
|
||||
|
||||
// isKnownSignal reports whether n is one of models.JobSignalMapTable's values.
|
||||
func isKnownSignal(n int) bool {
|
||||
for _, v := range models.JobSignalMapTable {
|
||||
if v == n {
|
||||
return true
|
||||
}
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
// parseSignal accepts a signal number or name (case-insensitive, with or
|
||||
// without the SIG prefix) and returns its number.
|
||||
func parseSignal(s string) (int, error) {
|
||||
if n, err := strconv.Atoi(s); err == nil {
|
||||
if !isKnownSignal(n) {
|
||||
return 0, fmt.Errorf("unsupported signal: %s", s)
|
||||
}
|
||||
return n, nil
|
||||
}
|
||||
name := strings.TrimPrefix(strings.ToUpper(s), "SIG")
|
||||
if n, ok := models.JobSignalMapTable[name]; ok {
|
||||
return n, nil
|
||||
}
|
||||
return 0, fmt.Errorf("unsupported signal: %s", s)
|
||||
}
|
||||
|
||||
// resolveExecUser returns the effective exec user for a job submission: the
|
||||
// caller-specified ExecUser if present, otherwise the submitting user
|
||||
// (bug-1780: an unset execuser in the startjob request reaches xcecd as
|
||||
// empty, which controller_connector.c defaults to the literal Linux user
|
||||
// 'nobody' -- a user with no home/shell that then fails to exec the job).
|
||||
func resolveExecUser(execUser, submitUser string) string {
|
||||
if execUser != "" {
|
||||
return execUser
|
||||
}
|
||||
return submitUser
|
||||
}
|
||||
|
||||
// rawStartInfo mirrors the C job_startinfo_t fields used by the Go model.
|
||||
type rawStartInfo struct {
|
||||
cmdline string
|
||||
|
|
@ -143,6 +378,55 @@ type rawStartInfo struct {
|
|||
execHome string
|
||||
execCWD string
|
||||
execUser string
|
||||
exdata string
|
||||
}
|
||||
|
||||
// submitStartInfo maps a Go JobInfo to the job_startinfo_t fields a
|
||||
// submission carries to the controller (and from there, unchanged, to the
|
||||
// worker's startjob request). modelJobToSubmit (convert.go) is the CGO
|
||||
// adapter that copies the result into the C struct.
|
||||
//
|
||||
// bug-1936: exdata (env, limits, transfers, interactive mode, built by the
|
||||
// submit handler into ExtendData), the exec cwd and the exec home were
|
||||
// never set, so none of them reached the worker.
|
||||
func submitStartInfo(job *models.JobInfo) rawStartInfo {
|
||||
execCWD := job.ExecCWD
|
||||
if execCWD == "" {
|
||||
execCWD = job.CWD
|
||||
}
|
||||
return rawStartInfo{
|
||||
cmdline: job.Command,
|
||||
launchType: job.LaunchType,
|
||||
preExec: job.PreExecCmd,
|
||||
postExec: job.PostExecCmd,
|
||||
execHome: job.ExecHome,
|
||||
execCWD: execCWD,
|
||||
execUser: resolveExecUser(job.ExecUser, job.SubmitUser),
|
||||
exdata: job.ExtendData.String,
|
||||
}
|
||||
}
|
||||
|
||||
// maxExdataLen is the longest exdata the worker can take: xcecd hands it to
|
||||
// xcecd_worker in job_assignment_t.exdata[EXDATA_LENGTH] (16 KiB,
|
||||
// src/worker/include/uv_ipc.h), NUL included.
|
||||
const maxExdataLen = 16*1024 - 1
|
||||
|
||||
// ValidateExdata checks a job's extended data before it is submitted: the
|
||||
// worker's parse_exdata() fails the job at launch on anything but a JSON
|
||||
// object, and a longer string than maxExdataLen would be cut short on its
|
||||
// way to the job process.
|
||||
func ValidateExdata(exdata string) error {
|
||||
if exdata == "" {
|
||||
return nil
|
||||
}
|
||||
if len(exdata) > maxExdataLen {
|
||||
return fmt.Errorf("extended job data is %d bytes, more than the limit of %d", len(exdata), maxExdataLen)
|
||||
}
|
||||
var obj map[string]interface{}
|
||||
if err := json.Unmarshal([]byte(exdata), &obj); err != nil {
|
||||
return errors.New("extended job data must be a JSON object")
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// rawSubmitData mirrors the C job_submit_data_t fields used by the Go model.
|
||||
|
|
@ -180,10 +464,10 @@ type rawResReason struct {
|
|||
// marshaled value lands in models.JobInfo.PendingReason as a JSONB object —
|
||||
// processJobData's JSONBJFields pass then surfaces it in the API response.
|
||||
type rawPendReason struct {
|
||||
MainReason int `json:"main_reason"`
|
||||
MainReasonName string `json:"main_reason_name"`
|
||||
HostReasons []rawHostReason `json:"host_reasons,omitempty"`
|
||||
ResReasons []rawResReason `json:"res_reasons,omitempty"`
|
||||
MainReason int `json:"main_reason"`
|
||||
MainReasonName string `json:"main_reason_name"`
|
||||
HostReasons []rawHostReason `json:"host_reasons,omitempty"`
|
||||
ResReasons []rawResReason `json:"res_reasons,omitempty"`
|
||||
}
|
||||
|
||||
// pendReasonName maps a PREASON_* main_reason code to a readable name. The
|
||||
|
|
@ -215,6 +499,33 @@ func pendReasonName(code int) string {
|
|||
}
|
||||
}
|
||||
|
||||
// rawHostSlot mirrors one xce_host_alloc_t's host name plus its "slots"
|
||||
// resource allocation count (xce_host_alloc_t.allocs[], scanned for
|
||||
// resname=="slots" -- mirrors the C scheduler's own get_nslots_in_alloc()/
|
||||
// repl_worker_alloc_nslots() convention, defaulting to 1 when no explicit
|
||||
// slots entry is present).
|
||||
type rawHostSlot struct {
|
||||
host string
|
||||
slots int
|
||||
}
|
||||
|
||||
// rawRunUsage mirrors the C xce_job_runusage_t job-level fields used by the
|
||||
// Go model (bug-1736): accumulated CPU time and memory, and the CPU peak and
|
||||
// efficiency values the controller computes with LSF bjobs -l definitions
|
||||
// (efficiencies in percent, memory in MB as the worker reports it).
|
||||
type rawRunUsage struct {
|
||||
maxMem float64
|
||||
avgMem float64
|
||||
usedMem float64
|
||||
cpuTime int
|
||||
peakCPUTime int
|
||||
peakCPUDuration int
|
||||
memEfficiency float64
|
||||
avgCPUEfficiency float64
|
||||
peakCPUEfficiency float64
|
||||
peakCPU float64
|
||||
}
|
||||
|
||||
// rawJobData mirrors the C xce_job_t fields used by the Go model.
|
||||
type rawJobData struct {
|
||||
jobID int64
|
||||
|
|
@ -225,8 +536,9 @@ type rawJobData struct {
|
|||
startTime int64
|
||||
finishTime int64
|
||||
submit *rawSubmitData
|
||||
execHosts []string
|
||||
execHosts []rawHostSlot
|
||||
pendReason *rawPendReason
|
||||
runUsage *rawRunUsage // nil until the job's first usage report
|
||||
}
|
||||
|
||||
// rawWorkerData mirrors the C xce_worker_t fields used by the Go model.
|
||||
|
|
@ -293,7 +605,9 @@ func rawJobToModel(r rawJobData) *models.JobInfo {
|
|||
info.PostExecCmd = si.postExec
|
||||
info.ExecHome = si.execHome
|
||||
info.ExecCWD = si.execCWD
|
||||
info.CWD = si.execCWD // the submit cwd is sent as the exec cwd (bug-1936)
|
||||
info.ExecUser = si.execUser
|
||||
info.ExtendData = models.Jsonb{String: si.exdata}
|
||||
}
|
||||
}
|
||||
info.SubmitTime = r.subTime
|
||||
|
|
@ -311,16 +625,57 @@ func rawJobToModel(r rawJobData) *models.JobInfo {
|
|||
}
|
||||
slots := make([]hostSlot, len(r.execHosts))
|
||||
for i, h := range r.execHosts {
|
||||
slots[i] = hostSlot{HostName: h, NumSlots: 1}
|
||||
slots[i] = hostSlot{HostName: h.host, NumSlots: h.slots}
|
||||
}
|
||||
data, _ := json.Marshal(execHostsJSON{HostSlots: slots, NumHosts: len(slots)})
|
||||
info.ExecHosts = models.Jsonb{String: string(data)}
|
||||
}
|
||||
if r.pendReason != nil {
|
||||
// main_reason is only set by an explicit add_job_preason(reasons,
|
||||
// code, NULL, NULL) (which erases the detail lists); the allocator's
|
||||
// ordinary shortfalls attach per-host/per-res DETAIL rows and leave
|
||||
// main_reason 0. A detail-only record promotes its first detail row
|
||||
// to main_reason_name here — the API-visible "overall reason"
|
||||
// (pend-reason P0: main_reason_name used to read unknown(0)) — while
|
||||
// the C record's main_reason value itself marshals unchanged (the
|
||||
// scheduler-framework trace/verify stack counts main_reason as its
|
||||
// own occurrence and its committed .replay baselines encode
|
||||
// main_reason=0 detail-only PENDs, so the promotion stays an
|
||||
// API-layer view).
|
||||
if r.pendReason.MainReasonName == "" && r.pendReason.MainReason != 0 {
|
||||
r.pendReason.MainReasonName = pendReasonName(r.pendReason.MainReason)
|
||||
} else if r.pendReason.MainReasonName == "" && r.pendReason.MainReason == 0 {
|
||||
for _, h := range r.pendReason.HostReasons {
|
||||
if h.Reason != 0 {
|
||||
r.pendReason.MainReasonName = pendReasonName(h.Reason)
|
||||
break
|
||||
}
|
||||
}
|
||||
if r.pendReason.MainReasonName == "" {
|
||||
for _, rr := range r.pendReason.ResReasons {
|
||||
if rr.Reason != 0 {
|
||||
r.pendReason.MainReasonName = pendReasonName(rr.Reason)
|
||||
break
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
if data, err := json.Marshal(r.pendReason); err == nil {
|
||||
info.PendingReason = models.Jsonb{String: string(data)}
|
||||
}
|
||||
}
|
||||
if u := r.runUsage; u != nil {
|
||||
info.MaxMem = u.maxMem
|
||||
info.AvgMem = u.avgMem
|
||||
info.UsedMem = u.usedMem
|
||||
info.CPUTime = u.cpuTime
|
||||
info.PeakCPU = u.peakCPU
|
||||
info.PeakCPUTime = u.peakCPUTime
|
||||
info.PeakCPUDuration = float64(u.peakCPUDuration)
|
||||
info.MemEfficiency = u.memEfficiency
|
||||
info.AvgCPUEfficiency = u.avgCPUEfficiency
|
||||
info.PeakCPUEfficiency = u.peakCPUEfficiency
|
||||
}
|
||||
return info
|
||||
}
|
||||
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue