Txpool v2 features for futures connections with other modules

CHANGELOG.md

@@ -6,6 +6,9 @@ and this project adheres to [Semantic Versioning](http://semver.org/).
 
 ## [Unreleased]
 
+### Added
+- [2216](https://github.com/FuelLabs/fuel-core/pull/2216): Add more function to the state and task of TxPoolV2 to handle the future interactions with others modules (PoA, BlockProducer, BlockImporter and P2P)
+
 ### Removed
 - [2306](https://github.com/FuelLabs/fuel-core/pull/2306): Removed hack for genesis asset contract from the code.
 

crates/services/Cargo.toml

@@ -15,11 +15,15 @@ async-trait = { workspace = true }
 fuel-core-metrics = { workspace = true }
 futures = { workspace = true }
 parking_lot = { workspace = true }
+rayon = { workspace = true, optional = true }
 tokio = { workspace = true, features = ["full"] }
 tracing = { workspace = true }
 
 [dev-dependencies]
+fuel-core-services = { path = ".", features = ["sync-processor"] }
+futures = { workspace = true }
 mockall = { workspace = true }
 
 [features]
 test-helpers = []
+sync-processor = ["dep:rayon"]

crates/services/src/async_processor.rs

@@ -0,0 +1,258 @@
+use std::{
+    future::Future,
+    sync::Arc,
+};
+use tokio::{
+    runtime,
+    sync::{
+        OwnedSemaphorePermit,
+        Semaphore,
+    },
+};
+
+/// A processor that can execute async tasks with a limit on the number of tasks that can be
+/// executed concurrently.
+pub struct AsyncProcessor {
+    semaphore: Arc<Semaphore>,
+    thread_pool: runtime::Runtime,
+}
+
+/// A reservation for a task to be executed by the `AsyncProcessor`.
+pub struct AsyncReservation(OwnedSemaphorePermit);
+
+/// Out of capacity error.
+#[derive(Debug, PartialEq, Eq)]
+pub struct OutOfCapacity;
+
+impl AsyncProcessor {
+    /// Create a new `AsyncProcessor` with the given number of threads and the number of pending
+    /// tasks.
+    pub fn new(
+        number_of_threads: usize,
+        number_of_pending_tasks: usize,
+    ) -> anyhow::Result<Self> {
+        let thread_pool = runtime::Builder::new_multi_thread()
+            .worker_threads(number_of_threads)
+            .enable_all()
+            .build()
+            .map_err(|e| anyhow::anyhow!("Failed to create a tokio pool: {}", e))?;
+        let semaphore = Arc::new(Semaphore::new(number_of_pending_tasks));
+        Ok(Self {
+            thread_pool,
+            semaphore,
+        })
+    }
+
+    /// Reserve a slot for a task to be executed.
+    pub fn reserve(&self) -> Result<AsyncReservation, OutOfCapacity> {
+        let permit = self.semaphore.clone().try_acquire_owned();
+        if let Ok(permit) = permit {
+            Ok(AsyncReservation(permit))
+        } else {
+            Err(OutOfCapacity)
+        }
+    }
+
+    /// Spawn a task with a reservation.
+    pub fn spawn_reserved<F>(&self, reservation: AsyncReservation, future: F)
+    where
+        F: Future<Output = ()> + Send + 'static,
+    {
+        let permit = reservation.0;
+        self.thread_pool.spawn(async move {
+            let _drop = permit;
+            future.await
+        });
+    }
+
+    /// Tries to spawn a task. If the task cannot be spawned, returns an error.
+    pub fn try_spawn<F>(&self, future: F) -> Result<(), OutOfCapacity>
+    where
+        F: Future<Output = ()> + Send + 'static,
+    {
+        let reservation = self.reserve()?;
+        self.spawn_reserved(reservation, future);
+        Ok(())
+    }
+}
+
+#[cfg(test)]
+#[allow(clippy::bool_assert_comparison)]
+mod tests {
+    use super::*;
+    use std::{
+        thread::sleep,
+        time::Duration,
+    };
+    use tokio::time::Instant;
+
+    #[test]
+    fn one_spawn_single_tasks_works() {
+        // Given
+        let number_of_pending_tasks = 1;
+        let heavy_task_processor =
+            AsyncProcessor::new(1, number_of_pending_tasks).unwrap();
+
+        // When
+        let (sender, mut receiver) = tokio::sync::oneshot::channel();
+        let result = heavy_task_processor.try_spawn(async move {
+            sender.send(()).unwrap();
+        });
+
+        // Then
+        assert_eq!(result, Ok(()));
+        sleep(Duration::from_secs(1));
+        receiver.try_recv().unwrap();
+    }
+
+    #[test]
+    fn second_spawn_fails_when_limit_is_one_and_first_in_progress() {
+        // Given
+        let number_of_pending_tasks = 1;
+        let heavy_task_processor =
+            AsyncProcessor::new(1, number_of_pending_tasks).unwrap();
+        let first_spawn_result = heavy_task_processor.try_spawn(async move {
+            sleep(Duration::from_secs(1));
+        });
+        assert_eq!(first_spawn_result, Ok(()));
+
+        // When
+        let second_spawn_result = heavy_task_processor.try_spawn(async move {
+            sleep(Duration::from_secs(1));
+        });
+
+        // Then
+        assert_eq!(second_spawn_result, Err(OutOfCapacity));
+    }
+
+    #[test]
+    fn second_spawn_works_when_first_is_finished() {
+        let number_of_pending_tasks = 1;
+        let heavy_task_processor =
+            AsyncProcessor::new(1, number_of_pending_tasks).unwrap();
+
+        // Given
+        let (sender, receiver) = tokio::sync::oneshot::channel();
+        let first_spawn = heavy_task_processor.try_spawn(async move {
+            sleep(Duration::from_secs(1));
+            sender.send(()).unwrap();
+        });
+        assert_eq!(first_spawn, Ok(()));
+        futures::executor::block_on(async move {
+            receiver.await.unwrap();
+        });
+
+        // When
+        let second_spawn = heavy_task_processor.try_spawn(async move {
+            sleep(Duration::from_secs(1));
+        });
+
+        // Then
+        assert_eq!(second_spawn, Ok(()));
+    }
+
+    #[test]
+    fn can_spawn_10_tasks_when_limit_is_10() {
+        // Given
+        let number_of_pending_tasks = 10;
+        let heavy_task_processor =
+            AsyncProcessor::new(1, number_of_pending_tasks).unwrap();
+
+        for _ in 0..number_of_pending_tasks {
+            // When
+            let result = heavy_task_processor.try_spawn(async move {
+                tokio::time::sleep(Duration::from_secs(1)).await;
+            });
+
+            // Then
+            assert_eq!(result, Ok(()));
+        }
+    }
+
+    #[test]
+    fn executes_10_tasks_for_10_seconds_with_one_thread() {
+        // Given
+        let number_of_pending_tasks = 10;
+        let number_of_threads = 1;
+        let heavy_task_processor =
+            AsyncProcessor::new(number_of_threads, number_of_pending_tasks).unwrap();
+
+        // When
+        let (broadcast_sender, mut broadcast_receiver) =
+            tokio::sync::broadcast::channel(1024);
+        let instant = Instant::now();
+        for _ in 0..number_of_pending_tasks {
+            let broadcast_sender = broadcast_sender.clone();
+            let result = heavy_task_processor.try_spawn(async move {
+                sleep(Duration::from_secs(1));
+                broadcast_sender.send(()).unwrap();
+            });
+            assert_eq!(result, Ok(()));
+        }
+        drop(broadcast_sender);
+
+        // Then
+        futures::executor::block_on(async move {
+            while broadcast_receiver.recv().await.is_ok() {}
+            assert!(instant.elapsed() >= Duration::from_secs(10));
+        });
+    }
+
+    #[test]
+    fn executes_10_tasks_for_2_seconds_with_10_thread() {
+        // Given
+        let number_of_pending_tasks = 10;
+        let number_of_threads = 10;
+        let heavy_task_processor =
+            AsyncProcessor::new(number_of_threads, number_of_pending_tasks).unwrap();
+
+        // When
+        let (broadcast_sender, mut broadcast_receiver) =
+            tokio::sync::broadcast::channel(1024);
+        let instant = Instant::now();
+        for _ in 0..number_of_pending_tasks {
+            let broadcast_sender = broadcast_sender.clone();
+            let result = heavy_task_processor.try_spawn(async move {
+                sleep(Duration::from_secs(1));
+                broadcast_sender.send(()).unwrap();
+            });
+            assert_eq!(result, Ok(()));
+        }
+        drop(broadcast_sender);
+
+        // Then
+        futures::executor::block_on(async move {
+            while broadcast_receiver.recv().await.is_ok() {}
+            assert!(instant.elapsed() <= Duration::from_secs(2));
+        });
+    }
+
+    #[test]
+    fn executes_10_tasks_for_2_seconds_with_1_thread() {
+        // Given
+        let number_of_pending_tasks = 10;
+        let number_of_threads = 10;
+        let heavy_task_processor =
+            AsyncProcessor::new(number_of_threads, number_of_pending_tasks).unwrap();
+
+        // When
+        let (broadcast_sender, mut broadcast_receiver) =
+            tokio::sync::broadcast::channel(1024);
+        let instant = Instant::now();
+        for _ in 0..number_of_pending_tasks {
+            let broadcast_sender = broadcast_sender.clone();
+            let result = heavy_task_processor.try_spawn(async move {
+                tokio::time::sleep(Duration::from_secs(1)).await;
+                broadcast_sender.send(()).unwrap();
+            });
+            assert_eq!(result, Ok(()));
+        }
+        drop(broadcast_sender);
+
+        // Then
+        futures::executor::block_on(async move {
+            while broadcast_receiver.recv().await.is_ok() {}
+            assert!(instant.elapsed() <= Duration::from_secs(2));
+        });
+    }
+}

crates/services/src/lib.rs

@@ -5,9 +5,12 @@
 #![deny(missing_docs)]
 #![deny(warnings)]
 
+mod async_processor;
 mod service;
 mod state;
 mod sync;
+#[cfg(feature = "sync-processor")]
+mod sync_processor;
 
 /// Re-exports for streaming utilities
 pub mod stream {
@@ -58,6 +61,7 @@ where
     }
 }
 
+pub use async_processor::AsyncProcessor;
 pub use service::{
     EmptyShared,
     RunnableService,
@@ -74,3 +78,10 @@ pub use sync::{
     Shared,
     SharedMutex,
 };
+#[cfg(feature = "sync-processor")]
+pub use sync_processor::SyncProcessor;
+
+// For tests
+use crate as fuel_core_services;
+#[allow(unused_imports)]
+use fuel_core_services as _;