chore: improve log messages for tokio tasks

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(),

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,

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 },
             );
         }

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,

crates/node/src/protocol.rs

@@ -47,7 +47,7 @@ pub async fn run_protocol<T>(
     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.

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;
             }
 

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

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<R> {
     handle: tokio::task::JoinHandle<R>,
@@ -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("")
+                );
             }
         })
     }