[blk-threaded - 12/12] tests: add unit tests and threaded coverage

src/vmm/src/devices/virtio/block/device.rs

@@ -315,3 +315,26 @@ impl Persist<'_> for Block {
         }
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::devices::virtio::block::virtio::device::FileEngineType;
+    use crate::devices::virtio::block::virtio::test_utils::default_block;
+
+    #[test]
+    fn test_spawn_worker_filter() {
+        let mut inline = Block::Virtio(default_block(FileEngineType::Sync));
+        inline.spawn_worker(None).unwrap();
+
+        let mut threaded = default_block(FileEngineType::Sync);
+        threaded.config.threaded = true;
+        let mut threaded = Block::Virtio(threaded);
+
+        assert!(matches!(
+            threaded.spawn_worker(None),
+            Err(BlockError::MissingSeccompFilter)
+        ));
+        threaded.spawn_worker(Some(Arc::new(vec![]))).unwrap();
+    }
+}

src/vmm/src/devices/virtio/block/virtio/device.rs

@@ -819,13 +819,14 @@ mod tests {
     use crate::devices::virtio::block::virtio::IO_URING_NUM_ENTRIES;
     use crate::devices::virtio::block::virtio::request::*;
     use crate::devices::virtio::block::virtio::test_utils::{
-        default_block, read_blk_req_descriptors, set_queue, set_rate_limiter,
-        simulate_async_completion_event, simulate_queue_and_async_completion_events,
-        simulate_queue_event,
+        default_block, default_threaded_block, read_blk_req_descriptors, set_queue,
+        set_rate_limiter, simulate_async_completion_event,
+        simulate_queue_and_async_completion_events, simulate_queue_event,
     };
     use crate::devices::virtio::queue::{VIRTQ_DESC_F_NEXT, VIRTQ_DESC_F_WRITE};
     use crate::devices::virtio::test_utils::{VirtQueue, default_interrupt, default_mem};
-    use crate::rate_limiter::TokenType;
+    use crate::rate_limiter::{TokenBucket, TokenType};
+    use crate::seccomp::BPF_MAX_LEN;
     use crate::vstate::memory::{Address, Bytes, GuestAddress};
 
     #[test]
@@ -1972,6 +1973,185 @@ mod tests {
         }
     }
 
+    #[test]
+    fn test_threaded_device() {
+        let mut block = default_threaded_block(FileEngineType::Sync);
+        let mem = default_mem();
+        let interrupt = default_interrupt();
+        let vq = VirtQueue::new(GuestAddress(0), &mem, 16);
+        set_queue(&mut block, 0, vq.create_queue());
+
+        block.activate(mem.clone(), interrupt).unwrap();
+
+        assert!(block.is_threaded_active());
+        assert_eq!(block.num_queues(), BLOCK_QUEUE_SIZES.len());
+        assert_eq!(block.queue_config(0).unwrap().size, vq.size());
+        assert!(block.queue_config_mut(0).is_some());
+        assert!(block.queue_event(0).is_some());
+    }
+
+    #[test]
+    fn test_threaded_update_disk() {
+        for engine in [FileEngineType::Sync, FileEngineType::Async] {
+            let mut block = default_threaded_block(engine);
+            let mem = default_mem();
+            let interrupt = default_interrupt();
+            let vq = VirtQueue::new(GuestAddress(0), &mem, 16);
+            set_queue(&mut block, 0, vq.create_queue());
+            block.activate(mem.clone(), interrupt.clone()).unwrap();
+
+            let disk = TempFile::new().unwrap();
+            disk.as_file().set_len(0x2000).unwrap();
+            let disk_path = disk.as_path().to_str().unwrap().to_string();
+
+            block.update_disk_image(disk_path.clone()).unwrap();
+
+            assert_eq!(block.config().path_on_host, disk_path);
+            assert_eq!(u64::from_le(block.config_space.capacity), 16);
+            assert!(
+                interrupt.has_pending_interrupt(VirtioInterruptType::Config),
+                "updating an active disk must notify the guest"
+            );
+        }
+    }
+
+    #[test]
+    fn test_threaded_update_disk_error() {
+        for engine in [FileEngineType::Sync, FileEngineType::Async] {
+            let mut block = default_threaded_block(engine);
+            let original_config = block.config();
+            let original_capacity = block.config_space.capacity;
+
+            let mem = default_mem();
+            let interrupt = default_interrupt();
+            let vq = VirtQueue::new(GuestAddress(0), &mem, 16);
+            set_queue(&mut block, 0, vq.create_queue());
+            block.activate(mem.clone(), interrupt.clone()).unwrap();
+
+            let mut missing_disk = TempFile::new().unwrap();
+            let missing_path = missing_disk.as_path().to_str().unwrap().to_string();
+            missing_disk.remove().unwrap();
+
+            assert!(block.update_disk_image(missing_path).is_err());
+            assert_eq!(block.config(), original_config);
+            assert_eq!(block.config_space.capacity, original_capacity);
+            assert!(!interrupt.has_pending_interrupt(VirtioInterruptType::Config));
+        }
+    }
+
+    #[test]
+    fn test_threaded_rate_limiter() {
+        let mut block = default_threaded_block(FileEngineType::Sync);
+        let mem = default_mem();
+        let vq = VirtQueue::new(GuestAddress(0), &mem, 16);
+        set_queue(&mut block, 0, vq.create_queue());
+        block.activate(mem.clone(), default_interrupt()).unwrap();
+
+        let bandwidth = TokenBucket::new(1000, 1001, 1002).unwrap();
+        let ops = TokenBucket::new(1003, 1004, 1005).unwrap();
+        block.update_rate_limiter(BucketUpdate::Update(bandwidth), BucketUpdate::Update(ops));
+
+        // Reset the device before reactivation.
+        assert!(block.reset());
+        assert_eq!(
+            block.config().rate_limiter,
+            RateLimiterConfig::from(&*block.rate_limiter()).into_option()
+        );
+
+        set_queue(&mut block, 0, vq.create_queue());
+        block.activate(mem, default_interrupt()).unwrap();
+        block.update_rate_limiter(BucketUpdate::Disabled, BucketUpdate::Disabled);
+
+        assert!(block.reset());
+        assert_eq!(block.config().rate_limiter, None);
+        assert!(block.rate_limiter().bandwidth().is_none());
+        assert!(block.rate_limiter().ops().is_none());
+    }
+
+    #[test]
+    fn test_threaded_queue_dirty() {
+        let mut block = default_threaded_block(FileEngineType::Sync);
+        let mem = default_mem();
+        let vq = VirtQueue::new(GuestAddress(0), &mem, 16);
+        set_queue(&mut block, 0, vq.create_queue());
+        block.activate(mem.clone(), default_interrupt()).unwrap();
+
+        assert!(matches!(
+            block.mark_queue_memory_dirty(&mem),
+            Err(QueueError::NotReady)
+        ));
+
+        block.prepare_save();
+
+        block.mark_queue_memory_dirty(&mem).unwrap();
+    }
+
+    #[test]
+    fn test_threaded_resume() {
+        let mut block = default_threaded_block(FileEngineType::Sync);
+        let mem = default_mem();
+        let vq = VirtQueue::new(GuestAddress(0), &mem, 16);
+        set_queue(&mut block, 0, vq.create_queue());
+        block.activate(mem.clone(), default_interrupt()).unwrap();
+
+        block.prepare_save();
+        read_blk_req_descriptors(&vq);
+        let request_type_addr = GuestAddress(vq.dtable[0].addr.get());
+        let status_addr = GuestAddress(vq.dtable[2].addr.get());
+        vq.dtable[1].len.set(VIRTIO_BLK_ID_BYTES);
+        mem.write_obj::<u32>(VIRTIO_BLK_T_GET_ID, request_type_addr)
+            .unwrap();
+        block.mark_queue_memory_dirty(&mem).unwrap();
+
+        block.kick();
+        // A second pause waits until the resumed worker has processed its queue.
+        block.prepare_save();
+
+        assert_eq!(vq.used.idx.get(), 1);
+        assert_eq!(vq.used.ring[0].get().id, 0);
+        assert_eq!(vq.used.ring[0].get().len, VIRTIO_BLK_ID_BYTES + 1);
+        assert_eq!(mem.read_obj::<u32>(status_addr).unwrap(), VIRTIO_BLK_S_OK);
+    }
+
+    #[test]
+    fn test_threaded_reset_paused() {
+        for engine in [FileEngineType::Sync, FileEngineType::Async] {
+            let mut block = default_threaded_block(engine);
+            let mem = default_mem();
+            let vq = VirtQueue::new(GuestAddress(0), &mem, 16);
+            set_queue(&mut block, 0, vq.create_queue());
+            block.set_acked_features(1);
+            block.activate(mem, default_interrupt()).unwrap();
+            block.prepare_save();
+
+            assert!(block.reset());
+            assert!(!block.is_activated());
+            assert_eq!(block.acked_features(), 0);
+            assert!(!block.queue_config(0).unwrap().ready);
+            assert!(matches!(&block.state, BlockState::Configuring(_, Some(_))));
+        }
+    }
+
+    #[test]
+    fn test_threaded_reset_error() {
+        let mut block = default_block(FileEngineType::Sync);
+        block.config.threaded = true;
+        block
+            .spawn_worker(Arc::new(vec![0; BPF_MAX_LEN + 1]))
+            .unwrap();
+
+        let mem = default_mem();
+        let vq = VirtQueue::new(GuestAddress(0), &mem, 16);
+        set_queue(&mut block, 0, vq.create_queue());
+        block.set_acked_features(1);
+        block.activate(mem, default_interrupt()).unwrap();
+
+        assert!(!block.reset());
+        assert!(block.is_threaded_active());
+        assert_eq!(block.acked_features(), 1);
+        assert!(block.queue_config(0).unwrap().ready);
+    }
+
     #[test]
     fn test_reset_and_reactivation() {
         for engine in [FileEngineType::Sync, FileEngineType::Async] {

src/vmm/src/devices/virtio/block/virtio/persist.rs

@@ -164,11 +164,14 @@ impl Persist<'_> for VirtioBlock {
 
 #[cfg(test)]
 mod tests {
+    use std::sync::Arc;
     use vmm_sys_util::tempfile::TempFile;
 
     use super::*;
+    use crate::devices::virtio::block::virtio::test_utils::{default_block_with_path, set_queue};
     use crate::devices::virtio::device::VirtioDevice;
-    use crate::devices::virtio::test_utils::{default_interrupt, default_mem};
+    use crate::devices::virtio::test_utils::{VirtQueue, default_interrupt, default_mem};
+    use crate::vstate::memory::GuestAddress;
 
     #[test]
     fn test_cache_semantic_ser() {
@@ -253,4 +256,37 @@ mod tests {
         // Test that block specific fields are the same.
         assert_eq!(restored_block.disk().file_path, block.disk().file_path);
     }
+
+    #[test]
+    fn test_threaded_persistence() {
+        for engine in [FileEngineType::Sync, FileEngineType::Async] {
+            let disk = TempFile::new().unwrap();
+            disk.as_file().set_len(0x1000).unwrap();
+            let mut block =
+                default_block_with_path(disk.as_path().to_str().unwrap().to_string(), engine);
+            block.config.threaded = true;
+            block.spawn_worker(Arc::new(vec![])).unwrap();
+            let mem = default_mem();
+            let vq = VirtQueue::new(GuestAddress(0), &mem, BLOCK_QUEUE_SIZES[0]);
+            set_queue(&mut block, 0, vq.create_queue());
+            block.set_acked_features(block.avail_features());
+            block.activate(mem.clone(), default_interrupt()).unwrap();
+            // Pause the worker for snapshotting
+            block.prepare_save();
+
+            let state = block.save();
+            let serialized = bitcode::serialize(&state).unwrap();
+            let restored_state = bitcode::deserialize(&serialized).unwrap();
+            let restored =
+                VirtioBlock::restore(BlockConstructorArgs { mem }, &restored_state).unwrap();
+
+            assert!(state.threaded);
+            assert!(state.virtio_state.activated);
+            assert!(restored.config().threaded);
+            assert!(!restored.is_activated());
+            assert_eq!(restored.acked_features(), block.acked_features());
+            assert_eq!(restored.queue_config(0), block.queue_config(0));
+            assert_eq!(restored.file_engine_type(), engine);
+        }
+    }
 }

src/vmm/src/devices/virtio/block/virtio/test_utils.rs

@@ -3,6 +3,8 @@
 
 #![doc(hidden)]
 
+#[cfg(test)]
+use std::sync::Arc;
 #[cfg(test)]
 use std::thread;
 #[cfg(test)]
@@ -35,6 +37,14 @@ pub fn default_block(file_engine_type: FileEngineType) -> VirtioBlock {
     default_block_with_path(f.as_path().to_str().unwrap().to_string(), file_engine_type)
 }
 
+#[cfg(test)]
+pub fn default_threaded_block(file_engine_type: FileEngineType) -> VirtioBlock {
+    let mut block = default_block(file_engine_type);
+    block.config.threaded = true;
+    block.spawn_worker(Arc::new(vec![])).unwrap();
+    block
+}
+
 /// Create a default Block instance using file at the specified path to be used in tests.
 pub fn default_block_with_path(path: String, file_engine_type: FileEngineType) -> VirtioBlock {
     let config = VirtioBlockConfig {