feat(rvnic): add poll support for RX ring notification

Author: minhuw

rvnic/kernel/rattan_vnic.c

@@ -29,6 +29,7 @@
 #include <linux/atomic.h>
 #include <linux/ktime.h>
 #include <linux/workqueue.h>
+#include <linux/poll.h>
 
 #include "rattan_vnic.h"
 
@@ -205,6 +206,10 @@ static netdev_tx_t rattan_vnic_start_xmit(struct sk_buff *skb,
 
 	/* Success */
 	rcu_read_unlock();
+
+	/* Wake up poll waiters */
+	wake_up_interruptible(&ctx->wait);
+
 	dev->stats.tx_packets++;
 	dev->stats.tx_bytes += skb->len;
 	return NETDEV_TX_OK;
@@ -1198,6 +1203,7 @@ static int rattan_vnic_fop_open(struct inode *inode, struct file *file)
 	ctx->started = false;
 	RCU_INIT_POINTER(ctx->umem, NULL);
 	RCU_INIT_POINTER(ctx->rings, NULL);
+	init_waitqueue_head(&ctx->wait);
 
 	/* Register network device */
 	err = register_netdev(netdev);
@@ -1270,11 +1276,49 @@ static int rattan_vnic_fop_release(struct inode *inode, struct file *file)
 	return 0;
 }
 
+/*
+ * Poll implementation - allows userspace to block until RX data is available
+ *
+ * Returns:
+ *   EPOLLIN | EPOLLRDNORM - RX ring has data (producer != consumer)
+ *   EPOLLERR - device not configured
+ */
+static __poll_t rattan_vnic_poll(struct file *file, poll_table *wait)
+{
+	struct rattan_vnic_ctx *ctx = file->private_data;
+	struct rattan_rings *rings;
+	__poll_t mask = 0;
+	u32 rx_prod, rx_cons;
+
+	if (!ctx)
+		return EPOLLERR;
+
+	poll_wait(file, &ctx->wait, wait);
+
+	rcu_read_lock();
+	rings = rcu_dereference(ctx->rings);
+	if (!rings || !rings->rx) {
+		rcu_read_unlock();
+		return EPOLLERR;
+	}
+
+	/* Check RX ring: kernel produces, userspace consumes */
+	rx_prod = smp_load_acquire(&rings->rx->ring.producer);
+	rx_cons = READ_ONCE(rings->rx->ring.consumer);
+	if (rx_prod != rx_cons)
+		mask |= EPOLLIN | EPOLLRDNORM;
+
+	rcu_read_unlock();
+
+	return mask;
+}
+
 static const struct file_operations rattan_vnic_fops = {
 	.owner = THIS_MODULE,
 	.open = rattan_vnic_fop_open,
 	.release = rattan_vnic_fop_release,
 	.unlocked_ioctl = rattan_vnic_ioctl,
+	.poll = rattan_vnic_poll,
 };
 
 static struct miscdevice rattan_vnic_misc = {

rvnic/kernel/rattan_vnic.h

@@ -16,6 +16,7 @@
 #include <linux/refcount.h>
 #include <linux/ktime.h>
 #include <linux/spinlock.h>
+#include <linux/wait.h>
 
 #include "rattan_vnic_uapi.h"
 
@@ -114,6 +115,9 @@ struct rattan_vnic_ctx {
 
 	struct rattan_umem __rcu *umem;
 	struct rattan_rings __rcu *rings;
+
+	/* Poll wait queue - for blocking waits on RX availability */
+	wait_queue_head_t wait;
 };
 
 #endif /* _RATTAN_VNIC_INTERNAL_H */

rvnic/librvnic/examples/test_forward.rs

@@ -17,6 +17,7 @@
 
 use rvnic::sys::RattanDesc;
 use rvnic::{CompRing, FillRing, RATTAN_RING_SIZE, Rings, RvnicDevice, RxRing, TxRing, UmemBuilder};
+use std::os::unix::io::AsRawFd;
 use std::process::{Child, Command, Stdio};
 use std::sync::Arc;
 use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
@@ -284,65 +285,105 @@ fn main() -> Result<(), Box<dyn std::error::Error>> {
     let packets_forwarded = Arc::new(AtomicU64::new(0));
     let packets_forwarded_clone = packets_forwarded.clone();
 
-    // Two forwarding threads - one per direction
-    let running_clone2 = running.clone();
-    let packets_forwarded_clone2 = packets_forwarded.clone();
+    // Get file descriptors for epoll
+    let dev0_fd = dev0.as_raw_fd();
+    let dev1_fd = dev1.as_raw_fd();
 
-    // Thread 1: rx0 → tx1, recycle comp1 → fill0
-    let thread1 = std::thread::spawn(move || {
+    // Single-threaded forwarder using epoll
+    let forward_thread = std::thread::spawn(move || {
         let mut descs = [RattanDesc::default(); BATCH_SIZE];
         let mut addrs = [0u64; BATCH_SIZE];
 
+        // Create epoll instance
+        let epfd = unsafe { libc::epoll_create1(0) };
+        if epfd < 0 {
+            eprintln!("epoll_create1 failed");
+            return (dev0, dev1);
+        }
+
+        // Add dev0 to epoll (data.u32 = 0)
+        let mut ev = libc::epoll_event {
+            events: libc::EPOLLIN as u32,
+            u64: 0,
+        };
+        if unsafe { libc::epoll_ctl(epfd, libc::EPOLL_CTL_ADD, dev0_fd, &mut ev) } < 0 {
+            eprintln!("epoll_ctl add dev0 failed");
+            unsafe { libc::close(epfd) };
+            return (dev0, dev1);
+        }
+
+        // Add dev1 to epoll (data.u32 = 1)
+        ev.u64 = 1;
+        if unsafe { libc::epoll_ctl(epfd, libc::EPOLL_CTL_ADD, dev1_fd, &mut ev) } < 0 {
+            eprintln!("epoll_ctl add dev1 failed");
+            unsafe { libc::close(epfd) };
+            return (dev0, dev1);
+        }
+
+        let mut events = [libc::epoll_event { events: 0, u64: 0 }; 2];
+
         while running_clone.load(Ordering::Relaxed) {
-            // Forward a batch
-            let fwd = forward_batch(&mut rx0, &mut tx1, &mut descs);
+            // Wait for events on either device (10ms timeout)
+            let nfds = unsafe { libc::epoll_wait(epfd, events.as_mut_ptr(), 2, 10) };
 
-            // Aggressively recycle all available chunks
-            loop {
-                let recycled = recycle_chunks(&mut comp1, &mut fill0, &mut addrs);
-                if recycled == 0 {
+            if nfds < 0 {
+                let err = std::io::Error::last_os_error();
+                if err.kind() != std::io::ErrorKind::Interrupted {
+                    eprintln!("epoll_wait failed: {}", err);
                     break;
                 }
+                continue;
             }
 
-            if fwd > 0 {
-                packets_forwarded_clone.fetch_add(fwd as u64, Ordering::Relaxed);
-                let _ = dev1.kick_rx();
-            } else {
-                std::hint::spin_loop();
+            // Process events - drain RX rings completely
+            for i in 0..nfds as usize {
+                let dev_idx = events[i].u64;
+
+                if dev_idx == 0 {
+                    // dev0 has RX data: forward rx0 → tx1
+                    // Drain all available packets (not just one batch)
+                    loop {
+                        let fwd = forward_batch(&mut rx0, &mut tx1, &mut descs);
+                        if fwd > 0 {
+                            packets_forwarded_clone.fetch_add(fwd as u64, Ordering::Relaxed);
+                            let _ = dev1.kick_rx();
+                        }
+                        if fwd < BATCH_SIZE {
+                            break; // Ring drained
+                        }
+                    }
+                } else {
+                    // dev1 has RX data: forward rx1 → tx0
+                    // Drain all available packets (not just one batch)
+                    loop {
+                        let fwd = forward_batch(&mut rx1, &mut tx0, &mut descs);
+                        if fwd > 0 {
+                            packets_forwarded_clone.fetch_add(fwd as u64, Ordering::Relaxed);
+                            let _ = dev0.kick_rx();
+                        }
+                        if fwd < BATCH_SIZE {
+                            break; // Ring drained
+                        }
+                    }
+                }
             }
-        }
-        dev1
-    });
-
-    // Thread 2: rx1 → tx0, recycle comp0 → fill1
-    let thread2 = std::thread::spawn(move || {
-        let mut descs = [RattanDesc::default(); BATCH_SIZE];
-        let mut addrs = [0u64; BATCH_SIZE];
 
-        while running_clone2.load(Ordering::Relaxed) {
-            // Forward a batch
-            let fwd = forward_batch(&mut rx1, &mut tx0, &mut descs);
-
-            // Aggressively recycle all available chunks
+            // Always recycle chunks from both directions
+            // This ensures chunks are returned promptly regardless of which device triggered
             loop {
-                let recycled = recycle_chunks(&mut comp0, &mut fill1, &mut addrs);
-                if recycled == 0 {
+                let r0 = recycle_chunks(&mut comp1, &mut fill0, &mut addrs);
+                let r1 = recycle_chunks(&mut comp0, &mut fill1, &mut addrs);
+                if r0 == 0 && r1 == 0 {
                     break;
                 }
             }
-
-            if fwd > 0 {
-                packets_forwarded_clone2.fetch_add(fwd as u64, Ordering::Relaxed);
-                let _ = dev0.kick_rx();
-            } else {
-                std::hint::spin_loop();
-            }
         }
-        dev0
+
+        unsafe { libc::close(epfd) };
+        (dev0, dev1)
     });
 
-    println!("   Forwarding loop started");
+    println!("   Forwarding loop started (single thread, epoll)");
 
     // Step 5: Run iperf test
     println!("\n5. Running iperf3 performance test...");
@@ -377,8 +418,7 @@ fn main() -> Result<(), Box<dyn std::error::Error>> {
     println!("\n7. Cleaning up...");
     running.store(false, Ordering::Relaxed);
 
-    let _dev1 = thread1.join().unwrap();
-    let _dev0 = thread2.join().unwrap();
+    let (_dev0, _dev1) = forward_thread.join().unwrap();
 
     // netns_guard will clean up namespaces on drop
     drop(netns_guard);