From b76119a1e786b51d012d358ba8711fc9d709e046 Mon Sep 17 00:00:00 2001 From: Andrei Vagin Date: Tue, 21 Dec 2021 10:16:59 -0800 Subject: [PATCH] pipe: a reader has to wait when all writers will be notified Otherwise, we can have a race when a reader cloes a pipe before a write detects this reader. PiperOrigin-RevId: 417645683 --- pkg/sentry/kernel/pipe/node.go | 10 ++++++++-- pkg/sentry/kernel/pipe/pipe.go | 21 ++++++++++++++++++++- pkg/sentry/kernel/pipe/vfs.go | 28 ++++++++++++++-------------- 3 files changed, 42 insertions(+), 17 deletions(-) diff --git a/pkg/sentry/kernel/pipe/node.go b/pkg/sentry/kernel/pipe/node.go index 91892e2d4..5581e1eb9 100644 --- a/pkg/sentry/kernel/pipe/node.go +++ b/pkg/sentry/kernel/pipe/node.go @@ -15,6 +15,8 @@ package pipe import ( + "sync/atomic" + "gvisor.dev/gvisor/pkg/abi/linux" "gvisor.dev/gvisor/pkg/context" "gvisor.dev/gvisor/pkg/errors/linuxerr" @@ -73,8 +75,10 @@ func NewInodeOperations(ctx context.Context, perms fs.FilePermissions, p *Pipe) func (i *inodeOperations) GetFile(ctx context.Context, d *fs.Dirent, flags fs.FileFlags) (*fs.File, error) { switch { case flags.Read && !flags.Write: // O_RDONLY. + tWriters := atomic.LoadInt32(&i.p.totalWriters) r := i.p.Open(ctx, d, flags) - for i.p.isNamed && !flags.NonBlocking && !i.p.HasWriters() { + for i.p.isNamed && !flags.NonBlocking && !i.p.HasWriters() && + tWriters == atomic.LoadInt32(&i.p.totalWriters) { if !ctx.BlockOn((*waitWriters)(i.p), waiter.EventInternal) { r.DecRef(ctx) return nil, linuxerr.ErrInterrupted @@ -87,8 +91,10 @@ func (i *inodeOperations) GetFile(ctx context.Context, d *fs.Dirent, flags fs.Fi return r, nil case flags.Write && !flags.Read: // O_WRONLY. + tReaders := atomic.LoadInt32(&i.p.totalReaders) w := i.p.Open(ctx, d, flags) - for i.p.isNamed && !i.p.HasReaders() { + for i.p.isNamed && !i.p.HasReaders() && + tReaders == atomic.LoadInt32(&i.p.totalReaders) { // On a nonblocking, write-only open, the open fails with ENXIO if the // read side isn't open yet. if flags.NonBlocking { diff --git a/pkg/sentry/kernel/pipe/pipe.go b/pkg/sentry/kernel/pipe/pipe.go index ccaad6319..ec7e5af39 100644 --- a/pkg/sentry/kernel/pipe/pipe.go +++ b/pkg/sentry/kernel/pipe/pipe.go @@ -124,11 +124,21 @@ type Pipe struct { // Access atomically. readers int32 - // The number of active writes for this pipe. + // The total number of readers for this pipe. + // + // Access atomically. + totalReaders int32 + + // The number of active writers for this pipe. // // Access atomically. writers int32 + // The total number of writers for this pipe. + // + // Access atomically. + totalWriters int32 + // mu protects all pipe internal state below. mu sync.Mutex `state:"nosave"` @@ -158,6 +168,13 @@ type Pipe struct { // // This is protected by mu. hadWriter bool + + // waitingWriters is used to wait when writers are initialized after a + // reader has opened the pipe. + waitingWriters sync.WaitGroup `state:"nosave"` + // waitingReaders is used to wait when readers are initialized after a + // write has opened the pipe. + waitingReaders sync.WaitGroup `state:"nosave"` } // NewPipe initializes and returns a pipe. @@ -373,6 +390,7 @@ func (p *Pipe) writeLocked(count int64, f func(safemem.BlockSeq) (uint64, error) // rOpen signals a new reader of the pipe. func (p *Pipe) rOpen() { atomic.AddInt32(&p.readers, 1) + atomic.AddInt32(&p.totalReaders, 1) // Notify for blocking openers. p.queue.Notify(waiter.EventInternal) @@ -383,6 +401,7 @@ func (p *Pipe) wOpen() { p.mu.Lock() p.hadWriter = true atomic.AddInt32(&p.writers, 1) + atomic.AddInt32(&p.totalWriters, 1) p.mu.Unlock() // Notify for blocking openers. diff --git a/pkg/sentry/kernel/pipe/vfs.go b/pkg/sentry/kernel/pipe/vfs.go index 9cb7f00d0..be59e10ef 100644 --- a/pkg/sentry/kernel/pipe/vfs.go +++ b/pkg/sentry/kernel/pipe/vfs.go @@ -15,6 +15,8 @@ package pipe import ( + "sync/atomic" + "gvisor.dev/gvisor/pkg/abi/linux" "gvisor.dev/gvisor/pkg/context" "gvisor.dev/gvisor/pkg/errors/linuxerr" @@ -55,11 +57,13 @@ func (vp *VFSPipe) ReaderWriterPair(ctx context.Context, mnt *vfs.Mount, vfsd *v if err != nil { return nil, nil, err } + vp.pipe.rOpen() w, err := vp.newFD(mnt, vfsd, linux.O_WRONLY|statusFlags, locks) if err != nil { r.DecRef(ctx) return nil, nil, err } + vp.pipe.wOpen() return r, w, nil } @@ -93,12 +97,17 @@ func (vp *VFSPipe) Open(ctx context.Context, mnt *vfs.Mount, vfsd *vfs.Dentry, s // FIFO for writing while there are no readers available." - fifo(7) switch { case readable && writable: + vp.pipe.rOpen() + vp.pipe.wOpen() // Pipes opened for read-write always succeed without blocking. case readable: + tWriters := atomic.LoadInt32(&vp.pipe.totalWriters) + vp.pipe.rOpen() // If this pipe is being opened as blocking and there's no // writer, we have to wait for a writer to open the other end. - for vp.pipe.isNamed && statusFlags&linux.O_NONBLOCK == 0 && !vp.pipe.HasWriters() { + for vp.pipe.isNamed && statusFlags&linux.O_NONBLOCK == 0 && !vp.pipe.HasWriters() && + tWriters == atomic.LoadInt32(&vp.pipe.totalWriters) { if !ctx.BlockOn((*waitWriters)(&vp.pipe), waiter.EventInternal) { fd.DecRef(ctx) return nil, linuxerr.EINTR @@ -106,7 +115,10 @@ func (vp *VFSPipe) Open(ctx context.Context, mnt *vfs.Mount, vfsd *vfs.Dentry, s } case writable: - for vp.pipe.isNamed && !vp.pipe.HasReaders() { + tReaders := atomic.LoadInt32(&vp.pipe.totalReaders) + vp.pipe.wOpen() + for vp.pipe.isNamed && !vp.pipe.HasReaders() && + tReaders == atomic.LoadInt32(&vp.pipe.totalReaders) { // Non-blocking, write-only opens fail with ENXIO when the read // side isn't open yet. if statusFlags&linux.O_NONBLOCK != 0 { @@ -140,18 +152,6 @@ func (vp *VFSPipe) newFD(mnt *vfs.Mount, vfsd *vfs.Dentry, statusFlags uint32, l return nil, err } - switch { - case fd.vfsfd.IsReadable() && fd.vfsfd.IsWritable(): - vp.pipe.rOpen() - vp.pipe.wOpen() - case fd.vfsfd.IsReadable(): - vp.pipe.rOpen() - case fd.vfsfd.IsWritable(): - vp.pipe.wOpen() - default: - panic("invalid pipe flags: must be readable, writable, or both") - } - return &fd.vfsfd, nil }