Merge queue into Unix transport

This queue only has a single user, so there is no need for it to use an
interface. Merging it into the same package as its sole user allows us to avoid
a circular dependency.

This simplifies the code and should slightly improve performance.

PiperOrigin-RevId: 217595889
Change-Id: Iabbd5164240b935f79933618c61581bc8dcd2822

6 files changed, 37 insertions, 79 deletions

diff --git a/pkg/sentry/socket/unix/transport/BUILD b/pkg/sentry/socket/unix/transport/BUILD
index 04ef0d438..75b5a2eb6 100644
--- a/pkg/sentry/socket/unix/transport/BUILD
+++ b/pkg/sentry/socket/unix/transport/BUILD
@@ -8,13 +8,13 @@ go_library(
         "connectioned.go",
         "connectioned_state.go",
         "connectionless.go",
+        "queue.go",
         "unix.go",
     ],
     importpath = "gvisor.googlesource.com/gvisor/pkg/sentry/socket/unix/transport",
     visibility = ["//:sandbox"],
     deps = [
         "//pkg/ilist",
-        "//pkg/sentry/socket/unix/transport/queue",
         "//pkg/tcpip",
         "//pkg/tcpip/buffer",
         "//pkg/waiter",
diff --git a/pkg/sentry/socket/unix/transport/connectioned.go b/pkg/sentry/socket/unix/transport/connectioned.go
index f09935765..566e3d57b 100644
--- a/pkg/sentry/socket/unix/transport/connectioned.go
+++ b/pkg/sentry/socket/unix/transport/connectioned.go
@@ -17,7 +17,6 @@ package transport
 import (
 	"sync"
 
-	"gvisor.googlesource.com/gvisor/pkg/sentry/socket/unix/transport/queue"
 	"gvisor.googlesource.com/gvisor/pkg/tcpip"
 	"gvisor.googlesource.com/gvisor/pkg/waiter"
 )
@@ -135,8 +134,8 @@ func NewPair(stype SockType, uid UniqueIDProvider) (Endpoint, Endpoint) {
 		stype:        stype,
 	}
 
-	q1 := queue.New(a.Queue, b.Queue, initialLimit)
-	q2 := queue.New(b.Queue, a.Queue, initialLimit)
+	q1 := newQueue(a.Queue, b.Queue, initialLimit)
+	q2 := newQueue(b.Queue, a.Queue, initialLimit)
 
 	if stype == SockStream {
 		a.receiver = &streamQueueReceiver{queueReceiver: queueReceiver{q1}}
@@ -283,8 +282,8 @@ func (e *connectionedEndpoint) BidirectionalConnect(ce ConnectingEndpoint, retur
 		idGenerator: e.idGenerator,
 		stype:       e.stype,
 	}
-	readQueue := queue.New(ce.WaiterQueue(), ne.Queue, initialLimit)
-	writeQueue := queue.New(ne.Queue, ce.WaiterQueue(), initialLimit)
+	readQueue := newQueue(ce.WaiterQueue(), ne.Queue, initialLimit)
+	writeQueue := newQueue(ne.Queue, ce.WaiterQueue(), initialLimit)
 	ne.connected = &connectedEndpoint{
 		endpoint:   ce,
 		writeQueue: readQueue,
diff --git a/pkg/sentry/socket/unix/transport/connectionless.go b/pkg/sentry/socket/unix/transport/connectionless.go
index fb2728010..86cd05199 100644
--- a/pkg/sentry/socket/unix/transport/connectionless.go
+++ b/pkg/sentry/socket/unix/transport/connectionless.go
@@ -15,7 +15,6 @@
 package transport
 
 import (
-	"gvisor.googlesource.com/gvisor/pkg/sentry/socket/unix/transport/queue"
 	"gvisor.googlesource.com/gvisor/pkg/tcpip"
 	"gvisor.googlesource.com/gvisor/pkg/waiter"
 )
@@ -34,7 +33,7 @@ type connectionlessEndpoint struct {
 // NewConnectionless creates a new unbound dgram endpoint.
 func NewConnectionless() Endpoint {
 	ep := &connectionlessEndpoint{baseEndpoint{Queue: &waiter.Queue{}}}
-	ep.receiver = &queueReceiver{readQueue: queue.New(&waiter.Queue{}, ep.Queue, initialLimit)}
+	ep.receiver = &queueReceiver{readQueue: newQueue(&waiter.Queue{}, ep.Queue, initialLimit)}
 	return ep
 }
 
diff --git a/pkg/sentry/socket/unix/transport/queue/queue.go b/pkg/sentry/socket/unix/transport/queue.go
index b3d2ea68b..203e31333 100644
--- a/pkg/sentry/socket/unix/transport/queue/queue.go
+++ b/pkg/sentry/socket/unix/transport/queue.go
@@ -12,9 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-// Package queue provides the implementation of buffer queue
-// and interface of queue entry with Length method.
-package queue
+package transport
 
 import (
 	"sync"
@@ -24,29 +22,10 @@ import (
 	"gvisor.googlesource.com/gvisor/pkg/waiter"
 )
 
-// Entry implements Linker interface and has additional required methods.
-type Entry interface {
-	ilist.Linker
-
-	// Length returns the number of bytes stored in the entry.
-	Length() int64
-
-	// Release releases any resources held by the entry.
-	Release()
-
-	// Peek returns a copy of the entry. It must be Released separately.
-	Peek() Entry
-
-	// Truncate reduces the number of bytes stored in the entry to n bytes.
-	//
-	// Preconditions: n <= Length().
-	Truncate(n int64)
-}
-
-// Queue is a buffer queue.
+// queue is a buffer queue.
 //
 // +stateify savable
-type Queue struct {
+type queue struct {
 	ReaderQueue *waiter.Queue
 	WriterQueue *waiter.Queue
 
@@ -57,9 +36,9 @@ type Queue struct {
 	dataList ilist.List
 }
 
-// New allocates and initializes a new queue.
-func New(ReaderQueue *waiter.Queue, WriterQueue *waiter.Queue, limit int64) *Queue {
-	return &Queue{ReaderQueue: ReaderQueue, WriterQueue: WriterQueue, limit: limit}
+// 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
@@ -68,7 +47,7 @@ func New(ReaderQueue *waiter.Queue, WriterQueue *waiter.Queue, limit int64) *Que
 // Both the read and write queues must be notified after closing:
 // q.ReaderQueue.Notify(waiter.EventIn)
 // q.WriterQueue.Notify(waiter.EventOut)
-func (q *Queue) Close() {
+func (q *queue) Close() {
 	q.mu.Lock()
 	q.closed = true
 	q.mu.Unlock()
@@ -79,10 +58,10 @@ func (q *Queue) Close() {
 // Both the read and write queues must be notified after resetting:
 // q.ReaderQueue.Notify(waiter.EventIn)
 // q.WriterQueue.Notify(waiter.EventOut)
-func (q *Queue) Reset() {
+func (q *queue) Reset() {
 	q.mu.Lock()
 	for cur := q.dataList.Front(); cur != nil; cur = cur.Next() {
-		cur.(Entry).Release()
+		cur.(*message).Release()
 	}
 	q.dataList.Reset()
 	q.used = 0
@@ -90,7 +69,7 @@ func (q *Queue) Reset() {
 }
 
 // IsReadable determines if q is currently readable.
-func (q *Queue) IsReadable() bool {
+func (q *queue) IsReadable() bool {
 	q.mu.Lock()
 	defer q.mu.Unlock()
 
@@ -103,12 +82,12 @@ func (q *Queue) IsReadable() bool {
 // free.
 //
 // See net/unix/af_unix.c:unix_writeable.
-func (q *Queue) bufWritable() bool {
+func (q *queue) bufWritable() bool {
 	return 4*q.used < q.limit
 }
 
 // IsWritable determines if q is currently writable.
-func (q *Queue) IsWritable() bool {
+func (q *queue) IsWritable() bool {
 	q.mu.Lock()
 	defer q.mu.Unlock()
 
@@ -122,7 +101,7 @@ func (q *Queue) IsWritable() bool {
 //
 // If notify is true, ReaderQueue.Notify must be called:
 // q.ReaderQueue.Notify(waiter.EventIn)
-func (q *Queue) Enqueue(e Entry, truncate bool) (l int64, notify bool, err *tcpip.Error) {
+func (q *queue) Enqueue(e *message, truncate bool) (l int64, notify bool, err *tcpip.Error) {
 	q.mu.Lock()
 
 	if q.closed {
@@ -171,7 +150,7 @@ func (q *Queue) Enqueue(e Entry, truncate bool) (l int64, notify bool, err *tcpi
 //
 // If notify is true, WriterQueue.Notify must be called:
 // q.WriterQueue.Notify(waiter.EventOut)
-func (q *Queue) Dequeue() (e Entry, notify bool, err *tcpip.Error) {
+func (q *queue) Dequeue() (e *message, notify bool, err *tcpip.Error) {
 	q.mu.Lock()
 
 	if q.dataList.Front() == nil {
@@ -186,7 +165,7 @@ func (q *Queue) Dequeue() (e Entry, notify bool, err *tcpip.Error) {
 
 	notify = !q.bufWritable()
 
-	e = q.dataList.Front().(Entry)
+	e = q.dataList.Front().(*message)
 	q.dataList.Remove(e)
 	q.used -= e.Length()
 
@@ -198,7 +177,7 @@ func (q *Queue) Dequeue() (e Entry, notify bool, err *tcpip.Error) {
 }
 
 // Peek returns the first entry in the data queue, if one exists.
-func (q *Queue) Peek() (Entry, *tcpip.Error) {
+func (q *queue) Peek() (*message, *tcpip.Error) {
 	q.mu.Lock()
 	defer q.mu.Unlock()
 
@@ -210,18 +189,18 @@ func (q *Queue) Peek() (Entry, *tcpip.Error) {
 		return nil, err
 	}
 
-	return q.dataList.Front().(Entry).Peek(), nil
+	return q.dataList.Front().(*message).Peek(), nil
 }
 
 // QueuedSize returns the number of bytes currently in the queue, that is, the
 // number of readable bytes.
-func (q *Queue) QueuedSize() int64 {
+func (q *queue) QueuedSize() int64 {
 	q.mu.Lock()
 	defer q.mu.Unlock()
 	return q.used
 }
 
 // MaxQueueSize returns the maximum number of bytes storable in the queue.
-func (q *Queue) MaxQueueSize() int64 {
+func (q *queue) MaxQueueSize() int64 {
 	return q.limit
 }
diff --git a/pkg/sentry/socket/unix/transport/queue/BUILD b/pkg/sentry/socket/unix/transport/queue/BUILD
deleted file mode 100644
index d914ecc23..000000000
--- a/pkg/sentry/socket/unix/transport/queue/BUILD
+++ /dev/null
@@ -1,15 +0,0 @@
-package(licenses = ["notice"])  # Apache 2.0
-
-load("//tools/go_stateify:defs.bzl", "go_library")
-
-go_library(
-    name = "queue",
-    srcs = ["queue.go"],
-    importpath = "gvisor.googlesource.com/gvisor/pkg/sentry/socket/unix/transport/queue",
-    visibility = ["//:sandbox"],
-    deps = [
-        "//pkg/ilist",
-        "//pkg/tcpip",
-        "//pkg/waiter",
-    ],
-)
diff --git a/pkg/sentry/socket/unix/transport/unix.go b/pkg/sentry/socket/unix/transport/unix.go
index 577aa87d5..9a0de9a06 100644
--- a/pkg/sentry/socket/unix/transport/unix.go
+++ b/pkg/sentry/socket/unix/transport/unix.go
@@ -20,7 +20,6 @@ import (
 	"sync/atomic"
 
 	"gvisor.googlesource.com/gvisor/pkg/ilist"
-	"gvisor.googlesource.com/gvisor/pkg/sentry/socket/unix/transport/queue"
 	"gvisor.googlesource.com/gvisor/pkg/tcpip"
 	"gvisor.googlesource.com/gvisor/pkg/tcpip/buffer"
 	"gvisor.googlesource.com/gvisor/pkg/waiter"
@@ -271,7 +270,7 @@ func (m *message) Release() {
 }
 
 // Peek returns a copy of the message.
-func (m *message) Peek() queue.Entry {
+func (m *message) Peek() *message {
 	return &message{Data: m.Data, Control: m.Control.Clone(), Address: m.Address}
 }
 
@@ -325,12 +324,12 @@ type Receiver interface {
 //
 // +stateify savable
 type queueReceiver struct {
-	readQueue *queue.Queue
+	readQueue *queue
 }
 
 // Recv implements Receiver.Recv.
 func (q *queueReceiver) Recv(data [][]byte, creds bool, numRights uintptr, peek bool) (uintptr, uintptr, ControlMessages, tcpip.FullAddress, bool, *tcpip.Error) {
-	var m queue.Entry
+	var m *message
 	var notify bool
 	var err *tcpip.Error
 	if peek {
@@ -341,15 +340,14 @@ func (q *queueReceiver) Recv(data [][]byte, creds bool, numRights uintptr, peek
 	if err != nil {
 		return 0, 0, ControlMessages{}, tcpip.FullAddress{}, false, err
 	}
-	msg := m.(*message)
-	src := []byte(msg.Data)
+	src := []byte(m.Data)
 	var copied uintptr
 	for i := 0; i < len(data) && len(src) > 0; i++ {
 		n := copy(data[i], src)
 		copied += uintptr(n)
 		src = src[n:]
 	}
-	return copied, uintptr(len(msg.Data)), msg.Control, msg.Address, notify, nil
+	return copied, uintptr(len(m.Data)), m.Control, m.Address, notify, nil
 }
 
 // RecvNotify implements Receiver.RecvNotify.
@@ -456,10 +454,9 @@ func (q *streamQueueReceiver) Recv(data [][]byte, wantCreds bool, numRights uint
 			return 0, 0, ControlMessages{}, tcpip.FullAddress{}, false, err
 		}
 		notify = n
-		msg := m.(*message)
-		q.buffer = []byte(msg.Data)
-		q.control = msg.Control
-		q.addr = msg.Address
+		q.buffer = []byte(m.Data)
+		q.control = m.Control
+		q.addr = m.Address
 	}
 
 	var copied uintptr
@@ -506,10 +503,9 @@ func (q *streamQueueReceiver) Recv(data [][]byte, wantCreds bool, numRights uint
 			break
 		}
 		notify = notify || n
-		msg := m.(*message)
-		q.buffer = []byte(msg.Data)
-		q.control = msg.Control
-		q.addr = msg.Address
+		q.buffer = []byte(m.Data)
+		q.control = m.Control
+		q.addr = m.Address
 
 		if wantCreds {
 			if (q.control.Credentials == nil) != (c.Credentials == nil) {
@@ -619,7 +615,7 @@ type connectedEndpoint struct {
 		Type() SockType
 	}
 
-	writeQueue *queue.Queue
+	writeQueue *queue
 }
 
 // Passcred implements ConnectedEndpoint.Passcred.