The paced walk keeps a tenth of the time, and reads the pressure since its last look (review of issue 433)
mesh/merge-gate pass: builds mesh-host → ace, g14, novox, shanks; no bus step; every machine composes with the change as it did without (4 of 4 compose)
mesh/repo-check pass: its merge-check.sh passed
mesh/delivery superseded: a newer head of the same pull request

The review of #79 found that pacing alone brought back issue 361's
fault. The control-node's own I/O pressure sits near the 10 % bound with
no walk running, so a walk that paused 30 s for every 250 ms it read
would read 0.8 % of the time: a 16-minute walk would take some 32 hours,
outlive the age a search judges for, and be begun again at an engine
restart, leaving the machine unjudged.

A run of pauses is now at most nine times as long as the walk read since
the one before, so the walk keeps at least a tenth of the time whatever
the pressure, and takes at most ten times as long as unpaced: about
2.7 hours for the 16-minute walk, inside the 12 hours the controller
waits quietly. A test ties the bound to rootsearch.Quiet.

The pressure is the share of the time since the last look in which some
task waited on I/O, from PSI's cumulative counter, not its ten-second
average, which the kernel moves only every two seconds. A pause is
500 ms, then the pressure is looked at again.

The engine's walker is built in one place, engineWalker, and a test
fails if it is not paced by /proc/pressure/io. What the walk cannot read
is said through the search's own channel (the cache's Say), as the rest
of the search is, with what it means for the disk.
This commit is contained in:
2026-10-11 01:53:25 +02:00
parent 8f7db549ae
commit 30242197fa
3 changed files with 257 additions and 105 deletions
+93 -54
View File
@@ -5,20 +5,23 @@ package accounts
// disks run mq-deadline, which does not, and there the walk read at full priority: on 2026-10-11 it kept two
// spinning disks saturated for sixteen minutes, the bus whose store shares those disks stalled for half a minute
// twice, and the controller lost its lease each time. So the walk also paces itself by the machine's I/O pressure
// (PSI), which reads the same whatever the scheduler: while some task has waited on I/O for more than PaceAbove
// of the last ten seconds, the walk pauses before it reads the next directory.
// (PSI), which reads the same whatever the scheduler.
//
// The walk's own reads count in that pressure, so on a machine where nothing else waits it slows itself to keep
// the disk below the bound; that is the yielding the idle class promised. A pause in a row is at most PaceAtMost,
// then one more directory is read: on a machine that is never quiet the walk is slow, but it still ends, so a
// machine is never left unjudged for good (the fault issue 361 repaired).
// It looks every PaceLook at the share of the time since its last look in which some task waited on I/O — from
// the kernel's cumulative stall counter, not its ten-second average, which moves only every two seconds and would
// let the walk read blind that long. Above PaceAbove it pauses, PaceStep at a time, until the share falls below.
//
// **The walk keeps at least a tenth of the time, whatever the pressure** (the review of #79): a run of pauses is
// at most PaceRatio times as long as the walk read since the one before. The control-node's own load sits near
// PaceAbove with no walk running, so a walk that paused for as long as the pressure stayed high would take days,
// outlive the age a search judges for, and leave the machine unjudged — the fault issue 361 repaired. With the
// floor a walk takes at most 1 + PaceRatio times as long as it would unpaced, and its bursts are PaceLook long.
import (
"bufio"
"bytes"
"context"
"fmt"
"log"
"os"
"strconv"
"strings"
@@ -26,14 +29,17 @@ import (
)
const (
// PaceAbove is the I/O pressure, as the percentage of the last ten seconds in which some task waited on I/O
// (PSI's "some avg10"), above which the walk pauses.
// PaceAbove is the share of the time since the last look, in percent, in which some task waited on I/O, above
// which the walk pauses.
PaceAbove = 10.0
// PaceLook is how often the walk looks at the pressure while it walks.
// PaceLook is how long the walk reads between two looks at the pressure: the longest it reads blind.
PaceLook = 250 * time.Millisecond
// PacePause is one pause.
PacePause = 2 * time.Second
// PaceAtMost is the longest the walk pauses in a row before it reads one more directory.
// PaceStep is one pause, after which the pressure is looked at again.
PaceStep = 500 * time.Millisecond
// PaceRatio bounds a run of pauses to this many times the time the walk read since the run before: the walk
// keeps at least 1/(1+PaceRatio) of the time.
PaceRatio = 9
// PaceAtMost is the longest run of pauses, whatever the walk read before it.
PaceAtMost = 30 * time.Second
// ioPressureFile is where the kernel says the machine's I/O pressure.
ioPressureFile = "/proc/pressure/io"
@@ -41,67 +47,100 @@ const (
// pacer is what the walk asks before it reads a directory. A nil pacer never pauses.
type pacer struct {
pressure func() (float64, error) // some avg10, in percent
sleep func(context.Context, time.Duration) error
now func() time.Time
logf func(format string, args ...any)
// path is the pressure file read, kept so a test can see the engine's own walker reads the machine's.
path string
// stalled is the cumulative time some task waited on I/O (PSI's "some total").
stalled func() (time.Duration, error)
sleep func(context.Context, time.Duration) error
now func() time.Time
say func(string)
lastLook time.Time
lastLook time.Time // when the pressure was last looked at
readSince time.Time // when the walk last began to read, after a run of pauses or at its start
prevAt time.Time // the time of the last sample of the counter
prevTotal time.Duration // the counter at that sample
saidBlind bool
}
// newPacer paces by the machine's own I/O pressure.
func newPacer() *pacer {
// newPacer paces by the machine's own I/O pressure, and says through say what it cannot do.
func newPacer(say func(string)) *pacer {
return &pacer{
pressure: func() (float64, error) { return readIOPressure(ioPressureFile) },
sleep: sleepCtx,
now: time.Now,
logf: log.Printf,
path: ioPressureFile,
stalled: func() (time.Duration, error) { return readIOStalled(ioPressureFile) },
sleep: sleepCtx,
now: time.Now,
say: say,
}
}
// yield pauses while the machine's I/O pressure is above PaceAbove, PacePause at a time and PaceAtMost in a row,
// looking at most every PaceLook. It ends early only with ctx, whose error ends the walk as incomplete. Pressure
// that cannot be read pauses nothing, and is said once: the walk then yields only where the idle class does.
// yield pauses while the machine's I/O pressure is above PaceAbove: looking at most every PaceLook, pausing
// PaceStep at a time, a run of pauses at most PaceRatio times what the walk read since the last run and never more
// than PaceAtMost. It ends early only with ctx, whose error ends the walk as incomplete. Pressure that cannot be
// read pauses nothing, and is said once.
func (pc *pacer) yield(ctx context.Context) error {
if pc == nil || pc.pressure == nil {
if pc == nil || pc.stalled == nil {
return nil
}
if !pc.lastLook.IsZero() && pc.now().Sub(pc.lastLook) < PaceLook {
now := pc.now()
if pc.readSince.IsZero() {
pc.readSince = now
}
if !pc.lastLook.IsZero() && now.Sub(pc.lastLook) < PaceLook {
return nil
}
defer func() { pc.lastLook = pc.now() }()
for waited := time.Duration(0); waited < PaceAtMost; waited += PacePause {
p, err := pc.pressure()
if err != nil {
if !pc.saidBlind && pc.logf != nil {
pc.logf("the walk for setuid programs cannot read the machine's I/O pressure, so it does not pace "+
"itself and yields only where the disk's scheduler honours the idle class: %v", err)
}
pc.saidBlind = true
return nil
}
if p <= PaceAbove {
return nil
}
if err := pc.sleep(ctx, PacePause); err != nil {
pc.lastLook = now
share, known, err := pc.sample(now)
if err != nil || !known || share <= PaceAbove {
return nil
}
budget := min(PaceAtMost, PaceRatio*now.Sub(pc.readSince))
for paused := time.Duration(0); paused < budget; {
step := min(PaceStep, budget-paused)
if err := pc.sleep(ctx, step); err != nil {
return err
}
paused += step
share, known, err = pc.sample(pc.now())
if err != nil || (known && share <= PaceAbove) {
break
}
}
pc.readSince, pc.lastLook = pc.now(), pc.now()
return nil
}
// readIOPressure is the "some avg10" of a PSI file: the percentage of the last ten seconds in which some task
// waited on I/O.
func readIOPressure(path string) (float64, error) {
// sample is the share of the time since the last sample, in percent, in which some task waited on I/O; known is
// false for the first sample, which only sets where the next one counts from.
func (pc *pacer) sample(now time.Time) (float64, bool, error) {
total, err := pc.stalled()
if err != nil {
if !pc.saidBlind && pc.say != nil {
pc.say(fmt.Sprintf("the walk for setuid programs cannot read the machine's I/O pressure (%v), so it does "+
"not pace itself: where the disk's scheduler ignores the idle class, it may load the disk at full "+
"priority", err))
}
pc.saidBlind = true
return 0, false, err
}
prevAt, prevTotal := pc.prevAt, pc.prevTotal
pc.prevAt, pc.prevTotal = now, total
elapsed := now.Sub(prevAt)
if prevAt.IsZero() || elapsed <= 0 || total < prevTotal {
return 0, false, nil
}
return 100 * float64(total-prevTotal) / float64(elapsed), true, nil
}
// readIOStalled is the "some total" of a PSI file: the time, cumulative since boot, in which some task waited on I/O.
func readIOStalled(path string) (time.Duration, error) {
b, err := os.ReadFile(path)
if err != nil {
return 0, err
}
return parseIOPressure(b)
return parseIOStalled(b)
}
func parseIOPressure(b []byte) (float64, error) {
func parseIOStalled(b []byte) (time.Duration, error) {
sc := bufio.NewScanner(bytes.NewReader(b))
for sc.Scan() {
fields := strings.Fields(sc.Text())
@@ -109,16 +148,16 @@ func parseIOPressure(b []byte) (float64, error) {
continue
}
for _, f := range fields[1:] {
if v, ok := strings.CutPrefix(f, "avg10="); ok {
p, err := strconv.ParseFloat(v, 64)
if v, ok := strings.CutPrefix(f, "total="); ok {
us, err := strconv.ParseUint(v, 10, 64)
if err != nil {
return 0, fmt.Errorf("the I/O pressure's avg10 is not a number: %q", v)
return 0, fmt.Errorf("the I/O pressure's total is not a count of microseconds: %q", v)
}
return p, nil
return time.Duration(us) * time.Microsecond, nil
}
}
}
return 0, fmt.Errorf("the I/O pressure names no \"some\" line with avg10")
return 0, fmt.Errorf("the I/O pressure names no \"some\" line with a total")
}
// sleepCtx waits d, or until ctx ends.
+153 -50
View File
@@ -3,35 +3,54 @@ package accounts
import (
"context"
"errors"
"fmt"
"slices"
"strings"
"testing"
"time"
"github.com/novox/mesh-host/rootsearch"
)
// The walk for setuid programs yields to a busy disk by the machine's I/O pressure (novox/hq issue 433), because
// the idle I/O class it is put in is ignored by a disk scheduler that does not schedule by class, as the
// control-node's mq-deadline does not.
// control-node's mq-deadline does not — and it keeps a tenth of the time whatever the pressure, so it still ends.
// fakePace is a pacer whose pressure readings, sleeps and clock a test controls.
// fakePace is a pacer whose stall counter, sleeps and clock a test controls.
type fakePace struct {
readings []float64 // said in turn; the last is said again once they run out
read int
slept []time.Duration
clock time.Time
said []string
fail error
// shares is the stall share, in percent, of each interval between two reads of the counter, in turn; the last
// is said again once they run out. The first read only sets the counter's start.
shares []float64
i int
total time.Duration
lastRead time.Time
reads int
clock time.Time
// tick moves the clock on at every look at it, as a walk reading directories takes time; zero does not.
tick time.Duration
slept []time.Duration
said []string
fail error
}
func newFake(shares ...float64) *fakePace {
return &fakePace{shares: shares, clock: time.Unix(1_000_000, 0)}
}
func (f *fakePace) pacer() *pacer {
return &pacer{
pressure: func() (float64, error) {
stalled: func() (time.Duration, error) {
if f.fail != nil {
return 0, f.fail
}
i := min(f.read, len(f.readings)-1)
f.read++
return f.readings[i], nil
f.reads++
if !f.lastRead.IsZero() {
s := f.shares[min(f.i, len(f.shares)-1)]
f.i++
f.total += time.Duration(float64(f.clock.Sub(f.lastRead)) * s / 100)
}
f.lastRead = f.clock
return f.total, nil
},
sleep: func(ctx context.Context, d time.Duration) error {
if err := ctx.Err(); err != nil {
@@ -41,54 +60,111 @@ func (f *fakePace) pacer() *pacer {
f.clock = f.clock.Add(d)
return nil
},
now: func() time.Time { return f.clock },
logf: func(format string, args ...any) { f.said = append(f.said, fmt.Sprintf(format, args...)) },
now: func() time.Time {
f.clock = f.clock.Add(f.tick)
return f.clock
},
say: func(s string) { f.said = append(f.said, s) },
}
}
func TestTheWalkPausesWhileTheDiskIsUnderPressure(t *testing.T) {
f := &fakePace{readings: []float64{42, 18, PaceAbove, 3}, clock: time.Unix(1_000_000, 0)}
func (f *fakePace) paused() time.Duration {
var total time.Duration
for _, d := range f.slept {
total += d
}
return total
}
func TestTheWalkPausesWhileTheDiskIsUnderPressureAndGoesOnWhenItEases(t *testing.T) {
f := newFake(60, 40, 3)
pc := f.pacer()
if err := pc.yield(t.Context()); err != nil { // the first look sets where the counter counts from
t.Fatal(err)
}
f.clock = f.clock.Add(300 * time.Millisecond) // the walk reads
if err := pc.yield(t.Context()); err != nil {
t.Fatal(err)
}
// 42 and 18 are above the bound; PaceAbove itself is not.
if want := []time.Duration{PacePause, PacePause}; !slices.Equal(f.slept, want) {
// 60 % since the first look: pause; 40 % during the first pause: pause again; 3 %: go on.
if want := []time.Duration{PaceStep, PaceStep}; !slices.Equal(f.slept, want) {
t.Fatalf("slept %v, want %v", f.slept, want)
}
if f.read != 3 {
t.Fatalf("read the pressure %d times, want 3", f.read)
}
func TestTheFirstLookOnlySetsWhereTheNextCountsFrom(t *testing.T) {
f := newFake(99)
if err := f.pacer().yield(t.Context()); err != nil {
t.Fatal(err)
}
if len(f.slept) != 0 {
t.Fatalf("slept %v on the first look, which has nothing to compare", f.slept)
}
}
func TestTheWalkDoesNotPauseOnAQuietDisk(t *testing.T) {
f := &fakePace{readings: []float64{0.4}, clock: time.Unix(1_000_000, 0)}
if err := f.pacer().yield(t.Context()); err != nil {
t.Fatal(err)
f := newFake(0.4)
pc := f.pacer()
for range 10 {
if err := pc.yield(t.Context()); err != nil {
t.Fatal(err)
}
f.clock = f.clock.Add(PaceLook)
}
if len(f.slept) != 0 {
t.Fatalf("slept %v on a quiet disk", f.slept)
}
}
// A machine that is never quiet still sees the walk end: a pause in a row is bounded, then one more directory.
func TestTheWalkPausesAtMostPaceAtMostInARow(t *testing.T) {
f := &fakePace{readings: []float64{95}, clock: time.Unix(1_000_000, 0)}
// The control-node's own load sits near the bound with no walk running: the walk must still end (issue 361).
func TestTheWalkKeepsATenthOfTheTimeWhateverThePressure(t *testing.T) {
f := newFake(100)
pc := f.pacer()
var read time.Duration
for range 400 {
if err := pc.yield(t.Context()); err != nil {
t.Fatal(err)
}
f.clock = f.clock.Add(PaceLook)
read += PaceLook
}
if f.paused() == 0 {
t.Fatal("never paused under constant pressure")
}
if f.paused() > PaceRatio*read {
t.Fatalf("paused %v while it read %v: more than %d times as long", f.paused(), read, PaceRatio)
}
}
// What the floor promises against the ages a search is judged by: the walk that took 16m31s unpaced on the
// control-node on 2026-10-11, slowed to its worst, still completes well inside the time the controller stays quiet
// while it waits for a search, and so judges for most of its fresh age.
func TestTheSlowestPacedWalkStillJudgesInTime(t *testing.T) {
const unpaced = 16*time.Minute + 31*time.Second
worst := (1 + PaceRatio) * unpaced
if worst >= rootsearch.Quiet {
t.Fatalf("a walk paced to its worst takes %v, not less than the %v the controller waits quietly", worst,
rootsearch.Quiet)
}
}
func TestARunOfPausesIsAtMostPaceAtMost(t *testing.T) {
f := newFake(100)
pc := f.pacer()
if err := pc.yield(t.Context()); err != nil {
t.Fatal(err)
}
var total time.Duration
for _, d := range f.slept {
total += d
f.clock = f.clock.Add(10 * time.Minute) // read a long time without a look: the ratio would allow 90 minutes
if err := pc.yield(t.Context()); err != nil {
t.Fatal(err)
}
if total != PaceAtMost {
t.Fatalf("paused %v in a row, want %v", total, PaceAtMost)
if f.paused() != PaceAtMost {
t.Fatalf("paused %v in a row, want %v", f.paused(), PaceAtMost)
}
}
func TestTheWalkLooksAtThePressureAtMostEveryPaceLook(t *testing.T) {
f := &fakePace{readings: []float64{1}, clock: time.Unix(1_000_000, 0)}
f := newFake(1)
pc := f.pacer()
for range 5 {
if err := pc.yield(t.Context()); err != nil {
@@ -96,21 +172,22 @@ func TestTheWalkLooksAtThePressureAtMostEveryPaceLook(t *testing.T) {
}
f.clock = f.clock.Add(PaceLook / 10)
}
if f.read != 1 {
t.Fatalf("read the pressure %d times within %v, want once", f.read, PaceLook)
if f.reads != 1 {
t.Fatalf("read the pressure %d times within %v, want once", f.reads, PaceLook)
}
f.clock = f.clock.Add(PaceLook)
if err := pc.yield(t.Context()); err != nil {
t.Fatal(err)
}
if f.read != 2 {
t.Fatalf("read the pressure %d times after %v, want twice", f.read, PaceLook)
if f.reads != 2 {
t.Fatalf("read the pressure %d times after %v, want twice", f.reads, PaceLook)
}
}
// Pressure that cannot be read pauses nothing, and is said once rather than at every directory.
// Pressure that cannot be read pauses nothing, and is said once, with what it means, rather than at every look.
func TestAPressureThatCannotBeReadIsSaidOnce(t *testing.T) {
f := &fakePace{fail: errors.New("no such file"), clock: time.Unix(1_000_000, 0)}
f := newFake(0)
f.fail = errors.New("no such file")
pc := f.pacer()
for range 3 {
if err := pc.yield(t.Context()); err != nil {
@@ -121,12 +198,38 @@ func TestAPressureThatCannotBeReadIsSaidOnce(t *testing.T) {
if len(f.slept) != 0 || len(f.said) != 1 {
t.Fatalf("slept %v, said %q; want no pause and one line", f.slept, f.said)
}
if !strings.Contains(f.said[0], "full priority") {
t.Errorf("the line does not say what it means for the disk: %q", f.said[0])
}
}
// The engine's own walk is paced by the machine's pressure, and says through the search's own channel.
func TestTheEnginesWalkIsPacedByTheMachinesPressure(t *testing.T) {
var said []string
w := engineWalker(nil, func(s string) { said = append(said, s) })
if w.pace == nil || w.pace.path != ioPressureFile || w.pace.stalled == nil {
t.Fatalf("the engine's walker is not paced by %s: %+v", ioPressureFile, w.pace)
}
w.pace.say("x")
if len(said) != 1 {
t.Fatal("the engine's walker does not say through the search's channel")
}
}
// many is a tree of enough directories that a walk reaches a look at the pressure partway.
func many(t *testing.T) string {
names := []string{"bin/", "bin/su!", "bin/ls", "sbin/", "sbin/mount!", "usr/lib/helper%"}
for _, d := range []string{"a", "b", "c", "d", "e", "f", "g", "h"} {
names = append(names, "usr/share/"+d+"/")
}
return tree(t, names...)
}
// A walk ended while it pauses is an incomplete walk, never one that found nothing.
func TestAWalkEndedWhileItPausesIsIncomplete(t *testing.T) {
top := tree(t, "bin/", "bin/su!")
f := &fakePace{readings: []float64{80}, clock: time.Unix(1_000_000, 0)}
top := many(t)
f := newFake(80)
f.tick = 100 * time.Millisecond
ctx, cancel := context.WithCancel(t.Context())
defer cancel()
pc := f.pacer()
@@ -138,8 +241,7 @@ func TestAWalkEndedWhileItPausesIsIncomplete(t *testing.T) {
}
w := mine()
w.pace = pc
p := &SetuidProgress{Mounts: []string{top}}
err := w.walk(ctx, p, func() {})
err := w.walk(ctx, &SetuidProgress{Mounts: []string{top}}, func() {})
if len(f.slept) == 0 {
t.Fatal("the walk never paused, so this proves nothing about a walk ended while it pauses")
}
@@ -150,10 +252,11 @@ func TestAWalkEndedWhileItPausesIsIncomplete(t *testing.T) {
// Paced, the walk finds what it finds unpaced: pausing changes when, never what.
func TestAPacedWalkFindsWhatAnUnpacedWalkFinds(t *testing.T) {
top := tree(t, "bin/", "bin/su!", "bin/ls", "sbin/", "sbin/mount!", "usr/", "usr/lib/", "usr/lib/helper%")
top := many(t)
plain := walkAll(t, mine(), top)
f := &fakePace{readings: []float64{50, 2}, clock: time.Unix(1_000_000, 0)}
f := newFake(80, 2)
f.tick = 100 * time.Millisecond
w := mine()
w.pace = f.pacer()
paced := walkAll(t, w, top)
@@ -166,13 +269,13 @@ func TestAPacedWalkFindsWhatAnUnpacedWalkFinds(t *testing.T) {
}
func TestTheIOPressureIsReadAsTheKernelWritesIt(t *testing.T) {
got, err := parseIOPressure([]byte("some avg10=26.24 avg60=20.07 avg300=27.47 total=123456789\n" +
got, err := parseIOStalled([]byte("some avg10=26.24 avg60=20.07 avg300=27.47 total=123456789\n" +
"full avg10=21.90 avg60=15.77 avg300=22.47 total=98765432\n"))
if err != nil || got != 26.24 {
t.Fatalf("read %v, %v; want 26.24", got, err)
if err != nil || got != 123456789*time.Microsecond {
t.Fatalf("read %v, %v; want 123456789 µs", got, err)
}
for _, bad := range []string{"", "full avg10=1.00 avg60=0 avg300=0 total=0\n", "some avg10=x avg60=0\n"} {
if _, err := parseIOPressure([]byte(bad)); err == nil {
for _, bad := range []string{"", "full avg10=1.00 avg60=0 avg300=0 total=1\n", "some avg10=1 total=x\n"} {
if _, err := parseIOStalled([]byte(bad)); err == nil {
t.Errorf("%q was read as a pressure", bad)
}
}
+11 -1
View File
@@ -426,7 +426,11 @@ func contains(xs []string, s string) bool {
func (e Exec) setuidSearch(ctx context.Context, p *SetuidProgress, checkpoint func(), mountPoints map[string]bool) ([]string, error) {
walk := e.Walk
if walk == nil {
walk = walker{every: CheckpointEvery, mountPoints: mountPoints, pace: newPacer()}.walk
var say func(string)
if e.Cache != nil {
say = e.Cache.Say
}
walk = engineWalker(mountPoints, say).walk
}
if err := walk(ctx, p, checkpoint); err != nil {
return nil, fmt.Errorf("the walk for setuid programs did not complete: %w", err)
@@ -439,6 +443,12 @@ func (e Exec) setuidSearch(ctx context.Context, p *SetuidProgress, checkpoint fu
return e.unpackaged(ctx, slices.Compact(found))
}
// engineWalker is the walk the engine runs: checkpointed every CheckpointEvery, leaving other filesystems unread,
// and paced by the machine's I/O pressure (setuid_pace.go, novox/hq issue 433), saying through say what it cannot do.
func engineWalker(mountPoints map[string]bool, say func(string)) walker {
return walker{every: CheckpointEvery, mountPoints: mountPoints, pace: newPacer(say)}
}
// errorLines are the lines a command said on its error stream, as the runner words a failure: "<name> exited N:
// <its words>", the prefix taken off the first line. Empty lines are dropped.
func errorLines(err error, name string) []string {