just use block_in_place

crates/polars-io/src/pl_async.rs

@@ -217,7 +217,6 @@ where
 
 pub struct RuntimeManager {
     rt: Runtime,
-    blocking_markers: Vec<AtomicBool>,
 }
 
 impl RuntimeManager {
@@ -229,14 +228,7 @@ impl RuntimeManager {
             .build()
             .unwrap();
 
-        let blocking_markers = (0..POOL.current_num_threads())
-            .map(|_| AtomicBool::new(false))
-            .collect();
-
-        Self {
-            rt,
-            blocking_markers,
-        }
+        Self { rt }
     }
 
     /// Keep track of rayon threads that drive the runtime. Every thread
@@ -248,29 +240,8 @@ impl RuntimeManager {
         F: Future + Send,
         F::Output: Send,
     {
-        let thread_idx = POOL.current_thread_index();
-
-        if thread_idx.is_none() {
-            self.rt.block_on(future)
-        } else {
-            let thread_idx = thread_idx.unwrap();
-            let tokio_entered = unsafe { self.blocking_markers.get_unchecked(thread_idx) };
-
-            if tokio_entered.load(Ordering::Relaxed) {
-                tokio_entered.store(true, Ordering::Relaxed);
-                let out = self.rt.block_on(future);
-                debug_assert_eq!(
-                    thread_idx,
-                    POOL.current_thread_index().unwrap(),
-                    "execution after block_on should be by the same rayon thread"
-                );
-                tokio_entered.store(false, Ordering::Relaxed);
-                out
-            } else {
-                // Safety: The tokio runtime flavor is multi-threaded
-                tokio::task::block_in_place(|| self.rt.block_on(future))
-            }
-        }
+        // Safety: The tokio runtime flavor is multi-threaded
+        tokio::task::block_in_place(|| self.rt.block_on(future))
     }
 
     pub fn block_on<F>(&self, future: F) -> F::Output