diff --git a/internal/fdbased_darwin/packet_dispatchers.go b/internal/fdbased_darwin/packet_dispatchers.go index c102f6c..afe362f 100644 --- a/internal/fdbased_darwin/packet_dispatchers.go +++ b/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) } diff --git a/internal/rawfile_darwin/rawfile.go b/internal/rawfile_darwin/rawfile.go index e3a061b..e586ef2 100644 --- a/internal/rawfile_darwin/rawfile.go +++ b/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) { - // Create kqueue +type Poller struct { + 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) - - // 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{ + kevents := []unix.Kevent_t{ + { + Ident: uint64(efd), + Filter: unix.EVFILT_READ, + Flags: unix.EV_ADD | unix.EV_ENABLE, + }, + { 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) - revents := make([]unix.Kevent_t, len(kevents)) - n, err := unix.Kevent(kq, nil, revents, nil) +func (p *Poller) Wait() (bool, unix.Errno) { + var revents [2]unix.Kevent_t + 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 { - // Check for errors or EOF + if int(ev.Ident) == p.fd { 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) +} diff --git a/stack_system.go b/stack_system.go index 46f6f87..3cb0cb0 100644 --- a/stack_system.go +++ b/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 { - tun Tun +func newSystemWriteback(tunInterface Tun, frontHeadroom int) ForwardWriteback { + 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 { - writeback := &systemWriteback{tun: tunInterface, frontHeadroom: frontHeadroom} - if linuxTUN, isLinuxTUN := tunInterface.(LinuxTUN); isLinuxTUN { - writeback.linuxTUN = linuxTUN +func (w *systemWritebackLinux) ReturnHeadroom() int { + return w.frontHeadroom + PacketOffset +} + +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 { diff --git a/tun_darwin.go b/tun_darwin.go index 8f7a186..4b00ac5 100644 --- a/tun_darwin.go +++ b/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)