netbird/gvisor
0
0
Fork 0
mirror of https://github.com/netbirdio/gvisor.git synced 2026-05-22 17:12:49 -07:00
Code Issues Packages Projects Releases Wiki Activity

Merge queue into Unix transport

Browse Source 
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
This commit is contained in:
Ian Gudger
2018-10-17 15:10:20 -07:00
committed by Shentubot
parent e4277cb6ff
commit 6922eee649
6 changed files with 37 additions and 79 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/sentry/socket/unix/transport/BUILD
+1 -1
View File
@@ -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",
pkg/sentry/socket/unix/transport/connectioned.go
+4 -5
View File
@@ -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,
pkg/sentry/socket/unix/transport/connectionless.go
+1 -2
View File
@@ -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
}
pkg/sentry/socket/unix/transport/queue/queue.go → pkg/sentry/socket/unix/transport/queue.go
+19 -40
View File
@@ -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
}
pkg/sentry/socket/unix/transport/queue/BUILD
-15
View File
@@ -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",
],
)
pkg/sentry/socket/unix/transport/unix.go
+12 -16
View File
@@ -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.