Revert "Peer timer teardown"

This reverts commit d73f960aab86b9a12b0b7d18aa80ce1d4f130695.

Problems with deadlocking arises due to orphaned packets in per-peer
queues. Additional work on this issue continues in seperate branch.

diff --git a/src/conn.go b/src/conn.go
index ddb7ed1b3227686468dbfa3518f969dcc653b679..6d292d36f74af8648e58bcb374ca2f5c7fa6dfe1 100644 (file)
--- a/src/conn.go
+++ b/src/conn.go
@@ -82,7 +82,7 @@ func updateBind(device *Device) error {
 
        // open new sockets
 
-       if device.isUp.Get() {
+       if device.tun.isUp.Get() {
 
                device.log.Debug.Println("UDP bind updating")
 

diff --git a/src/device.go b/src/device.go
index f4a087c3c053c876eb76ffa17de240508a1fe4ba..a3461adca1b6a74293fffc610bf91440c8e97bac 100644 (file)
--- a/src/device.go
+++ b/src/device.go
@@ -8,13 +8,13 @@ import (
 )
 
 type Device struct {
-       isUp      AtomicBool // device is up (TUN interface up)?
-       isClosed  AtomicBool // device is closed? (acting as guard)
+       closed    AtomicBool // device is closed? (acting as guard)
        log       *Logger    // collection of loggers for levels
        idCounter uint       // for assigning debug ids to peers
        fwMark    uint32
        tun       struct {
                device TUNDevice
+               isUp   AtomicBool
                mtu    int32
        }
        pool struct {
@@ -45,28 +45,6 @@ type Device struct {
        mac            CookieChecker
 }
 
-func (device *Device) Up() {
-       device.mutex.Lock()
-       defer device.mutex.Unlock()
-
-       device.isUp.Set(true)
-       updateBind(device)
-       for _, peer := range device.peers {
-               peer.Start()
-       }
-}
-
-func (device *Device) Down() {
-       device.mutex.Lock()
-       defer device.mutex.Unlock()
-
-       device.isUp.Set(false)
-       closeBind(device)
-       for _, peer := range device.peers {
-               peer.Stop()
-       }
-}
-
 /* Warning:
  * The caller must hold the device mutex (write lock)
  */
@@ -76,9 +54,9 @@ func removePeerUnsafe(device *Device, key NoisePublicKey) {
                return
        }
        peer.mutex.Lock()
-       peer.Stop()
        device.routingTable.RemovePeer(peer)
        delete(device.peers, key)
+       peer.Close()
 }
 
 func (device *Device) IsUnderLoad() bool {
@@ -120,7 +98,7 @@ func (device *Device) SetPrivateKey(sk NoisePrivateKey) error {
        device.publicKey = publicKey
        device.mac.Init(publicKey)
 
-       // do DH pre-computations
+       // do DH precomputations
 
        rmKey := device.privateKey.IsZero()
 
@@ -154,12 +132,10 @@ func NewDevice(tun TUNDevice, logger *Logger) *Device {
        device.mutex.Lock()
        defer device.mutex.Unlock()
 
-       device.isUp.Set(false)
-       device.isClosed.Set(false)
-
        device.log = logger
        device.peers = make(map[NoisePublicKey]*Peer)
        device.tun.device = tun
+       device.tun.isUp.Set(false)
 
        device.indices.Init()
        device.ratelimiter.Init()
@@ -220,13 +196,17 @@ func (device *Device) RemovePeer(key NoisePublicKey) {
 func (device *Device) RemoveAllPeers() {
        device.mutex.Lock()
        defer device.mutex.Unlock()
-       for key := range device.peers {
-               removePeerUnsafe(device, key)
+
+       for key, peer := range device.peers {
+               peer.mutex.Lock()
+               delete(device.peers, key)
+               peer.Close()
+               peer.mutex.Unlock()
        }
 }
 
 func (device *Device) Close() {
-       if device.isClosed.Swap(true) {
+       if device.closed.Swap(true) {
                return
        }
        device.log.Info.Println("Closing device")

diff --git a/src/peer.go b/src/peer.go
index 7c6ad476697ad21bb7d5a20c39ed21a119128116..f582556301813ca8e87ae8fa8c9e10456cfe9b9c 100644 (file)
--- a/src/peer.go
+++ b/src/peer.go
@@ -34,15 +34,15 @@ type Peer struct {
                flushNonceQueue    Signal // size 1, empty queued packets
                messageSend        Signal // size 1, message was send to peer
                messageReceived    Signal // size 1, authenticated message recv
-               stop               Signal // size 0, stop all goroutines in peer
+               stop               Signal // size 0, stop all goroutines
        }
        timer struct {
                // state related to WireGuard timers
 
                keepalivePersistent Timer // set for persistent keepalives
                keepalivePassive    Timer // set upon recieving messages
+               newHandshake        Timer // begin a new handshake (stale)
                zeroAllKeys         Timer // zero all key material
-               handshakeNew        Timer // begin a new handshake (stale)
                handshakeDeadline   Timer // complete handshake timeout
                handshakeTimeout    Timer // current handshake message timeout
 
@@ -69,8 +69,8 @@ func (device *Device) NewPeer(pk NoisePublicKey) (*Peer, error) {
 
        peer.timer.keepalivePersistent = NewTimer()
        peer.timer.keepalivePassive = NewTimer()
+       peer.timer.newHandshake = NewTimer()
        peer.timer.zeroAllKeys = NewTimer()
-       peer.timer.handshakeNew = NewTimer()
        peer.timer.handshakeDeadline = NewTimer()
        peer.timer.handshakeTimeout = NewTimer()
 
@@ -116,29 +116,32 @@ func (device *Device) NewPeer(pk NoisePublicKey) (*Peer, error) {
 
        // prepare signaling & routines
 
+       peer.signal.stop = NewSignal()
        peer.signal.newKeyPair = NewSignal()
        peer.signal.handshakeBegin = NewSignal()
        peer.signal.handshakeCompleted = NewSignal()
        peer.signal.flushNonceQueue = NewSignal()
 
+       go peer.RoutineNonce()
+       go peer.RoutineTimerHandler()
+       go peer.RoutineSequentialSender()
+       go peer.RoutineSequentialReceiver()
+
        return peer, nil
 }
 
 func (peer *Peer) SendBuffer(buffer []byte) error {
        peer.device.net.mutex.RLock()
        defer peer.device.net.mutex.RUnlock()
-
        peer.mutex.RLock()
        defer peer.mutex.RUnlock()
-
        if peer.endpoint == nil {
                return errors.New("No known endpoint for peer")
        }
-
        return peer.device.net.bind.Send(buffer, peer.endpoint)
 }
 
-/* Returns a short string identifier for logging
+/* Returns a short string identification for logging
  */
 func (peer *Peer) String() string {
        if peer.endpoint == nil {
@@ -156,32 +159,6 @@ func (peer *Peer) String() string {
        )
 }
 
-/* Starts all routines for a given peer
- *
- * Requires that the caller holds the exclusive peer lock!
- */
-func unsafePeerStart(peer *Peer) {
-       peer.signal.stop.Broadcast()
-       peer.signal.stop = NewSignal()
-
-       var wait sync.WaitGroup
-
-       wait.Add(1)
-
-       go peer.RoutineNonce()
-       go peer.RoutineTimerHandler(&wait)
-       go peer.RoutineSequentialSender()
-       go peer.RoutineSequentialReceiver()
-
-       wait.Wait()
-}
-
-func (peer *Peer) Start() {
-       peer.mutex.Lock()
-       unsafePeerStart(peer)
-       peer.mutex.Unlock()
-}
-
-func (peer *Peer) Stop() {
+func (peer *Peer) Close() {
        peer.signal.stop.Broadcast()
 }

diff --git a/src/timer.go b/src/timer.go
index f00ca499ed72be3766eb0bfab3802ca0698e88a2..3def2533112c64df0bdc8272ff26f8a26378a200 100644 (file)
--- a/src/timer.go
+++ b/src/timer.go
@@ -43,6 +43,12 @@ func (t *Timer) Reset(dur time.Duration) {
        t.Start(dur)
 }
 
+func (t *Timer) Push(dur time.Duration) {
+       if t.pending.Get() {
+               t.Reset(dur)
+       }
+}
+
 func (t *Timer) Wait() <-chan time.Time {
        return t.timer.C
 }

diff --git a/src/timers.go b/src/timers.go
index f2fed30ec28e101833a686b6b41078312114867d..ee47393cb28a6eb4a8b9dc1c5a375b47ea099573 100644 (file)
--- a/src/timers.go
+++ b/src/timers.go
@@ -4,17 +4,10 @@ import (
        "bytes"\r
        "encoding/binary"\r
        "math/rand"\r
-       "sync"\r
        "sync/atomic"\r
        "time"\r
 )\r
 \r
-/* NOTE:\r
- * Notion of validity\r
- *\r
- *\r
- */\r
-\r
 /* Called when a new authenticated message has been send\r
  *\r
  */\r
@@ -51,25 +44,25 @@ func (peer *Peer) KeepKeyFreshReceiving() {
        send := nonce > RekeyAfterMessages || time.Now().Sub(kp.created) > RekeyAfterTimeReceiving\r
        if send {\r
                // do a last minute attempt at initiating a new handshake\r
-               peer.timer.sendLastMinuteHandshake = true\r
                peer.signal.handshakeBegin.Send()\r
+               peer.timer.sendLastMinuteHandshake = true\r
        }\r
 }\r
 \r
 /* Queues a keep-alive if no packets are queued for peer\r
  */\r
 func (peer *Peer) SendKeepAlive() bool {\r
-       if len(peer.queue.nonce) != 0 {\r
-               return false\r
-       }\r
        elem := peer.device.NewOutboundElement()\r
        elem.packet = nil\r
-       select {\r
-       case peer.queue.nonce <- elem:\r
-               return true\r
-       default:\r
-               return false\r
+       if len(peer.queue.nonce) == 0 {\r
+               select {\r
+               case peer.queue.nonce <- elem:\r
+                       return true\r
+               default:\r
+                       return false\r
+               }\r
        }\r
+       return true\r
 }\r
 \r
 /* Event:\r
@@ -77,7 +70,9 @@ func (peer *Peer) SendKeepAlive() bool {
  */\r
 func (peer *Peer) TimerDataSent() {\r
        peer.timer.keepalivePassive.Stop()\r
-       peer.timer.handshakeNew.Start(NewHandshakeTime)\r
+       if peer.timer.newHandshake.Pending() {\r
+               peer.timer.newHandshake.Reset(NewHandshakeTime)\r
+       }\r
 }\r
 \r
 /* Event:\r
@@ -96,7 +91,7 @@ func (peer *Peer) TimerDataReceived() {
  * Any (authenticated) packet received\r
  */\r
 func (peer *Peer) TimerAnyAuthenticatedPacketReceived() {\r
-       peer.timer.handshakeNew.Stop()\r
+       peer.timer.newHandshake.Stop()\r
 }\r
 \r
 /* Event:\r
@@ -120,6 +115,10 @@ func (peer *Peer) TimerAnyAuthenticatedPacketTraversal() {
  * - First transport message under the "next" key\r
  */\r
 func (peer *Peer) TimerHandshakeComplete() {\r
+       atomic.StoreInt64(\r
+               &peer.stats.lastHandshakeNano,\r
+               time.Now().UnixNano(),\r
+       )\r
        peer.signal.handshakeCompleted.Send()\r
        peer.device.log.Info.Println("Negotiated new handshake for", peer.String())\r
 }\r
@@ -140,75 +139,13 @@ func (peer *Peer) TimerEphemeralKeyCreated() {
        peer.timer.zeroAllKeys.Reset(RejectAfterTime * 3)\r
 }\r
 \r
-/* Sends a new handshake initiation message to the peer (endpoint)\r
- */\r
-func (peer *Peer) sendNewHandshake() error {\r
-\r
-       // temporarily disable the handshake complete signal\r
-\r
-       peer.signal.handshakeCompleted.Disable()\r
-\r
-       // create initiation message\r
-\r
-       msg, err := peer.device.CreateMessageInitiation(peer)\r
-       if err != nil {\r
-               return err\r
-       }\r
-\r
-       // marshal handshake message\r
-\r
-       var buff [MessageInitiationSize]byte\r
-       writer := bytes.NewBuffer(buff[:0])\r
-       binary.Write(writer, binary.LittleEndian, msg)\r
-       packet := writer.Bytes()\r
-       peer.mac.AddMacs(packet)\r
-\r
-       // send to endpoint\r
-\r
-       peer.TimerAnyAuthenticatedPacketTraversal()\r
-\r
-       err = peer.SendBuffer(packet)\r
-       if err == nil {\r
-               peer.signal.handshakeCompleted.Enable()\r
-       }\r
-\r
-       // set timeout\r
-\r
-       jitter := time.Millisecond * time.Duration(rand.Uint32()%334)\r
-\r
-       peer.timer.keepalivePassive.Stop()\r
-       peer.timer.handshakeTimeout.Reset(RekeyTimeout + jitter)\r
-\r
-       return err\r
-}\r
-\r
-func (peer *Peer) RoutineTimerHandler(ready *sync.WaitGroup) {\r
+func (peer *Peer) RoutineTimerHandler() {\r
        device := peer.device\r
 \r
        logInfo := device.log.Info\r
        logDebug := device.log.Debug\r
        logDebug.Println("Routine, timer handler, started for peer", peer.String())\r
 \r
-       // reset all timers\r
-\r
-       peer.timer.keepalivePassive.Stop()\r
-       peer.timer.handshakeDeadline.Stop()\r
-       peer.timer.handshakeTimeout.Stop()\r
-       peer.timer.handshakeNew.Stop()\r
-       peer.timer.zeroAllKeys.Stop()\r
-\r
-       interval := atomic.LoadUint64(&peer.persistentKeepaliveInterval)\r
-       if interval > 0 {\r
-               duration := time.Duration(interval) * time.Second\r
-               peer.timer.keepalivePersistent.Reset(duration)\r
-       }\r
-\r
-       // signal that timers are reset\r
-\r
-       ready.Done()\r
-\r
-       // handle timer events\r
-\r
        for {\r
                select {\r
 \r
@@ -221,7 +158,6 @@ func (peer *Peer) RoutineTimerHandler(ready *sync.WaitGroup) {
                        interval := atomic.LoadUint64(&peer.persistentKeepaliveInterval)\r
                        if interval > 0 {\r
                                logDebug.Println("Sending keep-alive to", peer.String())\r
-                               peer.timer.keepalivePassive.Stop()\r
                                peer.SendKeepAlive()\r
                        }\r
 \r
@@ -232,8 +168,8 @@ func (peer *Peer) RoutineTimerHandler(ready *sync.WaitGroup) {
                        peer.SendKeepAlive()\r
 \r
                        if peer.timer.needAnotherKeepalive {\r
-                               peer.timer.needAnotherKeepalive = false\r
                                peer.timer.keepalivePassive.Reset(KeepaliveTimeout)\r
+                               peer.timer.needAnotherKeepalive = false\r
                        }\r
 \r
                // clear key material timer\r
@@ -277,7 +213,7 @@ func (peer *Peer) RoutineTimerHandler(ready *sync.WaitGroup) {
 \r
                // handshake timers\r
 \r
-               case <-peer.timer.handshakeNew.Wait():\r
+               case <-peer.timer.newHandshake.Wait():\r
                        logInfo.Println("Retrying handshake with", peer.String())\r
                        peer.signal.handshakeBegin.Send()\r
 \r
@@ -332,16 +268,48 @@ func (peer *Peer) RoutineTimerHandler(ready *sync.WaitGroup) {
                        logInfo.Println(\r
                                "Handshake completed for:", peer.String())\r
 \r
-                       atomic.StoreInt64(\r
-                               &peer.stats.lastHandshakeNano,\r
-                               time.Now().UnixNano(),\r
-                       )\r
-\r
                        peer.timer.handshakeTimeout.Stop()\r
                        peer.timer.handshakeDeadline.Stop()\r
                        peer.signal.handshakeBegin.Enable()\r
-\r
-                       peer.timer.sendLastMinuteHandshake = false\r
                }\r
        }\r
 }\r
+\r
+/* Sends a new handshake initiation message to the peer (endpoint)\r
+ */\r
+func (peer *Peer) sendNewHandshake() error {\r
+\r
+       // temporarily disable the handshake complete signal\r
+\r
+       peer.signal.handshakeCompleted.Disable()\r
+\r
+       // create initiation message\r
+\r
+       msg, err := peer.device.CreateMessageInitiation(peer)\r
+       if err != nil {\r
+               return err\r
+       }\r
+\r
+       // marshal handshake message\r
+\r
+       var buff [MessageInitiationSize]byte\r
+       writer := bytes.NewBuffer(buff[:0])\r
+       binary.Write(writer, binary.LittleEndian, msg)\r
+       packet := writer.Bytes()\r
+       peer.mac.AddMacs(packet)\r
+\r
+       // send to endpoint\r
+\r
+       err = peer.SendBuffer(packet)\r
+       if err == nil {\r
+               peer.TimerAnyAuthenticatedPacketTraversal()\r
+               peer.signal.handshakeCompleted.Enable()\r
+       }\r
+\r
+       // set timeout\r
+\r
+       jitter := time.Millisecond * time.Duration(rand.Uint32()%334)\r
+       peer.timer.handshakeTimeout.Reset(RekeyTimeout + jitter)\r
+\r
+       return err\r
+}\r

diff --git a/src/tun.go b/src/tun.go
index 024f0f07133bcadaddb811ec6bca4932ff5aaaad..54253b4f25dba54080bc770ef4914929940fc0d8 100644 (file)
--- a/src/tun.go
+++ b/src/tun.go
@@ -46,13 +46,21 @@ func (device *Device) RoutineTUNEventReader() {
                }
 
                if event&TUNEventUp != 0 {
-                       logInfo.Println("Interface set up")
-                       device.Up()
+                       if !device.tun.isUp.Get() {
+                               // begin listening for incomming datagrams
+                               logInfo.Println("Interface set up")
+                               device.tun.isUp.Set(true)
+                               updateBind(device)
+                       }
                }
 
                if event&TUNEventDown != 0 {
-                       logInfo.Println("Interface set down")
-                       device.Up()
+                       if device.tun.isUp.Get() {
+                               // stop listening for incomming datagrams
+                               logInfo.Println("Interface set down")
+                               device.tun.isUp.Set(false)
+                               closeBind(device)
+                       }
                }
        }
 }

diff --git a/src/uapi.go b/src/uapi.go
index a67bff1424578ca11d3f07ec0790568643808b1f..155f4833e3125fddc8288ceb86b0e404cf1630d3 100644 (file)
--- a/src/uapi.go
+++ b/src/uapi.go
@@ -296,7 +296,7 @@ func ipcSetOperation(device *Device, socket *bufio.ReadWriter) *IPCError {
                                                logError.Println("Failed to get tun device status:", err)
                                                return &IPCError{Code: ipcErrorIO}
                                        }
-                                       if device.isUp.Get() && !dummy {
+                                       if device.tun.isUp.Get() && !dummy {
                                                peer.SendKeepAlive()
                                        }
                                }