perf(vmnetsock): batch stream input and join shutdown

internal/vzbridge/vmnet_interface.m

@@ -164,21 +164,23 @@ int32_t vz_writeVmnetPacket(void *pointer, const void *buffer, uint64_t size, vo
     return 0;
 }
 
-// vmnet limits each call to 256 KB. Use decimal KB conservatively.
 #if __MAC_OS_X_VERSION_MAX_ALLOWED >= 260000
 static int32_t transferPackets(void *pointer, void *buffers, void *sizes, int32_t requested, bool reading, void **errorOut)
 {
     struct VmnetSession *session = pointer;
     uint64_t limit = reading ? session->readMaxPackets : session->writeMaxPackets;
     if (!limit)
         limit = 1;
+    // Match vmnet.packetBatchLimit so the descriptor arrays stay bounded.
     if (requested < 1 || requested > 32) {
         interfaceError(VMNET_INVALID_ARGUMENT, @"packet batch count", errorOut);
         return 0;
     }
     struct iovec vectors[32];
     struct vmpktdesc packets[32];
     uint64_t *lengths = sizes;
+    // Apple documents 256 KB per call. Use decimal KB, as in vmnet.packetBatchBytes.
+    // https://developer.apple.com/documentation/vmnet
     uint64_t remaining = 256000;
     int submitted = 0;
     for (; submitted < requested && submitted < limit; submitted++) {

vmnet/batch.go

@@ -11,7 +11,14 @@ import (
 	"github.com/Code-Hex/vz/v4/internal/vzbridge"
 )
 
+// Bound the native descriptor arrays to 32 packets per call. This is our batch
+// size; the bridge also checks vmnet's per-interface packet limits.
+// See https://developer.apple.com/documentation/vmnet/vmnet_read_max_packets_key
 const packetBatchLimit = 32
+
+// Apple documents a 256 KB limit per read/write call. Use decimal KB to keep
+// this cap conservative; the documentation does not give an exact byte count.
+// See https://developer.apple.com/documentation/vmnet
 const packetBatchBytes = 256000
 
 // ReadPackets reads Ethernet frames into frames and returns the number read.

vmnetsock/adapter.go

@@ -72,22 +72,17 @@ func New(ctx context.Context, iface *vmnet.Interface, transport Transport) (*Ada
 
 	var pumps sync.WaitGroup
 	pumps.Go(func() {
-		s.finish(s.forwardToInterface(iface, transport))
+		s.finish(s.forwardToInterface(run, iface, transport))
 	})
 	pumps.Go(func() {
 		s.finish(s.forwardToPeer(run, iface, transport))
 	})
 	go func() {
+		<-run.Done()
+		s.finish(run.Err())
 		pumps.Wait()
 		close(s.done)
 	}()
-	go func() {
-		select {
-		case <-ctx.Done():
-			s.finish(ctx.Err())
-		case <-s.done:
-		}
-	}()
 
 	return &Adapter{state: s}, nil
 }
@@ -116,25 +111,33 @@ func (a *Adapter) Dup() (*os.File, error) {
 	return os.NewFile(uintptr(fd), "vmnet-peer"), nil
 }
 
-func (s *adapterState) forwardToInterface(iface *vmnet.Interface, transport Transport) error {
-	frame := make([]byte, iface.MaxPacketSize())
+func (s *adapterState) forwardToInterface(ctx context.Context, iface *vmnet.Interface, transport Transport) error {
+	var reader *streamPacketReader
+	var frame []byte
+	if transport == Stream {
+		reader = newStreamPacketReader(s.conn, iface.MaxPacketSize())
+	} else {
+		frame = make([]byte, iface.MaxPacketSize())
+	}
 
 	for {
-		var n int
+		if err := ctx.Err(); err != nil {
+			return err
+		}
 		var err error
 		if transport == Stream {
-			n, err = readStreamFrame(s.conn, frame)
+			var frames [][]byte
+			frames, err = reader.readPackets()
+			if err == nil {
+				_, err = iface.WritePackets(frames)
+			}
 		} else {
+			var n int
 			n, err = readDatagramFrame(s.conn, frame)
-		}
-		if err != nil {
-			if errors.Is(err, io.EOF) {
-				return nil
+			if err == nil {
+				err = iface.WritePacket(frame[:n])
 			}
-			return err
 		}
-
-		err = iface.WritePacket(frame[:n])
 		if vmnetBufferExhausted(err) {
 			continue
 		}
@@ -149,13 +152,17 @@ func (s *adapterState) forwardToInterface(iface *vmnet.Interface, transport Tran
 
 func (s *adapterState) forwardToPeer(ctx context.Context, iface *vmnet.Interface, transport Transport) error {
 	size := iface.MaxPacketSize()
-	count := min(32, max(1, 256000/size))
+	count := packetBatchSize(size)
 	frames := make([][]byte, count)
 	for j := range frames {
 		frames[j] = make([]byte, size)
 	}
-	headers := make([][4]byte, count)
-	vectors := make(net.Buffers, 2*count)
+	var headers [][4]byte
+	var vectors net.Buffers
+	if transport == Stream {
+		headers = make([][4]byte, count)
+		vectors = make(net.Buffers, 2*count)
+	}
 	for {
 		n, err := iface.ReadPackets(ctx, frames)
 		if err != nil {

vmnetsock/adapter_test.go

@@ -43,10 +43,12 @@ func TestAdapterFramesAndLifetime(t *testing.T) {
 			if err != nil {
 				t.Fatal(err)
 			}
+			t.Cleanup(func() { objc.Release(c) })
 			n, err := vmnet.NewNetwork(c)
 			if err != nil {
 				t.Fatal(err)
 			}
+			t.Cleanup(func() { objc.Release((*objc.Pointer)(n)) })
 			a, b := newInterface(t, n), newInterface(t, n)
 			ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
 			defer cancel()
@@ -146,11 +148,16 @@ func TestAdapterCancellation(t *testing.T) {
 	if macos.Available(26, vzbridge.SDKVersion) != nil {
 		t.Skip("requires macOS 26 SDK")
 	}
-	c, _ := vmnet.NewNetworkConfiguration(vmnet.ModeHost)
+	c, err := vmnet.NewNetworkConfiguration(vmnet.ModeHost)
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(func() { objc.Release(c) })
 	n, err := vmnet.NewNetwork(c)
 	if err != nil {
 		t.Fatal(err)
 	}
+	t.Cleanup(func() { objc.Release((*objc.Pointer)(n)) })
 	for _, transport := range []vmnetsock.Transport{vmnetsock.Stream, vmnetsock.Datagram} {
 		i := newInterface(t, n)
 		ctx, cancel := context.WithCancel(context.Background())
@@ -186,10 +193,12 @@ func TestAdapterStreamTermination(t *testing.T) {
 	if err != nil {
 		t.Fatal(err)
 	}
+	t.Cleanup(func() { objc.Release(c) })
 	n, err := vmnet.NewNetwork(c)
 	if err != nil {
 		t.Fatal(err)
 	}
+	t.Cleanup(func() { objc.Release((*objc.Pointer)(n)) })
 	for _, test := range []struct {
 		name      string
 		payload   []byte

vmnetsock/backpressure_test.go

@@ -11,6 +11,7 @@ import (
 	"time"
 
 	"github.com/Code-Hex/vz/v4/internal/macos"
+	"github.com/Code-Hex/vz/v4/internal/objc"
 	"github.com/Code-Hex/vz/v4/internal/vzbridge"
 	"github.com/Code-Hex/vz/v4/vmnet"
 )
@@ -26,10 +27,12 @@ func TestAdapterBackpressureCleanup(t *testing.T) {
 				if err != nil {
 					t.Fatal(err)
 				}
+				t.Cleanup(func() { objc.Release(c) })
 				n, err := vmnet.NewNetwork(c)
 				if err != nil {
 					t.Fatal(err)
 				}
+				t.Cleanup(func() { objc.Release((*objc.Pointer)(n)) })
 				receiver, err := vmnet.StartInterfaceWithNetwork(n)
 				if err != nil {
 					t.Fatal(err)

vmnetsock/framing.go

@@ -1,6 +1,7 @@
 package vmnetsock
 
 import (
+	"bufio"
 	"encoding/binary"
 	"fmt"
 	"io"
@@ -9,22 +10,79 @@ import (
 	"golang.org/x/sys/unix"
 )
 
-func readStreamFrame(reader io.Reader, frame []byte) (int, error) {
-	var header [4]byte
-	if _, err := io.ReadFull(reader, header[:]); err != nil {
-		return 0, err
+// Apple documents a 256 KB limit per read/write call. Use the same conservative
+// decimal-KB cap as vmnet.WritePackets when grouping frames for the interface.
+// See https://developer.apple.com/documentation/vmnet
+const packetBatchBytes = 256000
+
+func packetBatchSize(size int) int {
+	// Match vmnet's internal 32-packet batch size, not Apple's packet limit.
+	return min(32, max(1, packetBatchBytes/size))
+}
+
+type streamPacketReader struct {
+	reader   *bufio.Reader
+	frames   [][]byte
+	consumed int
+	maxSize  int
+}
+
+func newStreamPacketReader(reader io.Reader, maxSize int) *streamPacketReader {
+	maxSize = min(maxSize, packetBatchBytes)
+	count := packetBatchSize(maxSize)
+	return &streamPacketReader{
+		reader:  bufio.NewReaderSize(reader, count*(maxSize+4)),
+		frames:  make([][]byte, 0, count),
+		maxSize: maxSize,
 	}
+}
+
+// readPackets borrows views of the reader's buffer until the next call.
+// Only the first packet waits for input; the rest must already be buffered.
+func (r *streamPacketReader) readPackets() ([][]byte, error) {
+	_, _ = r.reader.Discard(r.consumed)
+	r.consumed = 0
+	r.frames = r.frames[:0]
 
-	size := uint64(binary.BigEndian.Uint32(header[:]))
-	if size == 0 || size > uint64(len(frame)) {
-		return 0, fmt.Errorf("invalid stream frame length %d, maximum %d", size, len(frame))
+	header, err := r.reader.Peek(4)
+	if err != nil {
+		if err == io.EOF && len(header) != 0 {
+			err = io.ErrUnexpectedEOF
+		}
+		return nil, err
+	}
+	size, err := streamFrameSize(header, r.maxSize)
+	if err != nil {
+		return nil, err
+	}
+	if _, err := r.reader.Peek(4 + size); err != nil {
+		if err == io.EOF {
+			err = io.ErrUnexpectedEOF
+		}
+		return nil, err
 	}
 
-	_, err := io.ReadFull(reader, frame[:int(size)])
-	if err == io.EOF {
-		err = io.ErrUnexpectedEOF
+	data, _ := r.reader.Peek(r.reader.Buffered())
+	for len(data) >= 4 && len(r.frames) < cap(r.frames) {
+		size, err := streamFrameSize(data[:4], r.maxSize)
+		if err != nil || len(data) < 4+size {
+			// Leave an invalid or incomplete suffix for the next call.
+			break
+		}
+		end := 4 + size
+		r.frames = append(r.frames, data[4:end:end])
+		r.consumed += end
+		data = data[end:]
+	}
+	return r.frames, nil
+}
+
+func streamFrameSize(header []byte, maximum int) (int, error) {
+	size := uint64(binary.BigEndian.Uint32(header))
+	if size == 0 || size > uint64(maximum) {
+		return 0, fmt.Errorf("invalid stream frame length %d, maximum %d", size, maximum)
 	}
-	return int(size), err
+	return int(size), nil
 }
 
 func writeStreamFrames(writer io.Writer, frames [][]byte, headers [][4]byte, vectors net.Buffers) error {