diff --git a/cmd/aforge/chatv3_orchestrate.go b/cmd/aforge/chatv3_orchestrate.go index a0d7fa527f..15a857d09e 100644 --- a/cmd/aforge/chatv3_orchestrate.go +++ b/cmd/aforge/chatv3_orchestrate.go @@ -38,6 +38,17 @@ import ( // in the one the person is looking at, or the room that draws its graph would // ask the wrong session for a snapshot. +// v3MachineLanes is THE ACCOUNT OF THIS MACHINE'S RUNNING TASK LANES, and there +// is one for the life of the process because there is one machine under it. +// /new and /resume each build a conversation of their own, and the memory half +// of the reading each one's admission governor takes is of this whole process +// and everything it started — /proc cannot say which conversation started which +// compiler. So what that reading is divided by has to be every conversation's +// lanes and never one conversation's, or a neighbour's build becomes this +// conversation's per-node weight for the rest of the session (internal/session's +// task_pressure.go, issue #907). +var v3MachineLanes = session.NewTaskLanes() + // v3OpenSession builds one v3 conversation with its adaptive runner wired to // itself. // @@ -47,6 +58,9 @@ import ( // session that could start a run the surface cannot then find is worse than a // session that cannot start one. func v3OpenSession(cfg session.Config) (*session.Agent, error) { + // Every conversation this process opens counts its task lanes in one + // account, because they are all running on one machine. + cfg.TaskLanes = v3MachineLanes cfg, runs := v3Adaptive(cfg) agent, err := session.New(cfg) if err != nil { diff --git a/docs/changes/unreleased/947-governor-footprint.md b/docs/changes/unreleased/947-governor-footprint.md new file mode 100644 index 0000000000..3ea3bd8386 --- /dev/null +++ b/docs/changes/unreleased/947-governor-footprint.md @@ -0,0 +1,18 @@ +--- +kind: fixed +title: the machine reading is divided by the lanes the whole process is running, not one conversation's +pr: 947 +surface: [engine] +invalidates: + - "The admission governor used to divide its reading by ONE GRAPH'S count of running nodes, while the memory in that reading is the whole process tree's; a second conversation's build therefore became this conversation's per-node footprint, and that figure only rises. The count now covers every lane the reading does, kept in one `session.TaskLanes` the process hands to every conversation it opens (`Config.TaskLanes`, set by `cmd/aforge`'s `v3OpenSession`). A graph handed none is alone in its process and keeps an account of its own." + - "`TaskGraph.running` used to be incremented and decremented at five sites (the frontier's start, `handBackSlotLocked`, `park`, `unpark`, and `standingWideWork`'s `graph.running = 1`). There is now ONE door, `takeLaneLocked`/`giveLaneLocked`, it moves the process's account in the same breath, and `TestEveryLaneMovesThroughTheOneDoor` fails the build on a sixth site." + - "`docs/design/task-start/DESIGN.md` used to state the per-conversation attribution gap as a boundary owed to #907. It is closed, and the section now describes the account." +--- + +Two conversations share one process, and `/proc` cannot say which of them started +which compiler. Divided by one graph's own lanes, a neighbour's build read here as +that build over this conversation's node count — and a measured footprint only +rises, so one such reading narrowed every later fan in this conversation for the +rest of the session. The same gap ran the other way on the reservation, each graph +reading the other's visible memory as covering part of its own. Both halves now +read one count over the same population the reading covers. diff --git a/docs/design/task-start/DESIGN.md b/docs/design/task-start/DESIGN.md index 1a11c4a5a2..90ea14775c 100644 --- a/docs/design/task-start/DESIGN.md +++ b/docs/design/task-start/DESIGN.md @@ -305,15 +305,38 @@ re-read every second. **Boundaries, stated rather than hidden.** The reading is `/proc`'s, so on macOS and Windows the governor still says "cannot say" and never holds, exactly as -before. And each conversation's graph keeps its own governor and so its own -reservation, while the visible half is this whole process: two conversations in -one process fanning out at the same moment each read the other's visible work as -covering part of its own reservation, and one reading taken during the other's -build can raise this graph's measured footprint for the rest of the session, -because that figure only rises. The floor on the reading itself still holds for -both. The fix is one account for the whole process, which needs every -`TaskGraph.running` mutation behind one door; it is -[#907](https://github.com/Agent-Field/aforge-v2/issues/907), not a clamp here. +before. + +**One account for the whole process (#907).** Each conversation keeps its own +graph and its own governor, but the count those governors divide the reading by +is **the process's**, not the graph's. It has to be: the visible half is a +reading of the whole process tree, and /proc cannot say which conversation +started which compiler. Divided by one graph's own lanes, a second +conversation's build read here as that build over *this* conversation's node +count — and the measured footprint only rises, so one such reading narrowed +every later fan in this conversation for the rest of the session. The same gap +ran the other way on the reservation, each graph reading the other's visible +memory as covering part of its own. Both close with `session.TaskLanes`: one mutex and one int, +written by every graph through the one door every `TaskGraph.running` mutation +now goes through (`takeLaneLocked`/`giveLaneLocked`), read back by +`TaskGraph.lanesTaken` for `observe` and by `holdOnStartingLocked` for `admits`. +No clamp to a fraction of `MemTotal`, no timer decay and no per-graph +correction. + +**Which graphs share one is said, not assumed.** `cmd/aforge`'s `v3OpenSession` +— the one door every conversation is built through, launch, relaunch, `/new` +and `/resume` — builds one account for the life of the process and puts it on +every `session.Config` it hands out; `Agent.graph` and `standingWideWork` read +it off the config. A graph handed none is **alone in its process** and keeps an +account of its own (`newTaskGraph`), which is the truth about an embedder with +one conversation and about every scripted graph in the tests. A package-level +variable would have been the same claim made silently, and it is a claim a +library cannot make for its caller: a test binary is one process and a hundred +unrelated machines, and a lane one scene left running would have narrowed the +next scene's fan. +`TestEveryLaneMovesThroughTheOneDoor` (`go/ast`, so it is on the pull-request +gate) is what keeps there being one door, and the two graphs of +`task_pressure_account_test.go` are the scene itself. **Deleted.** `admissionGovernor.holds`, the once-per-pass `busy` bool and the `busy` parameter of `holdOnStartingLocked`; `TaskGraph.machineBusy`, which asked diff --git a/internal/manual/chat/how-tasks-run.md b/internal/manual/chat/how-tasks-run.md index 68452b02b1..fbccbb0878 100644 --- a/internal/manual/chat/how-tasks-run.md +++ b/internal/manual/chat/how-tasks-run.md @@ -2415,7 +2415,7 @@ minutes later — so each task that starts **sets aside a footprint** of memory reading shows it, and the next one is judged against what is left. A footprint is the larger of two figures this machine gives: one core's share of its memory (`MemTotal` ÷ cores, so 2 GiB on a 16 GiB eight-core laptop) and the most memory per task aforge has -watched this session's tasks actually hold. A quiet machine therefore starts roughly **one +watched a task actually hold. A quiet machine therefore starts roughly **one task per core's share of the memory above the floor**; the rest wait saying `machine busy` and start as the earlier ones finish or as a reading shows room. Nothing is counted twice — as a running task's memory appears in the reading, what is set aside for it falls by as @@ -2424,6 +2424,12 @@ much. This gates **starts only**. Nothing already running is ever touched; pressure drains as running tasks finish. +**Tasks in your other conversations count here.** aforge is one program, and the reading +covers that whole program and everything it started — it cannot tell which conversation +started which build. So every task aforge is running is set aside for, whichever +conversation started it: tasks fanning out beside you leave less room here too, and a +build running beside yours is never mistaken for what one of your own tasks weighs. + **The honest caveat:** these two governors read `/proc`, so they only work on Linux. On macOS and Windows there is no `/proc`, the machine cannot say, and silence is never treated as a hold — those platforms get no pressure gating at all, and `task.max_load` and diff --git a/internal/session/session.go b/internal/session/session.go index ae2bc64b66..92ce0eb2c9 100644 --- a/internal/session/session.go +++ b/internal/session/session.go @@ -1703,6 +1703,20 @@ type Config struct { // gets, which is what lets pressure drain instead of having to be relieved. TaskMinFreeMB int + // TaskLanes is THE ACCOUNT OF THIS MACHINE'S RUNNING TASK LANES, shared by + // every conversation this process opens ([NewTaskLanes]). The memory half + // of the reading above is `treeResidentMB(os.Getpid())` — this process and + // every descendant it started — and /proc cannot say which conversation + // started which compiler, so the count that reading is divided by has to + // cover the same work: every lane the process is running, not one graph's + // (task_pressure.go's ONE ACCOUNT FOR THE WHOLE PROCESS, #907). + // + // The process's own door sets it once and hands the same pointer to every + // conversation (cmd/aforge). Left nil, a graph is ALONE IN ITS PROCESS and + // keeps an account of its own — which is the truth for an embedder with one + // conversation, and for every scripted graph in the tests. + TaskLanes *TaskLanes + // InTask marks this agent as ONE TASK NODE'S RUNNER (task_run.go) rather // than the conversation. It changes exactly two things, and both are // consequences of the same fact — there is nobody to talk to: diff --git a/internal/session/standing_run.go b/internal/session/standing_run.go index 75dded078f..a70bef4fba 100644 --- a/internal/session/standing_run.go +++ b/internal/session/standing_run.go @@ -884,6 +884,9 @@ func standingWideWork(cfg Config, item standing.Item, brief string) (Config, *Ta // this firing leaves behind lands. graph.limit = cfg.TaskParallel graph.governor = newAdmissionGovernor(cfg.TaskMaxLoad, cfg.TaskMinFreeMB) + if cfg.TaskLanes != nil { + graph.lanes = cfg.TaskLanes + } graph.store = newTaskStore(taskCheckpointPath(cfg.SessionFile)) graph.run = graph.runOwned graph.report = graph.reportHome @@ -930,7 +933,12 @@ func standingWideWork(cfg Config, item standing.Item, brief string) (Config, *Ta // here. A firing on a machine capped at one task at a time has no second // pair of hands to give parts to, and [TaskGraph.freeHands] answers that // correctly only if the work already under way is counted as under way. - graph.running = 1 + // It goes through the same door every other start does, so the process's + // account sees a standing firing's lane exactly as it sees a task's — the + // machine carries this one no differently (task_pressure.go, #907). No lock + // is taken: the graph is still this function's, with nothing else able to + // reach it. + graph.takeLaneLocked() cfg.tasker = graph cfg.taskID = id diff --git a/internal/session/task_lane_door_law_test.go b/internal/session/task_lane_door_law_test.go new file mode 100644 index 0000000000..479c28bfdb --- /dev/null +++ b/internal/session/task_lane_door_law_test.go @@ -0,0 +1,122 @@ +package session + +import ( + "go/ast" + "go/parser" + "go/token" + "os" + "path/filepath" + "strings" + "testing" +) + +// theLaneDoor is the pair of functions that move a lane, and the only place in +// this package where either count of one may move. Both names are here rather +// than in the assertion below so that the failure can say what the door is. +var theLaneDoor = map[string]string{ + "takeLaneLocked": "take", + "giveLaneLocked": "give", +} + +// A LANE IS TAKEN AND HANDED BACK THROUGH ONE DOOR, AND THE DOOR KEEPS BOTH +// COUNTS IN STEP. +// +// Two numbers say how much work is running: this graph's, which the person's +// task.parallel cap is measured against, and the process's ([TaskLanes]), +// which the machine reading is divided by — and the reading is of the whole +// process tree, so a start written to one and not the other is exactly the +// attribution gap #907 closed. It was open in the first place because there +// were five places a lane moved and no one of them that could be made to tell +// the account; this law is what keeps there being one. +// +// It reads the package's own source, so it is on the pull-request gate +// (scripts/laws.sh) and decides in well under a second. +func TestEveryLaneMovesThroughTheOneDoor(t *testing.T) { + files := token.NewFileSet() + entries, err := os.ReadDir(".") + if err != nil { + t.Fatalf("read the package: %v", err) + } + // counter is every function that moves a graph's own count of running + // lanes, and account every function that writes the process's. + counter, account := map[string]bool{}, map[string]bool{} + for _, entry := range entries { + name := entry.Name() + if entry.IsDir() || !strings.HasSuffix(name, ".go") || strings.HasSuffix(name, "_test.go") { + continue + } + file, err := parser.ParseFile(files, filepath.Join(".", name), nil, 0) + if err != nil { + t.Fatalf("parse %s: %v", name, err) + } + for _, decl := range file.Decls { + fn, ok := decl.(*ast.FuncDecl) + if !ok || fn.Body == nil { + continue + } + where := fn.Name.Name + ast.Inspect(fn.Body, func(n ast.Node) bool { + switch node := n.(type) { + case *ast.IncDecStmt: + if isField(node.X, "running") { + counter[where] = true + } + case *ast.AssignStmt: + for index, lhs := range node.Lhs { + // A COUNT MOVES BY ARITHMETIC, and `a.running = true` + // elsewhere in this package is a different field on a + // different type: what is watched here is a number. + if !isField(lhs, "running") { + continue + } + if node.Tok != token.ASSIGN || (index < len(node.Rhs) && isNumber(node.Rhs[index])) { + counter[where] = true + } + } + case *ast.CallExpr: + if call, ok := node.Fun.(*ast.SelectorExpr); ok && isField(call.X, "lanes") { + if call.Sel.Name == "take" || call.Sel.Name == "give" { + account[where] = true + } + } + } + return true + }) + } + } + for where := range counter { + if _, isDoor := theLaneDoor[where]; !isDoor { + t.Errorf("%s moves a graph's count of running lanes itself. Every lane goes through takeLaneLocked or giveLaneLocked, "+ + "which are the only two places that can tell the process's account as well (task_pressure.go, #907).", where) + } + } + for where := range account { + if _, isDoor := theLaneDoor[where]; !isDoor { + t.Errorf("%s writes the process's lane account directly, outside the door that keeps it in step with the graph's own count", where) + } + } + // AND THE DOOR ITSELF DOES BOTH HALVES, which is the whole of what makes it + // a door: a take that forgot the account would pass every check above. + for door := range theLaneDoor { + if !counter[door] { + t.Errorf("%s no longer moves the graph's own count of running lanes", door) + } + if !account[door] { + t.Errorf("%s no longer tells the process's lane account, so the two counts can drift apart", door) + } + } +} + +// isField says whether an expression is a selector onto the named field, which +// is how both halves of the count are spelled at every site. +func isField(expr ast.Expr, name string) bool { + selector, ok := expr.(*ast.SelectorExpr) + return ok && selector.Sel.Name == name +} + +// isNumber says whether an expression is a literal number, which is how a +// count is set outright rather than stepped. +func isNumber(expr ast.Expr) bool { + lit, ok := expr.(*ast.BasicLit) + return ok && lit.Kind == token.INT +} diff --git a/internal/session/task_pressure.go b/internal/session/task_pressure.go index 9c3cd87b40..73f75b0a42 100644 --- a/internal/session/task_pressure.go +++ b/internal/session/task_pressure.go @@ -97,19 +97,33 @@ import ( // before the governor existed, which is the only honest thing to do with a // number nobody took. // -// ── ONE CONVERSATION'S RESERVATION ── -// -// Each conversation's graph keeps its own governor, and so its own -// reservation; the visible half is this whole process, because /proc cannot -// say which conversation started which compiler. A second conversation in the -// same process fanning out at the same moment as the first therefore reads -// the first one's visible work as covering part of its own reservation. The -// floor on the reading itself still holds for both, and the width that gap can -// add is the first conversation's visible memory divided by one footprint. The -// same attribution can raise a graph's measured footprint from a reading taken -// during the other conversation's build, and that figure only rises. Both close -// with one account for the whole process rather than one per graph, which is -// issue #907. +// ── ONE ACCOUNT FOR THE WHOLE PROCESS ── +// +// The visible half is this whole process and every descendant it started, +// because /proc cannot say which conversation started which compiler. A +// governor handed its OWN graph's count of running nodes therefore divided the +// whole tree's memory by one conversation's nodes: a second conversation's +// build, under way while this graph had two parts out, read here as three +// core-shares a node. And the measured footprint only rises, so that one +// reading narrowed every later fan in this conversation for the rest of the +// session. The same gap ran the other way on the reservation itself — each +// graph read the other's visible memory as covering part of its own — and both +// are issue #907. +// +// So THE COUNT THIS FILE IS HANDED COVERS EVERY LANE THE READING DOES, never +// one graph's. Both +// halves of `share = visible / running` and of `running × footprint − visible` +// are then read over the same population: every lane the process is running, +// and every byte the process holds above rest. Another conversation's build is +// counted as exactly what it is — memory held by lanes — against a divisor +// that already counts the lanes holding it, and no conversation can move +// another's per-node weight. The count comes from [TaskLanes] — one account, +// handed to every conversation the process opens ([Config.TaskLanes]) and +// written by every `TaskGraph.running` mutation through one door +// ([TaskGraph.takeLaneLocked], [TaskGraph.giveLaneLocked]). There is no clamp +// to a fraction of MemTotal, no timer decay and no per-graph correction, +// because the attribution gap is a fact about the machine the reading is of, +// and the account is kept where that machine is known. const ( // taskPressureTTL bounds how often the host is asked. Load average is a @@ -180,7 +194,7 @@ type admissionGovernor struct { known bool at time.Time - // restMB is workMB as the last reading that found this graph running + // restMB is workMB as the last reading that found the process running // nothing saw it: the conversation, its tools' servers, whatever this // process carries with no node at work. What is above it is the nodes'. // rested says there has been such a reading; until there is one, the @@ -188,14 +202,75 @@ type admissionGovernor struct { // under way as visible and so errs toward holding. restMB int rested bool - // peakShareMB is the most visible memory per running node any reading of - // this session has seen, and it only rises. A share that fell whenever the + // peakShareMB is the most visible memory per running LANE OF THE PROCESS + // any reading of this session has seen, and it only rises. A share that fell whenever the // nodes happened to be between builds would hand the room back just before // the next build needs it — the same reason the kernel's own ru_maxrss is // a high-water mark. peakShareMB int } +// TaskLanes is THE COUNT OF RUNNING LANES ON ONE MACHINE, and the one divisor +// the governor's reading is shared over. Every task graph that belongs to the +// same running aforge writes its own starts and hand-backs into one of these, +// so the count beside a reading of the whole process tree is drawn from the +// same population the reading is (#907). +// +// WHICH GRAPHS SHARE ONE IS SAID, NOT ASSUMED. The process's own door builds +// one and hands it to every conversation it opens ([Config.TaskLanes]); a graph +// given none is alone in its process and keeps one of its own ([newTaskGraph]). +// A package-level variable would have been the same claim made silently, and it +// is a claim no library can make for its caller — a test binary is one process +// and a hundred unrelated machines. +// +// It is one mutex and one int, and that is the whole design. IT IS THE +// INNERMOST LOCK here: every caller either holds a graph's lock already or +// holds nothing, and nothing inside it calls back out, so it cannot be half of +// a cycle. It is nil-safe throughout, because a graph assembled field by field +// in a test has no account and a count of nothing is the truth about it. +type TaskLanes struct { + mu sync.Mutex + n int +} + +// NewTaskLanes builds the account one machine's conversations share. The door +// that opens conversations builds one for the life of the process and puts it +// on every [Config] it hands out. +func NewTaskLanes() *TaskLanes { return &TaskLanes{} } + +// take records one lane taken anywhere in the process. +func (a *TaskLanes) take() { + if a == nil { + return + } + a.mu.Lock() + a.n++ + a.mu.Unlock() +} + +// give records one lane handed back anywhere in the process. +func (a *TaskLanes) give() { + if a == nil { + return + } + a.mu.Lock() + if a.n > 0 { + a.n-- + } + a.mu.Unlock() +} + +// running is how many lanes the whole process is running, which is the +// divisor the visible half is shared over. +func (a *TaskLanes) running() int { + if a == nil { + return 0 + } + a.mu.Lock() + defer a.mu.Unlock() + return a.n +} + // newAdmissionGovernor builds the gate over the real host, and returns nil // when both halves are off — a governor with nothing to check is not a // governor that always says yes, it is one that should not be consulted, and @@ -213,8 +288,12 @@ func newAdmissionGovernor(maxLoad float64, minFreeMB int) *admissionGovernor { } // observe asks the host, at most once per taskPressureTTL, and learns from -// the answer. running is how many of this graph's slot-taking nodes are -// running as the reading is taken. +// the answer. +// +// `running` IS THE WHOLE PROCESS'S COUNT OF RUNNING LANES and never one +// graph's ([TaskGraph.lanesTaken]), because the memory in the reading is the +// whole process tree's and /proc cannot attribute a compiler to the +// conversation that started it. // // IT IS CALLED BEFORE THE GRAPH'S LOCK, once a pass, and it is the only // method here that touches /proc: the graph's lock is held by everything that @@ -265,11 +344,17 @@ func (g *admissionGovernor) observe(running int) { } } -// admits reports whether one more node may start beside the `running` nodes -// of this graph that already are. It is asked once per admission, with the -// count that includes every node the same pass has already started, so a fan -// is admitted one node at a time against the machine each start leaves -// behind. +// admits reports whether one more node may start beside the `running` lanes +// the process already has out. It is asked once per admission, with a count +// that includes every node the same pass has already started, so a fan is +// admitted one node at a time against the machine each start leaves behind. +// +// The count is THE PROCESS'S, for the reason [admissionGovernor.observe] gives: +// the reservation is taken off a reading of the whole process tree, so it has +// to be the whole process's work that is reserved for. A conversation whose +// neighbour is fanning out therefore meets a machine that is genuinely fuller, +// rather than reading its neighbour's visible memory as cover for its own +// nodes (#907). // // It reads the sample [admissionGovernor.observe] took and nothing else, so // it is safe under the graph's lock. @@ -298,11 +383,11 @@ func (g *admissionGovernor) admits(running int) bool { return true } -// unseenLocked is the memory the running nodes are expected to need that the -// reading does not show yet: every one of them at one footprint, less what -// they are already visibly holding, and never below zero — a node carrying -// more than a footprint is carrying it IN the reading, where it is already -// counted. +// unseenLocked is the memory the process's running lanes are expected to need +// that the reading does not show yet: every one of them at one footprint, less +// what they are already visibly holding, and never below zero — a node +// carrying more than a footprint is carrying it IN the reading, where it is +// already counted. func (g *admissionGovernor) unseenLocked(running int) int { if running <= 0 { return 0 @@ -310,8 +395,8 @@ func (g *admissionGovernor) unseenLocked(running int) int { return max(0, running*g.footprintLocked()-g.visibleLocked()) } -// visibleLocked is the memory the running nodes visibly hold: this process's -// tree above what it held at rest. +// visibleLocked is the memory the running lanes visibly hold: this process's +// tree above what it held with no lane out anywhere in it. func (g *admissionGovernor) visibleLocked() int { return max(0, g.sample.workMB-g.restMB) } @@ -322,7 +407,7 @@ func (g *admissionGovernor) visibleLocked() int { // written here. // // The measured one is [admissionGovernor.peakShareMB]: the most visible memory -// per running node this session's own readings have seen. It is the footprint +// per running lane this session's own readings have seen. It is the footprint // of THIS project's builds on THIS machine, and it is the one that grows when // the work turns out to be heavier than a machine's rule of thumb. // diff --git a/internal/session/task_pressure_account_test.go b/internal/session/task_pressure_account_test.go new file mode 100644 index 0000000000..90a13c81c1 --- /dev/null +++ b/internal/session/task_pressure_account_test.go @@ -0,0 +1,132 @@ +package session + +import "testing" + +// ── one account for the process: another conversation's work is not this one's ── +// +// Two conversations share one process — tabs share the binary — and /proc +// cannot say which of them started which compiler: a reading that carries a +// build carries it for the whole tree. Both tests below state one machine and +// two graphs over it and ask the oldest question in accounting: whose work is +// this? A second conversation's build must not become this one's per-node +// weight, and its visible memory must not stand in for this one's reservation +// (#907). + +// ANOTHER CONVERSATION'S BUILD IS NOT THIS ONE'S PER-NODE WEIGHT. This +// conversation has two parts running and neither is holding anything yet; a +// second conversation in the same process starts one node whose build shows +// six core-shares in the tree every reading reads. Three lanes are out in the +// process and twelve gibibytes are visible above rest, so what a lane has been +// seen to weigh is four — and that is what this conversation's next fan is +// judged at. Divided over this conversation's own two lanes alone, the same +// reading says six core-shares a lane, and the measured footprint only rises, +// so that one reading would have narrowed every later fan here for the rest of +// the session. +func TestAnotherConversationsBuildIsNotThisOnesPerNodeWeight(t *testing.T) { + // What the reading says a lane weighs, over the whole process's lanes and + // over this conversation's own — the wrong answer is the defect's. + const ( + build = 6 * testFootprintMB + overProcess = build / 3 + overMine = build / 2 + ) + process := NewTaskLanes() + machine := &fakeMachine{} + machine.set(roomFor(12), true) + mine, _ := fanGraphIn(t, machine, process) + + // A reading with nothing running anywhere in the process, which is what + // fixes the tree's own weight for the visible half to count above. + mine.runFrontier() + + // This conversation starts two parts, and they hold nothing visible yet. + parts := admitChildren(mine, 0, 2) + mine.runFrontier() + if running, _ := partition(mine, parts); len(running) != 2 { + t.Fatalf("%d of this conversation's parts started, want both", len(running)) + } + + // A second conversation in the same process starts one node of its own. + theirs, _ := fanGraphIn(t, machine, process) + theirNode := admitChildren(theirs, 0, 1) + theirs.runFrontier() + if running, _ := partition(theirs, theirNode); len(running) != 1 { + t.Fatal("the second conversation's node did not start") + } + if lanes := process.running(); lanes != 3 { + t.Fatalf("the process counts %d lanes, want the 3 its two conversations have out", lanes) + } + + // And its build shows: six core-shares above rest, visible to every reading + // and attributable to nobody in particular. + busy := roomFor(12) + busy.availableMB -= build + busy.workMB += build + machine.set(busy, true) + + // This conversation reads the machine while that build is under way. + mine.runFrontier() + mine.governor.mu.Lock() + footprint := mine.governor.footprintLocked() + mine.governor.mu.Unlock() + if footprint == overMine { + t.Fatalf("a lane is measured at %d MiB, which is the other conversation's build divided by this conversation's own lanes", footprint) + } + if footprint != overProcess { + t.Fatalf("a lane is measured at %d MiB, want the %d the process's three lanes hold between them", footprint, overProcess) + } +} + +// ANOTHER CONVERSATION'S MEMORY DOES NOT STAND IN FOR THIS ONE'S RESERVATION. +// The machine has three core-shares of room above the floor, one lane of this +// conversation is running and the neighbour's node is visibly holding a share. +// Three more parts are asked for here, and the machine has room for two of +// them: the fourth lane of the process is the one it cannot carry. Read per +// conversation, the neighbour's visible share was counted as covering one of +// this conversation's own reservations and all three started, which is a part +// running in memory the machine does not have. +func TestAnotherConversationsMemoryDoesNotCoverThisOnesReservation(t *testing.T) { + process := NewTaskLanes() + machine := &fakeMachine{} + machine.set(roomFor(3), true) + mine, log := fanGraphIn(t, machine, process) + mine.runFrontier() + + // One part of this conversation is already running before the neighbour's + // memory shows, so the reading that carries it is not mistaken for what + // this process holds at rest. + first := admitChildren(mine, 0, 1) + mine.runFrontier() + if running, _ := partition(mine, first); len(running) != 1 { + t.Fatal("this conversation's first part did not start") + } + + theirs, _ := fanGraphIn(t, machine, process) + theirNode := admitChildren(theirs, 0, 1) + theirs.runFrontier() + if running, _ := partition(theirs, theirNode); len(running) != 1 { + t.Fatal("the second conversation's node did not start") + } + + // The second conversation's node is carrying one core's share, visibly. + busy := roomFor(3) + busy.availableMB -= testFootprintMB + busy.workMB += testFootprintMB + machine.set(busy, true) + + // Three more parts are asked for here: two start and one is held, which is + // the room left once the neighbour's lane is reserved for as well. + more := admitChildren(mine, 0, 3) + running, held := partition(mine, append(first, more...)) + if len(running) != 3 { + t.Fatalf("%d of this conversation's parts run beside the neighbour's node, want 3: the fourth lane of the process is one the machine cannot carry", len(running)) + } + if len(held) != 1 { + t.Fatalf("%d parts are held, want the one the machine has no room for", len(held)) + } + for _, id := range held { + if notice, said := log.last(id); !said || notice.Waiting != waitingMachineBusy { + t.Fatalf("held part %d says %q, want %q", id, notice.Waiting, waitingMachineBusy) + } + } +} diff --git a/internal/session/task_pressure_test.go b/internal/session/task_pressure_test.go index 7210b9bfd0..cd7e5f76f6 100644 --- a/internal/session/task_pressure_test.go +++ b/internal/session/task_pressure_test.go @@ -76,15 +76,27 @@ func settle(graph *TaskGraph, id uint64) { graph.complete(node, TaskDone) } -// fanGraph is a governed graph over a stated machine with a scripted runner. +// fanGraph is a governed graph over a stated machine with a scripted runner, +// ALONE IN A PROCESS OF ITS OWN, which is what [newTaskGraph] gives a graph +// nobody told about neighbours: the only lanes its governor counts are the ones +// this test starts (task_pressure.go's [TaskLanes]). // // ITS POLL NEVER FIRES INSIDE A TEST. The poll is the product's wake-up for a // machine that got quieter by itself, and here every pass is one the test // asks for, so a timer turning the frontier underneath it would be a second // author of the same pass. func fanGraph(t *testing.T, machine *fakeMachine) (*TaskGraph, *noticeLog) { + t.Helper() + return fanGraphIn(t, machine, NewTaskLanes()) +} + +// fanGraphIn is the same graph in a process a test names, so that two graphs +// can share one — two conversations in one aforge, which is the whole scene +// issue #907 is about. +func fanGraphIn(t *testing.T, machine *fakeMachine, process *TaskLanes) (*TaskGraph, *noticeLog) { t.Helper() graph := newTaskGraph() + graph.lanes = process log := ¬iceLog{} runner := &fanRunner{started: make(chan uint64, 4*fanWidth)} graph.report = log.record diff --git a/internal/session/task_run.go b/internal/session/task_run.go index c0ee63fb36..e1045987cb 100644 --- a/internal/session/task_run.go +++ b/internal/session/task_run.go @@ -989,6 +989,15 @@ type TaskGraph struct { order []uint64 seq uint64 running int + // lanes is the account of running lanes this graph shares with every other + // graph of the same running aforge ([Config.TaskLanes]), written through + // [TaskGraph.takeLaneLocked] and [TaskGraph.giveLaneLocked]. The governor's + // reading is of the whole process tree, so the count it divides that + // reading by has to cover the same work (task_pressure.go's ONE ACCOUNT FOR + // THE WHOLE PROCESS, #907). A graph that was handed none gets one of its + // own from [newTaskGraph] and is alone in its process, which is the truth + // about a lone embedder and about every scripted graph in the tests. + lanes *TaskLanes // quickGate serialises the ADMISSION of quick nodes, and it is the one lock // in this file that is not `mu` (task_quick.go). @@ -1173,7 +1182,12 @@ func (g *TaskGraph) runRowsLocked() []TaskNotice { } func newTaskGraph() *TaskGraph { - return &TaskGraph{nodes: make(map[uint64]*TaskNode, 1)} + // A GRAPH IS ALONE IN ITS PROCESS UNTIL IT IS TOLD OTHERWISE. The caller + // that knows several conversations share one machine says so by putting its + // own account on their Config ([Config.TaskLanes], read in [Agent.graph] + // and [standingWideWork]); until then the only lanes this graph's reading + // can be divided by are its own. + return &TaskGraph{nodes: make(map[uint64]*TaskNode, 1), lanes: NewTaskLanes()} } // graph is the session's graph, built on first use. Most conversations never @@ -1200,6 +1214,11 @@ func (a *Agent) graph() *TaskGraph { // mid-run would be a run whose rules changed under it. graph.limit = a.config.TaskParallel graph.governor = newAdmissionGovernor(a.config.TaskMaxLoad, a.config.TaskMinFreeMB) + // AND THE MACHINE'S OWN ACCOUNT, if this process opened more than one + // conversation onto the same machine (task_pressure.go, #907). + if a.config.TaskLanes != nil { + graph.lanes = a.config.TaskLanes + } // The checkpoint is per-journal, so a session with no file gets a graph // with no disk behind it rather than a session that refuses to run tasks // (task_store.go). @@ -1500,7 +1519,7 @@ func (g *TaskGraph) runFrontier() { // node is about to send carries no reason to be waiting. node.held = "" if node.takesSlot() { - g.running++ + g.takeLaneLocked() } // THE NODE'S CONTEXT IS MADE HERE, IN THE SAME HOLD OF THE LOCK THAT // MARKED IT RUNNING, and not inside the goroutine below. That is what @@ -1600,19 +1619,50 @@ func (g *TaskGraph) holdOnStartingLocked(node *TaskNode, ready bool) string { return "" case g.limit > 0 && g.running >= g.limit: return waitingSlot - case !g.governor.admits(g.running): + case !g.governor.admits(g.lanes.running()): return waitingMachineBusy } return "" } -// lanesTaken is how many slot-taking nodes are running, read for the one -// reader that must have it outside the lock: the governor's reading, which -// learns what one node weighs from the memory the running ones hold. +// lanesTaken is how many slot-taking nodes THE WHOLE PROCESS is running, read +// for the one reader that must have it outside the lock: the governor's +// reading, which learns what one lane weighs from the memory the running ones +// hold. It is the process's and not this graph's because the memory in that +// reading is the whole process tree's and no conversation can be told apart +// from another inside it (task_pressure.go, #907). func (g *TaskGraph) lanesTaken() int { - g.mu.Lock() - defer g.mu.Unlock() - return g.running + return g.lanes.running() +} + +// takeLaneLocked records one lane taken by this graph, and it is THE ONLY +// PLACE `g.running` RISES. Two counts move together here and nowhere else: +// this graph's, which is what the person's task.parallel cap is measured +// against, and the process's, which is what the machine reading is divided by. +// A start written to one and not the other is the attribution gap #907 closed, +// reopened by hand. +// +// The lock is the caller's, as the name says, and the account's own lock is +// taken under it — always that way round, never the other (see [TaskLanes]). +func (g *TaskGraph) takeLaneLocked() { + g.running++ + g.lanes.take() +} + +// giveLaneLocked hands one lane back, and it is THE ONLY PLACE `g.running` +// FALLS. A graph holding none hands nothing back: the counts would otherwise +// drift below what is really running, and the process's account is shared, so +// this graph's arithmetic slip would be every other conversation's fan. +// +// The lock is the caller's. WHETHER a lane is owed back at all is the caller's +// question too — a parked node, a node that takes no slot — because this door +// is about keeping the two counts in step and not about who may hold one. +func (g *TaskGraph) giveLaneLocked() { + if g.running <= 0 { + return + } + g.running-- + g.lanes.give() } // armPoll sets the one clock this scheduler has. @@ -1997,8 +2047,8 @@ func (g *TaskGraph) handBackSlotLocked(node *TaskNode) { node.parked = false return } - if g.running > 0 && node.takesSlot() { - g.running-- + if node.takesSlot() { + g.giveLaneLocked() } } @@ -4884,9 +4934,7 @@ func (g *TaskGraph) park(node *TaskNode) { // flag, so that no reader can ever see the flag of one park beside the // number of another (see [TaskNode.parkGen]). node.parkGen++ - if g.running > 0 { - g.running-- - } + g.giveLaneLocked() g.mu.Unlock() // AND THE ROW SAYS SO. A hold is not a state — nothing about this node moved — // so it travels as [TaskNotice.Waiting] on an update of its own, exactly as a @@ -4925,7 +4973,7 @@ func (g *TaskGraph) unpark(node *TaskNode) { return } node.parked = false - g.running++ + g.takeLaneLocked() g.mu.Unlock() // A hold ENDING is news the same way a hold starting is, and the row would // otherwise wear "waiting · its parts" until whatever this node does next