fix(grpc-lock): serialized read/write lock for grpc request

Signed-off-by: Hrudaya <hrudayaranjan.sahoo@datacore.com>

io-engine/src/bin/io-engine.rs

@@ -100,7 +100,9 @@ fn start_tokio_runtime(args: &MayastorCliArgs) {
 
     // Initialize Lock manager.
     let cfg = ResourceLockManagerConfig::default()
-        .with_subsystem(ProtectedSubsystems::NEXUS, 512);
+        .with_subsystem(ProtectedSubsystems::NEXUS, 512)
+        .with_subsystem(ProtectedSubsystems::REPLICA, 2048)
+        .with_subsystem(ProtectedSubsystems::POOL, 512);
     ResourceLockManager::initialize(cfg);
 
     Mthread::spawn_unaffinitized(move || {

io-engine/src/core/lock.rs

@@ -4,13 +4,16 @@ use std::{
     time::Duration,
 };
 
-use futures::lock::{Mutex, MutexGuard};
 use once_cell::sync::OnceCell;
+use std::sync::Arc;
+use tokio::sync::{RwLock, RwLockReadGuard, RwLockWriteGuard};
 
 /// Common IO engine resource subsystems.
 pub struct ProtectedSubsystems;
 impl ProtectedSubsystems {
     pub const NEXUS: &'static str = "nexus";
+    pub const REPLICA: &'static str = "replica";
+    pub const POOL: &'static str = "pool";
 }
 
 /// Configuration parameters for initialization of the Lock manager.
@@ -41,38 +44,39 @@ impl ResourceLockManagerConfig {
 }
 
 /// Resource subsystem that holds locks for all resources withing this system.
+#[derive(Clone)]
 pub struct ResourceSubsystem {
     id: String,
-    object_locks: Vec<Mutex<LockStats>>,
-    subsystem_lock: Mutex<LockStats>,
+    object_locks: Vec<Arc<RwLock<LockStats>>>,
+    subsystem_lock: Arc<RwLock<LockStats>>,
 }
 
 impl ResourceSubsystem {
     /// Create a new resource subsystem with target id and maximum number of
     /// objects.
     fn new(id: String, num_objects: usize) -> Self {
         let object_locks =
-            std::iter::repeat_with(|| Mutex::new(LockStats::default()))
+            std::iter::repeat_with(|| RwLock::new(LockStats::default()).into())
                 .take(num_objects)
                 .collect::<Vec<_>>();
 
         Self {
             id,
             object_locks,
-            subsystem_lock: Mutex::new(LockStats::default()),
+            subsystem_lock: RwLock::new(LockStats::default()).into(),
         }
     }
 
-    /// Acquire the subsystem lock.
-    pub async fn lock(
+    /// Acquire the subsystem write lock.
+    pub async fn write_lock_subsystem(
         &self,
         wait_timeout: Option<Duration>,
     ) -> Option<ResourceLockGuard<'_>> {
-        acquire_lock(&self.subsystem_lock, wait_timeout).await
+        acquire_write_lock(&self.subsystem_lock, wait_timeout).await
     }
 
-    /// Lock subsystem resource by its ID and obtain a lock guard.
-    pub async fn lock_resource<T: AsRef<str>>(
+    /// Write lock subsystem resource by its ID and obtain a lock guard.
+    pub async fn write_lock_resource<T: AsRef<str>>(
         &self,
         id: T,
         wait_timeout: Option<Duration>,
@@ -81,15 +85,35 @@ impl ResourceSubsystem {
         let mut hasher = DefaultHasher::new();
         id.as_ref().hash(&mut hasher);
         let mutex_id = hasher.finish() as usize % self.object_locks.len();
+        acquire_write_lock(&self.object_locks[mutex_id], wait_timeout).await
+    }
+
+    /// Acquire the subsystem read lock.
+    pub async fn read_lock_subsystem(
+        &self,
+        wait_timeout: Option<Duration>,
+    ) -> Option<ResourceLockGuard<'_>> {
+        acquire_read_lock(&self.subsystem_lock, wait_timeout).await
+    }
 
-        acquire_lock(&self.object_locks[mutex_id], wait_timeout).await
+    /// Read lock subsystem resource by its ID and obtain a lock guard.
+    pub async fn read_lock_resource<T: AsRef<str>>(
+        &self,
+        id: T,
+        wait_timeout: Option<Duration>,
+    ) -> Option<ResourceLockGuard<'_>> {
+        // Calculate hash of the object to get the mutex index.
+        let mut hasher = DefaultHasher::new();
+        id.as_ref().hash(&mut hasher);
+        let mutex_id = hasher.finish() as usize % self.object_locks.len();
+        acquire_read_lock(&self.object_locks[mutex_id], wait_timeout).await
     }
 }
 
 /// Structure that holds per-lock statistics.
 #[derive(Debug, Default)]
-struct LockStats {
-    num_acquires: usize,
+pub struct LockStats {
+    pub num_acquires: usize,
 }
 
 /// Lock manager which is used for protecting access to sensitive resources.
@@ -107,37 +131,53 @@ pub struct ResourceLockManager {
     /// All known resource subsystems with locks.
     subsystems: Vec<ResourceSubsystem>,
     /// Global resource lock,
-    mgr_lock: Mutex<LockStats>,
+    mgr_lock: Arc<RwLock<LockStats>>,
 }
 
-/// Automatically releases the lock once dropped.
-pub struct ResourceLockGuard<'a> {
-    _lock_guard: MutexGuard<'a, LockStats>,
+/// Resource guard for read and write which automatically released once dropped.
+pub enum ResourceLockGuard<'a> {
+    Read(RwLockReadGuard<'a, LockStats>),
+    Write(RwLockWriteGuard<'a, LockStats>),
 }
 
 /// Global instance of the resource lock manager.
 static LOCK_MANAGER: OnceCell<ResourceLockManager> = OnceCell::new();
 
-/// Helper function to abstract common lock acquisition logic.
-async fn acquire_lock(
-    lock: &Mutex<LockStats>,
+/// Helper function to aquire read lock.
+async fn acquire_read_lock(
+    lock: &Arc<RwLock<LockStats>>,
     wait_timeout: Option<Duration>,
 ) -> Option<ResourceLockGuard<'_>> {
-    let mut lock_guard = if let Some(d) = wait_timeout {
-        match tokio::time::timeout(d, lock.lock()).await {
+    let lock_guard = if let Some(d) = wait_timeout {
+        match tokio::time::timeout(d, lock.read()).await {
             Err(_) => return None,
             Ok(g) => g,
         }
     } else {
         // No timeout, wait for the lock indefinitely.
-        lock.lock().await
+        lock.read().await
     };
 
+    Some(ResourceLockGuard::Read(lock_guard))
+}
+
+/// Helper function to aquire write lock.
+async fn acquire_write_lock(
+    lock: &RwLock<LockStats>,
+    wait_timeout: Option<Duration>,
+) -> Option<ResourceLockGuard<'_>> {
+    let mut lock_guard = if let Some(d) = wait_timeout {
+        match tokio::time::timeout(d, lock.write()).await {
+            Err(_) => return None,
+            Ok(g) => g,
+        }
+    } else {
+        // No timeout, wait for the lock indefinitely.
+        lock.write().await
+    };
     lock_guard.num_acquires += 1;
 
-    Some(ResourceLockGuard {
-        _lock_guard: lock_guard,
-    })
+    Some(ResourceLockGuard::Write(lock_guard))
 }
 
 impl ResourceLockManager {
@@ -153,17 +193,25 @@ impl ResourceLockManager {
 
             ResourceLockManager {
                 subsystems,
-                mgr_lock: Mutex::new(LockStats::default()),
+                mgr_lock: RwLock::new(LockStats::default()).into(),
             }
         });
     }
 
-    /// Acquire the global Lock manager lock.
-    pub async fn lock(
+    /// Acquire the global Lock manager write lock.
+    pub async fn write_lock(
+        &self,
+        wait_timeout: Option<Duration>,
+    ) -> Option<ResourceLockGuard<'_>> {
+        acquire_write_lock(&self.mgr_lock, wait_timeout).await
+    }
+
+    /// Acquire the global Lock manager read lock.
+    pub async fn read_lock(
         &self,
         wait_timeout: Option<Duration>,
     ) -> Option<ResourceLockGuard<'_>> {
-        acquire_lock(&self.mgr_lock, wait_timeout).await
+        acquire_read_lock(&self.mgr_lock, wait_timeout).await
     }
 
     /// Get resource subsystem by its id.
@@ -185,5 +233,3 @@ impl ResourceLockManager {
         LOCK_MANAGER.get().expect("Lock Manager is not initialized")
     }
 }
-
-impl ResourceLockGuard<'_> {}

io-engine/src/core/mod.rs

@@ -56,7 +56,6 @@ pub use reactor::{
 
 pub use lock::{
     ProtectedSubsystems,
-    ResourceLockGuard,
     ResourceLockManager,
     ResourceLockManagerConfig,
     ResourceSubsystem,

io-engine/src/grpc/mod.rs

@@ -1,21 +1,28 @@
+use futures::{channel::oneshot::Receiver, FutureExt};
+use nix::errno::Errno;
+pub use server::MayastorGrpcServer;
 use std::{
     error::Error,
     fmt::{Debug, Display},
     future::Future,
     time::Duration,
 };
-
-use futures::channel::oneshot::Receiver;
-use nix::errno::Errno;
-pub use server::MayastorGrpcServer;
 use tokio::sync::RwLock;
 use tonic::{Request, Response, Status};
 
 use crate::{
     bdev_api::BdevError,
-    core::{CoreError, Reactor, VerboseError},
+    core::{
+        lock::ResourceLockManager,
+        CoreError,
+        Reactor,
+        ResourceSubsystem,
+        VerboseError,
+    },
 };
 
+use std::panic::AssertUnwindSafe;
+
 impl From<BdevError> for tonic::Status {
     fn from(e: BdevError) -> Self {
         match e {
@@ -131,6 +138,127 @@ pub(crate) trait RWSerializer<F, T> {
     async fn shared(&self, ctx: GrpcClientContext, f: F) -> Result<T, Status>;
 }
 
+/// Trait give the default implementation of the gRPC serialization.
+/// Each service can implement its own serialization logic.
+#[async_trait::async_trait]
+pub(crate) trait GrpcRWSerializer<F, T> {
+    async fn serialize_grpc(
+        &self,
+        ctx: GrpcClientContext,
+        resource: Option<String>,
+        subsystem: ResourceSubsystem,
+        global_operation: bool,
+        write_lock: bool,
+        f: F,
+    ) -> Result<T, Status>
+    where
+        T: Send + 'static,
+        F: core::future::Future<Output = Result<T, Status>> + Send + 'static,
+    {
+        async fn acquire_lock(
+            ctx: &GrpcClientContext,
+            resource: Option<String>,
+            subsystem: ResourceSubsystem,
+            write_lock: bool,
+            global_operation: bool,
+        ) -> Result<(), Status> {
+            let lock_manager = ResourceLockManager::get_instance();
+            if global_operation {
+                if write_lock {
+                    match lock_manager.write_lock(Some(ctx.timeout)).await {
+                        Some(_) => (),
+                        None => return Err(Status::deadline_exceeded(
+                            "Failed to acquire global write lock within given timeout"
+                            .to_string()
+                        )),
+                    }
+                } else {
+                    match lock_manager.read_lock(Some(ctx.timeout)).await {
+                        Some(_) => (),
+                        None => return Err(Status::deadline_exceeded(
+                            "Failed to acquire global read lock within given timeout"
+                            .to_string()
+                        )),
+                    }
+                }
+            }
+            if let Some(resource) = resource {
+                if write_lock {
+                    match subsystem.write_lock_resource(resource, Some(ctx.timeout))
+                        .await {
+                            Some(_) => (),
+                            None => return Err(Status::deadline_exceeded(
+                                "Failed to acquire write lock for the resource within given timeout"
+                                .to_string()
+                            )),
+                        };
+                } else {
+                    match subsystem.read_lock_resource(resource, Some(ctx.timeout))
+                        .await {
+                            Some(_) => (),
+                            None => return Err(Status::deadline_exceeded(
+                                "Failed to acquire read lock for the resource within given timeout"
+                                .to_string()
+                            )),
+                        };
+                }
+            } else if write_lock {
+                match subsystem.write_lock_subsystem(Some(ctx.timeout)).await {
+                        Some(_) => (),
+                        None => return Err(Status::deadline_exceeded(
+                            "Failed to acquire subsystem write lock within given timeout"
+                            .to_string()
+                        )),
+                    };
+            } else {
+                match subsystem.read_lock_subsystem(Some(ctx.timeout)).await {
+                        Some(_) => (),
+                        None => return Err(Status::deadline_exceeded(
+                            "Failed to acquire subsystem read lock within given timeout"
+                            .to_string()
+                        )),
+                    };
+            }
+            Ok(())
+        }
+
+        let fut = AssertUnwindSafe(f).catch_unwind();
+        // Schedule a Tokio task to detach it from the high-level gRPC future
+        // and avoid task cancellation when the top-level gRPC future is
+        // cancelled.
+        match tokio::spawn(async move {
+            acquire_lock(
+                &ctx,
+                resource,
+                subsystem,
+                write_lock,
+                global_operation,
+            )
+            .await?;
+
+            let r = fut.await;
+
+            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
+                    )))
+                }
+            }
+        })
+        .await
+        {
+            Ok(r) => r,
+            Err(_) => Err(Status::cancelled("gRPC call cancelled")),
+        }
+    }
+}
 /// Trait allows service implementing to return RWLock of itself to the
 /// caller.
 #[async_trait::async_trait]