async_backtrace

crates/librqbit/src/tests/e2e.rs

@@ -14,7 +14,9 @@ use tracing::{error_span, info, Instrument};
 
 use crate::{
     create_torrent,
-    tests::test_util::{create_default_random_dir_with_torrents, TestPeerMetadata},
+    tests::test_util::{
+        create_default_random_dir_with_torrents, spawn_debug_server, TestPeerMetadata,
+    },
     AddTorrentOptions, AddTorrentResponse, Session, SessionOptions,
 };
 
@@ -30,6 +32,8 @@ async fn test_e2e_download() {
 async fn _test_e2e_download() {
     let _ = tracing_subscriber::fmt::try_init();
 
+    spawn_debug_server();
+
     // 1. Create a torrent
     // Ideally (for a more complicated test) with N files, and at least N pieces that span 2 files.
 

crates/librqbit/src/tests/test_util.rs

@@ -1,8 +1,11 @@
 use std::{io::Write, path::Path};
 
+use anyhow::Context;
+use axum::{response::IntoResponse, routing::get, Router};
 use librqbit_core::Id20;
 use rand::{thread_rng, Rng, RngCore, SeedableRng};
 use tempfile::TempDir;
+use tracing::info;
 
 pub fn create_new_file_with_random_content(path: &Path, mut size: usize) {
     let mut file = std::fs::OpenOptions::new()
@@ -79,3 +82,28 @@ impl TestPeerMetadata {
         0f64
     }
 }
+
+async fn debug_server() -> anyhow::Result<()> {
+    async fn backtraces() -> impl IntoResponse {
+        async_backtrace::taskdump_tree(true)
+    }
+
+    let app = Router::new().route("/backtrace", get(backtraces));
+    let app = app.into_make_service();
+
+    let addr = "127.0.0.1:3032";
+
+    info!(%addr, "starting HTTP server");
+
+    use tokio::net::TcpListener;
+
+    let listener = TcpListener::bind(addr)
+        .await
+        .with_context(|| format!("error binding to {addr}"))?;
+    axum::serve(listener, app).await?;
+    Ok(())
+}
+
+pub fn spawn_debug_server() {
+    tokio::spawn(debug_server());
+}

crates/librqbit/src/torrent_state/live/mod.rs

@@ -353,6 +353,7 @@ impl TorrentStateLive {
         Ok(())
     }
 
+    #[async_backtrace::framed]
     async fn task_manage_incoming_peer(
         self: Arc<Self>,
         checked_peer: CheckedIncomingConnection,
@@ -413,6 +414,7 @@ impl TorrentStateLive {
         Ok(())
     }
 
+    #[async_backtrace::framed]
     async fn task_manage_outgoing_peer(
         self: Arc<Self>,
         addr: SocketAddr,
@@ -449,12 +451,12 @@ impl TorrentStateLive {
             state.meta.spawner,
             state.meta.connector.clone(),
         );
-        let requester = handler
+        let requester = async_backtrace::frame!(handler
             .task_peer_chunk_requester()
-            .instrument(error_span!("chunk_requester"));
-        let conn_manager = peer_connection
+            .instrument(error_span!("chunk_requester")));
+        let conn_manager = async_backtrace::frame!(peer_connection
             .manage_peer_outgoing(rx, state.have_broadcast_tx.subscribe())
-            .instrument(error_span!("peer_connection"));
+            .instrument(error_span!("peer_connection")));
 
         handler
             .counters
@@ -1213,7 +1215,7 @@ impl PeerHandler {
         }
 
         loop {
-            self.wait_for_unchoke().await;
+            async_backtrace::frame!(self.wait_for_unchoke()).await;
 
             if self.state.is_finished_and_dont_need_peers() {
                 debug!("nothing left to do, disconnecting peer");
@@ -1232,7 +1234,7 @@ impl PeerHandler {
                 Some(next) => next,
                 None => {
                     debug!("no pieces to request");
-                    tokio::time::sleep(Duration::from_secs(10)).await;
+                    async_backtrace::frame!(tokio::time::sleep(Duration::from_secs(10))).await;
                     continue;
                 }
             };
@@ -1266,7 +1268,12 @@ impl PeerHandler {
                 };
 
                 loop {
-                    match timeout(Duration::from_secs(10), self.requests_sem.acquire()).await {
+                    match async_backtrace::frame!(timeout(
+                        Duration::from_secs(10),
+                        async_backtrace::frame!(self.requests_sem.acquire())
+                    ))
+                    .await
+                    {
                         Ok(acq) => break acq?.forget(),
                         Err(_) => continue,
                     };