[PATCH] device: add write queue mutex for peer

[liuhaichao]

From: Haichao Liu <liuhaic...@bytedance.com>

fix panic: send on closed channel when remove peer

Change-Id: Ica0e2447cd941bd3a26948aacafc837904ea18b2
Signed-off-by: Haichao Liu <liuhaic...@bytedance.com>
---
 device/peer.go    | 6 +++++-
 device/receive.go | 2 ++
 device/send.go    | 4 ++++
 3 files changed, 11 insertions(+), 1 deletion(-)

diff --git a/device/peer.go b/device/peer.go
index ef6c010..78204bb 100644
--- a/device/peer.go
+++ b/device/peer.go
@@ -58,6 +58,7 @@ type Peer struct {
        }
 
        queue struct {
+               sync.RWMutex
                nonce                           chan *QueueOutboundElement // 
nonce / pre-handshake queue
                outbound                        chan *QueueOutboundElement // 
sequential ordering of work
                inbound                         chan *QueueInboundElement  // 
sequential ordering of work
@@ -195,10 +196,11 @@ func (peer *Peer) Start() {
        peer.routines.stopping.Add(PeerRoutineNumber)
 
        // prepare queues
-
+       peer.queue.Lock()
        peer.queue.nonce = make(chan *QueueOutboundElement, QueueOutboundSize)
        peer.queue.outbound = make(chan *QueueOutboundElement, 
QueueOutboundSize)
        peer.queue.inbound = make(chan *QueueInboundElement, QueueInboundSize)
+       peer.queue.Unlock()
 
        peer.timersInit()
        peer.handshake.lastSentHandshake = time.Now().Add(-(RekeyTimeout + 
time.Second))
@@ -284,9 +286,11 @@ func (peer *Peer) Stop() {
 
        // close queues
 
+       peer.queue.Lock()
        close(peer.queue.nonce)
        close(peer.queue.outbound)
        close(peer.queue.inbound)
+       peer.queue.Unlock()
 
        peer.ZeroAndFlushAll()
 }
diff --git a/device/receive.go b/device/receive.go
index b53c9c0..e4a94b5 100644
--- a/device/receive.go
+++ b/device/receive.go
@@ -184,11 +184,13 @@ func (device *Device) RoutineReceiveIncoming(IP int, bind 
conn.Bind) {
 
                        // add to decryption queues
 
+                       peer.queue.RLock()
                        if peer.isRunning.Get() {
                                if 
device.addToInboundAndDecryptionQueues(peer.queue.inbound, 
device.queue.decryption, elem) {
                                        buffer = device.GetMessageBuffer()
                                }
                        }
+                       peer.queue.RUnlock()
 
                        continue
 
diff --git a/device/send.go b/device/send.go
index c0bdba3..d202b62 100644
--- a/device/send.go
+++ b/device/send.go
@@ -107,6 +107,8 @@ func addToOutboundAndEncryptionQueues(outboundQueue chan 
*QueueOutboundElement,
 /* Queues a keepalive if no packets are queued for peer
  */
 func (peer *Peer) SendKeepalive() bool {
+       peer.queue.RLock()
+       defer peer.queue.RUnlock()
        if len(peer.queue.nonce) != 0 || 
peer.queue.packetInNonceQueueIsAwaitingKey.Get() || !peer.isRunning.Get() {
                return false
        }
@@ -310,6 +312,7 @@ func (device *Device) RoutineReadFromTUN() {
 
                // insert into nonce/pre-handshake queue
 
+               peer.queue.RLock()
                if peer.isRunning.Get() {
                        if peer.queue.packetInNonceQueueIsAwaitingKey.Get() {
                                peer.SendHandshakeInitiation(false)
@@ -317,6 +320,7 @@ func (device *Device) RoutineReadFromTUN() {
                        addToNonceQueue(peer.queue.nonce, elem, device)
                        elem = nil
                }
+               peer.queue.RUnlock()
        }
 }
 
-- 
2.19.1