Fix connectionedEndpoint.Readiness() to return EPOLL{RD}HUP appropriately.

Compare net/unix/af_unix.c:unix_poll().

Fixes #7331
Updates dart-lang/sdk#47899
Updates GoogleCloudPlatform/functions-framework-dart#302

PiperOrigin-RevId: 597625614

pkg/sentry/socket/unix/transport/connectioned.go

@@ -561,6 +561,12 @@ func (e *connectionedEndpoint) Readiness(mask waiter.EventMask) waiter.EventMask
 		if mask&waiter.WritableEvents != 0 && e.connected.Writable() {
 			ready |= waiter.WritableEvents
 		}
+		if mask&(waiter.EventHUp|waiter.EventRdHUp) != 0 && e.receiver.IsRecvClosed() {
+			ready |= waiter.EventRdHUp
+			if mask&waiter.EventHUp != 0 && e.connected.IsSendClosed() {
+				ready |= waiter.EventHUp
+			}
+		}
 	case e.ListeningLocked():
 		if mask&waiter.ReadableEvents != 0 && (len(e.acceptedChan) > 0 || e.isBoundSocketReadable()) {
 			ready |= waiter.ReadableEvents

pkg/sentry/socket/unix/transport/host.go

@@ -82,6 +82,12 @@ type HostConnectedEndpoint struct {
 
 	// stype is the type of Unix socket.
 	stype linux.SockType
+
+	// rdShutdown is true if receptions have been shutdown with SHUT_RD.
+	rdShutdown atomicbitops.Bool
+
+	// wrShutdown is true if transmissions have been shutdown with SHUT_WR.
+	wrShutdown atomicbitops.Bool
 }
 
 // init performs initialization required for creating new
@@ -192,11 +198,17 @@ func (c *HostConnectedEndpoint) CloseSend() {
 		// net/unix/af_unix.c:unix_shutdown.
 		panic(fmt.Sprintf("failed write shutdown on host socket %+v: %v", c, err))
 	}
+	c.wrShutdown.Store(true)
 }
 
 // CloseNotify implements ConnectedEndpoint.CloseNotify.
 func (c *HostConnectedEndpoint) CloseNotify() {}
 
+// IsSendClosed implements ConnectedEndpoint.IsSendClosed.
+func (c *HostConnectedEndpoint) IsSendClosed() bool {
+	return c.wrShutdown.Load()
+}
+
 // Writable implements ConnectedEndpoint.Writable.
 func (c *HostConnectedEndpoint) Writable() bool {
 	c.mu.RLock()
@@ -294,6 +306,12 @@ func (c *HostConnectedEndpoint) CloseRecv() {
 		// net/unix/af_unix.c:unix_shutdown.
 		panic(fmt.Sprintf("failed read shutdown on host socket %+v: %v", c, err))
 	}
+	c.rdShutdown.Store(true)
+}
+
+// IsRecvClosed implements Receiver.IsRecvClosed.
+func (c *HostConnectedEndpoint) IsRecvClosed() bool {
+	return c.rdShutdown.Load()
 }
 
 // Readable implements Receiver.Readable.

pkg/sentry/socket/unix/transport/queue.go

@@ -15,6 +15,7 @@
 package transport
 
 import (
+	"gvisor.dev/gvisor/pkg/atomicbitops"
 	"gvisor.dev/gvisor/pkg/context"
 	"gvisor.dev/gvisor/pkg/syserr"
 	"gvisor.dev/gvisor/pkg/waiter"
@@ -30,7 +31,7 @@ type queue struct {
 	WriterQueue *waiter.Queue
 
 	mu       queueMutex `state:"nosave"`
-	closed   bool
+	closed   atomicbitops.Bool
 	unread   bool
 	used     int64
 	limit    int64
@@ -45,10 +46,14 @@ type queue struct {
 // q.WriterQueue.Notify(waiter.WritableEvents)
 func (q *queue) Close() {
 	q.mu.Lock()
-	q.closed = true
+	q.closed.Store(true)
 	q.mu.Unlock()
 }
 
+func (q *queue) isClosed() bool {
+	return q.closed.Load()
+}
+
 // Reset empties the queue and Releases all of the Entries.
 //
 // Both the read and write queues must be notified after resetting:
@@ -80,7 +85,7 @@ func (q *queue) IsReadable() bool {
 	q.mu.Lock()
 	defer q.mu.Unlock()
 
-	return q.closed || q.dataList.Front() != nil
+	return q.closed.RacyLoad() || q.dataList.Front() != nil
 }
 
 // bufWritable returns true if there is space for writing.
@@ -98,7 +103,7 @@ func (q *queue) IsWritable() bool {
 	q.mu.Lock()
 	defer q.mu.Unlock()
 
-	return q.closed || q.bufWritable()
+	return q.closed.RacyLoad() || q.bufWritable()
 }
 
 // Enqueue adds an entry to the data queue if room is available.
@@ -115,7 +120,7 @@ func (q *queue) IsWritable() bool {
 func (q *queue) Enqueue(ctx context.Context, data [][]byte, c ControlMessages, from Address, discardEmpty bool, truncate bool) (l int64, notify bool, err *syserr.Error) {
 	q.mu.Lock()
 
-	if q.closed {
+	if q.closed.RacyLoad() {
 		q.mu.Unlock()
 		return 0, false, syserr.ErrClosedForSend
 	}
@@ -184,7 +189,7 @@ func (q *queue) Dequeue() (e *message, notify bool, err *syserr.Error) {
 
 	if q.dataList.Front() == nil {
 		err := syserr.ErrWouldBlock
-		if q.closed {
+		if q.closed.RacyLoad() {
 			err = syserr.ErrClosedForReceive
 			if q.unread {
 				err = syserr.ErrConnectionReset
@@ -215,7 +220,7 @@ func (q *queue) Peek() (*message, *syserr.Error) {
 
 	if q.dataList.Front() == nil {
 		err := syserr.ErrWouldBlock
-		if q.closed {
+		if q.closed.RacyLoad() {
 			if err = syserr.ErrClosedForReceive; q.unread {
 				err = syserr.ErrConnectionReset
 			}

pkg/sentry/socket/unix/transport/unix.go

@@ -378,6 +378,9 @@ type Receiver interface {
 	// called while holding any endpoint locks.
 	CloseNotify()
 
+	// IsRecvClosed returns true if reception of additional messages is closed.
+	IsRecvClosed() bool
+
 	// Readable returns if messages should be attempted to be received. This
 	// includes when read has been shutdown.
 	Readable() bool
@@ -454,6 +457,11 @@ func (q *queueReceiver) CloseRecv() {
 	q.readQueue.Close()
 }
 
+// IsRecvClosed implements Receiver.IsRecvClosed.
+func (q *queueReceiver) IsRecvClosed() bool {
+	return q.readQueue.isClosed()
+}
+
 // Readable implements Receiver.Readable.
 func (q *queueReceiver) Readable() bool {
 	return q.readQueue.IsReadable()
@@ -687,6 +695,9 @@ type ConnectedEndpoint interface {
 	// must not be called while holding any endpoint locks.
 	CloseNotify()
 
+	// IsSendClosed returns true if transmission of additional messages is closed.
+	IsSendClosed() bool
+
 	// Writable returns if messages should be attempted to be sent. This
 	// includes when write has been shutdown.
 	Writable() bool
@@ -782,6 +793,11 @@ func (e *connectedEndpoint) CloseSend() {
 	e.writeQueue.Close()
 }
 
+// IsSendClosed implements ConnectedEndpoint.IsSendClosed.
+func (e *connectedEndpoint) IsSendClosed() bool {
+	return e.writeQueue.isClosed()
+}
+
 // Writable implements ConnectedEndpoint.Writable.
 func (e *connectedEndpoint) Writable() bool {
 	return e.writeQueue.IsWritable()