Change queueing drop order and fix memory leaks

If the queues are full, we drop the present packet, which is better for
network traffic flow. Also, we try to fix up the memory leaks with not
putting buffers from our shared pool.

receive.go

@@ -43,59 +43,28 @@ func (elem *QueueInboundElement) IsDropped() bool {
 	return atomic.LoadInt32(&elem.dropped) == AtomicTrue
 }
 
-func (device *Device) addToInboundQueue(
+func (device *Device) addToInboundAndDecryptionQueues(inboundQueue chan *QueueInboundElement, decryptionQueue chan *QueueInboundElement, element *QueueInboundElement) bool {
-	queue chan *QueueInboundElement,
+	select {
-	element *QueueInboundElement,
+	case inboundQueue <- element:
-) {
-	for {
 		select {
-		case queue <- element:
+		case decryptionQueue <- element:
-			return
+			return true
 		default:
-			select {
+			element.Drop()
-			case old := <-queue:
+			element.mutex.Unlock()
-				old.Drop()
+			return false
-			default:
-			}
 		}
+	default:
+		return false
 	}
 }
 
-func (device *Device) addToDecryptionQueue(
+func (device *Device) addToHandshakeQueue(queue chan QueueHandshakeElement, element QueueHandshakeElement) bool {
-	queue chan *QueueInboundElement,
+	select {
-	element *QueueInboundElement,
+	case queue <- element:
-) {
+		return true
-	for {
+	default:
-		select {
+		return false
-		case queue <- element:
-			return
-		default:
-			select {
-			case old := <-queue:
-				// drop & release to potential consumer
-				old.Drop()
-				old.mutex.Unlock()
-			default:
-			}
-		}
-	}
-}
-
-func (device *Device) addToHandshakeQueue(
-	queue chan QueueHandshakeElement,
-	element QueueHandshakeElement,
-) {
-	for {
-		select {
-		case queue <- element:
-			return
-		default:
-			select {
-			case elem := <-queue:
-				device.PutMessageBuffer(elem.buffer)
-			default:
-			}
-		}
 	}
 }
 
@@ -154,6 +123,7 @@ func (device *Device) RoutineReceiveIncoming(IP int, bind Bind) {
 		}
 
 		if err != nil {
+			device.PutMessageBuffer(buffer)
 			return
 		}
 
@@ -212,9 +182,9 @@ func (device *Device) RoutineReceiveIncoming(IP int, bind Bind) {
 			// add to decryption queues
 
 			if peer.isRunning.Get() {
-				device.addToDecryptionQueue(device.queue.decryption, elem)
+				if device.addToInboundAndDecryptionQueues(peer.queue.inbound, device.queue.decryption, elem) {
-				device.addToInboundQueue(peer.queue.inbound, elem)
+					buffer = device.GetMessageBuffer()
-				buffer = device.GetMessageBuffer()
+				}
 			}
 
 			continue
@@ -235,7 +205,7 @@ func (device *Device) RoutineReceiveIncoming(IP int, bind Bind) {
 		}
 
 		if okay {
-			device.addToHandshakeQueue(
+			if (device.addToHandshakeQueue(
 				device.queue.handshake,
 				QueueHandshakeElement{
 					msgType:  msgType,
@@ -243,8 +213,9 @@ func (device *Device) RoutineReceiveIncoming(IP int, bind Bind) {
 					packet:   packet,
 					endpoint: endpoint,
 				},
-			)
+			)) {
-			buffer = device.GetMessageBuffer()
+				buffer = device.GetMessageBuffer()
+			}
 		}
 	}
 }
@@ -307,6 +278,8 @@ func (device *Device) RoutineDecryption() {
 			)
 			if err != nil {
 				elem.Drop()
+				device.PutMessageBuffer(elem.buffer)
+				elem.mutex.Unlock()
 			}
 			elem.mutex.Unlock()
 		}

send.go

@@ -66,10 +66,7 @@ func (elem *QueueOutboundElement) IsDropped() bool {
 	return atomic.LoadInt32(&elem.dropped) == AtomicTrue
 }
 
-func addToOutboundQueue(
+func addToNonceQueue(queue chan *QueueOutboundElement, element *QueueOutboundElement, device *Device) {
-	queue chan *QueueOutboundElement,
-	element *QueueOutboundElement,
-) {
 	for {
 		select {
 		case queue <- element:
@@ -78,32 +75,30 @@ func addToOutboundQueue(
 			select {
 			case old := <-queue:
 				old.Drop()
+				device.PutMessageBuffer(element.buffer)
 			default:
 			}
 		}
 	}
 }
 
-func addToEncryptionQueue(
+func addToOutboundAndEncryptionQueues(outboundQueue chan *QueueOutboundElement, encryptionQueue chan *QueueOutboundElement, element *QueueOutboundElement) {
-	queue chan *QueueOutboundElement,
+	select {
-	element *QueueOutboundElement,
+	case outboundQueue <- element:
-) {
-	for {
 		select {
-		case queue <- element:
+		case encryptionQueue <- element:
 			return
 		default:
-			select {
+			element.Drop()
-			case old := <-queue:
+			element.peer.device.PutMessageBuffer(element.buffer)
-				// drop & release to potential consumer
+			element.mutex.Unlock()
-				old.Drop()
-				old.mutex.Unlock()
-			default:
-			}
 		}
+	default:
+		element.peer.device.PutMessageBuffer(element.buffer)
 	}
 }
 
+
 /* Queues a keepalive if no packets are queued for peer
  */
 func (peer *Peer) SendKeepalive() bool {
@@ -117,6 +112,7 @@ func (peer *Peer) SendKeepalive() bool {
 		peer.device.log.Debug.Println(peer, "- Sending keepalive packet")
 		return true
 	default:
+		peer.device.PutMessageBuffer(elem.buffer)
 		return false
 	}
 }
@@ -267,6 +263,7 @@ func (device *Device) RoutineReadFromTUN() {
 				logError.Println("Failed to read packet from TUN device:", err)
 				device.Close()
 			}
+			device.PutMessageBuffer(elem.buffer)
 			return
 		}
 
@@ -308,7 +305,7 @@ func (device *Device) RoutineReadFromTUN() {
 			if peer.queue.packetInNonceQueueIsAwaitingKey.Get() {
 				peer.SendHandshakeInitiation(false)
 			}
-			addToOutboundQueue(peer.queue.nonce, elem)
+			addToNonceQueue(peer.queue.nonce, elem, device)
 			elem = device.NewOutboundElement()
 		}
 	}
@@ -342,7 +339,8 @@ func (peer *Peer) RoutineNonce() {
 	flush := func() {
 		for {
 			select {
-			case <-peer.queue.nonce:
+			case elem := <-peer.queue.nonce:
+				device.PutMessageBuffer(elem.buffer)
 			default:
 				return
 			}
@@ -402,10 +400,12 @@ func (peer *Peer) RoutineNonce() {
 					logDebug.Println(peer, "- Obtained awaited keypair")
 
 				case <-peer.signals.flushNonceQueue:
+					device.PutMessageBuffer(elem.buffer)
 					flush()
 					goto NextPacket
 
 				case <-peer.routines.stop:
+					device.PutMessageBuffer(elem.buffer)
 					return
 				}
 			}
@@ -420,6 +420,7 @@ func (peer *Peer) RoutineNonce() {
 
 			if elem.nonce >= RejectAfterMessages {
 				atomic.StoreUint64(&keypair.sendNonce, RejectAfterMessages)
+				device.PutMessageBuffer(elem.buffer)
 				goto NextPacket
 			}
 
@@ -428,9 +429,7 @@ func (peer *Peer) RoutineNonce() {
 			elem.mutex.Lock()
 
 			// add to parallel and sequential queue
-
+			addToOutboundAndEncryptionQueues(peer.queue.outbound, device.queue.encryption, elem)
-			addToEncryptionQueue(device.queue.encryption, elem)
-			addToOutboundQueue(peer.queue.outbound, elem)
 		}
 	}
 }