diff --git a/crates/node/src/coordinator.rs b/crates/node/src/coordinator.rs index 40ed1208f1..40157af71b 100644 --- a/crates/node/src/coordinator.rs +++ b/crates/node/src/coordinator.rs @@ -425,7 +425,7 @@ where let network_client = network_client.clone(); let mpc_config = mpc_config.clone(); - tracking::spawn_checked("key_resharing", async move { + tracking::spawn_checked("key resharing", async move { Self::run_key_resharing( &config_file, keyshare_storage.clone(), diff --git a/crates/node/src/key_events.rs b/crates/node/src/key_events.rs index 496b2b833f..8eb5e0421d 100644 --- a/crates/node/src/key_events.rs +++ b/crates/node/src/key_events.rs @@ -511,7 +511,7 @@ pub async fn keygen_follower( }; tasks.spawn_checked( - &format!("Key generation follower for {:?}", key_event_id), + &format!("key generation follower for {:?}", key_event_id), keygen_computation( key_event_receiver.clone(), channel, @@ -641,7 +641,7 @@ pub async fn resharing_follower( }; tasks.spawn_checked( - &format!("Key resharing follower for {:?}", key_event_id), + &format!("key resharing follower for {:?}", key_event_id), resharing_computation( key_event_receiver.clone(), channel, diff --git a/crates/node/src/mpc_client.rs b/crates/node/src/mpc_client.rs index 7fc34e142c..82f1b859a7 100644 --- a/crates/node/src/mpc_client.rs +++ b/crates/node/src/mpc_client.rs @@ -511,7 +511,7 @@ impl MpcClient { while let Some(channel) = channel_receiver.recv().await { let mpc_clone = mpc_client.clone(); tasks.spawn_checked( - &format!("passive task {:?}", channel.task_id()), + &format!("passive task; task_id: {:?}", channel.task_id()), async move { mpc_clone.process_channel_task(channel).await }, ); } diff --git a/crates/node/src/p2p.rs b/crates/node/src/p2p.rs index f5068ea02e..798d568da7 100644 --- a/crates/node/src/p2p.rs +++ b/crates/node/src/p2p.rs @@ -366,75 +366,80 @@ pub async fn new_tls_mesh_network( let connectivities_clone = connectivities.clone(); let my_id = config.my_participant_id; info!("Spawning incoming connections handler."); - let incoming_connections_task = tracking::spawn("Handle incoming connections", async move { - let mut tasks = AutoAbortTaskCollection::new(); - while let Ok((tcp_stream, _)) = tcp_listener.accept().await { - let message_sender = message_sender.clone(); - let participant_identities = participant_identities.clone(); - let tls_acceptor = tls_acceptor.clone(); - let connectivities = connectivities_clone.clone(); - tasks.spawn_checked::<_, ()>("Handle connection", async move { - let tcp_stream = configure_tcp_stream(tcp_stream)?; - let mut stream = tls_acceptor.accept(tcp_stream).await?; - - let peer_id = verify_peer_identity(stream.get_ref().1, &participant_identities)?; - tracking::set_progress(&format!("Authenticated as {}", peer_id)); - p2p_handshake(&mut stream, TlsConnection::HANDSHAKE_TIMEOUT) - .await - .context("p2p handshake")?; - tracing::info!("Incoming {} <-- {} connected", my_id, peer_id); - let incoming_conn = Arc::new(()); - connectivities - .get(peer_id)? - .set_incoming_connection(&incoming_conn); - let mut received_bytes: u64 = 0; - loop { - let len = tokio::time::timeout( - std::time::Duration::from_secs(MESSAGE_READ_TIMEOUT_SECS), - stream.read_u32(), - ) - .await??; - if len >= MAX_MESSAGE_LEN { - anyhow::bail!("Message too long"); - } - let mut buf = vec![0; len as usize]; - tokio::time::timeout( - std::time::Duration::from_secs(MESSAGE_READ_TIMEOUT_SECS), - stream.read_exact(&mut buf), - ) - .await??; - received_bytes += 4 + len as u64; - - let packet = - Packet::try_from_slice(&buf).context("Failed to deserialize packet")?; - match packet { - Packet::Ping => { - // Do nothing. Pings are just for TCP keepalive. - } - Packet::MpcMessage(mpc_message) => { - message_sender.send(PeerMessage::Mpc(MpcPeerMessage { - from: peer_id, - message: mpc_message, - }))?; - } - Packet::IndexerHeight(message) => { - message_sender.send(PeerMessage::IndexerHeight( - PeerIndexerHeightMessage { - from: peer_id, - message, - }, - ))?; + let incoming_connections_task = + tracking::spawn("incoming connection handler tls mesh network", async move { + let mut tasks = AutoAbortTaskCollection::new(); + while let Ok((tcp_stream, _)) = tcp_listener.accept().await { + let message_sender = message_sender.clone(); + let participant_identities = participant_identities.clone(); + let tls_acceptor = tls_acceptor.clone(); + let connectivities = connectivities_clone.clone(); + tasks.spawn_checked::<_, ()>( + "new connection handler tls mesh network", + async move { + let tcp_stream = configure_tcp_stream(tcp_stream)?; + let mut stream = tls_acceptor.accept(tcp_stream).await?; + + let peer_id = + verify_peer_identity(stream.get_ref().1, &participant_identities)?; + tracking::set_progress(&format!("Authenticated as {}", peer_id)); + p2p_handshake(&mut stream, TlsConnection::HANDSHAKE_TIMEOUT) + .await + .context("p2p handshake")?; + tracing::info!("Incoming {} <-- {} connected", my_id, peer_id); + let incoming_conn = Arc::new(()); + connectivities + .get(peer_id)? + .set_incoming_connection(&incoming_conn); + let mut received_bytes: u64 = 0; + loop { + let len = tokio::time::timeout( + std::time::Duration::from_secs(MESSAGE_READ_TIMEOUT_SECS), + stream.read_u32(), + ) + .await??; + if len >= MAX_MESSAGE_LEN { + anyhow::bail!("Message too long"); + } + let mut buf = vec![0; len as usize]; + tokio::time::timeout( + std::time::Duration::from_secs(MESSAGE_READ_TIMEOUT_SECS), + stream.read_exact(&mut buf), + ) + .await??; + received_bytes += 4 + len as u64; + + let packet = Packet::try_from_slice(&buf) + .context("Failed to deserialize packet")?; + match packet { + Packet::Ping => { + // Do nothing. Pings are just for TCP keepalive. + } + Packet::MpcMessage(mpc_message) => { + message_sender.send(PeerMessage::Mpc(MpcPeerMessage { + from: peer_id, + message: mpc_message, + }))?; + } + Packet::IndexerHeight(message) => { + message_sender.send(PeerMessage::IndexerHeight( + PeerIndexerHeightMessage { + from: peer_id, + message, + }, + ))?; + } + } + + tracking::set_progress(&format!( + "Received {} bytes from {}", + received_bytes, peer_id + )); } - } - - tracking::set_progress(&format!( - "Received {} bytes from {}", - received_bytes, peer_id - )); - } - }); - } - }); + }, + ); + } + }); let sender = TlsMeshSender { my_id: config.my_participant_id, diff --git a/crates/node/src/protocol.rs b/crates/node/src/protocol.rs index fe8373ca3e..c505e1f89e 100644 --- a/crates/node/src/protocol.rs +++ b/crates/node/src/protocol.rs @@ -47,7 +47,7 @@ pub async fn run_protocol( let sending_handle = { let counters = counters.clone(); let sender = channel.sender(); - tracking::spawn_checked("send messages", async move { + tracking::spawn_checked("message senders for all participants", async move { // One future for each recipient. For the same recipient it is OK to send messages // serially, but for multiple recipients we want them to not block each other. // These futures are IO-bound, so we don't have to spawn them separately. diff --git a/crates/node/src/providers/ecdsa/presign.rs b/crates/node/src/providers/ecdsa/presign.rs index a3813578d8..1fb347360c 100644 --- a/crates/node/src/providers/ecdsa/presign.rs +++ b/crates/node/src/providers/ecdsa/presign.rs @@ -111,30 +111,33 @@ impl EcdsaSignatureProvider { let presignature_store = presignature_store.clone(); let config_clone = config.clone(); let keygen_out = keygen_out.clone(); - tasks.spawn_checked(&format!("{:?}", task_id), async move { - let _in_flight = in_flight; - let _semaphore_guard = parallelism_limiter.acquire().await?; - let presignature = PresignComputation { - threshold, - triple0, - triple1, - keygen_out, - } - .perform_leader_centric_computation( - channel, - Duration::from_secs(config_clone.timeout_sec), - ) - .await?; - presignature_store.add_owned( - id, - PresignOutputWithParticipants { - presignature, - participants, - }, - ); + tasks.spawn_checked( + &format!("ecdsa presign; task_id: {:?}", task_id), + async move { + let _in_flight = in_flight; + let _semaphore_guard = parallelism_limiter.acquire().await?; + let presignature = PresignComputation { + threshold, + triple0, + triple1, + keygen_out, + } + .perform_leader_centric_computation( + channel, + Duration::from_secs(config_clone.timeout_sec), + ) + .await?; + presignature_store.add_owned( + id, + PresignOutputWithParticipants { + presignature, + participants, + }, + ); - anyhow::Ok(()) - }); + anyhow::Ok(()) + }, + ); continue; } diff --git a/crates/node/src/providers/ecdsa/triple.rs b/crates/node/src/providers/ecdsa/triple.rs index 68ae04c7b5..5db518895f 100644 --- a/crates/node/src/providers/ecdsa/triple.rs +++ b/crates/node/src/providers/ecdsa/triple.rs @@ -122,26 +122,30 @@ impl EcdsaSignatureProvider { let parallelism_limiter = parallelism_limiter.clone(); let triple_store = triple_store.clone(); let config_clone = config.clone(); - tasks.spawn_checked(&format!("{:?}", task_id), async move { - let _in_flight = in_flight; - let _semaphore_guard = parallelism_limiter.acquire().await?; - let triples = (ManyTripleGenerationComputation::< - SUPPORTED_TRIPLE_GENERATION_BATCH_SIZE, - > { - threshold, - }) - .perform_leader_centric_computation( - channel, - Duration::from_secs(config_clone.timeout_sec), - ) - .await?; + tasks.spawn_checked( + &format!("background triple generation; task_id: {:?}", task_id), + async move { + let _in_flight = in_flight; + let _semaphore_guard = parallelism_limiter.acquire().await?; + let triples = (ManyTripleGenerationComputation::< + SUPPORTED_TRIPLE_GENERATION_BATCH_SIZE, + > { + threshold, + }) + .perform_leader_centric_computation( + channel, + Duration::from_secs(config_clone.timeout_sec), + ) + .await?; - for (i, paired_triple) in triples.into_iter().enumerate() { - triple_store.add_owned(id_start.add_to_counter(i as u32)?, paired_triple); - } + for (i, paired_triple) in triples.into_iter().enumerate() { + triple_store + .add_owned(id_start.add_to_counter(i as u32)?, paired_triple); + } - anyhow::Ok(()) - }); + anyhow::Ok(()) + }, + ); // Before issuing another one, wait a bit. This can dramatically // improve throughput by avoiding thundering herd situations. // Further optimization can be done to avoid thundering herd diff --git a/crates/node/src/tracking.rs b/crates/node/src/tracking.rs index 06ba217270..b09de8a431 100644 --- a/crates/node/src/tracking.rs +++ b/crates/node/src/tracking.rs @@ -10,7 +10,7 @@ use tokio::task_local; /// A wrapper around JoinHandle, except that dropping this will abort the task /// behind the handle. This is very useful in making sure that background tasks -/// spawned by futures that are then dropped are are properly cleaned up. +/// spawned by futures that are then dropped are properly cleaned up. #[must_use = "Dropping this value will immediately abort the task"] pub struct AutoAbortTask { handle: tokio::task::JoinHandle, @@ -287,7 +287,12 @@ impl TaskHandle { )); current_task = task.parent.clone(); } - tracing::error!("Task failed: {}; trace:{}", err, task_trace.join("")); + tracing::error!( + "task failed; description: {}; error msg: {}; trace:{}", + child.description, + err, + task_trace.join("") + ); } }) }