chore: bootstrapping task-manager through feature flags

crates/topos-tce-broadcast/Cargo.toml

@@ -32,7 +32,7 @@ rand.workspace = true
 topos-test-sdk = { path = "../topos-test-sdk/" }
 
 [features]
-default = []
+task-manager-channels = []
 
 [[bench]]
 name = "double_echo"

crates/topos-tce-broadcast/benches/double_echo.rs

@@ -1,7 +1,9 @@
 use criterion::async_executor::FuturesExecutor;
 use criterion::{criterion_group, criterion_main, Criterion};
 
+#[cfg(feature = "task-manager-channels")]
 mod task_manager_channels;
+#[cfg(not(feature = "task-manager-channels"))]
 mod task_manager_futures;
 
 pub fn criterion_benchmark(c: &mut Criterion) {
@@ -11,6 +13,7 @@ pub fn criterion_benchmark(c: &mut Criterion) {
         .build()
         .unwrap();
 
+    #[cfg(feature = "task-manager-channels")]
     c.bench_function("double_echo with channels", |b| {
         b.to_async(FuturesExecutor).iter(|| async {
             runtime.block_on(async {
@@ -19,6 +22,7 @@ pub fn criterion_benchmark(c: &mut Criterion) {
         })
     });
 
+    #[cfg(not(feature = "task-manager-channels"))]
     c.bench_function("double_echo with futures", |b| {
         b.to_async(FuturesExecutor).iter(|| async {
             runtime.block_on(async {

crates/topos-tce-broadcast/benches/task_manager_channels.rs

@@ -1,31 +1,26 @@
-use std::collections::HashMap;
-
 use rand::Rng;
 use tokio::spawn;
 use tokio::sync::mpsc;
 
+use tce_transport::ReliableBroadcastParams;
 use topos_core::uci::CertificateId;
 use topos_p2p::PeerId;
-use topos_tce_broadcast::task_manager_channels::{TaskManager, Thresholds};
+use topos_tce_broadcast::task_manager_channels::TaskManager;
 use topos_tce_broadcast::DoubleEchoCommand;
 
 pub async fn processing_double_echo(n: u64) {
     let (message_sender, message_receiver) = mpsc::channel(1024);
-    let (task_completion_sender, task_completion_receiver) = mpsc::channel(1024);
     let (event_sender, mut event_receiver) = mpsc::channel(1024);
 
-    let task_manager = TaskManager {
-        message_receiver,
-        task_completion: task_completion_receiver,
-        task_context: HashMap::new(),
-        thresholds: Thresholds {
-            echo: n as usize,
-            ready: n as usize,
-            delivery: n as usize,
-        },
+    let threshold = ReliableBroadcastParams {
+        echo_threshold: n as usize,
+        ready_threshold: n as usize,
+        delivery_threshold: n as usize,
     };
 
-    spawn(task_manager.run(task_completion_sender, event_sender));
+    let (task_manager, shutdown_receiver) = TaskManager::new(message_receiver, threshold);
+
+    spawn(task_manager.run(event_sender, shutdown_receiver));
 
     let mut certificates = vec![];
 

crates/topos-tce-broadcast/benches/task_manager_futures.rs

@@ -1,31 +1,26 @@
-use futures::stream::FuturesUnordered;
 use rand::Rng;
+use tce_transport::ReliableBroadcastParams;
 use tokio::spawn;
 use tokio::sync::mpsc;
 use topos_core::uci::CertificateId;
 use topos_p2p::PeerId;
 use topos_tce_broadcast::DoubleEchoCommand;
 
-use topos_tce_broadcast::task_manager_futures::{TaskManager, Thresholds};
+use topos_tce_broadcast::task_manager_futures::TaskManager;
 
 pub async fn processing_double_echo(n: u64) {
     let (message_sender, message_receiver) = mpsc::channel(1024);
-    let (task_completion_sender, mut task_completion_receiver) = mpsc::channel(48_000);
-    let (shutdown_sender, shutdown_receiver) = mpsc::channel(1);
-
-    let task_manager = TaskManager {
-        message_receiver,
-        task_completion_sender,
-        tasks: Default::default(),
-        running_tasks: FuturesUnordered::new(),
-        thresholds: Thresholds {
-            echo: n as usize,
-            ready: n as usize,
-            delivery: n as usize,
-        },
-        shutdown_sender,
+    let (task_completion_sender, mut task_completion_receiver) = mpsc::channel(40_000);
+
+    let thresholds = ReliableBroadcastParams {
+        echo_threshold: n as usize,
+        ready_threshold: n as usize,
+        delivery_threshold: n as usize,
     };
 
+    let (task_manager, shutdown_receiver) =
+        TaskManager::new(message_receiver, task_completion_sender, thresholds);
+
     spawn(task_manager.run(shutdown_receiver));
 
     let mut certificates = vec![];

crates/topos-tce-broadcast/src/constant.rs

@@ -7,6 +7,24 @@ lazy_static! {
             .ok()
             .and_then(|s| s.parse().ok())
             .unwrap_or(2048);
+    /// Size of the channel between double echo and the task manager
+    pub static ref TASK_MANAGER_CHANNEL_SIZE: usize =
+        std::env::var("TOPOS_TASK_MANAGER_CHANNEL_SIZE")
+            .ok()
+            .and_then(|s| s.parse().ok())
+            .unwrap_or(20_480);
+    /// Size of the channel to send protocol events from the double echo
+    pub static ref PROTOCOL_CHANNEL_SIZE: usize =
+        std::env::var("TOPOS_PROTOCOL_CHANNEL_SIZE")
+            .ok()
+            .and_then(|s| s.parse().ok())
+            .unwrap_or(2048);
+    /// Size of the channel to send updated subscriptions views to the double echo
+    pub static ref SUBSCRIPTION_VIEW_CHANNEL_SIZE: usize =
+        std::env::var("TOPOS_SUBSCRIPTION_VIEW_CHANNEL_SIZE")
+            .ok()
+            .and_then(|s| s.parse().ok())
+            .unwrap_or(2048);
     /// Capacity alert threshold for the double echo command channel
     pub static ref COMMAND_CHANNEL_CAPACITY: usize = COMMAND_CHANNEL_SIZE
         .checked_mul(10)

crates/topos-tce-broadcast/src/double_echo/broadcast_state.rs

@@ -13,6 +13,7 @@ mod status;
 
 pub use status::Status;
 
+#[derive(Debug)]
 pub struct BroadcastState {
     subscriptions_view: SubscriptionsView,
     status: Status,