From 27d9a67dda63acd087cf8814a22eac60c89dc242 Mon Sep 17 00:00:00 2001 From: Wade Simmons Date: Thu, 25 Feb 2021 15:01:14 -0500 Subject: [PATCH] Proper multiqueue support for tun devices (#382) This change is for Linux only. Previously, when running with multiple tun.routines, we would only have one file descriptor. This change instead sets IFF_MULTI_QUEUE and opens a file descriptor for each routine. This allows us to process with multiple threads while preventing out of order packet reception issues. To attempt to distribute the flows across the queues, we try to write to the tun/UDP queue that corresponds with the one we read from. So if we read a packet from tun queue "2", we will write the outgoing encrypted packet to UDP queue "2". Because of the nature of how multi queue works with flows, a given host tunnel will be sticky to a given routine (so if you try to performance benchmark by only using one tunnel between two hosts, you are only going to be using a max of one thread for each direction). Because this system works much better when we can correlate flows between the tun and udp routines, we are deprecating the undocumented "tun.routines" and "listen.routines" parameters and introducing a new "routines" parameter that sets the value for both. If you use the old undocumented parameters, the max of the values will be used and a warning logged. Co-authored-by: Nate Brown --- examples/config.yml | 9 ++++++ inside.go | 12 ++++---- interface.go | 67 +++++++++++++++++++++++----------------- main.go | 74 +++++++++++++++++++++++++++++++-------------- outside.go | 8 ++--- tun_android.go | 6 +++- tun_darwin.go | 7 ++++- tun_disabled.go | 4 +++ tun_freebsd.go | 6 +++- tun_ios.go | 6 +++- tun_linux.go | 34 ++++++++++++++++++--- tun_windows.go | 7 ++++- udp_generic.go | 4 +-- udp_linux.go | 4 +-- 14 files changed, 175 insertions(+), 73 deletions(-) diff --git a/examples/config.yml b/examples/config.yml index 6c48c01..fbae07f 100644 --- a/examples/config.yml +++ b/examples/config.yml @@ -86,6 +86,15 @@ listen: #read_buffer: 10485760 #write_buffer: 10485760 +# EXPERIMENTAL: This option is currently only supported on linux and may +# change in future minor releases. +# +# Routines is the number of thread pairs to run that consume from the tun and UDP queues. +# Currently, this defaults to 1 which means we have 1 tun queue reader and 1 +# UDP queue reader. Setting this above one will set IFF_MULTI_QUEUE on the tun +# device and SO_REUSEPORT on the UDP socket to allow multiple queues. +#routines: 1 + punchy: # Continues to punch inbound/outbound at a regular interval to avoid expiration of firewall nat mappings punch: true diff --git a/inside.go b/inside.go index 3e36b49..6192a1c 100644 --- a/inside.go +++ b/inside.go @@ -7,7 +7,7 @@ import ( "github.com/sirupsen/logrus" ) -func (f *Interface) consumeInsidePacket(packet []byte, fwPacket *FirewallPacket, nb, out []byte) { +func (f *Interface) consumeInsidePacket(packet []byte, fwPacket *FirewallPacket, nb, out []byte, q int) { err := newPacket(packet, false, fwPacket) if err != nil { l.WithField("packet", packet).Debugf("Error while validating outbound packet: %s", err) @@ -54,7 +54,7 @@ func (f *Interface) consumeInsidePacket(packet []byte, fwPacket *FirewallPacket, dropReason := f.firewall.Drop(packet, *fwPacket, false, hostinfo, trustedCAs) if dropReason == nil { - mc := f.sendNoMetrics(message, 0, ci, hostinfo, hostinfo.remote, packet, nb, out) + mc := f.sendNoMetrics(message, 0, ci, hostinfo, hostinfo.remote, packet, nb, out, q) if f.lightHouse != nil && mc%5000 == 0 { f.lightHouse.Query(fwPacket.RemoteIP, f) } @@ -139,7 +139,7 @@ func (f *Interface) sendMessageNow(t NebulaMessageType, st NebulaMessageSubType, return } - f.sendNoMetrics(message, st, hostInfo.ConnectionState, hostInfo, hostInfo.remote, p, nb, out) + f.sendNoMetrics(message, st, hostInfo.ConnectionState, hostInfo, hostInfo.remote, p, nb, out, 0) if f.lightHouse != nil && *hostInfo.ConnectionState.messageCounter%5000 == 0 { f.lightHouse.Query(fp.RemoteIP, f) } @@ -211,10 +211,10 @@ func (f *Interface) sendMessageToAll(t NebulaMessageType, st NebulaMessageSubTyp func (f *Interface) send(t NebulaMessageType, st NebulaMessageSubType, ci *ConnectionState, hostinfo *HostInfo, remote *udpAddr, p, nb, out []byte) { f.messageMetrics.Tx(t, st, 1) - f.sendNoMetrics(t, st, ci, hostinfo, remote, p, nb, out) + f.sendNoMetrics(t, st, ci, hostinfo, remote, p, nb, out, 0) } -func (f *Interface) sendNoMetrics(t NebulaMessageType, st NebulaMessageSubType, ci *ConnectionState, hostinfo *HostInfo, remote *udpAddr, p, nb, out []byte) uint64 { +func (f *Interface) sendNoMetrics(t NebulaMessageType, st NebulaMessageSubType, ci *ConnectionState, hostinfo *HostInfo, remote *udpAddr, p, nb, out []byte, q int) uint64 { if ci.eKey == nil { //TODO: log warning return 0 @@ -240,7 +240,7 @@ func (f *Interface) sendNoMetrics(t NebulaMessageType, st NebulaMessageSubType, return c } - err = f.outside.WriteTo(out, remote) + err = f.writers[q].WriteTo(out, remote) if err != nil { hostinfo.logger().WithError(err). WithField("udpAddr", remote).Error("Failed to write outgoing packet") diff --git a/interface.go b/interface.go index ee90657..fd3483b 100644 --- a/interface.go +++ b/interface.go @@ -5,6 +5,7 @@ import ( "io" "net" "os" + "runtime" "time" "github.com/rcrowley/go-metrics" @@ -18,6 +19,7 @@ type Inside interface { CidrNet() *net.IPNet DeviceName() string WriteRaw([]byte) error + NewMultiQueueReader() (io.ReadWriteCloser, error) } type InterfaceConfig struct { @@ -35,8 +37,7 @@ type InterfaceConfig struct { DropLocalBroadcast bool DropMulticast bool UDPBatchSize int - udpQueues int - tunQueues int + routines int MessageMetrics *MessageMetrics version string } @@ -57,10 +58,12 @@ type Interface struct { dropLocalBroadcast bool dropMulticast bool udpBatchSize int - udpQueues int - tunQueues int + routines int version string + writers []*udpConn + readers []io.ReadWriteCloser + metricHandshakes metrics.Histogram messageMetrics *MessageMetrics } @@ -94,9 +97,10 @@ func NewInterface(c *InterfaceConfig) (*Interface, error) { dropLocalBroadcast: c.DropLocalBroadcast, dropMulticast: c.DropMulticast, udpBatchSize: c.UDPBatchSize, - udpQueues: c.udpQueues, - tunQueues: c.tunQueues, + routines: c.routines, version: c.version, + writers: make([]*udpConn, c.routines), + readers: make([]io.ReadWriteCloser, c.routines), metricHandshakes: metrics.GetOrRegisterHistogram("handshakes", nil, metrics.NewExpDecaySample(1028, 0.015)), messageMetrics: c.MessageMetrics, @@ -109,9 +113,6 @@ func NewInterface(c *InterfaceConfig) (*Interface, error) { func (f *Interface) run() { // actually turn on tun dev - if err := f.inside.Activate(); err != nil { - l.Fatal(err) - } addr, err := f.outside.LocalAddr() if err != nil { @@ -122,53 +123,61 @@ func (f *Interface) run() { WithField("build", f.version).WithField("udpAddr", addr). Info("Nebula interface is active") + metrics.GetOrRegisterGauge("routines", nil).Update(int64(f.routines)) + // Launch n queues to read packets from udp - for i := 0; i < f.udpQueues; i++ { + for i := 0; i < f.routines; i++ { go f.listenOut(i) } // Launch n queues to read packets from tun dev - for i := 0; i < f.tunQueues; i++ { - go f.listenIn(i) + var reader io.ReadWriteCloser = f.inside + for i := 0; i < f.routines; i++ { + if i > 0 { + reader, err = f.inside.NewMultiQueueReader() + if err != nil { + l.Fatal(err) + } + } + f.readers[i] = reader + go f.listenIn(reader, i) + } + + if err := f.inside.Activate(); err != nil { + l.Fatal(err) } } func (f *Interface) listenOut(i int) { - //TODO: handle error - addr, err := f.outside.LocalAddr() - if err != nil { - l.WithError(err).Error("failed to discover udp listening address") - } + runtime.LockOSThread() var li *udpConn + // TODO clean this up with a coherent interface for each outside connection if i > 0 { - //TODO: handle error - li, err = NewListener(udp2ip(addr).String(), int(addr.Port), i > 0) - if err != nil { - l.WithError(err).Error("failed to make a new udp listener") - } + li = f.writers[i] } else { li = f.outside } - - li.ListenOut(f) + li.ListenOut(f, i) } -func (f *Interface) listenIn(i int) { +func (f *Interface) listenIn(reader io.ReadWriteCloser, i int) { + runtime.LockOSThread() + packet := make([]byte, mtu) out := make([]byte, mtu) fwPacket := &FirewallPacket{} nb := make([]byte, 12, 12) for { - n, err := f.inside.Read(packet) + n, err := reader.Read(packet) if err != nil { l.WithError(err).Error("Error while reading outbound packet") // This only seems to happen when something fatal happens to the fd, so exit. os.Exit(2) } - f.consumeInsidePacket(packet[:n], fwPacket, nb, out) + f.consumeInsidePacket(packet[:n], fwPacket, nb, out, i) } } @@ -176,7 +185,9 @@ func (f *Interface) RegisterConfigChangeCallbacks(c *Config) { c.RegisterReloadCallback(f.reloadCA) c.RegisterReloadCallback(f.reloadCertKey) c.RegisterReloadCallback(f.reloadFirewall) - c.RegisterReloadCallback(f.outside.reloadConfig) + for _, udpConn := range f.writers { + c.RegisterReloadCallback(udpConn.reloadConfig) + } } func (f *Interface) reloadCA(c *Config) { diff --git a/main.go b/main.go index 73ec8e7..74db1b5 100644 --- a/main.go +++ b/main.go @@ -93,6 +93,30 @@ func Main(config *Config, configTest bool, buildVersion string, logger *logrus.L // tun config, listeners, anything modifying the computer should be below //////////////////////////////////////////////////////////////////////////////////////////////////////////////////// + var routines int + + // If `routines` is set, use that and ignore the specific values + if routines = config.GetInt("routines", 0); routines != 0 { + if routines < 1 { + routines = 1 + } + if routines > 1 { + l.WithField("routines", routines).Info("Using multiple routines") + } + } else { + // deprecated and undocumented + tunQueues := config.GetInt("tun.routines", 1) + udpQueues := config.GetInt("listen.routines", 1) + if tunQueues > udpQueues { + routines = tunQueues + } else { + routines = udpQueues + } + if routines != 1 { + l.WithField("routines", routines).Warn("Setting tun.routines and listen.routines is deprecated. Use `routines` instead") + } + } + var tun Inside if !configTest { config.CatchHUP() @@ -117,6 +141,7 @@ func Main(config *Config, configTest bool, buildVersion string, logger *logrus.L routes, unsafeRoutes, config.GetInt("tun.tx_queue", 500), + routines > 1, ) } @@ -126,15 +151,27 @@ func Main(config *Config, configTest bool, buildVersion string, logger *logrus.L } // set up our UDP listener - udpQueues := config.GetInt("listen.routines", 1) - var udpServer *udpConn + udpConns := make([]*udpConn, routines) + port := config.GetInt("listen.port", 0) if !configTest { - udpServer, err = NewListener(config.GetString("listen.host", "0.0.0.0"), config.GetInt("listen.port", 0), udpQueues > 1) - if err != nil { - return nil, NewContextualError("Failed to open udp listener", nil, err) + for i := 0; i < routines; i++ { + udpServer, err := NewListener(config.GetString("listen.host", "0.0.0.0"), port, routines > 1) + if err != nil { + return nil, NewContextualError("Failed to open udp listener", m{"queue": i}, err) + } + udpServer.reloadConfig(config) + udpConns[i] = udpServer + + // If port is dynamic, discover it + if port == 0 { + uPort, err := udpServer.LocalAddr() + if err != nil { + return nil, NewContextualError("Failed to get listening port", nil, err) + } + port = int(uPort.Port) + } } - udpServer.reloadConfig(config) } // Set up my internal host map @@ -190,17 +227,7 @@ func Main(config *Config, configTest bool, buildVersion string, logger *logrus.L punchy := NewPunchyFromConfig(config) if punchy.Punch && !configTest { l.Info("UDP hole punching enabled") - go hostMap.Punchy(udpServer) - } - - port := config.GetInt("listen.port", 0) - // If port is dynamic, discover it - if port == 0 && !configTest { - uPort, err := udpServer.LocalAddr() - if err != nil { - return nil, NewContextualError("Failed to get listening port", nil, err) - } - port = int(uPort.Port) + go hostMap.Punchy(udpConns[0]) } amLighthouse := config.GetBool("lighthouse.am_lighthouse", false) @@ -230,7 +257,7 @@ func Main(config *Config, configTest bool, buildVersion string, logger *logrus.L //TODO: change to a duration config.GetInt("lighthouse.interval", 10), port, - udpServer, + udpConns[0], punchy.Respond, punchy.Delay, config.GetBool("stats.lighthouse_metrics", false), @@ -304,7 +331,7 @@ func Main(config *Config, configTest bool, buildVersion string, logger *logrus.L messageMetrics: messageMetrics, } - handshakeManager := NewHandshakeManager(tunCidr, preferredRanges, hostMap, lightHouse, udpServer, handshakeConfig) + handshakeManager := NewHandshakeManager(tunCidr, preferredRanges, hostMap, lightHouse, udpConns[0], handshakeConfig) lightHouse.handshakeTrigger = handshakeManager.trigger //TODO: These will be reused for psk @@ -317,7 +344,7 @@ func Main(config *Config, configTest bool, buildVersion string, logger *logrus.L ifConfig := &InterfaceConfig{ HostMap: hostMap, Inside: tun, - Outside: udpServer, + Outside: udpConns[0], certState: cs, Cipher: config.GetString("cipher", "aes"), Firewall: fw, @@ -329,8 +356,7 @@ func Main(config *Config, configTest bool, buildVersion string, logger *logrus.L DropLocalBroadcast: config.GetBool("tun.drop_local_broadcast", false), DropMulticast: config.GetBool("tun.drop_multicast", false), UDPBatchSize: config.GetInt("listen.batch", 64), - udpQueues: udpQueues, - tunQueues: config.GetInt("tun.routines", 1), + routines: routines, MessageMetrics: messageMetrics, version: buildVersion, } @@ -351,6 +377,10 @@ func Main(config *Config, configTest bool, buildVersion string, logger *logrus.L return nil, fmt.Errorf("failed to initialize interface: %s", err) } + // TODO: Better way to attach these, probably want a new interface in InterfaceConfig + // I don't want to make this initial commit too far-reaching though + ifce.writers = udpConns + ifce.RegisterConfigChangeCallbacks(config) go handshakeManager.Run(ifce) diff --git a/outside.go b/outside.go index 7738e80..20aa4f1 100644 --- a/outside.go +++ b/outside.go @@ -17,7 +17,7 @@ const ( minFwPacketLen = 4 ) -func (f *Interface) readOutsidePackets(addr *udpAddr, out []byte, packet []byte, header *Header, fwPacket *FirewallPacket, lhh *LightHouseHandler, nb []byte) { +func (f *Interface) readOutsidePackets(addr *udpAddr, out []byte, packet []byte, header *Header, fwPacket *FirewallPacket, lhh *LightHouseHandler, nb []byte, q int) { err := header.Parse(packet) if err != nil { // TODO: best if we return this and let caller log @@ -45,7 +45,7 @@ func (f *Interface) readOutsidePackets(addr *udpAddr, out []byte, packet []byte, return } - f.decryptToTun(hostinfo, header.MessageCounter, out, packet, fwPacket, nb) + f.decryptToTun(hostinfo, header.MessageCounter, out, packet, fwPacket, nb, q) // Fallthrough to the bottom to record incoming traffic @@ -257,7 +257,7 @@ func (f *Interface) decrypt(hostinfo *HostInfo, mc uint64, out []byte, packet [] return out, nil } -func (f *Interface) decryptToTun(hostinfo *HostInfo, messageCounter uint64, out []byte, packet []byte, fwPacket *FirewallPacket, nb []byte) { +func (f *Interface) decryptToTun(hostinfo *HostInfo, messageCounter uint64, out []byte, packet []byte, fwPacket *FirewallPacket, nb []byte, q int) { var err error out, err = hostinfo.ConnectionState.dKey.DecryptDanger(out, packet[:HeaderLen], packet[HeaderLen:], messageCounter, nb) @@ -292,7 +292,7 @@ func (f *Interface) decryptToTun(hostinfo *HostInfo, messageCounter uint64, out } f.connectionManager.In(hostinfo.hostId) - err = f.inside.WriteRaw(out) + _, err = f.readers[q].Write(out) if err != nil { l.WithError(err).Error("Failed to write to tun") } diff --git a/tun_android.go b/tun_android.go index 957af0b..8f6463f 100644 --- a/tun_android.go +++ b/tun_android.go @@ -37,7 +37,7 @@ func newTunFromFd(deviceFd int, cidr *net.IPNet, defaultMTU int, routes []route, return } -func newTun(deviceName string, cidr *net.IPNet, defaultMTU int, routes []route, unsafeRoutes []route, txQueueLen int) (ifce *Tun, err error) { +func newTun(deviceName string, cidr *net.IPNet, defaultMTU int, routes []route, unsafeRoutes []route, txQueueLen int, multiqueue bool) (ifce *Tun, err error) { return nil, fmt.Errorf("newTun not supported in Android") } @@ -74,3 +74,7 @@ func (c *Tun) CidrNet() *net.IPNet { func (c *Tun) DeviceName() string { return c.Device } + +func (t *Tun) NewMultiQueueReader() (io.ReadWriteCloser, error) { + return nil, fmt.Errorf("TODO: multiqueue not implemented for android") +} diff --git a/tun_darwin.go b/tun_darwin.go index dc63b45..0dfbe3c 100644 --- a/tun_darwin.go +++ b/tun_darwin.go @@ -4,6 +4,7 @@ package nebula import ( "fmt" + "io" "net" "os/exec" "strconv" @@ -20,7 +21,7 @@ type Tun struct { *water.Interface } -func newTun(deviceName string, cidr *net.IPNet, defaultMTU int, routes []route, unsafeRoutes []route, txQueueLen int) (ifce *Tun, err error) { +func newTun(deviceName string, cidr *net.IPNet, defaultMTU int, routes []route, unsafeRoutes []route, txQueueLen int, multiqueue bool) (ifce *Tun, err error) { if len(routes) > 0 { return nil, fmt.Errorf("route MTU not supported in Darwin") } @@ -80,3 +81,7 @@ func (c *Tun) WriteRaw(b []byte) error { _, err := c.Write(b) return err } + +func (t *Tun) NewMultiQueueReader() (io.ReadWriteCloser, error) { + return nil, fmt.Errorf("TODO: multiqueue not implemented for darwin") +} diff --git a/tun_disabled.go b/tun_disabled.go index 2d96b6a..4ca7afc 100644 --- a/tun_disabled.go +++ b/tun_disabled.go @@ -50,6 +50,10 @@ func (t *disabledTun) WriteRaw(b []byte) error { return err } +func (t *disabledTun) NewMultiQueueReader() (io.ReadWriteCloser, error) { + return t, nil +} + func (t *disabledTun) Close() error { if t.block != nil { close(t.block) diff --git a/tun_freebsd.go b/tun_freebsd.go index b5a67e7..7e6f98c 100644 --- a/tun_freebsd.go +++ b/tun_freebsd.go @@ -26,7 +26,7 @@ func newTunFromFd(deviceFd int, cidr *net.IPNet, defaultMTU int, routes []route, return nil, fmt.Errorf("newTunFromFd not supported in FreeBSD") } -func newTun(deviceName string, cidr *net.IPNet, defaultMTU int, routes []route, unsafeRoutes []route, txQueueLen int) (ifce *Tun, err error) { +func newTun(deviceName string, cidr *net.IPNet, defaultMTU int, routes []route, unsafeRoutes []route, txQueueLen int, multiqueue bool) (ifce *Tun, err error) { if len(routes) > 0 { return nil, fmt.Errorf("Route MTU not supported in FreeBSD") } @@ -87,3 +87,7 @@ func (c *Tun) WriteRaw(b []byte) error { _, err := c.Write(b) return err } + +func (t *Tun) NewMultiQueueReader() (io.ReadWriteCloser, error) { + return nil, fmt.Errorf("TODO: multiqueue not implemented for freebsd") +} diff --git a/tun_ios.go b/tun_ios.go index 5440f21..2e0e784 100644 --- a/tun_ios.go +++ b/tun_ios.go @@ -18,7 +18,7 @@ type Tun struct { Cidr *net.IPNet } -func newTun(deviceName string, cidr *net.IPNet, defaultMTU int, routes []route, unsafeRoutes []route, txQueueLen int) (ifce *Tun, err error) { +func newTun(deviceName string, cidr *net.IPNet, defaultMTU int, routes []route, unsafeRoutes []route, txQueueLen int, multiqueue bool) (ifce *Tun, err error) { return nil, fmt.Errorf("newTun not supported in iOS") } @@ -111,3 +111,7 @@ func (c *Tun) CidrNet() *net.IPNet { func (c *Tun) DeviceName() string { return c.Device } + +func (t *Tun) NewMultiQueueReader() (io.ReadWriteCloser, error) { + return nil, fmt.Errorf("TODO: multiqueue not implemented for ios") +} diff --git a/tun_linux.go b/tun_linux.go index cde79d6..4d0707b 100644 --- a/tun_linux.go +++ b/tun_linux.go @@ -55,8 +55,9 @@ func ipv4(addr string) (o [4]byte, err error) { */ const ( - cIFF_TUN = 0x0001 - cIFF_NO_PI = 0x1000 + cIFF_TUN = 0x0001 + cIFF_NO_PI = 0x1000 + cIFF_MULTI_QUEUE = 0x0100 ) type ifreqAddr struct { @@ -94,7 +95,7 @@ func newTunFromFd(deviceFd int, cidr *net.IPNet, defaultMTU int, routes []route, return } -func newTun(deviceName string, cidr *net.IPNet, defaultMTU int, routes []route, unsafeRoutes []route, txQueueLen int) (ifce *Tun, err error) { +func newTun(deviceName string, cidr *net.IPNet, defaultMTU int, routes []route, unsafeRoutes []route, txQueueLen int, multiqueue bool) (ifce *Tun, err error) { fd, err := unix.Open("/dev/net/tun", os.O_RDWR, 0) if err != nil { return nil, err @@ -102,9 +103,12 @@ func newTun(deviceName string, cidr *net.IPNet, defaultMTU int, routes []route, var req ifReq req.Flags = uint16(cIFF_TUN | cIFF_NO_PI) + if multiqueue { + req.Flags |= cIFF_MULTI_QUEUE + } copy(req.Name[:], deviceName) if err = ioctl(uintptr(fd), uintptr(unix.TUNSETIFF), uintptr(unsafe.Pointer(&req))); err != nil { - return + return nil, err } name := strings.Trim(string(req.Name[:]), "\x00") @@ -131,6 +135,24 @@ func newTun(deviceName string, cidr *net.IPNet, defaultMTU int, routes []route, return } +func (c *Tun) NewMultiQueueReader() (io.ReadWriteCloser, error) { + fd, err := unix.Open("/dev/net/tun", os.O_RDWR, 0) + if err != nil { + return nil, err + } + + var req ifReq + req.Flags = uint16(cIFF_TUN | cIFF_NO_PI | cIFF_MULTI_QUEUE) + copy(req.Name[:], c.Device) + if err = ioctl(uintptr(fd), uintptr(unix.TUNSETIFF), uintptr(unsafe.Pointer(&req))); err != nil { + return nil, err + } + + file := os.NewFile(uintptr(fd), "/dev/net/tun") + + return file, nil +} + func (c *Tun) WriteRaw(b []byte) error { var nn int for { @@ -153,6 +175,10 @@ func (c *Tun) WriteRaw(b []byte) error { } } +func (c *Tun) Write(b []byte) (int, error) { + return len(b), c.WriteRaw(b) +} + func (c Tun) deviceBytes() (o [16]byte) { for i, c := range c.Device { o[i] = byte(c) diff --git a/tun_windows.go b/tun_windows.go index d9e12f4..040f1ce 100644 --- a/tun_windows.go +++ b/tun_windows.go @@ -2,6 +2,7 @@ package nebula import ( "fmt" + "io" "net" "os/exec" "strconv" @@ -22,7 +23,7 @@ func newTunFromFd(deviceFd int, cidr *net.IPNet, defaultMTU int, routes []route, return nil, fmt.Errorf("newTunFromFd not supported in Windows") } -func newTun(deviceName string, cidr *net.IPNet, defaultMTU int, routes []route, unsafeRoutes []route, txQueueLen int) (ifce *Tun, err error) { +func newTun(deviceName string, cidr *net.IPNet, defaultMTU int, routes []route, unsafeRoutes []route, txQueueLen int, multiqueue bool) (ifce *Tun, err error) { if len(routes) > 0 { return nil, fmt.Errorf("route MTU not supported in Windows") } @@ -100,3 +101,7 @@ func (c *Tun) WriteRaw(b []byte) error { _, err := c.Write(b) return err } + +func (t *Tun) NewMultiQueueReader() (io.ReadWriteCloser, error) { + return nil, fmt.Errorf("TODO: multiqueue not implemented for windows") +} diff --git a/udp_generic.go b/udp_generic.go index 0bafbb6..eccf616 100644 --- a/udp_generic.go +++ b/udp_generic.go @@ -100,7 +100,7 @@ type rawMessage struct { Len uint32 } -func (u *udpConn) ListenOut(f *Interface) { +func (u *udpConn) ListenOut(f *Interface, q int) { plaintext := make([]byte, mtu) buffer := make([]byte, mtu) header := &Header{} @@ -119,7 +119,7 @@ func (u *udpConn) ListenOut(f *Interface) { } udpAddr.UDPAddr = *rua - f.readOutsidePackets(udpAddr, plaintext[:0], buffer[:n], header, fwPacket, lhh, nb) + f.readOutsidePackets(udpAddr, plaintext[:0], buffer[:n], header, fwPacket, lhh, nb, q) } } diff --git a/udp_linux.go b/udp_linux.go index 92866e5..19b5f5e 100644 --- a/udp_linux.go +++ b/udp_linux.go @@ -139,7 +139,7 @@ func (u *udpConn) LocalAddr() (*udpAddr, error) { return addr, nil } -func (u *udpConn) ListenOut(f *Interface) { +func (u *udpConn) ListenOut(f *Interface, q int) { plaintext := make([]byte, mtu) header := &Header{} fwPacket := &FirewallPacket{} @@ -168,7 +168,7 @@ func (u *udpConn) ListenOut(f *Interface) { udpAddr.IP = binary.BigEndian.Uint32(names[i][4:8]) udpAddr.Port = binary.BigEndian.Uint16(names[i][2:4]) - f.readOutsidePackets(udpAddr, plaintext[:0], buffers[i][:msgs[i].Len], header, fwPacket, lhh, nb) + f.readOutsidePackets(udpAddr, plaintext[:0], buffers[i][:msgs[i].Len], header, fwPacket, lhh, nb, q) } } }