From ef9e8d913185e6967eca7321daacad5693178ca8 Mon Sep 17 00:00:00 2001 From: Kevin Krakauer Date: Mon, 25 Apr 2022 20:38:37 -0700 Subject: [PATCH] netstack: switch from sync/atomic to atomicbitops for 32 bit values PiperOrigin-RevId: 444446109 --- pkg/sentry/control/logging.go | 7 +- pkg/tcpip/link/fdbased/BUILD | 1 + pkg/tcpip/link/fdbased/endpoint.go | 8 +- pkg/tcpip/link/fdbased/mmap_unsafe.go | 6 +- pkg/tcpip/link/qdisc/fifo/BUILD | 1 + pkg/tcpip/link/qdisc/fifo/fifo.go | 10 +- pkg/tcpip/link/sharedmem/BUILD | 1 + pkg/tcpip/link/sharedmem/queue/BUILD | 2 + pkg/tcpip/link/sharedmem/queue/queue_test.go | 21 ++-- pkg/tcpip/link/sharedmem/queue/rx.go | 10 +- pkg/tcpip/link/sharedmem/queue/tx.go | 8 +- pkg/tcpip/link/sharedmem/rx.go | 9 +- pkg/tcpip/link/sharedmem/server_rx.go | 9 +- pkg/tcpip/link/sharedmem/server_tx.go | 7 +- pkg/tcpip/link/sharedmem/sharedmem.go | 13 ++- pkg/tcpip/link/sharedmem/sharedmem_server.go | 14 ++- pkg/tcpip/link/sharedmem/sharedmem_unsafe.go | 5 +- pkg/tcpip/link/sniffer/BUILD | 1 + pkg/tcpip/link/sniffer/sniffer.go | 14 +-- pkg/tcpip/network/arp/BUILD | 1 + pkg/tcpip/network/arp/arp.go | 12 +-- pkg/tcpip/network/ipv4/BUILD | 1 + pkg/tcpip/network/ipv4/igmp.go | 15 ++- pkg/tcpip/network/ipv4/ipv4.go | 48 ++++----- pkg/tcpip/network/ipv6/BUILD | 1 + pkg/tcpip/network/ipv6/ipv6.go | 47 ++++---- pkg/tcpip/ports/BUILD | 1 + pkg/tcpip/ports/ports.go | 8 +- pkg/tcpip/socketops.go | 102 +++++++++--------- pkg/tcpip/stack/BUILD | 1 + pkg/tcpip/stack/conntrack.go | 10 +- pkg/tcpip/stack/neighbor_cache_test.go | 10 +- pkg/tcpip/stack/nic.go | 12 +-- pkg/tcpip/stack/nic_test.go | 5 +- pkg/tcpip/stack/tcp.go | 7 +- pkg/tcpip/transport/internal/network/BUILD | 1 + .../transport/internal/network/endpoint.go | 10 +- pkg/tcpip/transport/tcp/connect.go | 4 +- pkg/tcpip/transport/tcp/endpoint.go | 61 ++++++----- pkg/tcpip/transport/tcp/endpoint_state.go | 15 +-- .../transport/tcp/test/e2e/tcp_rack_test.go | 12 +-- .../transport/tcp/test/e2e/tcp_sack_test.go | 6 +- pkg/tcpip/transport/tcp/test/e2e/tcp_test.go | 2 +- runsc/boot/loader.go | 5 +- test/benchmarks/tcp/tcp_proxy.go | 2 +- 45 files changed, 265 insertions(+), 281 deletions(-) diff --git a/pkg/sentry/control/logging.go b/pkg/sentry/control/logging.go index 7613dfcbc..d9d396ed0 100644 --- a/pkg/sentry/control/logging.go +++ b/pkg/sentry/control/logging.go @@ -16,7 +16,6 @@ package control import ( "fmt" - "sync/atomic" "gvisor.dev/gvisor/pkg/log" "gvisor.dev/gvisor/pkg/sentry/strace" @@ -83,11 +82,11 @@ func (l *Logging) Change(args *LoggingArgs, code *int) error { if args.SetLogPackets { if args.LogPackets { - atomic.StoreUint32(&sniffer.LogPackets, 1) + sniffer.LogPackets.Store(1) } else { - atomic.StoreUint32(&sniffer.LogPackets, 0) + sniffer.LogPackets.Store(0) } - log.Infof("LogPackets set to: %v", atomic.LoadUint32(&sniffer.LogPackets)) + log.Infof("LogPackets set to: %v", sniffer.LogPackets.Load()) } if args.SetStrace { diff --git a/pkg/tcpip/link/fdbased/BUILD b/pkg/tcpip/link/fdbased/BUILD index 0dc97431b..c83922307 100644 --- a/pkg/tcpip/link/fdbased/BUILD +++ b/pkg/tcpip/link/fdbased/BUILD @@ -14,6 +14,7 @@ go_library( ], visibility = ["//visibility:public"], deps = [ + "//pkg/atomicbitops", "//pkg/sync", "//pkg/tcpip", "//pkg/tcpip/buffer", diff --git a/pkg/tcpip/link/fdbased/endpoint.go b/pkg/tcpip/link/fdbased/endpoint.go index a683bab99..949c3ffef 100644 --- a/pkg/tcpip/link/fdbased/endpoint.go +++ b/pkg/tcpip/link/fdbased/endpoint.go @@ -42,9 +42,9 @@ package fdbased import ( "fmt" - "sync/atomic" "golang.org/x/sys/unix" + "gvisor.dev/gvisor/pkg/atomicbitops" "gvisor.dev/gvisor/pkg/sync" "gvisor.dev/gvisor/pkg/tcpip" "gvisor.dev/gvisor/pkg/tcpip/buffer" @@ -223,9 +223,7 @@ type Options struct { // Since fanoutID must be unique within the network namespace, we start with // the PID to avoid collisions. The only way to be sure of avoiding collisions // is to run in a new network namespace. -// -// Must be accessed using atomic operations. -var fanoutID int32 = int32(unix.Getpid()) +var fanoutID atomicbitops.Int32 = atomicbitops.FromInt32(int32(unix.Getpid())) // New creates a new fd-based endpoint. // @@ -282,7 +280,7 @@ func New(opts *Options) (stack.LinkEndpoint, error) { // Increment fanoutID to ensure that we don't re-use the same fanoutID for // the next endpoint. - fid := atomic.AddInt32(&fanoutID, 1) + fid := fanoutID.Add(1) // Create per channel dispatchers. for _, fd := range opts.FDs { diff --git a/pkg/tcpip/link/fdbased/mmap_unsafe.go b/pkg/tcpip/link/fdbased/mmap_unsafe.go index 5b786169a..7d62aa2a6 100644 --- a/pkg/tcpip/link/fdbased/mmap_unsafe.go +++ b/pkg/tcpip/link/fdbased/mmap_unsafe.go @@ -19,10 +19,10 @@ package fdbased import ( "fmt" - "sync/atomic" "unsafe" "golang.org/x/sys/unix" + "gvisor.dev/gvisor/pkg/atomicbitops" ) // tPacketHdrlen is the TPACKET_HDRLEN variable defined in . @@ -34,7 +34,7 @@ var tPacketHdrlen = tPacketAlign(unsafe.Sizeof(tPacketHdr{}) + unsafe.Sizeof(uni func (t tPacketHdr) tpStatus() uint32 { hdr := unsafe.Pointer(&t[0]) statusPtr := unsafe.Pointer(uintptr(hdr) + uintptr(tpStatusOffset)) - return atomic.LoadUint32((*uint32)(statusPtr)) + return (*atomicbitops.Uint32)(statusPtr).Load() } // setTPStatus set's the frame status to the provided status. @@ -43,7 +43,7 @@ func (t tPacketHdr) tpStatus() uint32 { func (t tPacketHdr) setTPStatus(status uint32) { hdr := unsafe.Pointer(&t[0]) statusPtr := unsafe.Pointer(uintptr(hdr) + uintptr(tpStatusOffset)) - atomic.StoreUint32((*uint32)(statusPtr), status) + (*atomicbitops.Uint32)(statusPtr).Store(status) } func newPacketMMapDispatcher(fd int, e *endpoint) (linkDispatcher, error) { diff --git a/pkg/tcpip/link/qdisc/fifo/BUILD b/pkg/tcpip/link/qdisc/fifo/BUILD index 9f3b75daa..8d9ae6a16 100644 --- a/pkg/tcpip/link/qdisc/fifo/BUILD +++ b/pkg/tcpip/link/qdisc/fifo/BUILD @@ -9,6 +9,7 @@ go_library( ], visibility = ["//visibility:public"], deps = [ + "//pkg/atomicbitops", "//pkg/sleep", "//pkg/sync", "//pkg/tcpip", diff --git a/pkg/tcpip/link/qdisc/fifo/fifo.go b/pkg/tcpip/link/qdisc/fifo/fifo.go index 607012723..2cd315037 100644 --- a/pkg/tcpip/link/qdisc/fifo/fifo.go +++ b/pkg/tcpip/link/qdisc/fifo/fifo.go @@ -18,8 +18,7 @@ package fifo import ( - "sync/atomic" - + "gvisor.dev/gvisor/pkg/atomicbitops" "gvisor.dev/gvisor/pkg/sleep" "gvisor.dev/gvisor/pkg/sync" "gvisor.dev/gvisor/pkg/tcpip" @@ -44,8 +43,7 @@ type discipline struct { wg sync.WaitGroup dispatchers []queueDispatcher - // +checkatomic - closed int32 + closed atomicbitops.Int32 } // queueDispatcher is responsible for dispatching all outbound packets in its @@ -134,7 +132,7 @@ func (qd *queueDispatcher) dispatchLoop() { // - pkt.GSOOptions // - pkt.NetworkProtocolNumber func (d *discipline) WritePacket(pkt *stack.PacketBuffer) tcpip.Error { - if atomic.LoadInt32(&d.closed) == qDiscClosed { + if d.closed.Load() == qDiscClosed { return &tcpip.ErrClosedForSend{} } qd := &d.dispatchers[int(pkt.Hash)%len(d.dispatchers)] @@ -154,7 +152,7 @@ func (d *discipline) WritePacket(pkt *stack.PacketBuffer) tcpip.Error { } func (d *discipline) Close() { - atomic.StoreInt32(&d.closed, qDiscClosed) + d.closed.Store(qDiscClosed) for i := range d.dispatchers { d.dispatchers[i].closeWaker.Assert() } diff --git a/pkg/tcpip/link/sharedmem/BUILD b/pkg/tcpip/link/sharedmem/BUILD index f7070bf56..bd8dfa218 100644 --- a/pkg/tcpip/link/sharedmem/BUILD +++ b/pkg/tcpip/link/sharedmem/BUILD @@ -18,6 +18,7 @@ go_library( "//visibility:public", ], deps = [ + "//pkg/atomicbitops", "//pkg/cleanup", "//pkg/eventfd", "//pkg/log", diff --git a/pkg/tcpip/link/sharedmem/queue/BUILD b/pkg/tcpip/link/sharedmem/queue/BUILD index 3ba06af73..32f89700b 100644 --- a/pkg/tcpip/link/sharedmem/queue/BUILD +++ b/pkg/tcpip/link/sharedmem/queue/BUILD @@ -10,6 +10,7 @@ go_library( ], visibility = ["//visibility:public"], deps = [ + "//pkg/atomicbitops", "//pkg/log", "//pkg/tcpip/link/sharedmem/pipe", ], @@ -22,6 +23,7 @@ go_test( ], library = ":queue", deps = [ + "//pkg/atomicbitops", "//pkg/tcpip/link/sharedmem/pipe", ], ) diff --git a/pkg/tcpip/link/sharedmem/queue/queue_test.go b/pkg/tcpip/link/sharedmem/queue/queue_test.go index b8a7f3d86..64f218d00 100644 --- a/pkg/tcpip/link/sharedmem/queue/queue_test.go +++ b/pkg/tcpip/link/sharedmem/queue/queue_test.go @@ -19,6 +19,7 @@ import ( "reflect" "testing" + "gvisor.dev/gvisor/pkg/atomicbitops" "gvisor.dev/gvisor/pkg/tcpip/link/sharedmem/pipe" ) @@ -35,7 +36,7 @@ func TestBasicTxQueue(t *testing.T) { txp.Init(pb2) var q Tx - var state uint32 + var state atomicbitops.Uint32 q.Init(pb1, pb2, &state) // Enqueue two buffers. @@ -204,7 +205,7 @@ func TestBadTxCompletion(t *testing.T) { txp.Init(pb2) var q Tx - var state uint32 + var state atomicbitops.Uint32 q.Init(pb1, pb2, &state) // Post a completion that is too short, and check that it is ignored. @@ -320,7 +321,7 @@ func TestFillTxPipe(t *testing.T) { txp.Init(pb2) var q Tx - var state uint32 + var state atomicbitops.Uint32 q.Init(pb1, pb2, &state) // Transmit twice, which should fill the tx pipe. @@ -389,7 +390,7 @@ func TestLotsOfTransmissions(t *testing.T) { txp.Init(pb2) var q Tx - var state uint32 + var state atomicbitops.Uint32 q.Init(pb1, pb2, &state) // Prepare packet with two buffers. @@ -495,13 +496,13 @@ func TestRxEnableNotification(t *testing.T) { pb1 := make([]byte, 100) pb2 := make([]byte, 100) - var state uint32 + var state atomicbitops.Uint32 var q Rx q.Init(pb1, pb2, &state) q.EnableNotification() - if state != EventFDEnabled { - t.Fatalf("Bad value in shared state: got %v, want %v", state, EventFDEnabled) + if state.Load() != EventFDEnabled { + t.Fatalf("Bad value in shared state: got %v, want %v", state.Load(), EventFDEnabled) } } @@ -510,12 +511,12 @@ func TestRxDisableNotification(t *testing.T) { pb1 := make([]byte, 100) pb2 := make([]byte, 100) - var state uint32 + var state atomicbitops.Uint32 var q Rx q.Init(pb1, pb2, &state) q.DisableNotification() - if state != EventFDDisabled { - t.Fatalf("Bad value in shared state: got %v, want %v", state, EventFDDisabled) + if state.Load() != EventFDDisabled { + t.Fatalf("Bad value in shared state: got %v, want %v", state.Load(), EventFDDisabled) } } diff --git a/pkg/tcpip/link/sharedmem/queue/rx.go b/pkg/tcpip/link/sharedmem/queue/rx.go index 89bdf5ef6..328ecd91a 100644 --- a/pkg/tcpip/link/sharedmem/queue/rx.go +++ b/pkg/tcpip/link/sharedmem/queue/rx.go @@ -18,8 +18,8 @@ package queue import ( "encoding/binary" - "sync/atomic" + "gvisor.dev/gvisor/pkg/atomicbitops" "gvisor.dev/gvisor/pkg/log" "gvisor.dev/gvisor/pkg/tcpip/link/sharedmem/pipe" ) @@ -77,12 +77,12 @@ type RxBuffer struct { type Rx struct { tx pipe.Tx rx pipe.Rx - sharedEventFDState *uint32 + sharedEventFDState *atomicbitops.Uint32 } // Init initializes the receive queue with the given pipes, and shared state // pointer -- the latter is used to enable/disable eventfd notifications. -func (r *Rx) Init(tx, rx []byte, sharedEventFDState *uint32) { +func (r *Rx) Init(tx, rx []byte, sharedEventFDState *atomicbitops.Uint32) { r.sharedEventFDState = sharedEventFDState r.tx.Init(tx) r.rx.Init(rx) @@ -91,13 +91,13 @@ func (r *Rx) Init(tx, rx []byte, sharedEventFDState *uint32) { // EnableNotification updates the shared state such that the peer will notify // the eventfd when there are packets to be dequeued. func (r *Rx) EnableNotification() { - atomic.StoreUint32(r.sharedEventFDState, EventFDEnabled) + r.sharedEventFDState.Store(EventFDEnabled) } // DisableNotification updates the shared state such that the peer will not // notify the eventfd. func (r *Rx) DisableNotification() { - atomic.StoreUint32(r.sharedEventFDState, EventFDDisabled) + r.sharedEventFDState.Store(EventFDDisabled) } // PostedBuffersLimit returns the maximum number of buffers that can be posted diff --git a/pkg/tcpip/link/sharedmem/queue/tx.go b/pkg/tcpip/link/sharedmem/queue/tx.go index 09907c761..27c6d4597 100644 --- a/pkg/tcpip/link/sharedmem/queue/tx.go +++ b/pkg/tcpip/link/sharedmem/queue/tx.go @@ -16,8 +16,8 @@ package queue import ( "encoding/binary" - "sync/atomic" + "gvisor.dev/gvisor/pkg/atomicbitops" "gvisor.dev/gvisor/pkg/log" "gvisor.dev/gvisor/pkg/tcpip/link/sharedmem/pipe" ) @@ -52,11 +52,11 @@ type TxBuffer struct { type Tx struct { tx pipe.Tx rx pipe.Rx - sharedEventFDState *uint32 + sharedEventFDState *atomicbitops.Uint32 } // Init initializes the transmit queue with the given pipes. -func (t *Tx) Init(tx, rx []byte, sharedEventFDState *uint32) { +func (t *Tx) Init(tx, rx []byte, sharedEventFDState *atomicbitops.Uint32) { t.tx.Init(tx) t.rx.Init(rx) t.sharedEventFDState = sharedEventFDState @@ -66,7 +66,7 @@ func (t *Tx) Init(tx, rx []byte, sharedEventFDState *uint32) { // peer of events (eg. packet transmit etc). func (t *Tx) NotificationsEnabled() bool { // Notifications are considered enabled unless explicitly disabled. - return atomic.LoadUint32(t.sharedEventFDState) != EventFDDisabled + return t.sharedEventFDState.Load() != EventFDDisabled } // Enqueue queues the given linked list of buffers for transmission as one diff --git a/pkg/tcpip/link/sharedmem/rx.go b/pkg/tcpip/link/sharedmem/rx.go index 87747dcc7..8cbf0c93a 100644 --- a/pkg/tcpip/link/sharedmem/rx.go +++ b/pkg/tcpip/link/sharedmem/rx.go @@ -18,9 +18,8 @@ package sharedmem import ( - "sync/atomic" - "golang.org/x/sys/unix" + "gvisor.dev/gvisor/pkg/atomicbitops" "gvisor.dev/gvisor/pkg/eventfd" "gvisor.dev/gvisor/pkg/tcpip/link/sharedmem/queue" ) @@ -111,7 +110,7 @@ func (r *rx) notify() { // that were read as well. // // This function will block if there aren't any available packets. -func (r *rx) postAndReceive(b []queue.RxBuffer, stopRequested *uint32) ([]queue.RxBuffer, uint32) { +func (r *rx) postAndReceive(b []queue.RxBuffer, stopRequested *atomicbitops.Uint32) ([]queue.RxBuffer, uint32) { // Post the buffers first. If we cannot post, sleep until we can. We // never post more than will fit concurrently, so it's safe to wait // until enough room is available. @@ -119,7 +118,7 @@ func (r *rx) postAndReceive(b []queue.RxBuffer, stopRequested *uint32) ([]queue. r.q.EnableNotification() for !r.q.PostBuffers(b) { r.eventFD.Wait() - if atomic.LoadUint32(stopRequested) != 0 { + if stopRequested.Load() != 0 { r.q.DisableNotification() return nil, 0 } @@ -143,7 +142,7 @@ func (r *rx) postAndReceive(b []queue.RxBuffer, stopRequested *uint32) ([]queue. // Wait for notification. r.eventFD.Wait() - if atomic.LoadUint32(stopRequested) != 0 { + if stopRequested.Load() != 0 { r.q.DisableNotification() return nil, 0 } diff --git a/pkg/tcpip/link/sharedmem/server_rx.go b/pkg/tcpip/link/sharedmem/server_rx.go index 40068334b..75e901be6 100644 --- a/pkg/tcpip/link/sharedmem/server_rx.go +++ b/pkg/tcpip/link/sharedmem/server_rx.go @@ -18,9 +18,8 @@ package sharedmem import ( - "sync/atomic" - "golang.org/x/sys/unix" + "gvisor.dev/gvisor/pkg/atomicbitops" "gvisor.dev/gvisor/pkg/cleanup" "gvisor.dev/gvisor/pkg/eventfd" "gvisor.dev/gvisor/pkg/tcpip/link/sharedmem/pipe" @@ -47,7 +46,7 @@ type serverRx struct { // sharedEventFDState is the memory region in sharedData used to enable // disable notifications on eventFD. - sharedEventFDState *uint32 + sharedEventFDState *atomicbitops.Uint32 } // init initializes all state needed by the serverTx queue based on the @@ -112,13 +111,13 @@ func (s *serverRx) cleanup() { // EnableNotification updates the shared state such that the peer will notify // the eventfd when there are packets to be dequeued. func (s *serverRx) EnableNotification() { - atomic.StoreUint32(s.sharedEventFDState, queue.EventFDEnabled) + s.sharedEventFDState.Store(queue.EventFDEnabled) } // DisableNotification updates the shared state such that the peer will not // notify the eventfd. func (s *serverRx) DisableNotification() { - atomic.StoreUint32(s.sharedEventFDState, queue.EventFDDisabled) + s.sharedEventFDState.Store(queue.EventFDDisabled) } // completionNotificationSize is size in bytes of a completion notification sent diff --git a/pkg/tcpip/link/sharedmem/server_tx.go b/pkg/tcpip/link/sharedmem/server_tx.go index be0ebf931..3144187b3 100644 --- a/pkg/tcpip/link/sharedmem/server_tx.go +++ b/pkg/tcpip/link/sharedmem/server_tx.go @@ -18,9 +18,8 @@ package sharedmem import ( - "sync/atomic" - "golang.org/x/sys/unix" + "gvisor.dev/gvisor/pkg/atomicbitops" "gvisor.dev/gvisor/pkg/cleanup" "gvisor.dev/gvisor/pkg/eventfd" "gvisor.dev/gvisor/pkg/tcpip/buffer" @@ -50,7 +49,7 @@ type serverTx struct { // sharedEventFDState is the memory region in sharedData used to enable/disable // notifications on eventFD. - sharedEventFDState *uint32 + sharedEventFDState *atomicbitops.Uint32 } // init initializes all tstate needed by the serverTx queue based on the @@ -198,7 +197,7 @@ func (s *serverTx) transmit(views []buffer.View) bool { func (s *serverTx) notificationsEnabled() bool { // notifications are considered to be enabled unless explicitly disabled. - return atomic.LoadUint32(s.sharedEventFDState) != queue.EventFDDisabled + return s.sharedEventFDState.Load() != queue.EventFDDisabled } func (s *serverTx) notify() { diff --git a/pkg/tcpip/link/sharedmem/sharedmem.go b/pkg/tcpip/link/sharedmem/sharedmem.go index 18e25f3d2..102b90019 100644 --- a/pkg/tcpip/link/sharedmem/sharedmem.go +++ b/pkg/tcpip/link/sharedmem/sharedmem.go @@ -25,8 +25,8 @@ package sharedmem import ( "fmt" - "sync/atomic" + "gvisor.dev/gvisor/pkg/atomicbitops" "gvisor.dev/gvisor/pkg/eventfd" "gvisor.dev/gvisor/pkg/log" "gvisor.dev/gvisor/pkg/sync" @@ -167,9 +167,8 @@ type endpoint struct { // rx is the receive queue. rx rx - // stopRequested is to be accessed atomically only, and determines if - // the worker goroutines should stop. - stopRequested uint32 + // stopRequested determines whether the worker goroutines should stop. + stopRequested atomicbitops.Uint32 // Wait group used to indicate that all workers have stopped. completed sync.WaitGroup @@ -236,7 +235,7 @@ func New(opts Options) (stack.LinkEndpoint, error) { func (e *endpoint) Close() { // Tell dispatch goroutine to stop, then write to the eventfd so that // it wakes up in case it's sleeping. - atomic.StoreUint32(&e.stopRequested, 1) + e.stopRequested.Store(1) e.rx.eventFD.Notify() // Cleanup the queues inline if the worker hasn't started yet; we also @@ -261,7 +260,7 @@ func (e *endpoint) Wait() { // reads packets from the rx queue. func (e *endpoint) Attach(dispatcher stack.NetworkDispatcher) { e.mu.Lock() - if !e.workerStarted && atomic.LoadUint32(&e.stopRequested) == 0 { + if !e.workerStarted && e.stopRequested.Load() == 0 { e.workerStarted = true e.completed.Add(1) @@ -396,7 +395,7 @@ func (e *endpoint) dispatchLoop(d stack.NetworkDispatcher) { // Read in a loop until a stop is requested. var rxb []queue.RxBuffer - for atomic.LoadUint32(&e.stopRequested) == 0 { + for e.stopRequested.Load() == 0 { var n uint32 rxb, n = e.rx.postAndReceive(rxb, &e.stopRequested) diff --git a/pkg/tcpip/link/sharedmem/sharedmem_server.go b/pkg/tcpip/link/sharedmem/sharedmem_server.go index 9d7773fb4..ad8398172 100644 --- a/pkg/tcpip/link/sharedmem/sharedmem_server.go +++ b/pkg/tcpip/link/sharedmem/sharedmem_server.go @@ -18,8 +18,7 @@ package sharedmem import ( - "sync/atomic" - + "gvisor.dev/gvisor/pkg/atomicbitops" "gvisor.dev/gvisor/pkg/sync" "gvisor.dev/gvisor/pkg/tcpip" "gvisor.dev/gvisor/pkg/tcpip/buffer" @@ -44,9 +43,8 @@ type serverEndpoint struct { // rx is the receive queue. rx serverRx - // stopRequested is to be accessed atomically only, and determines if the - // worker goroutines should stop. - stopRequested uint32 + // stopRequested determines whether the worker goroutines should stop. + stopRequested atomicbitops.Uint32 // Wait group used to indicate that all workers have stopped. completed sync.WaitGroup @@ -124,7 +122,7 @@ func NewServerEndpoint(opts Options) (stack.LinkEndpoint, error) { func (e *serverEndpoint) Close() { // Tell dispatch goroutine to stop, then write to the eventfd so that it wakes // up in case it's sleeping. - atomic.StoreUint32(&e.stopRequested, 1) + e.stopRequested.Store(1) e.rx.eventFD.Notify() // Cleanup the queues inline if the worker hasn't started yet; we also know it @@ -149,7 +147,7 @@ func (e *serverEndpoint) Wait() { // reads packets from the rx queue. func (e *serverEndpoint) Attach(dispatcher stack.NetworkDispatcher) { e.mu.Lock() - if !e.workerStarted && atomic.LoadUint32(&e.stopRequested) == 0 { + if !e.workerStarted && e.stopRequested.Load() == 0 { e.workerStarted = true e.completed.Add(1) if e.peerFD >= 0 { @@ -277,7 +275,7 @@ func (e *serverEndpoint) WritePackets(pkts stack.PacketBufferList) (int, tcpip.E // dispatchLoop reads packets from the rx queue in a loop and dispatches them // to the network stack. func (e *serverEndpoint) dispatchLoop(d stack.NetworkDispatcher) { - for atomic.LoadUint32(&e.stopRequested) == 0 { + for e.stopRequested.Load() == 0 { b := e.rx.receive() if b == nil { e.rx.EnableNotification() diff --git a/pkg/tcpip/link/sharedmem/sharedmem_unsafe.go b/pkg/tcpip/link/sharedmem/sharedmem_unsafe.go index d974c266e..657ddf485 100644 --- a/pkg/tcpip/link/sharedmem/sharedmem_unsafe.go +++ b/pkg/tcpip/link/sharedmem/sharedmem_unsafe.go @@ -20,13 +20,14 @@ import ( "unsafe" "golang.org/x/sys/unix" + "gvisor.dev/gvisor/pkg/atomicbitops" "gvisor.dev/gvisor/pkg/memutil" ) // sharedDataPointer converts the shared data slice into a pointer so that it // can be used in atomic operations. -func sharedDataPointer(sharedData []byte) *uint32 { - return (*uint32)(unsafe.Pointer(&sharedData[0:4][0])) +func sharedDataPointer(sharedData []byte) *atomicbitops.Uint32 { + return (*atomicbitops.Uint32)(unsafe.Pointer(&sharedData[0:4][0])) } // getBuffer returns a memory region mapped to the full contents of the given diff --git a/pkg/tcpip/link/sniffer/BUILD b/pkg/tcpip/link/sniffer/BUILD index 4aac12a8c..006196294 100644 --- a/pkg/tcpip/link/sniffer/BUILD +++ b/pkg/tcpip/link/sniffer/BUILD @@ -10,6 +10,7 @@ go_library( ], visibility = ["//visibility:public"], deps = [ + "//pkg/atomicbitops", "//pkg/log", "//pkg/tcpip", "//pkg/tcpip/buffer", diff --git a/pkg/tcpip/link/sniffer/sniffer.go b/pkg/tcpip/link/sniffer/sniffer.go index d172821b9..36c0ac99a 100644 --- a/pkg/tcpip/link/sniffer/sniffer.go +++ b/pkg/tcpip/link/sniffer/sniffer.go @@ -24,9 +24,9 @@ import ( "encoding/binary" "fmt" "io" - "sync/atomic" "time" + "gvisor.dev/gvisor/pkg/atomicbitops" "gvisor.dev/gvisor/pkg/log" "gvisor.dev/gvisor/pkg/tcpip" "gvisor.dev/gvisor/pkg/tcpip/buffer" @@ -38,16 +38,12 @@ import ( // LogPackets is a flag used to enable or disable packet logging via the log // package. Valid values are 0 or 1. -// -// LogPackets must be accessed atomically. -var LogPackets uint32 = 1 +var LogPackets atomicbitops.Uint32 = atomicbitops.FromUint32(1) // LogPacketsToPCAP is a flag used to enable or disable logging packets to a // pcap writer. Valid values are 0 or 1. A writer must have been specified when the // sniffer was created for this flag to have effect. -// -// LogPacketsToPCAP must be accessed atomically. -var LogPacketsToPCAP uint32 = 1 +var LogPacketsToPCAP atomicbitops.Uint32 = atomicbitops.FromUint32(1) type endpoint struct { nested.Endpoint @@ -142,10 +138,10 @@ func (e *endpoint) DeliverNetworkPacket(protocol tcpip.NetworkProtocolNumber, pk func (e *endpoint) dumpPacket(dir direction, protocol tcpip.NetworkProtocolNumber, pkt *stack.PacketBuffer) { writer := e.writer - if writer == nil && atomic.LoadUint32(&LogPackets) == 1 { + if writer == nil && LogPackets.Load() == 1 { logPacket(e.logPrefix, dir, protocol, pkt) } - if writer != nil && atomic.LoadUint32(&LogPacketsToPCAP) == 1 { + if writer != nil && LogPacketsToPCAP.Load() == 1 { packet := pcapPacket{ timestamp: time.Now(), packet: pkt, diff --git a/pkg/tcpip/network/arp/BUILD b/pkg/tcpip/network/arp/BUILD index ca9bba83a..7df369548 100644 --- a/pkg/tcpip/network/arp/BUILD +++ b/pkg/tcpip/network/arp/BUILD @@ -10,6 +10,7 @@ go_library( ], visibility = ["//visibility:public"], deps = [ + "//pkg/atomicbitops", "//pkg/sync", "//pkg/tcpip", "//pkg/tcpip/buffer", diff --git a/pkg/tcpip/network/arp/arp.go b/pkg/tcpip/network/arp/arp.go index 189d70dfe..65721a8f2 100644 --- a/pkg/tcpip/network/arp/arp.go +++ b/pkg/tcpip/network/arp/arp.go @@ -20,8 +20,8 @@ package arp import ( "fmt" "reflect" - "sync/atomic" + "gvisor.dev/gvisor/pkg/atomicbitops" "gvisor.dev/gvisor/pkg/sync" "gvisor.dev/gvisor/pkg/tcpip" "gvisor.dev/gvisor/pkg/tcpip/buffer" @@ -50,9 +50,7 @@ type endpoint struct { protocol *protocol // enabled is set to 1 when the NIC is enabled and 0 when it is disabled. - // - // Must be accessed using atomic operations. - enabled uint32 + enabled atomicbitops.Uint32 nic stack.NetworkInterface stats sharedStats @@ -104,15 +102,15 @@ func (e *endpoint) Enabled() bool { // isEnabled returns true if the endpoint is enabled, regardless of the // enabled status of the NIC. func (e *endpoint) isEnabled() bool { - return atomic.LoadUint32(&e.enabled) == 1 + return e.enabled.Load() == 1 } // setEnabled sets the enabled status for the endpoint. func (e *endpoint) setEnabled(v bool) { if v { - atomic.StoreUint32(&e.enabled, 1) + e.enabled.Store(1) } else { - atomic.StoreUint32(&e.enabled, 0) + e.enabled.Store(0) } } diff --git a/pkg/tcpip/network/ipv4/BUILD b/pkg/tcpip/network/ipv4/BUILD index 5e931277e..53c24f4c8 100644 --- a/pkg/tcpip/network/ipv4/BUILD +++ b/pkg/tcpip/network/ipv4/BUILD @@ -12,6 +12,7 @@ go_library( ], visibility = ["//visibility:public"], deps = [ + "//pkg/atomicbitops", "//pkg/sync", "//pkg/tcpip", "//pkg/tcpip/buffer", diff --git a/pkg/tcpip/network/ipv4/igmp.go b/pkg/tcpip/network/ipv4/igmp.go index 5df951b2b..d9e231335 100644 --- a/pkg/tcpip/network/ipv4/igmp.go +++ b/pkg/tcpip/network/ipv4/igmp.go @@ -16,9 +16,9 @@ package ipv4 import ( "fmt" - "sync/atomic" "time" + "gvisor.dev/gvisor/pkg/atomicbitops" "gvisor.dev/gvisor/pkg/tcpip" "gvisor.dev/gvisor/pkg/tcpip/buffer" "gvisor.dev/gvisor/pkg/tcpip/header" @@ -83,9 +83,8 @@ type igmpState struct { // MUST be based upon whether or not an IGMPv1 query was heard in the last // [Version 1 Router Present Timeout] seconds". // - // Must be accessed with atomic operations. Holds a value of 1 when true, 0 - // when false. - igmpV1Present uint32 + // Holds a value of 1 when true, 0 when false. + igmpV1Present atomicbitops.Uint32 // igmpV1Job is scheduled when this interface receives an IGMPv1 style // message, upon expiration the igmpV1Present flag is cleared. @@ -149,7 +148,7 @@ func (igmp *igmpState) init(ep *endpoint) { Protocol: igmp, MaxUnsolicitedReportDelay: UnsolicitedReportIntervalMax, }) - igmp.igmpV1Present = igmpV1PresentDefault + igmp.igmpV1Present = atomicbitops.FromUint32(igmpV1PresentDefault) igmp.igmpV1Job = tcpip.NewJob(ep.protocol.stack.Clock(), &ep.mu, func() { igmp.setV1Present(false) }) @@ -269,14 +268,14 @@ func (igmp *igmpState) handleIGMP(pkt *stack.PacketBuffer, hasRouterAlertOption } func (igmp *igmpState) v1Present() bool { - return atomic.LoadUint32(&igmp.igmpV1Present) == 1 + return igmp.igmpV1Present.Load() == 1 } func (igmp *igmpState) setV1Present(v bool) { if v { - atomic.StoreUint32(&igmp.igmpV1Present, 1) + igmp.igmpV1Present.Store(1) } else { - atomic.StoreUint32(&igmp.igmpV1Present, 0) + igmp.igmpV1Present.Store(0) } } diff --git a/pkg/tcpip/network/ipv4/ipv4.go b/pkg/tcpip/network/ipv4/ipv4.go index fbbbb1272..820c608d6 100644 --- a/pkg/tcpip/network/ipv4/ipv4.go +++ b/pkg/tcpip/network/ipv4/ipv4.go @@ -19,9 +19,9 @@ import ( "fmt" "math" "reflect" - "sync/atomic" "time" + "gvisor.dev/gvisor/pkg/atomicbitops" "gvisor.dev/gvisor/pkg/sync" "gvisor.dev/gvisor/pkg/tcpip" "gvisor.dev/gvisor/pkg/tcpip/buffer" @@ -85,24 +85,18 @@ type endpoint struct { // enabled is set to 1 when the endpoint is enabled and 0 when it is // disabled. - // - // +checkatomic - enabled uint32 + enabled atomicbitops.Uint32 // forwarding is set to forwardingEnabled when the endpoint has forwarding // enabled and forwardingDisabled when it is disabled. - // - // +checkatomic - forwarding uint32 + forwarding atomicbitops.Uint32 // multicastForwarding is set to forwardingEnabled when the endpoint has // forwarding enabled and forwardingDisabled when it is disabled. // // TODO(https://gvisor.dev/issue/7338): Implement support for multicast //forwarding. Currently, setting this value to true is a no-op. - // - // +checkatomic - multicastForwarding uint32 + multicastForwarding atomicbitops.Uint32 // mu protects below. mu sync.RWMutex @@ -195,7 +189,7 @@ func (p *protocol) forgetEndpoint(nicID tcpip.NICID) { // Forwarding implements stack.ForwardingNetworkEndpoint. func (e *endpoint) Forwarding() bool { - return atomic.LoadUint32(&e.forwarding) == forwardingEnabled + return e.forwarding.Load() == forwardingEnabled } // setForwarding sets the forwarding status for the endpoint. @@ -207,7 +201,7 @@ func (e *endpoint) setForwarding(v bool) bool { forwarding = forwardingEnabled } - return atomic.SwapUint32(&e.forwarding, forwarding) != forwardingDisabled + return e.forwarding.Swap(forwarding) != forwardingDisabled } // SetForwarding implements stack.ForwardingNetworkEndpoint. @@ -248,7 +242,7 @@ func (e *endpoint) SetForwarding(forwarding bool) bool { // MulticastForwarding implements stack.MulticastForwardingNetworkEndpoint. func (e *endpoint) MulticastForwarding() bool { - return atomic.LoadUint32(&e.multicastForwarding) == forwardingEnabled + return e.multicastForwarding.Load() == forwardingEnabled } // SetMulticastForwarding implements stack.MulticastForwardingNetworkEndpoint. @@ -258,7 +252,7 @@ func (e *endpoint) SetMulticastForwarding(forwarding bool) bool { updatedForwarding = forwardingEnabled } - return atomic.SwapUint32(&e.multicastForwarding, updatedForwarding) != forwardingDisabled + return e.multicastForwarding.Swap(updatedForwarding) != forwardingDisabled } // Enable implements stack.NetworkEndpoint. @@ -316,7 +310,7 @@ func (e *endpoint) Enabled() bool { // isEnabled returns true if the endpoint is enabled, regardless of the // enabled status of the NIC. func (e *endpoint) isEnabled() bool { - return atomic.LoadUint32(&e.enabled) == 1 + return e.enabled.Load() == 1 } // setEnabled sets the enabled status for the endpoint. @@ -324,9 +318,9 @@ func (e *endpoint) isEnabled() bool { // Returns true if the enabled status was updated. func (e *endpoint) setEnabled(v bool) bool { if v { - return atomic.SwapUint32(&e.enabled, 1) == 0 + return e.enabled.Swap(1) == 0 } - return atomic.SwapUint32(&e.enabled, 0) == 1 + return e.enabled.Swap(0) == 1 } // Disable implements stack.NetworkEndpoint. @@ -416,7 +410,7 @@ func (e *endpoint) addIPHeader(srcAddr, dstAddr tcpip.Address, pkt *stack.Packet // RFC 6864 section 4.3 mandates uniqueness of ID values for non-atomic // datagrams. Since the DF bit is never being set here, all datagrams // are non-atomic and need an ID. - id := atomic.AddUint32(&e.protocol.ids[hashRoute(srcAddr, dstAddr, params.Protocol, e.protocol.hashIV)%buckets], 1) + id := e.protocol.ids[hashRoute(srcAddr, dstAddr, params.Protocol, e.protocol.hashIV)%buckets].Add(1) ipH.Encode(&header.IPv4Fields{ TotalLength: uint16(length), ID: uint16(id), @@ -588,7 +582,7 @@ func (e *endpoint) WriteHeaderIncludedPacket(r *stack.Route, pkt *stack.PacketBu // non-atomic datagrams, so assign an ID to all such datagrams // according to the definition given in RFC 6864 section 4. if ipH.Flags()&header.IPv4FlagDontFragment == 0 || ipH.Flags()&header.IPv4FlagMoreFragments != 0 || ipH.FragmentOffset() > 0 { - ipH.SetID(uint16(atomic.AddUint32(&e.protocol.ids[hashRoute(r.LocalAddress(), r.RemoteAddress(), 0 /* protocol */, e.protocol.hashIV)%buckets], 1))) + ipH.SetID(uint16(e.protocol.ids[hashRoute(r.LocalAddress(), r.RemoteAddress(), 0 /* protocol */, e.protocol.hashIV)%buckets].Add(1))) } } @@ -1206,11 +1200,9 @@ type protocol struct { // defaultTTL is the current default TTL for the protocol. Only the // uint8 portion of it is meaningful. - // - // +checkatomic - defaultTTL uint32 + defaultTTL atomicbitops.Uint32 - ids []uint32 + ids []atomicbitops.Uint32 hashIV uint32 fragmentation *fragmentation.Fragmentation @@ -1258,12 +1250,12 @@ func (p *protocol) Option(option tcpip.GettableNetworkProtocolOption) tcpip.Erro // SetDefaultTTL sets the default TTL for endpoints created with this protocol. func (p *protocol) SetDefaultTTL(ttl uint8) { - atomic.StoreUint32(&p.defaultTTL, uint32(ttl)) + p.defaultTTL.Store(uint32(ttl)) } // DefaultTTL returns the default TTL for endpoints created with this protocol. func (p *protocol) DefaultTTL() uint8 { - return uint8(atomic.LoadUint32(&p.defaultTTL)) + return uint8(p.defaultTTL.Load()) } // Close implements stack.TransportProtocol. @@ -1426,12 +1418,12 @@ type Options struct { // NewProtocolWithOptions returns an IPv4 network protocol. func NewProtocolWithOptions(opts Options) stack.NetworkProtocolFactory { - ids := make([]uint32, buckets) + ids := make([]atomicbitops.Uint32, buckets) // Randomly initialize hashIV and the ids. r := hash.RandN32(1 + buckets) for i := range ids { - ids[i] = r[i] + ids[i] = atomicbitops.FromUint32(r[i]) } hashIV := r[buckets] @@ -1440,7 +1432,7 @@ func NewProtocolWithOptions(opts Options) stack.NetworkProtocolFactory { stack: s, ids: ids, hashIV: hashIV, - defaultTTL: DefaultTTL, + defaultTTL: atomicbitops.FromUint32(DefaultTTL), options: opts, } p.fragmentation = fragmentation.NewFragmentation(fragmentblockSize, fragmentation.HighFragThreshold, fragmentation.LowFragThreshold, ReassembleTimeout, s.Clock(), p) diff --git a/pkg/tcpip/network/ipv6/BUILD b/pkg/tcpip/network/ipv6/BUILD index 88b03cc8a..5e4c7fa7a 100644 --- a/pkg/tcpip/network/ipv6/BUILD +++ b/pkg/tcpip/network/ipv6/BUILD @@ -14,6 +14,7 @@ go_library( ], visibility = ["//visibility:public"], deps = [ + "//pkg/atomicbitops", "//pkg/sync", "//pkg/tcpip", "//pkg/tcpip/buffer", diff --git a/pkg/tcpip/network/ipv6/ipv6.go b/pkg/tcpip/network/ipv6/ipv6.go index 0813a9ceb..bc02b9e57 100644 --- a/pkg/tcpip/network/ipv6/ipv6.go +++ b/pkg/tcpip/network/ipv6/ipv6.go @@ -22,9 +22,9 @@ import ( "math" "reflect" "sort" - "sync/atomic" "time" + "gvisor.dev/gvisor/pkg/atomicbitops" "gvisor.dev/gvisor/pkg/sync" "gvisor.dev/gvisor/pkg/tcpip" "gvisor.dev/gvisor/pkg/tcpip/buffer" @@ -189,24 +189,18 @@ type endpoint struct { // enabled is set to 1 when the endpoint is enabled and 0 when it is // disabled. - // - // Must be accessed using atomic operations. - enabled uint32 + enabled atomicbitops.Uint32 // forwarding is set to forwardingEnabled when the endpoint has forwarding // enabled and forwardingDisabled when it is disabled. - // - // Must be accessed using atomic operations. - forwarding uint32 + forwarding atomicbitops.Uint32 // multicastForwarding is set to forwardingEnabled when the endpoint has // forwarding enabled and forwardingDisabled when it is disabled. // // TODO(https://gvisor.dev/issue/7338): Implement support for multicast // forwarding. Currently, setting this value to true is a no-op. - // - // Must be accessed using atomic operations. - multicastForwarding uint32 + multicastForwarding atomicbitops.Uint32 mu struct { sync.RWMutex @@ -444,7 +438,7 @@ func (e *endpoint) dupTentativeAddrDetected(addr tcpip.Address, holderLinkAddr t // Forwarding implements stack.ForwardingNetworkEndpoint. func (e *endpoint) Forwarding() bool { - return atomic.LoadUint32(&e.forwarding) == forwardingEnabled + return e.forwarding.Load() == forwardingEnabled } // setForwarding sets the forwarding status for the endpoint. @@ -456,7 +450,7 @@ func (e *endpoint) setForwarding(v bool) bool { forwarding = forwardingEnabled } - return atomic.SwapUint32(&e.forwarding, forwarding) != forwardingDisabled + return e.forwarding.Swap(forwarding) != forwardingDisabled } // SetForwarding implements stack.ForwardingNetworkEndpoint. @@ -517,7 +511,7 @@ func (e *endpoint) SetForwarding(forwarding bool) bool { // MulticastForwarding implements stack.MulticastForwardingNetworkEndpoint. func (e *endpoint) MulticastForwarding() bool { - return atomic.LoadUint32(&e.multicastForwarding) == forwardingEnabled + return e.multicastForwarding.Load() == forwardingEnabled } // SetMulticastForwarding implements stack.MulticastForwardingNetworkEndpoint. @@ -527,7 +521,7 @@ func (e *endpoint) SetMulticastForwarding(forwarding bool) bool { updatedForwarding = forwardingEnabled } - return atomic.SwapUint32(&e.multicastForwarding, updatedForwarding) != forwardingDisabled + return e.multicastForwarding.Swap(updatedForwarding) != forwardingDisabled } // Enable implements stack.NetworkEndpoint. @@ -621,7 +615,7 @@ func (e *endpoint) Enabled() bool { // isEnabled returns true if the endpoint is enabled, regardless of the // enabled status of the NIC. func (e *endpoint) isEnabled() bool { - return atomic.LoadUint32(&e.enabled) == 1 + return e.enabled.Load() == 1 } // setEnabled sets the enabled status for the endpoint. @@ -629,9 +623,9 @@ func (e *endpoint) isEnabled() bool { // Returns true if the enabled status was updated. func (e *endpoint) setEnabled(v bool) bool { if v { - return atomic.SwapUint32(&e.enabled, 1) == 0 + return e.enabled.Swap(1) == 0 } - return atomic.SwapUint32(&e.enabled, 0) == 1 + return e.enabled.Swap(0) == 1 } // Disable implements stack.NetworkEndpoint. @@ -754,7 +748,7 @@ func (e *endpoint) handleFragments(r *stack.Route, networkMTU uint32, pkt *stack } pf := fragmentation.MakePacketFragmenter(pkt, fragmentPayloadLen, calculateFragmentReserve(pkt)) - id := atomic.AddUint32(&e.protocol.ids[hashRoute(r, e.protocol.hashIV)%buckets], 1) + id := e.protocol.ids[hashRoute(r, e.protocol.hashIV)%buckets].Add(1) var n int for { @@ -1966,14 +1960,12 @@ type protocol struct { icmpRateLimitedTypes map[header.ICMPv6Type]struct{} } - ids []uint32 + ids []atomicbitops.Uint32 hashIV uint32 // defaultTTL is the current default TTL for the protocol. Only the // uint8 portion of it is meaningful. - // - // Must be accessed using atomic operations. - defaultTTL uint32 + defaultTTL atomicbitops.Uint32 fragmentation *fragmentation.Fragmentation icmpRateLimiter *stack.ICMPRateLimiter @@ -2097,12 +2089,12 @@ func (p *protocol) Option(option tcpip.GettableNetworkProtocolOption) tcpip.Erro // SetDefaultTTL sets the default TTL for endpoints created with this protocol. func (p *protocol) SetDefaultTTL(ttl uint8) { - atomic.StoreUint32(&p.defaultTTL, uint32(ttl)) + p.defaultTTL.Store(uint32(ttl)) } // DefaultTTL returns the default TTL for endpoints created with this protocol. func (p *protocol) DefaultTTL() uint8 { - return uint8(atomic.LoadUint32(&p.defaultTTL)) + return uint8(p.defaultTTL.Load()) } // Close implements stack.TransportProtocol. @@ -2277,12 +2269,17 @@ func NewProtocolWithOptions(opts Options) stack.NetworkProtocolFactory { ids := hash.RandN32(buckets) hashIV := hash.RandN32(1)[0] + atomicIds := make([]atomicbitops.Uint32, len(ids)) + for i := range ids { + atomicIds[i] = atomicbitops.FromUint32(ids[i]) + } + return func(s *stack.Stack) stack.NetworkProtocol { p := &protocol{ stack: s, options: opts, - ids: ids, + ids: atomicIds, hashIV: hashIV, } p.fragmentation = fragmentation.NewFragmentation(header.IPv6FragmentExtHdrFragmentOffsetBytesPerUnit, fragmentation.HighFragThreshold, fragmentation.LowFragThreshold, ReassembleTimeout, s.Clock(), p) diff --git a/pkg/tcpip/ports/BUILD b/pkg/tcpip/ports/BUILD index fe98a52af..8b1169c90 100644 --- a/pkg/tcpip/ports/BUILD +++ b/pkg/tcpip/ports/BUILD @@ -10,6 +10,7 @@ go_library( ], visibility = ["//visibility:public"], deps = [ + "//pkg/atomicbitops", "//pkg/sync", "//pkg/tcpip", "//pkg/tcpip/header", diff --git a/pkg/tcpip/ports/ports.go b/pkg/tcpip/ports/ports.go index fb8ef1ee2..b070b80e3 100644 --- a/pkg/tcpip/ports/ports.go +++ b/pkg/tcpip/ports/ports.go @@ -19,8 +19,8 @@ package ports import ( "math" "math/rand" - "sync/atomic" + "gvisor.dev/gvisor/pkg/atomicbitops" "gvisor.dev/gvisor/pkg/sync" "gvisor.dev/gvisor/pkg/tcpip" "gvisor.dev/gvisor/pkg/tcpip/header" @@ -231,7 +231,7 @@ type PortManager struct { // // hint must be accessed using the portHint/incPortHint helpers. // TODO(gvisor.dev/issue/940): S/R this field. - hint uint32 + hint atomicbitops.Uint32 } // NewPortManager creates new PortManager. @@ -264,12 +264,12 @@ func (pm *PortManager) PickEphemeralPort(rng *rand.Rand, testPort PortTester) (p // portHint atomically reads and returns the pm.hint value. func (pm *PortManager) portHint() uint32 { - return atomic.LoadUint32(&pm.hint) + return pm.hint.Load() } // incPortHint atomically increments pm.hint by 1. func (pm *PortManager) incPortHint() { - atomic.AddUint32(&pm.hint, 1) + pm.hint.Add(1) } // PickEphemeralPortStable starts at the specified offset + pm.portHint and diff --git a/pkg/tcpip/socketops.go b/pkg/tcpip/socketops.go index 00622c408..b57c0fb38 100644 --- a/pkg/tcpip/socketops.go +++ b/pkg/tcpip/socketops.go @@ -15,8 +15,6 @@ package tcpip import ( - "sync/atomic" - "gvisor.dev/gvisor/pkg/atomicbitops" "gvisor.dev/gvisor/pkg/sync" ) @@ -138,96 +136,96 @@ type SocketOptions struct { // broadcastEnabled determines whether datagram sockets are allowed to // send packets to a broadcast address. - broadcastEnabled uint32 + broadcastEnabled atomicbitops.Uint32 // passCredEnabled determines whether SCM_CREDENTIALS socket control // messages are enabled. - passCredEnabled uint32 + passCredEnabled atomicbitops.Uint32 // noChecksumEnabled determines whether UDP checksum is disabled while // transmitting for this socket. - noChecksumEnabled uint32 + noChecksumEnabled atomicbitops.Uint32 // reuseAddressEnabled determines whether Bind() should allow reuse of // local address. - reuseAddressEnabled uint32 + reuseAddressEnabled atomicbitops.Uint32 // reusePortEnabled determines whether to permit multiple sockets to be // bound to an identical socket address. - reusePortEnabled uint32 + reusePortEnabled atomicbitops.Uint32 // keepAliveEnabled determines whether TCP keepalive is enabled for this // socket. - keepAliveEnabled uint32 + keepAliveEnabled atomicbitops.Uint32 // multicastLoopEnabled determines whether multicast packets sent over a // non-loopback interface will be looped back. - multicastLoopEnabled uint32 + multicastLoopEnabled atomicbitops.Uint32 // receiveTOSEnabled is used to specify if the TOS ancillary message is // passed with incoming packets. - receiveTOSEnabled uint32 + receiveTOSEnabled atomicbitops.Uint32 // receiveTTLEnabled is used to specify if the TTL ancillary message is passed // with incoming packets. - receiveTTLEnabled uint32 + receiveTTLEnabled atomicbitops.Uint32 // receiveHopLimitEnabled is used to specify if the HopLimit ancillary message // is passed with incoming packets. - receiveHopLimitEnabled uint32 + receiveHopLimitEnabled atomicbitops.Uint32 // receiveTClassEnabled is used to specify if the IPV6_TCLASS ancillary // message is passed with incoming packets. - receiveTClassEnabled uint32 + receiveTClassEnabled atomicbitops.Uint32 // receivePacketInfoEnabled is used to specify if more information is // provided with incoming IPv4 packets. - receivePacketInfoEnabled uint32 + receivePacketInfoEnabled atomicbitops.Uint32 // receivePacketInfoEnabled is used to specify if more information is // provided with incoming IPv6 packets. - receiveIPv6PacketInfoEnabled uint32 + receiveIPv6PacketInfoEnabled atomicbitops.Uint32 // hdrIncludeEnabled is used to indicate for a raw endpoint that all packets // being written have an IP header and the endpoint should not attach an IP // header. - hdrIncludedEnabled uint32 + hdrIncludedEnabled atomicbitops.Uint32 // v6OnlyEnabled is used to determine whether an IPv6 socket is to be // restricted to sending and receiving IPv6 packets only. - v6OnlyEnabled uint32 + v6OnlyEnabled atomicbitops.Uint32 // quickAckEnabled is used to represent the value of TCP_QUICKACK option. // It currently does not have any effect on the TCP endpoint. - quickAckEnabled uint32 + quickAckEnabled atomicbitops.Uint32 // delayOptionEnabled is used to specify if data should be sent out immediately // by the transport protocol. For TCP, it determines if the Nagle algorithm // is on or off. - delayOptionEnabled uint32 + delayOptionEnabled atomicbitops.Uint32 // corkOptionEnabled is used to specify if data should be held until segments // are full by the TCP transport protocol. - corkOptionEnabled uint32 + corkOptionEnabled atomicbitops.Uint32 // receiveOriginalDstAddress is used to specify if the original destination of // the incoming packet should be returned as an ancillary message. - receiveOriginalDstAddress uint32 + receiveOriginalDstAddress atomicbitops.Uint32 // recvErrEnabled determines whether extended reliable error message passing // is enabled. - recvErrEnabled uint32 + recvErrEnabled atomicbitops.Uint32 // errQueue is the per-socket error queue. It is protected by errQueueMu. errQueueMu sync.Mutex `state:"nosave"` errQueue sockErrorList // bindToDevice determines the device to which the socket is bound. - bindToDevice int32 + bindToDevice atomicbitops.Int32 - // getSendBufferLimits provides the handler to get the min, default and - // max size for send buffer. It is initialized at the creation time and - // will not change. + // getSendBufferLimits provides the handler to get the min, default and max + // size for send buffer. It is initialized at the creation time and will not + // change. getSendBufferLimits GetSendBufferLimits `state:"manual"` // sendBufferSize determines the send buffer size for this socket. @@ -250,7 +248,7 @@ type SocketOptions struct { // rcvlowat specifies the minimum number of bytes which should be // received to indicate the socket as readable. - rcvlowat int32 + rcvlowat atomicbitops.Int32 } // InitHandler initializes the handler. This must be called before using the @@ -262,12 +260,12 @@ func (so *SocketOptions) InitHandler(handler SocketOptionsHandler, stack StackHa so.getReceiveBufferLimits = getReceiveBufferLimits } -func storeAtomicBool(addr *uint32, v bool) { +func storeAtomicBool(addr *atomicbitops.Uint32, v bool) { var val uint32 if v { val = 1 } - atomic.StoreUint32(addr, val) + addr.Store(val) } // SetLastError sets the last error for a socket. @@ -277,7 +275,7 @@ func (so *SocketOptions) SetLastError(err Error) { // GetBroadcast gets value for SO_BROADCAST option. func (so *SocketOptions) GetBroadcast() bool { - return atomic.LoadUint32(&so.broadcastEnabled) != 0 + return so.broadcastEnabled.Load() != 0 } // SetBroadcast sets value for SO_BROADCAST option. @@ -287,7 +285,7 @@ func (so *SocketOptions) SetBroadcast(v bool) { // GetPassCred gets value for SO_PASSCRED option. func (so *SocketOptions) GetPassCred() bool { - return atomic.LoadUint32(&so.passCredEnabled) != 0 + return so.passCredEnabled.Load() != 0 } // SetPassCred sets value for SO_PASSCRED option. @@ -297,7 +295,7 @@ func (so *SocketOptions) SetPassCred(v bool) { // GetNoChecksum gets value for SO_NO_CHECK option. func (so *SocketOptions) GetNoChecksum() bool { - return atomic.LoadUint32(&so.noChecksumEnabled) != 0 + return so.noChecksumEnabled.Load() != 0 } // SetNoChecksum sets value for SO_NO_CHECK option. @@ -307,7 +305,7 @@ func (so *SocketOptions) SetNoChecksum(v bool) { // GetReuseAddress gets value for SO_REUSEADDR option. func (so *SocketOptions) GetReuseAddress() bool { - return atomic.LoadUint32(&so.reuseAddressEnabled) != 0 + return so.reuseAddressEnabled.Load() != 0 } // SetReuseAddress sets value for SO_REUSEADDR option. @@ -318,7 +316,7 @@ func (so *SocketOptions) SetReuseAddress(v bool) { // GetReusePort gets value for SO_REUSEPORT option. func (so *SocketOptions) GetReusePort() bool { - return atomic.LoadUint32(&so.reusePortEnabled) != 0 + return so.reusePortEnabled.Load() != 0 } // SetReusePort sets value for SO_REUSEPORT option. @@ -329,7 +327,7 @@ func (so *SocketOptions) SetReusePort(v bool) { // GetKeepAlive gets value for SO_KEEPALIVE option. func (so *SocketOptions) GetKeepAlive() bool { - return atomic.LoadUint32(&so.keepAliveEnabled) != 0 + return so.keepAliveEnabled.Load() != 0 } // SetKeepAlive sets value for SO_KEEPALIVE option. @@ -340,7 +338,7 @@ func (so *SocketOptions) SetKeepAlive(v bool) { // GetMulticastLoop gets value for IP_MULTICAST_LOOP option. func (so *SocketOptions) GetMulticastLoop() bool { - return atomic.LoadUint32(&so.multicastLoopEnabled) != 0 + return so.multicastLoopEnabled.Load() != 0 } // SetMulticastLoop sets value for IP_MULTICAST_LOOP option. @@ -350,7 +348,7 @@ func (so *SocketOptions) SetMulticastLoop(v bool) { // GetReceiveTOS gets value for IP_RECVTOS option. func (so *SocketOptions) GetReceiveTOS() bool { - return atomic.LoadUint32(&so.receiveTOSEnabled) != 0 + return so.receiveTOSEnabled.Load() != 0 } // SetReceiveTOS sets value for IP_RECVTOS option. @@ -360,7 +358,7 @@ func (so *SocketOptions) SetReceiveTOS(v bool) { // GetReceiveTTL gets value for IP_RECVTTL option. func (so *SocketOptions) GetReceiveTTL() bool { - return atomic.LoadUint32(&so.receiveTTLEnabled) != 0 + return so.receiveTTLEnabled.Load() != 0 } // SetReceiveTTL sets value for IP_RECVTTL option. @@ -370,7 +368,7 @@ func (so *SocketOptions) SetReceiveTTL(v bool) { // GetReceiveHopLimit gets value for IP_RECVHOPLIMIT option. func (so *SocketOptions) GetReceiveHopLimit() bool { - return atomic.LoadUint32(&so.receiveHopLimitEnabled) != 0 + return so.receiveHopLimitEnabled.Load() != 0 } // SetReceiveHopLimit sets value for IP_RECVHOPLIMIT option. @@ -380,7 +378,7 @@ func (so *SocketOptions) SetReceiveHopLimit(v bool) { // GetReceiveTClass gets value for IPV6_RECVTCLASS option. func (so *SocketOptions) GetReceiveTClass() bool { - return atomic.LoadUint32(&so.receiveTClassEnabled) != 0 + return so.receiveTClassEnabled.Load() != 0 } // SetReceiveTClass sets value for IPV6_RECVTCLASS option. @@ -390,7 +388,7 @@ func (so *SocketOptions) SetReceiveTClass(v bool) { // GetReceivePacketInfo gets value for IP_PKTINFO option. func (so *SocketOptions) GetReceivePacketInfo() bool { - return atomic.LoadUint32(&so.receivePacketInfoEnabled) != 0 + return so.receivePacketInfoEnabled.Load() != 0 } // SetReceivePacketInfo sets value for IP_PKTINFO option. @@ -400,7 +398,7 @@ func (so *SocketOptions) SetReceivePacketInfo(v bool) { // GetIPv6ReceivePacketInfo gets value for IPV6_RECVPKTINFO option. func (so *SocketOptions) GetIPv6ReceivePacketInfo() bool { - return atomic.LoadUint32(&so.receiveIPv6PacketInfoEnabled) != 0 + return so.receiveIPv6PacketInfoEnabled.Load() != 0 } // SetIPv6ReceivePacketInfo sets value for IPV6_RECVPKTINFO option. @@ -410,7 +408,7 @@ func (so *SocketOptions) SetIPv6ReceivePacketInfo(v bool) { // GetHeaderIncluded gets value for IP_HDRINCL option. func (so *SocketOptions) GetHeaderIncluded() bool { - return atomic.LoadUint32(&so.hdrIncludedEnabled) != 0 + return so.hdrIncludedEnabled.Load() != 0 } // SetHeaderIncluded sets value for IP_HDRINCL option. @@ -420,7 +418,7 @@ func (so *SocketOptions) SetHeaderIncluded(v bool) { // GetV6Only gets value for IPV6_V6ONLY option. func (so *SocketOptions) GetV6Only() bool { - return atomic.LoadUint32(&so.v6OnlyEnabled) != 0 + return so.v6OnlyEnabled.Load() != 0 } // SetV6Only sets value for IPV6_V6ONLY option. @@ -432,7 +430,7 @@ func (so *SocketOptions) SetV6Only(v bool) { // GetQuickAck gets value for TCP_QUICKACK option. func (so *SocketOptions) GetQuickAck() bool { - return atomic.LoadUint32(&so.quickAckEnabled) != 0 + return so.quickAckEnabled.Load() != 0 } // SetQuickAck sets value for TCP_QUICKACK option. @@ -442,7 +440,7 @@ func (so *SocketOptions) SetQuickAck(v bool) { // GetDelayOption gets inverted value for TCP_NODELAY option. func (so *SocketOptions) GetDelayOption() bool { - return atomic.LoadUint32(&so.delayOptionEnabled) != 0 + return so.delayOptionEnabled.Load() != 0 } // SetDelayOption sets inverted value for TCP_NODELAY option. @@ -453,7 +451,7 @@ func (so *SocketOptions) SetDelayOption(v bool) { // GetCorkOption gets value for TCP_CORK option. func (so *SocketOptions) GetCorkOption() bool { - return atomic.LoadUint32(&so.corkOptionEnabled) != 0 + return so.corkOptionEnabled.Load() != 0 } // SetCorkOption sets value for TCP_CORK option. @@ -464,7 +462,7 @@ func (so *SocketOptions) SetCorkOption(v bool) { // GetReceiveOriginalDstAddress gets value for IP(V6)_RECVORIGDSTADDR option. func (so *SocketOptions) GetReceiveOriginalDstAddress() bool { - return atomic.LoadUint32(&so.receiveOriginalDstAddress) != 0 + return so.receiveOriginalDstAddress.Load() != 0 } // SetReceiveOriginalDstAddress sets value for IP(V6)_RECVORIGDSTADDR option. @@ -474,7 +472,7 @@ func (so *SocketOptions) SetReceiveOriginalDstAddress(v bool) { // GetRecvError gets value for IP*_RECVERR option. func (so *SocketOptions) GetRecvError() bool { - return atomic.LoadUint32(&so.recvErrEnabled) != 0 + return so.recvErrEnabled.Load() != 0 } // SetRecvError sets value for IP*_RECVERR option. @@ -649,7 +647,7 @@ func (so *SocketOptions) QueueLocalErr(err Error, net NetworkProtocolNumber, inf // GetBindToDevice gets value for SO_BINDTODEVICE option. func (so *SocketOptions) GetBindToDevice() int32 { - return atomic.LoadInt32(&so.bindToDevice) + return so.bindToDevice.Load() } // SetBindToDevice sets value for SO_BINDTODEVICE option. If bindToDevice is @@ -659,7 +657,7 @@ func (so *SocketOptions) SetBindToDevice(bindToDevice int32) Error { return &ErrUnknownDevice{} } - atomic.StoreInt32(&so.bindToDevice, bindToDevice) + so.bindToDevice.Store(bindToDevice) return nil } @@ -723,6 +721,6 @@ func (so *SocketOptions) GetRcvlowat() int32 { // SetRcvlowat sets value for SO_RCVLOWAT option. func (so *SocketOptions) SetRcvlowat(rcvlowat int32) Error { - atomic.StoreInt32(&so.rcvlowat, rcvlowat) + so.rcvlowat.Store(rcvlowat) return nil } diff --git a/pkg/tcpip/stack/BUILD b/pkg/tcpip/stack/BUILD index efb1d858a..1ecfd46a7 100644 --- a/pkg/tcpip/stack/BUILD +++ b/pkg/tcpip/stack/BUILD @@ -155,6 +155,7 @@ go_test( ], library = ":stack", deps = [ + "//pkg/atomicbitops", "//pkg/sync", "//pkg/tcpip", "//pkg/tcpip/buffer", diff --git a/pkg/tcpip/stack/conntrack.go b/pkg/tcpip/stack/conntrack.go index eb034b2ce..5bc0bf92c 100644 --- a/pkg/tcpip/stack/conntrack.go +++ b/pkg/tcpip/stack/conntrack.go @@ -20,9 +20,9 @@ import ( "math" "math/rand" "sync" - "sync/atomic" "time" + "gvisor.dev/gvisor/pkg/atomicbitops" "gvisor.dev/gvisor/pkg/tcpip" "gvisor.dev/gvisor/pkg/tcpip/hash/jenkins" "gvisor.dev/gvisor/pkg/tcpip/header" @@ -144,9 +144,7 @@ type conn struct { finalizeOnce sync.Once // Holds a finalizeResult. - // - // +checkatomics - finalizeResult uint32 + finalizeResult atomicbitops.Uint32 mu sync.RWMutex `state:"nosave"` // sourceManip indicates the source manipulation type. @@ -653,7 +651,7 @@ func (ct *ConnTrack) finalize(cn *conn) finalizeResult { } func (cn *conn) getFinalizeResult() finalizeResult { - return finalizeResult(atomic.LoadUint32(&cn.finalizeResult)) + return finalizeResult(cn.finalizeResult.Load()) } // finalize attempts to finalize the connection and returns true iff the @@ -667,7 +665,7 @@ func (cn *conn) getFinalizeResult() finalizeResult { // goroutines will block until the finalizing goroutine finishes finalizing. func (cn *conn) finalize() bool { cn.finalizeOnce.Do(func() { - atomic.StoreUint32(&cn.finalizeResult, uint32(cn.ct.finalize(cn))) + cn.finalizeResult.Store(uint32(cn.ct.finalize(cn))) }) switch res := cn.getFinalizeResult(); res { diff --git a/pkg/tcpip/stack/neighbor_cache_test.go b/pkg/tcpip/stack/neighbor_cache_test.go index 7de25fe37..2bc48784b 100644 --- a/pkg/tcpip/stack/neighbor_cache_test.go +++ b/pkg/tcpip/stack/neighbor_cache_test.go @@ -20,12 +20,12 @@ import ( "math/rand" "strings" "sync" - "sync/atomic" "testing" "time" "github.com/google/go-cmp/cmp" "github.com/google/go-cmp/cmp/cmpopts" + "gvisor.dev/gvisor/pkg/atomicbitops" "gvisor.dev/gvisor/pkg/tcpip" "gvisor.dev/gvisor/pkg/tcpip/faketime" ) @@ -1273,9 +1273,9 @@ func TestNeighborCacheResolutionFailed(t *testing.T) { clock := faketime.NewManualClock() linkRes := newTestNeighborResolver(&nudDisp, config, clock) - var requestCount uint32 + var requestCount atomicbitops.Uint32 linkRes.onLinkAddressRequest = func() { - atomic.AddUint32(&requestCount, 1) + requestCount.Add(1) } entry, ok := linkRes.entries.entry(0) @@ -1303,7 +1303,7 @@ func TestNeighborCacheResolutionFailed(t *testing.T) { } // Verify address resolution fails for an unknown address. - before := atomic.LoadUint32(&requestCount) + before := requestCount.Load() entry.Addr += "2" { @@ -1325,7 +1325,7 @@ func TestNeighborCacheResolutionFailed(t *testing.T) { } maxAttempts := linkRes.neigh.config().MaxUnicastProbes - if got, want := atomic.LoadUint32(&requestCount)-before, maxAttempts; got != want { + if got, want := requestCount.Load()-before, maxAttempts; got != want { t.Errorf("got link address request count = %d, want = %d", got, want) } } diff --git a/pkg/tcpip/stack/nic.go b/pkg/tcpip/stack/nic.go index fa1a66f52..7f7029442 100644 --- a/pkg/tcpip/stack/nic.go +++ b/pkg/tcpip/stack/nic.go @@ -17,8 +17,8 @@ package stack import ( "fmt" "reflect" - "sync/atomic" + "gvisor.dev/gvisor/pkg/atomicbitops" "gvisor.dev/gvisor/pkg/sync" "gvisor.dev/gvisor/pkg/tcpip" "gvisor.dev/gvisor/pkg/tcpip/header" @@ -61,9 +61,7 @@ type nic struct { duplicateAddressDetectors map[tcpip.NetworkProtocolNumber]DuplicateAddressDetector // enabled is set to 1 when the NIC is enabled and 0 when it is disabled. - // - // Must be accessed using atomic operations. - enabled uint32 + enabled atomicbitops.Uint32 // linkResQueue holds packets that are waiting for link resolution to // complete. @@ -221,7 +219,7 @@ func (n *nic) getNetworkEndpoint(proto tcpip.NetworkProtocolNumber) NetworkEndpo // Enabled implements NetworkInterface. func (n *nic) Enabled() bool { - return atomic.LoadUint32(&n.enabled) == 1 + return n.enabled.Load() == 1 } // setEnabled sets the enabled status for the NIC. @@ -229,9 +227,9 @@ func (n *nic) Enabled() bool { // Returns true if the enabled status was updated. func (n *nic) setEnabled(v bool) bool { if v { - return atomic.SwapUint32(&n.enabled, 1) == 0 + return n.enabled.Swap(1) == 0 } - return atomic.SwapUint32(&n.enabled, 0) == 1 + return n.enabled.Swap(0) == 1 } // disable disables n. diff --git a/pkg/tcpip/stack/nic_test.go b/pkg/tcpip/stack/nic_test.go index cdd6dede5..f42a32af7 100644 --- a/pkg/tcpip/stack/nic_test.go +++ b/pkg/tcpip/stack/nic_test.go @@ -18,6 +18,7 @@ import ( "reflect" "testing" + "gvisor.dev/gvisor/pkg/atomicbitops" "gvisor.dev/gvisor/pkg/tcpip" "gvisor.dev/gvisor/pkg/tcpip/buffer" "gvisor.dev/gvisor/pkg/tcpip/header" @@ -203,7 +204,7 @@ func TestDisabledRxStatsWhenNICDisabled(t *testing.T) { func TestPacketWithUnknownNetworkProtocolNumber(t *testing.T) { nic := nic{ stats: makeNICStats(tcpip.NICStats{}.FillIn()), - enabled: 1, + enabled: atomicbitops.FromUint32(1), } // IPv4 isn't recognized since we haven't initialized the NIC with an IPv4 // endpoint. @@ -223,7 +224,7 @@ func TestPacketWithUnknownTransportProtocolNumber(t *testing.T) { nic := nic{ stack: &Stack{}, stats: makeNICStats(tcpip.NICStats{}.FillIn()), - enabled: 1, + enabled: atomicbitops.FromUint32(1), } // UDP isn't recognized since we haven't initialized the NIC with a UDP // protocol. diff --git a/pkg/tcpip/stack/tcp.go b/pkg/tcpip/stack/tcp.go index a941091b0..44b866db5 100644 --- a/pkg/tcpip/stack/tcp.go +++ b/pkg/tcpip/stack/tcp.go @@ -17,6 +17,7 @@ package stack import ( "time" + "gvisor.dev/gvisor/pkg/atomicbitops" "gvisor.dev/gvisor/pkg/tcpip" "gvisor.dev/gvisor/pkg/tcpip/header" "gvisor.dev/gvisor/pkg/tcpip/internal/tcp" @@ -25,7 +26,7 @@ import ( // TCPProbeFunc is the expected function type for a TCP probe function to be // passed to stack.AddTCPProbe. -type TCPProbeFunc func(s TCPEndpointState) +type TCPProbeFunc func(s *TCPEndpointState) // TCPCubicState is used to hold a copy of the internal cubic state when the // TCPProbeFunc is invoked. @@ -396,9 +397,7 @@ type TCPSndBufState struct { // AutoTuneSndBufDisabled indicates that the auto tuning of send buffer // is disabled. - // - // Must be accessed using atomic operations. - AutoTuneSndBufDisabled uint32 + AutoTuneSndBufDisabled atomicbitops.Uint32 } // TCPEndpointStateInner contains the members of TCPEndpointState used directly diff --git a/pkg/tcpip/transport/internal/network/BUILD b/pkg/tcpip/transport/internal/network/BUILD index b6dc07b8b..085d0307d 100644 --- a/pkg/tcpip/transport/internal/network/BUILD +++ b/pkg/tcpip/transport/internal/network/BUILD @@ -14,6 +14,7 @@ go_library( "//pkg/tcpip/transport/udp:__pkg__", ], deps = [ + "//pkg/atomicbitops", "//pkg/sync", "//pkg/tcpip", "//pkg/tcpip/header", diff --git a/pkg/tcpip/transport/internal/network/endpoint.go b/pkg/tcpip/transport/internal/network/endpoint.go index 63c0eed3f..0280411ac 100644 --- a/pkg/tcpip/transport/internal/network/endpoint.go +++ b/pkg/tcpip/transport/internal/network/endpoint.go @@ -18,8 +18,8 @@ package network import ( "fmt" - "sync/atomic" + "gvisor.dev/gvisor/pkg/atomicbitops" "gvisor.dev/gvisor/pkg/sync" "gvisor.dev/gvisor/pkg/tcpip" "gvisor.dev/gvisor/pkg/tcpip/header" @@ -95,9 +95,7 @@ type Endpoint struct { // lock when delivering packets/errors to endpoints). // // Writes must be performed through setEndpointState. - // - // +checkatomics - state uint32 + state atomicbitops.Uint32 } // +stateify savable @@ -156,12 +154,12 @@ func (e *Endpoint) NetProto() tcpip.NetworkProtocolNumber { // // +checklocks:e.mu func (e *Endpoint) setEndpointState(state transport.DatagramEndpointState) { - atomic.StoreUint32(&e.state, uint32(state)) + e.state.Store(uint32(state)) } // State returns the state of the endpoint. func (e *Endpoint) State() transport.DatagramEndpointState { - return transport.DatagramEndpointState(atomic.LoadUint32(&e.state)) + return transport.DatagramEndpointState(e.state.Load()) } // Close cleans the endpoint's resources and leaves the endpoint in a closed diff --git a/pkg/tcpip/transport/tcp/connect.go b/pkg/tcpip/transport/tcp/connect.go index 1908f11c6..c13517437 100644 --- a/pkg/tcpip/transport/tcp/connect.go +++ b/pkg/tcpip/transport/tcp/connect.go @@ -1194,7 +1194,9 @@ func (e *endpoint) handleSegmentsLocked() tcpip.Error { // +checklocks:e.mu func (e *endpoint) probeSegmentLocked() { if fn := e.probe; fn != nil { - fn(e.completeStateLocked()) + var state stack.TCPEndpointState + e.completeStateLocked(&state) + fn(&state) } } diff --git a/pkg/tcpip/transport/tcp/endpoint.go b/pkg/tcpip/transport/tcp/endpoint.go index bea8049e6..78a93fce2 100644 --- a/pkg/tcpip/transport/tcp/endpoint.go +++ b/pkg/tcpip/transport/tcp/endpoint.go @@ -22,9 +22,9 @@ import ( "math" "runtime" "strings" - "sync/atomic" "time" + "gvisor.dev/gvisor/pkg/atomicbitops" "gvisor.dev/gvisor/pkg/sleep" "gvisor.dev/gvisor/pkg/sync" "gvisor.dev/gvisor/pkg/tcpip" @@ -283,6 +283,16 @@ type sndQueueInfo struct { sndWaker sleep.Waker `state:"manual"` } +// CloneState clones sq into other. It is not thread safe +func (sq *sndQueueInfo) CloneState(other *stack.TCPSndBufState) { + other.SndBufSize = sq.SndBufSize + other.SndBufUsed = sq.SndBufUsed + other.SndClosed = sq.SndClosed + other.PacketTooBigCount = sq.PacketTooBigCount + other.SndMTU = sq.SndMTU + other.AutoTuneSndBufDisabled = atomicbitops.FromUint32(sq.AutoTuneSndBufDisabled.RacyLoad()) +} + // rcvQueueInfo contains the endpoint's rcvQueue and associated metadata. // // +stateify savable @@ -392,9 +402,7 @@ type endpoint struct { // compute the window and the actual available buffer space. This is distinct // from rcvBufUsed above which is the actual number of payload bytes held in // the buffer not including any segment overheads. - // - // rcvMemUsed must be accessed atomically. - rcvMemUsed int32 + rcvMemUsed atomicbitops.Int32 // mu protects all endpoint fields unless documented otherwise. mu must // be acquired before interacting with the endpoint fields. @@ -402,11 +410,11 @@ type endpoint struct { // During handshake, mu is locked by the protocol listen goroutine and // released by the handshake completion goroutine. mu sync.CrossGoroutineMutex `state:"nosave"` - ownedByUser uint32 + ownedByUser atomicbitops.Uint32 // state must be read/set using the EndpointState()/setEndpointState() // methods. - state uint32 `state:".(EndpointState)"` + state atomicbitops.Uint32 `state:".(EndpointState)"` // origEndpointState is only used during a restore phase to save the // endpoint state at restore time as the socket is moved to it's correct @@ -632,9 +640,9 @@ func (e *endpoint) LockUser() { if !e.TryLock() { // If socket is owned by the user then just go to sleep // as the lock could be held for a reasonably long time. - if atomic.LoadUint32(&e.ownedByUser) == 1 { + if e.ownedByUser.Load() == 1 { e.mu.Lock() - atomic.StoreUint32(&e.ownedByUser, 1) + e.ownedByUser.Store(1) return } // Spin but yield the processor since the lower half @@ -642,7 +650,7 @@ func (e *endpoint) LockUser() { runtime.Gosched() continue } - atomic.StoreUint32(&e.ownedByUser, 1) + e.ownedByUser.Store(1) return } } @@ -660,7 +668,7 @@ func (e *endpoint) UnlockUser() { // and actually unlock the endpoint mutex. e.segmentQueue.mu.Lock() if e.segmentQueue.emptyLocked() { - if atomic.SwapUint32(&e.ownedByUser, 0) != 1 { + if e.ownedByUser.Swap(0) != 1 { panic("e.UnlockUser() called without calling e.LockUser()") } e.mu.Unlock() @@ -671,7 +679,7 @@ func (e *endpoint) UnlockUser() { // Since we are waking the processor goroutine here just unlock // and let it process the queued segments. - if atomic.SwapUint32(&e.ownedByUser, 0) != 1 { + if e.ownedByUser.Swap(0) != 1 { panic("e.UnlockUser() called without calling e.LockUser()") } processor := e.protocol.dispatcher.selectProcessor(e.ID) @@ -729,7 +737,7 @@ func (e *endpoint) TryLock() bool { // // +checklocks:e.mu func (e *endpoint) setEndpointState(state EndpointState) { - oldstate := EndpointState(atomic.SwapUint32(&e.state, uint32(state))) + oldstate := EndpointState(e.state.Swap(uint32(state))) switch state { case StateEstablished: e.stack.Stats().TCP.CurrentEstablished.Increment() @@ -750,7 +758,7 @@ func (e *endpoint) setEndpointState(state EndpointState) { // EndpointState returns the current state of the endpoint. func (e *endpoint) EndpointState() EndpointState { - return EndpointState(atomic.LoadUint32(&e.state)) + return EndpointState(e.state.Load()) } // setRecentTimestamp sets the recentTS field to the provided value. @@ -811,7 +819,7 @@ func newEndpoint(s *stack.Stack, protocol *protocol, netProto tcpip.NetworkProto }, }, waiterQueue: waiterQueue, - state: uint32(StateInitial), + state: atomicbitops.FromUint32(uint32(StateInitial)), keepalive: keepalive{ idle: DefaultKeepaliveIdle, interval: DefaultKeepaliveInterval, @@ -1804,7 +1812,7 @@ func (e *endpoint) OnSetReceiveBufferSize(rcvBufSz, oldSz int64) (newSz int64, p // OnSetSendBufferSize implements tcpip.SocketOptionsHandler.OnSetSendBufferSize. func (e *endpoint) OnSetSendBufferSize(sz int64) int64 { - atomic.StoreUint32(&e.sndQueueInfo.TCPSndBufState.AutoTuneSndBufDisabled, 1) + e.sndQueueInfo.TCPSndBufState.AutoTuneSndBufDisabled.Store(1) return sz } @@ -2978,12 +2986,12 @@ func (e *endpoint) receiveBufferUsed() int { // receiveMemUsed returns the total memory in use by segments held by this // endpoint. func (e *endpoint) receiveMemUsed() int { - return int(atomic.LoadInt32(&e.rcvMemUsed)) + return int(e.rcvMemUsed.Load()) } // updateReceiveMemUsed adds the provided delta to e.rcvMemUsed. func (e *endpoint) updateReceiveMemUsed(delta int) { - atomic.AddInt32(&e.rcvMemUsed, int32(delta)) + e.rcvMemUsed.Add(int32(delta)) } // maxReceiveBufferSize returns the stack wide maximum receive buffer size for @@ -3074,19 +3082,17 @@ func (e *endpoint) maxOptionSize() (size int) { // used before invoking the probe. // // +checklocks:e.mu -func (e *endpoint) completeStateLocked() stack.TCPEndpointState { - s := stack.TCPEndpointState{ - TCPEndpointStateInner: e.TCPEndpointStateInner, - ID: stack.TCPEndpointID(e.TransportEndpointInfo.ID), - SegTime: e.stack.Clock().NowMonotonic(), - Receiver: e.rcv.TCPReceiverState, - Sender: e.snd.TCPSenderState, - } +func (e *endpoint) completeStateLocked(s *stack.TCPEndpointState) { + s.TCPEndpointStateInner = e.TCPEndpointStateInner + s.ID = stack.TCPEndpointID(e.TransportEndpointInfo.ID) + s.SegTime = e.stack.Clock().NowMonotonic() + s.Receiver = e.rcv.TCPReceiverState + s.Sender = e.snd.TCPSenderState sndBufSize := e.getSendBufferSize() // Copy the send buffer atomically. e.sndQueueInfo.sndQueueMu.Lock() - s.SndBufState = e.sndQueueInfo.TCPSndBufState + e.sndQueueInfo.CloneState(&s.SndBufState) s.SndBufState.SndBufSize = sndBufSize e.sndQueueInfo.sndQueueMu.Unlock() @@ -3112,7 +3118,6 @@ func (e *endpoint) completeStateLocked() stack.TCPEndpointState { s.Sender.RACKState = e.snd.rc.TCPRACKState s.Sender.RetransmitTS = e.snd.retransmitTS s.Sender.SpuriousRecovery = e.snd.spuriousRecovery - return s } func (e *endpoint) initHardwareGSO() { @@ -3233,7 +3238,7 @@ func (e *endpoint) computeTCPSendBufferSize() int64 { // Auto tuning is disabled when the user explicitly sets the send // buffer size with SO_SNDBUF option. - if disabled := atomic.LoadUint32(&e.sndQueueInfo.TCPSndBufState.AutoTuneSndBufDisabled); disabled == 1 { + if disabled := e.sndQueueInfo.TCPSndBufState.AutoTuneSndBufDisabled.Load(); disabled == 1 { return curSndBufSz } diff --git a/pkg/tcpip/transport/tcp/endpoint_state.go b/pkg/tcpip/transport/tcp/endpoint_state.go index 527624760..38f7b50b0 100644 --- a/pkg/tcpip/transport/tcp/endpoint_state.go +++ b/pkg/tcpip/transport/tcp/endpoint_state.go @@ -16,8 +16,8 @@ package tcp import ( "fmt" - "sync/atomic" + "gvisor.dev/gvisor/pkg/atomicbitops" "gvisor.dev/gvisor/pkg/sync" "gvisor.dev/gvisor/pkg/tcpip" "gvisor.dev/gvisor/pkg/tcpip/header" @@ -109,15 +109,16 @@ func (e *endpoint) loadState(epState EndpointState) { // Directly update the state here rather than using e.setEndpointState // as the endpoint is still being loaded and the stack reference is not // yet initialized. - atomic.StoreUint32((*uint32)(&e.state), uint32(epState)) + e.state.Store(uint32(epState)) } // afterLoad is invoked by stateify. func (e *endpoint) afterLoad() { - e.origEndpointState = e.state + // RacyLoad() can be used because we are initializing e. + e.origEndpointState = e.state.RacyLoad() // Restore the endpoint to InitialState as it will be moved to // its origEndpointState during Resume. - e.state = uint32(StateInitial) + e.state = atomicbitops.FromUint32(uint32(StateInitial)) stack.StackFromEnv.RegisterRestoredEndpoint(e) } @@ -198,7 +199,7 @@ func (e *endpoint) Resume(s *stack.Stack) { if _, ok := err.(*tcpip.ErrConnectStarted); !ok { panic("endpoint connecting failed: " + err.String()) } - e.state = e.origEndpointState + e.state.Store(e.origEndpointState) // For FIN-WAIT-2 and TIME-WAIT we need to start the appropriate timers so // that the socket is closed correctly. switch epState { @@ -275,11 +276,11 @@ func (e *endpoint) Resume(s *stack.Stack) { }() case epState == StateClose: e.isPortReserved = false - e.state = uint32(StateClose) + e.state.Store(uint32(StateClose)) e.stack.CompleteTransportEndpointCleanup(e) tcpip.DeleteDanglingEndpoint(e) case epState == StateError: - e.state = uint32(StateError) + e.state.Store(uint32(StateError)) e.stack.CompleteTransportEndpointCleanup(e) tcpip.DeleteDanglingEndpoint(e) } diff --git a/pkg/tcpip/transport/tcp/test/e2e/tcp_rack_test.go b/pkg/tcpip/transport/tcp/test/e2e/tcp_rack_test.go index 005475b20..35872ddc8 100644 --- a/pkg/tcpip/transport/tcp/test/e2e/tcp_rack_test.go +++ b/pkg/tcpip/transport/tcp/test/e2e/tcp_rack_test.go @@ -46,7 +46,7 @@ func TestRACKUpdate(t *testing.T) { var xmitTime tcpip.MonotonicTime probeDone := make(chan struct{}) - c.Stack().AddTCPProbe(func(state stack.TCPEndpointState) { + c.Stack().AddTCPProbe(func(state *stack.TCPEndpointState) { // Validate that the endpoint Sender.RACKState is what we expect. if state.Sender.RACKState.XmitTime.Before(xmitTime) { t.Fatalf("RACK transmit time failed to update when an ACK is received") @@ -99,7 +99,7 @@ func TestRACKDetectReorder(t *testing.T) { var n int const ackNumToVerify = 2 probeDone := make(chan struct{}) - c.Stack().AddTCPProbe(func(state stack.TCPEndpointState) { + c.Stack().AddTCPProbe(func(state *stack.TCPEndpointState) { gotSeq := state.Sender.RACKState.FACK wantSeq := state.Sender.SndNxt // FACK should be updated to the highest ending sequence number of the @@ -160,7 +160,7 @@ const ( func addDSACKSeenCheckerProbe(t *testing.T, c *context.Context, numACK int, probeDone chan int) { var n int - c.Stack().AddTCPProbe(func(state stack.TCPEndpointState) { + c.Stack().AddTCPProbe(func(state *stack.TCPEndpointState) { // Validate that RACK detects DSACK. n++ if n < numACK { @@ -808,7 +808,7 @@ func TestRACKWithInvalidDSACKBlock(t *testing.T) { probeDone := make(chan struct{}) const ackNumToVerify = 2 var n int - c.Stack().AddTCPProbe(func(state stack.TCPEndpointState) { + c.Stack().AddTCPProbe(func(state *stack.TCPEndpointState) { // Validate that RACK does not detect DSACK when DSACK block is // not the first SACK block. n++ @@ -854,7 +854,7 @@ func TestRACKWithInvalidDSACKBlock(t *testing.T) { func addReorderWindowCheckerProbe(c *context.Context, numACK int, probeDone chan error) { var n int - c.Stack().AddTCPProbe(func(state stack.TCPEndpointState) { + c.Stack().AddTCPProbe(func(state *stack.TCPEndpointState) { // Validate that RACK detects DSACK. n++ if n < numACK { @@ -965,7 +965,7 @@ func TestRACKUpdateSackedOut(t *testing.T) { probeDone := make(chan struct{}) ackNum := 0 - c.Stack().AddTCPProbe(func(state stack.TCPEndpointState) { + c.Stack().AddTCPProbe(func(state *stack.TCPEndpointState) { // Validate that the endpoint Sender.SackedOut is what we expect. if state.Sender.SackedOut != 2 && ackNum == 0 { t.Fatalf("SackedOut got updated to wrong value got: %v want: 2", state.Sender.SackedOut) diff --git a/pkg/tcpip/transport/tcp/test/e2e/tcp_sack_test.go b/pkg/tcpip/transport/tcp/test/e2e/tcp_sack_test.go index a6898a3c1..e70ace16a 100644 --- a/pkg/tcpip/transport/tcp/test/e2e/tcp_sack_test.go +++ b/pkg/tcpip/transport/tcp/test/e2e/tcp_sack_test.go @@ -367,7 +367,7 @@ func TestSACKRecovery(t *testing.T) { c := context.New(t, uint32(header.TCPMinimumSize+header.IPv4MinimumSize+e2e.MaxTCPOptionSize+maxPayload)) defer c.Cleanup() - c.Stack().AddTCPProbe(func(s stack.TCPEndpointState) { + c.Stack().AddTCPProbe(func(s *stack.TCPEndpointState) { // We use log.Printf instead of t.Logf here because this probe // can fire even when the test function has finished. This is // because closing the endpoint in cleanup() does not mean the @@ -745,7 +745,7 @@ func TestDetectSpuriousRecoveryWithRTO(t *testing.T) { defer c.Cleanup() probeDone := make(chan struct{}) - c.Stack().AddTCPProbe(func(s stack.TCPEndpointState) { + c.Stack().AddTCPProbe(func(s *stack.TCPEndpointState) { if s.Sender.RetransmitTS == 0 { t.Fatalf("RetransmitTS did not get updated, got: 0 want > 0") } @@ -826,7 +826,7 @@ func TestSACKDetectSpuriousRecoveryWithDupACK(t *testing.T) { numAck := 0 probeDone := make(chan struct{}) - c.Stack().AddTCPProbe(func(s stack.TCPEndpointState) { + c.Stack().AddTCPProbe(func(s *stack.TCPEndpointState) { if numAck < 3 { numAck++ return diff --git a/pkg/tcpip/transport/tcp/test/e2e/tcp_test.go b/pkg/tcpip/transport/tcp/test/e2e/tcp_test.go index d1d8ffd37..49414f15c 100644 --- a/pkg/tcpip/transport/tcp/test/e2e/tcp_test.go +++ b/pkg/tcpip/transport/tcp/test/e2e/tcp_test.go @@ -5293,7 +5293,7 @@ func TestTCPEndpointProbe(t *testing.T) { defer c.Cleanup() invoked := make(chan struct{}) - c.Stack().AddTCPProbe(func(state stack.TCPEndpointState) { + c.Stack().AddTCPProbe(func(state *stack.TCPEndpointState) { // Validate that the endpoint ID is what we expect. // // We don't do an extensive validation of every field but a diff --git a/runsc/boot/loader.go b/runsc/boot/loader.go index afcf0887f..1d5918a55 100644 --- a/runsc/boot/loader.go +++ b/runsc/boot/loader.go @@ -21,7 +21,6 @@ import ( mrand "math/rand" "os" "runtime" - "sync/atomic" gtime "time" specs "github.com/opencontainers/runtime-spec/specs-go" @@ -383,10 +382,10 @@ func New(args Args) (*Loader, error) { // Turn on packet logging if enabled. if args.Conf.LogPackets { log.Infof("Packet logging enabled") - atomic.StoreUint32(&sniffer.LogPackets, 1) + sniffer.LogPackets.Store(1) } else { log.Infof("Packet logging disabled") - atomic.StoreUint32(&sniffer.LogPackets, 0) + sniffer.LogPackets.Store(0) } // Create a watchdog. diff --git a/test/benchmarks/tcp/tcp_proxy.go b/test/benchmarks/tcp/tcp_proxy.go index 308a7c3af..b0879fe82 100644 --- a/test/benchmarks/tcp/tcp_proxy.go +++ b/test/benchmarks/tcp/tcp_proxy.go @@ -324,7 +324,7 @@ func (n netstackImpl) installProbe(probeFileName string) (close func()) { } probeEncoder := gob.NewEncoder(probeFile) // Install a TCP Probe. - n.s.AddTCPProbe(func(state stack.TCPEndpointState) { + n.s.AddTCPProbe(func(state *stack.TCPEndpointState) { probeEncoder.Encode(state) }) return func() { probeFile.Close() }