fix(casperf): invalid next offset calculation

Also add easy block size specification, eg: allows "1MiB".
Disable cleanup of nexuses when they're not configured.
Prevent double cleanup with duplicate handlers.

Signed-off-by: Tiago Castro <[email protected]>

.gitignore

@@ -14,6 +14,7 @@ test-yamls/*
 /node_modules
 artifacts/
 .idea
+.vscode
 **/.pytest_cache
 **/__pycache__
 /chart/charts/

io-engine/src/bdev/nexus/mod.rs

@@ -149,6 +149,11 @@ pub fn register_module() {
 /// so that a possible remove event from SPDK also results in bdev removal
 #[allow(clippy::needless_collect)]
 pub async fn shutdown_nexuses() {
+    if NexusModule::current_opt().is_none() {
+        info!("Skipping nexus shutdown - Nexus Module not registered...");
+        return;
+    }
+
     info!("Shutting down nexuses...");
 
     // TODO: We need to collect list of Nexuses before destroying them,

io-engine/src/bdev/nexus/nexus_module.rs

@@ -21,13 +21,18 @@ pub(crate) struct NexusModule {}
 
 impl NexusModule {
     /// Returns Nexus Bdev module instance.
+    /// # Warning
     /// Panics if the Nexus module was not registered.
     pub fn current() -> BdevModule {
         match BdevModule::find_by_name(NEXUS_MODULE_NAME) {
             Ok(m) => m,
             Err(err) => panic!("{}", err),
         }
     }
+    /// Returns Nexus Bdev module instance, if registered.
+    pub fn current_opt() -> Option<BdevModule> {
+        BdevModule::find_by_name(NEXUS_MODULE_NAME).ok()
+    }
 }
 
 impl WithModuleInit for NexusModule {

io-engine/src/bin/casperf.rs

@@ -33,7 +33,7 @@ use spdk_rs::{
 };
 use version_info::version_info_str;
 
-#[derive(Debug)]
+#[derive(Debug, Clone, Copy)]
 enum IoType {
     /// perform random read operations
     Read,
@@ -62,7 +62,7 @@ struct Job {
     /// io_size the io_size is the number of blocks submit per IO
     io_size: u64,
     /// blk_size of the underlying device
-    blk_size: u32,
+    blk_size: u64,
     /// num_blocks the device has
     num_blocks: u64,
     /// aligned set of IOs we can do
@@ -130,12 +130,12 @@ impl Job {
             return;
         }
 
-        let offset = (job.rng.gen::<u64>() % job.io_size) * job.io_blocks;
+        let offset = job.rng.gen_range(0 .. job.io_blocks) * job.io_size;
         ioq.next(offset);
     }
 
     /// construct a new job
-    async fn new(bdev: &str, size: u64, qd: u64) -> Box<Self> {
+    async fn new(bdev: &str, size: u64, qd: u64, io_type: IoType) -> Box<Self> {
         let bdev = bdev_create(bdev)
             .await
             .map_err(|e| {
@@ -147,18 +147,18 @@ impl Job {
 
         let desc = bdev.open(true).unwrap();
 
-        let blk_size = bdev.block_len();
+        let blk_size = bdev.block_len() as u64;
         let num_blocks = bdev.num_blocks();
 
-        let io_size = size / blk_size as u64;
+        let io_size = size / blk_size;
         let io_blocks = num_blocks / io_size;
 
         let mut queue = Vec::new();
 
         (0 ..= qd).for_each(|offset| {
             queue.push(Io {
                 buf: DmaBuf::new(size, bdev.alignment()).unwrap(),
-                iot: IoType::Read,
+                iot: io_type,
                 offset,
                 job: NonNull::dangling(),
             });
@@ -229,21 +229,22 @@ impl Io {
 
     /// dispatch the read IO at given offset
     fn read(&mut self, offset: u64) {
+        let nbytes = self.buf.len();
         unsafe {
             if spdk_bdev_read(
                 self.job.as_ref().desc.legacy_as_ptr(),
                 self.job.as_ref().ch.as_ref().unwrap().legacy_as_ptr(),
                 self.buf.as_mut_ptr(),
                 offset,
-                self.buf.len(),
+                nbytes,
                 Some(Job::io_completion),
                 self as *const _ as *mut _,
             ) == 0
             {
                 self.job.as_mut().n_inflight += 1;
             } else {
                 eprintln!(
-                    "failed to submit read IO to {}",
+                    "failed to submit read IO to {}, offset={offset}, nbytes={nbytes}",
                     self.job.as_ref().bdev.name()
                 );
             }
@@ -348,6 +349,14 @@ fn main() {
                 .help("block size in bytes")
                 .takes_value(true),
         )
+        .arg(
+            Arg::with_name("io_type")
+                .value_name("io_type")
+                .short("t")
+                .help("type of IOs")
+                .possible_values(&["randread", "randwrite"])
+                .takes_value(true),
+        )
         .arg(
             Arg::with_name("queue_depth")
                 .value_name("queue_depth")
@@ -371,10 +380,22 @@ fn main() {
         .map(|u| u.to_string())
         .collect::<Vec<_>>();
 
-    let io_size = value_t!(matches.value_of("io_size"), u64).unwrap_or(IO_SIZE);
+    let io_size = match matches.value_of("io_size") {
+        Some(io_size) => {
+            byte_unit::Byte::from_str(io_size).unwrap().get_bytes() as u64
+        }
+        None => IO_SIZE,
+    };
+    let io_type = match matches.value_of("io_type").unwrap_or("randread") {
+        "randread" => IoType::Read,
+        "randwrite" => IoType::Write,
+        io_type => panic!("Invalid io_type: {}", io_type),
+    };
+
     let qd = value_t!(matches.value_of("queue_depth"), u64).unwrap_or(QD);
     let args = MayastorCliArgs {
         reactor_mask: "0x2".to_string(),
+        skip_sig_handler: true,
         ..Default::default()
     };
 
@@ -383,7 +404,7 @@ fn main() {
     Reactors::master().send_future(async move {
         let jobs = uris
             .iter_mut()
-            .map(|u| Job::new(u, io_size, qd))
+            .map(|u| Job::new(u, io_size, qd, io_type))
             .collect::<Vec<_>>();
 
         for j in jobs {

io-engine/src/core/env.rs

@@ -171,7 +171,7 @@ pub struct MayastorCliArgs {
         default_value = "30"
     )]
     pub nvmf_tgt_crdt: u16,
-    /// api Version
+    /// The gRPC api version.
     #[structopt(
         long,
         value_delimiter = ",",
@@ -192,6 +192,10 @@ pub struct MayastorCliArgs {
         env = "REACTOR_FREEZE_TIMEOUT"
     )]
     pub reactor_freeze_timeout: Option<u64>,
+    /// Skip install of the signal handler which will trigger process graceful
+    /// termination.
+    #[structopt(long, hidden = true)]
+    pub skip_sig_handler: bool,
 }
 
 /// Mayastor features.
@@ -242,6 +246,7 @@ impl Default for MayastorCliArgs {
             diagnose_stack: None,
             reactor_freeze_detection: false,
             reactor_freeze_timeout: None,
+            skip_sig_handler: false,
         }
     }
 }
@@ -338,6 +343,7 @@ pub struct MayastorEnvironment {
     nvmf_tgt_interface: Option<String>,
     pub nvmf_tgt_crdt: u16,
     api_versions: Vec<ApiVersion>,
+    skip_sig_handler: bool,
 }
 
 impl Default for MayastorEnvironment {
@@ -383,6 +389,7 @@ impl Default for MayastorEnvironment {
             nvmf_tgt_interface: None,
             nvmf_tgt_crdt: 30,
             api_versions: vec![ApiVersion::V0, ApiVersion::V1],
+            skip_sig_handler: false,
         }
     }
 }
@@ -501,6 +508,7 @@ impl MayastorEnvironment {
             nvmf_tgt_interface: args.nvmf_tgt_interface,
             nvmf_tgt_crdt: args.nvmf_tgt_crdt,
             api_versions: args.api_versions,
+            skip_sig_handler: args.skip_sig_handler,
             ..Default::default()
         }
         .setup_static()
@@ -880,7 +888,9 @@ impl MayastorEnvironment {
         );
 
         // setup our signal handlers
-        self.install_signal_handlers();
+        if !self.skip_sig_handler {
+            self.install_signal_handlers();
+        }
 
         // allocate a Reactor per core
         Reactors::init();