diff --git a/src/runtime/bake/dev_server/hot_reload_event.rs b/src/runtime/bake/dev_server/hot_reload_event.rs index d134cd6522f0..d210b2713a03 100644 --- a/src/runtime/bake/dev_server/hot_reload_event.rs +++ b/src/runtime/bake/dev_server/hot_reload_event.rs @@ -3,14 +3,14 @@ use super::*; /// `DevServer.HotReloadEvent` — produced by the watcher thread. -// Note: cache-line alignment makes each inline `WatcherAtomics.events: [3]` +// Note: cache-line alignment makes each inline `WatcherAtomics.events` // element occupy its own cache line, avoiding false sharing on // `contention_indicator` between watcher and dev-server threads. 128 matches // the cache line on x86_64/aarch64 (Bun's tier-1 targets) and absorbs Intel // adjacent-line prefetch. #[repr(align(128))] pub(crate) struct HotReloadEvent { - /// BACKREF (LIFETIMES.tsv): element of `WatcherAtomics.events: [3]`. + /// BACKREF (LIFETIMES.tsv): element of `WatcherAtomics.events`. /// Nulled by `Drop for DevServer` when an event is still queued; `run` /// checks for null before dereferencing. pub(crate) owner: *mut DevServer, @@ -292,6 +292,7 @@ impl HotReloadEvent { #[cfg(debug_assertions)] { + use core::sync::atomic::Ordering; // SAFETY: `first` is live and exclusively owned by this thread. debug_assert!(unsafe { (*first).debug_mutex.try_lock() }); // SAFETY: `first` is live (caller contract); atomic load needs no diff --git a/src/runtime/bake/dev_server/lifecycle.rs b/src/runtime/bake/dev_server/lifecycle.rs index ed76337d64e2..167820e72da2 100644 --- a/src/runtime/bake/dev_server/lifecycle.rs +++ b/src/runtime/bake/dev_server/lifecycle.rs @@ -664,15 +664,11 @@ impl Drop for DevServer { } // `Watcher::shutdown` serialises with `dispatch_file_updates` on - // `Watcher.mutex`, so `next_event` is stable from the watcher side. + // `Watcher.mutex`, so `state` is stable from the watcher side. let atomics = self.watcher_atomics.as_ptr(); // SAFETY: `atomics` is the `heap::alloc_nn` pointer from `init()`. - let next = unsafe { - (*atomics) - .next_event - .load(::core::sync::atomic::Ordering::Acquire) - }; - if next == crate::bake::dev_server::NextEvent::DONE.0 { + let posted = unsafe { (*atomics).state.lock().dev.is_some() }; + if !posted { // SAFETY: `atomics` is the unique owner; no event is queued. unsafe { for event in &mut (*atomics).events { diff --git a/src/runtime/bake/dev_server/mod.rs b/src/runtime/bake/dev_server/mod.rs index bd72c175faae..1f68b71b21df 100644 --- a/src/runtime/bake/dev_server/mod.rs +++ b/src/runtime/bake/dev_server/mod.rs @@ -54,8 +54,6 @@ bun_output::declare_scope!(SourceMapStore, visible); bun_output::define_scoped_log!(debug_log, crate::bake::dev_server::DevServer); bun_output::define_scoped_log!(map_log, crate::bake::dev_server::SourceMapStore); -use core::sync::atomic::Ordering; - use bun_collections::StringArrayHashMap; use bun_sys::FdExt as _; @@ -102,7 +100,7 @@ pub(crate) use http::{is_allowed_dev_host, is_allowed_host_header}; pub(crate) use lifecycle::init; pub(crate) use protocol::{ConsoleLogKind, HmrTopic, HmrTopicBits, IncomingMessageId, MessageId}; use router_glue::{OpaqueFileIdOrOptional, from_opaque_file_id, to_opaque_file_id}; -pub(crate) use watcher_atomics::{NextEvent, WatcherAtomics}; +pub(crate) use watcher_atomics::WatcherAtomics; /// `DevServer.FileKind` — kept in lockstep with `bun_bundler::bake_types::CacheKind` /// (the vtable boundary maps between them via an exhaustive match). @@ -560,7 +558,7 @@ pub(crate) struct DevServer { pub(crate) directory_watchers: DirectoryWatchStore, /// Heap-owned via `heap::alloc_nn` so a still-queued /// `HotReloadEvent.concurrent_task` can outlive `DevServer`; see the - /// `next_event` check in `Drop`. + /// `state` check in `Drop`. pub(crate) watcher_atomics: ::core::ptr::NonNull, pub(crate) testing_batch_events: TestingBatchEvents, diff --git a/src/runtime/bake/dev_server/watcher_atomics.rs b/src/runtime/bake/dev_server/watcher_atomics.rs index 9764e66f7b27..874f37f591cf 100644 --- a/src/runtime/bake/dev_server/watcher_atomics.rs +++ b/src/runtime/bake/dev_server/watcher_atomics.rs @@ -2,41 +2,22 @@ use super::*; -/// `DevServer.WatcherAtomics` — three pre-allocated `HotReloadEvent`s -/// rotated between the watcher thread and the main thread. +/// Two pre-allocated `HotReloadEvent`s. The dev server thread holds at most one of them at a time. pub(crate) struct WatcherAtomics { - pub(crate) events: [HotReloadEvent; 3], - /// Atomically encodes a `NextEvent`: values 0..3 are an index into - /// `events`, plus the `WAITING`/`DONE` sentinels. - // Rust cannot align individual fields, so this field is not cache-line - // aligned. Wrap in a `#[repr(align(128))]` newtype (init site: - // lifecycle.rs) if false sharing ever shows up in profiles. - pub(crate) next_event: core::sync::atomic::AtomicU8, - /// Watcher-thread-only; index into `events` currently being processed. - pub(crate) current_event: Option, - /// Watcher-thread-only; index into `events` queued behind `current_event`. - pub(crate) pending_event: Option, - // Debug fields to ensure methods are being called in the right order. + pub(crate) events: [HotReloadEvent; 2], + /// Not held across a call that locks `Watcher.mutex`, which the watcher thread takes first. + pub(crate) state: bun_threading::Guarded, + // Debug field to ensure methods are being called in the right order. #[cfg(debug_assertions)] pub(crate) dbg_watcher_event: Option<*mut HotReloadEvent>, - #[cfg(debug_assertions)] - pub(crate) dbg_server_event: Option<*mut HotReloadEvent>, } -/// Stored in `WatcherAtomics::next_event` (an `AtomicU8`). Modeled as a -/// transparent newtype rather than a `#[repr(u8)] enum` because any value -/// other than the named constants is an index into the `events` array, and -/// Rust enums cannot hold unlisted discriminants. -#[repr(transparent)] -#[derive(Copy, Clone, Eq, PartialEq)] -pub(crate) struct NextEvent(pub(crate) u8); - -impl NextEvent { - /// An event is running, and no next event is pending. - pub(crate) const WAITING: NextEvent = NextEvent(u8::MAX - 1); - /// No event is running. - pub(crate) const DONE: NextEvent = NextEvent(u8::MAX); - // Any other value represents an index into the `events` array. +#[derive(Default)] +pub(crate) struct State { + /// Index into `events` of the event the dev server thread was given and has not recycled. + pub(crate) dev: Option, + /// The other event has files and waits for `recycle_event_from_dev_server`. + queued: bool, } // These take `this: *mut Self` (not `&mut self`) so the returned @@ -62,57 +43,23 @@ impl WatcherAtomics { // SAFETY: caller contract — `this` and `old_event` are live; every // `(*this)` / `(*old_event)` below is a field access on those. unsafe { + let mut state = (*this).state.lock(); (*old_event).reset(); - #[cfg(debug_assertions)] - { - // Not atomic because watcher won't modify this value while an event is running. - let dbg_event = (*this).dbg_server_event; - (*this).dbg_server_event = None; - debug_assert!( - dbg_event == Some(old_event), - "recycleEventFromDevServer: old_event: expected {:?}, got {:p}", - dbg_event, - old_event, - ); + let old_index = state.dev.expect("the dev server thread holds an event"); + debug_assert!( + core::ptr::eq(old_event, &raw const (*this).events[old_index as usize]), + "recycleEventFromDevServer: old_event: expected index {}, got {:p}", + old_index, + old_event, + ); + if !core::mem::take(&mut state.queued) { + state.dev = None; + return None; // done running events } - - let event: *mut HotReloadEvent = loop { - let next = NextEvent( - (*this) - .next_event - .swap(NextEvent::WAITING.0, Ordering::AcqRel), - ); - match next { - NextEvent::WAITING => { - // Success order is not AcqRel because the swap above performed an Acquire load. - // Failure order is Relaxed because we're going to perform an Acquire load - // in the next loop iteration. - if (*this) - .next_event - .compare_exchange_weak( - NextEvent::WAITING.0, - NextEvent::DONE.0, - Ordering::Release, - Ordering::Relaxed, - ) - .is_err() - { - continue; // another event may have been added - } - return None; // done running events - } - NextEvent::DONE => unreachable!(), - _ => break &raw mut (*this).events[next.0 as usize], - } - }; - - #[cfg(debug_assertions)] - { - // Not atomic because watcher won't modify this value while an event is running. - (*this).dbg_server_event = Some(event); - } - Some(event) + let next = old_index ^ 1; + state.dev = Some(next); + Some(&raw mut (*this).events[next as usize]) } } @@ -129,21 +76,11 @@ impl WatcherAtomics { // SAFETY: caller contract — `this` is live; every `(*this)` / `(*ev)` // below is a field access on the heap `WatcherAtomics` allocation. unsafe { - let mut available = [true; 3]; - if let Some(i) = (*this).current_event { - available[i as usize] = false; - } - if let Some(i) = (*this).pending_event { - available[i as usize] = false; - } - - let index = 'find: { - for (i, &is_available) in available.iter().enumerate() { - if is_available { - break 'find i; - } - } - unreachable!() + let (index, taken_back): (usize, bool) = { + let mut state = (*this).state.lock(); + // Take the queued event back so that this batch joins it. + let taken_back = core::mem::take(&mut state.queued); + (if state.dev == Some(0) { 1 } else { 0 }, taken_back) }; let ev: *mut HotReloadEvent = &raw mut (*this).events[index]; @@ -156,9 +93,9 @@ impl WatcherAtomics { (*this).dbg_watcher_event = Some(ev); } - // `ev` points into `(*this).events[index]`, which the watcher thread - // has exclusive access to (neither `current_event` nor `pending_event`). + // Only this thread can reach `ev`: it is neither `state.dev` nor queued. let ev_ref = &mut *ev; + debug_assert_eq!(ev_ref.is_empty(), !taken_back); // Initialize the timer if it is empty. if ev_ref.is_empty() { @@ -226,26 +163,14 @@ impl WatcherAtomics { let ev_index: u8 = u8::try_from(ev.offset_from((&raw const (*this).events).cast::())) .unwrap(); - let old_next = NextEvent((*this).next_event.swap(ev_index, Ordering::AcqRel)); - match old_next { - NextEvent::DONE => { + let mut state = (*this).state.lock(); + match state.dev { + // The dev server thread takes `ev` when it recycles the event it holds. + Some(_) => state.queued = true, + None => { // Dev server is done running events. We need to schedule the event directly. - (*this).current_event = Some(ev_index); - (*this).pending_event = None; - // Relaxed because the dev server is not running events right now. - // (could technically be made non-atomic) - (*this) - .next_event - .store(NextEvent::WAITING.0, Ordering::Relaxed); - #[cfg(debug_assertions)] - { - debug_assert!( - (*this).dbg_server_event.is_none(), - "no event should be running right now", - ); - // Not atomic because the dev server is not running events right now. - (*this).dbg_server_event = Some(ev); - } + state.dev = Some(ev_index); + drop(state); (*ev).concurrent_task = bun_event_loop::ConcurrentTask::ConcurrentTask { task: bun_event_loop::Task::init(ev), ..Default::default() @@ -259,27 +184,6 @@ impl WatcherAtomics { core::ptr::NonNull::new_unchecked(&raw mut (*ev).concurrent_task), ); } - - NextEvent::WAITING => { - if (*this).pending_event.is_some() { - // `pending_event` is running, which means we're done with `current_event`. - (*this).current_event = (*this).pending_event; - } // else, no pending event yet, but not done with `current_event`. - (*this).pending_event = Some(ev_index); - } - - _ => { - // This is an index into the `events` array. - let old_index: u8 = old_next.0; - debug_assert!( - (*this).pending_event == Some(old_index), - "watcher_release_and_submit_event: expected `pending_event` to be {}; got {:?}", - old_index, - (*this).pending_event, - ); - // The old pending event hadn't been run yet, so we can replace it with `ev`. - (*this).pending_event = Some(ev_index); - } } } } @@ -289,14 +193,10 @@ impl WatcherAtomics { pub(crate) fn init(owner: *mut DevServer) -> core::ptr::NonNull { let mk_event = || HotReloadEvent::init_empty(owner); let atomics = bun_core::heap::alloc_nn(WatcherAtomics { - events: [mk_event(), mk_event(), mk_event()], - next_event: core::sync::atomic::AtomicU8::new(NextEvent::DONE.0), - current_event: None, - pending_event: None, + events: [mk_event(), mk_event()], + state: Default::default(), #[cfg(debug_assertions)] dbg_watcher_event: None, - #[cfg(debug_assertions)] - dbg_server_event: None, }); // SAFETY: `atomics` is the fresh `heap::alloc_nn` result; the // allocation is exclusively reachable through it here. diff --git a/test/bake/bake-harness.ts b/test/bake/bake-harness.ts index c5ba2f53efaf..115cfb086da9 100644 --- a/test/bake/bake-harness.ts +++ b/test/bake/bake-harness.ts @@ -1569,6 +1569,9 @@ if (!fs.existsSync(tempDir)) { fs.mkdirSync(tempDir, { recursive: true }); } +/** Holds the root directory of each test. */ +export const devServerTestingDir = tempDir; + // react-server-dom-bun's only published build was compiled against this react build. RSC // packages only pair with the exact react they were built for, so pin all three instead of // tracking the daily-moving `experimental` dist-tag (whose 2026-07-01 build broke the suite). diff --git a/test/bake/dev/hot.test.ts b/test/bake/dev/hot.test.ts index f0f230bce840..ff8f8acfbc38 100644 --- a/test/bake/dev/hot.test.ts +++ b/test/bake/dev/hot.test.ts @@ -1,7 +1,12 @@ // Hot tests ensure that the `import.meta.hot` interface is functional import { expect } from "bun:test"; -import { renameSync, unlinkSync, writeFileSync } from "node:fs"; -import { devTest, emptyHtmlFile } from "../bake-harness"; +import { isLinux } from "harness"; +import { closeSync, openSync, readFileSync, renameSync, statSync, unlinkSync, writeFileSync, writeSync } from "node:fs"; +import { join } from "node:path"; +import { devServerTestingDir, devTest, emptyHtmlFile, WAIT_MULTIPLIER } from "../bake-harness"; + +// Outside the dev server's root, so that the watcher does not report its own trace file. +const watcherTrace = join(devServerTestingDir, "hot-watcher-trace.log"); devTest("import.meta.hot.accept basic", { files: { @@ -691,3 +696,100 @@ devTest("dev.write resolves only after the new module body has run", { expect(await c.js`globalThis.marker`).toBe("updated"); }, }); + +devTest("watcher batches that arrive while a bundle runs all reach the next bundle", { + env: { BUN_WATCHER_TRACE: watcherTrace }, + files: { + "bunfig.toml": ` + [serve.static] + plugins = ["./hold-plugin.ts"] + `, + // Keeps a bundle that loads held.ts open for as long as hold/on exists. + "hold-plugin.ts": ` + import { existsSync, readFileSync } from "node:fs"; + import { join } from "node:path"; + + const on = join(import.meta.dir, "hold", "on"); + export default { + name: "hold", + setup(build) { + build.onLoad({ filter: /held\\.ts$/ }, async args => { + if (existsSync(on)) { + console.log("hold-plugin: the bundle of held.ts waits"); + while (existsSync(on)) await Bun.sleep(5); + } + return { contents: readFileSync(args.path, "utf8"), loader: "ts" }; + }); + }, + }; + `, + "hold/.keep": "", + "index.html": emptyHtmlFile({ scripts: ["index.ts"] }), + "index.ts": ` + import { held } from "./held"; + import { a } from "./a"; + import { b } from "./b"; + import { c } from "./c"; + import { d } from "./d"; + import { e } from "./e"; + console.log(held, a, b, c, d, e); + `, + "held.ts": `export const held = "held-0";`, + "a.ts": `export const a = "a-0";`, + "b.ts": `export const b = "b-0";`, + "c.ts": `export const c = "c-0";`, + "d.ts": `export const d = "d-0";`, + "e.ts": `export const e = "e-0";`, + }, + async test(dev) { + const served = async () => { + const html = await (await dev.fetch("/")).text(); + const src = html.match(/]+src="([^"]+)"[^>]*data-bun-dev-server-script/)![1]; + return await (await dev.fetch(src)).text(); + }; + // For the two signals that no event reports: a line in the watcher trace file, and the served bundle. + const until = async (what: string, done: () => boolean | Promise) => { + const deadline = Date.now() + 10_000 * WAIT_MULTIPLIER; + while (!(await done())) { + if (dev.output.exitCode !== null) throw new Error("The dev server exited while the test waited for " + what); + if (Date.now() > deadline) throw new Error("Timed out waiting for " + what); + await Bun.sleep(10); + } + }; + // One write and no truncate: the watcher gets one change per save. + const save = (name: string) => { + const fd = openSync(dev.join(name + ".ts"), "r+"); + writeSync(fd, `export const ${name} = "${name}-1";`, 0); + closeSync(fd); + }; + // The watcher logs one line per batch, before it handles the batch. + const batchesSince = (offset: number) => + readFileSync(watcherTrace).subarray(offset).toString().split("\n").filter(Boolean); + const names = (batch: string, name: string) => new RegExp(`[/\\\\"]${name}\\.ts"`).test(batch); + + expect(await served()).toContain('"held-0"'); + + writeFileSync(dev.join("hold/on"), ""); + save("held"); + await dev.output.waitForLine(/hold-plugin: the bundle of held\.ts waits/); + + const held = statSync(watcherTrace).size; + const saved = ["a", "b", "c", "d", "e"]; + for (const name of saved) { + const from = statSync(watcherTrace).size; + save(name); + await until(`the watcher to report ${name}.ts`, () => batchesSince(from).some(batch => names(batch, name))); + } + // One write is one inotify event, so on Linux each save is one batch. Other watchers can split a save. + if (isLinux) { + const batches = batchesSince(held).map(batch => saved.filter(name => names(batch, name)).join("+")); + expect(batches.filter(Boolean)).toEqual(saved); + } + // `on_load` locks `Watcher.mutex`, so the held bundle cannot end before the batch of e.ts is queued. + unlinkSync(dev.join("hold/on")); + + let text = ""; + await until("the bundle after the held one", async () => (text = await served()).includes('"e-1"')); + expect(saved.filter(name => !text.includes(`"${name}-1"`))).toEqual([]); + }, +});