1 files changed, 68 insertions, 117 deletions

diff --git a/pkg/sentry/fsimpl/fuse/dev.go b/pkg/sentry/fsimpl/fuse/dev.go
index 64c3e32e2..e522ff9a0 100644
--- a/pkg/sentry/fsimpl/fuse/dev.go
+++ b/pkg/sentry/fsimpl/fuse/dev.go
@@ -19,6 +19,7 @@ import (
 
 	"gvisor.dev/gvisor/pkg/abi/linux"
 	"gvisor.dev/gvisor/pkg/context"
+	"gvisor.dev/gvisor/pkg/log"
 	"gvisor.dev/gvisor/pkg/sentry/kernel"
 	"gvisor.dev/gvisor/pkg/sentry/kernel/auth"
 	"gvisor.dev/gvisor/pkg/sentry/vfs"
@@ -55,6 +56,9 @@ type DeviceFD struct {
 	vfs.DentryMetadataFileDescriptionImpl
 	vfs.NoLockFD
 
+	// mounted specifies whether a FUSE filesystem was mounted using the DeviceFD.
+	mounted bool
+
 	// nextOpID is used to create new requests.
 	nextOpID linux.FUSEOpID
 
@@ -96,15 +100,13 @@ type DeviceFD struct {
 
 // Release implements vfs.FileDescriptionImpl.Release.
 func (fd *DeviceFD) Release(context.Context) {
-	if fd.fs != nil {
-		fd.fs.conn.connected = false
-	}
+	fd.fs.conn.connected = false
 }
 
 // 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.mounted {
 		return 0, syserror.EPERM
 	}
 
@@ -114,16 +116,10 @@ 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.mounted {
 		return 0, syserror.EPERM
 	}
 
-	// Return ENODEV if the filesystem is umounted.
-	if fd.fs.umounted {
-		// TODO(gvisor.dev/issue/3525): return ECONNABORTED if aborted via fuse control fs.
-		return 0, syserror.ENODEV
-	}
-
 	// We require that any Read done on this filesystem have a sane minimum
 	// read buffer. It must have the capacity for the fixed parts of any request
 	// header (Linux uses the request header and the FUSEWriteIn header for this
@@ -147,82 +143,58 @@ 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.
 func (fd *DeviceFD) readLocked(ctx context.Context, dst usermem.IOSequence, opts vfs.ReadOptions) (int64, error) {
-	var req *Request
+	if fd.queue.Empty() {
+		return 0, syserror.ErrWouldBlock
+	}
 
-	// Find the first valid request.
-	// For the normal case this loop only execute once.
-	for !fd.queue.Empty() {
-		req = fd.queue.Front()
+	var readCursor uint32
+	var bytesRead int64
+	for {
+		req := fd.queue.Front()
+		if dst.NumBytes() < int64(req.hdr.Len) {
+			// The request is too large. Cannot process it. All requests must be smaller than the
+			// negotiated size as specified by Connection.MaxWrite set as part of the FUSE_INIT
+			// handshake.
+			errno := -int32(syscall.EIO)
+			if req.hdr.Opcode == linux.FUSE_SETXATTR {
+				errno = -int32(syscall.E2BIG)
+			}
 
-		if int64(req.hdr.Len)+int64(len(req.payload)) <= dst.NumBytes() {
-			break
-		}
+			// Return the error to the calling task.
+			if err := fd.sendError(ctx, errno, req); err != nil {
+				return 0, err
+			}
 
-		// The request is too large. Cannot process it. All requests must be smaller than the
-		// negotiated size as specified by Connection.MaxWrite set as part of the FUSE_INIT
-		// handshake.
-		errno := -int32(syscall.EIO)
-		if req.hdr.Opcode == linux.FUSE_SETXATTR {
-			errno = -int32(syscall.E2BIG)
-		}
+			// We're done with this request.
+			fd.queue.Remove(req)
 
-		// Return the error to the calling task.
-		if err := fd.sendError(ctx, errno, req.hdr.Unique); err != nil {
-			return 0, err
+			// Restart the read as this request was invalid.
+			log.Warningf("fuse.DeviceFD.Read: request found was too large. Restarting read.")
+			return fd.readLocked(ctx, dst, opts)
 		}
 
-		// We're done with this request.
-		fd.queue.Remove(req)
-		req = nil
-	}
-
-	if req == nil {
-		return 0, syserror.ErrWouldBlock
-	}
-
-	// We already checked the size: dst must be able to fit the whole request.
-	// Now we write the marshalled header, the payload,
-	// and the potential additional payload
-	// to the user memory IOSequence.
-
-	n, err := dst.CopyOut(ctx, req.data)
-	if err != nil {
-		return 0, err
-	}
-	if n != len(req.data) {
-		return 0, syserror.EIO
-	}
-
-	if req.hdr.Opcode == linux.FUSE_WRITE {
-		written, err := dst.DropFirst(n).CopyOut(ctx, req.payload)
+		n, err := dst.CopyOut(ctx, req.data[readCursor:])
 		if err != nil {
 			return 0, err
 		}
-		if written != len(req.payload) {
-			return 0, syserror.EIO
-		}
-		n += int(written)
-	}
+		readCursor += uint32(n)
+		bytesRead += int64(n)
 
-	// Fully done with this req, remove it from the queue.
-	fd.queue.Remove(req)
-
-	// Remove noReply ones from map of requests expecting a reply.
-	if req.noReply {
-		fd.numActiveRequests -= 1
-		delete(fd.completions, req.hdr.Unique)
+		if readCursor >= req.hdr.Len {
+			// Fully done with this req, remove it from the queue.
+			fd.queue.Remove(req)
+			break
+		}
 	}
 
-	return int64(n), nil
+	return bytesRead, nil
 }
 
 // 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.mounted {
 		return 0, syserror.EPERM
 	}
 
@@ -239,15 +211,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.
 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.mounted {
 		return 0, syserror.EPERM
 	}
 
-	// Return ENODEV if the filesystem is umounted.
-	if fd.fs.umounted {
-		return 0, syserror.ENODEV
-	}
-
 	var cn, n int64
 	hdrLen := uint32((*linux.FUSEHeaderOut)(nil).SizeBytes())
 
@@ -309,12 +276,7 @@ func (fd *DeviceFD) writeLocked(ctx context.Context, src usermem.IOSequence, opt
 
 			fut, ok := fd.completions[hdr.Unique]
 			if !ok {
-				if fut.hdr.Unique == linux.FUSE_RELEASE {
-					// Currently we simply discard the reply for FUSE_RELEASE.
-					return n + src.NumBytes(), nil
-				}
-				// Server sent us a response for a request we never sent,
-				// or for which we already received a reply (e.g. aborted), an unlikely event.
+				// Server sent us a response for a request we never sent?
 				return 0, syserror.EINVAL
 			}
 
@@ -345,23 +307,8 @@ func (fd *DeviceFD) writeLocked(ctx context.Context, src usermem.IOSequence, opt
 
 // Readiness implements vfs.FileDescriptionImpl.Readiness.
 func (fd *DeviceFD) Readiness(mask waiter.EventMask) waiter.EventMask {
-	fd.mu.Lock()
-	defer fd.mu.Unlock()
-	return fd.readinessLocked(mask)
-}
-
-// readinessLocked implements checking the readiness of the fuse device while
-// locked with DeviceFD.mu.
-func (fd *DeviceFD) readinessLocked(mask waiter.EventMask) waiter.EventMask {
 	var ready waiter.EventMask
-
-	if fd.fs.umounted {
-		ready |= waiter.EventErr
-		return ready & mask
-	}
-
-	// FD is always writable.
-	ready |= waiter.EventOut
+	ready |= waiter.EventOut // FD is always writable
 	if !fd.queue.Empty() {
 		// Have reqs available, FD is readable.
 		ready |= waiter.EventIn
@@ -383,7 +330,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.mounted {
 		return 0, syserror.EPERM
 	}
 
@@ -392,54 +339,58 @@ func (fd *DeviceFD) Seek(ctx context.Context, offset int64, whence int32) (int64
 
 // sendResponse sends a response to the waiting task (if any).
 func (fd *DeviceFD) sendResponse(ctx context.Context, fut *futureResponse) error {
-	// Signal the task waiting on a response if any.
-	defer close(fut.ch)
+	// See if the running task need to perform some action before returning.
+	// Since we just finished writing the future, we can be sure that
+	// getResponse generates a populated response.
+	if err := fd.noReceiverAction(ctx, fut.getResponse()); err != nil {
+		return err
+	}
 
 	// Signal that the queue is no longer full.
 	select {
 	case fd.fullQueueCh <- struct{}{}:
 	default:
 	}
-	fd.numActiveRequests--
-
-	if fut.async {
-		return fd.asyncCallBack(ctx, fut.getResponse())
-	}
+	fd.numActiveRequests -= 1
 
+	// Signal the task waiting on a response.
+	close(fut.ch)
 	return nil
 }
 
-// sendError sends an error response to the waiting task (if any) by calling sendResponse().
-func (fd *DeviceFD) sendError(ctx context.Context, errno int32, unique linux.FUSEOpID) error {
+// sendError sends an error response to the waiting task (if any).
+func (fd *DeviceFD) sendError(ctx context.Context, errno int32, req *Request) error {
 	// Return the error to the calling task.
 	outHdrLen := uint32((*linux.FUSEHeaderOut)(nil).SizeBytes())
 	respHdr := linux.FUSEHeaderOut{
 		Len:    outHdrLen,
 		Error:  errno,
-		Unique: unique,
+		Unique: req.hdr.Unique,
 	}
 
 	fut, ok := fd.completions[respHdr.Unique]
 	if !ok {
-		// A response for a request we never sent,
-		// or for which we already received a reply (e.g. aborted).
+		// Server sent us a response for a request we never sent?
 		return syserror.EINVAL
 	}
 	delete(fd.completions, respHdr.Unique)
 
 	fut.hdr = &respHdr
-	return fd.sendResponse(ctx, fut)
+	if err := fd.sendResponse(ctx, fut); err != nil {
+		return err
+	}
+
+	return nil
 }
 
-// asyncCallBack executes pre-defined callback function for async requests.
-// Currently used by: FUSE_INIT.
-func (fd *DeviceFD) asyncCallBack(ctx context.Context, r *Response) error {
-	switch r.opcode {
-	case linux.FUSE_INIT:
+// noReceiverAction has the calling kernel.Task do some action if its known that no
+// receiver is going to be waiting on the future channel. This is to be used by:
+// FUSE_INIT.
+func (fd *DeviceFD) noReceiverAction(ctx context.Context, r *Response) error {
+	if r.opcode == linux.FUSE_INIT {
 		creds := auth.CredentialsFromContext(ctx)
 		rootUserNs := kernel.KernelFromContext(ctx).RootUserNamespace()
 		return fd.fs.conn.InitRecv(r, creds.HasCapabilityIn(linux.CAP_SYS_ADMIN, rootUserNs))
-		// TODO(gvisor.dev/issue/3247): support async read: correctly process the response.
 	}
 
 	return nil