diff options
| -rw-r--r-- | pkg/tcpip/transport/queue/queue.go | 2 | ||||
| -rw-r--r-- | pkg/tcpip/transport/unix/unix.go | 12 |
2 files changed, 12 insertions, 2 deletions
diff --git a/pkg/tcpip/transport/queue/queue.go b/pkg/tcpip/transport/queue/queue.go index 2d2918504..0c90588ae 100644 --- a/pkg/tcpip/transport/queue/queue.go +++ b/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 } diff --git a/pkg/tcpip/transport/unix/unix.go b/pkg/tcpip/transport/unix/unix.go index 72c21a432..34bdb5877 100644 --- a/pkg/tcpip/transport/unix/unix.go +++ b/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. |