[runtime] Introduce Runtime-Agnostic, Asynchronous RwLock (#836)

Author: patrick-ogrady

Cargo.lock

@@ -67,6 +67,7 @@ opentelemetry-otlp = "0.28.0"
 opentelemetry_sdk =  "0.28.0"
 tracing-opentelemetry = "0.29.0"
 rayon = "1.10.0"
+async-lock = "3.4.0"
 
 [profile.bench]
 # Because we enable overflow checks in "release," we should benchmark with them.

Cargo.toml

@@ -26,6 +26,7 @@ tracing-subscriber = { workspace = true, features = ["fmt", "json", "env-filter"
 opentelemetry = { workspace = true }
 tracing-opentelemetry = { workspace = true }
 rayon = { workspace = true }
+async-lock = { workspace = true }
 
 # Enable "js" feature when WASM is target
 [target.'cfg(target_arch = "wasm32")'.dependencies.getrandom]

runtime/Cargo.toml

@@ -37,7 +37,9 @@ cfg_if::cfg_if! {
 mod storage;
 pub mod telemetry;
 mod utils;
-pub use utils::{create_pool, reschedule, Handle, Signal, Signaler};
+pub use utils::{
+    create_pool, reschedule, Handle, RwLock, RwLockReadGuard, RwLockWriteGuard, Signal, Signaler,
+};
 
 /// Prefix for runtime metrics.
 const METRICS_PREFIX: &str = "runtime";

runtime/src/lib.rs

@@ -346,6 +346,83 @@ pub fn create_pool<S: Spawner + Metrics>(
         .build()
 }
 
+/// Async reader–writer lock.
+///
+/// Powered by [async_lock::RwLock], `RwLock` provides both fair writer acquisition
+/// and `try_read` / `try_write` without waiting (without any runtime-specific dependencies).
+///
+/// Usage:
+/// ```rust
+/// use commonware_runtime::{Spawner, Runner, Signaler, deterministic, RwLock};
+///
+/// let executor = deterministic::Runner::default();
+/// executor.start(|context| async move {
+///     // Create a new RwLock
+///     let lock = RwLock::new(2);
+///
+///     // many concurrent readers
+///     let r1 = lock.read().await;
+///     let r2 = lock.read().await;
+///     assert_eq!(*r1 + *r2, 4);
+///
+///     // exclusive writer
+///     drop((r1, r2));
+///     let mut w = lock.write().await;
+///     *w += 1;
+/// });
+/// ```
+pub struct RwLock<T>(async_lock::RwLock<T>);
+
+/// Shared guard returned by [`RwLock::read`].
+pub type RwLockReadGuard<'a, T> = async_lock::RwLockReadGuard<'a, T>;
+
+/// Exclusive guard returned by [`RwLock::write`].
+pub type RwLockWriteGuard<'a, T> = async_lock::RwLockWriteGuard<'a, T>;
+
+impl<T> RwLock<T> {
+    /// Create a new lock.
+    #[inline]
+    pub const fn new(value: T) -> Self {
+        Self(async_lock::RwLock::new(value))
+    }
+
+    /// Acquire a shared read guard.
+    #[inline]
+    pub async fn read(&self) -> RwLockReadGuard<'_, T> {
+        self.0.read().await
+    }
+
+    /// Acquire an exclusive write guard.
+    #[inline]
+    pub async fn write(&self) -> RwLockWriteGuard<'_, T> {
+        self.0.write().await
+    }
+
+    /// Try to get a read guard without waiting.
+    #[inline]
+    pub fn try_read(&self) -> Option<RwLockReadGuard<'_, T>> {
+        self.0.try_read()
+    }
+
+    /// Try to get a write guard without waiting.
+    #[inline]
+    pub fn try_write(&self) -> Option<RwLockWriteGuard<'_, T>> {
+        self.0.try_write()
+    }
+
+    /// Get mutable access without locking (requires `&mut self`).
+    #[inline]
+    pub fn get_mut(&mut self) -> &mut T {
+        self.0.get_mut()
+    }
+
+    /// Consume the lock, returning the inner value.
+    #[inline]
+    pub fn into_inner(self) -> T {
+        self.0.into_inner()
+    }
+}
+
 #[cfg(test)]
 async fn task(i: usize) -> usize {
     for _ in 0..5 {
@@ -376,7 +453,7 @@ pub fn run_tasks(tasks: usize, runner: crate::deterministic::Runner) -> (String,
 #[cfg(test)]
 mod tests {
     use super::*;
-    use crate::{tokio, Metrics};
+    use crate::{deterministic, tokio, Metrics};
     use commonware_macros::test_traced;
     use rayon::iter::{IntoParallelRefIterator, ParallelIterator};
 
@@ -396,4 +473,26 @@ mod tests {
             });
         });
     }
+
+    #[test_traced]
+    fn test_rwlock() {
+        let executor = deterministic::Runner::default();
+        executor.start(|_| async move {
+            // Create a new RwLock
+            let lock = RwLock::new(100);
+
+            // many concurrent readers
+            let r1 = lock.read().await;
+            let r2 = lock.read().await;
+            assert_eq!(*r1 + *r2, 200);
+
+            // exclusive writer
+            drop((r1, r2)); // all readers must go away
+            let mut w = lock.write().await;
+            *w += 1;
+
+            // Check the value
+            assert_eq!(*w, 101);
+        });
+    }
 }