fix: deadlock for TokioMultiThreadExecutor (delta-io#1606)

## 🥞 Stacked PR
Use this
[link](https://github.com/delta-io/delta-kernel-rs/pull/1606/files) to
review incremental changes.
-
[**stack/dead-wait**](delta-io#1606)
[[Files
changed](https://github.com/delta-io/delta-kernel-rs/pull/1606/files)]
-
[stack/rename-checkpoint](delta-io#1608)
[[Files
changed](https://github.com/delta-io/delta-kernel-rs/pull/1608/files/d0750b490f0f817bf0fd185dd17c34314eb77f77..1bdeda0c00304a685baa8a655779ace8d34e74fc)]
-
[stack/perform-checkpoint](delta-io#1600)
[[Files
changed](https://github.com/delta-io/delta-kernel-rs/pull/1600/files/1bdeda0c00304a685baa8a655779ace8d34e74fc..847883e1503277d42da7039c975098ae8e587666)]
-
[stack/ffi-checkpoint](delta-io#1611)
[[Files
changed](https://github.com/delta-io/delta-kernel-rs/pull/1611/files/847883e1503277d42da7039c975098ae8e587666..17b307b2f9b34f9e5a6e2b2a33d0ea008b1634e1)]

---------
## What changes are proposed in this pull request?
Fix delta-io#1605
<!--
**Uncomment** this section if there are any changes affecting public
APIs. Else, **delete** this section.

### This PR affects the following public APIs

If there are breaking changes, please ensure the `breaking-changes`
label gets added by CI, and describe why the changes are needed.

Note that _new_ public APIs are not considered breaking.
-->

## How was this change tested?
Added unit test

---------

Co-authored-by: Ryan Johnson <scovich@users.noreply.github.com>

Author: dengsh12

kernel/src/engine/default/executor.rs

@@ -169,6 +169,9 @@ pub mod tokio {
     }
 
     impl TaskExecutor for TokioMultiThreadExecutor {
+        // `block_on` uses `block_in_place`; If concurrent `block_on` calls exceed Tokio's `max_blocking_threads`, this can deadlock
+        // See:
+        // https://docs.rs/tokio/latest/tokio/runtime/struct.Builder.html#method.max_blocking_threads
         fn block_on<T>(&self, task: T) -> T::Output
         where
             T: Future + Send + 'static,
@@ -191,9 +194,14 @@ pub mod tokio {
             // We throw away the handle, but it should continue on.
             self.handle.spawn(fut);
 
-            receiver
-                .recv()
-                .expect("TokioMultiThreadExecutor has crashed")
+            // Use block_in_place to tell Tokio we're about to block - this allows
+            // the runtime to move tasks off this worker's local queue so they can
+            // be stolen by other workers.
+            tokio::task::block_in_place(|| {
+                receiver
+                    .recv()
+                    .expect("TokioMultiThreadExecutor has crashed")
+            })
         }
 
         fn spawn<F>(&self, task: F)
@@ -246,5 +254,39 @@ pub mod tokio {
             let executor = TokioMultiThreadExecutor::new(tokio::runtime::Handle::current());
             test_executor(executor).await;
         }
+
+        #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+        async fn test_nested_block_on_does_not_deadlock() {
+            use std::sync::Arc;
+            use std::time::Duration;
+
+            let executor = Arc::new(TokioMultiThreadExecutor::new(
+                tokio::runtime::Handle::current(),
+            ));
+            let executor_clone = executor.clone();
+
+            let (tx, rx) = channel::<i32>();
+
+            let handle = std::thread::spawn(move || {
+                // Outer block_on
+                let result = executor.block_on(async move {
+                    // Inner block_on
+                    let inner_result = executor_clone.block_on(async {
+                        tokio::time::sleep(Duration::from_millis(1)).await;
+                        42
+                    });
+                    inner_result + 1
+                });
+                tx.send(result).ok();
+            });
+
+            // Wait with timeout - if this times out, we have a deadlock
+            let timeout = Duration::from_secs(5);
+            let result = rx
+                .recv_timeout(timeout)
+                .expect("Timeout - likely deadlock in TokioMultiThreadExecutor::block_on");
+            assert_eq!(result, 43);
+            handle.join().expect("thread panicked");
+        }
     }
 }