Functional parity with linux

lib/runtime/src/platform/linux_x86_64/runtime.rs

@@ -478,6 +478,99 @@ pub fn run() {
     }
 }
 
+/// Drains ready work on the current runtime thread without blocking for future work.
+///
+/// Unlike [`run`], this returns as soon as there are no immediately runnable
+/// microtasks or macrotasks left. It is intended for host integrations that
+/// need to re-enter the scheduler while an outer platform loop remains active.
+pub fn run_until_stalled() {
+    let _ = current_thread();
+
+    loop {
+        drain_all();
+
+        let mut microtasks_run: u64 = 0;
+        while let Some(task) = pop_microtask() {
+            task();
+            microtasks_run += 1;
+            drain_all();
+        }
+        if microtasks_run >= MICROTASK_STARVATION_THRESHOLD {
+            tracing::warn!(
+                target: trace_targets::SCHEDULER,
+                event = "microtask_starvation",
+                count = microtasks_run,
+                "microtask queue ran {microtasks_run} tasks in a single turn; macrotask handlers may be starved",
+            );
+        }
+
+        if let Some(task) = pop_macrotask() {
+            task();
+            continue;
+        }
+
+        drain_all();
+
+        if has_ready_work() {
+            continue;
+        }
+
+        current_thread()
+            .shared
+            .closing
+            .store(false, Ordering::Release);
+        return;
+    }
+}
+
+/// Drains already-queued work on the current runtime thread without polling the
+/// driver for timers or I/O readiness.
+///
+/// This is intended for host integrations that need to flush application work
+/// from inside a host callback without re-entering timer callbacks.
+pub fn run_ready_tasks() {
+    let _ = current_thread();
+
+    loop {
+        drain_remote_tasks();
+        drain_completed_workers();
+
+        let mut microtasks_run: u64 = 0;
+        while let Some(task) = pop_microtask() {
+            task();
+            microtasks_run += 1;
+            drain_remote_tasks();
+            drain_completed_workers();
+        }
+        if microtasks_run >= MICROTASK_STARVATION_THRESHOLD {
+            tracing::warn!(
+                target: trace_targets::SCHEDULER,
+                event = "microtask_starvation",
+                count = microtasks_run,
+                "microtask queue ran {microtasks_run} tasks in a single turn; macrotask handlers may be starved",
+            );
+        }
+
+        if let Some(task) = pop_macrotask() {
+            task();
+            continue;
+        }
+
+        drain_remote_tasks();
+        drain_completed_workers();
+
+        if has_ready_work() {
+            continue;
+        }
+
+        current_thread()
+            .shared
+            .closing
+            .store(false, Ordering::Release);
+        return;
+    }
+}
+
 fn drain_all() {
     drain_driver_events();
     drain_remote_tasks();

lib/runtime/src/platform/macos_aarch64/runtime.rs

@@ -480,6 +480,99 @@ pub fn run() {
     }
 }
 
+/// Drains ready work on the current runtime thread without blocking for future work.
+///
+/// Unlike [`run`], this returns as soon as there are no immediately runnable
+/// microtasks or macrotasks left. It is intended for host integrations that
+/// need to re-enter the scheduler while an outer platform loop remains active.
+pub fn run_until_stalled() {
+    let _ = current_thread();
+
+    loop {
+        drain_all();
+
+        let mut microtasks_run: u64 = 0;
+        while let Some(task) = pop_microtask() {
+            task();
+            microtasks_run += 1;
+            drain_all();
+        }
+        if microtasks_run >= MICROTASK_STARVATION_THRESHOLD {
+            tracing::warn!(
+                target: trace_targets::SCHEDULER,
+                event = "microtask_starvation",
+                count = microtasks_run,
+                "microtask queue ran {microtasks_run} tasks in a single turn; macrotask handlers may be starved",
+            );
+        }
+
+        if let Some(task) = pop_macrotask() {
+            task();
+            continue;
+        }
+
+        drain_all();
+
+        if has_ready_work() {
+            continue;
+        }
+
+        current_thread()
+            .shared
+            .closing
+            .store(false, Ordering::Release);
+        return;
+    }
+}
+
+/// Drains already-queued work on the current runtime thread without polling the
+/// driver for timers or I/O readiness.
+///
+/// This is intended for host integrations that need to flush application work
+/// from inside a host callback without re-entering timer callbacks.
+pub fn run_ready_tasks() {
+    let _ = current_thread();
+
+    loop {
+        drain_remote_tasks();
+        drain_completed_workers();
+
+        let mut microtasks_run: u64 = 0;
+        while let Some(task) = pop_microtask() {
+            task();
+            microtasks_run += 1;
+            drain_remote_tasks();
+            drain_completed_workers();
+        }
+        if microtasks_run >= MICROTASK_STARVATION_THRESHOLD {
+            tracing::warn!(
+                target: trace_targets::SCHEDULER,
+                event = "microtask_starvation",
+                count = microtasks_run,
+                "microtask queue ran {microtasks_run} tasks in a single turn; macrotask handlers may be starved",
+            );
+        }
+
+        if let Some(task) = pop_macrotask() {
+            task();
+            continue;
+        }
+
+        drain_remote_tasks();
+        drain_completed_workers();
+
+        if has_ready_work() {
+            continue;
+        }
+
+        current_thread()
+            .shared
+            .closing
+            .store(false, Ordering::Release);
+        return;
+    }
+}
+
 fn drain_all() {
     drain_driver_events();
     drain_remote_tasks();