Work on UAPI
Cross-platform API (get operation) Handshake initiation creation process Outbound packet flow Fixes from code-review
This commit is contained in:
Mathias Hall-Andersen
227
src/send.go
227
src/send.go
@@ -5,107 +5,159 @@ import (
|
||||
"golang.org/x/crypto/chacha20poly1305"
|
||||
"net"
|
||||
"sync"
|
||||
"time"
|
||||
)
|
||||
|
||||
/* Handles outbound flow
|
||||
*
|
||||
* 1. TUN queue
|
||||
* 2. Routing
|
||||
* 3. Per peer queuing
|
||||
* 4. (work queuing)
|
||||
* 2. Routing (sequential)
|
||||
* 3. Nonce assignment (sequential)
|
||||
* 4. Encryption (parallel)
|
||||
* 5. Transmission (sequential)
|
||||
*
|
||||
* The order of packets (per peer) is maintained.
|
||||
* The functions in this file occure (roughly) in the order packets are processed.
|
||||
*/
|
||||
|
||||
type OutboundWorkQueueElement struct {
|
||||
wg sync.WaitGroup
|
||||
/* A work unit
|
||||
*
|
||||
* The sequential consumers will attempt to take the lock,
|
||||
* workers release lock when they have completed work on the packet.
|
||||
*/
|
||||
type QueueOutboundElement struct {
|
||||
mutex sync.Mutex
|
||||
packet []byte
|
||||
nonce uint64
|
||||
keyPair *KeyPair
|
||||
}
|
||||
|
||||
func (peer *Peer) HandshakeWorker(handshakeQueue []byte) {
|
||||
|
||||
func (peer *Peer) FlushNonceQueue() {
|
||||
elems := len(peer.queue.nonce)
|
||||
for i := 0; i < elems; i += 1 {
|
||||
select {
|
||||
case <-peer.queue.nonce:
|
||||
default:
|
||||
return
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (device *Device) SendPacket(packet []byte) {
|
||||
|
||||
// lookup peer
|
||||
|
||||
var peer *Peer
|
||||
switch packet[0] >> 4 {
|
||||
case IPv4version:
|
||||
dst := packet[IPv4offsetDst : IPv4offsetDst+net.IPv4len]
|
||||
peer = device.routingTable.LookupIPv4(dst)
|
||||
|
||||
case IPv6version:
|
||||
dst := packet[IPv6offsetDst : IPv6offsetDst+net.IPv6len]
|
||||
peer = device.routingTable.LookupIPv6(dst)
|
||||
|
||||
default:
|
||||
device.log.Debug.Println("receieved packet with unknown IP version")
|
||||
return
|
||||
}
|
||||
|
||||
if peer == nil {
|
||||
return
|
||||
}
|
||||
|
||||
// insert into peer queue
|
||||
|
||||
func (peer *Peer) InsertOutbound(elem *QueueOutboundElement) {
|
||||
for {
|
||||
select {
|
||||
case peer.queueOutboundRouting <- packet:
|
||||
case peer.queue.outbound <- elem:
|
||||
default:
|
||||
select {
|
||||
case <-peer.queueOutboundRouting:
|
||||
case <-peer.queue.outbound:
|
||||
default:
|
||||
}
|
||||
continue
|
||||
}
|
||||
break
|
||||
}
|
||||
}
|
||||
|
||||
/* Go routine
|
||||
/* Reads packets from the TUN and inserts
|
||||
* into nonce queue for peer
|
||||
*
|
||||
*
|
||||
* 1. waits for handshake.
|
||||
* 2. assigns key pair & nonce
|
||||
* 3. inserts to working queue
|
||||
*
|
||||
* TODO: avoid dynamic allocation of work queue elements
|
||||
* Obs. Single instance per TUN device
|
||||
*/
|
||||
func (peer *Peer) RoutineOutboundNonceWorker() {
|
||||
func (device *Device) RoutineReadFromTUN(tun TUNDevice) {
|
||||
for {
|
||||
// read packet
|
||||
|
||||
packet := make([]byte, 1<<16) // TODO: Fix & avoid dynamic allocation
|
||||
size, err := tun.Read(packet)
|
||||
if err != nil {
|
||||
device.log.Error.Println("Failed to read packet from TUN device:", err)
|
||||
continue
|
||||
}
|
||||
packet = packet[:size]
|
||||
if len(packet) < IPv4headerSize {
|
||||
device.log.Error.Println("Packet too short, length:", len(packet))
|
||||
continue
|
||||
}
|
||||
|
||||
device.log.Debug.Println("New packet on TUN:", packet) // TODO: Slow debugging, remove.
|
||||
|
||||
// lookup peer
|
||||
|
||||
var peer *Peer
|
||||
switch packet[0] >> 4 {
|
||||
case IPv4version:
|
||||
dst := packet[IPv4offsetDst : IPv4offsetDst+net.IPv4len]
|
||||
peer = device.routingTable.LookupIPv4(dst)
|
||||
|
||||
case IPv6version:
|
||||
dst := packet[IPv6offsetDst : IPv6offsetDst+net.IPv6len]
|
||||
peer = device.routingTable.LookupIPv6(dst)
|
||||
|
||||
default:
|
||||
device.log.Debug.Println("Receieved packet with unknown IP version")
|
||||
return
|
||||
}
|
||||
|
||||
if peer == nil {
|
||||
device.log.Debug.Println("No peer configured for IP")
|
||||
return
|
||||
}
|
||||
|
||||
// insert into nonce/pre-handshake queue
|
||||
|
||||
for {
|
||||
select {
|
||||
case peer.queue.nonce <- packet:
|
||||
default:
|
||||
select {
|
||||
case <-peer.queue.nonce:
|
||||
default:
|
||||
}
|
||||
continue
|
||||
}
|
||||
break
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/* Queues packets when there is no handshake.
|
||||
* Then assigns nonces to packets sequentially
|
||||
* and creates "work" structs for workers
|
||||
*
|
||||
* TODO: Avoid dynamic allocation of work queue elements
|
||||
*
|
||||
* Obs. A single instance per peer
|
||||
*/
|
||||
func (peer *Peer) RoutineNonce() {
|
||||
var packet []byte
|
||||
var keyPair *KeyPair
|
||||
var flushTimer time.Timer
|
||||
|
||||
for {
|
||||
|
||||
// wait for packet
|
||||
|
||||
if packet == nil {
|
||||
packet = <-peer.queueOutboundRouting
|
||||
select {
|
||||
case packet = <-peer.queue.nonce:
|
||||
case <-peer.signal.stopSending:
|
||||
close(peer.queue.outbound)
|
||||
return
|
||||
}
|
||||
}
|
||||
|
||||
// wait for key pair
|
||||
|
||||
for keyPair == nil {
|
||||
flushTimer.Reset(time.Second * 10)
|
||||
// TODO: Handshake or NOP
|
||||
peer.signal.newHandshake <- true
|
||||
select {
|
||||
case <-peer.keyPairs.newKeyPair:
|
||||
keyPair = peer.keyPairs.Current()
|
||||
continue
|
||||
case <-flushTimer.C:
|
||||
size := len(peer.queueOutboundRouting)
|
||||
for i := 0; i < size; i += 1 {
|
||||
<-peer.queueOutboundRouting
|
||||
}
|
||||
case <-peer.signal.flushNonceQueue:
|
||||
peer.FlushNonceQueue()
|
||||
packet = nil
|
||||
continue
|
||||
case <-peer.signal.stopSending:
|
||||
close(peer.queue.outbound)
|
||||
return
|
||||
}
|
||||
break
|
||||
}
|
||||
|
||||
// process current packet
|
||||
@@ -114,14 +166,13 @@ func (peer *Peer) RoutineOutboundNonceWorker() {
|
||||
|
||||
// create work element
|
||||
|
||||
work := new(OutboundWorkQueueElement)
|
||||
work.wg.Add(1)
|
||||
work := new(QueueOutboundElement) // TODO: profile, maybe use pool
|
||||
work.keyPair = keyPair
|
||||
work.packet = packet
|
||||
work.nonce = keyPair.sendNonce
|
||||
work.mutex.Lock()
|
||||
|
||||
packet = nil
|
||||
peer.queueOutbound <- work
|
||||
keyPair.sendNonce += 1
|
||||
|
||||
// drop packets until there is space
|
||||
@@ -129,46 +180,36 @@ func (peer *Peer) RoutineOutboundNonceWorker() {
|
||||
func() {
|
||||
for {
|
||||
select {
|
||||
case peer.device.queueWorkOutbound <- work:
|
||||
case peer.device.queue.encryption <- work:
|
||||
return
|
||||
default:
|
||||
drop := <-peer.device.queueWorkOutbound
|
||||
drop := <-peer.device.queue.encryption
|
||||
drop.packet = nil
|
||||
drop.wg.Done()
|
||||
drop.mutex.Unlock()
|
||||
}
|
||||
}
|
||||
}()
|
||||
peer.queue.outbound <- work
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/* Go routine
|
||||
*
|
||||
* sequentially reads packets from queue and sends to endpoint
|
||||
/* Encrypts the elements in the queue
|
||||
* and marks them for sequential consumption (by releasing the mutex)
|
||||
*
|
||||
* Obs. One instance per core
|
||||
*/
|
||||
func (peer *Peer) RoutineSequential() {
|
||||
for work := range peer.queueOutbound {
|
||||
work.wg.Wait()
|
||||
if work.packet == nil {
|
||||
continue
|
||||
}
|
||||
if peer.endpoint == nil {
|
||||
continue
|
||||
}
|
||||
peer.device.conn.WriteToUDP(work.packet, peer.endpoint)
|
||||
}
|
||||
}
|
||||
|
||||
func (device *Device) RoutineEncryptionWorker() {
|
||||
func (device *Device) RoutineEncryption() {
|
||||
var nonce [chacha20poly1305.NonceSize]byte
|
||||
for work := range device.queueWorkOutbound {
|
||||
for work := range device.queue.encryption {
|
||||
|
||||
// pad packet
|
||||
|
||||
padding := device.mtu - len(work.packet)
|
||||
if padding < 0 {
|
||||
// drop
|
||||
work.packet = nil
|
||||
work.wg.Done()
|
||||
work.mutex.Unlock()
|
||||
}
|
||||
for n := 0; n < padding; n += 1 {
|
||||
work.packet = append(work.packet, 0)
|
||||
@@ -183,6 +224,30 @@ func (device *Device) RoutineEncryptionWorker() {
|
||||
work.packet,
|
||||
nil,
|
||||
)
|
||||
work.wg.Done()
|
||||
work.mutex.Unlock()
|
||||
}
|
||||
}
|
||||
|
||||
/* Sequentially reads packets from queue and sends to endpoint
|
||||
*
|
||||
* Obs. Single instance per peer.
|
||||
* The routine terminates then the outbound queue is closed.
|
||||
*/
|
||||
func (peer *Peer) RoutineSequential() {
|
||||
for work := range peer.queue.outbound {
|
||||
work.mutex.Lock()
|
||||
func() {
|
||||
peer.mutex.RLock()
|
||||
defer peer.mutex.RUnlock()
|
||||
if work.packet == nil {
|
||||
return
|
||||
}
|
||||
if peer.endpoint == nil {
|
||||
return
|
||||
}
|
||||
peer.device.conn.WriteToUDP(work.packet, peer.endpoint)
|
||||
peer.timer.sendKeepalive.Reset(peer.persistentKeepaliveInterval)
|
||||
}()
|
||||
work.mutex.Unlock()
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user