fix(base): avoid self-deadlock during runtime drop (#20103)

fix(runtime): avoid self-deadlock during runtime drop

Author: dqhl76

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

@@ -82,7 +82,8 @@ impl Runtime {
 
                 if cfg!(debug_assertions) {
                     let instant = Instant::now();
-                    // We wait up to 3 seconds to complete the runtime shutdown.
+                    // In debug builds, bound the Tokio runtime shutdown wait to 3 seconds.
+                    // The dropper joining this wait-to-drop thread is still an unbounded wait.
                     runtime.shutdown_timeout(Duration::from_secs(3));
                     instant.elapsed() >= Duration::from_secs(3)
                 } else {
@@ -95,6 +96,7 @@ impl Runtime {
             task_marker,
             _dropper: Dropper {
                 name,
+                runtime_id,
                 close: Some(watchdog_tx),
                 join_handler: Some(join_handler),
             },
@@ -296,6 +298,7 @@ impl Runtime {
 /// Dropping the dropper will cause runtime to shutdown.
 pub struct Dropper {
     name: Option<String>,
+    runtime_id: String,
     close: Option<std::sync::mpsc::Sender<WatchdogEvent>>,
     join_handler: Option<ThreadJoinHandle<bool>>,
 }
@@ -307,11 +310,20 @@ impl Drop for Dropper {
             if let Some(close_sender) = self.close.take()
                 && close_sender.send(WatchdogEvent::Stop).is_ok()
             {
+                if self.is_dropping_from_own_runtime() {
+                    // The wait-to-drop thread owns the Tokio runtime and will shut it down
+                    // after observing the stop signal. Joining it from one of the same
+                    // runtime's workers would deadlock: shutdown waits for this worker to
+                    // exit, while this worker waits for shutdown to finish.
+                    drop(self.join_handler.take());
+                    return;
+                }
+
                 match self.join_handler.take().unwrap().join() {
                     Err(e) => warn!("Runtime dropper panic, {:?}", e),
                     Ok(true) => {
-                        // When the runtime shutdown is blocked for more than 3 seconds,
-                        // we will print the backtrace in the warn log, which will help us debug.
+                        // If the debug shutdown timeout is fully consumed, log the
+                        // drop-site backtrace to help diagnose blocked runtime tasks.
                         warn!(
                             "Runtime dropper is blocked 3 seconds, runtime name: {:?}, drop backtrace: {:?}",
                             self.name,
@@ -325,6 +337,15 @@ impl Drop for Dropper {
     }
 }
 
+impl Dropper {
+    fn is_dropping_from_own_runtime(&self) -> bool {
+        match Handle::try_current() {
+            Ok(handle) => handle.id().to_string() == self.runtime_id,
+            Err(_) => false,
+        }
+    }
+}
+
 /// Run multiple futures parallel
 /// using a semaphore to limit the parallelism number, and a specified thread pool to run the futures.
 /// It waits for all futures to complete and returns their results.

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

@@ -15,6 +15,7 @@
 use std::sync::Arc;
 use std::sync::LazyLock;
 use std::sync::Mutex;
+use std::sync::mpsc;
 use std::time::Duration;
 use std::time::Instant;
 
@@ -86,6 +87,37 @@ async fn test_shutdown_long_run_runtime() -> anyhow::Result<()> {
     Ok(())
 }
 
+struct RuntimeOwner {
+    _runtime: Arc<Runtime>,
+}
+
+#[test]
+fn test_runtime_can_be_dropped_from_own_worker() -> anyhow::Result<()> {
+    let runtime = Arc::new(Runtime::with_worker_threads(
+        1,
+        Some("drop-runtime-from-own-worker".to_string()),
+    )?);
+    let owner = Arc::new(RuntimeOwner {
+        _runtime: runtime.clone(),
+    });
+    let owner_for_task = owner.clone();
+    let (release_tx, release_rx) = mpsc::channel();
+    let (done_tx, done_rx) = mpsc::channel();
+
+    runtime.spawn(async move {
+        release_rx.recv().unwrap();
+        drop(owner_for_task);
+        let _ = done_tx.send(());
+    });
+
+    drop(owner);
+    drop(runtime);
+    release_tx.send(())?;
+
+    done_rx.recv_timeout(Duration::from_secs(2))?;
+    Ok(())
+}
+
 #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
 async fn test_runtime_task_dump_contains_runtime_id() -> anyhow::Result<()> {
     let runtime = Runtime::with_worker_threads(1, Some("task-marker-runtime".to_string()))?;