vhost_user_fs: Add support for EVENT_IDX

Now that Queue supports EVENT_IDX, expose the feature and add support
for it in vhost_user_fs.

Signed-off-by: Sergio Lopez <slp@redhat.com>

src/bin/vhost_user_fs.rs

@@ -13,19 +13,23 @@ use clap::{App, Arg};
 use epoll;
 use libc::EFD_NONBLOCK;
 use log::*;
+use std::num::Wrapping;
 use std::sync::{Arc, RwLock};
 use std::{convert, error, fmt, io, process};
 
 use vhost_rs::vhost_user::message::*;
 use vhost_rs::vhost_user::SlaveFsCacheReq;
 use vhost_user_backend::{VhostUserBackend, VhostUserDaemon, Vring};
+use vhost_user_fs::descriptor_utils::Error as VufDescriptorError;
 use vhost_user_fs::descriptor_utils::{Reader, Writer};
 use vhost_user_fs::filesystem::FileSystem;
 use vhost_user_fs::passthrough::{self, PassthroughFs};
 use vhost_user_fs::server::Server;
 use vhost_user_fs::Error as VhostUserFsError;
 use virtio_bindings::bindings::virtio_net::*;
-use virtio_bindings::bindings::virtio_ring::VIRTIO_RING_F_INDIRECT_DESC;
+use virtio_bindings::bindings::virtio_ring::{
+    VIRTIO_RING_F_EVENT_IDX, VIRTIO_RING_F_INDIRECT_DESC,
+};
 use vm_memory::GuestMemoryMmap;
 use vmm_sys_util::eventfd::EventFd;
 
@@ -54,6 +58,12 @@ enum Error {
     NoMemoryConfigured,
     /// Processing queue failed.
     ProcessQueue(VhostUserFsError),
+    /// Creating a queue reader failed.
+    QueueReader(VufDescriptorError),
+    /// Creating a queue writer failed.
+    QueueWriter(VufDescriptorError),
+    /// Signaling queue failed.
+    SignalQueue(io::Error),
 }
 
 impl fmt::Display for Error {
@@ -76,6 +86,7 @@ struct VhostUserFsBackend<F: FileSystem + Send + Sync + 'static> {
     server: Arc<Server<F>>,
     // handle request from slave to master
     vu_req: Option<SlaveFsCacheReq>,
+    event_idx: bool,
 }
 
 impl<F: FileSystem + Send + Sync + 'static> Clone for VhostUserFsBackend<F> {
@@ -85,6 +96,7 @@ impl<F: FileSystem + Send + Sync + 'static> Clone for VhostUserFsBackend<F> {
             kill_evt: self.kill_evt.try_clone().unwrap(),
             server: self.server.clone(),
             vu_req: self.vu_req.clone(),
+            event_idx: self.event_idx,
         }
     }
 }
@@ -96,36 +108,39 @@ impl<F: FileSystem + Send + Sync + 'static> VhostUserFsBackend<F> {
             kill_evt: EventFd::new(EFD_NONBLOCK).map_err(Error::CreateKillEventFd)?,
             server: Arc::new(Server::new(fs)),
             vu_req: None,
+            event_idx: false,
         })
     }
 
-    fn process_queue(&mut self, vring: &mut Vring) -> Result<()> {
+    fn process_queue(&mut self, vring: &mut Vring) -> Result<bool> {
+        let mut used_any = false;
         let mem = self.mem.as_ref().ok_or(Error::NoMemoryConfigured)?;
 
-        let mut used_desc_heads = [(0, 0); QUEUE_SIZE];
-        let mut used_count = 0;
         while let Some(avail_desc) = vring.mut_queue().iter(&mem).next() {
             let head_index = avail_desc.index;
-            let reader = Reader::new(mem, avail_desc.clone()).unwrap();
-            let writer = Writer::new(mem, avail_desc.clone()).unwrap();
+            let reader = Reader::new(mem, avail_desc.clone()).map_err(Error::QueueReader)?;
+            let writer = Writer::new(mem, avail_desc.clone()).map_err(Error::QueueWriter)?;
 
-            let total = self
-                .server
+            self.server
                 .handle_message(reader, writer, self.vu_req.as_mut())
                 .map_err(Error::ProcessQueue)?;
-
-            used_desc_heads[used_count] = (head_index, total);
-            used_count += 1;
-        }
-
-        if used_count > 0 {
-            for &(desc_index, _) in &used_desc_heads[..used_count] {
-                vring.mut_queue().add_used(&mem, desc_index, 0);
+            if self.event_idx {
+                if let Some(used_idx) = vring.mut_queue().add_used(mem, head_index, 0) {
+                    let used_event = vring.mut_queue().get_used_event(mem);
+                    if vring.needs_notification(Wrapping(used_idx), used_event) {
+                        vring.signal_used_queue().map_err(Error::SignalQueue)?;
+                    }
+                    used_any = true;
+                }
+            } else {
+                vring.mut_queue().add_used(mem, head_index, 0);
+                vring.signal_used_queue().map_err(Error::SignalQueue)?;
+                used_any = true;
             }
-            vring.signal_used_queue().unwrap();
+            vring.signal_used_queue().map_err(Error::SignalQueue)?;
         }
 
-        Ok(())
+        Ok(used_any)
     }
 }
 
@@ -141,6 +156,7 @@ impl<F: FileSystem + Send + Sync + 'static> VhostUserBackend for VhostUserFsBack
     fn features(&self) -> u64 {
         1 << VIRTIO_F_VERSION_1
             | 1 << VIRTIO_RING_F_INDIRECT_DESC
+            | 1 << VIRTIO_RING_F_EVENT_IDX
             | VhostUserVirtioFeatures::PROTOCOL_FEATURES.bits()
     }
 
@@ -148,7 +164,9 @@ impl<F: FileSystem + Send + Sync + 'static> VhostUserBackend for VhostUserFsBack
         VhostUserProtocolFeatures::MQ | VhostUserProtocolFeatures::SLAVE_REQ
     }
 
-    fn set_event_idx(&mut self, _enabled: bool) {}
+    fn set_event_idx(&mut self, enabled: bool) {
+        self.event_idx = enabled;
+    }
 
     fn update_memory(&mut self, mem: GuestMemoryMmap) -> VhostUserBackendResult<()> {
         self.mem = Some(mem);
@@ -172,7 +190,23 @@ impl<F: FileSystem + Send + Sync + 'static> VhostUserBackend for VhostUserFsBack
             REQ_QUEUE_EVENT => {
                 debug!("REQ_QUEUE_EVENT");
                 let mut vring = vrings[1].write().unwrap();
-                self.process_queue(&mut vring)?;
+                if self.event_idx {
+                    // vm-virtio's Queue implementation only checks avail_index
+                    // once, so to properly support EVENT_IDX we need to keep
+                    // calling process_queue() until it stops finding new
+                    // requests on the queue.
+                    loop {
+                        vring.mut_queue().update_avail_event(
+                            self.mem.as_ref().ok_or(Error::NoMemoryConfigured)?,
+                        );
+                        if !self.process_queue(&mut vring)? {
+                            break;
+                        }
+                    }
+                } else {
+                    // Without EVENT_IDX, a single call is enough.
+                    self.process_queue(&mut vring)?;
+                }
             }
             _ => return Err(Error::HandleEventUnknownEvent.into()),
         }