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

@@ -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 {

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.

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
 }