Try #1711:

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

@@ -900,7 +900,7 @@ impl<'n> Nexus<'n> {
                         nexus_name,
                         child_device, "Unplugging nexus child device",
                     );
-                    child.unplug();
+                    child.unplug().await;
                 }
                 None => {
                     warn!(

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

@@ -24,8 +24,6 @@ use crate::{
         BlockDeviceHandle,
         CoreError,
         DeviceEventSink,
-        Reactor,
-        Reactors,
         VerboseError,
     },
     eventing::replica_events::state_change_event_meta,
@@ -1109,7 +1107,7 @@ impl<'c> NexusChild<'c> {
     /// underlying device is removed.
     ///
     /// Note: The descriptor *must* be dropped for the unplug to complete.
-    pub(crate) fn unplug(&mut self) {
+    pub(crate) async fn unplug(&mut self) {
         info!("{self:?}: unplugging child...");
 
         let state = self.state();
@@ -1139,12 +1137,10 @@ impl<'c> NexusChild<'c> {
         // device-related events directly.
         if state != ChildState::Faulted(FaultReason::IoError) {
             let nexus_name = self.parent.clone();
-            Reactor::block_on(async move {
-                match nexus_lookup_mut(&nexus_name) {
-                    Some(n) => n.reconfigure(DrEvent::ChildUnplug).await,
-                    None => error!("Nexus '{nexus_name}' not found"),
-                }
-            });
+            match nexus_lookup_mut(&nexus_name) {
+                Some(n) => n.reconfigure(DrEvent::ChildUnplug).await,
+                None => error!("Nexus '{nexus_name}' not found"),
+            }
         }
 
         if is_destroying {
@@ -1153,22 +1149,16 @@ impl<'c> NexusChild<'c> {
             self.device_descriptor.take();
         }
 
-        self.unplug_complete();
-        info!("{self:?}: child successfully unplugged");
+        self.unplug_complete().await;
     }
 
     /// Signal that the child unplug is complete.
-    fn unplug_complete(&self) {
-        let sender = self.remove_channel.0.clone();
-        let name = self.name.clone();
-        Reactors::current().send_future(async move {
-            if let Err(e) = sender.send(()).await {
-                error!(
-                    "Failed to send unplug complete for child '{}': {}",
-                    name, e
-                );
-            }
-        });
+    async fn unplug_complete(&self) {
+        if let Err(error) = self.remove_channel.0.send(()).await {
+            info!("{self:?}: failed to send unplug complete: {error}");
+        } else {
+            info!("{self:?}: child successfully unplugged");
+        }
     }
 
     /// create a new nexus child

io-engine/src/bdev/nvmx/controller.rs

@@ -1071,8 +1071,11 @@ pub(crate) mod options {
             self.admin_timeout_ms = Some(timeout);
             self
         }
-        pub fn with_fabrics_connect_timeout_us(mut self, timeout: u64) -> Self {
-            self.fabrics_connect_timeout_us = Some(timeout);
+        pub fn with_fabrics_connect_timeout_us<T: Into<Option<u64>>>(
+            mut self,
+            timeout: T,
+        ) -> Self {
+            self.fabrics_connect_timeout_us = timeout.into();
             self
         }
 

io-engine/src/bdev/nvmx/qpair.rs

@@ -467,9 +467,9 @@ impl<'a> Connection<'a> {
             0 => Ok(false),
             // Connection is still in progress, keep polling.
             1 => Ok(true),
-            // Error occured during polling.
+            // Error occurred during polling.
             e => {
-                let e = Errno::from_i32(-e);
+                let e = Errno::from_i32(e.abs());
                 error!(?self, "I/O qpair async connection polling error: {e}");
                 Err(e)
             }

io-engine/src/bdev/nvmx/uri.rs

@@ -227,6 +227,12 @@ impl<'probe> NvmeControllerContext<'probe> {
             )
             .with_transport_retry_count(
                 Config::get().nvme_bdev_opts.transport_retry_count as u8,
+            )
+            .with_fabrics_connect_timeout_us(
+                crate::subsys::config::opts::try_from_env(
+                    "NVMF_FABRICS_CONNECT_TIMEOUT",
+                    1_000_000,
+                ),
             );
 
         let hostnqn = template.hostnqn.clone().or_else(|| {

io-engine/src/subsys/config/opts.rs

@@ -156,7 +156,7 @@ pub struct NvmfTcpTransportOpts {
 }
 
 /// try to read an env variable or returns the default when not found
-fn try_from_env<T>(name: &str, default: T) -> T
+pub(crate) fn try_from_env<T>(name: &str, default: T) -> T
 where
     T: FromStr + Display + Copy,
     <T as FromStr>::Err: Debug + Display,

io-engine/src/subsys/nvmf/target.rs

@@ -27,7 +27,7 @@ use spdk_rs::libspdk::{
 
 use crate::{
     constants::NVME_CONTROLLER_MODEL_ID,
-    core::{Cores, Mthread, Reactor, Reactors},
+    core::{Cores, Mthread, Reactors},
     ffihelper::{AsStr, FfiResult},
     subsys::{
         nvmf::{
@@ -272,7 +272,7 @@ impl Target {
 
     /// enable discovery for the target -- note that the discovery system is not
     /// started
-    fn enable_discovery(&self) {
+    fn enable_discovery(&self) -> NvmfSubsystem {
         debug!("enabling discovery for target");
         let discovery = unsafe {
             NvmfSubsystem::from(spdk_nvmf_subsystem_create(
@@ -296,12 +296,7 @@ impl Target {
 
         discovery.allow_any(true);
 
-        Reactor::block_on(async {
-            let nqn = discovery.get_nqn();
-            if let Err(e) = discovery.start().await {
-                error!("Error starting subsystem '{}': {}", nqn, e.to_string());
-            }
-        });
+        discovery
     }
 
     /// stop all subsystems on this target we are borrowed here
@@ -355,13 +350,20 @@ impl Target {
 
     /// Final state for the target during init.
     pub fn running(&mut self) {
-        self.enable_discovery();
-        info!(
-            "nvmf target accepting new connections and is ready to roll..{}",
-            '\u{1F483}'
-        );
+        let discovery = self.enable_discovery();
 
-        unsafe { spdk_subsystem_init_next(0) }
+        Reactors::master().send_future(async move {
+            let nqn = discovery.get_nqn();
+            if let Err(error) = discovery.start().await {
+                error!("Error starting subsystem '{nqn}': {error}");
+            }
+
+            info!(
+                "nvmf target accepting new connections and is ready to roll..{}",
+                '\u{1F483}'
+            );
+            unsafe { spdk_subsystem_init_next(0) }
+        })
     }
 
     /// Shutdown procedure.