chore: avoid redundant opendal runtime layer spawns (#20159)

fix(storage): avoid redundant IO runtime spawns

Author: dqhl76

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

@@ -25,6 +25,7 @@ use futures::future;
 use log::warn;
 use tokio::runtime::Builder;
 use tokio::runtime::Handle;
+use tokio::runtime::Id;
 use tokio::sync::OwnedSemaphorePermit;
 use tokio::sync::Semaphore;
 use tokio::task::JoinHandle;
@@ -60,7 +61,7 @@ impl Runtime {
 
         let handle = runtime.handle().clone();
         let runtime_name = name.clone().unwrap_or_else(|| "unnamed".to_string());
-        let runtime_id = handle.id().to_string();
+        let runtime_id = handle.id();
         let runtime_label = format!("{runtime_name} id={runtime_id}");
         let task_marker = format!("[{runtime_label}]");
 
@@ -152,6 +153,12 @@ impl Runtime {
         self.handle.clone()
     }
 
+    /// Returns whether the caller is running in this runtime.
+    #[inline]
+    pub fn is_current(&self) -> bool {
+        is_current_runtime(self.handle.id())
+    }
+
     fn task_location_name(&self, location_name: String) -> String {
         format!("{} {}", self.task_marker, location_name)
     }
@@ -298,7 +305,7 @@ impl Runtime {
 /// Dropping the dropper will cause runtime to shutdown.
 pub struct Dropper {
     name: Option<String>,
-    runtime_id: String,
+    runtime_id: Id,
     close: Option<std::sync::mpsc::Sender<WatchdogEvent>>,
     join_handler: Option<ThreadJoinHandle<bool>>,
 }
@@ -310,7 +317,7 @@ 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() {
+                if is_current_runtime(self.runtime_id) {
                     // 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
@@ -337,12 +344,11 @@ 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,
-        }
+#[inline]
+fn is_current_runtime(runtime_id: Id) -> bool {
+    match Handle::try_current() {
+        Ok(handle) => handle.id() == runtime_id,
+        Err(_) => false,
     }
 }