personal_interest_mirrors/nebula
1
0
Fork 0
mirror of https://github.com/slackhq/nebula.git synced 2025-11-22 08:24:25 +01:00
Code Issues Actions 2 Packages Projects Releases Wiki Activity

multiqueue but it doesn't help

Browse Source
This commit is contained in:
JackDoan
2025-11-11 22:14:19 -06:00
parent 9bdc513da0
commit 1a83817cc2
6 changed files with 43 additions and 51 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
interface.go
View File

@@ -301,7 +301,7 @@ func (f *Interface) listenOut(q int) {
for i := 0; i < len(toSend); i += batch {
x := min(len(toSend[i:]), batch)
toSendThisTime := toSend[i : i+x]
_, err := f.readers[q].WriteMany(toSendThisTime)
_, err := f.readers[q].WriteMany(toSendThisTime, q)
if err != nil {
f.l.WithError(err).Error("Failed to write messages")
}
@@ -332,7 +332,7 @@ func (f *Interface) listenIn(reader overlay.TunDev, queueNum int) {
for {
n, err := reader.ReadMany(packets)
n, err := reader.ReadMany(packets, queueNum)
//todo!!
if err != nil {
if errors.Is(err, os.ErrClosed) && f.closed.Load() {

4
overlay/tun.go
View File

@@ -16,8 +16,8 @@ const DefaultMTU = 1300
type TunDev interface {
io.WriteCloser
ReadMany([]*packet.VirtIOPacket) (int, error)
WriteMany([][]byte) (int, error)
ReadMany(x []*packet.VirtIOPacket, q int) (int, error)
WriteMany(x [][]byte, q int) (int, error)
}
// TODO: We may be able to remove routines

4
overlay/tun_disabled.go
View File

@@ -111,7 +111,7 @@ func (t *disabledTun) Write(b []byte) (int, error) {
return len(b), nil
}
func (t *disabledTun) WriteMany(b [][]byte) (int, error) {
func (t *disabledTun) WriteMany(b [][]byte, _ int) (int, error) {
out := 0
for i := range b {
x, err := t.Write(b[i])
@@ -123,7 +123,7 @@ func (t *disabledTun) WriteMany(b [][]byte) (int, error) {
return out, nil
}
func (t *disabledTun) ReadMany(b []*packet.VirtIOPacket) (int, error) {
func (t *disabledTun) ReadMany(b []*packet.VirtIOPacket, _ int) (int, error) {
return t.Read(b[0].Payload)
}

71
overlay/tun_linux.go
View File

@@ -10,7 +10,6 @@ import (
"os"
"strings"
"sync/atomic"
"syscall"
"time"
"unsafe"
@@ -29,7 +28,7 @@ import (
type tun struct {
file *os.File
fd int
vdev *vhostnet.Device
vdev []*vhostnet.Device
Device string
vpnNetworks []netip.Prefix
MaxMTU int
@@ -108,7 +107,7 @@ func newTun(c *config.C, l *logrus.Logger, vpnNetworks []netip.Prefix, multiqueu
var req ifReq
req.Flags = uint16(unix.IFF_TUN | unix.IFF_NO_PI | unix.IFF_TUN_EXCL | unix.IFF_VNET_HDR)
if multiqueue {
//req.Flags |= unix.IFF_MULTI_QUEUE
req.Flags |= unix.IFF_MULTI_QUEUE
}
copy(req.Name[:], c.GetString("tun.dev", ""))
if err = ioctl(uintptr(fd), uintptr(unix.TUNSETIFF), uintptr(unsafe.Pointer(&req))); err != nil {
@@ -135,17 +134,6 @@ func newTun(c *config.C, l *logrus.Logger, vpnNetworks []netip.Prefix, multiqueu
return nil, fmt.Errorf("set offloads: %w", err)
}
//name := strings.Trim(c.GetString("tun.dev", ""), "\x00")
//tundev, err := tuntap.NewDevice(
// tuntap.WithName(name),
// tuntap.WithDeviceType(tuntap.DeviceTypeTUN), //todo wtf
// tuntap.WithVirtioNetHdr(true), //todo hmm
// tuntap.WithOffloads(unix.TUN_F_CSUM|unix.TUN_F_USO4|unix.TUN_F_USO6), //todo
//)
//if err != nil {
// return nil, err
//}
t, err := newTunGeneric(c, l, file, vpnNetworks)
if err != nil {
return nil, err
@@ -160,7 +148,7 @@ func newTun(c *config.C, l *logrus.Logger, vpnNetworks []netip.Prefix, multiqueu
if err != nil {
return nil, err
}
t.vdev = vdev
t.vdev = []*vhostnet.Device{vdev}
return t, nil
}
@@ -259,22 +247,29 @@ func (t *tun) reload(c *config.C, initial bool) error {
}
func (t *tun) NewMultiQueueReader() (TunDev, error) {
//fd, err := unix.Open("/dev/net/tun", os.O_RDWR, 0)
//if err != nil {
// return nil, err
//}
//
//var req ifReq
//req.Flags = uint16(unix.IFF_TUN | unix.IFF_NO_PI | unix.IFF_MULTI_QUEUE)
//copy(req.Name[:], t.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")
fd, err := unix.Open("/dev/net/tun", os.O_RDWR, 0)
if err != nil {
return nil, err
}
//return file, nil
return nil, syscall.ENOTSUP
var req ifReq
req.Flags = uint16(unix.IFF_TUN | unix.IFF_NO_PI | unix.IFF_MULTI_QUEUE)
copy(req.Name[:], t.Device)
if err = ioctl(uintptr(fd), uintptr(unix.TUNSETIFF), uintptr(unsafe.Pointer(&req))); err != nil {
return nil, err
}
vdev, err := vhostnet.NewDevice(
vhostnet.WithBackendFD(fd),
vhostnet.WithQueueSize(8192), //todo config
)
if err != nil {
return nil, err
}
t.vdev = append(t.vdev, vdev)
return t, nil
}
func (t *tun) RoutesFor(ip netip.Addr) routing.Gateways {
@@ -694,8 +689,10 @@ func (t *tun) Close() error {
close(t.routeChan)
}
if t.vdev != nil {
_ = t.vdev.Close()
for _, v := range t.vdev {
if v != nil {
_ = v.Close()
}
}
if t.file != nil {
@@ -709,8 +706,8 @@ func (t *tun) Close() error {
return nil
}
func (t *tun) ReadMany(p []*packet.VirtIOPacket) (int, error) {
n, err := t.vdev.ReceivePackets(p) //we are TXing
func (t *tun) ReadMany(p []*packet.VirtIOPacket, q int) (int, error) {
n, err := t.vdev[q].ReceivePackets(p) //we are TXing
if err != nil {
return 0, err
}
@@ -730,7 +727,7 @@ func (t *tun) Write(b []byte) (int, error) {
NumBuffers: 0,
}
err := t.vdev.TransmitPackets(hdr, [][]byte{b})
err := t.vdev[0].TransmitPackets(hdr, [][]byte{b})
if err != nil {
t.l.WithError(err).Error("Transmitting packet")
return 0, err
@@ -738,7 +735,7 @@ func (t *tun) Write(b []byte) (int, error) {
return maximum, nil
}
func (t *tun) WriteMany(b [][]byte) (int, error) {
func (t *tun) WriteMany(b [][]byte, q int) (int, error) {
maximum := len(b) //we are RXing
if maximum == 0 {
return 0, nil
@@ -753,7 +750,7 @@ func (t *tun) WriteMany(b [][]byte) (int, error) {
NumBuffers: 0,
}
err := t.vdev.TransmitPackets(hdr, b)
err := t.vdev[q].TransmitPackets(hdr, b)
if err != nil {
t.l.WithError(err).Error("Transmitting packet")
return 0, err

9
overlay/user.go
View File

@@ -6,7 +6,6 @@ import (
"github.com/sirupsen/logrus"
"github.com/slackhq/nebula/config"
"github.com/slackhq/nebula/overlay/virtqueue"
"github.com/slackhq/nebula/packet"
"github.com/slackhq/nebula/routing"
)
@@ -68,11 +67,11 @@ func (d *UserDevice) Close() error {
return nil
}
func (d *UserDevice) ReadMany(b []*packet.VirtIOPacket) (int, error) {
func (d *UserDevice) ReadMany(b []*packet.VirtIOPacket, _ int) (int, error) {
return d.Read(b[0].Payload)
}
func (d *UserDevice) WriteMany(b [][]byte) (int, error) {
func (d *UserDevice) WriteMany(b [][]byte, _ int) (int, error) {
out := 0
for i := range b {
x, err := d.Write(b[i])
@@ -83,7 +82,3 @@ func (d *UserDevice) WriteMany(b [][]byte) (int, error) {
}
return out, nil
}
func (*UserDevice) GetQueues() []*virtqueue.SplitQueue {
return nil
}

2
overlay/vhostnet/device.go
View File

@@ -77,7 +77,7 @@ func NewDevice(options ...Option) (*Device, error) {
// Advertise the supported features. This isn't much for now.
// TODO: Add feature options and implement proper feature negotiation.
getFeatures, err := vhost.GetFeatures(dev.controlFD)
getFeatures, err := vhost.GetFeatures(dev.controlFD) //0x1033D008000 but why
if err != nil {
return nil, fmt.Errorf("get features: %w", err)
}