Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
115 changes: 87 additions & 28 deletions src/bundler/BundleThread.rs
Original file line number Diff line number Diff line change
Expand Up @@ -45,6 +45,11 @@ pub enum BundleV2Result {
// provides the `CompletionStruct` impl for the forward-decl.
pub(crate) struct BundleThread<C: Node> {
pub(crate) waker: Async::Waker,
/// Set by the bundle thread instead of `waker` when `Waker::init()` fails
/// (`eventfd`/`kqueue` at the open-files limit). Published by
/// `ready_event`, read once by `spawn()`, which then reports it and the
/// allocation is freed.
pub(crate) waker_error: Option<Async::Error>,
pub(crate) ready_event: ResetEvent,
// `bun.UnboundedQueue(CompletionStruct, .next)` — intrusive over `C.next`;
// the field offset is encoded via the `Node` supertrait on `CompletionStruct`.
Expand Down Expand Up @@ -125,10 +130,13 @@ impl<C: CompletionStruct> BundleThread<C> {
// Windows), so zeroing them is *language-level* UB even if never read.
// `placeholder()` yields a fully-initialized inert value instead.
// `ready_event.wait()` in `spawn()` blocks until `thread_main` overwrites
// it via `ptr::write`, so the placeholder is never observed live.
// it via `ptr::write`, so the placeholder is never observed live; if
// `thread_main` fails instead, the whole struct is freed and dropping the
// placeholder is harmless (it owns nothing).
pub(crate) fn uninitialized() -> Self {
Self {
waker: Async::Waker::placeholder(),
waker_error: None,
queue: UnboundedQueue::new(),
generation: 0,
ready_event: ResetEvent::default(),
Expand All @@ -138,11 +146,11 @@ impl<C: CompletionStruct> BundleThread<C> {
/// # Safety
/// `instance` must be valid for `'static`: after `Ok` the bundle thread accesses it
/// forever, so callers must only touch it through the raw-pointer methods on this
/// impl (e.g. `enqueue`) and never materialize a `&mut Self`. After `Err` no thread
/// exists and the caller still owns it.
pub(crate) unsafe fn spawn(
instance: *mut Self,
) -> std::io::Result<std::thread::JoinHandle<()>> {
/// impl (e.g. `enqueue`) and never materialize a `&mut Self`. After `Err` nothing
/// refers to `*instance` any more and the caller still owns it: either no thread
/// was started, or the one that was has been joined without ever looking at the
/// queue.
pub(crate) unsafe fn spawn(instance: *mut Self) -> std::io::Result<()> {
// `std::thread::Builder` (not `std::thread::spawn`) so the spawn error
// is surfaced to the caller.
struct SendPtr<T>(*mut T);
Expand All @@ -156,15 +164,32 @@ impl<C: CompletionStruct> BundleThread<C> {
let ptr = ptr;
// SAFETY: caller guarantees `instance` is valid for 'static; `thread_main`
// accesses fields only via raw-ptr projection (never `&Self`/`&mut Self`)
// and is the sole writer of `waker`/`generation`, so concurrent `enqueue()`
// from other threads is sound.
// and is the sole writer of `waker`/`waker_error`/`generation`, so
// concurrent `enqueue()` from other threads is sound.
unsafe { Self::thread_main(ptr.0) }
})?;
// SAFETY: field projection via raw ptr — the spawned thread is concurrently
// writing `waker`, so we must not hold `&Self`/`&mut Self` here. `ready_event`
// itself is a sync primitive safe to wait on from this thread.
unsafe { (*instance).ready_event.wait() };
Ok(thread)
// SAFETY: field projections via raw ptr — the spawned thread is concurrently
// writing `waker` / `waker_error`, so we must not hold `&Self`/`&mut Self` here.
// `ready_event` itself is a sync primitive safe to wait on from this thread, and
// the bundle thread writes `waker_error` before `set()`, so the read after
// `wait()` observes it.
let waker_error = unsafe {
(*instance).ready_event.wait();
(*instance).waker_error
};
match waker_error {
None => {
// `std.Thread.detach()` — drop the JoinHandle without joining.
drop(thread);
Ok(())
}
Some(err) => {
// `thread_main` returns right after `set()`; wait for the thread to be
// gone so the caller can free `*instance`.
let _ = thread.join();
Err(waker_error_to_io(err))
}
}
}

/// # Safety
Expand All @@ -189,12 +214,27 @@ impl<C: CompletionStruct> BundleThread<C> {
unsafe fn thread_main(instance: *mut Self) {
Output::Source::configure_named_thread(zstr!("Bundler"));

// The waker has to be created on this thread (on Windows it is this
// thread's libuv loop), which is why `spawn()` cannot create it up front
// and instead learns of a failure through `waker_error`.
let waker = match Async::Waker::init() {
Ok(waker) => waker,
Err(err) => {
// SAFETY: `spawn()` is blocked in `ready_event.wait()` and reads
// `waker_error` only once that returns; `set()` is this thread's last
// access to `*instance`, and `spawn()` joins this thread before the
// allocation is freed.
unsafe {
core::ptr::addr_of_mut!((*instance).waker_error).write(Some(err));
(*instance).ready_event.set();
}
return;
}
};

// SAFETY: `waker` is written exactly once here, before `ready_event.set()`
// releases any thread that could call `enqueue` (which reads `waker`).
unsafe {
core::ptr::addr_of_mut!((*instance).waker)
.write(Async::Waker::init().unwrap_or_else(|_| panic!("Failed to create waker")));
}
unsafe { core::ptr::addr_of_mut!((*instance).waker).write(waker) };

// Unblock the calling thread so it can continue.
// SAFETY: raw-ptr field projection; spawning thread is blocked in `ready_event.wait()`.
Expand Down Expand Up @@ -343,6 +383,23 @@ impl<C: CompletionStruct> BundleThread<C> {
}
}

/// Carries a failed waker creation out of `spawn()` in the form `singleton::get`
/// reports: `SystemErrno::from_io_error` recovers the errno for the named
/// message, the rest fall through to the OS-description path.
fn waker_error_to_io(err: Async::Error) -> std::io::Error {
match err {
// The POSIX wakers store the OS's errno and `SystemErrno`'s
// discriminants are those values, so the code round-trips. (Windows'
// waker cannot fail; its `raw_os_error` space is Win32 codes, so a
// `Sys` from it must not take this path.)
Async::Error::Sys(errno) => std::io::Error::from_raw_os_error(errno as i32),
// No OS code to carry; the log then shows the variant's Display.
Async::Error::MachportCreationFailed | Async::Error::Unexpected => {
std::io::Error::other(err)
}
}
}

/// Lazily-initialized singleton. This is used for `Bun.build` since the
/// bundle thread may not be needed.
// Rust forbids generic statics, so the storage is
Expand All @@ -364,8 +421,9 @@ pub mod singleton {

static INSTANCE: bun_threading::Guarded<Option<Instance>> = bun_threading::Guarded::new(None);

/// Starts the bundle thread on first use; a failed spawn leaves the slot empty
/// for the next build to retry.
/// Starts the bundle thread on first use; a failed start (`pthread_create`,
/// or the thread's waker at the open-files limit) leaves the slot empty for
/// the next build to retry.
///
/// # Safety
/// All calls (across the process) must use the same `C`; the static is
Expand All @@ -380,14 +438,11 @@ pub mod singleton {
bun_core::heap::into_raw_nn(Box::new(BundleThread::<C>::uninitialized()));
// SAFETY: a leaked Box, valid for 'static; passed as a raw pointer so no
// `&mut` aliases the bundle thread's own access.
match unsafe { BundleThread::spawn(bundle_thread.as_ptr()) } {
// `std.Thread.detach()` — drop the JoinHandle without joining.
Ok(os_thread) => drop(os_thread),
Err(err) => {
// SAFETY: `spawn` started nothing, so this is still the sole owner.
unsafe { bun_core::heap::destroy(bundle_thread.as_ptr()) };
return Err(err);
}
if let Err(err) = unsafe { BundleThread::spawn(bundle_thread.as_ptr()) } {
// SAFETY: per `spawn`'s contract no thread refers to the allocation
// after `Err`, so this is the sole owner again.
unsafe { bun_core::heap::destroy(bundle_thread.as_ptr()) };
return Err(err);
}
*instance = Some(Instance(bundle_thread.cast::<()>()));
Ok(bundle_thread.as_ptr())
Expand Down Expand Up @@ -427,9 +482,13 @@ pub mod singleton {
let mut log = bun_ast::Log::init();
match errno {
Some(errno) => {
// Unix reports an exhausted thread or pid limit as EAGAIN.
// Unix reports an exhausted thread or pid limit as EAGAIN;
// EMFILE/ENFILE is the waker's eventfd/kqueue at the open-files
// limit.
let hint = if cfg!(not(windows)) && errno == SystemErrno::EAGAIN {
" The process or thread limit may have been reached (ulimit -u, or the container's pids limit)."
} else if matches!(errno, SystemErrno::EMFILE | SystemErrno::ENFILE) {
" The open file limit may have been reached (ulimit -n, or the system-wide limit)."
} else {
""
};
Expand Down
42 changes: 22 additions & 20 deletions src/io/lib.rs
Original file line number Diff line number Diff line change
Expand Up @@ -2107,9 +2107,11 @@ pub mod waker {

#[cfg(windows)]
pub struct WindowsWaker {
/// Process-global `WindowsLoop` singleton. `BackRef` invariant (pointee
/// outlives holder) holds trivially: the loop is never freed. `None`
/// only between [`placeholder`] and [`init`]; every dispatch path
/// The `WindowsLoop` of the thread that called [`init`] (uws keeps one
/// per thread, freed when that thread exits). `BackRef` invariant
/// (pointee outlives holder) holds because the only user, the bundle
/// thread, never exits once it holds a waker. `None` only between
/// [`placeholder`] and [`init`]; every dispatch path
/// (`wake`/`wait`/`uv_loop`) unwraps and would have UB-derefed the old
/// raw null anyway.
///
Expand All @@ -2130,7 +2132,9 @@ pub mod waker {
Self { loop_: None }
}

pub fn init() -> crate::Result<Self> {
/// Binds the calling thread's loop, so call it on the thread that will
/// `wait()`. Same signature as the POSIX wakers; cannot fail today.
pub fn init() -> crate::error::Result<Self> {
Ok(Self {
loop_: Some(bun_ptr::BackRef::from(
core::ptr::NonNull::new(bun_uws_sys::WindowsLoop::get())
Expand All @@ -2149,15 +2153,14 @@ pub mod waker {

pub fn wait(&self) {
// Do NOT route through `WindowsLoop::wait(&mut self)`: that would
// materialize a `&mut WindowsLoop` over the process-global
// singleton for the entire duration of `us_loop_run`/`uv_run`,
// and a concurrent `wake()` from a worker thread (BundleThread,
// HTTPThread) would alias it — two live `&mut T` to one
// materialize a `&mut WindowsLoop` over the loop for the entire
// duration of `us_loop_run`/`uv_run`, and a concurrent `wake()`
// from another thread would alias it — two live `&mut T` to one
// allocation is UB under Stacked/Tree Borrows. Call the C entry
// point with the raw pointer directly so no Rust reference is
// ever formed.
// SAFETY: `loop_` is the live `WindowsLoop::get()` singleton,
// non-null after `init()`.
// SAFETY: `loop_` is live for as long as this waker is (see the
// field doc) and non-null after `init()`.
unsafe { bun_uws_sys::loop_::us_loop_run(self.loop_ref().as_const_ptr().cast_mut()) };
}

Expand All @@ -2166,23 +2169,22 @@ pub mod waker {
// `&mut WindowsLoop` here would alias the event-loop thread's
// borrow held across `us_loop_run`. Pass the raw pointer to the
// thread-safe C wake (`uv_async_send`) instead.
// SAFETY: `loop_` is the live `WindowsLoop::get()` singleton;
// `us_wakeup_loop` → `uv_async_send` is documented thread-safe.
// SAFETY: `loop_` is live for as long as this waker is (see the
// field doc); `us_wakeup_loop` → `uv_async_send` is documented
// thread-safe.
unsafe {
bun_uws_sys::loop_::us_wakeup_loop(self.loop_ref().as_const_ptr().cast_mut())
};
}

/// Raw libuv `uv_loop_t*` underlying this waker's `WindowsLoop`.
///
/// `loop_` is the process-global singleton from `WindowsLoop::get()`
/// (set in [`init`]), so the returned pointer has process lifetime —
/// safe to hand to `uv::Timer::init` and friends without an `unsafe`
/// block at the call site.
/// Raw libuv `uv_loop_t*` underlying this waker's `WindowsLoop`. It
/// lives as long as `loop_` does (see the field doc), so it is safe to
/// hand to `uv::Timer::init` and friends without an `unsafe` block at
/// the call site.
#[inline]
pub fn uv_loop(&self) -> *mut bun_sys::windows::libuv::Loop {
// `BackRef` deref is safe (process-lifetime singleton); `uv_loop`
// is a `Copy` field set once by C `us_create_loop`.
// `BackRef` deref is safe for the same reason; `uv_loop` is a
// `Copy` field set once by C `us_create_loop`.
self.loop_ref().uv_loop
}
}
Expand Down
3 changes: 2 additions & 1 deletion src/uws_sys/Loop.rs
Original file line number Diff line number Diff line change
Expand Up @@ -397,7 +397,8 @@ impl WindowsLoop {
}

pub fn get() -> *mut WindowsLoop {
// SAFETY: uv::Loop::get() returns the libuv default loop; uws wraps it
// SAFETY: uv::Loop::get() returns this thread's live libuv loop; uws
// wraps it in a per-thread loop of its own.
unsafe { c::uws_get_loop_with_native(uv::Loop::get() as *mut c_void) }
}

Expand Down
Loading
Loading