feat: map virtio-blk virtqueues across an IOThread pool (#963)

* feat: map virtio-blk virtqueues across an IOThread pool

Boot volumes ran every virtqueue through a single IOThread. Size a pool
from the vCPU count and hand QEMU an iothread-vq-mapping so the queues
spread across it. Falls back to the scalar iothread= form for MMIO
machines, single-queue devices and a pool of one.

SPINIFEX_BLK_IOTHREADS is experiment scaffolding for sweeping the pool
size without a rebuild, and comes out before this merges.

* feat: name QEMU guests with debug-threads=on

Config.Name held the instance ID but was never emitted, so QEMU left
every thread called qemu-system-x86 and a host-side pidstat -t could
not tell an IOThread from a vCPU thread. Emit -name guest=<id> with
debug-threads=on, which is what libvirt does for the same reason.

* refactor: drop the IOThread pool-size experiment override

SPINIFEX_BLK_IOTHREADS existed to sweep the pool size on dev-prod
without a rebuild and redeploy per value. The sweep is done, so the
derived size is the only path and buildDrives keeps its own doc comment.

spinifex/vm/devices.go

@@ -1,6 +1,7 @@
 package vm
 
 import (
+	"encoding/json"
 	"fmt"
 	"strconv"
 	"strings"
@@ -93,6 +94,109 @@ func NetDevice(machineType, netdev, mac string, queues, mtu int) Device {
 	return Device{Value: b.String()}
 }
 
+// Blk IOThread pool sizing. One IOThread per virtio-blk device services every
+// virtqueue on it, so a guest with several queues still funnels through a
+// single host thread. QEMU 9.0+ can map queues across a pool instead.
+//
+// The pool is deliberately much smaller than the vCPU count. Red Hat's RHEL 9.4
+// measurements used 4 IOThreads against 192 vCPUs and 96 queues, and OpenShift
+// Virtualization documents 4 as the starting point with 8-16 reserved for very
+// fast local storage — which an NBD export backed by object storage is not.
+const (
+	// BlkIOThreadsPerVCPU divides the vCPU count. The threshold lands at 4
+	// vCPUs, matching the guest size below which neither source expects the
+	// feature to do anything.
+	BlkIOThreadsPerVCPU = 2
+
+	// MaxBlkIOThreads caps the pool. These are real host threads charged to
+	// the node, and a per-device pool multiplies by the volume count.
+	MaxBlkIOThreads = 4
+)
+
+// BlkIOThreadPoolSize returns how many IOThreads to create for one virtio-blk
+// device on a guest with vcpus processors: vcpus/2, clamped to 1..4. A result
+// of 1 reproduces the single-IOThread command line exactly.
+func BlkIOThreadPoolSize(vcpus int) int {
+	n := vcpus / BlkIOThreadsPerVCPU
+	if n < 1 {
+		return 1
+	}
+	if n > MaxBlkIOThreads {
+		return MaxBlkIOThreads
+	}
+	return n
+}
+
+// BlkIOThreadID names the i'th IOThread in a device's pool. The single-thread
+// case keeps the original unsuffixed name so an existing guest's command line
+// is unchanged.
+func BlkIOThreadID(base string, i, poolSize int) string {
+	if poolSize <= 1 {
+		return base
+	}
+	return fmt.Sprintf("%s-%d", base, i)
+}
+
+// iothreadVQMapping is one entry of virtio-blk's iothread-vq-mapping list.
+// Marshalled rather than concatenated: the property is a QAPI struct list and
+// the value has to be valid JSON, so hand-built strings are a silent-corruption
+// risk for no gain.
+type iothreadVQMapping struct {
+	IOThread string `json:"iothread"`
+	VQs      []int  `json:"vqs"`
+}
+
+// blkDeviceJSON is the -device value for a virtio-blk device carrying an
+// iothread-vq-mapping. QEMU accepts a JSON object as one argv value, which is
+// the only command-line form that can express a nested list.
+type blkDeviceJSON struct {
+	Driver    string              `json:"driver"`
+	Drive     string              `json:"drive"`
+	NumQueues int                 `json:"num-queues,omitempty"`
+	BootIndex int                 `json:"bootindex,omitempty"`
+	Mapping   []iothreadVQMapping `json:"iothread-vq-mapping"`
+}
+
+// BlkDeviceMapped returns a virtio-blk -device that spreads its virtqueues
+// across a pool of poolSize IOThreads, round-robin. Falls back to BlkDevice's
+// scalar iothread= form when poolSize <= 1, so the common small-guest case is
+// byte-for-byte what it was before this existed.
+//
+// Every queue is mapped exactly once and the scalar iothread= property is
+// absent: QEMU treats the two as alternative assignments and rejects both.
+func BlkDeviceMapped(machineType, drive, iothreadBase string, queues, bootIdx, poolSize int) (Device, error) {
+	if poolSize <= 1 || queues <= 1 || IsMMIO(machineType) {
+		return BlkDevice(machineType, drive, iothreadBase, queues, bootIdx), nil
+	}
+	if poolSize > queues {
+		poolSize = queues
+	}
+
+	vqs := make([][]int, poolSize)
+	for q := range queues {
+		vqs[q%poolSize] = append(vqs[q%poolSize], q)
+	}
+	mapping := make([]iothreadVQMapping, 0, poolSize)
+	for i := range vqs {
+		mapping = append(mapping, iothreadVQMapping{
+			IOThread: BlkIOThreadID(iothreadBase, i, poolSize),
+			VQs:      vqs[i],
+		})
+	}
+
+	b, err := json.Marshal(blkDeviceJSON{
+		Driver:    "virtio-blk-pci",
+		Drive:     drive,
+		NumQueues: queues,
+		BootIndex: bootIdx,
+		Mapping:   mapping,
+	})
+	if err != nil {
+		return Device{}, fmt.Errorf("marshal virtio-blk iothread-vq-mapping: %w", err)
+	}
+	return Device{Value: string(b)}, nil
+}
+
 // BlkDevice returns the appropriate QEMU virtio-blk device string for
 // machineType. iothread and queues are omitted when empty/zero. bootIdx is
 // only emitted for PCI machines (MMIO has no bootindex concept).

spinifex/vm/devices_test.go

@@ -1,6 +1,8 @@
 package vm
 
 import (
+	"encoding/json"
+	"strings"
 	"testing"
 
 	"github.com/stretchr/testify/assert"
@@ -201,3 +203,101 @@ func TestNetDevice_MMIO_NoQueueSizes(t *testing.T) {
 	assert.NotContains(t, d.Value, "mq=on")
 	assert.Contains(t, d.Value, "host_mtu=1442")
 }
+
+func TestBlkIOThreadPoolSize(t *testing.T) {
+	// The threshold sits at 4 vCPUs and the ceiling at 4 threads, matching the
+	// only configurations Red Hat's two write-ups actually recommend.
+	for _, tc := range []struct{ vcpus, want int }{
+		{1, 1}, {2, 1}, {3, 1}, {4, 2}, {6, 3}, {8, 4}, {16, 4}, {192, 4},
+	} {
+		if got := BlkIOThreadPoolSize(tc.vcpus); got != tc.want {
+			t.Errorf("BlkIOThreadPoolSize(%d) = %d, want %d", tc.vcpus, got, tc.want)
+		}
+	}
+}
+
+func TestBlkIOThreadID(t *testing.T) {
+	// A pool of one keeps the unsuffixed name, so an existing guest's command
+	// line does not change when this code lands.
+	if got := BlkIOThreadID("ioth-os", 0, 1); got != "ioth-os" {
+		t.Errorf("pool of 1 = %q, want ioth-os", got)
+	}
+	if got := BlkIOThreadID("ioth-os", 1, 2); got != "ioth-os-1" {
+		t.Errorf("pool of 2 index 1 = %q, want ioth-os-1", got)
+	}
+}
+
+func TestBlkDeviceMapped_SingleThreadUnchanged(t *testing.T) {
+	// Byte-for-byte identical to the scalar form: this is the path every guest
+	// below 4 vCPUs takes, so a difference here is a silent behaviour change.
+	want := BlkDevice("q35", "os", "ioth-os", 2, 1)
+	for _, pool := range []int{0, 1} {
+		got, err := BlkDeviceMapped("q35", "os", "ioth-os", 2, 1, pool)
+		if err != nil {
+			t.Fatalf("pool %d: %v", pool, err)
+		}
+		if got.Value != want.Value {
+			t.Errorf("pool %d = %q, want %q", pool, got.Value, want.Value)
+		}
+	}
+}
+
+func TestBlkDeviceMapped_MMIOFallsBack(t *testing.T) {
+	// microvm has no PCI bus, so the JSON form names a driver it cannot use.
+	got, err := BlkDeviceMapped("microvm", "os", "ioth-os", 4, 1, 2)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if !strings.HasPrefix(got.Value, "virtio-blk-device") {
+		t.Errorf("MMIO = %q, want the virtio-blk-device form", got.Value)
+	}
+}
+
+func TestBlkDeviceMapped_EveryQueueMappedOnce(t *testing.T) {
+	for _, tc := range []struct{ queues, pool int }{{4, 2}, {4, 4}, {8, 4}, {2, 4}, {3, 2}} {
+		got, err := BlkDeviceMapped("q35", "os", "ioth-os", tc.queues, 1, tc.pool)
+		if err != nil {
+			t.Fatalf("queues=%d pool=%d: %v", tc.queues, tc.pool, err)
+		}
+		var dev struct {
+			Driver    string `json:"driver"`
+			NumQueues int    `json:"num-queues"`
+			IOThread  string `json:"iothread"`
+			Mapping   []struct {
+				IOThread string `json:"iothread"`
+				VQs      []int  `json:"vqs"`
+			} `json:"iothread-vq-mapping"`
+		}
+		if err := json.Unmarshal([]byte(got.Value), &dev); err != nil {
+			t.Fatalf("queues=%d pool=%d: not valid JSON: %v (%s)", tc.queues, tc.pool, err, got.Value)
+		}
+		// The scalar property and the mapping are alternative assignments;
+		// QEMU rejects a device carrying both.
+		if dev.IOThread != "" {
+			t.Errorf("queues=%d pool=%d: scalar iothread= present alongside the mapping", tc.queues, tc.pool)
+		}
+		if dev.NumQueues != tc.queues {
+			t.Errorf("queues=%d pool=%d: num-queues=%d", tc.queues, tc.pool, dev.NumQueues)
+		}
+		seen := map[int]int{}
+		for _, m := range dev.Mapping {
+			for _, q := range m.VQs {
+				seen[q]++
+			}
+		}
+		for q := 0; q < tc.queues; q++ {
+			if seen[q] != 1 {
+				t.Errorf("queues=%d pool=%d: vq %d mapped %d times, want exactly 1 (%s)",
+					tc.queues, tc.pool, q, seen[q], got.Value)
+			}
+		}
+		if len(seen) != tc.queues {
+			t.Errorf("queues=%d pool=%d: %d distinct queues mapped", tc.queues, tc.pool, len(seen))
+		}
+		// A pool larger than the queue count would declare IOThread objects
+		// that no mapping entry references, which QEMU rejects.
+		if want := min(tc.pool, tc.queues); len(dev.Mapping) != want {
+			t.Errorf("queues=%d pool=%d: %d mapping entries, want %d", tc.queues, tc.pool, len(dev.Mapping), want)
+		}
+	}
+}

spinifex/vm/lifecycle.go

@@ -1380,9 +1380,18 @@ func buildDrives(requests []types.EBSRequest, cpuCount int, machineType string)
 				ReconnectDelay: NBDReconnectDelaySeconds,
 			}
 
-			iothreadID := "ioth-os"
-			cfg.IOThreads = append(cfg.IOThreads, IOThread{ID: iothreadID})
-			cfg.Devices = append(cfg.Devices, BlkDevice(machineType, drive.ID, iothreadID, cpuCount, 1))
+			// The boot disk's virtqueues are spread across a pool rather than
+			// all served by one host thread. A pool of 1 emits exactly the
+			// command line this used to, which is every guest below 4 vCPUs.
+			pool := BlkIOThreadPoolSize(cpuCount)
+			for i := range pool {
+				cfg.IOThreads = append(cfg.IOThreads, IOThread{ID: BlkIOThreadID("ioth-os", i, pool)})
+			}
+			dev, err := BlkDeviceMapped(machineType, drive.ID, "ioth-os", cpuCount, 1, pool)
+			if err != nil {
+				return driveConfig{}, fmt.Errorf("boot volume %s: %w", v.Name, err)
+			}
+			cfg.Devices = append(cfg.Devices, dev)
 			cfg.Drives = append(cfg.Drives, drive)
 
 		case v.EFI:

spinifex/vm/lifecycle_test.go

@@ -122,7 +122,25 @@ func TestBuildDrives(t *testing.T) {
 		wantErr          string
 	}{
 		{
-			name: "boot volume",
+			// 2 vCPUs resolves to a pool of one, which must keep the scalar
+			// iothread= form: this is the shape every small guest boots with.
+			name: "boot volume, pool of one keeps the scalar iothread",
+			requests: []types.EBSRequest{
+				{Name: "vol-boot", NBDURI: "nbd:unix:/tmp/boot.sock", Boot: true},
+			},
+			cpuCount: 2,
+			wantDrives: []Drive{
+				{File: "nbd:unix:/tmp/boot.sock", Format: "raw", If: "none", Media: "disk", ID: "os", Cache: "none", Werror: "stop", Rerror: "stop", ReconnectDelay: NBDReconnectDelaySeconds},
+			},
+			wantIOThreads: []IOThread{{ID: "ioth-os"}},
+			wantDevices: []Device{
+				{Value: "virtio-blk-pci,drive=os,iothread=ioth-os,num-queues=2,bootindex=1"},
+			},
+		},
+		{
+			// 4 vCPUs resolves to a pool of two, so the four virtqueues are
+			// split across two host threads rather than served by one.
+			name: "boot volume, four vCPUs map their queues across a pool of two",
 			requests: []types.EBSRequest{
 				{Name: "vol-boot", NBDURI: "nbd:unix:/tmp/boot.sock", Boot: true},
 			},
@@ -130,9 +148,9 @@ func TestBuildDrives(t *testing.T) {
 			wantDrives: []Drive{
 				{File: "nbd:unix:/tmp/boot.sock", Format: "raw", If: "none", Media: "disk", ID: "os", Cache: "none", Werror: "stop", Rerror: "stop", ReconnectDelay: NBDReconnectDelaySeconds},
 			},
-			wantIOThreads: []IOThread{{ID: "ioth-os"}},
+			wantIOThreads: []IOThread{{ID: "ioth-os-0"}, {ID: "ioth-os-1"}},
 			wantDevices: []Device{
-				{Value: "virtio-blk-pci,drive=os,iothread=ioth-os,num-queues=4,bootindex=1"},
+				{Value: `{"driver":"virtio-blk-pci","drive":"os","num-queues":4,"bootindex":1,"iothread-vq-mapping":[{"iothread":"ioth-os-0","vqs":[0,2]},{"iothread":"ioth-os-1","vqs":[1,3]}]}`},
 			},
 		},
 		{
@@ -234,7 +252,10 @@ func TestBuildDrives(t *testing.T) {
 				{Name: "vol-efi", NBDURI: "nbd:unix:/tmp/efi.sock", EFI: true},
 				{Name: "vol-data-a", NBDURI: "nbd:unix:/tmp/data-a.sock", HotplugPort: 3},
 			},
-			cpuCount: 4,
+			// 2 vCPUs, so the boot disk keeps the legacy scalar shape and this
+			// case still asserts what its name says it does. Data volumes are
+			// unaffected by the pool either way — they keep one IOThread each.
+			cpuCount: 2,
 			wantDrives: []Drive{
 				{File: "nbd:unix:/tmp/boot.sock", Format: "raw", If: "none", Media: "disk", ID: "os", Cache: "none", Werror: "stop", Rerror: "stop", ReconnectDelay: NBDReconnectDelaySeconds},
 				{File: "nbd:unix:/tmp/efi.sock", Format: "raw", If: "pflash", Unit: 1},
@@ -247,7 +268,7 @@ func TestBuildDrives(t *testing.T) {
 				{Value: "driver=nbd,node-name=nbd-vol-data-a,server.type=unix,server.path=/tmp/data-a.sock,export=,reconnect-delay=30"},
 			},
 			wantDevices: []Device{
-				{Value: "virtio-blk-pci,drive=os,iothread=ioth-os,num-queues=4,bootindex=1"},
+				{Value: "virtio-blk-pci,drive=os,iothread=ioth-os,num-queues=2,bootindex=1"},
 				{Value: "virtio-blk-pci,id=vdisk-vol-data-a,drive=nbd-vol-data-a,iothread=ioth-vol-data-a,serial=voldataa,bus=hotplug-ebs3,werror=stop,rerror=stop"},
 			},
 			wantHotplugPorts: []int{0, 0, 3},

spinifex/vm/vm.go

@@ -383,6 +383,13 @@ type FwCfgEntry struct {
 func (cfg *Config) Execute() (*exec.Cmd, error) {
 	args := []string{}
 
+	// debug-threads=on makes QEMU name its threads after what they do, so a
+	// host-side `pidstat -t` can tell an IOThread from a vCPU thread. Without it
+	// every row reads qemu-system-x86 and per-thread profiling is guesswork.
+	if cfg.Name != "" {
+		args = append(args, "-name", fmt.Sprintf("guest=%s,debug-threads=on", cfg.Name))
+	}
+
 	if cfg.PIDFile != "" {
 		args = append(args, "-pidfile", cfg.PIDFile)
 	}

spinifex/vm/vm_test.go

@@ -57,6 +57,7 @@ func TestExecute(t *testing.T) {
 	assert.NotNil(t, cmd)
 
 	expectedArgs := []string{
+		"-name", "guest=test-vm,debug-threads=on",
 		"-smp", "2",
 		"-m", "1024",
 		"-drive", "file=disk.img,format=qcow2",
@@ -644,3 +645,32 @@ func TestExecute_FullConfig(t *testing.T) {
 	assert.Equal(t, "iothread,id=io0", argValue(args, "-object"))
 	assert.Equal(t, "user,id=net0", argValue(args, "-netdev"))
 }
+
+// debug-threads=on is what makes host-side per-thread profiling readable: QEMU
+// names each thread after its role instead of leaving them all qemu-system-x86.
+func TestExecute_NameCarriesDebugThreads(t *testing.T) {
+	cfg := Config{
+		Name:         "i-0123456789abcdef0",
+		CPUCount:     2,
+		Memory:       512,
+		Architecture: "x86_64",
+		Drives:       []Drive{{File: "disk.img", Format: "raw"}},
+	}
+
+	cmd, err := cfg.Execute()
+	assert.NoError(t, err)
+	assert.Equal(t, "guest=i-0123456789abcdef0,debug-threads=on", argValue(cmd.Args[1:], "-name"))
+}
+
+func TestExecute_NoNameEmitsNoNameArg(t *testing.T) {
+	cfg := Config{
+		CPUCount:     2,
+		Memory:       512,
+		Architecture: "x86_64",
+		Drives:       []Drive{{File: "disk.img", Format: "raw"}},
+	}
+
+	cmd, err := cfg.Execute()
+	assert.NoError(t, err)
+	assert.False(t, argExists(cmd.Args[1:], "-name"))
+}