Fix bug in umount logic, causing long delays.

Signed-off-by: Thomas Hallgren <[email protected]>

Author: thallgren

pkg/fs/ftp.go

@@ -3,7 +3,6 @@ package fs
 
 import (
 	"bytes"
-	"context"
 	"errors"
 	"io"
 	"io/fs"
@@ -30,9 +29,6 @@ type fuseImpl struct {
 	// connPool is the pool of control connections to the remote FTP server.
 	pool connPool
 
-	// cancel the GC loop
-	cancel context.CancelFunc
-
 	// Mutex protects nextHandle, current, and shuttingDown
 	sync.RWMutex
 
@@ -114,10 +110,8 @@ type FTPClient interface {
 // NewFTPClient returns an implementation of the fuse.FileSystemInterface that is backed by
 // an FTP server connection tp the address. The dir parameter is the directory that the
 // FTP server changes to when connecting.
-func NewFTPClient(ctx context.Context, addr netip.AddrPort, dir string, ro bool, readTimeout time.Duration) (FTPClient, error) {
-	ctx, cancel := context.WithCancel(ctx)
+func NewFTPClient(done <-chan struct{}, addr netip.AddrPort, dir string, ro bool, readTimeout time.Duration) (FTPClient, error) {
 	f := &fuseImpl{
-		cancel:   cancel,
 		current:  make(map[uint64]*info),
 		readOnly: ro,
 		pool: connPool{
@@ -126,10 +120,11 @@ func NewFTPClient(ctx context.Context, addr netip.AddrPort, dir string, ro bool,
 		},
 	}
 	go func() {
+		defer f.Destroy()
 		ticker := time.NewTicker(stalePeriod)
 		for {
 			select {
-			case <-ctx.Done():
+			case <-done:
 				return
 			case <-ticker.C:
 				f.pool.tidy()
@@ -138,7 +133,6 @@ func NewFTPClient(ctx context.Context, addr netip.AddrPort, dir string, ro bool,
 	}()
 
 	if err := f.pool.setAddr(addr); err != nil {
-		cancel()
 		return nil, err
 	}
 	return f, nil
@@ -191,7 +185,6 @@ func (f *fuseImpl) Destroy() {
 	}
 	wg.Wait()
 	f.pool.quit()
-	f.cancel()
 }
 
 // Flush is a noop in this implementation

pkg/fs/ftp_test.go

@@ -139,7 +139,7 @@ func startFUSEHost(t *testing.T, ctx context.Context, port uint16, dir string, r
 	// Start the client
 	dir = filepath.Join(dir, "mount")
 	require.NoError(t, os.Mkdir(dir, 0755))
-	fsh, err := NewFTPClient(ctx, netip.MustParseAddrPort(fmt.Sprintf("127.0.0.1:%d", port)), remoteDir, readOnly, 60*time.Second)
+	fsh, err := NewFTPClient(ctx.Done(), netip.MustParseAddrPort(fmt.Sprintf("127.0.0.1:%d", port)), remoteDir, readOnly, 60*time.Second)
 	require.NoError(t, err)
 	mp := dir
 	if runtime.GOOS == "windows" {
@@ -153,7 +153,7 @@ func startFUSEHost(t *testing.T, ctx context.Context, port uint16, dir string, r
 
 func TestConnectFailure(t *testing.T) {
 	ctx := testContext(t)
-	_, err := NewFTPClient(ctx, netip.MustParseAddrPort("198.51.100.32:21"), "", false, time.Second)
+	_, err := NewFTPClient(ctx.Done(), netip.MustParseAddrPort("198.51.100.32:21"), "", false, time.Second)
 	require.Error(t, err)
 }
 

pkg/fs/fuse.go

@@ -51,9 +51,7 @@ func (fh *FuseHost) Start(ctx context.Context, startTimeout time.Duration) error
 	go fh.detectFuseStarted(startCtx, started)
 
 	mCh := make(chan bool, 1)
-	fh.wg.Add(1)
 	go func() {
-		defer fh.wg.Done()
 		mCh <- fh.host.Mount(fh.mountPoint, opts)
 	}()
 	go func() {
@@ -66,7 +64,9 @@ func (fh *FuseHost) Start(ctx context.Context, startTimeout time.Duration) error
 		defer fh.wg.Done()
 		select {
 		case <-ctx.Done():
+			logrus.Debug("FuseHost unmounting")
 			fh.host.Unmount()
+			logrus.Debug("FuseHost unmount complete")
 		case mountResult := <-mCh:
 			if !mountResult {
 				logrus.Errorf("fuse mount of %s failed", fh.mountPoint)

pkg/main/main.go

@@ -75,7 +75,7 @@ func (s *service) Mount(_ context.Context, rq *rpc.MountRequest) (*rpc.MountIden
 		return nil, err
 	}
 	ctx, cancel := context.WithCancel(s.ctx)
-	fi, err := fs.NewFTPClient(ctx, ap, rq.Directory, rq.ReadOnly, rq.ReadTimeout.AsDuration())
+	fi, err := fs.NewFTPClient(ctx.Done(), ap, rq.Directory, rq.ReadOnly, rq.ReadTimeout.AsDuration())
 	if err != nil {
 		cancel()
 		return nil, status.Error(codes.Internal, err.Error())
@@ -94,19 +94,23 @@ func (s *service) Mount(_ context.Context, rq *rpc.MountRequest) (*rpc.MountIden
 			s.Lock()
 			delete(s.mounts, id)
 			s.Unlock()
+			logrus.Debug("FuseHost Stop")
 			host.Stop()
+			logrus.Debug("Cancelling mount context")
 			cancel()
 		},
 	}
 	s.nextID++
 	return &rpc.MountIdentifier{Id: id}, nil
 }
 
-func (s *service) Unmount(_ context.Context, rq *rpc.MountIdentifier) (*emptypb.Empty, error) {
+func (s *service) Unmount(ctx context.Context, rq *rpc.MountIdentifier) (*emptypb.Empty, error) {
+	logrus.Debug("Unmount called")
 	s.Lock()
 	m, ok := s.mounts[rq.Id]
 	s.Unlock()
 	if ok {
+		logrus.Debug("cancelling mount")
 		m.cancel()
 	}
 	return &emptypb.Empty{}, nil