Fix FUSE connection control lock ordering and race in unit test

3 files changed, 18 insertions, 21 deletions

diff --git a/pkg/sentry/fsimpl/fuse/connection_control.go b/pkg/sentry/fsimpl/fuse/connection_control.go
index a63c66e7c..50cf4a761 100644
--- a/pkg/sentry/fsimpl/fuse/connection_control.go
+++ b/pkg/sentry/fsimpl/fuse/connection_control.go
@@ -95,6 +95,8 @@ func (conn *connection) InitSend(creds *auth.Credentials, pid uint32) error {
 }
 
 // InitRecv receives a FUSE_INIT reply and process it.
+//
+// Preconditions: conn.asyncMu must not be held if minor verion is newer than 13.
 func (conn *connection) InitRecv(res *Response, hasSysAdminCap bool) error {
 	if err := res.Error(); err != nil {
 		return err
@@ -189,6 +191,8 @@ func (conn *connection) initProcessReply(out *linux.FUSEInitOut, hasSysAdminCap
 // All possible requests waiting or blocking will be aborted.
 func (conn *connection) Abort(ctx context.Context) {
 	conn.fd.mu.Lock()
+	defer conn.fd.mu.Unlock()
+
 	conn.mu.Lock()
 	conn.asyncMu.Lock()
 
@@ -217,10 +221,9 @@ func (conn *connection) Abort(ctx context.Context) {
 		terminate = append(terminate, unique)
 	}
 
-	// Release all locks to avoid deadlock.
+	// Release locks to avoid deadlock.
 	conn.asyncMu.Unlock()
 	conn.mu.Unlock()
-	conn.fd.mu.Unlock()
 
 	// 1. The requets blocked before initialization.
 	// Will reach call() `connected` check and return.
diff --git a/pkg/sentry/fsimpl/fuse/connection_test.go b/pkg/sentry/fsimpl/fuse/connection_test.go
index 6aa77b36d..91d16c1cf 100644
--- a/pkg/sentry/fsimpl/fuse/connection_test.go
+++ b/pkg/sentry/fsimpl/fuse/connection_test.go
@@ -62,9 +62,9 @@ func TestConnectionAbort(t *testing.T) {
 	creds := auth.CredentialsFromContext(s.Ctx)
 	task := kernel.TaskFromContext(s.Ctx)
 
-	const maxActiveRequest uint64 = 10
+	const numRequests uint64 = 256
 
-	conn, _, err := newTestConnection(s, k, maxActiveRequest)
+	conn, _, err := newTestConnection(s, k, numRequests)
 	if err != nil {
 		t.Fatalf("newTestConnection: %v", err)
 	}
@@ -75,32 +75,22 @@ func TestConnectionAbort(t *testing.T) {
 
 	var futNormal []*futureResponse
 
-	for i := 0; i < 2*int(maxActiveRequest); i++ {
+	for i := 0; i < int(numRequests); i++ {
 		req, err := conn.NewRequest(creds, uint32(i), uint64(i), 0, testObj)
 		if err != nil {
 			t.Fatalf("NewRequest creation failed: %v", err)
 		}
-		if i < int(maxActiveRequest) {
-			// Issue the requests that will not be blocked due to maxActiveRequest.
-			fut, err := conn.callFutureLocked(task, req)
-			if err != nil {
-				t.Fatalf("callFutureLocked failed: %v", err)
-			}
-			futNormal = append(futNormal, fut)
-		} else {
-			go func(t *testing.T) {
-				// The requests beyond maxActiveRequest will be blocked and receive expected errors.
-				_, err := conn.callFutureLocked(task, req)
-				if err != syserror.ECONNABORTED && err != syserror.ENOTCONN {
-					t.Fatalf("Incorrect error code received for blocked callFutureLocked() on aborted connection")
-				}
-			}(t)
+		fut, err := conn.callFutureLocked(task, req)
+		if err != nil {
+			t.Fatalf("callFutureLocked failed: %v", err)
 		}
+		futNormal = append(futNormal, fut)
 	}
 
 	conn.Abort(s.Ctx)
 
 	// Abort should unblock the initialization channel.
+	// Note: no test requests are actually blocked on `conn.initializedChan`.
 	select {
 	case <-conn.initializedChan:
 	default:
@@ -110,7 +100,7 @@ func TestConnectionAbort(t *testing.T) {
 	// Abort will return ECONNABORTED error to unblocked requests.
 	for _, fut := range futNormal {
 		if fut.getResponse().hdr.Error != -int32(syscall.ECONNABORTED) {
-			t.Fatalf("Incorrect error code received for aborted connection")
+			t.Fatalf("Incorrect error code received for aborted connection: %v", fut.getResponse().hdr.Error)
 		}
 	}
 
diff --git a/pkg/sentry/fsimpl/fuse/dev.go b/pkg/sentry/fsimpl/fuse/dev.go
index 64c3e32e2..a19544fe9 100644
--- a/pkg/sentry/fsimpl/fuse/dev.go
+++ b/pkg/sentry/fsimpl/fuse/dev.go
@@ -391,6 +391,8 @@ func (fd *DeviceFD) Seek(ctx context.Context, offset int64, whence int32) (int64
 }
 
 // sendResponse sends a response to the waiting task (if any).
+//
+// Preconditions: fd.mu must be held.
 func (fd *DeviceFD) sendResponse(ctx context.Context, fut *futureResponse) error {
 	// Signal the task waiting on a response if any.
 	defer close(fut.ch)
@@ -410,6 +412,8 @@ func (fd *DeviceFD) sendResponse(ctx context.Context, fut *futureResponse) error
 }
 
 // sendError sends an error response to the waiting task (if any) by calling sendResponse().
+//
+// Preconditions: fd.mu must be held.
 func (fd *DeviceFD) sendError(ctx context.Context, errno int32, unique linux.FUSEOpID) error {
 	// Return the error to the calling task.
 	outHdrLen := uint32((*linux.FUSEHeaderOut)(nil).SizeBytes())