Merge pull request #3659 from autonomys/move-net-utils

Move process utilities out of subspace-networking

Author: teor2345

Cargo.lock

@@ -257,9 +257,9 @@ subspace-erasure-coding = { version = "0.1.0", path = "crates/subspace-erasure-c
 subspace-farmer-components = { version = "0.1.0", path = "crates/subspace-farmer-components" }
 subspace-gateway-rpc = { version = "0.1.0", path = "crates/subspace-gateway-rpc" }
 subspace-kzg = { version = "0.1.0", path = "shared/subspace-kzg", default-features = false }
-subspace-logging = { version = "0.0.1", path = "shared/subspace-logging" }
 subspace-metrics = { version = "0.1.0", path = "shared/subspace-metrics" }
 subspace-networking = { version = "0.1.0", path = "crates/subspace-networking" }
+subspace-process = { version = "0.0.1", path = "shared/subspace-process" }
 subspace-proof-of-space = { version = "0.1.0", path = "crates/subspace-proof-of-space", default-features = false }
 subspace-proof-of-space-gpu = { version = "0.1.0", path = "shared/subspace-proof-of-space-gpu" }
 subspace-proof-of-time = { version = "0.1.0", path = "crates/subspace-proof-of-time" }

Cargo.toml

@@ -26,8 +26,8 @@ libp2p = { workspace = true, features = ["kad"] }
 prometheus-client.workspace = true
 serde = { workspace = true, features = ["derive"] }
 serde_json.workspace = true
-subspace-logging.workspace = true
 subspace-metrics.workspace = true
 subspace-networking.workspace = true
+subspace-process.workspace = true
 tokio = { workspace = true, features = ["macros", "parking_lot", "rt-multi-thread"] }
 tracing.workspace = true

crates/subspace-bootstrap-node/Cargo.toml

@@ -15,11 +15,12 @@ use std::net::{IpAddr, Ipv4Addr, Ipv6Addr, SocketAddr};
 use std::panic;
 use std::process::exit;
 use std::sync::Arc;
-use subspace_logging::init_logger;
 use subspace_metrics::{RegistryAdapter, start_prometheus_metrics_server};
 use subspace_networking::libp2p::multiaddr::Protocol;
-use subspace_networking::utils::{raise_fd_limit, run_future_in_dedicated_thread, shutdown_signal};
 use subspace_networking::{Config, KademliaMode, peer_id};
+use subspace_process::{
+    init_logger, raise_fd_limit, run_future_in_dedicated_thread, shutdown_signal,
+};
 use tracing::{debug, info};
 
 /// Size of the LRU cache for peers.
@@ -185,11 +186,12 @@ async fn main() -> Result<(), Box<dyn Error>> {
 
             let node_runner_fut = run_future_in_dedicated_thread(
                 move || async move { node_runner.run().await },
-                "bootstrap-node-networking".to_string(),
+                "bootstrap-networking".to_string(),
             )?;
 
             info!("Subspace Bootstrap Node started");
 
+            // TODO: spawn this in a dedicated thread
             let prometheus_task = should_start_prometheus_server
                 .then(|| {
                     start_prometheus_metrics_server(
@@ -199,36 +201,26 @@ async fn main() -> Result<(), Box<dyn Error>> {
                 })
                 .transpose()?;
 
-            // If a spawned future is running for a long time, it can block receiving exit signals.
-            // Rather than hunting down every possible blocking future, we give the exit signal itself a
-            // dedicated thread to run on.
-            let exit_signal_select_fut = run_future_in_dedicated_thread(
-                move || async move {
-                    if let Some(prometheus_task) = prometheus_task {
-                        select! {
-                            // Signal future
-                            () = signal.fuse() => {},
-                            _ = node_runner_fut.fuse() => {
-                                info!("DSN network runner exited.");
-                            },
-                            _ = prometheus_task.fuse() => {
-                                info!("Prometheus server exited.");
-                            },
-                        }
-                    } else {
-                        select! {
-                            // Signal future
-                            () = signal.fuse() => {},
-                            _ = node_runner_fut.fuse() => {
-                                info!("DSN network runner exited.");
-                            },
-                        }
-                    }
-                },
-                "bootstrap-node-exit-signal-select".to_string(),
-            )?;
-
-            exit_signal_select_fut.await?;
+            if let Some(prometheus_task) = prometheus_task {
+                select! {
+                    // Signal future
+                    () = signal.fuse() => {},
+                    _ = node_runner_fut.fuse() => {
+                        info!("DSN network runner exited.");
+                    },
+                    _ = prometheus_task.fuse() => {
+                        info!("Prometheus server exited.");
+                    },
+                }
+            } else {
+                select! {
+                    // Signal future
+                    () = signal.fuse() => {},
+                    _ = node_runner_fut.fuse() => {
+                        info!("DSN network runner exited.");
+                    },
+                }
+            }
         }
         Command::GenerateKeypair { json } => {
             let output = KeypairOutput::new(Keypair::generate());

crates/subspace-bootstrap-node/src/main.rs

@@ -55,9 +55,9 @@ subspace-data-retrieval.workspace = true
 subspace-erasure-coding.workspace = true
 subspace-farmer-components.workspace = true
 subspace-kzg.workspace = true
-subspace-logging = { workspace = true, optional = true }
 subspace-metrics = { workspace = true, optional = true }
 subspace-networking.workspace = true
+subspace-process = { workspace = true, optional = true }
 subspace-proof-of-space.workspace = true
 subspace-rpc-primitives.workspace = true
 subspace-verification = { workspace = true, features = ["kzg"] }
@@ -97,5 +97,5 @@ binary = [
     "dep:criterion",
     "dep:mimalloc",
     "dep:subspace-metrics",
-    "dep:subspace-logging",
+    "dep:subspace-process",
 ]

crates/subspace-farmer/Cargo.toml

@@ -92,7 +92,7 @@ fn create_thread_pool(
         .unwrap_or_else(recommended_number_of_farming_threads);
 
     ThreadPoolBuilder::new()
-        .thread_name(|thread_index| format!("benchmark.{thread_index}"))
+        .thread_name(|thread_index| format!("benchmark.{thread_index:02}"))
         .num_threads(farming_thread_pool_size)
         .spawn_handler(tokio_rayon_spawn_handler())
         .build()

crates/subspace-farmer/src/bin/subspace-farmer/commands/benchmark.rs

@@ -21,9 +21,7 @@ use std::pin::Pin;
 use std::time::Duration;
 use subspace_farmer::cluster::nats_client::NatsClient;
 use subspace_metrics::{RegistryAdapter, start_prometheus_metrics_server};
-use subspace_networking::utils::{
-    AsyncJoinOnDrop, run_future_in_dedicated_thread, shutdown_signal,
-};
+use subspace_process::{AsyncJoinOnDrop, shutdown_signal};
 use subspace_proof_of_space::Table;
 
 const REQUEST_RETRY_MAX_ELAPSED_TIME: Duration = Duration::from_mins(1);
@@ -159,33 +157,22 @@ where
             RegistryAdapter::PrometheusClient(registry),
         )?;
 
+        // TODO: spawn this in a dedicated thread
         let join_handle = tokio::spawn(prometheus_task);
         tasks.push(Box::pin(async move {
             Ok(AsyncJoinOnDrop::new(join_handle, true).await??)
         }));
     }
 
-    // If a spawned future is running for a long time, it can block receiving exit signals.
-    // Rather than hunting down every possible blocking future, we give the exit signal itself a
-    // dedicated thread to run on.
-    let exit_signal_select_fut = run_future_in_dedicated_thread(
-        move || async move {
-            select! {
-                // Signal future
-                _ = signal.fuse() => {
-                    Ok(())
-                },
-
-                // Run future
-                result = tasks.next() => {
-                    result.expect("List of tasks is not empty; qed")
-                },
-            }
-        },
-        "farmer-cluster-exit-signal-select".to_string(),
-    )?;
+    select! {
+        // Signal future
+        () = signal.fuse() => {},
 
-    exit_signal_select_fut.await??;
+        // Run future
+        result = tasks.next() => {
+            result.expect("List of tasks is not empty; qed")?;
+        },
+    }
 
     anyhow::Ok(())
 }

crates/subspace-farmer/src/bin/subspace-farmer/commands/cluster.rs

@@ -14,7 +14,7 @@ use std::time::Duration;
 use subspace_farmer::cluster::cache::cache_service;
 use subspace_farmer::cluster::nats_client::NatsClient;
 use subspace_farmer::disk_piece_cache::DiskPieceCache;
-use subspace_networking::utils::AsyncJoinOnDrop;
+use subspace_process::AsyncJoinOnDrop;
 
 /// Interval between cache self-identification broadcast messages
 pub(super) const CACHE_IDENTIFICATION_BROADCAST_INTERVAL: Duration = Duration::from_secs(30);

crates/subspace-farmer/src/bin/subspace-farmer/commands/cluster/cache.rs

@@ -29,7 +29,7 @@ use subspace_farmer::node_client::rpc_node_client::RpcNodeClient;
 use subspace_farmer::single_disk_farm::identity::Identity;
 use subspace_kzg::Kzg;
 use subspace_networking::utils::piece_provider::PieceProvider;
-use subspace_networking::utils::{AsyncJoinOnDrop, run_future_in_dedicated_thread};
+use subspace_process::{AsyncJoinOnDrop, run_future_in_dedicated_thread};
 use tracing::{Instrument, info, info_span};
 
 /// Get piece retry attempts number.
@@ -210,7 +210,7 @@ pub(super) async fn controller(
 
             async move {
                 let fut =
-                    run_future_in_dedicated_thread(move || fut, format!("cache-worker-{index}"));
+                    run_future_in_dedicated_thread(move || fut, format!("cache-worker-{index:02}"));
                 anyhow::Ok(fut?.await?)
             }
         })
@@ -311,7 +311,7 @@ pub(super) async fn controller(
                         )
                         .await
                     },
-                    format!("caches-{index}"),
+                    format!("caches-{index:02}"),
                 );
                 anyhow::Ok(fut?.await?)
             }

crates/subspace-farmer/src/bin/subspace-farmer/commands/cluster/controller.rs

@@ -33,7 +33,7 @@ use subspace_farmer::utils::recommended_number_of_farming_threads;
 use subspace_farmer::utils::ss58::parse_ss58_reward_address;
 use subspace_farmer_components::reading::ReadSectorRecordChunksMode;
 use subspace_kzg::Kzg;
-use subspace_networking::utils::{AsyncJoinOnDrop, run_future_in_dedicated_thread};
+use subspace_process::{AsyncJoinOnDrop, run_future_in_dedicated_thread};
 use subspace_proof_of_space::Table;
 use tracing::{Instrument, error, info, info_span, warn};