Fix data races in Unix sockets

PiperOrigin-RevId: 202175558
Change-Id: I0113cb9a90d7a0cd7964bf43eef67f70c92d9589

pkg/tcpip/transport/queue/queue.go

@@ -157,6 +157,8 @@ func (q *Queue) Peek() (Entry, *tcpip.Error) {
 // QueuedSize returns the number of bytes currently in the queue, that is, the
 // number of readable bytes.
 func (q *Queue) QueuedSize() int64 {
+	q.mu.Lock()
+	defer q.mu.Unlock()
 	return q.used
 }
 

pkg/tcpip/transport/unix/unix.go

@@ -384,14 +384,22 @@ func vecCopy(data [][]byte, buf []byte) (uintptr, [][]byte, []byte) {
 
 // Readable implements Receiver.Readable.
 func (q *streamQueueReceiver) Readable() bool {
+	q.mu.Lock()
+	bl := len(q.buffer)
+	r := q.readQueue.IsReadable()
+	q.mu.Unlock()
 	// We're readable if we have data in our buffer or if the queue receiver is
 	// readable.
-	return len(q.buffer) > 0 || q.readQueue.IsReadable()
+	return bl > 0 || r
 }
 
 // RecvQueuedSize implements Receiver.RecvQueuedSize.
 func (q *streamQueueReceiver) RecvQueuedSize() int64 {
-	return int64(len(q.buffer)) + q.readQueue.QueuedSize()
+	q.mu.Lock()
+	bl := len(q.buffer)
+	qs := q.readQueue.QueuedSize()
+	q.mu.Unlock()
+	return int64(bl) + qs
 }
 
 // RecvMaxQueueSize implements Receiver.RecvMaxQueueSize.