From 2206234a85584307e1e7872cbac25fb896b64fcd Mon Sep 17 00:00:00 2001 From: Prince Kumar Date: Thu, 4 Jun 2026 14:53:19 +0000 Subject: [PATCH 1/3] feat(max_thread): adding cap for the goroutines which handles the fuse_request --- connection.go | 18 +++++ fuseutil/file_system.go | 11 ++- mount_config.go | 4 ++ mount_test.go | 156 ++++++++++++++++++++++++++++++++++++++++ 4 files changed, 188 insertions(+), 1 deletion(-) diff --git a/connection.go b/connection.go index ba65e67b..8e059655 100644 --- a/connection.go +++ b/connection.go @@ -121,6 +121,18 @@ func newConnection( cancelFuncs: make(map[uint64]func()), } + const defaultMaxThreads = 100000 + const hardMaxThreads = 100000 + + if c.cfg.MaxThreads <= 0 { + c.cfg.MaxThreads = defaultMaxThreads + } else if c.cfg.MaxThreads > hardMaxThreads { + if errorLogger != nil { + errorLogger.Printf("fuse: max_threads %d is beyond the hard limit of %d, ignoring configuration", c.cfg.MaxThreads, hardMaxThreads) + } + c.cfg.MaxThreads = defaultMaxThreads + } + // Initialize. if err := c.Init(); err != nil { c.close() @@ -618,6 +630,12 @@ func (c *Connection) callbackForOp(op interface{}) func() { return nil } +// MaxThreads returns the maximum number of concurrent worker goroutines +// configured for this connection. +func (c *Connection) MaxThreads() int { + return c.cfg.MaxThreads +} + // Close the connection. Must not be called until operations that were read // from the connection have been responded to. func (c *Connection) close() error { diff --git a/fuseutil/file_system.go b/fuseutil/file_system.go index b1c27f56..7c4c966c 100644 --- a/fuseutil/file_system.go +++ b/fuseutil/file_system.go @@ -104,6 +104,9 @@ func (s *fileSystemServer) ServeOps(c *fuse.Connection) { s.fs.Destroy() }() + maxThreads := c.MaxThreads() + sem := make(chan struct{}, maxThreads) + for { ctx, op, err := c.ReadOp() if err == io.EOF { @@ -122,7 +125,13 @@ func (s *fileSystemServer) ServeOps(c *fuse.Connection) { // cheap for the file system to handle s.handleOp(c, ctx, op) } else { - go s.handleOp(c, ctx, op) + sem <- struct{}{} + go func(ctx context.Context, op interface{}) { + defer func() { + <-sem + }() + s.handleOp(c, ctx, op) + }(ctx, op) } } } diff --git a/mount_config.go b/mount_config.go index f95895ad..0e0f655b 100644 --- a/mount_config.go +++ b/mount_config.go @@ -241,6 +241,10 @@ type MountConfig struct { // to always provide ReadFileOp.Dst. If the file system populates ReadFileOp.Data, // that data will be used for a vectored read, irrespective of this flag's value. UseVectoredRead bool + + // The maximum number of concurrent worker goroutines that can handle FUSE + // requests. If not set (or 0), defaults to 100,000. + MaxThreads int } type FUSEImpl uint8 diff --git a/mount_test.go b/mount_test.go index 87cc36a2..66bfad54 100644 --- a/mount_test.go +++ b/mount_test.go @@ -6,7 +6,9 @@ import ( "os" "path" "strings" + "sync" "testing" + "time" "github.com/jacobsa/fuse" "github.com/jacobsa/fuse/fuseops" @@ -93,3 +95,157 @@ func TestNonexistentMountPoint(t *testing.T) { t.Errorf("Unexpected error: %v", got) } } + +//////////////////////////////////////////////////////////////////////// +// blockingFS +//////////////////////////////////////////////////////////////////////// + +type blockingFS struct { + fuseutil.NotImplementedFileSystem + mu sync.Mutex + activeOps int + maxActive int + releaseCh chan struct{} +} + +func (fs *blockingFS) StatFS( + ctx context.Context, + op *fuseops.StatFSOp) error { + return nil +} + +func (fs *blockingFS) GetInodeAttributes( + ctx context.Context, + op *fuseops.GetInodeAttributesOp) error { + if op.Inode == fuseops.RootInodeID { + op.Attributes = fuseops.InodeAttributes{ + Mode: 0777 | os.ModeDir, + } + return nil + } + return fuse.ENOENT +} + +func (fs *blockingFS) LookUpInode( + ctx context.Context, + op *fuseops.LookUpInodeOp) error { + + fs.mu.Lock() + fs.activeOps++ + if fs.activeOps > fs.maxActive { + fs.maxActive = fs.activeOps + } + fs.mu.Unlock() + + defer func() { + fs.mu.Lock() + fs.activeOps-- + fs.mu.Unlock() + }() + + // Block until released. + <-fs.releaseCh + + if op.Name == "foo" || op.Name == "bar" || op.Name == "baz" { + op.Entry = fuseops.ChildInodeEntry{ + Child: 100, // Canned ID + Attributes: fuseops.InodeAttributes{ + Mode: 0444, + }, + } + return nil + } + + return fuse.ENOENT +} + +func TestMaxThreads(t *testing.T) { + ctx := context.Background() + + // Set up a temporary directory. + dir, err := os.MkdirTemp("", "mount_test") + if err != nil { + t.Fatalf("os.MkdirTemp: %v", err) + } + defer os.RemoveAll(dir) + + // Mount with MaxThreads = 2. + fs := &blockingFS{ + releaseCh: make(chan struct{}), + } + mfs, err := fuse.Mount( + dir, + fuseutil.NewFileSystemServer(fs), + &fuse.MountConfig{ + MaxThreads: 2, + }) + if err != nil { + t.Fatalf("fuse.Mount: %v", err) + } + defer func() { + if err := mfs.Join(ctx); err != nil { + t.Errorf("Joining: %v", err) + } + }() + defer fuse.Unmount(mfs.Dir()) + + // Start 3 goroutines, each doing a path lookup. + errChan := make(chan error, 3) + + go func() { + _, err := os.Stat(path.Join(dir, "foo")) + errChan <- err + }() + go func() { + _, err := os.Stat(path.Join(dir, "bar")) + errChan <- err + }() + go func() { + _, err := os.Stat(path.Join(dir, "baz")) + errChan <- err + }() + + // Wait a bit to let all 3 requests be sent by the OS to FUSE. + // Since we set MaxThreads to 2, the third request should be blocked in the server + // and never reach LookUpInode. + time.Sleep(100 * time.Millisecond) + + fs.mu.Lock() + active := fs.activeOps + maxActive := fs.maxActive + fs.mu.Unlock() + + if active > 2 { + t.Errorf("activeOps was %d, expected <= 2", active) + } + if maxActive > 2 { + t.Errorf("maxActive was %d, expected <= 2", maxActive) + } + + // Release one operation. + fs.releaseCh <- struct{}{} + + // Give the 3rd operation time to enter. + time.Sleep(100 * time.Millisecond) + + fs.mu.Lock() + active = fs.activeOps + fs.mu.Unlock() + + if active > 2 { + t.Errorf("activeOps after one release was %d, expected <= 2", active) + } + + // Release the other two. + fs.releaseCh <- struct{}{} + fs.releaseCh <- struct{}{} + + // Wait for all 3 goroutines to finish. + for i := 0; i < 3; i++ { + // Note: os.Stat might fail with ENOENT but should not return other errors. + err := <-errChan + if err != nil && !os.IsNotExist(err) { + t.Errorf("os.Stat failed: %v", err) + } + } +} From 09385e2d87403c51d66701934f5e629acb9ea930 Mon Sep 17 00:00:00 2001 From: Prince Kumar Date: Thu, 4 Jun 2026 15:23:00 +0000 Subject: [PATCH 2/3] removing the handling by default --- connection.go | 7 +++---- fuseutil/file_system.go | 23 +++++++++++++++-------- mount_config.go | 2 +- 3 files changed, 19 insertions(+), 13 deletions(-) diff --git a/connection.go b/connection.go index 8e059655..323ec911 100644 --- a/connection.go +++ b/connection.go @@ -121,16 +121,15 @@ func newConnection( cancelFuncs: make(map[uint64]func()), } - const defaultMaxThreads = 100000 const hardMaxThreads = 100000 - if c.cfg.MaxThreads <= 0 { - c.cfg.MaxThreads = defaultMaxThreads + if c.cfg.MaxThreads < 0 { + c.cfg.MaxThreads = 0 } else if c.cfg.MaxThreads > hardMaxThreads { if errorLogger != nil { errorLogger.Printf("fuse: max_threads %d is beyond the hard limit of %d, ignoring configuration", c.cfg.MaxThreads, hardMaxThreads) } - c.cfg.MaxThreads = defaultMaxThreads + c.cfg.MaxThreads = 0 } // Initialize. diff --git a/fuseutil/file_system.go b/fuseutil/file_system.go index 7c4c966c..9172528c 100644 --- a/fuseutil/file_system.go +++ b/fuseutil/file_system.go @@ -105,7 +105,10 @@ func (s *fileSystemServer) ServeOps(c *fuse.Connection) { }() maxThreads := c.MaxThreads() - sem := make(chan struct{}, maxThreads) + var sem chan struct{} + if maxThreads > 0 { + sem = make(chan struct{}, maxThreads) + } for { ctx, op, err := c.ReadOp() @@ -125,13 +128,17 @@ func (s *fileSystemServer) ServeOps(c *fuse.Connection) { // cheap for the file system to handle s.handleOp(c, ctx, op) } else { - sem <- struct{}{} - go func(ctx context.Context, op interface{}) { - defer func() { - <-sem - }() - s.handleOp(c, ctx, op) - }(ctx, op) + if sem != nil { + sem <- struct{}{} + go func(ctx context.Context, op interface{}) { + defer func() { + <-sem + }() + s.handleOp(c, ctx, op) + }(ctx, op) + } else { + go s.handleOp(c, ctx, op) + } } } } diff --git a/mount_config.go b/mount_config.go index 0e0f655b..6be2c7b3 100644 --- a/mount_config.go +++ b/mount_config.go @@ -243,7 +243,7 @@ type MountConfig struct { UseVectoredRead bool // The maximum number of concurrent worker goroutines that can handle FUSE - // requests. If not set (or 0), defaults to 100,000. + // requests. If not set (or 0), defaults to unlimited (no capping). MaxThreads int } From 23fa40745cc3396053504cbf4d8eb1c8df2081a6 Mon Sep 17 00:00:00 2001 From: Prince Kumar Date: Thu, 4 Jun 2026 15:38:04 +0000 Subject: [PATCH 3/3] simplified the test a bit --- connection.go | 6 ++---- mount_config.go | 2 +- mount_test.go | 43 +++++++++---------------------------------- 3 files changed, 12 insertions(+), 39 deletions(-) diff --git a/connection.go b/connection.go index 323ec911..c8f93f82 100644 --- a/connection.go +++ b/connection.go @@ -123,9 +123,7 @@ func newConnection( const hardMaxThreads = 100000 - if c.cfg.MaxThreads < 0 { - c.cfg.MaxThreads = 0 - } else if c.cfg.MaxThreads > hardMaxThreads { + if c.cfg.MaxThreads > hardMaxThreads { if errorLogger != nil { errorLogger.Printf("fuse: max_threads %d is beyond the hard limit of %d, ignoring configuration", c.cfg.MaxThreads, hardMaxThreads) } @@ -631,7 +629,7 @@ func (c *Connection) callbackForOp(op interface{}) func() { // MaxThreads returns the maximum number of concurrent worker goroutines // configured for this connection. -func (c *Connection) MaxThreads() int { +func (c *Connection) MaxThreads() uint32 { return c.cfg.MaxThreads } diff --git a/mount_config.go b/mount_config.go index 6be2c7b3..95a91f4a 100644 --- a/mount_config.go +++ b/mount_config.go @@ -244,7 +244,7 @@ type MountConfig struct { // The maximum number of concurrent worker goroutines that can handle FUSE // requests. If not set (or 0), defaults to unlimited (no capping). - MaxThreads int + MaxThreads uint32 } type FUSEImpl uint8 diff --git a/mount_test.go b/mount_test.go index 66bfad54..cb5563dd 100644 --- a/mount_test.go +++ b/mount_test.go @@ -6,7 +6,6 @@ import ( "os" "path" "strings" - "sync" "testing" "time" @@ -102,9 +101,7 @@ func TestNonexistentMountPoint(t *testing.T) { type blockingFS struct { fuseutil.NotImplementedFileSystem - mu sync.Mutex - activeOps int - maxActive int + enteredCh chan string releaseCh chan struct{} } @@ -129,19 +126,7 @@ func (fs *blockingFS) GetInodeAttributes( func (fs *blockingFS) LookUpInode( ctx context.Context, op *fuseops.LookUpInodeOp) error { - - fs.mu.Lock() - fs.activeOps++ - if fs.activeOps > fs.maxActive { - fs.maxActive = fs.activeOps - } - fs.mu.Unlock() - - defer func() { - fs.mu.Lock() - fs.activeOps-- - fs.mu.Unlock() - }() + fs.enteredCh <- op.Name // Block until released. <-fs.releaseCh @@ -171,13 +156,15 @@ func TestMaxThreads(t *testing.T) { // Mount with MaxThreads = 2. fs := &blockingFS{ + enteredCh: make(chan string, 3), releaseCh: make(chan struct{}), } mfs, err := fuse.Mount( dir, fuseutil.NewFileSystemServer(fs), &fuse.MountConfig{ - MaxThreads: 2, + MaxThreads: 2, + EnableParallelDirOps: true, }) if err != nil { t.Fatalf("fuse.Mount: %v", err) @@ -210,16 +197,8 @@ func TestMaxThreads(t *testing.T) { // and never reach LookUpInode. time.Sleep(100 * time.Millisecond) - fs.mu.Lock() - active := fs.activeOps - maxActive := fs.maxActive - fs.mu.Unlock() - - if active > 2 { - t.Errorf("activeOps was %d, expected <= 2", active) - } - if maxActive > 2 { - t.Errorf("maxActive was %d, expected <= 2", maxActive) + if got := len(fs.enteredCh); got != 2 { + t.Errorf("enteredOps was %d, expected 2", got) } // Release one operation. @@ -228,12 +207,8 @@ func TestMaxThreads(t *testing.T) { // Give the 3rd operation time to enter. time.Sleep(100 * time.Millisecond) - fs.mu.Lock() - active = fs.activeOps - fs.mu.Unlock() - - if active > 2 { - t.Errorf("activeOps after one release was %d, expected <= 2", active) + if got := len(fs.enteredCh); got != 3 { + t.Errorf("enteredOps after one release was %d, expected 3", got) } // Release the other two.