diff --git a/pkg/sentry/socket/unix/transport/connectioned.go b/pkg/sentry/socket/unix/transport/connectioned.go index c8bdcafa4..4c57804f2 100644 --- a/pkg/sentry/socket/unix/transport/connectioned.go +++ b/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 diff --git a/pkg/sentry/socket/unix/transport/host.go b/pkg/sentry/socket/unix/transport/host.go index 571089413..a80bf10f1 100644 --- a/pkg/sentry/socket/unix/transport/host.go +++ b/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. diff --git a/pkg/sentry/socket/unix/transport/queue.go b/pkg/sentry/socket/unix/transport/queue.go index 81beb7362..5a86e643e 100644 --- a/pkg/sentry/socket/unix/transport/queue.go +++ b/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 } diff --git a/pkg/sentry/socket/unix/transport/unix.go b/pkg/sentry/socket/unix/transport/unix.go index 4fb430daf..0fb070f00 100644 --- a/pkg/sentry/socket/unix/transport/unix.go +++ b/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()