From 2ad9837e6cc15a90a094abbc0bc8e680989ac042 Mon Sep 17 00:00:00 2001 From: Brad Fitzpatrick Date: Sun, 26 Apr 2026 15:02:22 +0000 Subject: [PATCH] device: refactor container locking for lock-order clarity Device-side portion of upstream tailscale/wireguard-go e3ac4a0 (device, cmd/check-lockorder: add static analysis tool for lock ordering); the analyzer itself is not carried in this fork. --- device/allowedips.go | 1 - device/channels.go | 4 +- device/lock-ordering.md | 27 ----- device/pools.go | 2 - device/receive.go | 234 +++++++++++++++++++++++----------------- device/send.go | 128 +++++++++++++--------- 6 files changed, 214 insertions(+), 182 deletions(-) delete mode 100644 device/lock-ordering.md diff --git a/device/allowedips.go b/device/allowedips.go index 8724802..2271af1 100644 --- a/device/allowedips.go +++ b/device/allowedips.go @@ -329,7 +329,6 @@ func (table *AllowedIPs) Remove(prefix netip.Prefix, peer *Peer) { node.remove() } - // setPeerPrefixes atomically removes all of peer's existing prefixes and adds // the provided ones. func (table *AllowedIPs) setPeerPrefixes(peer *Peer, prefixes []netip.Prefix) { diff --git a/device/channels.go b/device/channels.go index 45b2a76..9ac767f 100644 --- a/device/channels.go +++ b/device/channels.go @@ -97,7 +97,7 @@ func (device *Device) flushInboundQueue(q *autodrainingInboundQueue) { for { select { case elemsContainer := <-q.c: - elemsContainer.Lock() + elemsContainer.filling.Wait() for _, elem := range elemsContainer.elems { device.PutMessageBuffer(elem.buffer) device.PutInboundElement(elem) @@ -136,7 +136,7 @@ func (device *Device) flushOutboundQueue(q *autodrainingOutboundQueue) { for { select { case elemsContainer := <-q.c: - elemsContainer.Lock() + elemsContainer.filling.Wait() for _, elem := range elemsContainer.elems { device.PutOutboundBuffer(elem.buffer) device.PutOutboundElement(elem) diff --git a/device/lock-ordering.md b/device/lock-ordering.md deleted file mode 100644 index 55a15c0..0000000 --- a/device/lock-ordering.md +++ /dev/null @@ -1,27 +0,0 @@ -# Lock Ordering in wireguard-go/device - -## Lock hierarchy - -Locks must be acquired in the order listed below. A goroutine holding a -lock with a higher number must never attempt to acquire a lock with a -lower number. - -``` -Level 0 device.state.Mutex -Level 1 device.ipcMutex (sync.RWMutex) -Level 2 device.net.RWMutex -Level 3 device.staticIdentity.RWMutex -Level 4 device.peers.RWMutex -Level 5 peer.state.Mutex -Level 6 peer.handshake.mutex (sync.RWMutex) -Level 7 peer.keypairs.RWMutex -Level 8 device.allowedips.mu (sync.RWMutex) -Level 9 device.indexTable.RWMutex -Level 10 peer.endpoint.Mutex -Level 11 device.cookieChecker.RWMutex -Level 12 peer.cookieGenerator.RWMutex -Level 13 Timer.modifyingLock / Timer.runningLock -``` - -Not every pair of locks appears in practice; the ordering above is the -transitive closure of the pairs that do. diff --git a/device/pools.go b/device/pools.go index 173486e..6a52472 100644 --- a/device/pools.go +++ b/device/pools.go @@ -74,7 +74,6 @@ func (device *Device) PopulatePools() { func (device *Device) GetInboundElementsContainer() *QueueInboundElementsContainer { c := device.pool.inboundElementsContainer.Get().(*QueueInboundElementsContainer) - c.Mutex = sync.Mutex{} return c } @@ -88,7 +87,6 @@ func (device *Device) PutInboundElementsContainer(c *QueueInboundElementsContain func (device *Device) GetOutboundElementsContainer() *QueueOutboundElementsContainer { c := device.pool.outboundElementsContainer.Get().(*QueueOutboundElementsContainer) - c.Mutex = sync.Mutex{} return c } diff --git a/device/receive.go b/device/receive.go index b6edc56..e11e30a 100644 --- a/device/receive.go +++ b/device/receive.go @@ -8,6 +8,7 @@ package device import ( "encoding/binary" "errors" + "fmt" "net" "net/netip" "sync" @@ -35,8 +36,13 @@ type QueueInboundElement struct { } type QueueInboundElementsContainer struct { - sync.Mutex - elems []*QueueInboundElement + // filling is a one-shot barrier signaling decryption→receive + // handoff. RoutineReceiveIncoming calls Add(1) before sending the + // container down the decryption and inbound queues; RoutineDecryption + // calls Done after decrypting; RoutineSequentialReceiver calls Wait + // before reading the decrypted packets. + filling sync.WaitGroup + elems []*QueueInboundElement } // clearPointers clears elem fields that contain pointers. @@ -178,7 +184,6 @@ func (device *Device) RoutineReceiveIncoming(maxBatchSize int, recv conn.Receive elemsForPeer, ok := elemsByPeer[peer] if !ok { elemsForPeer = device.GetInboundElementsContainer() - elemsForPeer.Lock() elemsByPeer[peer] = elemsForPeer } elemsForPeer.elems = append(elemsForPeer.elems, elem) @@ -222,6 +227,7 @@ func (device *Device) RoutineReceiveIncoming(maxBatchSize int, recv conn.Receive } for peer, elemsContainer := range elemsByPeer { if peer.isRunning.Load() { + elemsContainer.filling.Add(1) peer.queue.inbound.c <- elemsContainer device.queue.decryption.c <- elemsContainer } else { @@ -263,7 +269,7 @@ func (device *Device) RoutineDecryption(id int) { elem.packet = nil } } - elemsContainer.Unlock() + elemsContainer.filling.Done() } } @@ -440,102 +446,128 @@ func (peer *Peer) RoutineSequentialReceiver(maxBatchSize int) { if elemsContainer == nil { return } - elemsContainer.Lock() - validTailPacket := -1 - dataPacketReceived := false - rxBytesLen := uint64(0) - for i, elem := range elemsContainer.elems { - if elem.packet == nil { - // decryption failed - continue - } - - if !elem.keypair.replayFilter.ValidateCounter(elem.counter, RejectAfterMessages) { - continue - } - - validTailPacket = i - if peer.ReceivedWithKeypair(elem.keypair) { - peer.SetEndpointFromPacket(elem.endpoint) - peer.timersHandshakeComplete() - peer.SendStagedPackets() - } - if ep, ok := elem.endpoint.(conn.PeerAwareEndpoint); ok { - ep.FromPeer(peer.handshake.remoteStatic) - } - rxBytesLen += uint64(len(elem.packet) + MinMessageSize) - - if len(elem.packet) == 0 { - device.log.Verbosef("%v - Receiving keepalive packet", peer) - continue - } - dataPacketReceived = true - - switch elem.packet[0] >> 4 { - case 4: - if len(elem.packet) < ipv4.HeaderLen { - continue - } - field := elem.packet[IPv4offsetTotalLength : IPv4offsetTotalLength+2] - length := binary.BigEndian.Uint16(field) - if int(length) > len(elem.packet) || int(length) < ipv4.HeaderLen { - continue - } - elem.packet = elem.packet[:length] - src := elem.packet[IPv4offsetSrc : IPv4offsetSrc+net.IPv4len] - srcAddr, _ := netip.AddrFromSlice(src) - if !peer.AllowedPeerSourceIP(srcAddr) { - device.log.Verbosef("IPv4 packet with disallowed source address from %v", peer) - continue - } - - case 6: - if len(elem.packet) < ipv6.HeaderLen { - continue - } - field := elem.packet[IPv6offsetPayloadLength : IPv6offsetPayloadLength+2] - length := binary.BigEndian.Uint16(field) - length += ipv6.HeaderLen - if int(length) > len(elem.packet) { - continue - } - elem.packet = elem.packet[:length] - src := elem.packet[IPv6offsetSrc : IPv6offsetSrc+net.IPv6len] - srcAddr, _ := netip.AddrFromSlice(src) - if !peer.AllowedPeerSourceIP(srcAddr) { - device.log.Verbosef("IPv6 packet with disallowed source address from %v", peer) - continue - } - - default: - device.log.Verbosef("Packet with invalid IP version from %v", peer) - continue - } - - bufs = append(bufs, elem.buffer[:MessageTransportOffsetContent+len(elem.packet)]) - } - - peer.rxBytes.Add(rxBytesLen) - if validTailPacket >= 0 { - peer.SetEndpointFromPacket(elemsContainer.elems[validTailPacket].endpoint) - peer.keepKeyFreshReceiving() - peer.timersAnyAuthenticatedPacketTraversal() - peer.timersAnyAuthenticatedPacketReceived() - } - if dataPacketReceived { - peer.timersDataReceived() - } - if len(bufs) > 0 { - _, err := device.tun.device.Write(bufs, MessageTransportOffsetContent) - if err != nil && !device.isClosed() { - device.log.Errorf("Failed to write packets to TUN device: %v", err) - } - } - for _, elem := range elemsContainer.elems { - device.PutMessageBuffer(elem.buffer) - device.PutInboundElement(elem) - } - bufs = bufs[:0] - device.PutInboundElementsContainer(elemsContainer) + peer.processInboundContainer(elemsContainer, bufs[:0]) + } +} + +// processInboundContainer waits for the decryption routine to finish +// filling elemsContainer, then writes the valid packets to the TUN +// device and returns the container to the pool. +// +// scratch is a length-0 slice used to assemble the per-packet buffers +// passed to tun.device.Write; its backing array is reused across calls. +func (peer *Peer) processInboundContainer(elemsContainer *QueueInboundElementsContainer, scratch [][]byte) { + // Invariants from RoutineSequentialReceiver; all should be unreachable. + if len(scratch) != 0 || cap(scratch) == 0 { + panic(fmt.Sprintf("processInboundContainer: scratch must be empty with non-zero cap; got len=%d cap=%d", + len(scratch), cap(scratch))) + } + if cap(scratch) < len(elemsContainer.elems) { + panic(fmt.Sprintf("processInboundContainer: scratch cap %d < elems %d", + cap(scratch), len(elemsContainer.elems))) + } + + device := peer.device + defer device.PutInboundElementsContainer(elemsContainer) + + // Wait for RoutineDecryption to finish filling the container. After + // Wait returns we have happens-before with that goroutine and are the + // sole owner of the container until Put hands it back to the pool. + elemsContainer.filling.Wait() + elems := elemsContainer.elems + + validTailPacket := -1 + dataPacketReceived := false + rxBytesLen := uint64(0) + for i, elem := range elems { + if elem.packet == nil { + // decryption failed + continue + } + + if !elem.keypair.replayFilter.ValidateCounter(elem.counter, RejectAfterMessages) { + continue + } + + validTailPacket = i + if peer.ReceivedWithKeypair(elem.keypair) { + peer.SetEndpointFromPacket(elem.endpoint) + peer.timersHandshakeComplete() + peer.SendStagedPackets() + } + if ep, ok := elem.endpoint.(conn.PeerAwareEndpoint); ok { + ep.FromPeer(peer.handshake.remoteStatic) + } + rxBytesLen += uint64(len(elem.packet) + MinMessageSize) + + if len(elem.packet) == 0 { + device.log.Verbosef("%v - Receiving keepalive packet", peer) + continue + } + dataPacketReceived = true + + switch elem.packet[0] >> 4 { + case 4: + if len(elem.packet) < ipv4.HeaderLen { + continue + } + field := elem.packet[IPv4offsetTotalLength : IPv4offsetTotalLength+2] + length := binary.BigEndian.Uint16(field) + if int(length) > len(elem.packet) || int(length) < ipv4.HeaderLen { + continue + } + elem.packet = elem.packet[:length] + src := elem.packet[IPv4offsetSrc : IPv4offsetSrc+net.IPv4len] + srcAddr, _ := netip.AddrFromSlice(src) + if !peer.AllowedPeerSourceIP(srcAddr) { + device.log.Verbosef("IPv4 packet with disallowed source address from %v", peer) + continue + } + + case 6: + if len(elem.packet) < ipv6.HeaderLen { + continue + } + field := elem.packet[IPv6offsetPayloadLength : IPv6offsetPayloadLength+2] + length := binary.BigEndian.Uint16(field) + length += ipv6.HeaderLen + if int(length) > len(elem.packet) { + continue + } + elem.packet = elem.packet[:length] + src := elem.packet[IPv6offsetSrc : IPv6offsetSrc+net.IPv6len] + srcAddr, _ := netip.AddrFromSlice(src) + if !peer.AllowedPeerSourceIP(srcAddr) { + device.log.Verbosef("IPv6 packet with disallowed source address from %v", peer) + continue + } + + default: + device.log.Verbosef("Packet with invalid IP version from %v", peer) + continue + } + + scratch = append(scratch, elem.buffer[:MessageTransportOffsetContent+len(elem.packet)]) + } + + peer.rxBytes.Add(rxBytesLen) + if validTailPacket >= 0 { + peer.SetEndpointFromPacket(elems[validTailPacket].endpoint) + peer.keepKeyFreshReceiving() + peer.timersAnyAuthenticatedPacketTraversal() + peer.timersAnyAuthenticatedPacketReceived() + } + if dataPacketReceived { + peer.timersDataReceived() + } + if len(scratch) > 0 { + _, err := device.tun.device.Write(scratch, MessageTransportOffsetContent) + if err != nil && !device.isClosed() { + device.log.Errorf("Failed to write packets to TUN device: %v", err) + } + } + for _, elem := range elems { + device.PutMessageBuffer(elem.buffer) + device.PutInboundElement(elem) } } diff --git a/device/send.go b/device/send.go index 6bd1ec9..fae5c41 100644 --- a/device/send.go +++ b/device/send.go @@ -8,6 +8,7 @@ package device import ( "encoding/binary" "errors" + "fmt" "net" "net/netip" "os" @@ -58,8 +59,13 @@ type QueueOutboundElement struct { } type QueueOutboundElementsContainer struct { - sync.Mutex - elems []*QueueOutboundElement + // filling is a one-shot barrier signaling encryption→send handoff. + // SendStagedPackets calls Add(1) before sending the container down + // the encryption and outbound queues; RoutineEncryption calls Done + // after encrypting; RoutineSequentialSender calls Wait before + // reading the encrypted packets. + filling sync.WaitGroup + elems []*QueueOutboundElement } func (device *Device) NewOutboundElement() *QueueOutboundElement { @@ -486,7 +492,6 @@ top: elem.keypair = keypair } - elemsContainer.Lock() elemsContainer.elems = elemsContainer.elems[:i] if elemsContainerOOO != nil { @@ -502,6 +507,7 @@ top: // add to parallel and sequential queue if peer.isRunning.Load() { + elemsContainer.filling.Add(1) peer.queue.outbound.c <- elemsContainer peer.device.queue.encryption.c <- elemsContainer } else { @@ -595,7 +601,7 @@ func (device *Device) RoutineEncryption(id int) { // re-slice packet to include encapsulating transport space elem.packet = elem.buffer[:MessageEncapsulatingTransportSize+len(elem.packet)] } - elemsContainer.Unlock() + elemsContainer.filling.Done() } } @@ -610,60 +616,84 @@ func (peer *Peer) RoutineSequentialSender(maxBatchSize int) { bufs := make([][]byte, 0, maxBatchSize) for elemsContainer := range peer.queue.outbound.c { - bufs = bufs[:0] if elemsContainer == nil { return } - if !peer.isRunning.Load() { - // peer has been stopped; return re-usable elems to the shared pool. - // This is an optimization only. It is possible for the peer to be stopped - // immediately after this check, in which case, elem will get processed. - // The timers and SendBuffers code are resilient to a few stragglers. - // TODO: rework peer shutdown order to ensure - // that we never accidentally keep timers alive longer than necessary. - elemsContainer.Lock() - peer.queuedOutboundPackets.Add(-int32(len(elemsContainer.elems))) - for _, elem := range elemsContainer.elems { - device.PutOutboundBuffer(elem.buffer) - device.PutOutboundElement(elem) - } - device.PutOutboundElementsContainer(elemsContainer) - continue - } - dataSent := false - elemsContainer.Lock() - for _, elem := range elemsContainer.elems { - if len(elem.packet) != MessageKeepaliveSize { - dataSent = true - } - bufs = append(bufs, elem.packet) - } + peer.processOutboundContainer(elemsContainer, bufs[:0]) + } +} - peer.timersAnyAuthenticatedPacketTraversal() - peer.timersAnyAuthenticatedPacketSent() +// processOutboundContainer waits for the encryption routine to finish +// filling elemsContainer, then sends the batch (or drops it, if the peer +// has been stopped) and returns the container to the pool. +// +// scratch is a length-0 slice used to assemble the per-packet buffers +// passed to SendBuffers; its backing array is reused across calls. +func (peer *Peer) processOutboundContainer(elemsContainer *QueueOutboundElementsContainer, scratch [][]byte) { + // Invariants from RoutineSequentialSender; all should be unreachable. + if len(scratch) != 0 || cap(scratch) == 0 { + panic(fmt.Sprintf("processOutboundContainer: scratch must be empty with non-zero cap; got len=%d cap=%d", + len(scratch), cap(scratch))) + } + if cap(scratch) < len(elemsContainer.elems) { + panic(fmt.Sprintf("processOutboundContainer: scratch cap %d < elems %d", + cap(scratch), len(elemsContainer.elems))) + } - err := peer.SendBuffers(bufs) - if dataSent { - peer.timersDataSent() - } + device := peer.device + defer device.PutOutboundElementsContainer(elemsContainer) + + // Wait for RoutineEncryption to finish filling the container. After + // Wait returns we have happens-before with that goroutine and are the + // sole owner of the container until Put hands it back to the pool. + elemsContainer.filling.Wait() + + if !peer.isRunning.Load() { + // peer has been stopped; return re-usable elems to the shared pool. + // This is an optimization only. It is possible for the peer to be stopped + // immediately after this check, in which case, elem will get processed. + // The timers and SendBuffers code are resilient to a few stragglers. + // TODO: rework peer shutdown order to ensure + // that we never accidentally keep timers alive longer than necessary. peer.queuedOutboundPackets.Add(-int32(len(elemsContainer.elems))) for _, elem := range elemsContainer.elems { device.PutOutboundBuffer(elem.buffer) device.PutOutboundElement(elem) } - device.PutOutboundElementsContainer(elemsContainer) - if err != nil { - var errGSO conn.ErrUDPGSODisabled - if errors.As(err, &errGSO) { - device.log.Verbosef(err.Error()) - err = errGSO.RetryErr - } - } - if err != nil { - device.log.Errorf("%v - Failed to send data packets: %v", peer, err) - continue - } - - peer.keepKeyFreshSending() + return } + + dataSent := false + for _, elem := range elemsContainer.elems { + if len(elem.packet[MessageEncapsulatingTransportSize:]) != MessageKeepaliveSize { + dataSent = true + } + scratch = append(scratch, elem.packet) + } + + peer.timersAnyAuthenticatedPacketTraversal() + peer.timersAnyAuthenticatedPacketSent() + + err := peer.SendBuffers(scratch) + if dataSent { + peer.timersDataSent() + } + peer.queuedOutboundPackets.Add(-int32(len(elemsContainer.elems))) + for _, elem := range elemsContainer.elems { + device.PutOutboundBuffer(elem.buffer) + device.PutOutboundElement(elem) + } + if err != nil { + var errGSO conn.ErrUDPGSODisabled + if errors.As(err, &errGSO) { + device.log.Verbosef(err.Error()) + err = errGSO.RetryErr + } + } + if err != nil { + device.log.Errorf("%v - Failed to send data packets: %v", peer, err) + return + } + + peer.keepKeyFreshSending() }