log *Logger // collection of loggers for levels
idCounter uint // for assigning debug ids to peers
fwMark uint32
- net struct {
+ pool struct {
+ // pools objects for reuse
+ messageBuffers sync.Pool
+ }
+ net struct {
// seperate for performance reasons
mutex sync.RWMutex
addr *net.UDPAddr // UDP source address
}
}
+func (device *Device) GetMessageBuffer() *[MaxMessageSize]byte {
+ return device.pool.messageBuffers.Get().(*[MaxMessageSize]byte)
+}
+
+func (device *Device) PutMessageBuffer(msg *[MaxMessageSize]byte) {
+ device.pool.messageBuffers.Put(msg)
+}
+
func NewDevice(tun TUNDevice, logLevel int) *Device {
device := new(Device)
device.net.addr, _ = net.ResolveUDPAddr(addr.Network(), addr.String())
device.net.mutex.Unlock()
+ // setup pools
+
+ device.pool.messageBuffers = sync.Pool{
+ New: func() interface{} {
+ return new([MaxMessageSize]byte)
+ },
+ }
+
// create queues
device.queue.handshake = make(chan QueueHandshakeElement, QueueHandshakeSize)
type QueueHandshakeElement struct {
msgType uint32
packet []byte
+ buffer *[MaxMessageSize]byte
source *net.UDPAddr
}
type QueueInboundElement struct {
dropped int32
mutex sync.Mutex
+ buffer *[MaxMessageSize]byte
packet []byte
counter uint64
keyPair *KeyPair
return atomic.LoadInt32(&elem.dropped) == AtomicTrue
}
-func addToInboundQueue(
+func (device *Device) addToInboundQueue(
queue chan *QueueInboundElement,
element *QueueInboundElement,
) {
}
}
-func addToHandshakeQueue(
+func (device *Device) addToHandshakeQueue(
queue chan QueueHandshakeElement,
element QueueHandshakeElement,
) {
return
default:
select {
- case <-queue:
+ case elem := <-queue:
+ device.PutMessageBuffer(elem.buffer)
default:
}
}
}
/* Routine determining the busy state of the interface
- *
- * TODO: prehaps nicer to do this in response to events
- * TODO: more well reasoned definition of "busy"
*/
func (device *Device) RoutineBusyMonitor() {
samples := 0
func (device *Device) RoutineReceiveIncomming() {
+ logInfo := device.log.Info
logDebug := device.log.Debug
logDebug.Println("Routine, receive incomming, started")
- var buffer []byte
+ var buffer *[MaxMessageSize]byte
for {
// read next datagram
if buffer == nil {
- buffer = make([]byte, MaxMessageSize)
+ buffer = device.GetMessageBuffer()
}
device.net.mutex.RLock()
conn.SetReadDeadline(time.Now().Add(time.Second))
- size, raddr, err := conn.ReadFromUDP(buffer)
+ size, raddr, err := conn.ReadFromUDP(buffer[:])
if err != nil || size < MinMessageSize {
continue
}
// add to handshake queue
- addToHandshakeQueue(
+ device.addToHandshakeQueue(
device.queue.handshake,
QueueHandshakeElement{
msgType: msgType,
+ buffer: buffer,
packet: packet,
source: raddr,
},
// add to peer queue
peer := value.peer
- work := new(QueueInboundElement)
- work.packet = packet
- work.keyPair = keyPair
- work.dropped = AtomicFalse
+ work := &QueueInboundElement{
+ packet: packet,
+ buffer: buffer,
+ keyPair: keyPair,
+ dropped: AtomicFalse,
+ }
work.mutex.Lock()
// add to decryption queues
- addToInboundQueue(device.queue.decryption, work)
- addToInboundQueue(peer.queue.inbound, work)
+ device.addToInboundQueue(device.queue.decryption, work)
+ device.addToInboundQueue(peer.queue.inbound, work)
buffer = nil
default:
- // unknown message type
- logDebug.Println("Got unknown message from:", raddr)
+ logInfo.Println("Got unknown message from:", raddr)
}
}()
}
var err error
copy(nonce[4:], counter)
elem.counter = binary.LittleEndian.Uint64(counter)
- elem.packet, err = elem.keyPair.receive.Open(elem.packet[:0], nonce[:], content, nil)
+ elem.packet, err = elem.keyPair.receive.Open(
+ elem.buffer[:0],
+ nonce[:],
+ content,
+ nil,
+ )
if err != nil {
elem.Drop()
}
logDebug.Println("Creating response message for", peer.String())
outElem := device.NewOutboundElement()
- writer := bytes.NewBuffer(outElem.data[:0])
+ writer := bytes.NewBuffer(outElem.buffer[:0])
binary.Write(writer, binary.LittleEndian, response)
outElem.packet = writer.Bytes()
peer.mac.AddMacs(outElem.packet)
addToOutboundQueue(peer.queue.outbound, outElem)
+ // create new keypair
+
+ peer.NewKeyPair()
+
case MessageResponseType:
// unmarshal
peer.EventHandshakeComplete()
default:
- device.log.Error.Println("Invalid message type in handshake queue")
+ logError.Println("Invalid message type in handshake queue")
}
}()
}
}
atomic.AddUint64(&peer.rxBytes, uint64(len(elem.packet)))
- addToInboundQueue(device.queue.inbound, elem)
+ device.addToInboundQueue(device.queue.inbound, elem)
}()
}
}
return
case elem := <-device.queue.inbound:
_, err := tun.Write(elem.packet)
+ device.PutMessageBuffer(elem.buffer)
if err != nil {
logError.Println("Failed to write packet to TUN device:", err)
}
type QueueOutboundElement struct {
dropped int32
mutex sync.Mutex
- data [MaxMessageSize]byte // slice holding the packet data
- packet []byte // slice of "data" (always!)
- nonce uint64 // nonce for encryption
- keyPair *KeyPair // key-pair for encryption
- peer *Peer // related peer
+ buffer *[MaxMessageSize]byte // slice holding the packet data
+ packet []byte // slice of "data" (always!)
+ nonce uint64 // nonce for encryption
+ keyPair *KeyPair // key-pair for encryption
+ peer *Peer // related peer
}
func (peer *Peer) FlushNonceQueue() {
}
}
-/*
- * Assumption: The mutex of the returned element is released
- */
func (device *Device) NewOutboundElement() *QueueOutboundElement {
- // TODO: profile, consider sync.Pool
- elem := new(QueueOutboundElement)
- return elem
+ return &QueueOutboundElement{
+ dropped: AtomicFalse,
+ buffer: device.pool.messageBuffers.Get().(*[MaxMessageSize]byte),
+ }
}
func (elem *QueueOutboundElement) Drop() {
elem = device.NewOutboundElement()
}
- elem.packet = elem.data[MessageTransportHeaderSize:]
+ elem.packet = elem.buffer[MessageTransportHeaderSize:]
size, err := tun.Read(elem.packet)
if err != nil {
// populate header fields
func() {
- header := work.data[:MessageTransportHeaderSize]
+ header := work.buffer[:MessageTransportHeaderSize]
fieldType := header[0:4]
fieldReceiver := header[4:8]
nil,
)
length := MessageTransportHeaderSize + len(work.packet)
- work.packet = work.data[:length]
+ work.packet = work.buffer[:length]
work.mutex.Unlock()
// refresh key if necessary
case work := <-peer.queue.outbound:
work.mutex.Lock()
- if work.IsDropped() {
- continue
- }
func() {
+ // return buffer to pool after processing
+
+ defer device.PutMessageBuffer(work.buffer)
+ if work.IsDropped() {
+ return
+ }
+
// send to endpoint
peer.mutex.RLock()
return
}
+ // send message and return buffer to pool
+
_, err := device.net.conn.WriteToUDP(work.packet, peer.endpoint)
if err != nil {
return
}
+
atomic.AddUint64(&peer.txBytes, uint64(len(work.packet)))
// reset keep-alive
projects / thirdparty / wireguard-go.git / commitdiff
