diff --git a/internal/accounts/setuid_pace.go b/internal/accounts/setuid_pace.go new file mode 100644 index 0000000..9cb23d2 --- /dev/null +++ b/internal/accounts/setuid_pace.go @@ -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 + } +} diff --git a/internal/accounts/setuid_pace_test.go b/internal/accounts/setuid_pace_test.go new file mode 100644 index 0000000..5d117ef --- /dev/null +++ b/internal/accounts/setuid_pace_test.go @@ -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") + } +} diff --git a/internal/accounts/setuid_walk.go b/internal/accounts/setuid_walk.go index 30c66f2..dac285a 100644 --- a/internal/accounts/setuid_walk.go +++ b/internal/accounts/setuid_walk.go @@ -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 diff --git a/internal/accounts/ways.go b/internal/accounts/ways.go index 90d68e9..5e6f912 100644 --- a/internal/accounts/ways.go +++ b/internal/accounts/ways.go @@ -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: " exited N: // ", the prefix taken off the first line. Empty lines are dropped. func errorLines(err error, name string) []string {