device: remove mutex from Peer send/receive

The immediate motivation for this change is an observed deadlock.

1. A goroutine calls peer.Stop. That calls peer.queue.Lock().
2. Another goroutine is in RoutineSequentialReceiver.
   It receives an elem from peer.queue.inbound.
3. The peer.Stop goroutine calls close(peer.queue.inbound),
   close(peer.queue.outbound), and peer.stopping.Wait().
   It blocks waiting for RoutineSequentialReceiver
   and RoutineSequentialSender to exit.
4. The RoutineSequentialReceiver goroutine calls peer.SendStagedPackets().
   SendStagedPackets attempts peer.queue.RLock().
   That blocks forever because the peer.Stop
   goroutine holds a write lock on that mutex.

A background motivation for this change is that it can be expensive
to have a mutex in the hot code path of RoutineSequential*.

The mutex was necessary to avoid attempting to send elems on a closed channel.
This commit removes that danger by never closing the channel.
Instead, we send a sentinel nil value on the channel to indicate
to the receiver that it should exit.

The only problem with this is that if the receiver exits,
we could write an elem into the channel which would never get received.
If it never gets received, it cannot get returned to the device pools.

To work around this, we use a finalizer. When the channel can be GC'd,
the finalizer drains any remaining elements from the channel and
restores them to the device pool.

After that change, peer.queue.RWMutex no longer makes sense where it is.
It is only used to prevent concurrent calls to Start and Stop.
Move it to a more sensible location and make it a plain sync.Mutex.

Signed-off-by: Josh Bleecher Snyder <josh@tailscale.com>

1 files changed, 13 insertions, 10 deletions

diff --git a/device/peer.go b/device/peer.go
index 3e4f4ec..49b9acb 100644
--- a/device/peer.go
+++ b/device/peer.go
@@ -51,8 +51,11 @@ type Peer struct {
 		sentLastMinuteHandshake AtomicBool
 	}
 
+	state struct {
+		mu sync.Mutex // protects against concurrent Start/Stop
+	}
+
 	queue struct {
-		sync.RWMutex
 		staged   chan *QueueOutboundElement // staged packets before a handshake is available
 		outbound chan *QueueOutboundElement // sequential ordering of udp transmission
 		inbound  chan *QueueInboundElement  // sequential ordering of tun writing
@@ -158,8 +161,8 @@ func (peer *Peer) Start() {
 	}
 
 	// prevent simultaneous start/stop operations
-	peer.queue.Lock()
-	defer peer.queue.Unlock()
+	peer.state.mu.Lock()
+	defer peer.state.mu.Unlock()
 
 	if peer.isRunning.Get() {
 		return
@@ -177,8 +180,8 @@ func (peer *Peer) Start() {
 	peer.handshake.mutex.Unlock()
 
 	// prepare queues
-	peer.queue.outbound = make(chan *QueueOutboundElement, QueueOutboundSize)
-	peer.queue.inbound = make(chan *QueueInboundElement, QueueInboundSize)
+	peer.queue.outbound = newAutodrainingOutboundQueue(device)
+	peer.queue.inbound = newAutodrainingInboundQueue(device)
 	if peer.queue.staged == nil {
 		peer.queue.staged = make(chan *QueueOutboundElement, QueueStagedSize)
 	}
@@ -239,8 +242,8 @@ func (peer *Peer) ExpireCurrentKeypairs() {
 }
 
 func (peer *Peer) Stop() {
-	peer.queue.Lock()
-	defer peer.queue.Unlock()
+	peer.state.mu.Lock()
+	defer peer.state.mu.Unlock()
 
 	if !peer.isRunning.Swap(false) {
 		return
@@ -249,9 +252,9 @@ func (peer *Peer) Stop() {
 	peer.device.log.Verbosef("%v - Stopping...", peer)
 
 	peer.timersStop()
-
-	close(peer.queue.inbound)
-	close(peer.queue.outbound)
+	// Signal that RoutineSequentialSender and RoutineSequentialReceiver should exit.
+	peer.queue.inbound <- nil
+	peer.queue.outbound <- nil
 	peer.stopping.Wait()
 	peer.device.queue.encryption.wg.Done() // no more writes to encryption queue from us