Improve darwin forwarding

internal/fdbased_darwin/packet_dispatchers.go

@@ -90,6 +90,8 @@ type recvMMsgDispatcher struct {
 	// fd is the file descriptor used to send and receive packets.
 	fd int
 
+	poller *rawfile.Poller
+
 	// e is the endpoint this dispatcher is attached to.
 	e *endpoint
 
@@ -123,9 +125,15 @@ func newRecvMMsgDispatcher(fd int, e *endpoint, opts *Options) (linkDispatcher,
 	} else {
 		batchSize = 1
 	}
+	poller, err := rawfile.NewPoller(stopFD.ReadFD, fd)
+	if err != nil {
+		stopFD.Close()
+		return nil, err
+	}
 	d := &recvMMsgDispatcher{
 		StopFD:  stopFD,
 		fd:      fd,
+		poller:  poller,
 		e:       e,
 		bufs:    make([]*iovecBuffer, batchSize),
 		msgHdrs: make([]rawfile.MsgHdrX, batchSize),
@@ -144,6 +152,7 @@ func (d *recvMMsgDispatcher) release() {
 	for _, iov := range d.bufs {
 		iov.release()
 	}
+	_ = d.poller.Close()
 	d.mgr.close()
 }
 
@@ -161,7 +170,7 @@ func (d *recvMMsgDispatcher) dispatch() (bool, tcpip.Error) {
 		d.msgHdrs[k].Msg.SetIovlen(iovLen)
 	}
 
-	nMsgs, errno := rawfile.BlockingRecvMMsgUntilStopped(d.ReadFD, d.fd, d.msgHdrs)
+	nMsgs, errno := rawfile.BlockingRecvMMsgUntilStopped(d.poller, d.fd, d.msgHdrs)
 	if errno != 0 {
 		return false, TranslateErrno(errno)
 	}

internal/rawfile_darwin/rawfile.go

@@ -69,7 +69,7 @@ func NonBlockingWriteIovec(fd int, iovec []unix.Iovec) unix.Errno {
 	return e
 }
 
-func BlockingReadvUntilStopped(efd int, fd int, iovecs []unix.Iovec) (int, unix.Errno) {
+func BlockingReadvUntilStopped(poller *Poller, fd int, iovecs []unix.Iovec) (int, unix.Errno) {
 	for {
 		//nolint:staticcheck
 		n, _, e := unix.RawSyscall(unix.SYS_READV, uintptr(fd), uintptr(unsafe.Pointer(&iovecs[0])), uintptr(len(iovecs)))
@@ -79,7 +79,7 @@ func BlockingReadvUntilStopped(efd int, fd int, iovecs []unix.Iovec) (int, unix.
 		if e != 0 && e != unix.EWOULDBLOCK {
 			return 0, e
 		}
-		stopped, e := BlockingPollUntilStopped(efd, fd, unix.POLLIN)
+		stopped, e := poller.Wait()
 		if stopped {
 			return -1, e
 		}
@@ -89,7 +89,7 @@ func BlockingReadvUntilStopped(efd int, fd int, iovecs []unix.Iovec) (int, unix.
 	}
 }
 
-func BlockingRecvMMsgUntilStopped(efd int, fd int, msgHdrs []MsgHdrX) (int, unix.Errno) {
+func BlockingRecvMMsgUntilStopped(poller *Poller, fd int, msgHdrs []MsgHdrX) (int, unix.Errno) {
 	for {
 		//nolint:staticcheck
 		n, _, e := unix.RawSyscall6(unix.SYS_RECVMSG_X, uintptr(fd), uintptr(unsafe.Pointer(&msgHdrs[0])), uintptr(len(msgHdrs)), unix.MSG_DONTWAIT, 0, 0)
@@ -101,7 +101,7 @@ func BlockingRecvMMsgUntilStopped(efd int, fd int, msgHdrs []MsgHdrX) (int, unix
 			return 0, e
 		}
 
-		stopped, e := BlockingPollUntilStopped(efd, fd, unix.POLLIN)
+		stopped, e := poller.Wait()
 		if stopped {
 			return -1, e
 		}
@@ -111,71 +111,58 @@ func BlockingRecvMMsgUntilStopped(efd int, fd int, msgHdrs []MsgHdrX) (int, unix
 	}
 }
 
-func BlockingPollUntilStopped(efd int, fd int, events int16) (bool, unix.Errno) {
+type Poller struct {
-	// Create kqueue
+	kq  int
+	efd int
+	fd  int
+}
+
+func NewPoller(efd int, fd int) (*Poller, error) {
 	kq, err := unix.Kqueue()
 	if err != nil {
-		return false, unix.Errno(err.(unix.Errno))
+		return nil, err
 	}
-	defer unix.Close(kq)
+	kevents := []unix.Kevent_t{
-
+		{
-	// Prepare kevents for registration
-	var kevents []unix.Kevent_t
-
-	// Always monitor efd for read events
-	kevents = append(kevents, unix.Kevent_t{
 			Ident:  uint64(efd),
 			Filter: unix.EVFILT_READ,
 			Flags:  unix.EV_ADD | unix.EV_ENABLE,
-	})
+		},
-
+		{
-	// Monitor fd based on requested events
-	// Convert poll events to kqueue filters
-	if events&unix.POLLIN != 0 {
-		kevents = append(kevents, unix.Kevent_t{
 			Ident:  uint64(fd),
 			Filter: unix.EVFILT_READ,
 			Flags:  unix.EV_ADD | unix.EV_ENABLE,
-		})
+		},
 	}
-	if events&unix.POLLOUT != 0 {
-		kevents = append(kevents, unix.Kevent_t{
-			Ident:  uint64(fd),
-			Filter: unix.EVFILT_WRITE,
-			Flags:  unix.EV_ADD | unix.EV_ENABLE,
-		})
-	}
-
-	// Register events
 	_, err = unix.Kevent(kq, kevents, nil, nil)
 	if err != nil {
-		return false, unix.Errno(err.(unix.Errno))
+		unix.Close(kq)
+		return nil, err
 	}
+	return &Poller{kq: kq, efd: efd, fd: fd}, nil
+}
 
-	// Wait for events (blocking)
+func (p *Poller) Wait() (bool, unix.Errno) {
-	revents := make([]unix.Kevent_t, len(kevents))
+	var revents [2]unix.Kevent_t
-	n, err := unix.Kevent(kq, nil, revents, nil)
+	n, err := unix.Kevent(p.kq, nil, revents[:], nil)
 	if err != nil {
-		return false, unix.Errno(err.(unix.Errno))
+		return false, err.(unix.Errno)
 	}
 
-	// Check results
 	var efdHasData bool
 	var errno unix.Errno
 
 	for i := range n {
 		ev := &revents[i]
 
-		if int(ev.Ident) == efd && ev.Filter == unix.EVFILT_READ {
+		if int(ev.Ident) == p.efd && ev.Filter == unix.EVFILT_READ {
 			efdHasData = true
 		}
 
-		if int(ev.Ident) == fd {
+		if int(ev.Ident) == p.fd {
-			// Check for errors or EOF
 			if ev.Flags&unix.EV_EOF != 0 {
 				errno = unix.ECONNRESET
 			} else if ev.Flags&unix.EV_ERROR != 0 {
-				// Extract error from Data field
 				if ev.Data != 0 {
 					errno = unix.Errno(ev.Data)
 				} else {
@@ -187,3 +174,7 @@ func BlockingPollUntilStopped(efd int, fd int, events int16) (bool, unix.Errno)
 
 	return efdHasData, errno
 }
+
+func (p *Poller) Close() error {
+	return unix.Close(p.kq)
+}

stack_system.go

@@ -750,18 +750,49 @@ func (w *systemUDPPacketWriter6) WritePacket(buffer *buf.Buffer, destination M.S
 	return common.Error(w.tun.Write(newPacket.Bytes()))
 }
 
-type systemWriteback struct {
+func newSystemWriteback(tunInterface Tun, frontHeadroom int) ForwardWriteback {
-	tun           Tun
+	if linuxTUN, isLinuxTUN := tunInterface.(LinuxTUN); isLinuxTUN {
+		return &systemWritebackLinux{linuxTUN: linuxTUN, frontHeadroom: frontHeadroom}
+	}
+	if darwinTUN, isDarwinTUN := tunInterface.(DarwinTUN); isDarwinTUN {
+		return &systemWritebackDarwin{darwinTUN: darwinTUN, frontHeadroom: frontHeadroom}
+	}
+	return &systemWriteback{tun: tunInterface, frontHeadroom: frontHeadroom}
+}
+
+type systemWritebackLinux struct {
 	linuxTUN      LinuxTUN
 	frontHeadroom int
 }
 
-func newSystemWriteback(tunInterface Tun, frontHeadroom int) *systemWriteback {
+func (w *systemWritebackLinux) ReturnHeadroom() int {
-	writeback := &systemWriteback{tun: tunInterface, frontHeadroom: frontHeadroom}
+	return w.frontHeadroom + PacketOffset
-	if linuxTUN, isLinuxTUN := tunInterface.(LinuxTUN); isLinuxTUN {
+}
-		writeback.linuxTUN = linuxTUN
+
+func (w *systemWritebackLinux) WriteReturnPackets(packets [][]byte) error {
+	return common.Error(w.linuxTUN.BatchWrite(packets, w.frontHeadroom))
+}
+
+type systemWritebackDarwin struct {
+	darwinTUN     DarwinTUN
+	frontHeadroom int
+}
+
+func (w *systemWritebackDarwin) ReturnHeadroom() int {
+	return w.frontHeadroom + PacketOffset
+}
+
+func (w *systemWritebackDarwin) WriteReturnPackets(packets [][]byte) error {
+	buffers := make([]*buf.Buffer, 0, len(packets))
+	for _, packet := range packets {
+		buffers = append(buffers, buf.As(packet[PacketOffset:]))
 	}
-	return writeback
+	return w.darwinTUN.BatchWrite(buffers)
+}
+
+type systemWriteback struct {
+	tun           Tun
+	frontHeadroom int
 }
 
 func (w *systemWriteback) ReturnHeadroom() int {
@@ -769,9 +800,6 @@ func (w *systemWriteback) ReturnHeadroom() int {
 }
 
 func (w *systemWriteback) WriteReturnPackets(packets [][]byte) error {
-	if w.linuxTUN != nil {
-		return common.Error(w.linuxTUN.BatchWrite(packets, w.frontHeadroom))
-	}
 	var writeErrors []error
 	for _, packet := range packets {
 		if PacketOffset > 0 {

tun_darwin.go

@@ -6,6 +6,7 @@ import (
 	"net"
 	"net/netip"
 	"os"
+	"sync"
 	"syscall"
 	"unsafe"
 
@@ -36,6 +37,8 @@ type NativeTun struct {
 	msgHdrsOutput       []rawfile.MsgHdrX
 	buffers             []*buf.Buffer
 	stopFd              stopfd.StopFD
+	readPoller          *rawfile.Poller
+	writeAccess         sync.Mutex
 	options             Options
 	inet4Address        [4]byte
 	inet6Address        [16]byte
@@ -132,6 +135,7 @@ func New(options Options) (Tun, error) {
 		stopFd:        common.Must1(stopfd.New()),
 		sendMsgX:      options.EXP_SendMsgX,
 	}
+	nativeTun.readPoller = common.Must1(rawfile.NewPoller(nativeTun.stopFd.ReadFD, tunFd))
 	for i := range batchSize {
 		nativeTun.iovecs[i] = newIovecBuffer(int(options.MTU))
 		nativeTun.iovecsOutput[i] = newIovecBuffer(int(options.MTU))
@@ -155,15 +159,26 @@ func (t *NativeTun) Start() error {
 
 func (t *NativeTun) Close() error {
 	if t.options.EXP_ExternalConfiguration {
-		return t.tunFile.Close()
+		t.stopFd.Stop()
+		err := t.tunFile.Close()
+		t.closePollers()
+		t.stopFd.Close()
+		return err
 	}
 	defer flushDNSCache()
 	t.stopFd.Stop()
 	err := E.Errors(t.unsetRoutes(), t.tunFile.Close())
+	t.closePollers()
 	t.stopFd.Close()
 	return err
 }
 
+func (t *NativeTun) closePollers() {
+	if t.readPoller != nil {
+		_ = t.readPoller.Close()
+	}
+}
+
 func (t *NativeTun) Read(p []byte) (n int, err error) {
 	return t.tunFile.Read(p)
 }
@@ -350,7 +365,7 @@ func (t *NativeTun) BatchRead() ([]*buf.Buffer, error) {
 		t.msgHdrs[i].Msg.Iov = &iovecs[0]
 		t.msgHdrs[i].Msg.Iovlen = 2
 	}
-	n, errno := rawfile.BlockingRecvMMsgUntilStopped(t.stopFd.ReadFD, t.tunFd, t.msgHdrs)
+	n, errno := rawfile.BlockingRecvMMsgUntilStopped(t.readPoller, t.tunFd, t.msgHdrs)
 	if errno != 0 {
 		for k := range n {
 			t.iovecs[k].buffer.Release()
@@ -377,6 +392,23 @@ func (t *NativeTun) BatchRead() ([]*buf.Buffer, error) {
 }
 
 func (t *NativeTun) BatchWrite(buffers []*buf.Buffer) error {
+	t.writeAccess.Lock()
+	defer t.writeAccess.Unlock()
+	for len(buffers) > 0 {
+		chunk := buffers
+		if len(chunk) > t.batchSize {
+			chunk = chunk[:t.batchSize]
+		}
+		buffers = buffers[len(chunk):]
+		err := t.batchWriteChunk(chunk)
+		if err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+func (t *NativeTun) batchWriteChunk(buffers []*buf.Buffer) error {
 	if !t.sendMsgX {
 		for i, buffer := range buffers {
 			t.iovecsOutput[i].nextIovecsOutput(buffer)