fuel-core-sync

.gitignore

@@ -6,4 +6,5 @@
 .vscode
 .cov
 lcov.info
-version-compatibility/Cargo.lock
+version-compatibility/Cargo.lock
+benches/benches-outputs/Cargo.lock

crates/services/Cargo.toml

@@ -1,18 +1,19 @@
 [package]
-name = "fuel-core-services"
-version = "0.15.1"
 authors = ["Fuel Labs <contact@fuel.sh>"]
+description = "The common code for fuel core services."
 edition = "2021"
 homepage = "https://fuel.network/"
-keywords = ["blockchain", "fuel", "consensus", "bft"]
+keywords = ["bft", "blockchain", "consensus", "fuel"]
 license = "BUSL-1.1"
+name = "fuel-core-services"
 repository = "https://github.com/FuelLabs/fuel-core"
-description = "The common code for fuel core services."
+version = "0.15.1"
 
 [dependencies]
 anyhow = "1.0"
 async-trait = "0.1"
 futures = "0.3"
+parking_lot = "0.12"
 tokio = { version = "1.21", features = ["full"] }
 tracing = "0.1"
 

crates/services/src/lib.rs

@@ -19,13 +19,27 @@ pub mod stream {
         core::pin::Pin<Box<dyn Stream<Item = T> + Send + Sync + 'static>>;
 }
 
+/// The source of some network data.
+pub struct SourcePeer<T> {
+    /// The source of the data.
+    pub peer_id: PeerId,
+    /// The data.
+    pub data: T,
+}
+
+/// Placeholder for a peer id
+pub struct PeerId;
+
 pub use service::{
     EmptyShared,
+    KillSwitch,
     RunnableService,
     RunnableTask,
     Service,
     ServiceRunner,
     Shared,
+    SharedMutex,
+    Shutdown,
 };
 pub use state::{
     State,

crates/services/src/service.rs

@@ -1,20 +1,44 @@
+use std::sync::Arc;
+
 use crate::state::{
     State,
     StateWatcher,
 };
 use anyhow::anyhow;
 use tokio::{
-    sync::watch,
+    sync::{
+        watch,
+        Semaphore,
+    },
     task::JoinHandle,
 };
 
 /// Alias for Arc<T>
 pub type Shared<T> = std::sync::Arc<T>;
 
+/// A mutex that can safely be in async contexts and avoids deadlocks.
+#[derive(Debug)]
+pub struct SharedMutex<T>(Shared<parking_lot::Mutex<T>>);
+
+impl<T> Clone for SharedMutex<T> {
+    fn clone(&self) -> Self {
+        Self(self.0.clone())
+    }
+}
+
 /// Used if services have no asynchronously shared data
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
 pub struct EmptyShared;
 
+#[derive(Debug, Clone)]
+/// A handle that can wait for shutdown.
+pub struct Shutdown(Arc<Semaphore>);
+
+#[derive(Debug)]
+/// A switch that can be used to shutdown all
+/// shutdown handles that are waiting.
+pub struct KillSwitch(Arc<Semaphore>);
+
 /// Trait for service runners, providing a minimal interface for managing
 /// the lifecycle of services such as start/stop and health status.
 #[async_trait::async_trait]
@@ -290,6 +314,55 @@ where
     })
 }
 
+impl<T> SharedMutex<T> {
+    /// Creates a new `SharedMutex` with the given value.
+    pub fn new(t: T) -> Self {
+        Self(Shared::new(parking_lot::Mutex::new(t)))
+    }
+
+    /// Apply a function to the inner value and return a value.
+    pub fn apply<R>(&self, f: impl FnOnce(&mut T) -> R) -> R {
+        let mut t = self.0.lock();
+        f(&mut t)
+    }
+}
+
+impl Shutdown {
+    /// Wait for shutdown.
+    pub async fn wait(&self) {
+        let _ = self.0.acquire().await;
+    }
+}
+
+impl KillSwitch {
+    /// Create a new kill switch.
+    pub fn new() -> Self {
+        Self(Arc::new(Semaphore::new(0)))
+    }
+
+    /// Get a handle to await shutdown.
+    pub fn handle(&self) -> Shutdown {
+        Shutdown(self.0.clone())
+    }
+
+    /// Shutdown all handles.
+    pub fn kill_all(&mut self) {
+        self.0.close();
+    }
+}
+
+impl Default for KillSwitch {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
+impl Drop for KillSwitch {
+    fn drop(&mut self) {
+        self.0.close();
+    }
+}
+
 // TODO: Add tests
 #[cfg(test)]
 mod tests {

crates/services/sync/Cargo.toml

@@ -1,16 +1,25 @@
 [package]
-name = "fuel-core-sync"
-version = "0.15.1"
 authors = ["Fuel Labs <contact@fuel.sh>"]
+description = "Fuel Synchronizer"
 edition = "2021"
 homepage = "https://fuel.network/"
 keywords = ["blockchain", "fuel", "fuel-vm"]
 license = "BUSL-1.1"
+name = "fuel-core-sync"
 repository = "https://github.com/FuelLabs/fuel-core"
-description = "Fuel Synchronizer"
+version = "0.15.1"
 
 [dependencies]
 anyhow = "1.0"
+async-trait = "0.1.60"
+fuel-core-services = { path = "../" }
 fuel-core-types = { path = "../../types", version = "0.15.1" }
-parking_lot = "0.12"
+futures = "0.3.25"
 tokio = { version = "1.21", features = ["full"] }
+
+[dev-dependencies]
+fuel-core-types = { path = "../../types", features = [
+    "test-helpers",
+] }
+mockall = "0.11.3"
+test-case = "2.2.2"