vfs/disk: Harden pending-window test against empty-dir race
Empty-dir disk init closes initDone almost immediately, so TestDiskTierSignalMixedPendingReady can observe DiskTierReady=0 then a heartbeat that already saw ready. Register a per-root init hold so the test can keep the pending window open without slowing production. Link: #36
This commit is contained in:
@@ -14,6 +14,7 @@ import (
|
|||||||
"path/filepath"
|
"path/filepath"
|
||||||
"runtime"
|
"runtime"
|
||||||
"s1d3sw1ped/steamcache2/steamcache/metrics"
|
"s1d3sw1ped/steamcache2/steamcache/metrics"
|
||||||
|
"s1d3sw1ped/steamcache2/vfs/disk"
|
||||||
"s1d3sw1ped/steamcache2/vfs/eviction"
|
"s1d3sw1ped/steamcache2/vfs/eviction"
|
||||||
"s1d3sw1ped/steamcache2/vfs/memory"
|
"s1d3sw1ped/steamcache2/vfs/memory"
|
||||||
"s1d3sw1ped/steamcache2/vfs/vfserror"
|
"s1d3sw1ped/steamcache2/vfs/vfserror"
|
||||||
@@ -1139,19 +1140,31 @@ func TestDiskTierSignalMemoryOnly(t *testing.T) {
|
|||||||
// TestDiskTierSignalMixedPendingReady covers mixed mode: DiskTierReady=0 (header
|
// TestDiskTierSignalMixedPendingReady covers mixed mode: DiskTierReady=0 (header
|
||||||
// pending) while the disk attach is in the Size barrier, then DiskTierReady=1
|
// pending) while the disk attach is in the Size barrier, then DiskTierReady=1
|
||||||
// (header ready) after the barrier opens and the attach goroutine sets SetSlow.
|
// (header ready) after the barrier opens and the attach goroutine sets SetSlow.
|
||||||
|
// An init hold keeps the empty-dir attach from finishing between the pending
|
||||||
|
// metric check and the heartbeat, which otherwise races under CI load.
|
||||||
func TestDiskTierSignalMixedPendingReady(t *testing.T) {
|
func TestDiskTierSignalMixedPendingReady(t *testing.T) {
|
||||||
td := t.TempDir()
|
td := t.TempDir()
|
||||||
diskPath := filepath.Join(td, "disk")
|
diskPath := filepath.Join(td, "disk")
|
||||||
if err := os.MkdirAll(diskPath, 0755); err != nil {
|
if err := os.MkdirAll(diskPath, 0755); err != nil {
|
||||||
t.Fatal(err)
|
t.Fatal(err)
|
||||||
}
|
}
|
||||||
|
hold := make(chan struct{})
|
||||||
|
var holdOnce sync.Once
|
||||||
|
closeHold := func() { holdOnce.Do(func() { close(hold) }) }
|
||||||
|
disk.RegisterInitHold(diskPath, hold)
|
||||||
|
t.Cleanup(func() {
|
||||||
|
closeHold()
|
||||||
|
disk.ClearInitHold(diskPath)
|
||||||
|
})
|
||||||
|
|
||||||
sc, err := New("127.0.0.1:0", "1MB", "10MB", diskPath, "", "lru", "lru", 10, 1, "0", nil)
|
sc, err := New("127.0.0.1:0", "1MB", "10MB", diskPath, "", "lru", "lru", 10, 1, "0", nil)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
t.Fatalf("New mixed: %v", err)
|
t.Fatalf("New mixed: %v", err)
|
||||||
}
|
}
|
||||||
t.Cleanup(func() { sc.Shutdown() })
|
t.Cleanup(func() { sc.Shutdown() })
|
||||||
|
t.Cleanup(closeHold) // before Shutdown: attach is blocked in Size() until the hold closes
|
||||||
|
|
||||||
// Immediately in the pending window
|
// Pending window is held open until closeHold; attach cannot finish.
|
||||||
if got := sc.GetMetrics().DiskTierReady; got != 0 {
|
if got := sc.GetMetrics().DiskTierReady; got != 0 {
|
||||||
t.Errorf("immediate DiskTierReady=%d, want 0 (pending)", got)
|
t.Errorf("immediate DiskTierReady=%d, want 0 (pending)", got)
|
||||||
}
|
}
|
||||||
@@ -1162,7 +1175,7 @@ func TestDiskTierSignalMixedPendingReady(t *testing.T) {
|
|||||||
t.Errorf("heartbeat header=%q, want pending", got)
|
t.Errorf("heartbeat header=%q, want pending", got)
|
||||||
}
|
}
|
||||||
|
|
||||||
// Wait the barrier, then retry until the attach goroutine flips the flag
|
closeHold()
|
||||||
_ = sc.disk.Size()
|
_ = sc.disk.Size()
|
||||||
deadline := time.Now().Add(2 * time.Second)
|
deadline := time.Now().Add(2 * time.Second)
|
||||||
for time.Now().Before(deadline) {
|
for time.Now().Before(deadline) {
|
||||||
|
|||||||
+39
-2
@@ -45,6 +45,25 @@ type DiskFS struct {
|
|||||||
initCloseOnce sync.Once
|
initCloseOnce sync.Once
|
||||||
startupEvict func(vfs.VFS, uint) uint // passed to New (via gc.GetGCAlgorithm); invoked as last step of bg init if over cap (no post-ctor race)
|
startupEvict func(vfs.VFS, uint) uint // passed to New (via gc.GetGCAlgorithm); invoked as last step of bg init if over cap (no post-ctor race)
|
||||||
metrics *metrics.Metrics
|
metrics *metrics.Metrics
|
||||||
|
// initHold, if non-nil, is received on before closing initDone (test pending-window hold).
|
||||||
|
initHold <-chan struct{}
|
||||||
|
}
|
||||||
|
|
||||||
|
// initHolds is a per-root registry of optional init holds (root path -> <-chan struct{}).
|
||||||
|
// Tests call RegisterInitHold before New so that instance copies the channel and waits
|
||||||
|
// before closing initDone; production never registers, so init is unchanged.
|
||||||
|
var initHolds sync.Map
|
||||||
|
|
||||||
|
// RegisterInitHold registers a channel that DiskFS.New for this root copies onto that
|
||||||
|
// instance. calculateSizeAndPopulateIndex receives on it before closing initDone, so
|
||||||
|
// tests can observe the pending-attach window. Other DiskFS roots are unaffected.
|
||||||
|
func RegisterInitHold(root string, ch <-chan struct{}) {
|
||||||
|
initHolds.Store(root, ch)
|
||||||
|
}
|
||||||
|
|
||||||
|
// ClearInitHold removes a previously registered hold for root.
|
||||||
|
func ClearInitHold(root string) {
|
||||||
|
initHolds.Delete(root)
|
||||||
}
|
}
|
||||||
|
|
||||||
// shardPath converts a Steam cache key to a sharded directory path to reduce inode pressure
|
// shardPath converts a Steam cache key to a sharded directory path to reduce inode pressure
|
||||||
@@ -129,6 +148,12 @@ func New(root string, capacity int64, evict func(vfs.VFS, uint) uint) (*DiskFS,
|
|||||||
startupEvict: evict,
|
startupEvict: evict,
|
||||||
}
|
}
|
||||||
|
|
||||||
|
if v, ok := initHolds.Load(root); ok {
|
||||||
|
if ch, ok := v.(<-chan struct{}); ok {
|
||||||
|
d.initHold = ch
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
d.initDone = make(chan struct{})
|
d.initDone = make(chan struct{})
|
||||||
// Launch heavy population asynchronously so New returns fast (scans millions of files without blocking ctor or using O(N) temp RAM).
|
// Launch heavy population asynchronously so New returns fast (scans millions of files without blocking ctor or using O(N) temp RAM).
|
||||||
// The initDone barrier ensures first Size() and subsequent ops (including late tier attach) see fully populated + post-eviction state.
|
// The initDone barrier ensures first Size() and subsequent ops (including late tier attach) see fully populated + post-eviction state.
|
||||||
@@ -152,7 +177,7 @@ func (d *DiskFS) calculateSizeAndPopulateIndex() {
|
|||||||
if r := recover(); r != nil {
|
if r := recover(); r != nil {
|
||||||
logger.Logger.Error().Interface("recovered_panic", r).Msg("calculateSizeAndPopulateIndex panicked; ensuring initDone closed to unblock Size waiters and prevent hang")
|
logger.Logger.Error().Interface("recovered_panic", r).Msg("calculateSizeAndPopulateIndex panicked; ensuring initDone closed to unblock Size waiters and prevent hang")
|
||||||
}
|
}
|
||||||
d.initCloseOnce.Do(func() { close(d.initDone) })
|
d.closeInitDone()
|
||||||
}()
|
}()
|
||||||
|
|
||||||
tstart := time.Now()
|
tstart := time.Now()
|
||||||
@@ -243,7 +268,19 @@ func (d *DiskFS) calculateSizeAndPopulateIndex() {
|
|||||||
|
|
||||||
// Signal readiness: Size() and callers (late tier attach + Evict*) now see correct populated + post-eviction state.
|
// Signal readiness: Size() and callers (late tier attach + Evict*) now see correct populated + post-eviction state.
|
||||||
// Use Once (recover path also uses it) to guarantee exactly one close even under panic.
|
// Use Once (recover path also uses it) to guarantee exactly one close even under panic.
|
||||||
d.initCloseOnce.Do(func() { close(d.initDone) })
|
d.closeInitDone()
|
||||||
|
}
|
||||||
|
|
||||||
|
// closeInitDone receives on a copied test hold (if any) then closes initDone once.
|
||||||
|
// Both the normal end of calculateSizeAndPopulateIndex and the panic-recovery defer
|
||||||
|
// call this so Size() waiters unblock in either path.
|
||||||
|
func (d *DiskFS) closeInitDone() {
|
||||||
|
d.initCloseOnce.Do(func() {
|
||||||
|
if d.initHold != nil {
|
||||||
|
<-d.initHold
|
||||||
|
}
|
||||||
|
close(d.initDone)
|
||||||
|
})
|
||||||
}
|
}
|
||||||
|
|
||||||
// insertBatch populates info/LRU under lock for a bounded batch (follows maxEvictBatch pattern for short critical sections).
|
// insertBatch populates info/LRU under lock for a bounded batch (follows maxEvictBatch pattern for short critical sections).
|
||||||
|
|||||||
@@ -662,3 +662,57 @@ func TestDiskFS_NewMkdirError(t *testing.T) {
|
|||||||
t.Errorf("expected mkdir failure error for file-as-dir, got: %v", err)
|
t.Errorf("expected mkdir failure error for file-as-dir, got: %v", err)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// TestDiskFS_InitHoldBlocksOnlyRegisteredRoot covers the per-root init hold:
|
||||||
|
// Size() stays blocked while the hold is open, and a DiskFS on a different root
|
||||||
|
// does not wait on that hold.
|
||||||
|
func TestDiskFS_InitHoldBlocksOnlyRegisteredRoot(t *testing.T) {
|
||||||
|
td := t.TempDir()
|
||||||
|
hold := make(chan struct{})
|
||||||
|
var holdOnce sync.Once
|
||||||
|
closeHold := func() { holdOnce.Do(func() { close(hold) }) }
|
||||||
|
RegisterInitHold(td, hold)
|
||||||
|
t.Cleanup(func() {
|
||||||
|
closeHold()
|
||||||
|
ClearInitHold(td)
|
||||||
|
})
|
||||||
|
|
||||||
|
d, err := New(td, 10*1024*1024, nil)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
|
||||||
|
blocked := make(chan struct{})
|
||||||
|
go func() {
|
||||||
|
_ = d.Size()
|
||||||
|
close(blocked)
|
||||||
|
}()
|
||||||
|
select {
|
||||||
|
case <-blocked:
|
||||||
|
t.Fatal("Size returned while init hold still open")
|
||||||
|
case <-time.After(50 * time.Millisecond):
|
||||||
|
}
|
||||||
|
|
||||||
|
td2 := t.TempDir()
|
||||||
|
d2, err := New(td2, 10*1024*1024, nil)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
other := make(chan struct{})
|
||||||
|
go func() {
|
||||||
|
_ = d2.Size()
|
||||||
|
close(other)
|
||||||
|
}()
|
||||||
|
select {
|
||||||
|
case <-other:
|
||||||
|
case <-time.After(2 * time.Second):
|
||||||
|
t.Fatal("unrelated DiskFS Size hung; init hold leaked across roots")
|
||||||
|
}
|
||||||
|
|
||||||
|
closeHold()
|
||||||
|
select {
|
||||||
|
case <-blocked:
|
||||||
|
case <-time.After(2 * time.Second):
|
||||||
|
t.Fatal("Size did not return after init hold released")
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user