Replace the pending-signal ring with one counter per signal number

src/jsc/PosixSignalHandle.rs

@@ -5,15 +5,16 @@ use crate::VirtualMachineRef as VirtualMachine;
 use crate::event_loop::EventLoop;
 use crate::{JSGlobalObject, Task};
 use bun_event_loop::{Taskable, task_tag};
-use bun_threading::SignalRing;
+use bun_threading::PendingSignals;
 
-const BUFFER_SIZE: usize = 8192;
+/// Cap on JS listener calls per signal number per loop tick.
+const MAX_TASKS_PER_SIGNAL: u32 = 8192;
 
 /// Signal numbers queued by signal handlers on any thread for the main loop.
 #[derive(Default)]
 pub struct PosixSignalHandle {
     #[allow(dead_code)]
-    ring: SignalRing<BUFFER_SIZE>,
+    pending: PendingSignals,
 }
 
 impl PosixSignalHandle {
@@ -23,29 +24,30 @@ impl PosixSignalHandle {
         Box::new(init)
     }
 
-    /// Returns `false` if the ring is full. The caller wakes the loop on `true`.
+    /// Async-signal-safe and never full.
     #[allow(dead_code)]
-    pub(crate) fn enqueue(&self, signal: u8) -> bool {
-        self.ring.enqueue(signal)
+    pub(crate) fn enqueue(&self, signal: u8) {
+        self.pending.add(signal);
     }
 
-    /// Drain as many signals as possible and enqueue them as tasks in the event loop.
-    /// Called by the main thread, the ring's single consumer.
+    /// Main thread: enqueue one task per pending arrival, lowest number first.
     #[allow(dead_code)]
     pub(crate) fn drain(&self, event_loop: &mut EventLoop) {
-        while let Some(signal) = self.ring.dequeue() {
-            // `Task` is a plain `{ tag, ptr }` pair (no bitfield packing), so build it
-            // directly — `bun_runtime::dispatch::run_task` unpacks `task.ptr as usize as u8`.
-            let task = Task::new(
-                <PosixSignalTask as Taskable>::TAG,
-                signal as usize as *mut (),
-            );
-            event_loop.enqueue_task(task);
-        }
+        self.pending.take(|signal, count| {
+            for _ in 0..count.min(MAX_TASKS_PER_SIGNAL) {
+                // `Task` is a plain `{ tag, ptr }` pair (no bitfield packing), so build it
+                // directly — `bun_runtime::dispatch::run_task` unpacks `task.ptr as usize as u8`.
+                let task = Task::new(
+                    <PosixSignalTask as Taskable>::TAG,
+                    signal as usize as *mut (),
+                );
+                event_loop.enqueue_task(task);
+            }
+        });
     }
 }
 
-/// This is the signal handler entry point. Calls enqueue on the ring buffer.
+/// This is the signal handler entry point. Counts the signal as pending.
 /// Note: Must be minimal logic here. Only do atomics & signal-safe calls.
 #[unsafe(no_mangle)]
 extern "C" fn Bun__onPosixSignal(number: i32) {
@@ -76,11 +78,10 @@ extern "C" fn Bun__onPosixSignal(number: i32) {
             let Some(signal) = u8::try_from(number).ok().filter(|&s| s != 0) else {
                 return;
             };
-            if handler.enqueue(signal) {
-                // SAFETY: same process-lifetime event loop as above; `wakeup`
-                // is one async-signal-safe write to the loop's wakeup fd.
-                unsafe { (*(*vm).event_loop()).wakeup() };
-            }
+            handler.enqueue(signal);
+            // SAFETY: same process-lifetime event loop as above; `wakeup`
+            // is one async-signal-safe write to the loop's wakeup fd.
+            unsafe { (*(*vm).event_loop()).wakeup() };
         }
     }
     #[cfg(not(unix))]

src/jsc/event_loop.rs

@@ -103,7 +103,7 @@ pub struct EventLoop {
     pub imminent_gc_timer: AtomicPtr<()>,
 
     #[cfg(unix)]
-    /// Boxed `PosixSignalHandle` ring buffer, leaked once by
+    /// Boxed `PosixSignalHandle` (pending-signal counters), leaked once by
     /// `Bun__ensureSignalHandler` and live for the process lifetime. Stored as
     /// a [`bun_ptr::BackRef`] so the per-tick `drain()` / signal-context
     /// `enqueue()` reads go through the single audited `BackRef::deref`
@@ -584,7 +584,7 @@ impl EventLoop {
         {
             if let Some(signal_handler) = self.signal_handler {
                 // `signal_handler` is a `BackRef` to the leaked process-lifetime
-                // `PosixSignalHandle` (see field doc); the ring-buffer backing is
+                // `PosixSignalHandle` (see field doc); the counter array is
                 // disjoint from `*self`, so the `&PosixSignalHandle` materialised
                 // by `BackRef::deref` does not alias the `&mut self` passed here.
                 signal_handler.drain(self);

src/threading/lib.rs

@@ -18,7 +18,7 @@ pub mod thread_pool;
 pub mod work_pool;
 
 pub mod guarded;
-pub mod signal_ring;
+pub mod pending_signals;
 pub mod unbounded_queue;
 #[path = "WaitGroup.rs"]
 pub mod wait_group;
@@ -32,10 +32,10 @@ pub use condition::{Condition, Condvar};
 pub use futex as Futex;
 pub use guarded::{Guarded, GuardedLock};
 pub use mutex::{Mutex, MutexGuard};
+pub use pending_signals::PendingSignals;
 pub use reset_event::ResetEvent;
 pub use rwlock::RwLock;
 pub use semaphore::Semaphore;
-pub use signal_ring::SignalRing;
 pub use thread_pool::ThreadPool;
 pub use unbounded_queue::{Link, Linked, UnboundedQueue};
 pub use wait_group::WaitGroup;

src/threading/pending_signals.rs

@@ -0,0 +1,135 @@
+//! Counts of pending POSIX signals, one per signal number.
+
+use core::sync::atomic::{AtomicU32, AtomicU64, Ordering};
+
+/// Multi-producer, multi-consumer pending-signal counts. Never full.
+pub struct PendingSignals {
+    /// How many times each signal number arrived since it was last taken.
+    counts: [AtomicU32; 256],
+    /// Bit `n` set: `counts[n]` may be nonzero (a hint, never authoritative).
+    mask: [AtomicU64; 4],
+}
+
+#[inline]
+const fn mask_bit(signal: u8) -> (usize, u64) {
+    (signal as usize / 64, 1u64 << (signal % 64))
+}
+
+impl Default for PendingSignals {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
+impl PendingSignals {
+    pub const fn new() -> Self {
+        Self {
+            counts: [const { AtomicU32::new(0) }; 256],
+            mask: [const { AtomicU64::new(0) }; 4],
+        }
+    }
+
+    /// Producer side, any thread, async-signal-safe. `signal` is never 0.
+    pub fn add(&self, signal: u8) {
+        debug_assert_ne!(signal, 0, "signal numbers start at 1");
+        self.counts[usize::from(signal)].fetch_add(1, Ordering::Relaxed);
+        let (word, bit) = mask_bit(signal);
+        self.mask[word].fetch_or(bit, Ordering::Release);
+    }
+
+    /// Consumer side, any number of them. Calls `f(signal, count)` for each
+    /// pending number, lowest first, and resets its count.
+    pub fn take(&self, mut f: impl FnMut(u8, u32)) {
+        for (index, word) in self.mask.iter().enumerate() {
+            if word.load(Ordering::Acquire) == 0 {
+                continue;
+            }
+            let mut bits = word.swap(0, Ordering::AcqRel);
+            while bits != 0 {
+                let bit = bits.trailing_zeros();
+                bits &= bits - 1;
+                let signal = (index * 64 + bit as usize) as u8;
+                let count = self.counts[usize::from(signal)].swap(0, Ordering::AcqRel);
+                // 0 when another consumer took it or the bit outran the count.
+                if count != 0 {
+                    f(signal, count);
+                }
+            }
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use std::sync::atomic::AtomicUsize;
+
+    // Miri interprets every atomic op, so the stress stays short there.
+    #[cfg(miri)]
+    const PRODUCERS: u8 = 4;
+    #[cfg(not(miri))]
+    const PRODUCERS: u8 = 8;
+    #[cfg(miri)]
+    const PER_PRODUCER: u32 = 200;
+    #[cfg(not(miri))]
+    const PER_PRODUCER: u32 = 50_000;
+
+    #[test]
+    fn counts_are_exact_and_lowest_number_first() {
+        let pending = PendingSignals::new();
+        pending.add(15);
+        pending.add(2);
+        pending.add(15);
+        pending.add(255);
+        pending.add(64);
+        pending.add(15);
+
+        let mut seen = Vec::new();
+        pending.take(|signal, count| seen.push((signal, count)));
+        assert_eq!(seen, [(2, 1), (15, 3), (64, 1), (255, 1)]);
+
+        seen.clear();
+        pending.take(|signal, count| seen.push((signal, count)));
+        assert_eq!(seen, []);
+
+        pending.add(2);
+        pending.take(|signal, count| seen.push((signal, count)));
+        assert_eq!(seen, [(2, 1)]);
+    }
+
+    /// Producers on several threads never lose an arrival.
+    #[test]
+    fn concurrent_producers_deliver_every_arrival_once() {
+        let pending = PendingSignals::new();
+        let done = AtomicUsize::new(0);
+        let mut taken = [0u32; 256];
+
+        std::thread::scope(|scope| {
+            for signal in 1..=PRODUCERS {
+                let (pending, done) = (&pending, &done);
+                scope.spawn(move || {
+                    for _ in 0..PER_PRODUCER {
+                        pending.add(signal);
+                    }
+                    done.fetch_add(1, Ordering::Release);
+                });
+            }
+            loop {
+                let all_done = done.load(Ordering::Acquire) == usize::from(PRODUCERS);
+                pending.take(|signal, count| taken[usize::from(signal)] += count);
+                if all_done {
+                    break;
+                }
+                std::hint::spin_loop();
+            }
+        });
+
+        for signal in 1..=PRODUCERS {
+            assert_eq!(taken[usize::from(signal)], PER_PRODUCER, "signal {signal}");
+        }
+        assert_eq!(
+            taken.iter().map(|&n| u64::from(n)).sum::<u64>(),
+            u64::from(PRODUCERS) * u64::from(PER_PRODUCER)
+        );
+    }
+}