fuse: Attempt to fix five data races.

I am not fully familiar with this code, but I added some `checklocks`
annotations wherever it seemed appropriate and obvious from existing comments.

PiperOrigin-RevId: 426042114

pkg/sentry/fsimpl/fuse/connection.go

@@ -48,7 +48,7 @@ const (
 type connection struct {
 	fd *DeviceFD
 
-	// mu protects access to struct memebers.
+	// mu protects access to struct members.
 	mu sync.Mutex `state:"nosave"`
 
 	// attributeVersion is the version of connection's attributes.
@@ -84,15 +84,18 @@ type connection struct {
 	//   umount,
 	//   connection abort,
 	//   device release.
+	// +checklocks:mu
 	connected bool
 
 	// connInitError if FUSE_INIT encountered error (major version mismatch).
 	// Only set in INIT.
+	// +checklocks:mu
 	connInitError bool
 
 	// connInitSuccess if FUSE_INIT is successful.
 	// Only set in INIT.
-	// Used for destory (not yet implemented).
+	// Used for destroy (not yet implemented).
+	// +checklocks:mu
 	connInitSuccess bool
 
 	// aborted via sysfs, and will send ECONNABORTED to read after disconnection (instead of ENODEV).
@@ -100,7 +103,7 @@ type connection struct {
 	// TODO(gvisor.dev/issue/3525): set this to true when user aborts.
 	aborted bool
 
-	// numWating is the number of requests waiting to be
+	// numWaiting is the number of requests waiting to be
 	// sent to FUSE device or being processed by FUSE daemon.
 	numWaiting uint32
 
@@ -118,19 +121,19 @@ type connection struct {
 	asyncMu sync.Mutex `state:"nosave"`
 
 	// asyncNum is the number of async requests.
-	// Protected by asyncMu.
+	// +checklocks:asyncMu
 	asyncNum uint16
 
 	// asyncCongestionThreshold the number of async requests.
 	// Negotiated in FUSE_INIT as "CongestionThreshold".
 	// TODO(gvisor.dev/issue/3529): add congestion control.
-	// Protected by asyncMu.
+	// +checklocks:asyncMu
 	asyncCongestionThreshold uint16
 
 	// asyncNumMax is the maximum number of asyncNum.
 	// Connection blocks the async requests when it is reached.
 	// Negotiated in FUSE_INIT as "MaxBackground".
-	// Protected by asyncMu.
+	// +checklocks:asyncMu
 	asyncNumMax uint16
 
 	// maxRead is the maximum size of a read buffer in in bytes.
@@ -201,10 +204,12 @@ func newFUSEConnection(_ context.Context, fuseFD *DeviceFD, opts *filesystemOpti
 
 	// Create the writeBuf for the header to be stored in.
 	hdrLen := uint32((*linux.FUSEHeaderOut)(nil).SizeBytes())
+	fuseFD.mu.Lock()
 	fuseFD.writeBuf = make([]byte, hdrLen)
 	fuseFD.completions = make(map[linux.FUSEOpID]*futureResponse)
 	fuseFD.fullQueueCh = make(chan struct{}, opts.maxActiveRequests)
 	fuseFD.writeCursor = 0
+	fuseFD.mu.Unlock()
 
 	return &connection{
 		fd:                       fuseFD,
@@ -251,15 +256,24 @@ func (conn *connection) Call(t *kernel.Task, r *Request) (*Response, error) {
 		}
 	}
 
-	if !conn.connected {
+	conn.fd.mu.Lock()
+	conn.mu.Lock()
+	connected := conn.connected
+	connInitError := conn.connInitError
+	conn.mu.Unlock()
+
+	if !connected {
+		conn.fd.mu.Unlock()
 		return nil, linuxerr.ENOTCONN
 	}
 
-	if conn.connInitError {
+	if connInitError {
+		conn.fd.mu.Unlock()
 		return nil, linuxerr.ECONNREFUSED
 	}
 
 	fut, err := conn.callFuture(t, r)
+	conn.fd.mu.Unlock()
 	if err != nil {
 		return nil, err
 	}
@@ -269,10 +283,8 @@ func (conn *connection) Call(t *kernel.Task, r *Request) (*Response, error) {
 
 // callFuture makes a request to the server and returns a future response.
 // Call resolve() when the response needs to be fulfilled.
+// +checklocks:conn.fd.mu
 func (conn *connection) callFuture(t *kernel.Task, r *Request) (*futureResponse, error) {
-	conn.fd.mu.Lock()
-	defer conn.fd.mu.Unlock()
-
 	// Is the queue full?
 	//
 	// We must busy wait here until the request can be queued. We don't
@@ -299,6 +311,7 @@ func (conn *connection) callFuture(t *kernel.Task, r *Request) (*futureResponse,
 }
 
 // callFutureLocked makes a request to the server and returns a future response.
+// +checklocks:conn.fd.mu
 func (conn *connection) callFutureLocked(t *kernel.Task, r *Request) (*futureResponse, error) {
 	// Check connected again holding conn.mu.
 	conn.mu.Lock()

pkg/sentry/fsimpl/fuse/connection_control.go

@@ -109,9 +109,13 @@ func (conn *connection) InitRecv(res *Response, hasSysAdminCap bool) error {
 // Process the FUSE_INIT reply from the FUSE server.
 // It tries to acquire the conn.asyncMu lock if minor version is newer than 13.
 func (conn *connection) initProcessReply(out *linux.FUSEInitOut, hasSysAdminCap bool) error {
+	conn.mu.Lock()
 	// No matter error or not, always set initialzied.
 	// to unblock the blocked requests.
-	defer conn.SetInitialized()
+	defer func() {
+		conn.SetInitialized()
+		conn.mu.Unlock()
+	}()
 
 	// No support for old major fuse versions.
 	if out.Major != linux.FUSE_KERNEL_VERSION {
@@ -219,7 +223,7 @@ func (conn *connection) Abort(ctx context.Context) {
 	conn.asyncMu.Unlock()
 	conn.mu.Unlock()
 
-	// 1. The requets blocked before initialization.
+	// 1. The request blocked before initialization.
 	// Will reach call() `connected` check and return.
 	if !conn.Initialized() {
 		conn.SetInitialized()

pkg/sentry/fsimpl/fuse/dev.go

@@ -118,10 +118,17 @@ func (fd *DeviceFD) Release(ctx context.Context) {
 	}
 }
 
+// filesystemIsInitialized returns true if fd.fs is set and the connection is
+// initialized.
+func (fd *DeviceFD) filesystemIsInitialized() bool {
+	// FIXME(gvisor.dev/issue/4813): Access to fd.fs should be synchronized.
+	return fd.fs != nil
+}
+
 // PRead implements vfs.FileDescriptionImpl.PRead.
 func (fd *DeviceFD) PRead(ctx context.Context, dst usermem.IOSequence, offset int64, opts vfs.ReadOptions) (int64, error) {
 	// Operations on /dev/fuse don't make sense until a FUSE filesystem is mounted.
-	if fd.fs == nil {
+	if !fd.filesystemIsInitialized() {
 		return 0, linuxerr.EPERM
 	}
 
@@ -131,7 +138,7 @@ func (fd *DeviceFD) PRead(ctx context.Context, dst usermem.IOSequence, offset in
 // Read implements vfs.FileDescriptionImpl.Read.
 func (fd *DeviceFD) Read(ctx context.Context, dst usermem.IOSequence, opts vfs.ReadOptions) (int64, error) {
 	// Operations on /dev/fuse don't make sense until a FUSE filesystem is mounted.
-	if fd.fs == nil {
+	if !fd.filesystemIsInitialized() {
 		return 0, linuxerr.EPERM
 	}
 
@@ -142,7 +149,9 @@ func (fd *DeviceFD) Read(ctx context.Context, dst usermem.IOSequence, opts vfs.R
 	minBuffSize := linux.FUSE_MIN_READ_BUFFER
 	inHdrLen := uint32((*linux.FUSEHeaderIn)(nil).SizeBytes())
 	writeHdrLen := uint32((*linux.FUSEWriteIn)(nil).SizeBytes())
+	fd.fs.conn.mu.Lock()
 	negotiatedMinBuffSize := inHdrLen + writeHdrLen + fd.fs.conn.maxWrite
+	fd.fs.conn.mu.Unlock()
 	if minBuffSize < negotiatedMinBuffSize {
 		minBuffSize = negotiatedMinBuffSize
 	}
@@ -160,6 +169,7 @@ func (fd *DeviceFD) Read(ctx context.Context, dst usermem.IOSequence, opts vfs.R
 // readLocked implements the reading of the fuse device while locked with DeviceFD.mu.
 //
 // Preconditions: dst is large enough for any reasonable request.
+// +checklocks:fd.mu
 func (fd *DeviceFD) readLocked(ctx context.Context, dst usermem.IOSequence, opts vfs.ReadOptions) (int64, error) {
 	var req *Request
 
@@ -233,7 +243,7 @@ func (fd *DeviceFD) readLocked(ctx context.Context, dst usermem.IOSequence, opts
 // PWrite implements vfs.FileDescriptionImpl.PWrite.
 func (fd *DeviceFD) PWrite(ctx context.Context, src usermem.IOSequence, offset int64, opts vfs.WriteOptions) (int64, error) {
 	// Operations on /dev/fuse don't make sense until a FUSE filesystem is mounted.
-	if fd.fs == nil {
+	if !fd.filesystemIsInitialized() {
 		return 0, linuxerr.EPERM
 	}
 
@@ -248,9 +258,10 @@ func (fd *DeviceFD) Write(ctx context.Context, src usermem.IOSequence, opts vfs.
 }
 
 // writeLocked implements writing to the fuse device while locked with DeviceFD.mu.
+// +checklocks:fd.mu
 func (fd *DeviceFD) writeLocked(ctx context.Context, src usermem.IOSequence, opts vfs.WriteOptions) (int64, error) {
 	// Operations on /dev/fuse don't make sense until a FUSE filesystem is mounted.
-	if fd.fs == nil {
+	if !fd.filesystemIsInitialized() {
 		return 0, linuxerr.EPERM
 	}
 
@@ -359,10 +370,11 @@ func (fd *DeviceFD) Readiness(mask waiter.EventMask) waiter.EventMask {
 
 // readinessLocked implements checking the readiness of the fuse device while
 // locked with DeviceFD.mu.
+// +checklocks:fd.mu
 func (fd *DeviceFD) readinessLocked(mask waiter.EventMask) waiter.EventMask {
 	var ready waiter.EventMask
 
-	if fd.fs == nil || fd.fs.umounted {
+	if !fd.filesystemIsInitialized() || fd.fs.umounted {
 		ready |= waiter.EventErr
 		return ready & mask
 	}
@@ -391,7 +403,7 @@ func (fd *DeviceFD) EventUnregister(e *waiter.Entry) {
 // Seek implements vfs.FileDescriptionImpl.Seek.
 func (fd *DeviceFD) Seek(ctx context.Context, offset int64, whence int32) (int64, error) {
 	// Operations on /dev/fuse don't make sense until a FUSE filesystem is mounted.
-	if fd.fs == nil {
+	if !fd.filesystemIsInitialized() {
 		return 0, linuxerr.EPERM
 	}
 

pkg/sentry/fsimpl/fuse/dev_test.go

@@ -209,7 +209,11 @@ func ReadTest(serverTask *kernel.Task, fd *vfs.FileDescription, inIOseq usermem.
 // a header, a payload, calls the server, waits for the response, and processes
 // the response.
 func fuseClientRun(t *testing.T, s *testutil.System, k *kernel.Kernel, conn *connection, creds *auth.Credentials, pid uint32, inode uint64, clientDone chan struct{}) {
-	defer func() { clientDone <- struct{}{} }()
+	defer func() {
+		if !t.Failed() {
+			clientDone <- struct{}{}
+		}
+	}()
 
 	tc := k.NewThreadGroup(nil, k.RootPIDNamespace(), kernel.NewSignalHandlers(), linux.SIGCHLD, k.GlobalInit().Limits())
 	clientTask, err := testutil.CreateTask(s.Ctx, fmt.Sprintf("fuse-client-%v", pid), tc, s.MntNs, s.Root, s.Root)
@@ -251,7 +255,11 @@ func fuseClientRun(t *testing.T, s *testutil.System, k *kernel.Kernel, conn *con
 // that simply reads a request and echos the same struct back as a response using the
 // appropriate headers.
 func fuseServerRun(t *testing.T, s *testutil.System, k *kernel.Kernel, fd *vfs.FileDescription, serverDone, killServer chan struct{}) {
-	defer func() { serverDone <- struct{}{} }()
+	defer func() {
+		if !t.Failed() {
+			serverDone <- struct{}{}
+		}
+	}()
 
 	// Create the tasks that the server will be using.
 	tc := k.NewThreadGroup(nil, k.RootPIDNamespace(), kernel.NewSignalHandlers(), linux.SIGCHLD, k.GlobalInit().Limits())

pkg/sentry/fsimpl/fuse/fusefs.go

@@ -272,9 +272,12 @@ func newFUSEFilesystem(ctx context.Context, vfsObj *vfs.VirtualFilesystem, fsTyp
 	// reference on fuseFD, since conn uses fuseFD for communication with the
 	// server? Wouldn't doing so create a circular reference?
 	fs.VFSFilesystem().IncRef() // for fuseFD.fs
-	// FIXME(gvisor.dev/issue/4813): fuseFD.fs is accessed without
-	// synchronization.
+
+	fuseFD.mu.Lock()
+	fs.conn.mu.Lock()
 	fuseFD.fs = fs
+	fs.conn.mu.Unlock()
+	fuseFD.mu.Unlock()
 
 	return fs, nil
 }