"golang.zx2c4.com/wireguard/conn"
)
-const (
- PeerRoutineNumber = 2
-)
-
type Peer struct {
isRunning AtomicBool
sync.RWMutex // Mostly protects endpoint, but is generally taken whenever we modify peer
sentLastMinuteHandshake AtomicBool
}
- signals struct {
- newKeypairArrived chan struct{}
- flushNonceQueue chan 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
- packetInNonceQueueIsAwaitingKey AtomicBool
+ staged chan *QueueOutboundElement // staged packets before a handshake is available
+ outbound chan *QueueOutboundElement // sequential ordering of work
+ inbound chan *QueueInboundElement // sequential ordering of work
}
routines struct {
peer.routines.stopping.Wait()
peer.routines.stop = make(chan struct{})
- peer.routines.stopping.Add(PeerRoutineNumber)
+ peer.routines.stopping.Add(1)
// prepare queues
peer.queue.Lock()
- peer.queue.nonce = make(chan *QueueOutboundElement, QueueOutboundSize)
+ peer.queue.staged = make(chan *QueueOutboundElement, QueueStagedSize)
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))
- peer.signals.newKeypairArrived = make(chan struct{}, 1)
- peer.signals.flushNonceQueue = make(chan struct{}, 1)
// wait for routines to start
- // RoutineNonce writes to the encryption queue; keep it alive until we are done.
- device.queue.encryption.wg.Add(1)
- go peer.RoutineNonce()
go peer.RoutineSequentialSender()
go peer.RoutineSequentialReceiver()
handshake.Clear()
handshake.mutex.Unlock()
- peer.FlushNonceQueue()
+ peer.FlushStagedPackets()
}
func (peer *Peer) ExpireCurrentKeypairs() {
// close queues
peer.queue.Lock()
- close(peer.queue.nonce)
close(peer.queue.inbound)
+ close(peer.queue.outbound)
peer.queue.Unlock()
peer.ZeroAndFlushAll()
elem.peer = nil
}
-func addToNonceQueue(queue chan *QueueOutboundElement, elem *QueueOutboundElement, device *Device) {
- for {
- select {
- case queue <- elem:
- return
- default:
- select {
- case old := <-queue:
- device.PutMessageBuffer(old.buffer)
- device.PutOutboundElement(old)
- default:
- }
- }
- }
-}
-
/* Queues a keepalive if no packets are queued for peer
*/
-func (peer *Peer) SendKeepalive() bool {
+func (peer *Peer) SendKeepalive() {
+ var elem *QueueOutboundElement
peer.queue.RLock()
- defer peer.queue.RUnlock()
- if len(peer.queue.nonce) != 0 || peer.queue.packetInNonceQueueIsAwaitingKey.Get() || !peer.isRunning.Get() {
- return false
+ if len(peer.queue.staged) != 0 || !peer.isRunning.Get() {
+ goto out
}
- elem := peer.device.NewOutboundElement()
+ elem = peer.device.NewOutboundElement()
elem.packet = nil
select {
- case peer.queue.nonce <- elem:
+ case peer.queue.staged <- elem:
peer.device.log.Verbosef("%v - Sending keepalive packet", peer)
- return true
default:
peer.device.PutMessageBuffer(elem.buffer)
peer.device.PutOutboundElement(elem)
- return false
}
+out:
+ peer.queue.RUnlock()
+ peer.SendStagedPackets()
}
func (peer *Peer) SendHandshakeInitiation(isRetry bool) error {
}
/* Reads packets from the TUN and inserts
- * into nonce queue for peer
+ * into staged queue for peer
*
* Obs. Single instance per TUN device
*/
if peer == nil {
continue
}
-
- // insert into nonce/pre-handshake queue
-
- peer.queue.RLock()
if peer.isRunning.Get() {
- if peer.queue.packetInNonceQueueIsAwaitingKey.Get() {
- peer.SendHandshakeInitiation(false)
- }
- addToNonceQueue(peer.queue.nonce, elem, device)
+ peer.StagePacket(elem)
elem = nil
+ peer.SendStagedPackets()
}
- peer.queue.RUnlock()
}
}
-func (peer *Peer) FlushNonceQueue() {
- select {
- case peer.signals.flushNonceQueue <- struct{}{}:
- default:
- }
-}
-
-/* Queues packets when there is no handshake.
- * Then assigns nonces to packets sequentially
- * and creates "work" structs for workers
- *
- * Obs. A single instance per peer
- */
-func (peer *Peer) RoutineNonce() {
- var keypair *Keypair
- device := peer.device
-
- flush := func() {
- for {
+func (peer *Peer) StagePacket(elem *QueueOutboundElement) {
+ for {
+ select {
+ case peer.queue.staged <- elem:
+ return
+ default:
select {
- case elem := <-peer.queue.nonce:
- device.PutMessageBuffer(elem.buffer)
- device.PutOutboundElement(elem)
+ case tooOld := <-peer.queue.staged:
+ peer.device.PutMessageBuffer(tooOld.buffer)
+ peer.device.PutOutboundElement(tooOld)
default:
- return
}
}
}
+}
- defer func() {
- flush()
- device.log.Verbosef("%v - Routine: nonce worker - stopped", peer)
- peer.queue.packetInNonceQueueIsAwaitingKey.Set(false)
- device.queue.encryption.wg.Done() // no more writes from us
- close(peer.queue.outbound) // no more writes to this channel
- peer.routines.stopping.Done()
- }()
+func (peer *Peer) SendStagedPackets() {
+top:
+ if len(peer.queue.staged) == 0 || !peer.device.isUp.Get() {
+ return
+ }
- device.log.Verbosef("%v - Routine: nonce worker - started", peer)
+ keypair := peer.keypairs.Current()
+ if keypair == nil || atomic.LoadUint64(&keypair.sendNonce) >= RejectAfterMessages || time.Since(keypair.created) >= RejectAfterTime {
+ peer.SendHandshakeInitiation(false)
+ return
+ }
+ peer.device.queue.encryption.wg.Add(1)
+ defer peer.device.queue.encryption.wg.Done()
-NextPacket:
for {
- peer.queue.packetInNonceQueueIsAwaitingKey.Set(false)
-
select {
- case <-peer.routines.stop:
- return
-
- case <-peer.signals.flushNonceQueue:
- flush()
- continue NextPacket
-
- case elem, ok := <-peer.queue.nonce:
-
- if !ok {
- return
- }
-
- // make sure to always pick the newest key
-
- for {
-
- // check validity of newest key pair
-
- keypair = peer.keypairs.Current()
- if keypair != nil && atomic.LoadUint64(&keypair.sendNonce) < RejectAfterMessages {
- if time.Since(keypair.created) < RejectAfterTime {
- break
- }
- }
- peer.queue.packetInNonceQueueIsAwaitingKey.Set(true)
-
- // no suitable key pair, request for new handshake
-
- select {
- case <-peer.signals.newKeypairArrived:
- default:
- }
-
- peer.SendHandshakeInitiation(false)
-
- // wait for key to be established
-
- device.log.Verbosef("%v - Awaiting keypair", peer)
-
- select {
- case <-peer.signals.newKeypairArrived:
- device.log.Verbosef("%v - Obtained awaited keypair", peer)
-
- case <-peer.signals.flushNonceQueue:
- device.PutMessageBuffer(elem.buffer)
- device.PutOutboundElement(elem)
- flush()
- continue NextPacket
-
- case <-peer.routines.stop:
- device.PutMessageBuffer(elem.buffer)
- device.PutOutboundElement(elem)
- return
- }
- }
- peer.queue.packetInNonceQueueIsAwaitingKey.Set(false)
-
- // populate work element
-
+ case elem := <-peer.queue.staged:
elem.peer = peer
elem.nonce = atomic.AddUint64(&keypair.sendNonce, 1) - 1
-
- // double check in case of race condition added by future code
-
if elem.nonce >= RejectAfterMessages {
atomic.StoreUint64(&keypair.sendNonce, RejectAfterMessages)
- device.PutMessageBuffer(elem.buffer)
- device.PutOutboundElement(elem)
- continue NextPacket
+ peer.StagePacket(elem) // XXX: Out of order, but we can't front-load go chans
+ goto top
}
elem.keypair = keypair
// add to parallel and sequential queue
peer.queue.outbound <- elem
- device.queue.encryption.c <- elem
+ peer.device.queue.encryption.c <- elem
+ default:
+ return
+ }
+ }
+}
+
+func (peer *Peer) FlushStagedPackets() {
+ for {
+ select {
+ case elem := <-peer.queue.staged:
+ peer.device.PutMessageBuffer(elem.buffer)
+ peer.device.PutOutboundElement(elem)
+ default:
+ return
}
}
}
projects / thirdparty / wireguard-go.git / commitdiff
