mesh-host/internal/accounts/setuid_pace.go

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
	}
}