feat(query): try support dump running async task stack (#10830)

* refactor(query): add dump

* refactor(query): add framed for sql

* refactor(query): add framed for sharing-endpoint

* refactor(query): add framed for storages

* refactor(query): add framed for users

* refactor(query): add framed for services

* refactor(query): add framed for spawn

* refactor(query): try fix test failure

* refactor(query): try fix test failure

* refactor(query): add tasks sort

* refactor(query): add framed for storage

Author: zhang2014

Cargo.lock

@@ -145,6 +145,9 @@ reqwest = { version = "0.11", default-features = false, features = [
 # runtime
 tokio = { version = "1.26.0", features = ["full"] }
 
+# backtrace
+async-backtrace = "0.2.2"
+
 [profile.release]
 debug = 1
 lto = "thin"

Cargo.toml

@@ -60,6 +60,7 @@ sharing-endpoint = { path = "../query/sharing-endpoint" }
 # Crates.io dependencies
 anyerror = { workspace = true }
 anyhow = { workspace = true }
+async-backtrace = { workspace = true }
 clap = { workspace = true }
 comfy-table = "6.1.3"
 limits-rs = "0.2.0"

src/binaries/Cargo.toml

@@ -54,7 +54,7 @@ fn main() {
             std::process::exit(cause.code() as i32);
         }
         Ok(rt) => {
-            if let Err(cause) = rt.block_on(main_entrypoint()) {
+            if let Err(cause) = rt.block_on(async_backtrace::location!().frame(main_entrypoint())) {
                 eprintln!("Databend Query start failure, cause: {:?}", cause);
                 std::process::exit(cause.code() as i32);
             }

src/binaries/query/main.rs

@@ -26,6 +26,7 @@ common-exception = { path = "../exception" }
 # Github dependencies
 
 # Crates.io dependencies
+async-backtrace = "0.2.2"
 async-channel = "1.7.1"
 async-trait = "0.1.57"
 bytesize = "1.1.0"

src/common/base/Cargo.toml

@@ -155,6 +155,8 @@ impl Runtime {
                     runtime_builder.thread_name(thread_name);
                 }
             }
+
+            runtime_builder.thread_stack_size(5 * 1024 * 1024);
         }
 
         Self::create(None, mem_stat, &mut runtime_builder)
@@ -179,6 +181,8 @@ impl Runtime {
                     thread_name = Some(cur_thread_name.to_string());
                 }
             }
+
+            runtime_builder.thread_stack_size(5 * 1024 * 1024);
         }
 
         if let Some(thread_name) = &thread_name {
@@ -255,10 +259,12 @@ impl Runtime {
             let permit = semaphore.acquire_owned().await.map_err(|e| {
                 ErrorCode::Internal(format!("semaphore closed, acquire permit failure. {}", e))
             })?;
-            let handler = self.handle.spawn(async move {
-                // take the ownership of the permit, (implicitly) drop it when task is done
-                fut(permit).await
-            });
+            let handler = self
+                .handle
+                .spawn(async_backtrace::location!().frame(async move {
+                    // take the ownership of the permit, (implicitly) drop it when task is done
+                    fut(permit).await
+                }));
             handlers.push(handler)
         }
 
@@ -273,7 +279,7 @@ impl TrySpawn for Runtime {
         T: Future + Send + 'static,
         T::Output: Send + 'static,
     {
-        Ok(self.handle.spawn(task))
+        Ok(self.handle.spawn(async_backtrace::location!().frame(task)))
     }
 }
 

src/common/base/src/runtime/runtime.rs

@@ -670,7 +670,7 @@ mod tests {
                     .unwrap();
 
                 rt.block_on(async {
-                    let h = tokio::spawn(f);
+                    let h = tokio::spawn(async_backtrace::location!().frame(f));
                     let res = h.await;
                     assert!(res.is_err(), "panicked");
                 });

src/common/base/src/runtime/runtime_tracker.rs

@@ -63,6 +63,8 @@ impl Thread {
                     name = Some(thread_name.to_string());
                 }
             }
+
+            thread_builder = thread_builder.stack_size(5 * 1024 * 1024);
         }
 
         let mut mem_stat_name = String::from("UnnamedThread");

src/common/base/src/runtime/thread.rs

@@ -85,6 +85,7 @@ async fn test_shutdown_long_run_runtime() -> Result<()> {
 }
 
 static START_TIME: Lazy<Instant> = Lazy::new(Instant::now);
+
 // println can more clearly know if they are parallel
 async fn mock_get_page(i: usize) -> Vec<usize> {
     let millis = Uniform::from(0..10).sample(&mut rand::thread_rng());

src/common/base/tests/it/runtime.rs

@@ -24,6 +24,7 @@ common-exception = { path = "../exception" }
 
 # Crates.io dependencies
 anyerror = { workspace = true }
+async-backtrace = { workspace = true }
 futures = "0.3.24"
 poem = { version = "1", features = ["rustls"] }
 serde = { workspace = true }