amadan

carlos / platform Public

cgroup: emptiness and coldness reads agree with the in-flight registry (#272)

Whole-branch review round 1, finding 1. A reader that asked the registry
and the kernel one after the other could miss a placement moving from one
to the other between the two reads.

- readQuiet brackets each of Cleanup's kernel reads (population, members):
  quiet only if no placement under the target was in flight at any moment
  of the read. Not quiet is populated, or ErrPlacementInFlight.
- ensureParent takes holdPlacements over the host's parent: no placement
  in or below it registers from before the first read until the cold
  parent's controllers and limits are written. The counts in flight are
  taken when the hold begins. The hold spans cgroupfs reads and writes only.
- KillAndEmpty's doc says what it guarantees: a placement that registers
  after the wait is never missed, but it is not killed (cgroup.kill is
  written once), so the proof ends unproven at ctx's end. Never a false
  empty.

Deterministic tests through two seams (placementReadHook,
placementHeldHook) for both schedules: a registration during a read, and
a placement that ends between the kernel read and the registry's.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude 306aa76d9fddba691503c3e7693d305c5dbb7bf6 parent 02472fe
7 files changed, +535 −40
  • internal/cgroup/cgroup.go +5 −1
  • internal/cgroup/cleanup_linux.go +42 −25
  • internal/cgroup/cleanup_linux_test.go +107 −0
  • internal/cgroup/gen.go +17 −2
  • internal/cgroup/gen_test.go +113 −0
  • internal/cgroup/inflight.go +159 −12
  • internal/cgroup/inflight_linux_test.go +92 −0
diff --git a/internal/cgroup/cgroup.go b/internal/cgroup/cgroup.go
index bab3730a5..43f449ba1 100644
--- a/internal/cgroup/cgroup.go
+++ b/internal/cgroup/cgroup.go
@@ -179,10 +179,14 @@ type Manager struct {
locks map[string]*sync.Mutex
// placing is the in-flight registry (inflight.go): placements by
// cleaned target directory. placeWake is closed and dropped whenever a
- // count drops, so waitPlacementsUnder wakes and rechecks.
+ // count drops or a hold ends, so waitPlacementsUnder and a registration
+ // waiting on a hold wake and recheck. windows are the readQuiet reads in
+ // progress; holds, the directories no placement may register under.
placeMu sync.Mutex
placing map[string]int
placeWake chan struct{}
+ windows map[*placeWindow]struct{}
+ holds map[string]int
}
// hostLock is host's start lock. A start holds it while it checks for, and
diff --git a/internal/cgroup/cleanup_linux.go b/internal/cgroup/cleanup_linux.go
index de8918ac8..66095cc62 100644
--- a/internal/cgroup/cleanup_linux.go
+++ b/internal/cgroup/cleanup_linux.go
@@ -143,26 +143,30 @@ func (c *Cleanup) Populated() (bool, error) {
}
// populationLocked is the pinned cgroup's population. A process being
-// placed there or below it (the in-flight registry) makes it populated
-// before cgroup.events is read: a shim spends its first milliseconds in the
-// agent's cgroup, and a flat fallback's child runs there until its PID is
-// moved, so the kernel cannot see either yet.
+// placed there or below it (the in-flight registry) makes it populated: a
+// shim spends its first milliseconds in the agent's cgroup, and a flat
+// fallback's child runs there until its PID is moved, so the kernel cannot
+// see either yet. cgroup.events is read inside readQuiet, so a placement
+// that registers during the read, or lands and ends its registration
+// between the read and the registry's, still counts (whole-branch review,
+// finding 1).
func (c *Cleanup) populationLocked() (bool, error) {
p := c.state.pin
if err := p.identityOK(); err != nil {
return false, err
}
- if c.state.m.placementsUnder(c.state.path) > 0 {
- return true, nil
- }
- v, err := strictPopulation(p.events)
+ var (
+ v bool
+ err error
+ )
+ quiet := c.state.m.readQuiet(c.state.path, func() { v, err = strictPopulation(p.events) })
if err != nil {
return false, err
}
if err := p.identityOK(); err != nil {
return false, err
}
- return v, nil
+ return v || !quiet, nil
}
func (c *Cleanup) Members() ([]int, error) {
@@ -188,22 +192,30 @@ const (
// generations'. A flat tenant runs as the agent's uid and can make cgroups
// of its own under its generation, so the walk goes to any depth, bounded.
// While a process is being placed there or below it, its PID is not in any
-// cgroup.procs yet, so the list would be short: ErrPlacementInFlight.
+// cgroup.procs yet, so the list would be short: ErrPlacementInFlight. The
+// walk runs inside readQuiet, so a placement in flight at any moment of it
+// counts, not only one in flight before it began.
func (c *Cleanup) membersLocked() ([]int, error) {
p := c.state.pin
if err := p.identityOK(); err != nil {
return nil, err
}
- if c.state.m.placementsUnder(c.state.path) > 0 {
- return nil, ErrPlacementInFlight
- }
- members, err := readProcs(c.state.procs)
+ var (
+ members []int
+ err error
+ )
+ quiet := c.state.m.readQuiet(c.state.path, func() {
+ if members, err = readProcs(c.state.procs); err != nil {
+ return
+ }
+ groups := 0
+ err = walkMembers(p.dir, 1, &groups, &members)
+ })
if err != nil {
return nil, err
}
- groups := 0
- if err := walkMembers(p.dir, 1, &groups, &members); err != nil {
- return nil, err
+ if !quiet {
+ return nil, ErrPlacementInFlight
}
slices.Sort(members)
if err := p.identityOK(); err != nil {
@@ -320,13 +332,18 @@ var cleanupKillWrite = func(f *os.File) error { return strictWrite(f, "1") }
// registry, Ruling 6) is a member the kernel cannot see yet. KillAndEmpty
// first waits for every such placement to resolve, bounded by ctx; an
// error there is unproven (ErrPlacementInFlight with ctx's error), and
-// nothing is killed. Every emptiness check after the kill goes through the
-// same registry, so a placement that registers after the wait keeps the
-// proof from finishing until the placed process has been killed with the
-// rest, or ctx ends (unproven). Callers still exclude new starts during a
-// proof as they always have (the lease, and the before hook a start asks
-// immediately before exec); the registry only closes the window a shim, or
-// a flat start awaiting its PID move, spends outside its cgroup.
+// nothing is killed. cgroup.kill is then written once. Every emptiness
+// check after it reads the kernel inside readQuiet, so a placement in
+// flight at any moment of a check makes that check populated. A placement
+// that registers after the wait is therefore never missed, but it is not
+// killed either: the kill was written before it landed, and is not written
+// again. Its process keeps the cgroup populated, so the proof ends
+// unproven when ctx ends, unless that process exits first. Never a false
+// empty; a late placement costs the proof, not the tenant. Callers still
+// exclude new starts during a proof as they always have (the lease, and
+// the before hook a start asks immediately before exec); the registry only
+// closes the window a shim, or a flat start awaiting its PID move, spends
+// outside its cgroup.
func (c *Cleanup) KillAndEmpty(ctx context.Context) error {
if c == nil || c.state == nil {
return ErrStrictState
@@ -383,7 +400,7 @@ func (c *Cleanup) KillAndEmpty(ctx context.Context) error {
}
members, err := c.membersLocked()
if errors.Is(err, ErrPlacementInFlight) {
- // Registered since the population read: not empty yet.
+ // In flight during the members read: not empty yet.
populated, members, err = true, nil, nil
}
if err != nil {
diff --git a/internal/cgroup/cleanup_linux_test.go b/internal/cgroup/cleanup_linux_test.go
index 331059b5d..35249e87e 100644
--- a/internal/cgroup/cleanup_linux_test.go
+++ b/internal/cgroup/cleanup_linux_test.go
@@ -408,3 +408,110 @@ func TestKillAndEmptySeesAPlacementRegisteredMidProof(t *testing.T) {
t.Fatalf("KillAndEmpty with nothing in flight = %v, want the proof", err)
}
}
+
+// Whole-branch review, finding 1, first schedule: a placement registers
+// while the kernel is being read (the hook runs right after the read, before
+// the registry is asked again). The fake cgroupfs says empty, so only the
+// registry can stop a false empty: Populated must answer populated and
+// Members must refuse, even though nothing was in flight when either began.
+func TestCleanupCountsAPlacementRegisteredDuringTheRead(t *testing.T) {
+ m, host, dir := cleanupFixture(t)
+ writeFixture(t, filepath.Join(dir, "cgroup.events"), "populated 0\n")
+ writeFixture(t, filepath.Join(dir, "cgroup.procs"), "")
+ c := fixtureCleanup(t, m, host)
+ var dones []func()
+ t.Cleanup(func() {
+ for _, d := range dones {
+ d()
+ }
+ })
+ reads := 0
+ setPlacementReadHook(t, func(string) {
+ reads++
+ dones = append(dones, m.beginPlacement(filepath.Join(dir, "g1")))
+ })
+ if populated, err := c.Populated(); err != nil || !populated {
+ t.Fatalf("Populated = %v, %v with a placement registered during the read: want true", populated, err)
+ }
+ for _, d := range dones {
+ d()
+ }
+ if members, err := c.Members(); !errors.Is(err, ErrPlacementInFlight) {
+ t.Fatalf("Members = %v, %v with a placement registered during the read: want ErrPlacementInFlight", members, err)
+ }
+ if reads != 2 {
+ t.Fatalf("the read hook ran %d times, want 2: the test proves nothing", reads)
+ }
+ // The control: the same reads with nothing registering answer empty.
+ for _, d := range dones {
+ d()
+ }
+ placementReadHook = nil
+ if populated, err := c.Populated(); err != nil || populated {
+ t.Fatalf("Populated = %v, %v with nothing in flight: want false", populated, err)
+ }
+ if members, err := c.Members(); err != nil || len(members) != 0 {
+ t.Fatalf("Members = %v, %v with nothing in flight: want none", members, err)
+ }
+}
+
+// Finding 1, second schedule: a placement in flight when the kernel is read
+// lands and ends its registration between the kernel read and the
+// registry's. Asked in that order, the registry would read zero beside a
+// kernel view taken before the process arrived. Both answers must count it.
+func TestCleanupCountsAPlacementThatEndsBetweenTheReads(t *testing.T) {
+ m, host, dir := cleanupFixture(t)
+ writeFixture(t, filepath.Join(dir, "cgroup.events"), "populated 0\n")
+ writeFixture(t, filepath.Join(dir, "cgroup.procs"), "")
+ c := fixtureCleanup(t, m, host)
+ var done func()
+ setPlacementReadHook(t, func(string) {
+ if done != nil {
+ done()
+ }
+ })
+ done = m.beginPlacement(filepath.Join(dir, "g1"))
+ if populated, err := c.Populated(); err != nil || !populated {
+ t.Fatalf("Populated = %v, %v with a placement that ended after the kernel read: want true", populated, err)
+ }
+ done = m.beginPlacement(filepath.Join(dir, "g1"))
+ if members, err := c.Members(); !errors.Is(err, ErrPlacementInFlight) {
+ t.Fatalf("Members = %v, %v with a placement that ended after the kernel read: want ErrPlacementInFlight", members, err)
+ }
+ if m.placementsUnder(dir) != 0 {
+ t.Fatal("the hook did not end the placements: the test proves nothing")
+ }
+}
+
+// Finding 1 through KillAndEmpty: a placement registered during each
+// emptiness read after the kill keeps the proof from finishing, so it ends
+// unproven at ctx's end and never answers empty.
+func TestKillAndEmptyCountsAPlacementRegisteredDuringARead(t *testing.T) {
+ m, host, dir := cleanupFixture(t)
+ writeFixture(t, filepath.Join(dir, "cgroup.events"), "populated 0\n")
+ writeFixture(t, filepath.Join(dir, "cgroup.procs"), "")
+ c := fixtureCleanup(t, m, host)
+ var dones []func()
+ t.Cleanup(func() {
+ for _, d := range dones {
+ d()
+ }
+ })
+ setPlacementReadHook(t, func(string) {
+ // Each read sees one registration begin; the previous one ends, so
+ // a check that asked the registry only before its read would see
+ // nothing in flight.
+ for _, d := range dones {
+ d()
+ }
+ dones = append(dones, m.beginPlacement(filepath.Join(dir, "g1")))
+ })
+ ctx, cancel := context.WithTimeout(t.Context(), 100*time.Millisecond)
+ defer cancel()
+ if err := c.KillAndEmpty(ctx); !errors.Is(err, context.DeadlineExceeded) {
+ t.Fatalf("KillAndEmpty = %v, want unproven at ctx's end (context.DeadlineExceeded)", err)
+ }
+ if len(dones) == 0 {
+ t.Fatal("no emptiness read ran: the test proves nothing")
+ }
+}
diff --git a/internal/cgroup/gen.go b/internal/cgroup/gen.go
index 0b7b604be..6aba3fce2 100644
--- a/internal/cgroup/gen.go
+++ b/internal/cgroup/gen.go
@@ -191,6 +191,15 @@ func parentProcs(dir string) (int, error) {
// as it is, because a swap raised it on purpose, once it is shown to
// delegate memory to its children.
//
+// The decision and the writes it allows are one step against the registry
+// (whole-branch review, finding 1): from before the first read until the
+// limits are written, no placement into the parent or below it can
+// register (holdPlacements), and the placements already in flight are
+// counted when the hold begins. So no sibling can move from "being placed"
+// to "in the kernel's view" unseen between the reads, and none can start
+// while a cold parent is being reset. A placement that waits on the hold
+// waits only for these cgroupfs reads and writes.
+//
// Nothing here is best-effort (Task 7): a generation started past any of
// these failures would run without a limit, with nothing to say so. A
// populated parent is ErrLegacyParentPopulated, and an unreadable
@@ -205,6 +214,8 @@ func (m *Manager) ensureParent(host string, tier Limit) error {
m.logf("cgroup: %s: create parent cgroup: %v (generation start refused)", host, err)
return fmt.Errorf("%w: %s: create parent: %w", ErrLimitNotApplied, host, err)
}
+ inflightAt, inflightUnder, release := m.holdPlacements(dir)
+ defer release()
n, err := parentProcs(dir)
if err != nil {
m.logf("cgroup: %s: generation start refused: cannot read the parent's cgroup.procs: %v", host, err)
@@ -212,14 +223,18 @@ func (m *Manager) ensureParent(host string, tier Limit) error {
}
// A flat start being placed into the parent itself is a direct process
// the moment it lands (Ruling 6): count it now.
- n += m.placementsAt(dir)
+ n += inflightAt
if n > 0 {
m.logf("cgroup: %s: generation start refused: the parent cgroup holds %d process(es) directly, so no memory limit could apply to a generation under it; stop them or let them exit", host, n)
return fmt.Errorf("%w: %s holds %d process(es)", ErrLegacyParentPopulated, host, n)
}
// A sibling generation being placed is a live one: the parent is not
// cold, and its limits (a swap may have raised them) are left alone.
- if parentPopulated(dir) || m.placementsUnder(dir) > 0 {
+ populated := parentPopulated(dir)
+ if placementReadHook != nil {
+ placementReadHook(dir)
+ }
+ if populated || inflightUnder > 0 {
if err := delegatesMemory(dir); err != nil {
m.logf("cgroup: %s: generation start refused: %v", host, err)
return fmt.Errorf("%s: %w", host, err)
diff --git a/internal/cgroup/gen_test.go b/internal/cgroup/gen_test.go
index 7bfdb716d..b9e37ce18 100644
--- a/internal/cgroup/gen_test.go
+++ b/internal/cgroup/gen_test.go
@@ -10,8 +10,10 @@ import (
"slices"
"strconv"
"strings"
+ "sync"
"syscall"
"testing"
+ "time"
)
// quietOOM points the oom_score_adj write at a temp dir, so a fake start
@@ -901,3 +903,114 @@ func TestGenerationCancelledStartIsNotABoxCondition(t *testing.T) {
})
}
}
+
+// Whole-branch review, finding 1, at ensureParent: a sibling generation in
+// flight when the parent's population is read lands, and ends its
+// registration, before the registry is asked. Read in that order, the
+// parent would look cold (the kernel read predates the process, the
+// registry no longer lists it) and its limits, which a swap may have raised,
+// would be reset. The hook ends the placement right after the kernel read.
+func TestEnsureParentCountsASiblingThatLandsBetweenTheReads(t *testing.T) {
+ const host = "h.example.com"
+ m, parent := genManager(t, host)
+ fakeStart(t, func(*exec.Cmd) error { return nil })
+ writeFixture(t, filepath.Join(parent, "cgroup.events"), "populated 0\nfrozen 0\n")
+ writeFixture(t, filepath.Join(parent, "cgroup.subtree_control"), "cpu memory io pids\n")
+ writeFixture(t, filepath.Join(parent, "memory.max"), "1073741824")
+ done := m.beginPlacement(filepath.Join(parent, GenChild(1)))
+ defer done()
+ hooked := false
+ setPlacementReadHook(t, func(string) {
+ if !hooked {
+ hooked = true
+ done() // g1's shim answered placed: the registry drops it
+ }
+ })
+ if _, err := m.StartChildGated(host, GenChild(2), TierDefault, TierDefault, buildTrue, nil); err != nil {
+ t.Fatalf("StartChildGated beside a sibling that landed: %v", err)
+ }
+ if !hooked {
+ t.Fatal("the read hook never ran: the test proves nothing")
+ }
+ if got := readFixture(t, filepath.Join(parent, "memory.max")); got != "1073741824" {
+ t.Errorf("the parent's memory.max was rewritten to %q: a sibling that landed between the reads was taken for a cold parent", got)
+ }
+ if got := readFixture(t, filepath.Join(parent, "cgroup.subtree_control")); got != "cpu memory io pids\n" {
+ t.Errorf("the parent's subtree_control was rewritten to %q", got)
+ }
+}
+
+// Finding 1, the other schedule at ensureParent: a sibling placement that
+// tries to register after the parent was found cold must wait until the
+// cold parent's controllers and limits are written. Otherwise the reset
+// could land on a parent that already has a live generation under it. The
+// hook starts the sibling's registration right after the population read
+// and waits until the registration reports that a hold stopped it.
+func TestEnsureParentHoldsPlacementsUntilTheColdParentIsWritten(t *testing.T) {
+ const host = "h.example.com"
+ m, parent := genManager(t, host)
+ fakeStart(t, func(*exec.Cmd) error { return nil })
+ writeFixture(t, filepath.Join(parent, "cgroup.events"), "populated 0\nfrozen 0\n")
+ blocked := make(chan struct{})
+ var once sync.Once
+ origHeld := placementHeldHook
+ placementHeldHook = func(string) { once.Do(func() { close(blocked) }) }
+ t.Cleanup(func() { placementHeldHook = origHeld })
+
+ type seen struct {
+ max, subtree string
+ done func()
+ }
+ registered := make(chan seen, 1)
+ hooked := false
+ setPlacementReadHook(t, func(string) {
+ if hooked {
+ return
+ }
+ hooked = true
+ go func() {
+ done := m.beginPlacement(filepath.Join(parent, GenChild(1)))
+ // What the sibling's start would find once it is registered.
+ max, _ := os.ReadFile(filepath.Join(parent, "memory.max"))
+ sub, _ := os.ReadFile(filepath.Join(parent, "cgroup.subtree_control"))
+ registered <- seen{string(max), string(sub), done}
+ }()
+ select {
+ case <-blocked:
+ case r := <-registered:
+ r.done()
+ t.Errorf("a sibling registered while the cold parent was being written (memory.max %q)", r.max)
+ case <-time.After(5 * time.Second):
+ t.Error("the sibling's registration neither registered nor waited on a hold")
+ }
+ })
+ if _, err := m.StartChildGated(host, GenChild(2), TierDefault, TierDefault, buildTrue, nil); err != nil {
+ t.Fatalf("StartChildGated on a cold parent: %v", err)
+ }
+ if !hooked {
+ t.Fatal("the read hook never ran: the test proves nothing")
+ }
+ select {
+ case r := <-registered:
+ defer r.done()
+ if want := strconv.FormatInt(TierDefault.MemoryMax, 10); r.max != want {
+ t.Errorf("the sibling registered with the parent's memory.max %q, want the cold parent's tier %q written first", r.max, want)
+ }
+ if !strings.Contains(r.subtree, "+memory") {
+ t.Errorf("the sibling registered with the parent's subtree_control %q, want the controllers delegated first", r.subtree)
+ }
+ case <-time.After(5 * time.Second):
+ t.Fatal("the sibling's registration was never released after the parent was written")
+ }
+ if n := m.placementsUnder(parent); n != 1 {
+ t.Fatalf("%d placements registered, want the sibling's one", n)
+ }
+}
+
+// setPlacementReadHook sets placementReadHook for one test.
+func setPlacementReadHook(t *testing.T, f func(dir string)) {
+ t.Helper()
+ orig := placementReadHook
+ placementReadHook = f
+ t.Cleanup(func() { placementReadHook = orig })
+}
diff --git a/internal/cgroup/inflight.go b/internal/cgroup/inflight.go
index f86596b96..a11197f5c 100644
--- a/internal/cgroup/inflight.go
+++ b/internal/cgroup/inflight.go
@@ -15,19 +15,69 @@ import (
// PID-move fallback (startAndMoveIn) registers its target as well, from
// its start until its PID write has returned. Readers: Cleanup's
// population and members (cleanup_linux.go) and ensureParent (gen.go).
+//
+// A registration ends only once its process is in the kernel's view of its
+// cgroup (or never will be), so a placement is always visible to one of the
+// two: the registry, or the kernel. But a reader that asks them one after
+// the other can miss a placement that moves from one to the other between
+// the two reads (whole-branch review, finding 1). Two tools close that:
+//
+// - readQuiet brackets one kernel read: it answers quiet only when no
+// placement into the directory or below it was in flight at any moment
+// of that read (none when it began, none registered or ended during it,
+// none at its end). A reader that is not quiet treats the directory as
+// populated, or its member list as unproven. Cleanup's reads use it.
+// - holdPlacements stops any placement into the directory or below it
+// from registering until it is released, so a decision taken under it
+// (ensureParent's cold parent) holds until its writes are done. It is
+// held across cgroupfs reads and writes only, never across a before
+// callback, a shim wait or a start.
+//
+// No lock of the registry is ever held across a before callback, a shim
+// wait or a start: placeMu guards only the maps, and a registration blocked
+// by a hold waits on a channel with placeMu released.
+
+// placementReadHook, when set, runs right after each kernel read that
+// readQuiet brackets, and after ensureParent's population read. Only test
+// files set it, to register or end a placement at exactly that point.
+var placementReadHook func(dir string)
+
+// placementHeldHook, when set, runs when a registration finds a hold over
+// its target and is about to wait for it. Only test files set it.
+var placementHeldHook func(dir string)
+
+// placeWindow is one readQuiet in progress: busy means a placement under
+// dir was in flight when it began; changed, that one registered or ended
+// under dir since.
+type placeWindow struct {
+ dir string
+ busy bool
+ changed bool
+}
// beginPlacement registers one placement into dir. done ends it, and is
// idempotent: a second call does nothing. A nil manager registers nothing.
+// While a hold covers dir (holdPlacements), it waits for the hold to end.
func (m *Manager) beginPlacement(dir string) (done func()) {
if m == nil {
return func() {}
}
dir = filepath.Clean(dir)
m.placeMu.Lock()
+ for m.heldLocked(dir) {
+ wake := m.wakeLocked()
+ m.placeMu.Unlock()
+ if placementHeldHook != nil {
+ placementHeldHook(dir)
+ }
+ <-wake
+ m.placeMu.Lock()
+ }
if m.placing == nil {
m.placing = make(map[string]int)
}
m.placing[dir]++
+ m.touchLocked(dir)
m.placeMu.Unlock()
var once sync.Once
return func() {
@@ -37,11 +87,71 @@ func (m *Manager) beginPlacement(dir string) (done func()) {
if m.placing[dir]--; m.placing[dir] <= 0 {
delete(m.placing, dir)
}
+ m.touchLocked(dir)
// Wake every waiter; each rechecks its own subtree.
- if m.placeWake != nil {
- close(m.placeWake)
- m.placeWake = nil
+ m.broadcastLocked()
+ })
+ }
+}
+
+// readQuiet runs read, a read of the kernel's view of dir, and reports
+// whether no placement into dir or below it was in flight at any moment
+// while read ran. Only a quiet read covers every process bound for dir: a
+// placement that ended before read began is in the kernel's view, and one
+// that registers after read ended had not forked while it ran. A nil
+// manager has no placements: always quiet.
+func (m *Manager) readQuiet(dir string, read func()) (quiet bool) {
+ dir = filepath.Clean(dir)
+ if m == nil {
+ read()
+ return true
+ }
+ w := &placeWindow{dir: dir}
+ m.placeMu.Lock()
+ w.busy = m.countUnderLocked(dir) > 0
+ if m.windows == nil {
+ m.windows = make(map[*placeWindow]struct{})
+ }
+ m.windows[w] = struct{}{}
+ m.placeMu.Unlock()
+ read()
+ if placementReadHook != nil {
+ placementReadHook(dir)
+ }
+ m.placeMu.Lock()
+ defer m.placeMu.Unlock()
+ delete(m.windows, w)
+ return !w.busy && !w.changed && m.countUnderLocked(dir) == 0
+}
+
+// holdPlacements keeps every placement into dir or below it from
+// registering until release is called, and answers the placements already
+// in flight: at, into dir itself, and under, into dir or below it. With
+// the hold in place those counts can only fall, and a count of zero stays
+// zero, so a kernel read taken under the hold, with both zero, sees every
+// process bound for dir. release is idempotent. Hold it only across
+// cgroupfs reads and writes.
+func (m *Manager) holdPlacements(dir string) (at, under int, release func()) {
+ if m == nil {
+ return 0, 0, func() {}
+ }
+ dir = filepath.Clean(dir)
+ m.placeMu.Lock()
+ if m.holds == nil {
+ m.holds = make(map[string]int)
+ }
+ m.holds[dir]++
+ at, under = m.placing[dir], m.countUnderLocked(dir)
+ m.placeMu.Unlock()
+ var once sync.Once
+ return at, under, func() {
+ once.Do(func() {
+ m.placeMu.Lock()
+ defer m.placeMu.Unlock()
+ if m.holds[dir]--; m.holds[dir] <= 0 {
+ delete(m.holds, dir)
}
+ m.broadcastLocked()
})
}
}
@@ -79,10 +189,7 @@ func (m *Manager) waitPlacementsUnder(ctx context.Context, dir string) error {
m.placeMu.Unlock()
return nil
}
- if m.placeWake == nil {
- m.placeWake = make(chan struct{})
- }
- wake := m.placeWake
+ wake := m.wakeLocked()
m.placeMu.Unlock()
select {
case <-ctx.Done():
@@ -92,16 +199,56 @@ func (m *Manager) waitPlacementsUnder(ctx context.Context, dir string) error {
}
}
-func (m *Manager) countUnderLocked(dir string) int {
- prefix := dir
- if !strings.HasSuffix(prefix, "/") {
- prefix += "/"
+// wakeLocked is the channel the next broadcastLocked closes.
+func (m *Manager) wakeLocked() chan struct{} {
+ if m.placeWake == nil {
+ m.placeWake = make(chan struct{})
+ }
+ return m.placeWake
+}
+
+// broadcastLocked wakes every waiter: a registration ended, or a hold did.
+func (m *Manager) broadcastLocked() {
+ if m.placeWake != nil {
+ close(m.placeWake)
+ m.placeWake = nil
}
+}
+
+// touchLocked marks every open readQuiet over target as changed.
+func (m *Manager) touchLocked(target string) {
+ for w := range m.windows {
+ if isUnder(target, w.dir) {
+ w.changed = true
+ }
+ }
+}
+
+// heldLocked reports whether a hold covers target.
+func (m *Manager) heldLocked(target string) bool {
+ for h := range m.holds {
+ if isUnder(target, h) {
+ return true
+ }
+ }
+ return false
+}
+
+func (m *Manager) countUnderLocked(dir string) int {
n := 0
for target, c := range m.placing {
- if target == dir || strings.HasPrefix(target, prefix) {
+ if isUnder(target, dir) {
n += c
}
}
return n
}
+
+// isUnder reports whether target is dir or below it. Both are clean.
+func isUnder(target, dir string) bool {
+ prefix := dir
+ if !strings.HasSuffix(prefix, "/") {
+ prefix += "/"
+ }
+ return target == dir || strings.HasPrefix(target, prefix)
+}
diff --git a/internal/cgroup/inflight_linux_test.go b/internal/cgroup/inflight_linux_test.go
index f6253fe22..9371132fc 100644
--- a/internal/cgroup/inflight_linux_test.go
+++ b/internal/cgroup/inflight_linux_test.go
@@ -5,6 +5,7 @@ package cgroup
import (
"context"
"errors"
+ "sync"
"testing"
"time"
)
@@ -91,3 +92,94 @@ func TestNilManagerPlacements(t *testing.T) {
t.Fatal(err)
}
}
+
+// readQuiet answers quiet only when nothing under its directory was in
+// flight at any moment of the read: not at its start, not registered or
+// ended during it, not at its end. Placements elsewhere do not count.
+func TestReadQuiet(t *testing.T) {
+ m := &Manager{root: "/r", logf: t.Logf}
+ if !m.readQuiet("/r/h", func() {}) {
+ t.Fatal("nothing in flight read as not quiet")
+ }
+ other := m.beginPlacement("/r/hx/g1")
+ defer other()
+ if !m.readQuiet("/r/h", func() { m.beginPlacement("/r/other")() }) {
+ t.Fatal("placements outside the directory made the read not quiet")
+ }
+ done := m.beginPlacement("/r/h/g1")
+ if m.readQuiet("/r/h", func() {}) {
+ t.Fatal("a placement in flight throughout read as quiet")
+ }
+ if m.readQuiet("/r/h", done) {
+ t.Fatal("a placement that ended during the read read as quiet")
+ }
+ var late func()
+ if m.readQuiet("/r/h", func() { late = m.beginPlacement("/r/h/g2"); late() }) {
+ t.Fatal("a placement that registered and ended during the read read as quiet")
+ }
+ if !m.readQuiet("/r/h", func() {}) {
+ t.Fatal("the read after everything ended is not quiet")
+ }
+ m.placeMu.Lock()
+ open := len(m.windows)
+ m.placeMu.Unlock()
+ if open != 0 {
+ t.Fatalf("%d windows left open", open)
+ }
+ var nilm *Manager
+ ran := false
+ if !nilm.readQuiet("/r", func() { ran = true }) || !ran {
+ t.Fatal("a nil manager's read did not run or was not quiet")
+ }
+}
+
+// holdPlacements stops registrations in and below its directory until it
+// is released, counts the ones already in flight, and leaves others alone.
+func TestHoldPlacementsBlocksRegistrationUntilReleased(t *testing.T) {
+ m := &Manager{root: "/r", logf: t.Logf}
+ before := m.beginPlacement("/r/h/g1")
+ at, under, release := m.holdPlacements("/r/h")
+ if at != 0 || under != 1 {
+ t.Fatalf("hold counted %d at, %d under: want 0, 1", at, under)
+ }
+ // A placement already in flight still ends under the hold.
+ before()
+ // One elsewhere is not held.
+ m.beginPlacement("/r/elsewhere")()
+
+ blocked := make(chan struct{})
+ var once sync.Once
+ orig := placementHeldHook
+ placementHeldHook = func(string) { once.Do(func() { close(blocked) }) }
+ t.Cleanup(func() { placementHeldHook = orig })
+ got := make(chan func(), 1)
+ go func() { got <- m.beginPlacement("/r/h/g2") }()
+ select {
+ case <-blocked:
+ case d := <-got:
+ d()
+ t.Fatal("a placement registered under a hold")
+ case <-time.After(5 * time.Second):
+ t.Fatal("the registration neither registered nor waited")
+ }
+ if n := m.placementsUnder("/r/h"); n != 0 {
+ t.Fatalf("%d placements registered under the hold", n)
+ }
+ release()
+ release() // idempotent
+ select {
+ case d := <-got:
+ if n := m.placementsAt("/r/h/g2"); n != 1 {
+ t.Fatalf("after the release %d placements at g2, want 1", n)
+ }
+ d()
+ case <-time.After(5 * time.Second):
+ t.Fatal("the registration did not go ahead once the hold was released")
+ }
+ var nilm *Manager
+ if a, u, r := nilm.holdPlacements("/r"); a != 0 || u != 0 {
+ t.Fatal("a nil manager counted placements")
+ } else {
+ r()
+ }
+}