fix(nexus): fixing I/O race when disconnecting retired device handles

Signed-off-by: Dmitry Savitskiy <[email protected]>

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

@@ -1143,6 +1143,19 @@ impl<'n> Nexus<'n> {
         unsafe { Pin::new_unchecked(self.bdev_mut()) }
     }
 
+    /// Gets a nexus reference from an untyped bdev.
+    /// No checks are performed (e.g. bdev module name check), as it is assumed
+    /// that the provided bdev is a nexus bdev.
+    #[inline(always)]
+    pub(crate) unsafe fn unsafe_from_untyped_bdev(
+        bdev: spdk_rs::UntypedBdev,
+    ) -> &'n Nexus<'n> {
+        spdk_rs::Bdev::<Nexus<'n>>::unsafe_from_inner_ptr(
+            bdev.unsafe_inner_ptr() as *mut _,
+        )
+        .data()
+    }
+
     /// Sets the required alignment of the Nexus.
     pub(crate) unsafe fn set_required_alignment(
         self: Pin<&mut Self>,

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

@@ -326,9 +326,10 @@ impl<'n> Nexus<'n> {
         // Close and remove the child.
         let res = match self.lookup_child(uri) {
             Some(child) => {
-                // Remove child from the I/O path.
+                // Detach the child from the I/O path, and close its handles.
                 if let Some(device) = child.get_device_name() {
-                    self.disconnect_device(&device).await;
+                    self.detach_device(&device).await;
+                    self.disconnect_detached_devices().await;
                 }
 
                 // Close child's device.
@@ -974,19 +975,29 @@ impl<'n> Nexus<'n> {
             return Ok(());
         }
 
-        // Disconnect the device from all the channels.
+        // Detach the device from all the channels.
+        //
         // After disconnecting, the device will no longer be used by the
         // channels, and all I/Os failing due to this device will eventually
         // resubmit and succeeded (if any healthy children are left).
-        self.disconnect_device(&dev).await;
+        //
+        // Device disconnection is done in two steps (detach, than disconnect)
+        // in order to prevent an I/O race when retiring a device.
+        self.detach_device(&dev).await;
 
-        // Destroy (close) the device. The subsystem must be paused to do this
-        // properly.
+        // Disconnect, destroy and close the device. The subsystem must be
+        // paused to do this properly.
         {
             debug!("{self:?}: retire: pausing...");
-            self.as_mut().pause().await?;
+            let res = self.as_mut().pause().await;
             debug!("{self:?}: retire: pausing ok");
 
+            // Disconnect the previously detached device handles. This has to be
+            // done after the pause to prevent an I/O race.
+            self.disconnect_detached_devices().await;
+
+            res?;
+
             self.child_retire_destroy_device(&dev).await;
 
             debug!("{self:?}: resuming...");
@@ -1055,20 +1066,39 @@ impl<'n> Nexus<'n> {
         Ok(())
     }
 
-    /// Disconnects a device from all I/O channels.
-    pub(crate) async fn disconnect_device(&self, dev: &str) {
+    /// Detaches the device's handles from all I/O channels.
+    ///
+    /// The detached handles must be disconnected and dropped by a
+    /// `disconnect_detached_devices()` call.
+    ///
+    /// Device disconnection is done in two steps (detach, than disconnect) in
+    /// order to prevent an I/O race when retiring a device.
+    pub(crate) async fn detach_device(&self, dev: &str) {
         if !self.has_io_device {
             return;
         }
 
-        debug!("{self:?}: disconnecting '{dev}' from all channels ...");
+        debug!("{self:?}: detaching '{dev}' from all channels...");
 
         self.traverse_io_channels_async(dev, |channel, dev| {
-            channel.disconnect_device(dev);
+            channel.detach_device(dev);
+        })
+        .await;
+
+        debug!("{self:?}: '{dev}' detached from all I/O channels");
+    }
+
+    /// Disconnects all the detached devices on all I/O channels by dropping
+    /// their handles.
+    pub(crate) async fn disconnect_detached_devices(&self) {
+        debug!("{self:?}: disconnecting all detached devices ...");
+
+        self.traverse_io_channels_async((), |channel, _| {
+            channel.disconnect_detached_devices();
         })
         .await;
 
-        debug!("{self:?}: '{dev}' disconnected from all I/O channels");
+        debug!("{self:?}: disconnected all detached devices");
     }
 
     /// Destroys the device being retired.
@@ -1143,7 +1173,8 @@ impl<'n> Nexus<'n> {
 
                 // Step 1: Close I/O channels for all children.
                 for dev in nexus.child_devices() {
-                    nexus.disconnect_device(&dev).await;
+                    nexus.detach_device(&dev).await;
+                    nexus.disconnect_detached_devices().await;
 
                     device_cmd_queue().enqueue(DeviceCommand::RetireDevice {
                         nexus_name: nexus.name.clone(),

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

@@ -17,6 +17,7 @@ use spdk_rs::Thread;
 pub struct NexusChannel<'n> {
     writers: Vec<Box<dyn BlockDeviceHandle>>,
     readers: Vec<Box<dyn BlockDeviceHandle>>,
+    detached: Vec<Box<dyn BlockDeviceHandle>>,
     io_logs: Vec<IOLogChannel>,
     previous_reader: UnsafeCell<usize>,
     fail_fast: u32,
@@ -123,6 +124,7 @@ impl<'n> NexusChannel<'n> {
         Self {
             writers,
             readers,
+            detached: Vec::new(),
             io_logs: nexus.io_log_channels(),
             previous_reader: UnsafeCell::new(0),
             nexus: unsafe { nexus.pinned_mut() },
@@ -209,16 +211,43 @@ impl<'n> NexusChannel<'n> {
         }
     }
 
-    /// Disconnects a child device from the I/O path.
-    pub fn disconnect_device(&mut self, device_name: &str) {
+    /// Detaches a child device from this I/O channel, moving the device's
+    /// handles to the list of detached devices to disconnect later.
+    ///
+    /// The detached handles must be disconnected and dropped by a
+    /// `disconnect_detached_devices()` call.
+    pub(super) fn detach_device(&mut self, device_name: &str) {
         self.previous_reader = UnsafeCell::new(0);
 
-        self.readers
-            .retain(|c| c.get_device().device_name() != device_name);
-        self.writers
-            .retain(|c| c.get_device().device_name() != device_name);
+        if let Some(d) = self
+            .readers
+            .iter()
+            .position(|c| c.get_device().device_name() == device_name)
+        {
+            let t = self.readers.remove(d);
+            self.detached.push(t);
+        }
+
+        if let Some(d) = self
+            .writers
+            .iter()
+            .position(|c| c.get_device().device_name() == device_name)
+        {
+            let t = self.writers.remove(d);
+            self.detached.push(t);
+        }
 
-        debug!("{self:?}: device '{device_name}' disconnected");
+        debug!("{self:?}: device '{device_name}' detached");
+    }
+
+    /// Disconnects all previously detached device handles by dropping them.
+    pub(super) fn disconnect_detached_devices(&mut self) {
+        debug!(
+            "{self:?}: disconnecting {n} detached device handles...",
+            n = self.detached.len()
+        );
+        self.detached.clear();
+        debug!("{self:?}: all detached device handles disconnected");
     }
 
     /// Refreshing our channels simply means that we either have a child going

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

@@ -642,7 +642,8 @@ impl<'n> NexusBio<'n> {
             // set the IO as failed in the submission stage.
             self.ctx_mut().failed += 1;
 
-            self.channel_mut().disconnect_device(&device);
+            self.channel_mut().detach_device(&device);
+            self.channel_mut().disconnect_detached_devices();
 
             if let Some(log) = self.fault_device(
                 &device,