Merge pull request 'The setuid walk paces itself by the machine's I/O pressure (issue 433)' (#79) from fix/433-the-setuid-walk-yields-to-a-busy-disk into main

This commit was merged in pull request #79.
This commit is contained in:
2026-10-11 00:03:46 +00:00
4 changed files with 498 additions and 1 deletions
+175
View File
@@ -0,0 +1,175 @@
package accounts
// The walk for setuid programs yields to a busy disk (novox/hq issue 433). It asks before each directory and each
// file it reads, so no burst between two looks is longer than PaceLook, however large a directory. The walk is put in the idle I/O class
// (idle_linux.go), but only a disk scheduler that schedules by class honours that (BFQ does). The control-node's
// 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.
//
// 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"
"os"
"strconv"
"strings"
"time"
)
const (
// 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 long the walk reads between two looks at the pressure: the longest it reads blind.
PaceLook = 250 * time.Millisecond
// 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"
)
// pacer is what the walk asks before it reads a directory. A nil pacer never pauses.
type pacer struct {
// 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 // 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, and says through say what it cannot do.
func newPacer(say func(string)) *pacer {
return &pacer{
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: 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.stalled == nil {
return nil
}
now := pc.now()
if pc.readSince.IsZero() {
pc.readSince = now
}
if !pc.lastLook.IsZero() && now.Sub(pc.lastLook) < PaceLook {
return 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
}
}
resumed := pc.now()
pc.readSince, pc.lastLook = resumed, resumed
return nil
}
// 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 parseIOStalled(b)
}
func parseIOStalled(b []byte) (time.Duration, error) {
sc := bufio.NewScanner(bytes.NewReader(b))
for sc.Scan() {
fields := strings.Fields(sc.Text())
if len(fields) == 0 || fields[0] != "some" {
continue
}
for _, f := range fields[1:] {
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 total is not a count of microseconds: %q", v)
}
return time.Duration(us) * time.Microsecond, nil
}
}
}
return 0, fmt.Errorf("the I/O pressure names no \"some\" line with a total")
}
// sleepCtx waits d, or until ctx ends.
func sleepCtx(ctx context.Context, d time.Duration) error {
t := time.NewTimer(d)
defer t.Stop()
select {
case <-ctx.Done():
return ctx.Err()
case <-t.C:
return nil
}
}
+302
View File
@@ -0,0 +1,302 @@
package accounts
import (
"context"
"errors"
"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 — and it keeps a tenth of the time whatever the pressure, so it still ends.
// fakePace is a pacer whose stall counter, sleeps and clock a test controls.
type fakePace struct {
// 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{
stalled: func() (time.Duration, error) {
if f.fail != nil {
return 0, f.fail
}
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 {
return err
}
f.slept = append(f.slept, d)
f.clock = f.clock.Add(d)
return nil
},
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 (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)
}
// 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)
}
}
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 := 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)
}
}
// 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. A guard on the constants only: that the code keeps to the ratio is proven by
// TestTheWalkKeepsATenthOfTheTimeWhateverThePressure.
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)
}
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 f.paused() != PaceAtMost {
t.Fatalf("paused %v in a row, want %v", f.paused(), PaceAtMost)
}
}
func TestTheWalkLooksAtThePressureAtMostEveryPaceLook(t *testing.T) {
f := newFake(1)
pc := f.pacer()
for range 5 {
if err := pc.yield(t.Context()); err != nil {
t.Fatal(err)
}
f.clock = f.clock.Add(PaceLook / 10)
}
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.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, with what it means, rather than at every look.
func TestAPressureThatCannotBeReadIsSaidOnce(t *testing.T) {
f := newFake(0)
f.fail = errors.New("no such file")
pc := f.pacer()
for range 3 {
if err := pc.yield(t.Context()); err != nil {
t.Fatal(err)
}
f.clock = f.clock.Add(PaceLook)
}
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 := many(t)
f := newFake(80)
f.tick = 100 * time.Millisecond
ctx, cancel := context.WithCancel(t.Context())
defer cancel()
pc := f.pacer()
// The engine stops while the walk waits for the disk: the context ends inside the pause, not before the walk.
pc.sleep = func(ctx context.Context, d time.Duration) error {
f.slept = append(f.slept, d)
cancel()
return ctx.Err()
}
w := mine()
w.pace = pc
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")
}
if !errors.Is(err, context.Canceled) {
t.Fatalf("a walk ended while it paused answered %v", err)
}
}
// Paced, the walk finds what it finds unpaced: pausing changes when, never what.
func TestAPacedWalkFindsWhatAnUnpacedWalkFinds(t *testing.T) {
top := many(t)
plain := walkAll(t, mine(), top)
f := newFake(80, 2)
f.tick = 100 * time.Millisecond
w := mine()
w.pace = f.pacer()
paced := walkAll(t, w, top)
if !slices.Equal(plain, paced) {
t.Fatalf("paced walk found %v, unpaced %v", paced, plain)
}
if len(f.slept) == 0 {
t.Fatal("the walk read its directories without asking the pacer: it never paused under pressure")
}
}
func TestTheIOPressureIsReadAsTheKernelWritesIt(t *testing.T) {
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 != 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=1\n", "some avg10=1 total=x\n"} {
if _, err := parseIOStalled([]byte(bad)); err == nil {
t.Errorf("%q was read as a pressure", bad)
}
}
}
// One directory of many files is paced as it is read, not only between directories: a file's own details may be
// read from the disk too.
func TestADirectoryOfManyFilesIsPacedAsItIsRead(t *testing.T) {
names := []string{"bin/"}
for i := range 20 {
names = append(names, "bin/f"+strings.Repeat("x", i+1))
}
top := tree(t, names...)
f := newFake(80, 2)
f.tick = 100 * time.Millisecond
w := mine()
w.pace = f.pacer()
walkAll(t, w, top)
// Two directories alone would be looked at once, 200 ms apart, within one look: no pause at all.
if len(f.slept) == 0 {
t.Fatal("a directory of twenty files was read without a pause under pressure")
}
}
+10
View File
@@ -125,6 +125,8 @@ type walker struct {
rootUID, rootGID uint32
every time.Duration
mountPoints map[string]bool
// pace is asked before each directory is read (setuid_pace.go); nil never pauses.
pace *pacer
}
// walk carries p on to its end, calling checkpoint every w.every and after each filesystem. It ends early only
@@ -232,7 +234,15 @@ func (w walker) walkMount(ctx context.Context, p *SetuidProgress, mount string,
if d, ok := deviceOf(info); !ok || d != dev {
return filepath.SkipDir // another filesystem, walked as its own or not at all (find's -xdev)
}
// Before its entries are read: a busy disk is waited for here (novox/hq issue 433).
if err := w.pace.yield(ctx); err != nil {
return err
}
case d.Type().IsRegular():
// A file's own details may be read from the disk too: a directory of many files is paced as it is read.
if err := w.pace.yield(ctx); err != nil {
return err
}
info, err := d.Info() // the entry's own, never through a link
if errors.Is(err, fs.ErrNotExist) {
return nil
+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}.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 {