vmnet: Refactor *FileAdaptorForInterfaces

- `StreamFileAdaptorForInterface`:
  - Support partial read on `unix.Readv` in `readPacketsFromConn`

- `Datagram*FileAdaptorForInterface`:
  - Add wait on `syscall.ENOBUFS` in `writePacketsToPacketConn`

Signed-off-by: Norio Nomura <norio.nomura@gmail.com>

Author: norio-nomura

vmnet/datagram_darwin.go

@@ -12,6 +12,7 @@ import (
 	"net"
 	"os"
 	"syscall"
+	"time"
 )
 
 // MARK: - DatagramFileAdaptorForInterface
@@ -160,7 +161,7 @@ func (v *pktDescsManager) buffersForWritingToPacketConn(packetCount int) (net.Bu
 			return nil, fmt.Errorf("vm_pkt_size %d exceeds maxPacketSize %d", vmPktDesc.vm_pkt_size, v.maxPacketSize)
 		}
 		// Resize buffer to exclude the 4-byte header
-		v.writingBuffers[i] = v.backingBuffers[i][headerSize : headerSize+uintptr(vmPktDesc.vm_pkt_size)]
+		v.writingBuffers[i] = v.packetBufferAt(i, 0)
 	}
 	return v.writingBuffers[:packetCount], nil
 }
@@ -180,8 +181,12 @@ func (v *pktDescsManager) writePacketsToPacketConn(conn net.PacketConn, packetCo
 		for sentCount < packetCount {
 			// send packet from buffer
 			if err := syscall.Sendmsg(int(fd), buffers[sentCount], nil, nil, 0); err != nil {
-				if errors.Is(err, syscall.EAGAIN) || errors.Is(err, syscall.ENOBUFS) {
+				if errors.Is(err, syscall.EAGAIN) {
 					return false // try again later
+				} else if errors.Is(err, syscall.ENOBUFS) {
+					// Wait and try to send next packet
+					time.Sleep(100 * time.Microsecond)
+					continue
 				}
 				sendErr = fmt.Errorf("syscall.Sendmsg failed: %w", err)
 				return true

vmnet/msg_x_darwin.go

@@ -14,6 +14,7 @@ import (
 	"os"
 	"runtime"
 	"syscall"
+	"time"
 	"unsafe"
 )
 
@@ -191,8 +192,12 @@ func (m *msgHdrXArray) writePacketsToPacketConn(conn net.PacketConn, packetCount
 			// send packet from msgHdrX array
 			n, err := C.sendmsg_x(C.int(fd), (*C.struct_msghdr_x)(m.at(sentCount)), C.u_int(packetCount-sentCount), 0)
 			if n < 0 {
-				if errors.Is(err, syscall.EAGAIN) || errors.Is(err, syscall.ENOBUFS) {
+				if errors.Is(err, syscall.EAGAIN) {
 					return false // try again later
+				} else if errors.Is(err, syscall.ENOBUFS) {
+					// Wait and try to send next packet
+					time.Sleep(100 * time.Microsecond)
+					continue
 				}
 				sendErr = fmt.Errorf("sendmsg_x failed: %w", err)
 				return true

vmnet/pktdesc_darwin.go

@@ -14,7 +14,7 @@ import (
 	"unsafe"
 )
 
-const headerSize = unsafe.Sizeof(C.uint32_t(0))
+const headerSize = int(unsafe.Sizeof(C.uint32_t(0)))
 const virtioNetHdrSize = 12 // Size of virtio_net_hdr_v1
 
 // VMPktDesc is a Go representation of C.struct_vmpktdesc.
@@ -26,6 +26,11 @@ func (v *VMPktDesc) SetPacketSize(size int) {
 	v.vm_pkt_iov.iov_len = C.size_t(size)
 }
 
+// GetPacketSize gets the packet size from VMPktDesc.
+func (v *VMPktDesc) GetPacketSize() int {
+	return int(v.vm_pkt_size)
+}
+
 // pktDescsManager manages pktDescs and their backing buffers.
 type pktDescsManager struct {
 	pktDescs       *VMPktDesc
@@ -65,6 +70,11 @@ func (v *pktDescsManager) at(index int) *VMPktDesc {
 	return (*VMPktDesc)(unsafe.Add(unsafe.Pointer(v.pktDescs), index*int(unsafe.Sizeof(VMPktDesc{}))))
 }
 
+// headerBufferAt returns the 4-byte header buffer at the given index.
+func (v *pktDescsManager) headerBufferAt(index int) []byte {
+	return v.backingBuffers[index][:headerSize]
+}
+
 // iter iterates over pktDescs and their corresponding buffers.
 func (v *pktDescsManager) iter(packetCount int) iter.Seq2[int, *VMPktDesc] {
 	return func(yield func(int, *VMPktDesc) bool) {
@@ -76,6 +86,12 @@ func (v *pktDescsManager) iter(packetCount int) iter.Seq2[int, *VMPktDesc] {
 	}
 }
 
+// packetBufferAt returns the packet buffer at the given index and offset.
+func (v *pktDescsManager) packetBufferAt(index, offset int) []byte {
+	end := headerSize + v.at(index).GetPacketSize()
+	return v.backingBuffers[index][headerSize+offset : end]
+}
+
 // reset resets pktDescs to initial state.
 func (v *pktDescsManager) reset() {
 	C.resetVMPktDescArray((*C.struct_vmpktdesc)(v.pktDescs), C.int(v.maxPacketCount), C.uint64_t(v.maxPacketSize))

vmnet/stream_darwin.go

@@ -139,9 +139,9 @@ func (v *pktDescsManager) buffersForWritingToConn(packetCount int) (net.Buffers,
 			return nil, fmt.Errorf("vm_pkt_size %d exceeds maxPacketSize %d", vmPktDesc.vm_pkt_size, v.maxPacketSize)
 		}
 		// Write packet size to the 4-byte header
-		binary.BigEndian.PutUint32(v.backingBuffers[i][:headerSize], uint32(vmPktDesc.vm_pkt_size))
+		binary.BigEndian.PutUint32(v.headerBufferAt(i), uint32(vmPktDesc.vm_pkt_size))
 		// Resize buffer to include header and packet size
-		v.writingBuffers[i] = v.backingBuffers[i][:headerSize+uintptr(vmPktDesc.vm_pkt_size)]
+		v.writingBuffers[i] = v.backingBuffers[i][:headerSize+vmPktDesc.GetPacketSize()]
 	}
 	return v.writingBuffers[:packetCount], nil
 }
@@ -165,6 +165,28 @@ func (v *pktDescsManager) writePacketsToConn(conn net.Conn, packetCount int) err
 	return nil
 }
 
+// buffersForReadingFromConn returns [net.Buffers] to read from the [net.Conn]
+// for the given index and offset.
+// It prepares buffer for the next header read as well if possible.
+func (v *pktDescsManager) buffersForReadingFromConn(index, offset int) net.Buffers {
+	if offset < v.at(index).GetPacketSize() {
+		if index+1 < v.maxPacketCount {
+			// prepare next header read as well
+			return net.Buffers{
+				v.packetBufferAt(index, offset),
+				v.headerBufferAt(index + 1),
+			}
+		}
+		return net.Buffers{
+			v.packetBufferAt(index, offset),
+		}
+	}
+	headerOffset := offset - v.at(index).GetPacketSize()
+	return net.Buffers{
+		v.headerBufferAt(index)[headerOffset:],
+	}
+}
+
 // readPacketsFromConn reads packets from the [net.Conn] into [VMPktDesc]s.
 //   - It returns the number of packets read.
 //   - The packets are expected to come one by one with 4-byte big-endian header indicating the packet size.
@@ -173,63 +195,51 @@ func (v *pktDescsManager) writePacketsToConn(conn net.Conn, packetCount int) err
 func (v *pktDescsManager) readPacketsFromConn(conn net.Conn) (int, error) {
 	var packetCount int
 	// Wait until 4-byte header is read
-	if _, err := conn.Read(v.backingBuffers[packetCount][:headerSize]); err != nil {
+	if _, err := conn.Read(v.headerBufferAt(packetCount)); err != nil {
 		return 0, fmt.Errorf("conn.Read failed: %w", err)
 	}
 	// Get rawConn for Readv
 	rawConn, _ := conn.(syscall.Conn).SyscallConn()
 	// Read available packets
-	var packetLen uint32
-	var bufs net.Buffers
 	for {
-		packetLen = binary.BigEndian.Uint32(v.backingBuffers[packetCount][:headerSize])
+		packetLen := int(binary.BigEndian.Uint32(v.headerBufferAt(packetCount)))
 		if packetLen == 0 || uint64(packetLen) > v.maxPacketSize {
 			return 0, fmt.Errorf("invalid packetLen: %d (max %d)", packetLen, v.maxPacketSize)
 		}
+		v.at(packetCount).SetPacketSize(packetLen)
 
-		// prepare buffers for reading packet and next header if any
-		if packetCount+1 < v.maxPacketCount {
-			// prepare next header read as well
-			bufs = net.Buffers{
-				v.backingBuffers[packetCount][headerSize : headerSize+uintptr(packetLen)],
-				v.backingBuffers[packetCount+1][:headerSize],
-			}
-		} else {
-			// prepare only packet read to avoid exceeding maxPacketCount
-			bufs = net.Buffers{
-				v.backingBuffers[packetCount][headerSize : headerSize+uintptr(packetLen)],
-			}
-		}
-
-		// Read packet from the connection
+		// Read packet from the connection until full packet is read, including next header if possible.
 		var bytesHasBeenRead int
-		var err error
-		rawConnReadErr := rawConn.Read(func(fd uintptr) (done bool) {
+		var readErr error
+		if rawConnReadErr := rawConn.Read(func(fd uintptr) (done bool) {
+			bufs := v.buffersForReadingFromConn(packetCount, bytesHasBeenRead)
 			// read packet into buffers
-			bytesHasBeenRead, err = unix.Readv(int(fd), bufs)
-			if bytesHasBeenRead <= 0 {
+			n, err := unix.Readv(int(fd), bufs)
+			if n <= 0 {
 				if errors.Is(err, syscall.EAGAIN) {
 					return false // try again later
 				}
-				err = fmt.Errorf("unix.Readv failed: %w", err)
+				readErr = fmt.Errorf("unix.Readv failed: %w", err)
+				return true
+			}
+			bytesHasBeenRead += n
+			if bytesHasBeenRead == packetLen+headerSize || bytesHasBeenRead == packetLen {
 				return true
 			}
-			// assumes partial read of a packet does not happen since packet len is already known
-			return true
-		})
-		if rawConnReadErr != nil {
+			// Partial read, read again
+			return false
+		}); rawConnReadErr != nil {
 			return 0, fmt.Errorf("rawConn.Read failed: %w", rawConnReadErr)
 		}
-		if err != nil {
-			return 0, fmt.Errorf("closure in rawConn.Read failed: %w", err)
+		if readErr != nil {
+			return 0, fmt.Errorf("closure in rawConn.Read failed: %w", readErr)
 		}
-		v.at(packetCount).SetPacketSize(int(packetLen))
 		packetCount++
-		if bytesHasBeenRead == int(packetLen) {
+		if bytesHasBeenRead == packetLen+headerSize {
+			// next packet header is also read, continue to read next packet
+		} else if bytesHasBeenRead == packetLen {
 			// next packet seems not available now, or reached maxPacketCount
 			break
-		} else if bytesHasBeenRead != int(packetLen)+int(headerSize) {
-			return 0, fmt.Errorf("unexpected bytesHasBeenRead: %d (expected %d or %d)", bytesHasBeenRead, packetLen, packetLen+uint32(headerSize))
 		}
 	}
 	return packetCount, nil