Fix usages of readmsg_x

Author: nekohasekai

internal/fdbased_darwin/endpoint.go

@@ -285,16 +285,12 @@ func New(opts *Options) (stack.LinkEndpoint, error) {
 			return nil, fmt.Errorf("unix.SetNonblock(%v) failed: %v", fd, err)
 		}
 
-		isSocket, err := isSocketFD(fd)
-		if err != nil {
-			return nil, err
-		}
-		e.fds = append(e.fds, fdInfo{fd: fd, isSocket: isSocket})
+		e.fds = append(e.fds, fdInfo{fd: fd, isSocket: true})
 		if opts.ProcessorsPerChannel == 0 {
 			opts.ProcessorsPerChannel = common.Max(1, runtime.GOMAXPROCS(0)/len(opts.FDs))
 		}
 
-		inboundDispatcher, err := createInboundDispatcher(e, fd, isSocket, opts)
+		inboundDispatcher, err := newRecvMMsgDispatcher(fd, e, opts)
 		if err != nil {
 			return nil, fmt.Errorf("createInboundDispatcher(...) = %v", err)
 		}
@@ -304,40 +300,6 @@ func New(opts *Options) (stack.LinkEndpoint, error) {
 	return e, nil
 }
 
-func createInboundDispatcher(e *endpoint, fd int, isSocket bool, opts *Options) (linkDispatcher, error) {
-	// By default use the readv() dispatcher as it works with all kinds of
-	// FDs (tap/tun/unix domain sockets and af_packet).
-	inboundDispatcher, err := newReadVDispatcher(fd, e, opts)
-	if err != nil {
-		return nil, fmt.Errorf("newReadVDispatcher(%d, %+v) = %v", fd, e, err)
-	}
-
-	if isSocket {
-		switch e.packetDispatchMode {
-		case RecvMMsg:
-			// If the provided FD is a socket then we optimize
-			// packet reads by using recvmmsg() instead of read() to
-			// read packets in a batch.
-			inboundDispatcher, err = newRecvMMsgDispatcher(fd, e, opts)
-			if err != nil {
-				return nil, fmt.Errorf("newRecvMMsgDispatcher(%d, %+v) = %v", fd, e, err)
-			}
-		case Readv:
-		default:
-			return nil, fmt.Errorf("unknown dispatch mode %d", e.packetDispatchMode)
-		}
-	}
-	return inboundDispatcher, nil
-}
-
-func isSocketFD(fd int) (bool, error) {
-	var stat unix.Stat_t
-	if err := unix.Fstat(fd, &stat); err != nil {
-		return false, fmt.Errorf("unix.Fstat(%v,...) failed: %v", fd, err)
-	}
-	return (stat.Mode & unix.S_IFSOCK) == unix.S_IFSOCK, nil
-}
-
 // Attach launches the goroutine that reads packets from the file descriptor and
 // dispatches them via the provided dispatcher. If one is already attached,
 // then nothing happens.

internal/fdbased_darwin/packet_dispatchers.go

@@ -79,70 +79,6 @@ 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.
 //
@@ -216,7 +152,7 @@ func (d *recvMMsgDispatcher) dispatch() (bool, tcpip.Error) {
 	for k := range d.msgHdrs {
 		iovecs := d.bufs[k].nextIovecs()
 		iovLen := len(iovecs)
-		d.msgHdrs[k].DataLen = 0
+		d.msgHdrs[k] = rawfile.MsgHdrX{}
 		d.msgHdrs[k].Msg.Iov = &iovecs[0]
 		d.msgHdrs[k].Msg.SetIovlen(iovLen)
 	}

stack_mixed.go

@@ -3,9 +3,6 @@
 package tun
 
 import (
-	"errors"
-	"os"
-
 	"github.com/sagernet/gvisor/pkg/buffer"
 	"github.com/sagernet/gvisor/pkg/tcpip"
 	gHdr "github.com/sagernet/gvisor/pkg/tcpip/header"
@@ -81,9 +78,8 @@ func (m *Mixed) tunLoop() {
 		}
 	}
 	if darwinTUN, isDarwinTUN := m.tun.(DarwinTUN); isDarwinTUN {
-		if m.batchLoopDarwin(darwinTUN) {
-			return
-		}
+		m.batchLoopDarwin(darwinTUN)
+		return
 	}
 	packetBuffer := make([]byte, m.mtu+PacketOffset)
 	for {
@@ -167,15 +163,13 @@ func (m *Mixed) batchLoopLinux(linuxTUN LinuxTUN, batchSize int) {
 	}
 }
 
-func (m *Mixed) batchLoopDarwin(darwinTUN DarwinTUN) bool {
+func (m *Mixed) batchLoopDarwin(darwinTUN DarwinTUN) {
 	var writeBuffers []*buf.Buffer
 	for {
 		buffers, err := darwinTUN.BatchRead()
 		if err != nil {
-			if errors.Is(err, os.ErrInvalid) {
-				return false
-			} else if E.IsClosed(err) {
-				return true
+			if E.IsClosed(err) {
+				return
 			}
 			m.logger.Error(E.Cause(err, "batch read packet"))
 		}

stack_system.go

@@ -5,7 +5,6 @@ import (
 	"errors"
 	"net"
 	"net/netip"
-	"os"
 	"syscall"
 	"time"
 
@@ -176,9 +175,8 @@ func (s *System) tunLoop() {
 		}
 	}
 	if darwinTUN, isDarwinTUN := s.tun.(DarwinTUN); isDarwinTUN {
-		if s.batchLoopDarwin(darwinTUN) {
-			return
-		}
+		s.batchLoopDarwin(darwinTUN)
+		return
 	}
 	packetBuffer := make([]byte, s.mtu+PacketOffset)
 	for {
@@ -262,15 +260,13 @@ func (s *System) batchLoopLinux(linuxTUN LinuxTUN, batchSize int) {
 	}
 }
 
-func (s *System) batchLoopDarwin(darwinTUN DarwinTUN) bool {
+func (s *System) batchLoopDarwin(darwinTUN DarwinTUN) {
 	var writeBuffers []*buf.Buffer
 	for {
 		buffers, err := darwinTUN.BatchRead()
 		if err != nil {
-			if errors.Is(err, os.ErrInvalid) {
-				return false
-			} else if E.IsClosed(err) {
-				return true
+			if E.IsClosed(err) {
+				return
 			}
 			s.logger.Error(E.Cause(err, "batch read packet"))
 		}

tun_darwin.go

@@ -354,7 +354,7 @@ func (t *NativeTun) BatchSize() int {
 func (t *NativeTun) BatchRead() ([]*buf.Buffer, error) {
 	for i := 0; i < t.batchSize; i++ {
 		iovecs := t.iovecs[i].nextIovecs()
-		t.msgHdrs[i].DataLen = 0
+		t.msgHdrs[i] = rawfile.MsgHdrX{}
 		t.msgHdrs[i].Msg.Iov = &iovecs[0]
 		t.msgHdrs[i].Msg.Iovlen = 2
 	}
@@ -365,11 +365,7 @@ func (t *NativeTun) BatchRead() ([]*buf.Buffer, error) {
 			t.iovecs[k].buffer = nil
 		}
 		t.buffers = t.buffers[:0]
-		if errors.Is(errno, unix.EINVAL) {
-			return nil, os.ErrInvalid
-		} else {
-			return nil, errno
-		}
+		return nil, errno
 	}
 	if n < 1 {
 		return nil, nil