Add readv dispatcher for darwin

Author: wwqgtxx

internal/fdbased_darwin/endpoint.go

@@ -60,57 +60,14 @@ type linkDispatcher interface {
 	release()
 }
 
-// PacketDispatchMode are the various supported methods of receiving and
-// dispatching packets from the underlying FD.
-type PacketDispatchMode int
-
-// BatchSize is the number of packets to write in each syscall. It is 47
-// because when GVisorGSO is in use then a single 65KB TCP segment can get
-// split into 46 segments of 1420 bytes and a single 216 byte segment.
-const BatchSize = 47
-
-const (
-	// Readv is the default dispatch mode and is the least performant of the
-	// dispatch options but the one that is supported by all underlying FD
-	// types.
-	Readv PacketDispatchMode = iota
-	// RecvMMsg enables use of recvmmsg() syscall instead of readv() to
-	// read inbound packets. This reduces # of syscalls needed to process
-	// packets.
-	//
-	// NOTE: recvmmsg() is only supported for sockets, so if the underlying
-	// FD is not a socket then the code will still fall back to the readv()
-	// path.
-	RecvMMsg
-	// PacketMMap enables use of PACKET_RX_RING to receive packets from the
-	// NIC. PacketMMap requires that the underlying FD be an AF_PACKET. The
-	// primary use-case for this is runsc which uses an AF_PACKET FD to
-	// receive packets from the veth device.
-	PacketMMap
-)
-
-func (p PacketDispatchMode) String() string {
-	switch p {
-	case Readv:
-		return "Readv"
-	case RecvMMsg:
-		return "RecvMMsg"
-	case PacketMMap:
-		return "PacketMMap"
-	default:
-		return fmt.Sprintf("unknown packet dispatch mode '%d'", p)
-	}
-}
-
 var (
 	_ stack.LinkEndpoint = (*endpoint)(nil)
 	_ stack.GSOEndpoint  = (*endpoint)(nil)
 )
 
 // +stateify savable
 type fdInfo struct {
-	fd       int
-	isSocket bool
+	fd int
 }
 
 // +stateify savable
@@ -137,10 +94,6 @@ type endpoint struct {
 	// +checklocks:mu
 	dispatcher stack.NetworkDispatcher
 
-	// packetDispatchMode controls the packet dispatcher used by this
-	// endpoint.
-	packetDispatchMode PacketDispatchMode
-
 	// wg keeps track of running goroutines.
 	wg sync.WaitGroup `state:"nosave"`
 
@@ -168,7 +121,7 @@ type endpoint struct {
 	mtu uint32
 
 	batchSize int
-	writeMsgX bool
+	sendMsgX  bool
 }
 
 // Options specify the details about the fd-based endpoint to be created.
@@ -201,10 +154,6 @@ type Options struct {
 	// include CapabilityDisconnectOk.
 	DisconnectOk bool
 
-	// PacketDispatchMode specifies the type of inbound dispatcher to be
-	// used for this endpoint.
-	PacketDispatchMode PacketDispatchMode
-
 	// TXChecksumOffload if true, indicates that this endpoints capability
 	// set should include CapabilityTXChecksumOffload.
 	TXChecksumOffload bool
@@ -225,8 +174,8 @@ type Options struct {
 	// from each FD.
 	ProcessorsPerChannel int
 
-	MultiPendingPackets bool
-	WriteMsgX           bool
+	RecvMsgX bool
+	SendMsgX bool
 }
 
 // New creates a new fd-based endpoint.
@@ -266,7 +215,7 @@ func New(opts *Options) (stack.LinkEndpoint, error) {
 		return nil, fmt.Errorf("opts.MaxSyscallHeaderBytes is negative")
 	}
 	var batchSize int
-	if opts.MultiPendingPackets {
+	if opts.RecvMsgX {
 		batchSize = int((512*1024)/(opts.MTU)) + 1
 	} else {
 		batchSize = 1
@@ -278,11 +227,10 @@ func New(opts *Options) (stack.LinkEndpoint, error) {
 		closed:                opts.ClosedFunc,
 		addr:                  opts.Address,
 		hdrSize:               hdrSize,
-		packetDispatchMode:    opts.PacketDispatchMode,
 		maxSyscallHeaderBytes: uintptr(opts.MaxSyscallHeaderBytes),
 		writevMaxIovs:         rawfile.MaxIovs,
 		batchSize:             batchSize,
-		writeMsgX:             opts.WriteMsgX,
+		sendMsgX:              opts.SendMsgX,
 	}
 	if e.maxSyscallHeaderBytes != 0 {
 		if max := int(e.maxSyscallHeaderBytes / rawfile.SizeofIovec); max < e.writevMaxIovs {
@@ -296,14 +244,23 @@ func New(opts *Options) (stack.LinkEndpoint, error) {
 			return nil, fmt.Errorf("unix.SetNonblock(%v) failed: %v", fd, err)
 		}
 
-		e.fds = append(e.fds, fdInfo{fd: fd, isSocket: true})
+		e.fds = append(e.fds, fdInfo{fd: fd})
 		if opts.ProcessorsPerChannel == 0 {
 			opts.ProcessorsPerChannel = common.Max(1, runtime.GOMAXPROCS(0)/len(opts.FDs))
 		}
 
-		inboundDispatcher, err := newRecvMMsgDispatcher(fd, e, opts)
-		if err != nil {
-			return nil, fmt.Errorf("createInboundDispatcher(...) = %v", err)
+		var inboundDispatcher linkDispatcher
+		var err error
+		if opts.RecvMsgX {
+			inboundDispatcher, err = newRecvMMsgDispatcher(fd, e, opts)
+			if err != nil {
+				return nil, fmt.Errorf("newRecvMMsgDispatcher(%d, %+v) = %v", fd, e, err)
+			}
+		} else {
+			inboundDispatcher, err = newReadVDispatcher(fd, e, opts)
+			if err != nil {
+				return nil, fmt.Errorf("newReadVDispatcher(%d, %+v) = %v", fd, e, err)
+			}
 		}
 		e.inboundDispatchers = append(e.inboundDispatchers, inboundDispatcher)
 	}
@@ -489,7 +446,7 @@ func (e *endpoint) writePacket(pkt *stack.PacketBuffer) tcpip.Error {
 
 func (e *endpoint) sendBatch(batchFDInfo fdInfo, pkts []*stack.PacketBuffer) (int, tcpip.Error) {
 	// Degrade to writePacket if underlying fd is not a socket.
-	if !batchFDInfo.isSocket || !e.writeMsgX {
+	if !e.sendMsgX {
 		var written int
 		var err tcpip.Error
 		for written < len(pkts) {
@@ -597,7 +554,7 @@ func (e *endpoint) sendBatch(batchFDInfo fdInfo, pkts []*stack.PacketBuffer) (in
 func (e *endpoint) WritePackets(pkts stack.PacketBufferList) (int, tcpip.Error) {
 	// Preallocate to avoid repeated reallocation as we append to batch.
 	batch := make([]*stack.PacketBuffer, 0, e.batchSize)
-	batchFDInfo := fdInfo{fd: -1, isSocket: false}
+	batchFDInfo := fdInfo{fd: -1}
 	sentPackets := 0
 	for _, pkt := range pkts.AsSlice() {
 		if len(batch) == 0 {

internal/fdbased_darwin/packet_dispatchers.go

@@ -79,6 +79,70 @@ func (b *iovecBuffer) release() {
 	}
 }
 
+// readVDispatcher uses readv() system call to read inbound packets and
+// dispatches them.
+//
+// +stateify savable
+type readVDispatcher struct {
+	stopfd.StopFD
+	// fd is the file descriptor used to send and receive packets.
+	fd int
+
+	// e is the endpoint this dispatcher is attached to.
+	e *endpoint
+
+	// buf is the iovec buffer that contains the packet contents.
+	buf *iovecBuffer
+
+	// mgr is the processor goroutine manager.
+	mgr *processorManager
+}
+
+func newReadVDispatcher(fd int, e *endpoint, opts *Options) (linkDispatcher, error) {
+	stopFD, err := stopfd.New()
+	if err != nil {
+		return nil, err
+	}
+	d := &readVDispatcher{
+		StopFD: stopFD,
+		fd:     fd,
+		e:      e,
+	}
+	d.buf = newIovecBuffer(opts.MTU)
+	d.mgr = newProcessorManager(opts, e)
+	d.mgr.start()
+	return d, nil
+}
+
+func (d *readVDispatcher) release() {
+	d.buf.release()
+	d.mgr.close()
+}
+
+// dispatch reads one packet from the file descriptor and dispatches it.
+func (d *readVDispatcher) dispatch() (bool, tcpip.Error) {
+	n, errno := rawfile.BlockingReadvUntilStopped(d.ReadFD, d.fd, d.buf.nextIovecs())
+	if n <= 0 || errno != 0 {
+		return false, TranslateErrno(errno)
+	}
+
+	payload := d.buf.pullBuffer(n)
+	pkt := stack.NewPacketBuffer(stack.PacketBufferOptions{
+		Payload: payload,
+	})
+	defer pkt.DecRef()
+
+	d.e.mu.RLock()
+	addr := d.e.addr
+	d.e.mu.RUnlock()
+	if !d.e.parseInboundHeader(pkt, addr) {
+		return false, nil
+	}
+	d.mgr.queuePacket(pkt, d.e.hdrSize > 0)
+	d.mgr.wakeReady()
+	return true, nil
+}
+
 // recvMMsgDispatcher uses the recvmmsg system call to read inbound packets and
 // dispatches them.
 //
@@ -115,12 +179,7 @@ func newRecvMMsgDispatcher(fd int, e *endpoint, opts *Options) (linkDispatcher,
 	if err != nil {
 		return nil, err
 	}
-	var batchSize int
-	if opts.MultiPendingPackets {
-		batchSize = int((512*1024)/(opts.MTU)) + 1
-	} else {
-		batchSize = 1
-	}
+	batchSize := int((512*1024)/(opts.MTU)) + 1
 	d := &recvMMsgDispatcher{
 		StopFD:  stopFD,
 		fd:      fd,

stack_mixed.go

@@ -78,7 +78,7 @@ func (m *Mixed) tunLoop() {
 			return
 		}
 	}
-	if darwinTUN, isDarwinTUN := m.tun.(DarwinTUN); isDarwinTUN && m.multiPendingPackets {
+	if darwinTUN, isDarwinTUN := m.tun.(DarwinTUN); isDarwinTUN && m.recvMsgX {
 		m.batchLoopDarwin(darwinTUN)
 		return
 	}

stack_system.go

@@ -49,7 +49,7 @@ type System struct {
 	interfaceFinder      control.InterfaceFinder
 	frontHeadroom        int
 	txChecksumOffload    bool
-	multiPendingPackets  bool
+	recvMsgX             bool
 }
 
 type Session struct {
@@ -75,7 +75,7 @@ func NewSystem(options StackOptions) (Stack, error) {
 		broadcastAddr:        BroadcastAddr(options.TunOptions.Inet4Address),
 		bindInterface:        options.ForwarderBindInterface,
 		interfaceFinder:      options.InterfaceFinder,
-		multiPendingPackets:  options.TunOptions.EXP_MultiPendingPackets,
+		recvMsgX:             options.TunOptions.EXP_RecvMsgX,
 	}
 	if len(options.TunOptions.Inet4Address) > 0 {
 		if !HasNextAddress(options.TunOptions.Inet4Address[0], 1) {
@@ -176,7 +176,7 @@ func (s *System) tunLoop() {
 			return
 		}
 	}
-	if darwinTUN, isDarwinTUN := s.tun.(DarwinTUN); isDarwinTUN && s.multiPendingPackets {
+	if darwinTUN, isDarwinTUN := s.tun.(DarwinTUN); isDarwinTUN && s.recvMsgX {
 		s.batchLoopDarwin(darwinTUN)
 		return
 	}

tun.go

@@ -97,8 +97,9 @@ type Options struct {
 	// For library usages.
 	EXP_DisableDNSHijack bool
 
-	EXP_MultiPendingPackets bool
-	EXP_WriteMsgX           bool
+	// For darwin tun
+	EXP_RecvMsgX bool
+	EXP_SendMsgX bool
 }
 
 func (o *Options) Inet4GatewayAddr() netip.Addr {

tun_darwin.go

@@ -110,14 +110,14 @@ func New(options Options) (Tun, error) {
 			unix.Close(tunFd)
 			return nil, err
 		}
-		err = configure(tunFd, options.EXP_MultiPendingPackets, batchSize)
+		err = configure(tunFd, options.EXP_RecvMsgX, batchSize)
 		if err != nil {
 			unix.Close(tunFd)
 			return nil, err
 		}
 	} else {
 		tunFd = options.FileDescriptor
-		err := configure(tunFd, options.EXP_MultiPendingPackets, batchSize)
+		err := configure(tunFd, options.EXP_RecvMsgX, batchSize)
 		if err != nil {
 			return nil, err
 		}
@@ -132,7 +132,7 @@ func New(options Options) (Tun, error) {
 		msgHdrs:       make([]rawfile.MsgHdrX, batchSize),
 		msgHdrsOutput: make([]rawfile.MsgHdrX, batchSize),
 		stopFd:        common.Must1(stopfd.New()),
-		writeMsgX:     options.EXP_WriteMsgX,
+		writeMsgX:     options.EXP_SendMsgX,
 	}
 	for i := 0; i < batchSize; i++ {
 		nativeTun.iovecs[i] = newIovecBuffer(int(options.MTU))
@@ -316,12 +316,12 @@ func create(tunFd int, ifIndex int, name string, options Options) error {
 	return nil
 }
 
-func configure(tunFd int, multiPendingPackets bool, batchSize int) error {
+func configure(tunFd int, recvMsgX bool, batchSize int) error {
 	err := unix.SetNonblock(tunFd, true)
 	if err != nil {
 		return os.NewSyscallError("SetNonblock", err)
 	}
-	if multiPendingPackets {
+	if recvMsgX {
 		const UTUN_OPT_MAX_PENDING_PACKETS = 16
 		err = unix.SetsockoptInt(tunFd, 2, UTUN_OPT_MAX_PENDING_PACKETS, batchSize)
 		if err != nil {

tun_darwin_gvisor.go

@@ -37,12 +37,11 @@ func (t *NativeTun) WritePacket(pkt *stack.PacketBuffer) (int, error) {
 
 func (t *NativeTun) NewEndpoint() (stack.LinkEndpoint, stack.NICOptions, error) {
 	ep, err := fdbased.New(&fdbased.Options{
-		FDs:                 []int{t.tunFd},
-		MTU:                 t.options.MTU,
-		RXChecksumOffload:   true,
-		PacketDispatchMode:  fdbased.RecvMMsg,
-		MultiPendingPackets: t.options.EXP_MultiPendingPackets,
-		WriteMsgX:           t.options.EXP_WriteMsgX,
+		FDs:               []int{t.tunFd},
+		MTU:               t.options.MTU,
+		RXChecksumOffload: true,
+		RecvMsgX:          t.options.EXP_RecvMsgX,
+		SendMsgX:          t.options.EXP_SendMsgX,
 	})
 	if err != nil {
 		return nil, stack.NICOptions{}, err