diff --git a/pkg/sentry/fsimpl/fuse/connection.go b/pkg/sentry/fsimpl/fuse/connection.go index d404edaf0..7ac672d83 100644 --- a/pkg/sentry/fsimpl/fuse/connection.go +++ b/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() diff --git a/pkg/sentry/fsimpl/fuse/connection_control.go b/pkg/sentry/fsimpl/fuse/connection_control.go index 948502db6..498fd1b54 100644 --- a/pkg/sentry/fsimpl/fuse/connection_control.go +++ b/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() diff --git a/pkg/sentry/fsimpl/fuse/dev.go b/pkg/sentry/fsimpl/fuse/dev.go index 55d86b933..05f2a44f9 100644 --- a/pkg/sentry/fsimpl/fuse/dev.go +++ b/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 } diff --git a/pkg/sentry/fsimpl/fuse/dev_test.go b/pkg/sentry/fsimpl/fuse/dev_test.go index 6e9dfe5ef..cf494d2f8 100644 --- a/pkg/sentry/fsimpl/fuse/dev_test.go +++ b/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()) diff --git a/pkg/sentry/fsimpl/fuse/fusefs.go b/pkg/sentry/fsimpl/fuse/fusefs.go index 039dc6996..9e0783b6f 100644 --- a/pkg/sentry/fsimpl/fuse/fusefs.go +++ b/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 }