feat(vmnet): add batched packet I/O

internal/vzbridge/native.h

@@ -102,8 +102,9 @@ void vz_releaseCFObject(void *object);
 void *vz_startVmnetInterface(VZBridgeObject network, const char *name, uint32_t options, uint64_t callback, uint64_t context, void *maxPacketSize, void *mac, VZBridgeErrorOut errorOut) __attribute__((annotate("vzbridge:noqueue")));
 VZBridgeOwnedObject vz_copyVmnetBridgedInterfaces(VZBridgeErrorOut errorOut) __attribute__((annotate("vzbridge:noqueue")));
 void vz_stopVmnetInterface(void *session, VZBridgeErrorOut errorOut) __attribute__((annotate("vzbridge:noqueue")));
-int32_t vz_readVmnetPacket(void *session, void *buffer, uint64_t size, VZBridgeErrorOut errorOut) __attribute__((annotate("vzbridge:noqueue")));
-void vz_writeVmnetPacket(void *session, const void *buffer, uint64_t size, VZBridgeErrorOut errorOut) __attribute__((annotate("vzbridge:noqueue")));
+int32_t vz_writeVmnetPacket(void *session, const void *buffer, uint64_t size, VZBridgeErrorOut errorOut) __attribute__((annotate("vzbridge:noqueue")));
+int32_t vz_readVmnetPackets(void *session, void *buffers, void *sizes, int32_t count, VZBridgeErrorOut errorOut) __attribute__((annotate("vzbridge:noqueue")));
+int32_t vz_writeVmnetPackets(void *session, void *buffers, void *sizes, int32_t count, VZBridgeErrorOut errorOut) __attribute__((annotate("vzbridge:noqueue")));
 void *vz_copyVmnetNetworkSerialization(VZBridgeObject network, VZBridgeErrorOut errorOut);
 void *vz_newVmnetNetworkWithSerialization(VZBridgeObject serialization, VZBridgeErrorOut errorOut);
 void vz_releaseXPCObject(void *object) __attribute__((annotate("vzbridge:noqueue")));

internal/vzbridge/vmnet_interface.m

@@ -7,6 +7,7 @@
     interface_ref interface;
     dispatch_queue_t events;
     dispatch_block_t disableCallback;
+    uint64_t readMaxPackets, writeMaxPackets;
 };
 static void interfaceError(vmnet_return_t status, NSString *operation, void **output)
 {
@@ -43,6 +44,8 @@ static void interfaceError(vmnet_return_t status, NSString *operation, void **ou
             result = status;
             if (status == VMNET_SUCCESS && parameters) {
                 *(uint64_t *)maxPacketSize = xpc_dictionary_get_uint64(parameters, vmnet_max_packet_size_key);
+                session->readMaxPackets = xpc_dictionary_get_uint64(parameters, vmnet_read_max_packets_key);
+                session->writeMaxPackets = xpc_dictionary_get_uint64(parameters, vmnet_write_max_packets_key);
                 const char *address = xpc_dictionary_get_string(parameters, vmnet_mac_address_key);
                 unsigned int bytes[6];
                 if (address && sscanf(address, "%x:%x:%x:%x:%x:%x", &bytes[0], &bytes[1], &bytes[2], &bytes[3], &bytes[4], &bytes[5]) == 6)
@@ -137,30 +140,83 @@ void vz_stopVmnetInterface(void *pointer, void **errorOut)
     }
 #endif
 }
-int32_t vz_readVmnetPacket(void *pointer, void *buffer, uint64_t size, void **errorOut)
+int32_t vz_writeVmnetPacket(void *pointer, const void *buffer, uint64_t size, void **errorOut)
 {
 #if __MAC_OS_X_VERSION_MAX_ALLOWED >= 260000
     if (@available(macOS 26, *)) {
         struct VmnetSession *session = pointer;
-        struct iovec vec = { .iov_base = buffer, .iov_len = size };
-        struct vmpktdesc packet = { .vm_pkt_size = size, .vm_pkt_iov = &vec, .vm_pkt_iovcnt = 1, .vm_flags = 0 };
+        struct iovec vector = { .iov_base = (void *)buffer, .iov_len = size };
+        struct vmpktdesc packet = { .vm_pkt_size = size, .vm_pkt_iov = &vector, .vm_pkt_iovcnt = 1, .vm_flags = 0 };
         int count = 1;
-        vmnet_return_t result = vmnet_read(session->interface, &packet, &count);
-        interfaceError(result, @"vmnet_read", errorOut);
-        return result == VMNET_SUCCESS && count ? (int32_t)packet.vm_pkt_size : 0;
+        vmnet_return_t result = vmnet_write(session->interface, &packet, &count);
+        interfaceError(result, @"vmnet_write", errorOut);
+        return result == VMNET_SUCCESS ? count : 0;
     }
 #endif
     return 0;
 }
-void vz_writeVmnetPacket(void *pointer, const void *buffer, uint64_t size, void **errorOut)
-{
+
+// vmnet limits each call to 256 KB. Use decimal KB conservatively.
 #if __MAC_OS_X_VERSION_MAX_ALLOWED >= 260000
-    if (@available(macOS 26, *)) {
-        struct VmnetSession *session = pointer;
-        struct iovec vec = { .iov_base = (void *)buffer, .iov_len = size };
-        struct vmpktdesc packet = { .vm_pkt_size = size, .vm_pkt_iov = &vec, .vm_pkt_iovcnt = 1, .vm_flags = 0 };
-        int count = 1;
-        interfaceError(vmnet_write(session->interface, &packet, &count), @"vmnet_write", errorOut);
+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;
+    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;
+    uint64_t remaining = 256000;
+    int submitted = 0;
+    for (; submitted < requested && submitted < limit; submitted++) {
+        uint64_t size = lengths[submitted];
+        if (!size || size > remaining)
+            break;
+        remaining -= size;
+        vectors[submitted] = (struct iovec) { .iov_base = ((void **)buffers)[submitted], .iov_len = size };
+        packets[submitted] = (struct vmpktdesc) { .vm_pkt_size = size, .vm_pkt_iov = &vectors[submitted], .vm_pkt_iovcnt = 1, .vm_flags = 0 };
+    }
+    if (!submitted) {
+        interfaceError(VMNET_INVALID_ARGUMENT, @"packet batch size", errorOut);
+        return 0;
+    }
+    int count = submitted;
+    vmnet_return_t result = reading ? vmnet_read(session->interface, packets, &count)
+                                    : vmnet_write(session->interface, packets, &count);
+    interfaceError(result, reading ? @"vmnet_read" : @"vmnet_write", errorOut);
+    // An error may leave count unchanged. Only successful calls report progress.
+    if (result != VMNET_SUCCESS)
+        return 0;
+    if (count < 0 || count > submitted) {
+        interfaceError(VMNET_FAILURE, @"packet batch result count", errorOut);
+        return 0;
+    }
+    if (reading)
+        for (int i = 0; i < count; i++)
+            lengths[i] = packets[i].vm_pkt_size;
+    return count;
+}
 #endif
+
+int32_t vz_readVmnetPackets(void *pointer, void *buffers, void *sizes, int32_t count, void **errorOut)
+{
+#if __MAC_OS_X_VERSION_MAX_ALLOWED >= 260000
+    if (@available(macOS 26, *))
+        return transferPackets(pointer, buffers, sizes, count, true, errorOut);
+#endif
+    return 0;
+}
+
+int32_t vz_writeVmnetPackets(void *pointer, void *buffers, void *sizes, int32_t count, void **errorOut)
+{
+#if __MAC_OS_X_VERSION_MAX_ALLOWED >= 260000
+    if (@available(macOS 26, *))
+        return transferPackets(pointer, buffers, sizes, count, false, errorOut);
+#endif
+    return 0;
 }

vmnet/batch.go

@@ -0,0 +1,181 @@
+package vmnet
+
+import (
+	"context"
+	"fmt"
+	"io"
+	"net"
+	"runtime"
+	"unsafe"
+
+	"github.com/Code-Hex/vz/v4/internal/vzbridge"
+)
+
+const packetBatchLimit = 32
+const packetBatchBytes = 256000
+
+// ReadPackets reads Ethernet frames into frames and returns the number read.
+// It waits for at least one frame, but does not wait to fill the batch.
+// A read may return fewer frames than fit, even when more frames are queued.
+//
+// Each buffer must have capacity for [Interface.MaxPacketSize] bytes. On success,
+// ReadPackets sets the lengths of frames[:n] to the received frame lengths.
+// The same buffers can be passed to the next call without resetting their lengths.
+// The buffers must not overlap. The caller must not use frames or its buffers
+// during the call, and ReadPackets does not retain them after it returns.
+//
+// Canceling ctx or calling [Interface.Stop] interrupts a blocked read, including
+// a read waiting for another reader to finish. On an open interface, an empty
+// batch returns (0, nil).
+//
+// See https://developer.apple.com/documentation/vmnet/vmnet_read(_:_:_:)?language=objc
+func (i *Interface) ReadPackets(ctx context.Context, frames [][]byte) (int, error) {
+	if i == nil || i.owner == nil {
+		return 0, fmt.Errorf("vmnet interface is nil or uninitialized")
+	}
+	defer runtime.KeepAlive(i)
+	r := i.owner.resource
+	select {
+	case <-r.done:
+		return 0, net.ErrClosed
+	default:
+	}
+	for _, frame := range frames {
+		if cap(frame) < r.maxSize {
+			return 0, io.ErrShortBuffer
+		}
+	}
+	if len(frames) == 0 {
+		return 0, nil
+	}
+	if r.maxSize > packetBatchBytes {
+		return 0, fmt.Errorf("vmnet maximum packet size %d exceeds batch byte limit %d", r.maxSize, packetBatchBytes)
+	}
+	select {
+	case r.readers <- struct{}{}:
+	case <-ctx.Done():
+		return 0, ctx.Err()
+	case <-r.done:
+		return 0, net.ErrClosed
+	}
+	defer func() { <-r.readers }()
+	frames = frames[:min(len(frames), packetBatchLimit, packetBatchBytes/r.maxSize)]
+	for {
+		if err := ctx.Err(); err != nil {
+			return 0, err
+		}
+		r.mu.Lock()
+		if r.stopping {
+			r.mu.Unlock()
+			return 0, net.ErrClosed
+		}
+		n, err := r.transferPackets(frames, true)
+		r.mu.Unlock()
+		if err != nil || n > 0 {
+			return n, err
+		}
+		// Events are wake hints, not packet counts. Drain native packets before waiting.
+		select {
+		case <-r.wake:
+		case <-ctx.Done():
+			return 0, ctx.Err()
+		case <-r.done:
+			return 0, net.ErrClosed
+		}
+	}
+}
+
+// WritePackets sends Ethernet frames and returns the number sent.
+// It checks every frame before sending any, then splits the frames into native
+// batches as needed. Each frame must contain between 1 and
+// min(MaxPacketSize(), 256000) bytes.
+//
+// On error, n counts only frames confirmed by successful native calls. The failed
+// call may have sent additional frames, so retrying frames[n:] can send duplicates.
+// Concurrent writers may interleave their batches.
+//
+// The caller must not change frames or its buffers during the call.
+// WritePackets does not retain them after it returns. On an open interface,
+// an empty batch returns (0, nil).
+//
+// See https://developer.apple.com/documentation/vmnet/vmnet_write(_:_:_:)?language=objc
+func (i *Interface) WritePackets(frames [][]byte) (int, error) {
+	if i == nil || i.owner == nil {
+		return 0, fmt.Errorf("vmnet interface is nil or uninitialized")
+	}
+	defer runtime.KeepAlive(i)
+	r := i.owner.resource
+	select {
+	case <-r.done:
+		return 0, net.ErrClosed
+	default:
+	}
+	for _, frame := range frames {
+		if len(frame) == 0 || len(frame) > min(r.maxSize, packetBatchBytes) {
+			return 0, fmt.Errorf("invalid vmnet frame length %d, maximum %d", len(frame), min(r.maxSize, packetBatchBytes))
+		}
+	}
+	sent := 0
+	for sent < len(frames) {
+		r.mu.Lock()
+		if r.stopping {
+			r.mu.Unlock()
+			return sent, net.ErrClosed
+		}
+		n, err := r.transferPackets(frames[sent:min(len(frames), sent+packetBatchLimit)], false)
+		r.mu.Unlock()
+		if err != nil {
+			return sent, err
+		}
+		if n == 0 {
+			return sent, io.ErrShortWrite
+		}
+		sent += n
+	}
+	return sent, nil
+}
+
+// The resource mutex protects the session throughout this synchronous call.
+func (r *interfaceResource) transferPackets(frames [][]byte, reading bool) (int, error) {
+	var buffers [packetBatchLimit]unsafe.Pointer
+	var sizes [packetBatchLimit]uint64
+	var pinned runtime.Pinner
+	defer pinned.Unpin()
+	for j, frame := range frames {
+		size := len(frame)
+		if reading {
+			size = r.maxSize
+			frame = frame[:size]
+		}
+		buffers[j] = unsafe.Pointer(&frame[0])
+		sizes[j] = uint64(size)
+		pinned.Pin(buffers[j])
+	}
+	pinned.Pin(&buffers[0])
+	pinned.Pin(&sizes[0])
+	var errorPtr unsafe.Pointer
+	var count int32
+	if reading {
+		count = vzbridge.ReadVmnetPackets(r.session, unsafe.Pointer(&buffers[0]), unsafe.Pointer(&sizes[0]), int32(len(frames)), &errorPtr)
+	} else {
+		count = vzbridge.WriteVmnetPackets(r.session, unsafe.Pointer(&buffers[0]), unsafe.Pointer(&sizes[0]), int32(len(frames)), &errorPtr)
+	}
+	if err := vmnetError(errorPtr); err != nil {
+		return 0, err
+	}
+	n := int(count)
+	if n < 0 || n > len(frames) {
+		return 0, fmt.Errorf("vmnet returned invalid packet count %d", n)
+	}
+	if reading {
+		for j := range n {
+			if sizes[j] == 0 || sizes[j] > uint64(r.maxSize) {
+				return 0, fmt.Errorf("vmnet returned invalid frame length %d", sizes[j])
+			}
+		}
+		for j := range n {
+			frames[j] = frames[j][:int(sizes[j])]
+		}
+	}
+	return n, nil
+}