feat: add yield_now to drive reactor from sync code

src/driver.rs

@@ -293,6 +293,32 @@ pub fn block_on<T>(future: impl Future<Output = T>) -> T {
     })
 }
 
+/// Processes any pending I/O events without blocking.
+///
+/// This function is intended to be called periodically from synchronous code that needs to
+/// drive `async-io` reactor without using [`block_on()`]. If another thread is currently
+/// driving the reactor, this function returns immediately without doing any work.
+///
+/// # Examples
+///
+/// ```
+/// // Periodically yield to async-io tasks from synchronous code.
+/// for _ in 0..10 {
+///     // ... do some synchronous work ...
+///     async_io::yield_now();
+/// }
+/// ```
+pub fn yield_now() {
+    #[cfg(feature = "tracing")]
+    let span = tracing::trace_span!("async_io::yield_now");
+    #[cfg(feature = "tracing")]
+    let _enter = span.enter();
+
+    if let Some(mut reactor_lock) = Reactor::get().try_lock() {
+        reactor_lock.react(Some(Duration::from_secs(0))).ok();
+    }
+}
+
 /// Runs a closure when dropped.
 struct CallOnDrop<F: Fn()>(F);
 

src/lib.rs

@@ -93,7 +93,7 @@ mod reactor;
 
 pub mod os;
 
-pub use driver::block_on;
+pub use driver::{block_on, yield_now};
 pub use reactor::{Readable, ReadableOwned, Writable, WritableOwned};
 
 /// A future or stream that emits timed events.

tests/yield_now.rs

@@ -0,0 +1,32 @@
+use async_io::{yield_now, Timer};
+use std::time::{Duration, Instant};
+
+#[test]
+fn yield_now_returns_without_blocking() {
+    let start = Instant::now();
+    for _ in 0..10 {
+        yield_now();
+    }
+    assert!(start.elapsed() < Duration::from_secs(1));
+}
+
+#[test]
+fn yield_now_drives_timer() {
+    let start = Instant::now();
+    let timer = Timer::after(Duration::from_millis(50));
+    futures_lite::pin!(timer);
+
+    let waker = futures_lite::future::block_on(async { std::task::Waker::noop().clone() });
+    let mut cx = std::task::Context::from_waker(&waker);
+
+    use std::future::Future;
+    while timer.as_mut().poll(&mut cx).is_pending() {
+        if start.elapsed() > Duration::from_secs(5) {
+            panic!("timer never fired");
+        }
+        yield_now();
+        std::thread::sleep(Duration::from_millis(5));
+    }
+
+    assert!(start.elapsed() >= Duration::from_millis(50));
+}