Fix recv blocking for connectionless Unix sockets.

Connectionless Unix sockets (DGRAM Unix sockets created with the socket system
call) inherently only have a read queue. They do not establish bidirectional
connections, instead, the connect system call only sets a default send
location. Writes give the data to the other endpoint which has its own read
queue.

To simplify the code, connectionless Unix sockets still get read and write
queues, but the write queue is a dummy and never waited on. The read queue is
the connectionless endpoint's queue. This change fixes a bug where the dummy
queue was incorrectly set as the read queue and the endpoint's queue was
incorrectly set as the write queue. This meant that read notifications went
to the dummy queue and were black holed.

PiperOrigin-RevId: 225921042
Change-Id: I8d9059def787a2c3c305185b92d05093fbd2be2a

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

@@ -135,8 +135,8 @@ func NewPair(stype SockType, uid UniqueIDProvider) (Endpoint, Endpoint) {
 		stype:        stype,
 	}
 
-	q1 := newQueue(a.Queue, b.Queue, initialLimit)
-	q2 := newQueue(b.Queue, a.Queue, initialLimit)
+	q1 := &queue{ReaderQueue: a.Queue, WriterQueue: b.Queue, limit: initialLimit}
+	q2 := &queue{ReaderQueue: b.Queue, WriterQueue: a.Queue, limit: initialLimit}
 
 	if stype == SockStream {
 		a.receiver = &streamQueueReceiver{queueReceiver: queueReceiver{q1}}
@@ -286,13 +286,13 @@ func (e *connectionedEndpoint) BidirectionalConnect(ce ConnectingEndpoint, retur
 		stype:       e.stype,
 	}
 
-	readQueue := newQueue(ce.WaiterQueue(), ne.Queue, initialLimit)
+	readQueue := &queue{ReaderQueue: ce.WaiterQueue(), WriterQueue: ne.Queue, limit: initialLimit}
 	ne.connected = &connectedEndpoint{
 		endpoint:   ce,
 		writeQueue: readQueue,
 	}
 
-	writeQueue := newQueue(ne.Queue, ce.WaiterQueue(), initialLimit)
+	writeQueue := &queue{ReaderQueue: ne.Queue, WriterQueue: ce.WaiterQueue(), limit: initialLimit}
 	if e.stype == SockStream {
 		ne.receiver = &streamQueueReceiver{queueReceiver: queueReceiver{readQueue: writeQueue}}
 	} else {

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

@@ -34,7 +34,7 @@ type connectionlessEndpoint struct {
 // NewConnectionless creates a new unbound dgram endpoint.
 func NewConnectionless() Endpoint {
 	ep := &connectionlessEndpoint{baseEndpoint{Queue: &waiter.Queue{}}}
-	ep.receiver = &queueReceiver{readQueue: newQueue(&waiter.Queue{}, ep.Queue, initialLimit)}
+	ep.receiver = &queueReceiver{readQueue: &queue{ReaderQueue: ep.Queue, WriterQueue: &waiter.Queue{}, limit: initialLimit}}
 	return ep
 }
 

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

@@ -38,11 +38,6 @@ type queue struct {
 	dataList messageList
 }
 
-// newQueue allocates and initializes a new queue.
-func newQueue(ReaderQueue *waiter.Queue, WriterQueue *waiter.Queue, limit int64) *queue {
-	return &queue{ReaderQueue: ReaderQueue, WriterQueue: WriterQueue, limit: limit}
-}
-
 // Close closes q for reading and writing. It is immediately not writable and
 // will become unreadable when no more data is pending.
 //

test/syscalls/linux/socket_unix_blocking_local.cc

@@ -33,14 +33,12 @@ std::vector<SocketPairKind> GetSocketPairs() {
       ApplyVec<SocketPairKind>(
           FilesystemBoundUnixDomainSocketPair,
           AllBitwiseCombinations(
-              // FIXME: Add SOCK_DGRAM once blocking is fixed.
-              List<int>{SOCK_STREAM, SOCK_SEQPACKET},
+              List<int>{SOCK_STREAM, SOCK_SEQPACKET, SOCK_DGRAM},
               List<int>{0, SOCK_CLOEXEC})),
       ApplyVec<SocketPairKind>(
           AbstractBoundUnixDomainSocketPair,
           AllBitwiseCombinations(
-              // FIXME: Add SOCK_DGRAM once blocking is fixed.
-              List<int>{SOCK_STREAM, SOCK_SEQPACKET},
+              List<int>{SOCK_STREAM, SOCK_SEQPACKET, SOCK_DGRAM},
               List<int>{0, SOCK_CLOEXEC})));
 }