Various fixes for idempotency, error messages and concurrency

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

@@ -131,11 +131,7 @@ pub enum Error {
         name
     ))]
     RemoveLastHealthyChild { child: String, name: String },
-    #[snafu(display(
-        "Cannot remove or offline the last healthy child {} of nexus {}",
-        child,
-        name
-    ))]
+    #[snafu(display("Child {} of nexus {} not found", child, name))]
     ChildNotFound { child: String, name: String },
     #[snafu(display("Child {} of nexus {} is not open", child, name))]
     ChildDeviceNotOpen { child: String, name: String },

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

@@ -255,7 +255,9 @@ impl<'n> From<&Nexus<'n>> for NexusPtpl {
 impl PtplFileOps for NexusPtpl {
     fn destroy(&self) -> Result<(), std::io::Error> {
         if let Some(path) = self.path() {
-            std::fs::remove_file(path)?;
+            if path.exists() {
+                std::fs::remove_file(path)?;
+            }
         }
         Ok(())
     }

io-engine/src/bdev_api.rs

@@ -80,10 +80,7 @@ pub enum BdevError {
     #[snafu(display("BDEV '{}' could not be found", name))]
     BdevNotFound { name: String },
     // Invalid creation parameters.
-    #[snafu(display(
-        "Failed to create a BDEV '{}': invalid parameters",
-        name
-    ))]
+    #[snafu(display("Failed to create a BDEV '{}'", name))]
     CreateBdevInvalidParams { source: Errno, name: String },
     // Generic creation failure.
     #[snafu(display("Failed to create a BDEV '{}'", name))]

io-engine/src/bin/io-engine-client/v0/nexus_cli.rs

@@ -708,17 +708,18 @@ async fn nexus_publish(
         .get_one::<String>("key")
         .cloned()
         .unwrap_or_default();
-    let protocol = match matches.get_one::<&str>("protocol") {
-        None => v0::ShareProtocolNexus::NexusNbd,
-        Some(&"nvmf") => v0::ShareProtocolNexus::NexusNvmf,
-        Some(_) => {
-            return Err(Status::new(
-                Code::Internal,
-                "Invalid value of share protocol".to_owned(),
-            ))
-            .context(GrpcStatus);
-        }
-    };
+    let protocol =
+        match matches.get_one::<String>("protocol").map(|s| s.as_str()) {
+            None => v0::ShareProtocolNexus::NexusNbd,
+            Some("nvmf") => v0::ShareProtocolNexus::NexusNvmf,
+            Some(_) => {
+                return Err(Status::new(
+                    Code::Internal,
+                    "Invalid value of share protocol".to_owned(),
+                ))
+                .context(GrpcStatus);
+            }
+        };
     let allowed_hosts = matches
         .get_many::<String>("allowed-host")
         .unwrap_or_default()

io-engine/src/bin/io-engine-client/v1/nexus_cli.rs

@@ -579,17 +579,18 @@ async fn nexus_publish(
         .cloned()
         .unwrap_or_default();
 
-    let protocol = match matches.get_one::<&str>("protocol") {
-        None => v1::common::ShareProtocol::Nvmf as i32,
-        Some(&"nvmf") => v1::common::ShareProtocol::Nvmf as i32,
-        Some(_) => {
-            return Err(Status::new(
-                Code::Internal,
-                "Invalid value of share protocol".to_owned(),
-            ))
-            .context(GrpcStatus);
-        }
-    };
+    let protocol =
+        match matches.get_one::<String>("protocol").map(|s| s.as_str()) {
+            None => v1::common::ShareProtocol::Nvmf as i32,
+            Some("nvmf") => v1::common::ShareProtocol::Nvmf as i32,
+            Some(_) => {
+                return Err(Status::new(
+                    Code::Internal,
+                    "Invalid value of share protocol".to_owned(),
+                ))
+                .context(GrpcStatus);
+            }
+        };
     let allowed_hosts = matches
         .get_many::<String>("allowed-host")
         .unwrap_or_default()

io-engine/src/bin/io-engine-client/v1/test_cli.rs

@@ -183,10 +183,14 @@ async fn replica_wipe(
         .map_err(|e| Status::invalid_argument(e.to_string()))
         .context(GrpcStatus)?;
 
-    let chunk_size =
-        parse_size(matches.get_one::<&str>("chunk-size").unwrap_or(&"0"))
-            .map_err(|s| Status::invalid_argument(format!("Bad size '{s}'")))
-            .context(GrpcStatus)?;
+    let chunk_size = parse_size(
+        matches
+            .get_one::<String>("chunk-size")
+            .map(|s| s.as_str())
+            .unwrap_or("0"),
+    )
+    .map_err(|s| Status::invalid_argument(format!("Bad size '{s}'")))
+    .context(GrpcStatus)?;
     let response = ctx
         .v1
         .test

io-engine/src/grpc/mod.rs

@@ -12,7 +12,7 @@ use tonic::{Request, Response, Status};
 
 use crate::{
     bdev_api::BdevError,
-    core::{CoreError, Reactor},
+    core::{CoreError, Reactor, VerboseError},
 };
 
 impl From<BdevError> for tonic::Status {
@@ -33,15 +33,28 @@ impl From<BdevError> for tonic::Status {
             BdevError::BoolParamParseFailed {
                 ..
             } => Status::invalid_argument(e.to_string()),
-            BdevError::CreateBdevInvalidParams {
+            BdevError::UuidParamParseFailed {
+                ..
+            } => Status::invalid_argument(e.to_string()),
+            BdevError::BdevWrongUuid {
+                ..
+            } => Status::invalid_argument(e.to_string()),
+            BdevError::CreateBdevFailed {
+                source, ..
+            }
+            | BdevError::CreateBdevInvalidParams {
                 source, ..
             } => match source {
-                Errno::EINVAL => Status::invalid_argument(e.to_string()),
-                Errno::ENOENT => Status::not_found(e.to_string()),
-                Errno::EEXIST => Status::already_exists(e.to_string()),
-                _ => Status::invalid_argument(e.to_string()),
+                Errno::EINVAL => Status::invalid_argument(e.verbose()),
+                Errno::ENOENT => Status::not_found(e.verbose()),
+                Errno::ENODEV => Status::not_found(e.verbose()),
+                Errno::EEXIST => Status::already_exists(e.verbose()),
+                _ => Status::invalid_argument(e.verbose()),
             },
-            e => Status::internal(e.to_string()),
+            BdevError::BdevNotFound {
+                ..
+            } => Status::not_found(e.to_string()),
+            e => Status::internal(e.verbose()),
         }
     }
 }
@@ -107,6 +120,12 @@ pub(crate) trait Serializer<F, T> {
     async fn locked(&self, ctx: GrpcClientContext, f: F) -> Result<T, Status>;
 }
 
+#[async_trait::async_trait]
+pub(crate) trait RWSerializer<F, T> {
+    async fn locked(&self, ctx: GrpcClientContext, f: F) -> Result<T, Status>;
+    async fn shared(&self, ctx: GrpcClientContext, f: F) -> Result<T, Status>;
+}
+
 pub type GrpcResult<T> = std::result::Result<Response<T>, Status>;
 
 /// call the given future within the context of the reactor on the first core

io-engine/src/grpc/server.rs

@@ -93,10 +93,14 @@ impl MayastorGrpcServer {
                 v1::replica::ReplicaRpcServer::new(replica_v1.clone())
             }))
             .add_optional_service(enable_v1.map(|_| {
-                v1::test::TestRpcServer::new(TestService::new(replica_v1))
+                v1::test::TestRpcServer::new(TestService::new(
+                    replica_v1.clone(),
+                ))
             }))
             .add_optional_service(enable_v1.map(|_| {
-                v1::snapshot::SnapshotRpcServer::new(SnapshotService::new())
+                v1::snapshot::SnapshotRpcServer::new(SnapshotService::new(
+                    replica_v1,
+                ))
             }))
             .add_optional_service(enable_v1.map(|_| {
                 v1::host::HostRpcServer::new(HostService::new(

io-engine/src/grpc/v1/replica.rs

@@ -11,7 +11,7 @@ use crate::{
         UntypedBdev,
         UpdateProps,
     },
-    grpc::{rpc_submit, GrpcClientContext, GrpcResult, Serializer},
+    grpc::{rpc_submit, GrpcClientContext, GrpcResult, RWSerializer},
     lvs::{Error as LvsError, Lvol, LvolSpaceUsage, Lvs, LvsLvol},
 };
 use ::function_name::named;
@@ -26,17 +26,17 @@ pub struct ReplicaService {
     #[allow(unused)]
     name: String,
     client_context:
-        std::sync::Arc<tokio::sync::Mutex<Option<GrpcClientContext>>>,
+        std::sync::Arc<tokio::sync::RwLock<Option<GrpcClientContext>>>,
 }
 
 #[async_trait::async_trait]
-impl<F, T> Serializer<F, T> for ReplicaService
+impl<F, T> RWSerializer<F, T> for ReplicaService
 where
     T: Send + 'static,
     F: core::future::Future<Output = Result<T, Status>> + Send + 'static,
 {
     async fn locked(&self, ctx: GrpcClientContext, f: F) -> Result<T, Status> {
-        let mut context_guard = self.client_context.lock().await;
+        let mut context_guard = self.client_context.write().await;
 
         // Store context as a marker of to detect abnormal termination of the
         // request. Even though AssertUnwindSafe() allows us to
@@ -57,6 +57,27 @@ where
         // Request completed, remove the marker.
         let ctx = context_guard.take().expect("gRPC context disappeared");
 
+        match r {
+            Ok(r) => r,
+            Err(_e) => {
+                warn!("{}: gRPC method panicked, args: {}", ctx.id, ctx.args);
+                Err(Status::cancelled(format!(
+                    "{}: gRPC method panicked",
+                    ctx.id
+                )))
+            }
+        }
+    }
+    async fn shared(&self, ctx: GrpcClientContext, f: F) -> Result<T, Status> {
+        let context_guard = self.client_context.read().await;
+
+        if let Some(c) = context_guard.as_ref() {
+            warn!("{}: gRPC method timed out, args: {}", c.id, c.args);
+        }
+
+        let fut = AssertUnwindSafe(f).catch_unwind();
+        let r = fut.await;
+
         match r {
             Ok(r) => r,
             Err(_e) => {
@@ -120,7 +141,7 @@ impl ReplicaService {
     pub fn new() -> Self {
         Self {
             name: String::from("ReplicaSvc"),
-            client_context: std::sync::Arc::new(tokio::sync::Mutex::new(None)),
+            client_context: std::sync::Arc::new(tokio::sync::RwLock::new(None)),
         }
     }
 }
@@ -311,7 +332,7 @@ impl ReplicaRpc for ReplicaService {
         &self,
         request: Request<ListReplicaOptions>,
     ) -> GrpcResult<ListReplicasResponse> {
-        self.locked(GrpcClientContext::new(&request, function_name!()), async {
+        self.shared(GrpcClientContext::new(&request, function_name!()), async {
             let args = request.into_inner();
             trace!("{:?}", args);
             let rx = rpc_submit::<_, _, LvsError>(async move {

io-engine/src/grpc/v1/snapshot.rs

@@ -21,7 +21,7 @@ use crate::{
         v1::nexus::nexus_lookup,
         GrpcClientContext,
         GrpcResult,
-        Serializer,
+        RWSerializer,
     },
     lvs::{Error as LvsError, Lvol, Lvs, LvsLvol},
     spdk_rs::ffihelper::IntoCString,
@@ -45,7 +45,7 @@ const SNAPSHOT_READY_AS_SOURCE: bool = false;
 #[allow(dead_code)]
 pub struct SnapshotService {
     name: String,
-    client_context: tokio::sync::Mutex<Option<GrpcClientContext>>,
+    replica_svc: super::replica::ReplicaService,
 }
 
 #[derive(Debug)]
@@ -201,55 +201,24 @@ impl From<VolumeSnapshotDescriptor> for SnapshotInfo {
     }
 }
 #[async_trait::async_trait]
-impl<F, T> Serializer<F, T> for SnapshotService
+impl<F, T> RWSerializer<F, T> for SnapshotService
 where
     T: Send + 'static,
     F: core::future::Future<Output = Result<T, Status>> + Send + 'static,
 {
     async fn locked(&self, ctx: GrpcClientContext, f: F) -> Result<T, Status> {
-        let mut context_guard = self.client_context.lock().await;
-
-        // Store context as a marker of to detect abnormal termination of the
-        // request. Even though AssertUnwindSafe() allows us to
-        // intercept asserts in underlying method strategies, such a
-        // situation can still happen when the high-level future that
-        // represents gRPC call at the highest level (i.e. the one created
-        // by gRPC server) gets cancelled (due to timeout or somehow else).
-        // This can't be properly intercepted by 'locked' function itself in the
-        // first place, so the state needs to be cleaned up properly
-        // upon subsequent gRPC calls.
-        if let Some(c) = context_guard.replace(ctx) {
-            warn!("{}: gRPC method timed out, args: {}", c.id, c.args);
-        }
-
-        let fut = AssertUnwindSafe(f).catch_unwind();
-        let r = fut.await;
-
-        // Request completed, remove the marker.
-        let ctx = context_guard.take().expect("gRPC context disappeared");
-
-        match r {
-            Ok(r) => r,
-            Err(_e) => {
-                warn!("{}: gRPC method panicked, args: {}", ctx.id, ctx.args);
-                Err(Status::cancelled(format!(
-                    "{}: gRPC method panicked",
-                    ctx.id
-                )))
-            }
-        }
+        self.replica_svc.locked(ctx, f).await
     }
-}
-impl Default for SnapshotService {
-    fn default() -> Self {
-        Self::new()
+    async fn shared(&self, ctx: GrpcClientContext, f: F) -> Result<T, Status> {
+        self.replica_svc.shared(ctx, f).await
     }
 }
+
 impl SnapshotService {
-    pub fn new() -> Self {
+    pub fn new(replica_svc: super::replica::ReplicaService) -> Self {
         Self {
             name: String::from("SnapshotSvc"),
-            client_context: tokio::sync::Mutex::new(None),
+            replica_svc,
         }
     }
     async fn serialized<T, F>(
@@ -490,7 +459,7 @@ impl SnapshotRpc for SnapshotService {
         &self,
         request: Request<ListSnapshotsRequest>,
     ) -> GrpcResult<ListSnapshotsResponse> {
-        self.locked(
+        self.shared(
             GrpcClientContext::new(&request, function_name!()),
             async move {
                 let args = request.into_inner();
@@ -739,7 +708,7 @@ impl SnapshotRpc for SnapshotService {
         &self,
         request: Request<ListSnapshotCloneRequest>,
     ) -> GrpcResult<ListSnapshotCloneResponse> {
-        self.locked(
+        self.shared(
             GrpcClientContext::new(&request, function_name!()),
             async move {
                 let args = request.into_inner();

io-engine/src/grpc/v1/test.rs

@@ -5,7 +5,7 @@ use crate::{
         Bdev,
         VerboseError,
     },
-    grpc::{rpc_submit, GrpcClientContext, GrpcResult, Serializer},
+    grpc::{rpc_submit, GrpcClientContext, GrpcResult, RWSerializer},
     lvs::{Error as LvsError, Lvol, Lvs, LvsLvol},
 };
 use ::function_name::named;
@@ -74,7 +74,7 @@ impl TestRpc for TestService {
 
         crate::core::spawn(async move {
             let result = replica_svc
-                .locked(
+                .shared(
                     GrpcClientContext::new(&request, function_name!()),
                     async move {
                         let args = request.into_inner();

io-engine/src/lvs/lvs_store.rs

@@ -906,7 +906,9 @@ impl From<&Lvs> for LvsPtpl {
 impl PtplFileOps for LvsPtpl {
     fn destroy(&self) -> Result<(), std::io::Error> {
         if let Some(path) = self.path() {
-            std::fs::remove_dir_all(path)?;
+            if path.exists() {
+                std::fs::remove_dir_all(path)?;
+            }
         }
         Ok(())
     }