src/driver/virtio.c

Walkthrough, interview notes & deep dive

Implementation Overview

The driver/virtio.c file implements the ixy userspace driver for Virtio-net devices, specifically focusing on the Legacy PCI interface. Like the ixgbe implementation, it abstracts the hardware-specific details behind the ixy_device vtable. This allows higher-level applications to use the same rx_batch and tx_batch logic regardless of whether they are running on a physical 10GbE NIC or a virtualized environment.

During virtio_init, the driver allocates an ixy_device and populates it with function pointers. It performs a PCI resource check to ensure the device is a legacy Virtio network card (0x1000) and maps the BAR0 I/O space for register access.

dev->ixy.rx_batch = virtio_rx_batch;
dev->ixy.tx_batch = virtio_tx_batch;
dev->ixy.read_stats = virtio_read_stats;
dev->ixy.set_promisc = virtio_set_promisc;
enable_dma(pci_addr);
dev->fd = pci_open_resource(pci_addr, "resource0", O_RDWR);
virtio_legacy_init(dev);

Legacy PCI Interface and Feature Negotiation

The driver follows the standard Virtio-net initialization sequence. It resets the device by writing to VIRTIO_PCI_STATUS, acknowledges the device, and then negotiates features. In this implementation, the driver requires features like VIRTIO_NET_F_CSUM, VIRTIO_NET_F_CTRL_VQ, and VIRTIO_F_ANY_LAYOUT. Once the features are set in VIRTIO_PCI_GUEST_FEATURES, the driver proceeds to set up the Virtqueues.

Register access in legacy Virtio is performed via I/O port offsets from BAR0. For example, selecting a queue is done by writing the queue index to VIRTIO_PCI_QUEUE_SEL, and then reading VIRTIO_PCI_QUEUE_NUM to determine the ring size. The physical address of the queue's memory is then written to VIRTIO_PCI_QUEUE_PFN (Page Frame Number), shifted appropriately.

The Virtqueue Mechanism

The core of Virtio data movement is the virtqueue, implemented here as a vring. It consists of three distinct memory areas shared between the guest (driver) and the host (hypervisor):

• Descriptor Table: An array of struct vring_desc containing the physical addresses, lengths, and flags for data buffers.
• Available Ring: A ring of indices into the descriptor table that the driver has filled and is offering to the device.
• Used Ring: A ring of indices into the descriptor table that the device has processed and is returning to the driver.

struct virtqueue {
	struct vring vring;
	uint32_t notification_offset;
	uint16_t vq_used_last_idx;
	struct mempool* mempool;
	void* virtual_addresses[];
};

Memory for these structures is allocated using memory_allocate_dma to ensure it is physically contiguous and suitable for DMA. The virtio_legacy_vring_init helper calculates the offsets for the desc, avail, and used pointers based on the queue size and a 4096-byte alignment requirement.

Data Path: RX and TX Batching

The virtio_rx_batch function reaps used buffers by comparing the local vq_used_last_idx with the vring.used->idx updated by the host. For each packet, it extracts the pkt_buf from the virtual_addresses array, calculates the size (subtracting the virtio_net_hdr), and updates statistics. Crucially, it then refills the descriptor table with new buffers from the mempool and updates the available ring to keep the device supplied with RX descriptors.

In virtio_tx_batch, the driver first reclaims descriptors that the host has marked as "used." It then populates new descriptors for the packets provided in the batch. Unlike hardware NICs that might use a tail pointer register, Virtio requires the driver to write the descriptor index into the available ring and increment vring.avail->idx.

vq->vring.desc[idx].len = buf->size + sizeof(net_hdr);
vq->vring.desc[idx].addr = buf->buf_addr_phy + offsetof(struct pkt_buf, head_room) + sizeof(buf->head_room) - sizeof(net_hdr);
vq->vring.desc[idx].flags = 0;
vq->vring.desc[idx].next = 0;
vq->vring.avail->ring[(vq->vring.avail->idx + buf_idx) % vq->vring.num] = idx;

Memory Ordering and Notification

Because the driver and the hypervisor act as two independent processors sharing memory, memory barriers are vital. The driver uses _mm_mfence() to ensure that descriptor updates are visible in RAM before the available ring index is updated, and again before "kicking" the device. The "kick" is performed by virtio_legacy_notify_queue, which writes the queue index to the VIRTIO_PCI_QUEUE_NOTIFY register, alerting the hypervisor to check the rings.

Interview angles

Q: How does Virtio's "Split Ring" design (avail/used rings) differ from a traditional NIC's circular ring buffer? A: Traditional NICs like the ixgbe use a single ring where the head/tail pointers are managed via registers. Virtio decouples the actual buffer descriptors from the notification mechanism. The available/used rings only store indices, which allows for out-of-order completion of descriptors and avoids constant register writes for every packet.

Q: Why are memory barriers (`_mm_mfence`) used before updating `avail->idx` and after the `notify` write? A: The first barrier ensures the hypervisor doesn't see an updated avail->idx before the actual data in the descriptor table is written to memory. The second barrier ensures the notification write to the I/O port is synchronized with the memory state, preventing race conditions where the hypervisor wakes up but sees stale ring data.

Q: What is the purpose of the `virtio_net_hdr` prepended to every packet in this driver? A: The Virtio-net specification requires a header (defined here as net_hdr) that contains metadata such as checksum offload flags and GSO (Generic Segmentation Offload) information. Even when offloads are disabled, a zeroed header must be present for the hypervisor to correctly parse the descriptor.

Would you like to see the analysis for the next chapter?

Going deeper

vq->vring.desc[idx].len = 2;
vq->vring.desc[idx].flags = VRING_DESC_F_NEXT;
vq->vring.desc[idx].next = idx + 1;
vq->vring.desc[idx + 1].len = cmd_len - 2 - 1;
vq->vring.desc[idx + 1].flags = VRING_DESC_F_NEXT;
vq->vring.desc[idx + 1].next = idx + 2;
vq->vring.desc[idx + 2].len = 1;
vq->vring.desc[idx + 2].flags = VRING_DESC_F_WRITE;

Chained control descriptors The virtio_legacy_send_command function utilizes a three-element descriptor chain to satisfy the virtio-net control-plane requirements. It splits a single pkt_buf into a 2-byte header, a variable-length payload, and a 1-byte status tail. Note the rigid idx + 1 and idx + 2 indexing; this implementation assumes three contiguous descriptors are available and does not handle ring wrap-around, which is safe only if the control queue is never concurrently accessed and has sufficient headroom. The cmd_len - 2 - 1 math accounts for the fixed-size header and the final device-writable acknowledgment byte.

for (idx = 0; idx < vq->vring.num; ++idx) {
	struct vring_desc* desc = &vq->vring.desc[idx];
	if (desc->addr == 0) {
		break;
	}
}

Linear descriptor scanning Unlike high-performance drivers that maintain a stack or bitmap of free descriptor indices, this driver performs an O(n) linear scan to find a free slot by checking for a null addr. This sentinel approach is efficient for small rings but introduces cache misses and latency as the ring size increases. It also relies on the driver explicitly zeroing addr during the reaping phase of virtio_rx_batch or virtio_tx_batch to signal availability back to the allocation logic.

_mm_mfence();
vq->vring.avail->idx += buf_idx;
_mm_mfence();
virtio_legacy_notify_queue(dev, 1);

Memory ordering in TX The transmission path demonstrates the cost-correctness tradeoff of batched updates. It populates the avail->ring entries for all buffers in the batch before a single _mm_mfence() ensures that these writes are visible to the device. Only then is the avail->idx updated and another fence issued before the VIRTIO_PCI_QUEUE_NOTIFY PIO write (the "kick"). This avoids the overhead of per-packet notifications, contrasting with the RX path which triggers a notification inside the refill loop, potentially causing excessive VM exits in high-throughput scenarios.

Harder interview questions

• Q: What happens if the virtio_rx_batch refill loop fails to find enough free descriptors? A: The function will return the packets it successfully reaped, but the subsequent refill logic will attempt to allocate new pkt_buf objects from the mempool. If the mempool is exhausted, it calls error(), which is a fatal teardown. A more robust implementation would silently stop refilling and try again on the next poll.

• Q: Why does the RX path use an exact check desc->flags != VRING_DESC_F_WRITE instead of a bitmask? A: The driver is designed for simplicity and only supports "flat" buffers. By checking for exact equality, it implicitly rejects VRING_DESC_F_NEXT (chained) or VRING_DESC_F_INDIRECT descriptors. If the device were to use these features, the driver's assumption that one vring_used_elem corresponds to exactly one pkt_buf would break, leading to memory corruption.

• Q: How does VRING_AVAIL_F_NO_INTERRUPT impact system performance? A: Setting this flag in avail->flags tells the host/hypervisor not to send PCI interrupts when the used ring is updated. This is ideal for ixy's busy-polling model, as it eliminates the overhead of interrupt handling and context switching, but it keeps a CPU core at 100% utilization even when there is no traffic.

• Q: The _mm_mfence() is used before and after updating avail->idx. Is this strictly necessary on x86? A: On x86, TSO (Total Store Ordering) ensures that stores are not reordered with other stores. However, _mm_mfence also acts as a compiler barrier, preventing the compiler from reordering the avail->idx update before the descriptor writes. Furthermore, it ensures visibility to the device's MMIO/PIO logic, which may not be guaranteed by simpler compiler barriers like asm volatile("" ::: "memory").

• Q: Why is the RX mempool sized at max_queue_size * 4? A: In userspace networking, applications often hold onto buffers for processing, lookups, or transmission on other ports. If the mempool were only the size of the ring, a slow application would quickly starve the RX path, leading to dropped packets. Over-provisioning the mempool provides a "cushion" for high-latency buffer processing.

Gotchas

• Control Queue Overrun: The virtio_legacy_send_command function finds one free slot but consumes three (idx, idx+1, idx+2). If idx is num - 1, the increments will point outside the allocated vring_desc array, causing a heap overflow.
• Legacy Limit: The driver hardcodes a check for device_id == 0x1000. It will fail to initialize on "Modern" Virtio 1.0+ devices (which use 0x1041 and a different capability-based PCI layout) or transitional devices operating in non-legacy mode.
• Synchronous Control Plane: The virtio_legacy_send_command uses a usleep(100000) busy-wait loop. While acceptable for initialization, this blocks the entire thread for 100ms per command, which would be disastrous if called during a runtime reconfiguration.
• Unaligned DMA: The vring is initialized with 4096 alignment. While standard for legacy virtio, the driver uses memory_allocate_dma without explicitly verifying that the underlying physical page is contiguous beyond the ring size, though ixy's mempool logic usually guarantees this.

From ixy to a production driver

For an interview, frame ixy as a teaching driver: it exposes minimal virtio-net mechanics without the compatibility, scaling, and recovery machinery a shipping driver needs. vendor/ixy/src/driver/virtio.c only accepts legacy PCI device 0x1000, negotiates through BAR0 ports such as VIRTIO_PCI_STATUS, VIRTIO_PCI_HOST_FEATURES, VIRTIO_PCI_GUEST_FEATURES, and VIRTIO_PCI_QUEUE_PFN, then builds split virtqueues from vring_desc, vring_avail, and vring_used.

Transport compatibility: Linux virtio_net.ko runs on the kernel virtio bus, so modern virtio 1.0+ devices use PCI capability structures for common config, notify, ISR, and device config rather than old QUEUE_PFN registers. Modern virtio-net IDs include 0x1041, and virtio 1.x defines split and packed rings. DPDK's virtio PMD covers PCI, vhost-user, and packed/split deployments. ixy can hardcode legacy split rings because the paper wants the structure visible; production code cannot strand users on one transport.

Queue scaling: ixy rejects more than one RX or TX queue and never negotiates VIRTIO_NET_F_MQ or VIRTIO_NET_F_RSS. Linux and DPDK use multiple queue pairs so cores can poll, interrupt, steer, and account independently. Single-queue code is readable, but shipping paths need RSS, CPU affinity, queue state, and per-queue stats to scale.

Interrupts and polling: ixy sets VRING_AVAIL_F_NO_INTERRUPT and busy-polls with _mm_mfence() around shared-ring updates. Linux uses interrupts, MSI-X, NAPI, and a poll budget so busy traffic is batched while idle traffic does not burn a core. DPDK PMDs poll like ixy, but inside a framework that owns lcores, queues, mempools, link handling, and device operations. The distinction is whether the driver controls the deployment model.

Feature negotiation and offloads: ixy sends a fixed zeroed virtio_net_hdr and does not implement checksum, GSO/TSO, mergeable RX buffers, or header variants. Real drivers negotiate VIRTIO_NET_F_CSUM, VIRTIO_NET_F_GUEST_CSUM, VIRTIO_NET_F_HOST_TSO4, VIRTIO_NET_F_HOST_TSO6, guest TSO/GSO, VIRTIO_NET_F_MRG_RXBUF, and virtio_net_hdr_mrg_rxbuf. Without them, throughput, MTU behavior, and checksum correctness can regress.

Control and robustness: ixy uses VIRTIO_NET_F_CTRL_VQ only for promiscuous mode, posts a 3-descriptor chain, kicks queue 2, then waits with usleep(100000). It finds descriptors by O(n) scan and has a real idx, idx+1, idx+2 wrap-around bug. Kernel and DPDK code maintain free descriptor state, complete commands asynchronously or with bounded waits, handle VIRTIO_NET_F_STATUS link changes, expose ethtool/statistics, coordinate locking and teardown, and avoid fatal error() on ordinary pressure. vhost-net and vhost-user add another production concern: cooperating with optimized host backends. ixy omits this for explainability; a production driver is judged by surviving hotplug, reset, queue exhaustion, feature mismatch, and concurrent dataplane/control-plane activity.

Sources

• Virtual I/O Device (VIRTIO) Version 1.2
• Linux virtio_net.c source
• DPDK Poll Mode Driver for Emulated Virtio NIC
• User Space Network Drivers

Source

#include <emmintrin.h>
#include <stdlib.h>
#include <string.h>
#include <sys/file.h>
#include <unistd.h>

#include "driver/device.h"
#include "log.h"
#include "memory.h"
#include "pci.h"
#include "virtio.h"
#include "virtio_type.h"

static const char* driver_name = "ixy-virtio";

static inline void virtio_legacy_notify_queue(struct virtio_device* dev, uint16_t idx) {
	write_io16(dev->fd, idx, VIRTIO_PCI_QUEUE_NOTIFY);
}

static uint8_t virtio_legacy_get_status(struct virtio_device* dev) {
	return read_io8(dev->fd, VIRTIO_PCI_STATUS);
}

static void virtio_legacy_check_status(struct virtio_device* dev) {
	if (read_io8(dev->fd, VIRTIO_PCI_STATUS) == VIRTIO_CONFIG_STATUS_FAILED) {
		error("Device signaled unrecoverable error");
	}
}

static inline size_t virtio_legacy_vring_size(unsigned int num, unsigned long align) {
	size_t size;

	size = num * sizeof(struct vring_desc);
	size += sizeof(struct vring_avail) + (num * sizeof(uint16_t));
	size = RTE_ALIGN_CEIL(size, align);
	size += sizeof(struct vring_used) + (num * sizeof(struct vring_used_elem));
	return size;
}

static inline void virtio_legacy_vring_init(struct vring* vr, unsigned int num, uint8_t* p, unsigned long align) {
	vr->num = num;
	vr->desc = (struct vring_desc*)p;
	vr->avail = (struct vring_avail*)(p + num * sizeof(struct vring_desc));
	vr->used = (void*)RTE_ALIGN_CEIL((uintptr_t)(&vr->avail->ring[num]), align);
}

static void virtio_legacy_setup_tx_queue(struct virtio_device* dev, uint16_t idx) {
	if (idx != 1 && idx != 2) {
		error("Can't setup queue %u as Tx queue", idx);
	}

	// Create virt queue itself - Section 4.1.5.1.3
	write_io16(dev->fd, idx, VIRTIO_PCI_QUEUE_SEL);
	uint32_t max_queue_size = read_io16(dev->fd, VIRTIO_PCI_QUEUE_NUM);
	debug("Max queue size of tx queue #%u: %u", idx, max_queue_size);
	if (max_queue_size == 0) {
		return;
	}
	size_t virt_queue_mem_size = virtio_legacy_vring_size(max_queue_size, 4096);
	struct dma_memory mem = memory_allocate_dma(virt_queue_mem_size, true);
	memset(mem.virt, 0xab, virt_queue_mem_size);
	debug("Allocated %zu bytes for virt queue at %p", virt_queue_mem_size, mem.virt);
	write_io32(dev->fd, mem.phy >> VIRTIO_PCI_QUEUE_ADDR_SHIFT, VIRTIO_PCI_QUEUE_PFN);

	// Section 2.4.2 for layout
	struct virtqueue* vq = calloc(1, sizeof(*vq) + sizeof(void*) * max_queue_size);
	virtio_legacy_vring_init(&vq->vring, max_queue_size, mem.virt, 4096);
	debug("vring desc: %p, vring avail: %p, vring used: %p", vq->vring.desc, vq->vring.avail, vq->vring.used);
	for (size_t i = 0; i < vq->vring.num; ++i) {
		vq->vring.desc[i].len = 0;
		vq->vring.desc[i].addr = 0;
		vq->vring.desc[i].flags = 0;
		vq->vring.desc[i].next = 0;
		vq->vring.avail->ring[i] = 0;
		vq->vring.used->ring[i].id = 0;
		vq->vring.used->ring[i].len = 0;
	}
	vq->vring.used->idx = 0;
	vq->vring.avail->idx = 0;
	vq->vq_used_last_idx = 0;

	// Section 4.1.4.4
	uint32_t notify_offset = read_io16(dev->fd, VIRTIO_PCI_QUEUE_NOTIFY);
	debug("vq notifcation offset %u", notify_offset);
	vq->notification_offset = notify_offset;

	// Ctrl queue packets are not supplied by the user
	if (idx == 2) {
		vq->mempool = memory_allocate_mempool(max_queue_size, 2048);
	}

	// Disable interrupts - Section 2.4.7
	vq->vring.avail->flags = VRING_AVAIL_F_NO_INTERRUPT;
	vq->vring.used->flags = 0;

	if (idx == 1) {
		dev->tx_queue = vq;
	} else {
		dev->ctrl_queue = vq;
	}
}

static void virtio_legacy_send_command(struct virtio_device* dev, void* cmd, size_t cmd_len) {
	struct virtqueue* vq = dev->ctrl_queue;

	if (cmd_len < sizeof(struct virtio_net_ctrl_hdr)) {
		error("Command can not be shorter than control header");
	}
	if (((uint8_t*)cmd)[0] != VIRTIO_NET_CTRL_RX) {
		error("Command class is not supported");
	}

	_mm_mfence();
	// Find free desciptor slot
	uint16_t idx = 0;
	for (idx = 0; idx < vq->vring.num; ++idx) {
		struct vring_desc* desc = &vq->vring.desc[idx];
		if (desc->addr == 0) {
			break;
		}
	}
	if (idx == vq->vring.num) {
		error("command queue full");
	} else {
		debug("Found free desc slot at %u (%u)", idx, vq->vring.num);
	}

	struct pkt_buf* buf = pkt_buf_alloc(vq->mempool);
	if (!buf) {
		error("Control queue ran out of buffers");
	}
	memcpy(buf->data, cmd, cmd_len);
	vq->virtual_addresses[idx] = buf;

	/* The following descriptor setup kills QEMU, but should be allowed with VIRTIO_F_ANY_LAYOUT
	 * Error: kvm: virtio-net ctrl missing headers
	 * Version: QEMU emulator version 2.7.1 pve-qemu-kvm_2.7.1-4
	 */
	// All in one descriptor
	// vq->vring.desc[idx].len = cmd_len;
	// vq->vring.desc[idx].addr = buf->buf_addr_phy + offsetof(struct pkt_buf, data);
	// vq->vring.desc[idx].flags = VRING_DESC_F_WRITE;
	// vq->vring.desc[idx].next = 0;

	// Device-readable head: cmd header
	vq->vring.desc[idx].len = 2;
	vq->vring.desc[idx].addr = buf->buf_addr_phy + offsetof(struct pkt_buf, data);
	vq->vring.desc[idx].flags = VRING_DESC_F_NEXT;
	vq->vring.desc[idx].next = idx + 1;
	// Device-readable payload: data
	vq->vring.desc[idx + 1].len = cmd_len - 2 - 1; // Header and ack byte
	vq->vring.desc[idx + 1].addr = buf->buf_addr_phy + offsetof(struct pkt_buf, data) + 2;
	vq->vring.desc[idx + 1].flags = VRING_DESC_F_NEXT;
	vq->vring.desc[idx + 1].next = idx + 2;
	// Device-writable tail: ack flag
	vq->vring.desc[idx + 2].len = 1;
	vq->vring.desc[idx + 2].addr = buf->buf_addr_phy + offsetof(struct pkt_buf, data) + cmd_len - 1;
	vq->vring.desc[idx + 2].flags = VRING_DESC_F_WRITE;
	vq->vring.desc[idx + 2].next = 0;
	vq->vring.avail->ring[vq->vring.avail->idx % vq->vring.num] = idx;
	_mm_mfence();
	vq->vring.avail->idx++;
	_mm_mfence();

	virtio_legacy_notify_queue(dev, 2);
	_mm_mfence();

	// Wait until the buffer got processed
	while (vq->vq_used_last_idx == vq->vring.used->idx) {
		_mm_mfence();
		debug("Waiting...");
		usleep(100000);
	}
	vq->vq_used_last_idx++;
	// Check status and free buffer
	struct vring_used_elem* e = &vq->vring.used->ring[vq->vring.used->idx];
	debug("e %p: id %u len %u", e, e->id, e->len);
	if (e->id != idx) {
		error("Used buffer has different index as sent one");
	}
	if (vq->virtual_addresses[idx] != buf) {
		error("buffer differ");
	}
	pkt_buf_free(buf);
	vq->vring.desc[idx] = (struct vring_desc){};
	vq->vring.desc[idx + 1] = (struct vring_desc){};
	vq->vring.desc[idx + 2] = (struct vring_desc){};
}

static void virtio_legacy_set_promiscuous(struct virtio_device* dev, bool on) {
	struct {
		struct virtio_net_ctrl_hdr hdr;
		uint8_t on;
		uint8_t ack;
	} __attribute__((__packed__)) cmd = {};
	static_assert(sizeof(cmd) == 4, "Size of command struct wrong");

	cmd.hdr.class = VIRTIO_NET_CTRL_RX;
	cmd.hdr.cmd = VIRTIO_NET_CTRL_RX_PROMISC;
	cmd.on = on ? 1 : 0;

	virtio_legacy_send_command(dev, &cmd, sizeof(cmd));
	info("Set promisc to %u", on);
}

void virtio_set_promisc(struct ixy_device* ixy, bool enabled) {
	struct virtio_device* dev = IXY_TO_VIRTIO(ixy);
	virtio_legacy_set_promiscuous(dev, enabled);
}

uint32_t virtio_get_link_speed(const struct ixy_device* dev) {
	return 1000;
}

static const struct virtio_legacy_net_hdr net_hdr = {
	.flags = 0,
	.gso_type = VIRTIO_NET_HDR_GSO_NONE,
	.hdr_len = 14 + 20 + 8,
};

static void virtio_legacy_setup_rx_queue(struct virtio_device* dev, uint16_t idx) {
	if (idx != 0) {
		error("Can't setup Tx queue as Rx");
	}

	// Create virt queue itself - Section 4.1.5.1.3
	write_io16(dev->fd, idx, VIRTIO_PCI_QUEUE_SEL);
	uint32_t max_queue_size = read_io16(dev->fd, VIRTIO_PCI_QUEUE_NUM);
	debug("Max queue size of rx queue #%u: %u", idx, max_queue_size);
	if (max_queue_size == 0) {
		return;
	}
	uint32_t notify_offset = read_io16(dev->fd, VIRTIO_PCI_QUEUE_NOTIFY);
	debug("Notifcation offset %u", notify_offset);
	size_t virt_queue_mem_size = virtio_legacy_vring_size(max_queue_size, 4096);
	struct dma_memory mem = memory_allocate_dma(virt_queue_mem_size, true);
	memset(mem.virt, 0xab, virt_queue_mem_size);
	debug("Allocated %zu bytes for virt queue at %p", virt_queue_mem_size, mem.virt);
	write_io32(dev->fd, mem.phy >> VIRTIO_PCI_QUEUE_ADDR_SHIFT, VIRTIO_PCI_QUEUE_PFN);

	// Section 2.4.2 for layout
	struct virtqueue* vq = calloc(1, sizeof(*vq) + sizeof(void*) * max_queue_size);
	virtio_legacy_vring_init(&vq->vring, max_queue_size, mem.virt, 4096);
	debug("vring desc: %p, vring avail: %p, vring used: %p", vq->vring.desc, vq->vring.avail, vq->vring.used);
	for (size_t i = 0; i < vq->vring.num; ++i) {
		vq->vring.desc[i].len = 0;
		vq->vring.desc[i].addr = 0;
		vq->vring.desc[i].flags = 0;
		vq->vring.desc[i].next = 0;
		vq->vring.avail->ring[i] = 0;
		vq->vring.used->ring[i].id = 0;
		vq->vring.used->ring[i].len = 0;
	}
	vq->vring.used->idx = 0;
	vq->vring.avail->idx = 0;
	vq->vq_used_last_idx = 0;

	// Section 4.1.4.4
	vq->notification_offset = notify_offset;

	// Disable interrupts - Section 2.4.7
	vq->vring.avail->flags = VRING_AVAIL_F_NO_INTERRUPT;
	vq->vring.used->flags = 0;

	// Allocate buffers and fill descriptor table - Section 3.2.1
	// We allocate more bufs than what would fit in the queue,
	// because we don't want to stall rx if users hold bufs for longer
	vq->mempool = memory_allocate_mempool(max_queue_size * 4, 2048);

	dev->rx_queue = vq;
}

static void virtio_legacy_init(struct virtio_device* dev) {
	// Section 3.1
	debug("Configuring bar0");
	write_io8(dev->fd, VIRTIO_CONFIG_STATUS_RESET, VIRTIO_PCI_STATUS);
	while (read_io8(dev->fd, VIRTIO_PCI_STATUS) != VIRTIO_CONFIG_STATUS_RESET) {
		usleep(100);
	}
	write_io8(dev->fd, VIRTIO_CONFIG_STATUS_ACK, VIRTIO_PCI_STATUS);
	write_io8(dev->fd, VIRTIO_CONFIG_STATUS_DRIVER, VIRTIO_PCI_STATUS);
	// Negotiate features
	uint32_t host_features = read_io32(dev->fd, VIRTIO_PCI_HOST_FEATURES);
	debug("Host features: %x", host_features);
	const uint32_t required_features = (1u << VIRTIO_NET_F_CSUM) | (1u << VIRTIO_NET_F_GUEST_CSUM) |
					   (1u << VIRTIO_NET_F_CTRL_VQ) | (1u << VIRTIO_F_ANY_LAYOUT) |
					   (1u << VIRTIO_NET_F_CTRL_RX) /*| (1u<<VIRTIO_NET_F_MQ)*/;
	if ((host_features & required_features) != required_features) {
		error("Device does not support required features");
	}
	debug("Guest features before negotiation: %x", read_io32(dev->fd, VIRTIO_PCI_GUEST_FEATURES));
	write_io32(dev->fd, required_features, VIRTIO_PCI_GUEST_FEATURES);
	debug("Guest features after negotiation: %x", read_io32(dev->fd, VIRTIO_PCI_GUEST_FEATURES));
	// Queue setup - Section 5.1.2 for queue index calculation
	// Legacy devices only have 3 queues
	virtio_legacy_setup_rx_queue(dev, 0); // Rx
	virtio_legacy_setup_tx_queue(dev, 1); // Tx
	virtio_legacy_setup_tx_queue(dev, 2); // Control
	_mm_mfence();
	// Signal OK
	write_io8(dev->fd, VIRTIO_CONFIG_STATUS_DRIVER_OK, VIRTIO_PCI_STATUS);
	info("Setup complete");
	// Recheck status
	virtio_legacy_check_status(dev);
	virtio_legacy_set_promiscuous(dev, true);
}

// read stat counters and accumulate in stats
// stats may be NULL to just reset the counters
// this is not thread-safe, (but we only support one queue anyways)
// a proper thread-safe implementation would collect per-queue stats
// and perform a read with relaxed memory ordering here without resetting the stats
void virtio_read_stats(struct ixy_device* ixy, struct device_stats* stats) {
	struct virtio_device* dev = IXY_TO_VIRTIO(ixy);
	if (stats) {
		stats->rx_pkts += dev->rx_pkts;
		stats->tx_pkts += dev->tx_pkts;
		stats->rx_bytes += dev->rx_bytes;
		stats->tx_bytes += dev->tx_bytes;
	}
	dev->rx_pkts = dev->tx_pkts = dev->rx_bytes = dev->tx_bytes = 0;
}


struct ixy_device* virtio_init(const char* pci_addr, uint16_t rx_queues, uint16_t tx_queues) {
	if (getuid()) {
		warn("Not running as root, this will probably fail");
	}
	if (rx_queues > 1) {
		error("cannot configure %d rx queues: limit is %d", rx_queues, 1);
	}
	if (tx_queues > 1) {
		error("cannot configure %d tx queues: limit is %d", tx_queues, 1);
	}
	remove_driver(pci_addr);
	struct virtio_device* dev = calloc(1, sizeof(*dev));
	dev->ixy.pci_addr = strdup(pci_addr);
	dev->ixy.driver_name = driver_name;
	dev->ixy.num_rx_queues = rx_queues;
	dev->ixy.num_tx_queues = tx_queues;
	dev->ixy.rx_batch = virtio_rx_batch;
	dev->ixy.tx_batch = virtio_tx_batch;
	dev->ixy.read_stats = virtio_read_stats;
	dev->ixy.set_promisc = virtio_set_promisc;
	dev->ixy.get_link_speed = virtio_get_link_speed;
	enable_dma(pci_addr);
	int config = pci_open_resource(pci_addr, "config", O_RDONLY);
	uint16_t device_id = read_io16(config, 2);
	close(config);
	// Check config if device is legacy network card
	if (device_id == 0x1000) {
		info("Detected virtio legacy network card");
		dev->fd = pci_open_resource(pci_addr, "resource0", O_RDWR);
		virtio_legacy_init(dev);
	} else {
		error("Modern device not supported");
	}
	return &dev->ixy;
}

uint32_t virtio_rx_batch(struct ixy_device* ixy, uint16_t queue_id, struct pkt_buf* bufs[], uint32_t num_bufs) {
	struct virtio_device* dev = IXY_TO_VIRTIO(ixy);
	struct virtqueue* vq = dev->rx_queue;
	uint32_t buf_idx;

	_mm_mfence();
	// Retrieve used bufs from the device
	for (buf_idx = 0; buf_idx < num_bufs; ++buf_idx) {
		// Section 3.2.2
		if (vq->vq_used_last_idx == vq->vring.used->idx) {
			break;
		}
		// info("Rx packet: last used %u, used idx %u", vq->vq_used_last_idx,
		// vq->vring.used->idx);
		struct vring_used_elem* e = vq->vring.used->ring + (vq->vq_used_last_idx % vq->vring.num);
		// info("Used elem %p, id %u len %u", e, e->id, e->len);
		struct vring_desc* desc = &vq->vring.desc[e->id];
		vq->vq_used_last_idx++;
		// We don't support chaining or indirect descriptors
		if (desc->flags != VRING_DESC_F_WRITE) {
			error("unsupported rx flags on descriptor: %x", desc->flags);
		}
		// info("Desc %lu %u %u %u", desc->addr, desc->len, desc->flags,
		// desc->next);
		*desc = (struct vring_desc){};

		// Section 5.1.6.4
		struct pkt_buf* buf = vq->virtual_addresses[e->id];
		buf->size = e->len - sizeof(net_hdr);
		bufs[buf_idx] = buf;
		//struct virtio_net_hdr* hdr = (void*)(buf->head_room + sizeof(buf->head_room) - sizeof(net_hdr));

		// Update rx counter
		dev->rx_bytes += buf->size;
		dev->rx_pkts++;
	}
	// Fill empty slots in descriptor table
	for (uint16_t idx = 0; idx < vq->vring.num; ++idx) {
		struct vring_desc* desc = &vq->vring.desc[idx];
		if (desc->addr != 0) { // descriptor points to something, therefore it is in use
			continue;
		}
		// info("Found free desc slot at %u (%u)", idx, vq->vring.num);
		struct pkt_buf* buf = pkt_buf_alloc(vq->mempool);
		if (!buf) {
			error("failed to allocate new mbuf for rx, you are either leaking memory or your mempool is too small");
		}
		buf->size = vq->mempool->buf_size;
		memcpy(buf->head_room + sizeof(buf->head_room) - sizeof(net_hdr), &net_hdr, sizeof(net_hdr));
		vq->vring.desc[idx].len = buf->size + sizeof(net_hdr);
		vq->vring.desc[idx].addr =
			buf->buf_addr_phy + offsetof(struct pkt_buf, head_room) + sizeof(buf->head_room) - sizeof(net_hdr);
		vq->vring.desc[idx].flags = VRING_DESC_F_WRITE;
		vq->vring.desc[idx].next = 0;
		vq->virtual_addresses[idx] = buf;
		vq->vring.avail->ring[vq->vring.avail->idx % vq->vring.num] = idx;
		_mm_mfence(); // Make sure exposed descriptors reach device before index is updated
		vq->vring.avail->idx++;
		_mm_mfence(); // Make sure the index update reaches device before it is triggered
		virtio_legacy_notify_queue(dev, 0);
	}
	return buf_idx;
}

uint32_t virtio_tx_batch(struct ixy_device* ixy, uint16_t queue_id, struct pkt_buf* bufs[], uint32_t num_bufs) {
	struct virtio_device* dev = IXY_TO_VIRTIO(ixy);
	struct virtqueue* vq = dev->tx_queue;

	_mm_mfence();
	// Free sent buffers
	while (vq->vq_used_last_idx != vq->vring.used->idx) {
		// info("We can free some buffers: %u != %u", vq->vq_used_last_idx,
		// vq->vring.used->idx);
		struct vring_used_elem* e = vq->vring.used->ring + (vq->vq_used_last_idx % vq->vring.num);
		// info("e %p, id %u", e, e->id);
		struct vring_desc* desc = &vq->vring.desc[e->id];
		desc->addr = 0;
		desc->len = 0;
		pkt_buf_free(vq->virtual_addresses[e->id]);
		vq->vq_used_last_idx++;
		_mm_mfence();
	}
	// Send buffers
	uint32_t buf_idx;
	uint16_t idx = 0; // Keep index of last found free descriptor and start searching from there
	for (buf_idx = 0; buf_idx < num_bufs; ++buf_idx) {
		struct pkt_buf* buf = bufs[buf_idx];
		// Find free desc index
		for (; idx < vq->vring.num; ++idx) {
			struct vring_desc* desc = &vq->vring.desc[idx];
			if (desc->addr == 0) {
				break;
			}
		}
		if (idx == vq->vring.num) {
			break;
		}
		// info("Found free desc slot at %u (%u)", idx, vq->vring.num);

		// Update tx counter
		dev->tx_bytes += buf->size;
		dev->tx_pkts++;

		vq->virtual_addresses[idx] = buf;

		// Copy header to headroom in front of data buffer
		memcpy(buf->head_room + sizeof(buf->head_room) - sizeof(net_hdr), &net_hdr, sizeof(net_hdr));

		vq->vring.desc[idx].len = buf->size + sizeof(net_hdr);
		vq->vring.desc[idx].addr =
			buf->buf_addr_phy + offsetof(struct pkt_buf, head_room) + sizeof(buf->head_room) - sizeof(net_hdr);
		vq->vring.desc[idx].flags = 0;
		vq->vring.desc[idx].next = 0;
		vq->vring.avail->ring[(vq->vring.avail->idx + buf_idx) % vq->vring.num] = idx;
	}
	_mm_mfence();
	vq->vring.avail->idx += buf_idx;
	_mm_mfence();
	virtio_legacy_notify_queue(dev, 1);
	return buf_idx;
}