perf(local): shared stat scheduler replacing per-entry timeout machinery

Replace the per-entry withTimeout/withTimeoutRetry goroutine+channel
machinery and producer-consumer channel pipeline with a shared per-Fs
stat scheduler using microbatch lease-gated publication.

Key changes:
- Sequential ReadDir(1024) → ProcessBatch → materialize loop
- Batch-scoped raw dirfd via Fd(), openat(readfd, ".") for stat fds
- StatScheduler with watchdog, retry queues, replacement workers
- Removed SyscallConn.Control per-entry overhead, errReopen, old
  pipeline, withTimeout, statPool, cloneDirForListStat
- Scheduler optimizations: pooled jobs, scratch reuse, in-place
  compaction, worker ID recycling, pre-sized result slices

Signed-off-by: Anagh Kumar Baranwal <6824881+darthShadow@users.noreply.github.com>

Author: darthShadow

backend/local/list.go

@@ -1,11 +1,11 @@
 //go:build linux && amd64
 
-// This file implements the linux/amd64 fast-stat path: cloneDirForListStat
-// dup's the directory fd via F_DUPFD_CLOEXEC, and statDirEntry calls
-// fstatat(dirfd, name, AT_SYMLINK_NOFOLLOW) to stat each entry without
-// constructing an absolute path. On other platforms the build tag excludes
-// this file and list_fstatat_other.go provides stubs that fall back to
-// entry.Info().
+// This file implements the linux/amd64 fast-stat path used by the sequential
+// listing loop. openDirAtReadFD opens "." relative to the listing read fd to
+// create a batch-scoped stat handle, listStatDirFD snapshots that handle's raw
+// dirfd once per batch, and statDirEntry uses fstatat(dirfd, name,
+// AT_SYMLINK_NOFOLLOW) without constructing absolute paths. Other platforms use
+// list_fstatat_other.go and fall back to entry.Info().
 
 package local
 
@@ -113,40 +113,48 @@ func fstatatNoFollow(dirfd int, name string) (os.FileInfo, error) {
 	}, nil
 }
 
-func cloneDirForListStat(fd *os.File) (*os.File, error) {
-	sc, err := fd.SyscallConn()
-	if err != nil {
-		return nil, err
+// openDirAtReadFD opens "." relative to the active listing fd so one ReadDir
+// batch gets its own directory handle for fstatat work.
+func openDirAtReadFD(fd *os.File) (*os.File, error) {
+	if fd == nil {
+		return nil, os.ErrClosed
 	}
-	var dupFD int
-	var dupErr error
-	ctrlErr := sc.Control(func(rawfd uintptr) {
-		dupFD, dupErr = unix.FcntlInt(rawfd, unix.F_DUPFD_CLOEXEC, 0)
-	})
-	if ctrlErr != nil {
-		return nil, ctrlErr
+
+	rawfd := fd.Fd()
+	if rawfd == ^uintptr(0) {
+		return nil, os.ErrClosed
 	}
-	if dupErr != nil {
-		return nil, dupErr
+
+	for {
+		statFD, err := unix.Openat(int(rawfd), ".", unix.O_RDONLY|unix.O_DIRECTORY|unix.O_CLOEXEC, 0)
+		if err == syscall.EINTR {
+			continue
+		}
+		if err != nil {
+			return nil, err
+		}
+		return os.NewFile(uintptr(statFD), fd.Name()), nil
 	}
-	return os.NewFile(uintptr(dupFD), fd.Name()), nil
 }
 
-func statDirEntry(entry cachedDirEntry) (os.FileInfo, error) {
-	if entry.statDir == nil {
-		return entry.Info()
+// listStatDirFD snapshots the raw dirfd from the batch-owned stat handle so the
+// batch can reuse it for every entry without reacquiring it per stat call.
+func listStatDirFD(fd *os.File) (int, bool) {
+	if fd == nil {
+		return 0, false
 	}
-	sc, err := entry.statDir.SyscallConn()
-	if err != nil {
-		return nil, err
+	rawfd := fd.Fd()
+	if rawfd == ^uintptr(0) {
+		return 0, false
 	}
-	var fi os.FileInfo
-	var fiErr error
-	ctrlErr := sc.Control(func(rawfd uintptr) {
-		fi, fiErr = fstatatNoFollow(int(rawfd), entry.Name())
-	})
-	if ctrlErr != nil {
-		return nil, ctrlErr
+	return int(rawfd), true
+}
+
+// statDirEntry stats one cached entry through the batch-scoped dirfd when
+// available, otherwise it falls back to the DirEntry.Info path.
+func statDirEntry(entry *cachedDirEntry) (os.FileInfo, error) {
+	if entry.useStatFD {
+		return fstatatNoFollow(entry.statDirFD, entry.Name())
 	}
-	return fi, fiErr
+	return entry.Info()
 }

backend/local/list_fstatat_linux.go

@@ -135,7 +135,7 @@ func TestFstatatNoFollow_ENOENT(t *testing.T) {
 	}()
 
 	fi, err := fstatatNoFollow(int(fd.Fd()), "missing.txt")
-	require.Nil(t, fi)
+	assert.Nil(t, fi)
 	require.True(t, os.IsNotExist(err), "expected not-exist error, got %v", err)
 }
 
@@ -172,7 +172,30 @@ func TestList_FstatatPath_TranslateSymlinksFilterParity(t *testing.T) {
 	assert.Equal(t, []string{"alias.txt" + fs.LinkSuffix}, dirEntryRemotes(entries))
 }
 
-func TestStatDirEntry_ClosedStatDirReturnsErrClosed(t *testing.T) {
+func TestListStatDirFD_ClosedFileReturnsFalse(t *testing.T) {
+	root := t.TempDir()
+	fd, err := os.Open(root)
+	require.NoError(t, err)
+
+	require.NoError(t, fd.Close())
+
+	rawfd, ok := listStatDirFD(fd)
+	assert.Zero(t, rawfd)
+	assert.False(t, ok)
+}
+
+func TestOpenDirAtReadFD_ClosedFileReturnsErrClosed(t *testing.T) {
+	root := t.TempDir()
+	fd, err := os.Open(root)
+	require.NoError(t, err)
+	require.NoError(t, fd.Close())
+
+	statDir, err := openDirAtReadFD(fd)
+	require.Nil(t, statDir)
+	require.True(t, errors.Is(err, os.ErrClosed), "expected os.ErrClosed, got %v", err)
+}
+
+func TestStatDirEntry_UsesBatchScopedDirFD(t *testing.T) {
 	root := t.TempDir()
 	require.NoError(t, os.WriteFile(filepath.Join(root, "file.txt"), []byte("hello"), 0o600))
 
@@ -189,36 +212,28 @@ func TestStatDirEntry_ClosedStatDirReturnsErrClosed(t *testing.T) {
 	}
 	require.NotNil(t, entry)
 
-	fd, err := os.Open(root)
+	statDir, err := os.Open(root)
 	require.NoError(t, err)
 	defer func() {
-		require.NoError(t, fd.Close())
+		require.NoError(t, statDir.Close())
 	}()
 
-	statDir, err := cloneDirForListStat(fd)
-	require.NoError(t, err)
-	require.NotNil(t, statDir)
+	statDirFD, ok := listStatDirFD(statDir)
+	require.True(t, ok)
 
-	closedFD := int(statDir.Fd())
-	require.NoError(t, statDir.Close())
-
-	var reused *os.File
-	for i := 0; i < 32; i++ {
-		probe, err := os.Open(root)
-		require.NoError(t, err)
-		if int(probe.Fd()) == closedFD {
-			reused = probe
-			break
-		}
-		require.NoError(t, probe.Close())
-	}
-	if reused != nil {
-		defer func() {
-			require.NoError(t, reused.Close())
-		}()
+	cachedEntry := cachedDirEntry{
+		DirEntry:  entry,
+		statDirFD: statDirFD,
+		useStatFD: true,
 	}
+	fi, err := statDirEntry(&cachedEntry)
+	require.NoError(t, err)
 
-	fi, err := statDirEntry(cachedDirEntry{DirEntry: entry, statDir: statDir})
-	require.Nil(t, fi)
-	require.True(t, errors.Is(err, os.ErrClosed), "expected os.ErrClosed, got %v", err)
+	want, err := os.Lstat(filepath.Join(root, "file.txt"))
+	require.NoError(t, err)
+
+	assert.Equal(t, want.Name(), fi.Name())
+	assert.Equal(t, want.Size(), fi.Size())
+	assert.Equal(t, want.Mode(), fi.Mode())
+	assert.True(t, fi.ModTime().Equal(want.ModTime()), "expected modtime %v, got %v", want.ModTime(), fi.ModTime())
 }

backend/local/list_fstatat_linux_test.go

@@ -1,18 +1,27 @@
 //go:build !(linux && amd64)
 
-// This file is the fallback for all platforms other than linux/amd64.
-// It provides no-op stubs for cloneDirForListStat and statDirEntry so that
-// list.go compiles everywhere; statDirEntry simply delegates to entry.Info()
-// which uses the DirEntry cached from ReadDir without an extra syscall.
+// This file is the fallback for all platforms other than linux/amd64. The
+// sequential listing loop still reads one ReadDir batch at a time, but there is
+// no batch-scoped fstatat helper on these targets, so statDirEntry delegates to
+// entry.Info().
 
 package local
 
 import "os"
 
-func cloneDirForListStat(fd *os.File) (*os.File, error) {
+// openDirAtReadFD is a no-op stub: this platform has no batch-scoped dirfd fast
+// path, so it returns a nil handle. The caller detects nil and falls back to
+// entry.Info().
+func openDirAtReadFD(fd *os.File) (*os.File, error) {
 	return nil, nil
 }
 
-func statDirEntry(entry cachedDirEntry) (os.FileInfo, error) {
+// listStatDirFD reports that there is no raw dirfd to reuse on this platform.
+func listStatDirFD(fd *os.File) (int, bool) {
+	return 0, false
+}
+
+// statDirEntry falls back to the cached DirEntry.Info implementation.
+func statDirEntry(entry *cachedDirEntry) (os.FileInfo, error) {
 	return entry.Info()
 }

backend/local/list_fstatat_other.go

@@ -11,57 +11,60 @@ import (
 	"github.com/stretchr/testify/require"
 )
 
-func TestListFileInfos_PipelineCancellationBufferedBatch(t *testing.T) {
+func TestListFileInfos_CancellationLargeDirectory(t *testing.T) {
 	f, root := newTestLocalFs(t)
 	for i := 0; i < 2050; i++ {
 		writeTestFile(t, root, fmt.Sprintf("item-%04d.txt", i))
 	}
 
-	baseCtx := newAsyncTestContext(4)
-	ctx, cancel := context.WithCancel(baseCtx)
-	defer cancel()
+	ctx, cancel := context.WithCancel(context.Background())
+	f.statScheduler.Close()
+	f.statScheduler = newStatScheduler(f, statSchedulerOptions{
+		Workers:          1,
+		MaxWorkers:       1,
+		QueueDepth:       16,
+		LeaseTimeout:     time.Second,
+		RetryBackoff:     time.Millisecond,
+		WatchdogInterval: time.Second,
+		WarnAfter:        time.Second,
+		ReplaceAfter:     time.Second,
+	})
+	t.Cleanup(f.statScheduler.Close)
 
 	fd, err := os.Open(root)
 	require.NoError(t, err)
 
-	openDir := func() (*os.File, error) {
-		return os.Open(root)
-	}
-
 	started := make(chan struct{}, 1)
-	statFunc := func(entry os.DirEntry) os.FileInfo {
-		select {
-		case started <- struct{}{}:
-		default:
-		}
-		<-ctx.Done()
-		return nil
-	}
-
 	done := make(chan error, 1)
 	go func() {
-		_, err := f.listFileInfos(ctx, fd, openDir, nil, statFunc, nil)
+		_, err := f.listFileInfos(ctx, fd, nil, nil, func(entry os.DirEntry) os.FileInfo {
+			select {
+			case started <- struct{}{}:
+			default:
+			}
+			<-ctx.Done()
+			return nil
+		})
 		done <- err
 	}()
 
 	select {
 	case <-started:
 	case <-time.After(2 * time.Second):
-		t.Fatal("timed out waiting for pipeline stat work to start")
+		t.Fatal("timed out waiting for scheduled stat work to start")
 	}
 
-	time.Sleep(50 * time.Millisecond)
 	cancel()
 
 	select {
 	case err := <-done:
 		require.True(t, errors.Is(err, context.Canceled), "expected context cancellation, got %v", err)
 	case <-time.After(2 * time.Second):
-		t.Fatal("timed out waiting for buffered pipeline cancellation")
+		t.Fatal("timed out waiting for large-directory cancellation")
 	}
 }
 
-func TestStatReadBatch_ClosesStatDirOnPanic(t *testing.T) {
+func TestListController_ProcessBatch_ClosesStatDirOnPanic(t *testing.T) {
 	dir := t.TempDir()
 	statDir, err := os.Open(dir)
 	require.NoError(t, err)
@@ -70,22 +73,24 @@ func TestStatReadBatch_ClosesStatDirOnPanic(t *testing.T) {
 		entries: fakeEntries("panic.txt"),
 		statDir: statDir,
 	}
+	controller := newListController(nil, nil, listControllerOptions{})
+	defer controller.Close()
 
 	var recovered any
 	func() {
 		defer func() {
 			recovered = recover()
 		}()
-		_, _, _ = (&Fs{}).statReadBatch(context.Background(), &batch, nil, func(os.DirEntry) (cachedDirEntry, bool) {
+		_, _, _ = controller.ProcessBatch(context.Background(), &batch, nil, func(os.DirEntry) (cachedDirEntry, bool) {
 			panic("boom")
-		}, nil, nil)
+		}, func(entry *cachedDirEntry) (os.FileInfo, error) {
+			t.Fatalf("unexpected stat call for %q", entry.Name())
+			return nil, nil
+		})
 	}()
 
 	require.NotNil(t, recovered)
 	require.Nil(t, batch.statDir)
-
-	require.Eventually(t, func() bool {
-		_, err := statDir.Stat()
-		return errors.Is(err, os.ErrClosed)
-	}, time.Second, 10*time.Millisecond, "expected statDir to be closed after panic cleanup")
+	_, err = statDir.Stat()
+	require.True(t, errors.Is(err, os.ErrClosed), "expected statDir to be closed after panic cleanup, got %v", err)
 }