pipe: have separate notifiers for readers and writers

This change fixes a busy loop in the pipe code. VFSPipe.Open calls ctx.BlockOn
to wait an opposite side, but waitQueue.EventRegister always triggers
EventInternal, so we never block.

Reported-by: syzbot+773e19ca2574516c9e00@syzkaller.appspotmail.com
PiperOrigin-RevId: 415428542

pkg/sentry/kernel/pipe/node.go

@@ -75,7 +75,7 @@ func (i *inodeOperations) GetFile(ctx context.Context, d *fs.Dirent, flags fs.Fi
 	case flags.Read && !flags.Write: // O_RDONLY.
 		r := i.p.Open(ctx, d, flags)
 		for i.p.isNamed && !flags.NonBlocking && !i.p.HasWriters() {
-			if !ctx.BlockOn((*waitQueue)(i.p), waiter.EventInternal) {
+			if !ctx.BlockOn((*waitWriters)(i.p), waiter.EventInternal) {
 				r.DecRef(ctx)
 				return nil, linuxerr.ErrInterrupted
 			}
@@ -95,7 +95,7 @@ func (i *inodeOperations) GetFile(ctx context.Context, d *fs.Dirent, flags fs.Fi
 				w.DecRef(ctx)
 				return nil, linuxerr.ENXIO
 			}
-			if !ctx.BlockOn((*waitQueue)(i.p), waiter.EventInternal) {
+			if !ctx.BlockOn((*waitReaders)(i.p), waiter.EventInternal) {
 				w.DecRef(ctx)
 				return nil, linuxerr.ErrInterrupted
 			}

pkg/sentry/kernel/pipe/pipe.go

@@ -49,23 +49,23 @@ const (
 	atomicIOBytes = 4096
 )
 
-// waitQueue is a wrapper around Pipe.
+// waitReaders is a wrapper around Pipe.
 //
 // This is used for ctx.Block operations that require the synchronization of
 // readers and writers, along with the careful grabbing and releasing of locks.
-type waitQueue Pipe
+type waitReaders Pipe
 
 // Readiness implements waiter.Waitable.Readiness.
-func (wq *waitQueue) Readiness(mask waiter.EventMask) waiter.EventMask {
+func (wq *waitReaders) Readiness(mask waiter.EventMask) waiter.EventMask {
 	return ((*Pipe)(wq)).rwReadiness() & mask
 }
 
 // EventRegister implements waiter.Waitable.EventRegister.
-func (wq *waitQueue) EventRegister(e *waiter.Entry) error {
+func (wq *waitReaders) EventRegister(e *waiter.Entry) error {
 	((*Pipe)(wq)).queue.EventRegister(e)
 
 	// Notify synchronously.
-	if ((*Pipe)(wq)).HasReaders() || ((*Pipe)(wq)).HasWriters() {
+	if ((*Pipe)(wq)).HasReaders() {
 		e.NotifyEvent(waiter.EventInternal)
 	}
 
@@ -73,7 +73,35 @@ func (wq *waitQueue) EventRegister(e *waiter.Entry) error {
 }
 
 // EventUnregister implements waiter.Waitable.EventUnregister.
-func (wq *waitQueue) EventUnregister(e *waiter.Entry) {
+func (wq *waitReaders) EventUnregister(e *waiter.Entry) {
+	((*Pipe)(wq)).queue.EventUnregister(e)
+}
+
+// waitWriters is a wrapper around Pipe.
+//
+// This is used for ctx.Block operations that require the synchronization of
+// readers and writers, along with the careful grabbing and releasing of locks.
+type waitWriters Pipe
+
+// Readiness implements waiter.Waitable.Readiness.
+func (wq *waitWriters) Readiness(mask waiter.EventMask) waiter.EventMask {
+	return ((*Pipe)(wq)).rwReadiness() & mask
+}
+
+// EventRegister implements waiter.Waitable.EventRegister.
+func (wq *waitWriters) EventRegister(e *waiter.Entry) error {
+	((*Pipe)(wq)).queue.EventRegister(e)
+
+	// Notify synchronously.
+	if ((*Pipe)(wq)).HasWriters() {
+		e.NotifyEvent(waiter.EventInternal)
+	}
+
+	return nil
+}
+
+// EventUnregister implements waiter.Waitable.EventUnregister.
+func (wq *waitWriters) EventUnregister(e *waiter.Entry) {
 	((*Pipe)(wq)).queue.EventUnregister(e)
 }
 

pkg/sentry/kernel/pipe/vfs.go

@@ -99,7 +99,7 @@ func (vp *VFSPipe) Open(ctx context.Context, mnt *vfs.Mount, vfsd *vfs.Dentry, s
 		// 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() {
-			if !ctx.BlockOn((*waitQueue)(&vp.pipe), waiter.EventInternal) {
+			if !ctx.BlockOn((*waitWriters)(&vp.pipe), waiter.EventInternal) {
 				fd.DecRef(ctx)
 				return nil, linuxerr.EINTR
 			}
@@ -113,7 +113,7 @@ func (vp *VFSPipe) Open(ctx context.Context, mnt *vfs.Mount, vfsd *vfs.Dentry, s
 				fd.DecRef(ctx)
 				return nil, linuxerr.ENXIO
 			}
-			if !ctx.BlockOn((*waitQueue)(&vp.pipe), waiter.EventInternal) {
+			if !ctx.BlockOn((*waitReaders)(&vp.pipe), waiter.EventInternal) {
 				fd.DecRef(ctx)
 				return nil, linuxerr.EINTR
 			}