perf(runtime): eliminate redundant asyncpipe microtask work

changelog.d/8409-redundant-microtask-wakes.md

@@ -0,0 +1,10 @@
+## Suppress redundant promise-drain wakeups
+
+Promise and `queueMicrotask` work queued by the main thread while it is already
+draining microtasks no longer signals the event loop. Cross-thread producers,
+timer callbacks, and rejection processing retain the normal wake path.
+
+`Promise.all` over plain native promises also uses compact settlement records
+instead of allocating a closure and `AlreadyCalled` array for every element;
+observable constructor, resolver, `then`, and lifecycle-hook paths still use
+the spec dispatch.

crates/perry-runtime/src/event_pump.rs

@@ -231,6 +231,7 @@ static NOTIFIED: AtomicBool = AtomicBool::new(false);
 static WAITER_COUNT: AtomicI64 = AtomicI64::new(0);
 pub static PROFILE_NOTIFY_COUNT: AtomicI64 = AtomicI64::new(0);
 pub static PROFILE_NOTIFY_DURING_DRAIN_COUNT: AtomicI64 = AtomicI64::new(0);
+pub static PROFILE_NOTIFY_DRAIN_SUPPRESSED_COUNT: AtomicI64 = AtomicI64::new(0);
 pub static PROFILE_WAIT_COUNT: AtomicI64 = AtomicI64::new(0);
 pub static PROFILE_WAIT_FAST_COUNT: AtomicI64 = AtomicI64::new(0);
 pub static PROFILE_WAIT_ZERO_COUNT: AtomicI64 = AtomicI64::new(0);
@@ -367,6 +368,23 @@ pub extern "C" fn js_notify_main_thread() {
     PUMP.cvar.notify_one();
 }
 
+/// Notify after queueing promise/microtask work, unless this same thread is
+/// already draining that queue to quiescence.
+///
+/// Cross-thread producers see a thread-local drain depth of zero and retain
+/// the full wake path. Timer and rejection phases are outside the job-drain
+/// scope as well, because work they queue is consumed by a later pump turn.
+#[inline(always)]
+pub(crate) fn js_notify_promise_progress() {
+    if crate::promise::microtasks::microtask_job_drain_active() {
+        if crate::promise::mt_profile_enabled() {
+            PROFILE_NOTIFY_DRAIN_SUPPRESSED_COUNT.fetch_add(1, Ordering::Relaxed);
+        }
+        return;
+    }
+    js_notify_main_thread();
+}
+
 // ============================================================================
 // #1088 — Unified Event Loop FFI facade for host embedding.
 //
@@ -673,6 +691,44 @@ mod tests {
     /// timer state — there is no per-thread injection point.)
     static SERIAL: StdMutex<()> = StdMutex::new(());
 
+    /// A promise settled while promise jobs are already draining must not
+    /// leave a redundant event-loop wake behind. The active runner consumes
+    /// the propagated job before returning; outside that scope the same
+    /// helper must still publish a wake.
+    #[test]
+    fn promise_progress_notify_is_suppressed_during_job_drain() {
+        let _g = SERIAL.lock().unwrap();
+
+        NOTIFIED.store(false, Ordering::Release);
+        js_notify_promise_progress();
+        assert!(NOTIFIED.swap(false, Ordering::Acquire));
+
+        let undefined = f64::from_bits(crate::value::TAG_UNDEFINED);
+        let mut observed_clean_drain = false;
+        for _ in 0..32 {
+            let promise = crate::promise::js_promise_new();
+            crate::promise::js_promise_then(promise, std::ptr::null(), std::ptr::null());
+            crate::promise::js_promise_resolve(promise, undefined);
+
+            // Consume the required outside-drain notify, then run only the
+            // promise-job phase so unrelated global timers cannot wake us.
+            NOTIFIED.store(false, Ordering::Release);
+            crate::promise::microtasks::js_promise_run_microtasks_checkpoint();
+            if !NOTIFIED.load(Ordering::Acquire) {
+                observed_clean_drain = true;
+                break;
+            }
+            // A parallel runtime test may use the process-global notifier;
+            // retry until we observe an uncontended drain.
+            NOTIFIED.store(false, Ordering::Release);
+        }
+
+        assert!(
+            observed_clean_drain,
+            "promise propagation repeatedly left an event-loop wake behind"
+        );
+    }
+
     struct ForcedZeroBudgetGuard;
 
     impl ForcedZeroBudgetGuard {

crates/perry-runtime/src/promise/assimilate.rs

@@ -151,7 +151,7 @@ pub(super) fn enqueue_thenable_job(promise: *mut Promise, thenable: f64, then_ac
             trigger_async_id: ids.trigger_async_id,
         });
     });
-    crate::event_pump::js_notify_main_thread();
+    crate::event_pump::js_notify_promise_progress();
 }
 
 pub(crate) fn promise_resolve_assimilating(promise: *mut Promise, value: f64) {
@@ -243,7 +243,7 @@ pub(super) fn enqueue_native_adoption_job(outer: *mut Promise, inner: *mut Promi
             trigger_async_id: ids.trigger_async_id,
         });
     });
-    crate::event_pump::js_notify_main_thread();
+    crate::event_pump::js_notify_promise_progress();
 }
 
 /// Job body — the intrinsic-`then` invocation of the adoption job. For a
@@ -280,7 +280,7 @@ extern "C" fn native_promise_adoption_job(closure: *const crate::closure::Closur
                     std::ptr::null_mut(),
                 ));
             });
-            crate::event_pump::js_notify_main_thread();
+            crate::event_pump::js_notify_promise_progress();
         }
         None => {
             super::then::js_promise_resolve_with_promise(outer, inner);

crates/perry-runtime/src/promise/async_step.rs

@@ -270,7 +270,7 @@ pub extern "C" fn js_promise_resolved_then(
                 capture_context(),
             ));
         });
-        crate::event_pump::js_notify_main_thread();
+        crate::event_pump::js_notify_promise_progress();
         // Suppress the rejection-handler bookkeeping: it would only
         // matter if `value` were a Promise, which it isn't here.
         let _ = on_rejected;
@@ -594,7 +594,7 @@ pub extern "C" fn js_async_step_chain(value: f64, step_closure: ClosurePtr) -> *
             trap.box_activation,
         ));
     });
-    crate::event_pump::js_notify_main_thread();
+    crate::event_pump::js_notify_promise_progress();
     next
 }
 

crates/perry-runtime/src/promise/combinators.rs

@@ -60,6 +60,53 @@ pub(super) fn promise_all_settle(state: PromiseAllState, value: f64, is_fulfille
     }
 }
 
+/// Attach the allocation-free `Promise.all` state used when every observable
+/// combinator hook has been ruled out. Settled inputs become ordinary
+/// microtasks; pending inputs park the state in the same GC-scanned keyed table
+/// that `js_promise_resolve` drains on settlement.
+pub(super) fn attach_promise_all_state(promise: *mut Promise, state: PromiseAllState) {
+    if promise.is_null() {
+        return;
+    }
+    mark_rejection_handled(promise);
+    let mut queued = false;
+    unsafe {
+        match (*promise).state {
+            PromiseState::Fulfilled => {
+                TASK_QUEUE.with(|q| {
+                    q.borrow_mut().push_back(Task::PromiseAll(
+                        state,
+                        (*promise).value,
+                        true,
+                        context_for_promise(promise),
+                    ));
+                });
+                queued = true;
+            }
+            PromiseState::Rejected => {
+                TASK_QUEUE.with(|q| {
+                    q.borrow_mut().push_back(Task::PromiseAll(
+                        state,
+                        (*promise).reason,
+                        false,
+                        context_for_promise(promise),
+                    ));
+                });
+                queued = true;
+            }
+            PromiseState::Pending => {
+                PROMISE_ALL_STATES.with(|states| {
+                    states.borrow_mut().push(promise as usize, state);
+                });
+                set_promise_callback_context(promise);
+            }
+        }
+    }
+    if queued {
+        crate::event_pump::js_notify_promise_progress();
+    }
+}
+
 pub(super) fn scan_promise_all_states_mut(visitor: &mut crate::gc::RuntimeRootVisitor<'_>) {
     PROMISE_ALL_STATES.with(|states| {
         let mut states = states.borrow_mut();

crates/perry-runtime/src/promise/microtasks.rs

@@ -4,6 +4,12 @@
 
 use super::*;
 
+#[derive(Clone, Copy)]
+struct MicrotaskRunDepths {
+    pump: u32,
+    jobs: u32,
+}
+
 crate::perry_thread_local! {
     /// Promise currently being dispatched by the microtask runner after its
     /// task has been popped from TASK_QUEUE. While user callbacks run this is
@@ -22,14 +28,21 @@ crate::perry_thread_local! {
     pub(super) static CURRENT_MICROTASK_NEXT: std::cell::Cell<*mut Promise>
         = const { std::cell::Cell::new(std::ptr::null_mut()) };
 
-    /// Nesting depth for `js_promise_run_microtasks` on this thread.
+    /// Nesting depths for `js_promise_run_microtasks` on this thread.
     ///
     /// Await lowering can re-enter the microtask runner from inside a
     /// microtask or timer callback. Re-entrant drains may run promise jobs,
     /// but they must not recursively enter the timer queues: timers are
     /// macrotasks, and running them from a nested microtask checkpoint can
     /// build an unbounded stack of exception traps.
-    static MICROTASK_RUN_DEPTH: std::cell::Cell<u32> = const { std::cell::Cell::new(0) };
+    /// `pump` covers the whole checkpoint. `jobs` covers only the part that is
+    /// about to drain (or is actively draining) promise/queueMicrotask jobs;
+    /// it deliberately ends before rejection processing and timer phases, as
+    /// work queued there needs to wake an immediately-following event-loop
+    /// wait. Keep both counters in this already-audited TLS holder.
+    static MICROTASK_RUN_DEPTH: std::cell::Cell<MicrotaskRunDepths> = const {
+        std::cell::Cell::new(MicrotaskRunDepths { pump: 0, jobs: 0 })
+    };
 
     /// One-shot: the entry module is ESM and its evaluation checkpoint has
     /// not happened yet. Consumed by the first `run_microtasks` drain, which
@@ -39,15 +52,22 @@ crate::perry_thread_local! {
     static ESM_EVAL_CHECKPOINT_PENDING: std::cell::Cell<bool> = const { std::cell::Cell::new(false) };
 }
 
-/// Whether this thread is already executing a microtask checkpoint.
-///
-/// A same-thread enqueue during a checkpoint does not need to wake the event
-/// loop: the runner drains the queue to quiescence before returning. Producers
-/// on other threads observe their own depth (zero) and still take the normal
-/// cross-thread wake path.
+/// Whether this thread is already executing a microtask pump, including its
+/// rejection and timer phases. Used by the event-path profiler.
 #[inline(always)]
 pub(crate) fn microtask_drain_active() -> bool {
-    MICROTASK_RUN_DEPTH.with(|depth| depth.get() != 0)
+    MICROTASK_RUN_DEPTH.with(|depth| depth.get().pump != 0)
+}
+
+/// Whether a same-thread promise job enqueue will be consumed by the active
+/// drain before it can return to the event loop.
+///
+/// Producers on other threads observe their own depth (zero) and still take
+/// the normal cross-thread wake path. The timer/rejection tail also observes
+/// zero so work queued there wakes the next event-loop wait.
+#[inline(always)]
+pub(crate) fn microtask_job_drain_active() -> bool {
+    MICROTASK_RUN_DEPTH.with(|depth| depth.get().jobs != 0)
 }
 
 /// Called once from the compiled entry (before top-level statements) when the
@@ -186,9 +206,12 @@ fn run_microtasks(mode: MicrotaskDrainMode) -> i32 {
     bump(&MT_DRAIN_COUNT);
     let async_box_ref_depth = async_box_execution_ref_depth();
     let reentrant = MICROTASK_RUN_DEPTH.with(|depth| {
-        let current = depth.get();
-        depth.set(current.saturating_add(1));
-        current > 0
+        let mut current = depth.get();
+        let reentrant = current.pump > 0;
+        current.pump = current.pump.saturating_add(1);
+        current.jobs = current.jobs.saturating_add(1);
+        depth.set(current);
+        reentrant
     });
     let mut ran = 0;
 
@@ -970,6 +993,15 @@ fn run_microtasks(mode: MicrotaskDrainMode) -> i32 {
         }
     }
 
+    // Promise/queueMicrotask jobs are quiescent. Notifications after this
+    // point (rejection processing or timer callbacks) must remain observable:
+    // the generated event loop may wait before its next drain.
+    MICROTASK_RUN_DEPTH.with(|depth| {
+        let mut current = depth.get();
+        current.jobs = current.jobs.saturating_sub(1);
+        depth.set(current);
+    });
+
     // #6077: the microtask checkpoint is over — the queue drained to empty.
     // This is where Node decides whether a rejection went unhandled
     // (`processTicksAndRejections` → `processPromiseRejections`), BEFORE the
@@ -1019,7 +1051,7 @@ fn run_microtasks(mode: MicrotaskDrainMode) -> i32 {
     // compacting, O(survivors) young collection instead of the non-moving
     // alloc-point fallback. Gated (default off); additive.
     if crate::gc::gc_moving_safepoint_enabled()
-        && MICROTASK_RUN_DEPTH.with(|depth| depth.get()) == 1
+        && MICROTASK_RUN_DEPTH.with(|depth| depth.get().pump) == 1
     {
         crate::gc::gc_safepoint_moving_minor();
     }
@@ -1028,14 +1060,16 @@ fn run_microtasks(mode: MicrotaskDrainMode) -> i32 {
     // activation (principally direct runtime tests). Production async frames
     // publish at their own queued/running AsyncStep refcount reaching zero;
     // they do not wait for this global pump boundary.
-    if MICROTASK_RUN_DEPTH.with(|depth| depth.get()) == 1
+    if MICROTASK_RUN_DEPTH.with(|depth| depth.get().pump) == 1
         && TASK_QUEUE.with(|q| q.borrow().is_empty())
     {
         crate::r#box::flush_released_boxes();
     }
 
     MICROTASK_RUN_DEPTH.with(|depth| {
-        depth.set(depth.get().saturating_sub(1));
+        let mut current = depth.get();
+        current.pump = current.pump.saturating_sub(1);
+        depth.set(current);
     });
 
     ran

crates/perry-runtime/src/promise/mod.rs

@@ -232,9 +232,10 @@ extern "C" fn mt_profile_atexit() {
         crate::timer::PROFILE_INTERVAL_TIMERS_FIRED.load(Ordering::Relaxed),
     );
     eprintln!(
-        "[mt-profile] event_notify={{sent:{},during_drain:{}}} event_wait={{total:{},fast:{},zero:{},driver:{},condvar:{}}}",
+        "[mt-profile] event_notify={{sent:{},during_drain:{},drain_suppressed:{}}} event_wait={{total:{},fast:{},zero:{},driver:{},condvar:{}}}",
         crate::event_pump::PROFILE_NOTIFY_COUNT.load(Ordering::Relaxed),
         crate::event_pump::PROFILE_NOTIFY_DURING_DRAIN_COUNT.load(Ordering::Relaxed),
+        crate::event_pump::PROFILE_NOTIFY_DRAIN_SUPPRESSED_COUNT.load(Ordering::Relaxed),
         crate::event_pump::PROFILE_WAIT_COUNT.load(Ordering::Relaxed),
         crate::event_pump::PROFILE_WAIT_FAST_COUNT.load(Ordering::Relaxed),
         crate::event_pump::PROFILE_WAIT_ZERO_COUNT.load(Ordering::Relaxed),
@@ -736,7 +737,7 @@ pub(crate) fn enqueue_queue_microtask(callback: i64) {
             trigger_async_id: ids.trigger_async_id,
         });
     });
-    crate::event_pump::js_notify_main_thread();
+    crate::event_pump::js_notify_promise_progress();
 }
 
 #[derive(Default)]