virtio: isolate Tx as dedicated thread & resolve dependency using int…

…erface

Signed-off-by: Nobuhiro MIKI <nob@bobuhiro11.net>

machine/machine.go

@@ -162,15 +162,12 @@ func New(nCpus int) (*Machine, error) {
 		}
 	}
 
-	virtioTxCallback := func(packet []byte) {
-		if _, err := t.Write(packet); err != nil {
-			panic(err)
-		}
-	}
+	v := virtio.NewNet(virtioIRQCallback, t, m.mem).(*virtio.Net)
+	go v.TxThreadEntry()
 
 	m.pci = pci.New(
 		pci.NewBridge(), // 00:00.0 for PCI bridge
-		virtio.NewNet(virtioIRQCallback, virtioTxCallback, m.mem), // 00:01.0 for Virtio PCI
+		v,               // 00:01.0 for Virtio PCI
 	)
 
 	return m, nil

virtio/net.go

@@ -5,12 +5,17 @@ import (
 	"encoding/binary"
 	"errors"
 	"fmt"
+	"io"
 	"unsafe"
 
 	"github.com/bobuhiro11/gokvm/pci"
 )
 
-var ErrIONotPermit = errors.New("IO is not permitted for virtio device")
+var (
+	ErrInvalidSel  = errors.New("queue sel is invalid")
+	ErrIONotPermit = errors.New("IO is not permitted for virtio device")
+	ErrNoTxPacket  = errors.New("no packet for tx")
+)
 
 const (
 	IOPortStart = 0x6200
@@ -38,11 +43,12 @@ type Net struct {
 	Mem          []byte
 	LastAvailIdx [2]uint16
 
+	tap io.ReadWriter
+
+	txKick chan interface{}
+
 	// This callback is called when virtio request IRQ.
 	irqCallback func(irq, level uint32)
-
-	// This callback is called when virtio transmit packet.
-	txCallBack func(packet []byte)
 }
 
 func (h Hdr) Bytes() ([]byte, error) {
@@ -63,7 +69,7 @@ type commonHeader struct {
 	queueSEL uint16
 	_        uint16 // queueNotify
 	_        uint8  // status
-	_        uint8  // isr
+	isr      uint8
 }
 
 type netHeader struct {
@@ -108,11 +114,26 @@ func (v Net) IOInHandler(port uint64, bytes []byte) error {
 	return nil
 }
 
-func (v *Net) QueueNotifyHandler() {
+func (v *Net) TxThreadEntry() {
+	for range v.txKick {
+		for v.Tx() == nil {
+		}
+	}
+}
+
+func (v *Net) Tx() error {
 	sel := v.Hdr.commonHeader.queueSEL
+	if sel == 0 {
+		return ErrInvalidSel
+	}
+
 	availRing := &v.VirtQueue[sel].AvailRing
 	usedRing := &v.VirtQueue[sel].UsedRing
 
+	if v.LastAvailIdx[sel] == availRing.Idx {
+		return ErrNoTxPacket
+	}
+
 	for v.LastAvailIdx[sel] != availRing.Idx {
 		buf := []byte{}
 		descID := availRing.Ring[v.LastAvailIdx[sel]%QueueSize]
@@ -142,12 +163,15 @@ func (v *Net) QueueNotifyHandler() {
 		// refs https://github.com/torvalds/linux/blob/38f80f42/include/uapi/linux/virtio_net.h#L178-L191
 		buf = buf[10:]
 
-		v.txCallBack(buf)
+		if _, err := v.tap.Write(buf); err != nil {
+			return err
+		}
 		usedRing.Idx++
 		v.LastAvailIdx[sel]++
 	}
-
 	v.InjectIRQ()
+
+	return nil
 }
 
 func (v *Net) IOOutHandler(port uint64, bytes []byte) error {
@@ -161,7 +185,8 @@ func (v *Net) IOOutHandler(port uint64, bytes []byte) error {
 	case 14:
 		v.Hdr.commonHeader.queueSEL = uint16(pci.BytesToNum(bytes))
 	case 16:
-		v.QueueNotifyHandler()
+		v.Hdr.commonHeader.isr = 0x0
+		v.txKick <- true
 	case 19:
 		fmt.Printf("ISR was written!\r\n")
 	default:
@@ -174,15 +199,17 @@ func (v Net) GetIORange() (start, end uint64) {
 	return IOPortStart, IOPortStart + IOPortSize
 }
 
-func NewNet(irqCallBack func(irq, level uint32), txCallBack func(packet []byte), mem []byte) pci.Device {
+func NewNet(irqCallBack func(irq, level uint32), tap io.ReadWriter, mem []byte) pci.Device {
 	res := &Net{
 		Hdr: Hdr{
 			commonHeader: commonHeader{
 				queueNUM: QueueSize,
+				isr:      0x0,
 			},
 		},
 		irqCallback:  irqCallBack,
-		txCallBack:   txCallBack,
+		txKick:       make(chan interface{}),
+		tap:          tap,
 		Mem:          mem,
 		VirtQueue:    [2]*VirtQueue{},
 		LastAvailIdx: [2]uint16{0, 0},

virtio/net_test.go

@@ -11,7 +11,7 @@ import (
 func TestGetDeviceHeader(t *testing.T) {
 	t.Parallel()
 
-	v := virtio.NewNet(func(_, _ uint32) {}, func(_ []byte) {}, []byte{})
+	v := virtio.NewNet(func(_, _ uint32) {}, bytes.NewBuffer([]byte{}), []byte{})
 	expected := uint16(0x1000)
 	actual := v.GetDeviceHeader().DeviceID
 
@@ -24,7 +24,7 @@ func TestGetIORange(t *testing.T) {
 	t.Parallel()
 
 	expected := uint64(virtio.IOPortSize)
-	s, e := virtio.NewNet(func(_, _ uint32) {}, func(_ []byte) {}, []byte{}).GetIORange()
+	s, e := virtio.NewNet(func(_, _ uint32) {}, bytes.NewBuffer([]byte{}), []byte{}).GetIORange()
 	actual := e - s
 
 	if actual != expected {
@@ -36,7 +36,7 @@ func TestIOInHandler(t *testing.T) {
 	t.Parallel()
 
 	expected := []byte{0x20, 0x00}
-	v := virtio.NewNet(func(_, _ uint32) {}, func(_ []byte) {}, []byte{})
+	v := virtio.NewNet(func(_, _ uint32) {}, bytes.NewBuffer([]byte{}), []byte{})
 	actual := make([]byte, 2)
 	_ = v.IOInHandler(virtio.IOPortStart+12, actual)
 
@@ -49,7 +49,7 @@ func TestSetQueuePhysAddr(t *testing.T) {
 	t.Parallel()
 
 	mem := make([]byte, 0x1000000)
-	v := virtio.NewNet(func(_, _ uint32) {}, func(_ []byte) {}, mem)
+	v := virtio.NewNet(func(_, _ uint32) {}, bytes.NewBuffer([]byte{}), mem)
 	base := uint32(uintptr(unsafe.Pointer(&(v.(*virtio.Net).Mem[0]))))
 
 	expected := [2]uint32{
@@ -84,13 +84,10 @@ func TestQueueNotifyHandler(t *testing.T) {
 	}
 
 	expected := []byte{0xaa, 0xbb, 0xcc, 0xdd}
-	actual := []byte{}
-	txCallback := func(bytes []byte) {
-		actual = append(actual, bytes...)
-	}
+	b := bytes.NewBuffer([]byte{})
 
 	mem := make([]byte, 0x1000000)
-	v := virtio.NewNet(irqCallback, txCallback, mem).(*virtio.Net)
+	v := virtio.NewNet(irqCallback, b, mem).(*virtio.Net)
 
 	// Size of struct virtio_net_hdr
 	const K = 10
@@ -116,13 +113,15 @@ func TestQueueNotifyHandler(t *testing.T) {
 	vq.AvailRing.Idx = 1
 	v.VirtQueue[sel] = &vq
 
-	v.QueueNotifyHandler()
+	if err := v.Tx(); err != nil {
+		t.Fatalf("err: %v\n", err)
+	}
 
 	if !irqInjected {
 		t.Fatalf("irqInjected = false\n")
 	}
 
-	if !bytes.Equal(expected, actual) {
-		t.Fatalf("expected: %v, actual: %v", expected, actual)
+	if !bytes.Equal(expected, b.Bytes()) {
+		t.Fatalf("expected: %v, actual: %v", expected, b.Bytes())
 	}
 }