Merge #1775

1775: Backport fixes for release/2.7 r=tiagolobocastro a=tiagolobocastro test(bdd): make nvme controller usage more robust Caters for when the device is /dev/nvmeX but X not the same as the controller! Signed-off-by: Tiago Castro <[email protected]> --- Merge #1755 1755: Reuse Rebuild IO handles r=tiagolobocastro a=tiagolobocastro fix(rebuild): reuse rebuild IO handles Reuses the rebuild IO handles, rather than attempting to allocate them per rebuild task. The main issue with handle allocation on the fly is that the target may have not cleaned up a previous IO qpair connection, and so the connect may fail. We started seeing this more on CI because we forgot to cherry-pick a commit increasing the retry delay. However, after inspecting a bunch of user support bundles I see that we still have occasional connect errors. Rather than increasing the timeout, we attempt here to reuse the handles, thus avoid the problem almost entirely. Signed-off-by: Tiago Castro <[email protected]> --- refactor(rebuild): rebuild completion is not an error When the rebuild has been complete, if we wait for it this fails because the channels are not longer available. Instead, simply return the rebuild state, since this is what we want anyway. Signed-off-by: Tiago Castro <[email protected]> Co-authored-by: Tiago Castro <[email protected]> --- fix: check valid sct and sc combinations for pi error Signed-off-by: Diwakar Sharma <[email protected]> --- fix: use auto-detected sector size for blockdev This fixes the behaviour where we pass 512 as sector size if the disk uri doesn't contain blk_size parameter. This causes pool creation failure if the underlying disk has a different sector size e.g. 4096. Instead of passing 512, we now pass 0 which lets spdk detect the device's sector size and use that value. Signed-off-by: Diwakar Sharma <[email protected]> Co-authored-by: Diwakar Sharma <[email protected]> Co-authored-by: mayastor-bors <[email protected]> Co-authored-by: Tiago Castro <[email protected]>
openebs · Nov 27, 2024 · 0d8ac9c · 0d8ac9c
2 parents 40572c9 + ece9dd0
commit 0d8ac9c
Show file tree

Hide file tree

Showing 18 changed files with 321 additions and 79 deletions.
diff --git a/.gitmodules b/.gitmodules
@@ -1,6 +1,6 @@
 [submodule "spdk-rs"]
 	path = spdk-rs
-	url = https://github.com/openebs/spdk-rs
+	url = ../spdk-rs.git
 	branch = release/2.7
 [submodule "utils/dependencies"]
 	path = utils/dependencies

diff --git a/Cargo.lock b/Cargo.lock
diff --git a/io-engine-tests/src/lib.rs b/io-engine-tests/src/lib.rs
@@ -188,6 +188,28 @@ pub fn truncate_file_bytes(path: &str, size: u64) {
     assert!(output.status.success());
 }
 
+/// Automatically assign a loopdev to path
+pub fn setup_loopdev_file(path: &str, sector_size: Option<u64>) -> String {
+    let log_sec = sector_size.unwrap_or(512);
+
+    let output = Command::new("losetup")
+        .args(["-f", "--show", "-b", &format!("{log_sec}"), path])
+        .output()
+        .expect("failed exec losetup");
+    assert!(output.status.success());
+    // return the assigned loop device
+    String::from_utf8(output.stdout).unwrap().trim().to_string()
+}
+
+/// Detach the provided loop device.
+pub fn detach_loopdev(dev: &str) {
+    let output = Command::new("losetup")
+        .args(["-d", dev])
+        .output()
+        .expect("failed exec losetup");
+    assert!(output.status.success());
+}
+
 pub fn fscheck(device: &str) {
     let output = Command::new("fsck")
         .args([device, "-n"])

diff --git a/io-engine/Cargo.toml b/io-engine/Cargo.toml
@@ -70,7 +70,7 @@ libc = "0.2.149"
 log = "0.4.20"
 md5 = "0.7.0"
 merge = "0.1.0"
-nix = { version = "0.27.1", default-features = false, features = [ "hostname", "net", "socket", "ioctl" ] }
+nix = { version = "0.27.1", default-features = false, features = ["hostname", "net", "socket", "ioctl"] }
 once_cell = "1.18.0"
 parking_lot = "0.12.1"
 pin-utils = "0.1.0"
@@ -98,9 +98,10 @@ async-process = { version = "1.8.1" }
 rstack = { version = "0.3.3" }
 tokio-stream = "0.1.14"
 rustls = "0.21.12"
+either = "1.9.0"
 
 devinfo = { path = "../utils/dependencies/devinfo" }
-jsonrpc = { path = "../jsonrpc"}
+jsonrpc = { path = "../jsonrpc" }
 io-engine-api = { path = "../utils/dependencies/apis/io-engine" }
 spdk-rs = { path = "../spdk-rs" }
 sysfs = { path = "../sysfs" }

diff --git a/io-engine/src/bdev/aio.rs b/io-engine/src/bdev/aio.rs
@@ -3,6 +3,7 @@ use std::{
     convert::TryFrom,
     ffi::CString,
     fmt::{Debug, Formatter},
+    os::unix::fs::FileTypeExt,
 };
 
 use async_trait::async_trait;
@@ -29,7 +30,7 @@ pub(super) struct Aio {
 
 impl Debug for Aio {
     fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
-        write!(f, "Aio '{}'", self.name)
+        write!(f, "Aio '{}', 'blk_size: {}'", self.name, self.blk_size)
     }
 }
 
@@ -47,6 +48,10 @@ impl TryFrom<&Url> for Aio {
             });
         }
 
+        let path_is_blockdev = std::fs::metadata(url.path())
+            .ok()
+            .map_or(false, |meta| meta.file_type().is_block_device());
+
         let mut parameters: HashMap<String, String> =
             url.query_pairs().into_owned().collect();
 
@@ -58,9 +63,14 @@ impl TryFrom<&Url> for Aio {
                     value: value.clone(),
                 })?
             }
-            None => 512,
+            None => {
+                if path_is_blockdev {
+                    0
+                } else {
+                    512
+                }
+            }
         };
-
         let uuid = uri::uuid(parameters.remove("uuid")).context(
             bdev_api::UuidParamParseFailed {
                 uri: url.to_string(),

diff --git a/io-engine/src/bdev/nexus/nexus_bdev_children.rs b/io-engine/src/bdev/nexus/nexus_bdev_children.rs
@@ -1152,10 +1152,11 @@ impl<'n> Nexus<'n> {
         // Cancel rebuild job for this child, if any.
         if let Some(job) = child.rebuild_job() {
             debug!("{self:?}: retire: stopping rebuild job...");
-            let terminated = job.force_fail();
-            Reactors::master().send_future(async move {
-                terminated.await.ok();
-            });
+            if let either::Either::Left(terminated) = job.force_fail() {
+                Reactors::master().send_future(async move {
+                    terminated.await.ok();
+                });
+            }
         }
 
         debug!("{child:?}: retire: enqueuing device '{dev}' to retire");

diff --git a/io-engine/src/bdev/nexus/nexus_bdev_rebuild.rs b/io-engine/src/bdev/nexus/nexus_bdev_rebuild.rs
@@ -247,15 +247,18 @@ impl<'n> Nexus<'n> {
     async fn terminate_rebuild(&self, child_uri: &str) {
         // If a rebuild job is not found that's ok
         // as we were just going to remove it anyway.
-        if let Ok(rj) = self.rebuild_job_mut(child_uri) {
-            let ch = rj.force_stop();
-            if let Err(e) = ch.await {
-                error!(
-                    "Failed to wait on rebuild job for child {child_uri} \
+        let Ok(rj) = self.rebuild_job_mut(child_uri) else {
+            return;
+        };
+        let either::Either::Left(ch) = rj.force_stop() else {
+            return;
+        };
+        if let Err(e) = ch.await {
+            error!(
+                "Failed to wait on rebuild job for child {child_uri} \
                     to terminate with error {}",
-                    e.verbose()
-                );
-            }
+                e.verbose()
+            );
         }
     }
 
@@ -355,6 +358,9 @@ impl<'n> Nexus<'n> {
 
         // wait for the jobs to complete terminating
         for job in terminated_jobs {
+            let either::Either::Left(job) = job else {
+                continue;
+            };
             if let Err(e) = job.await {
                 error!(
                     "{:?}: error when waiting for the rebuild job \

diff --git a/io-engine/src/bdev/nvmx/utils.rs b/io-engine/src/bdev/nvmx/utils.rs
@@ -40,8 +40,9 @@ pub enum NvmeAerInfoNvmCommandSet {
 
 /// Check if the Completion Queue Entry indicates abnormal termination of
 /// request due to any of the following conditions:
-///   - Any media specific errors that occur in the NVM or data integrity type
-///     errors.
+///   - An Status Code Type(SCT) of media specific errors that occur in the NVM
+///     or data integrity type errors, AND a Status Code(SC) value pertaining to
+///     one of the below:
 ///   - The command was aborted due to an end-to-end guard check failure.
 ///   - The command was aborted due to an end-to-end application tag check
 ///     failure.
@@ -59,9 +60,9 @@ pub(crate) fn nvme_cpl_is_pi_error(cpl: *const spdk_nvme_cpl) -> bool {
     }
 
     sct == NvmeStatusCodeType::MediaError as u16
-        || sc == NvmeMediaErrorStatusCode::Guard as u16
-        || sc == NvmeMediaErrorStatusCode::ApplicationTag as u16
-        || sc == NvmeMediaErrorStatusCode::ReferenceTag as u16
+        && (sc == NvmeMediaErrorStatusCode::Guard as u16
+            || sc == NvmeMediaErrorStatusCode::ApplicationTag as u16
+            || sc == NvmeMediaErrorStatusCode::ReferenceTag as u16)
 }
 
 #[inline]

diff --git a/io-engine/src/bdev/uring.rs b/io-engine/src/bdev/uring.rs
@@ -1,4 +1,9 @@
-use std::{collections::HashMap, convert::TryFrom, ffi::CString};
+use std::{
+    collections::HashMap,
+    convert::TryFrom,
+    ffi::CString,
+    os::unix::fs::FileTypeExt,
+};
 
 use async_trait::async_trait;
 use futures::channel::oneshot;
@@ -36,6 +41,10 @@ impl TryFrom<&Url> for Uring {
             });
         }
 
+        let path_is_blockdev = std::fs::metadata(url.path())
+            .ok()
+            .map_or(false, |meta| meta.file_type().is_block_device());
+
         let mut parameters: HashMap<String, String> =
             url.query_pairs().into_owned().collect();
 
@@ -47,7 +56,13 @@ impl TryFrom<&Url> for Uring {
                     value: value.clone(),
                 })?
             }
-            None => 512,
+            None => {
+                if path_is_blockdev {
+                    0
+                } else {
+                    512
+                }
+            }
         };
 
         let uuid = uri::uuid(parameters.remove("uuid")).context(

diff --git a/io-engine/src/grpc/v1/snapshot_rebuild.rs b/io-engine/src/grpc/v1/snapshot_rebuild.rs
@@ -110,7 +110,10 @@ impl SnapshotRebuildRpc for SnapshotRebuildService {
             let Ok(job) = SnapshotRebuildJob::lookup(&args.uuid) else {
                 return Err(tonic::Status::not_found(""));
             };
-            let rx = job.force_stop().await.ok();
+            let rx = match job.force_stop() {
+                either::Either::Left(chan) => chan.await,
+                either::Either::Right(stopped) => Ok(stopped),
+            };
             info!("Snapshot Rebuild stopped: {rx:?}");
             job.destroy();
             Ok(())

diff --git a/io-engine/src/rebuild/mod.rs b/io-engine/src/rebuild/mod.rs
@@ -57,7 +57,10 @@ impl WithinRange<u64> for std::ops::Range<u64> {
 /// Shutdown all pending snapshot rebuilds.
 pub(crate) async fn shutdown_snapshot_rebuilds() {
     let jobs = SnapshotRebuildJob::list().into_iter();
-    for recv in jobs.map(|job| job.force_stop()).collect::<Vec<_>>() {
+    for recv in jobs
+        .flat_map(|job| job.force_stop().left())
+        .collect::<Vec<_>>()
+    {
         recv.await.ok();
     }
 }

diff --git a/io-engine/src/rebuild/rebuild_descriptor.rs b/io-engine/src/rebuild/rebuild_descriptor.rs
@@ -48,9 +48,11 @@ pub(super) struct RebuildDescriptor {
     /// Pre-opened descriptor for the source block device.
     #[allow(clippy::non_send_fields_in_send_ty)]
     pub(super) src_descriptor: Box<dyn BlockDeviceDescriptor>,
+    pub(super) src_handle: Box<dyn BlockDeviceHandle>,
     /// Pre-opened descriptor for destination block device.
     #[allow(clippy::non_send_fields_in_send_ty)]
     pub(super) dst_descriptor: Box<dyn BlockDeviceDescriptor>,
+    pub(super) dst_handle: Box<dyn BlockDeviceHandle>,
     /// Start time of this rebuild.
     pub(super) start_time: DateTime<Utc>,
 }
@@ -90,9 +92,8 @@ impl RebuildDescriptor {
             });
         }
 
-        let source_hdl = RebuildDescriptor::io_handle(&*src_descriptor).await?;
-        let destination_hdl =
-            RebuildDescriptor::io_handle(&*dst_descriptor).await?;
+        let src_handle = RebuildDescriptor::io_handle(&*src_descriptor).await?;
+        let dst_handle = RebuildDescriptor::io_handle(&*dst_descriptor).await?;
 
         let range = match range {
             None => {
@@ -105,8 +106,8 @@ impl RebuildDescriptor {
         };
 
         if !Self::validate(
-            source_hdl.get_device(),
-            destination_hdl.get_device(),
+            src_handle.get_device(),
+            dst_handle.get_device(),
             &range,
         ) {
             return Err(RebuildError::InvalidSrcDstRange {});
@@ -123,7 +124,9 @@ impl RebuildDescriptor {
             block_size,
             segment_size_blks,
             src_descriptor,
+            src_handle,
             dst_descriptor,
+            dst_handle,
             start_time: Utc::now(),
         })
     }
@@ -173,18 +176,14 @@ impl RebuildDescriptor {
 
     /// Get a `BlockDeviceHandle` for the source.
     #[inline(always)]
-    pub(super) async fn src_io_handle(
-        &self,
-    ) -> Result<Box<dyn BlockDeviceHandle>, RebuildError> {
-        Self::io_handle(&*self.src_descriptor).await
+    pub(super) fn src_io_handle(&self) -> &dyn BlockDeviceHandle {
+        self.src_handle.as_ref()
     }
 
     /// Get a `BlockDeviceHandle` for the destination.
     #[inline(always)]
-    pub(super) async fn dst_io_handle(
-        &self,
-    ) -> Result<Box<dyn BlockDeviceHandle>, RebuildError> {
-        Self::io_handle(&*self.dst_descriptor).await
+    pub(super) fn dst_io_handle(&self) -> &dyn BlockDeviceHandle {
+        self.dst_handle.as_ref()
     }
 
     /// Get a `BlockDeviceHandle` for the given block device descriptor.
@@ -231,7 +230,6 @@ impl RebuildDescriptor {
     ) -> Result<bool, RebuildError> {
         match self
             .src_io_handle()
-            .await?
             .readv_blocks_async(
                 iovs,
                 offset_blk,
@@ -269,7 +267,6 @@ impl RebuildDescriptor {
         iovs: &[IoVec],
     ) -> Result<(), RebuildError> {
         self.dst_io_handle()
-            .await?
             .writev_blocks_async(
                 iovs,
                 offset_blk,
@@ -291,7 +288,6 @@ impl RebuildDescriptor {
     ) -> Result<(), RebuildError> {
         // Read the source again.
         self.src_io_handle()
-            .await?
             .readv_blocks_async(
                 iovs,
                 offset_blk,
@@ -306,7 +302,6 @@ impl RebuildDescriptor {
 
         match self
             .dst_io_handle()
-            .await?
             .comparev_blocks_async(
                 iovs,
                 offset_blk,

diff --git a/io-engine/src/rebuild/rebuild_job.rs b/io-engine/src/rebuild/rebuild_job.rs
@@ -163,13 +163,17 @@ impl RebuildJob {
 
     /// Forcefully stops the job, overriding any pending client operation
     /// returns an async channel which can be used to await for termination.
-    pub(crate) fn force_stop(&self) -> oneshot::Receiver<RebuildState> {
+    pub(crate) fn force_stop(
+        &self,
+    ) -> either::Either<oneshot::Receiver<RebuildState>, RebuildState> {
         self.force_terminate(RebuildOperation::Stop)
     }
 
     /// Forcefully fails the job, overriding any pending client operation
     /// returns an async channel which can be used to await for termination.
-    pub(crate) fn force_fail(&self) -> oneshot::Receiver<RebuildState> {
+    pub(crate) fn force_fail(
+        &self,
+    ) -> either::Either<oneshot::Receiver<RebuildState>, RebuildState> {
         self.force_terminate(RebuildOperation::Fail)
     }
 
@@ -179,10 +183,13 @@ impl RebuildJob {
     fn force_terminate(
         &self,
         op: RebuildOperation,
-    ) -> oneshot::Receiver<RebuildState> {
+    ) -> either::Either<oneshot::Receiver<RebuildState>, RebuildState> {
         self.exec_internal_op(op).ok();
-        self.add_completion_listener()
-            .unwrap_or_else(|_| oneshot::channel().1)
+
+        match self.add_completion_listener() {
+            Ok(chan) => either::Either::Left(chan),
+            Err(_) => either::Either::Right(self.state()),
+        }
     }
 
     /// Get the rebuild stats.