diff --git a/packages/orchestrator/pkg/cfg/model.go b/packages/orchestrator/pkg/cfg/model.go index 4067cddc42..4ca62108b2 100644 --- a/packages/orchestrator/pkg/cfg/model.go +++ b/packages/orchestrator/pkg/cfg/model.go @@ -149,11 +149,45 @@ type Config struct { NBDPoolSize int `env:"NBD_POOL_SIZE" envDefault:"64"` Services []string `env:"ORCHESTRATOR_SERVICES" envDefault:"orchestrator"` PersistentVolumeMounts map[string]string `env:"PERSISTENT_VOLUME_MOUNTS"` + // VolumeReadDirConcurrency is how the parallelism of a volume directory + // listing's per-entry stat calls scales with the number of entries: the + // stats are serial below Threshold, then run on Base workers plus one + // more per Step additional entries, capped at Max. Every stat is a round + // trip on the volume's storage, so this bounds the listing's cost there. + VolumeReadDirConcurrency ReadDirConcurrencyConfig // TLS is the mutual TLS settings of the shared port and the peer hop. TLS TLSConfig } +// ReadDirConcurrencyConfig holds the VOLUME_READ_DIR_CONCURRENCY_* settings +// that scale volume directory listings with the entry count; see the +// VolumeReadDirConcurrency field of Config. +type ReadDirConcurrencyConfig struct { + Threshold int `env:"VOLUME_READ_DIR_CONCURRENCY_THRESHOLD" envDefault:"1000"` + Base int `env:"VOLUME_READ_DIR_CONCURRENCY_BASE" envDefault:"8"` + Step int `env:"VOLUME_READ_DIR_CONCURRENCY_STEP" envDefault:"5000"` + Max int `env:"VOLUME_READ_DIR_CONCURRENCY_MAX" envDefault:"16"` +} + +// Validate reports scaling settings a listing could not run with. +func (c ReadDirConcurrencyConfig) Validate() error { + if c.Threshold < 0 { + return fmt.Errorf("VOLUME_READ_DIR_CONCURRENCY_THRESHOLD must be at least 0, got %d", c.Threshold) + } + if c.Base < 1 { + return fmt.Errorf("VOLUME_READ_DIR_CONCURRENCY_BASE must be at least 1, got %d", c.Base) + } + if c.Step < 1 { + return fmt.Errorf("VOLUME_READ_DIR_CONCURRENCY_STEP must be at least 1, got %d", c.Step) + } + if c.Max < c.Base { + return fmt.Errorf("VOLUME_READ_DIR_CONCURRENCY_MAX must be at least VOLUME_READ_DIR_CONCURRENCY_BASE, got %d < %d", c.Max, c.Base) + } + + return nil +} + // AdditionalClickhouseEndpoints returns the non-blank entries from // CLICKHOUSE_CONNECTION_STRINGS that are *in addition to* the singular // CLICKHOUSE_CONNECTION_STRING. Order is preserved; first occurrence wins on @@ -218,6 +252,10 @@ func Parse() (Config, error) { return config, err } + if err := config.VolumeReadDirConcurrency.Validate(); err != nil { + return config, err + } + if config.PersistentVolumeMounts != nil { for name, path := range config.PersistentVolumeMounts { path = filepath.Clean(path) diff --git a/packages/orchestrator/pkg/cfg/model_test.go b/packages/orchestrator/pkg/cfg/model_test.go index 029845d460..d510f92e25 100644 --- a/packages/orchestrator/pkg/cfg/model_test.go +++ b/packages/orchestrator/pkg/cfg/model_test.go @@ -71,6 +71,49 @@ func TestParse(t *testing.T) { require.NoError(t, err) assert.True(t, config.DisableStartupReclaim) }) + + t.Run("volume read dir concurrency gets defaults", func(t *testing.T) { + config, err := Parse() + require.NoError(t, err) + + assert.Equal(t, ReadDirConcurrencyConfig{Threshold: 1000, Base: 8, Step: 5000, Max: 16}, config.VolumeReadDirConcurrency) + }) + + t.Run("volume read dir concurrency can be overridden", func(t *testing.T) { + t.Setenv("VOLUME_READ_DIR_CONCURRENCY_THRESHOLD", "100") + t.Setenv("VOLUME_READ_DIR_CONCURRENCY_BASE", "2") + t.Setenv("VOLUME_READ_DIR_CONCURRENCY_STEP", "50") + t.Setenv("VOLUME_READ_DIR_CONCURRENCY_MAX", "6") + + config, err := Parse() + require.NoError(t, err) + + assert.Equal(t, ReadDirConcurrencyConfig{Threshold: 100, Base: 2, Step: 50, Max: 6}, config.VolumeReadDirConcurrency) + }) + + t.Run("volume read dir concurrency rejects unusable scaling", func(t *testing.T) { + tests := []struct { + name string + envs map[string]string + }{ + {"negative threshold", map[string]string{"VOLUME_READ_DIR_CONCURRENCY_THRESHOLD": "-1"}}, + {"zero base", map[string]string{"VOLUME_READ_DIR_CONCURRENCY_BASE": "0"}}, + {"zero step", map[string]string{"VOLUME_READ_DIR_CONCURRENCY_STEP": "0"}}, + {"max below base", map[string]string{"VOLUME_READ_DIR_CONCURRENCY_BASE": "4", "VOLUME_READ_DIR_CONCURRENCY_MAX": "3"}}, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + for key, value := range test.envs { + t.Setenv(key, value) + } + + _, err := Parse() + require.Error(t, err) + assert.ErrorContains(t, err, "VOLUME_READ_DIR_CONCURRENCY") + }) + } + }) } func TestAdditionalClickhouseEndpoints(t *testing.T) { diff --git a/packages/orchestrator/pkg/chrooted/builder.go b/packages/orchestrator/pkg/chrooted/builder.go index 80f7300873..5dc4fc79f1 100644 --- a/packages/orchestrator/pkg/chrooted/builder.go +++ b/packages/orchestrator/pkg/chrooted/builder.go @@ -28,7 +28,15 @@ func (b *Builder) Chroot(volumeType string, teamID, volumeID uuid.UUID) (*Chroot return nil, err } - fs, err := Chroot(fullPath, WithMetadata("volume-id", volumeID.String())) + fs, err := Chroot(fullPath, + WithMetadata("volume-id", volumeID.String()), + WithReadDirScaling(ReadDirScaling{ + Threshold: b.config.VolumeReadDirConcurrency.Threshold, + Base: b.config.VolumeReadDirConcurrency.Base, + Step: b.config.VolumeReadDirConcurrency.Step, + Max: b.config.VolumeReadDirConcurrency.Max, + }), + ) if err != nil { return nil, err } diff --git a/packages/orchestrator/pkg/chrooted/chroot.go b/packages/orchestrator/pkg/chrooted/chroot.go index 77a6ce32fb..f8bc5edbde 100644 --- a/packages/orchestrator/pkg/chrooted/chroot.go +++ b/packages/orchestrator/pkg/chrooted/chroot.go @@ -45,6 +45,10 @@ type Chrooted struct { // Waiters never initiate or retry that close. closeDone chan struct{} closeErr error + + // readDirScaling is how the parallelism of ReadDir's per-entry stat + // calls scales with the number of entries; see WithReadDirScaling. + readDirScaling ReadDirScaling } type Option func(*Chrooted) @@ -59,6 +63,14 @@ func WithMetadata(key, value string) Option { } } +// WithReadDirScaling sets how the parallelism of ReadDir's per-entry stat +// calls scales with the number of entries listed. +func WithReadDirScaling(scaling ReadDirScaling) Option { + return func(fs *Chrooted) { + fs.readDirScaling = scaling + } +} + var ( // ErrClosed is returned by operations on a Chrooted after Close. ErrClosed = errors.New("chrooted filesystem is closed") @@ -124,9 +136,10 @@ func Chroot(source string, opts ...Option) (*Chrooted, error) { } fs := &Chrooted{ - ActualRoot: source, - rootFD: rootFD, - closeDone: make(chan struct{}), + ActualRoot: source, + rootFD: rootFD, + closeDone: make(chan struct{}), + readDirScaling: defaultReadDirScaling, } for _, opt := range opts { diff --git a/packages/orchestrator/pkg/chrooted/fs.go b/packages/orchestrator/pkg/chrooted/fs.go index 282c398cf2..9a90662948 100644 --- a/packages/orchestrator/pkg/chrooted/fs.go +++ b/packages/orchestrator/pkg/chrooted/fs.go @@ -14,6 +14,7 @@ import ( "strings" "syscall" + "golang.org/x/sync/errgroup" "golang.org/x/sys/unix" "github.com/e2b-dev/infra/packages/shared/pkg/filesystem" @@ -408,9 +409,49 @@ func (fs *Chrooted) TempFile(dir, prefix string) (*os.File, error) { return nil, &os.PathError{Op: "createtemp", Path: path.Join(dir, prefix+"*"), Err: os.ErrExist} } +// ReadDirScaling says how many per-entry stat calls ReadDir issues in +// parallel as the listed directory grows: the stats are serial below +// Threshold, then run on Base workers plus one more per Step additional +// entries, capped at Max. Each stat is a single fstatat through the listing's +// directory descriptor, so the calls are independent of each other and safe +// to run concurrently. Workers is total: the zero value and degenerate +// fields yield 1, never a panic. +type ReadDirScaling struct { + Threshold int + Base int + Step int + Max int +} + +// Workers returns how many of a listing's entries may be statted in parallel. +func (s ReadDirScaling) Workers(entries int) int { + if entries < s.Threshold { + return 1 + } + + workers := s.Base + if s.Step > 0 { + workers += (entries - s.Threshold) / s.Step + } + if s.Max > 0 && workers > s.Max { + workers = s.Max + } + + return max(workers, 1) +} + +// defaultReadDirScaling is the scaling a Chroot gets unless an option +// overrides it; production values come from the orchestrator's config. +var defaultReadDirScaling = ReadDirScaling{Threshold: 1000, Base: 8, Step: 5000, Max: 16} + // ReadDir lists the directory at dirPath sorted by name, as os.ReadDir does. // Each entry carries its own attributes, not those of a symlink's target. An // entry removed while the directory is being read is left out. +// +// The per-entry stats run in parallel once the directory is large enough, +// scaled by the fs's ReadDirScaling: the tree is usually an NFS volume, where +// every stat is a network round trip. Smaller listings stay serial on the +// calling goroutine — a worker pool would cost more than the stats. func (fs *Chrooted) ReadDir(dirPath string) ([]os.FileInfo, error) { release, err := fs.acquire() if err != nil { @@ -432,21 +473,52 @@ func (fs *Chrooted) ReadDir(dirPath string) ([]os.FileInfo, error) { } slices.Sort(names) - infos := make([]os.FileInfo, 0, len(names)) - for _, name := range names { - info, err := lstatAt(fd, name) - if err != nil { - if errors.Is(err, unix.ENOENT) { - continue + infos := make([]os.FileInfo, len(names)) + statEntry := func(index int) error { + info, statErr := lstatAt(fd, names[index]) + if statErr != nil { + if errors.Is(statErr, unix.ENOENT) { + // The entry was removed while the directory was being read. + return nil } - return nil, pathError("readdir", path.Join(dirPath, name), err) + return pathError("readdir", path.Join(dirPath, names[index]), statErr) } - infos = append(infos, info) + infos[index] = info + + return nil + } + + if workers := fs.readDirScaling.Workers(len(names)); workers > 1 { + group := new(errgroup.Group) + group.SetLimit(workers) + + for index := range names { + group.Go(func() error { + return statEntry(index) + }) + } + + if err := group.Wait(); err != nil { + return nil, err + } + } else { + for index := range names { + if err := statEntry(index); err != nil { + return nil, err + } + } + } + + listed := make([]os.FileInfo, 0, len(infos)) + for _, info := range infos { + if info != nil { + listed = append(listed, info) + } } - return infos, nil + return listed, nil } func (fs *Chrooted) Mkdir(filename string, perm os.FileMode) error { diff --git a/packages/orchestrator/pkg/chrooted/fs_test.go b/packages/orchestrator/pkg/chrooted/fs_test.go new file mode 100644 index 0000000000..05768f21eb --- /dev/null +++ b/packages/orchestrator/pkg/chrooted/fs_test.go @@ -0,0 +1,107 @@ +//go:build linux + +package chrooted + +import ( + "fmt" + "os" + "path/filepath" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestReadDirScalingWorkers(t *testing.T) { + t.Parallel() + + tests := []struct { + name string + scaling ReadDirScaling + entries int + want int + }{ + // The default scaling: serial below 1000 entries, 8 workers from + // there, one more per 5000 additional entries, capped at 16. + {"default: empty directory", defaultReadDirScaling, 0, 1}, + {"default: below the threshold", defaultReadDirScaling, 999, 1}, + {"default: at the threshold", defaultReadDirScaling, 1000, 8}, + {"default: one below the first step", defaultReadDirScaling, 5999, 8}, + {"default: at the first step", defaultReadDirScaling, 6000, 9}, + {"default: at the second step", defaultReadDirScaling, 11000, 10}, + {"default: one below the cap", defaultReadDirScaling, 40999, 15}, + {"default: at the cap", defaultReadDirScaling, 41000, 16}, + {"default: far above the cap", defaultReadDirScaling, 1000000, 16}, + + // The zero value is serial rather than explosive. + {"zero value", ReadDirScaling{}, 1000000, 1}, + + // A step of 0 adds no workers rather than dividing by zero. + {"zero step", ReadDirScaling{Threshold: 1, Base: 4, Step: 0, Max: 16}, 1000000, 4}, + + // A max below the base caps at the max. + {"max below the base", ReadDirScaling{Threshold: 1, Base: 8, Step: 1, Max: 2}, 1000000, 2}, + + // A base of 1 stays serial until a step adds workers. + {"base 1 grows with steps", ReadDirScaling{Threshold: 1, Base: 1, Step: 1, Max: 16}, 5, 5}, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + t.Parallel() + + assert.Equal(t, test.want, test.scaling.Workers(test.entries)) + }) + } +} + +// TestReadDirScalingListing lists the same small directory once through the +// default scaling (serial) and once through a scaling tiny enough to put the +// per-entry stats on the worker pool, and expects the same sorted listing. +func TestReadDirScalingListing(t *testing.T) { + t.Parallel() + + source := t.TempDir() + const files = 40 + + want := make([]string, 0, files+1) + for i := range files { + name := fmt.Sprintf("file-%03d", i) + require.NoError(t, os.WriteFile(filepath.Join(source, name), []byte(name), 0o644)) + want = append(want, name) + } + require.NoError(t, os.Mkdir(filepath.Join(source, "zz-dir"), 0o755)) + want = append(want, "zz-dir") + + serial, err := Chroot(source) + require.NoError(t, err) + t.Cleanup(func() { + require.NoError(t, serial.Close()) + }) + + concurrent, err := Chroot(source, + WithReadDirScaling(ReadDirScaling{Threshold: 2, Base: 4, Step: 1, Max: 8}), + ) + require.NoError(t, err) + t.Cleanup(func() { + require.NoError(t, concurrent.Close()) + }) + + serialInfos, err := serial.ReadDir("/") + require.NoError(t, err) + + concurrentInfos, err := concurrent.ReadDir("/") + require.NoError(t, err) + + assert.Equal(t, want, entryNames(serialInfos)) + assert.Equal(t, want, entryNames(concurrentInfos)) +} + +func entryNames(infos []os.FileInfo) []string { + names := make([]string, len(infos)) + for i, info := range infos { + names[i] = info.Name() + } + + return names +}