Skip to content
Open
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
26 changes: 26 additions & 0 deletions src/event_loop/AnyEventLoop.rs
Original file line number Diff line number Diff line change
Expand Up @@ -462,6 +462,32 @@ impl EventLoopHandle {
EnteredEventLoop(self)
}

/// Owning thread: queue `owner` to run once pending I/O has had a turn —
/// the JS loop's after-yield queue (delivered by `T::TAG`), or for a mini
/// loop one tick of its platform loop, then its task queue (delivered to
/// `R`). For work that re-queues itself and would otherwise starve I/O.
pub fn enqueue_boxed_after_yield<T, R>(
self,
owner: Box<T>,
node: fn(&mut T) -> &mut AnyTaskWithExtraContext,
) where
T: crate::Taskable,
R: crate::AnyTaskWithExtraContext::BoxedMiniTaskRunner<T>,
{
match self {
EventLoopHandle::Js { owner: js } => {
js.enqueue_task_after_yield(crate::Task::from_boxed(owner));
}
EventLoopHandle::Mini(mut mini) => {
let mini = mini_mut(&mut mini);
mini.tick_platform_loop();
mini.enqueue_task_concurrent(AnyTaskWithExtraContext::arm_boxed::<T, R>(
owner, node,
));
}
}
}

/// Owning thread: the weak poster other threads use to deliver JS-loop
/// tasks to this handle's VM; `None` for a mini loop (post to it directly —
/// it is owned by, and outlives the work of, its thread).
Expand Down
7 changes: 7 additions & 0 deletions src/event_loop/MiniEventLoop.rs
Original file line number Diff line number Diff line change
Expand Up @@ -351,6 +351,13 @@ impl MiniEventLoop {
}
}

/// One tick of the platform (uSockets) loop, so pending I/O is serviced
/// before the caller queues more work.
pub fn tick_platform_loop(&mut self) {
// SAFETY: see `loop_ptr()` invariant.
unsafe { (*self.loop_ptr()).tick() };
}

/// `task` must outlive the queued work item; ownership of the intrusive
/// node stays with the caller until the callback runs.
pub fn enqueue_task_concurrent(&mut self, task: NonNull<AnyTaskWithExtraContext>) {
Expand Down
13 changes: 5 additions & 8 deletions src/runtime/dispatch.rs
Original file line number Diff line number Diff line change
Expand Up @@ -539,16 +539,13 @@ fn run_task_cold(task: Task) {
task_tag::ShellMvBatchedTask => shell_dispatch!(ShellMvBatchedTask),
task_tag::ShellMvCheckTargetTask => shell_dispatch!(ShellMvCheckTargetTask),
task_tag::ShellRmTask => shell_dispatch!(ShellRmTask),
task_tag::ShellRmDirTask => {
let t = cast_ptr!(ShellRmDirTask);
ShellRmDirTask::run_from_main_thread(t);
}
task_tag::ShellRmDirTask => shell_dispatch!(ShellRmDirTask),
task_tag::ShellGlobTask => shell_dispatch!(ShellGlobTask),
task_tag::ShellYesTask => {
// SAFETY: §Dispatch — tag identifies pointee; enqueued by
// `YesTask::enqueue`, storage lives inside `Box<Yes>` in the
// interpreter arena and is stable until the builtin deinits.
ShellYesTask::run_from_main_thread(unsafe { &*cast_ptr!(ShellYesTask) });
// SAFETY: §Dispatch — tag identifies pointee: the box
// `YesTask::enqueue` leaked when posting.
let t = unsafe { bun_core::heap::take(cast_ptr!(ShellYesTask)) };
t.run();
}

// ── bake dev-server ──────────────────────────────────────────────
Expand Down
36 changes: 15 additions & 21 deletions src/runtime/node/node_fs.rs
Original file line number Diff line number Diff line change
Expand Up @@ -1371,10 +1371,9 @@ mod _async_tasks {
/// enqueued once the count reaches zero, so subtasks still running on the
/// thread pool never dereference a freed parent.
pub(crate) subtask_count: AtomicUsize,
/// BACKREF — `Some` iff `IS_SHELL`. The shell `ShellCpTask` owns and
/// outlives this task; `ParentRef` gives a safe `&ShellCpTask` projection
/// for `cp_on_copy` and round-trips the `*mut` for `cp_on_finish`.
pub(crate) shelltask: Option<bun_ptr::ParentRef<ShellCpTask, bun_ptr::Mut>>,
/// `Some` iff `IS_SHELL`: the shell `cp` task this copy reports to
/// (`cp_on_copy` per file; handed back through `cp_on_finish`).
pub(crate) shelltask: Option<Box<ShellCpTask>>,
}

bun_threading::intrusive_work_task!([const IS_SHELL: bool] NewAsyncCpTask<IS_SHELL>, task);
Expand Down Expand Up @@ -1507,10 +1506,10 @@ mod _async_tasks {
if !IS_SHELL {
return;
}
// When IS_SHELL, `shelltask` is `Some` (ParentRef invariant: owner
// outlives this task). Shared borrow only — concurrent subtasks may
// call this in parallel; `cp_on_copy` serialises via its internal mutex.
// Shared borrow only — concurrent subtasks may call this in
// parallel; `cp_on_copy` serialises via its internal mutex.
self.shelltask
.as_deref()
.expect("IS_SHELL ⇒ shelltask")
.cp_on_copy(src.as_ref(), dest.as_ref());
}
Expand All @@ -1531,7 +1530,7 @@ mod _async_tasks {
EventLoopHandle::init(vm.event_loop.cast()),
bun_jsc::ConcurrentPoster::Js(vm.ticket()),
tracker,
core::ptr::null_mut(),
None,
);
// SAFETY: `schedule_new` returns a Box::leak'd pointer; valid until destroy()
unsafe { &*task }.promise.value()
Expand All @@ -1544,16 +1543,16 @@ mod _async_tasks {
cp_args: ThreadIsolated<args::Cp<'static>>,
evtloop: EventLoopHandle,
poster: bun_jsc::ConcurrentPoster,
shelltask: *mut ShellCpTask,
) -> *mut Self {
shelltask: Box<ShellCpTask>,
) {
Self::schedule_new(
JSPromiseStrong::default(),
cp_args,
evtloop,
poster,
AsyncTaskTracker { id: 0 },
shelltask,
)
Some(shelltask),
);
}

fn schedule_new(
Expand All @@ -1562,7 +1561,7 @@ mod _async_tasks {
evtloop: EventLoopHandle,
poster: bun_jsc::ConcurrentPoster,
tracker: AsyncTaskTracker,
shelltask: *mut ShellCpTask,
shelltask: Option<Box<ShellCpTask>>,
) -> *mut Self {
let mut task = Box::new(Self {
promise,
Expand All @@ -1577,9 +1576,7 @@ mod _async_tasks {
r#ref: KeepAlive::default(),
tracker,
subtask_count: AtomicUsize::new(1),
// SAFETY: `shelltask` (when non-null) is the live heap-alloc'd `ShellCpTask`
// that owns and outlives this task; pointer carries write provenance.
shelltask: unsafe { bun_ptr::ParentRef::from_nullable_mut(shelltask) },
shelltask,
});
if !IS_SHELL {
task.r#ref.ref_(event_loop_handle_to_ctx(task.evtloop));
Expand Down Expand Up @@ -1678,16 +1675,13 @@ mod _async_tasks {

pub(crate) fn run_from_js_thread(&mut self) -> JsResult<()> {
if IS_SHELL {
// SAFETY: shelltask is set by create_for_shell and outlives this task
// Move the result out — `Maybe<ret::Cp>` (= `Maybe<()>`) has a cheap
// `Ok(())` placeholder.
let result = core::mem::replace(self.result.get_mut(), Ok(()));
let src = core::mem::take(&mut self.args.src);
let dest = core::mem::take(&mut self.args.dest);
let shelltask = self.shelltask.expect("IS_SHELL ⇒ shelltask").as_mut_ptr();
// SAFETY: shelltask is non-null in the IS_SHELL specialization and
// outlives this task; `cp_on_finish` reclaims it.
unsafe { ShellCpTask::cp_on_finish(shelltask, src, dest, result) };
let shelltask = self.shelltask.take().expect("IS_SHELL ⇒ shelltask");
shelltask.cp_on_finish(src, dest, result);
// SAFETY: self was Box::leak'd in create*(); destroyed exactly once here
unsafe { Self::destroy(std::ptr::from_mut::<Self>(self)) };
return Ok(());
Expand Down
6 changes: 3 additions & 3 deletions src/runtime/shell/Builtin.rs
Original file line number Diff line number Diff line change
Expand Up @@ -158,7 +158,7 @@ macro_rules! shell_builtins {

/// Hoisted dispatch for the `onIOWriterChunk` callback. `seq != 0`
/// names a parked [`OutputTask`](crate::shell::interpreter::OutputTask)
/// chunk of an ls/mkdir/touch/cp.
/// chunk of an ls/mkdir/touch/cp, or one of rm's numbered chunks.
pub fn on_io_writer_chunk(
interp: &Interpreter,
cmd: NodeId,
Expand All @@ -174,8 +174,8 @@ macro_rules! shell_builtins {
Kind::Mkdir => OutputTask::<Mkdir>::on_chunk(interp, cmd, seq, written, err),
Kind::Touch => OutputTask::<Touch>::on_chunk(interp, cmd, seq, written, err),
Kind::Cp => OutputTask::<Cp>::on_chunk(interp, cmd, seq, written, err),
// rm numbers its verbose chunks only so that each is
// called back.
// rm numbers its verbose and error chunks only so that
// each is called back.
Kind::Rm => crate::shell::builtins::rm::Rm::on_io_writer_chunk(interp, cmd, written, err),
other => unreachable!("{} queues no numbered chunks", other.as_str()),
};
Expand Down
84 changes: 36 additions & 48 deletions src/runtime/shell/EnvMap.rs
Original file line number Diff line number Diff line change
@@ -1,4 +1,6 @@
use bun_collections::array_hash_map::{self, ArrayHashContext, ArrayHashMap, Iter};
use bun_collections::array_hash_map::{
self, ArrayHashAdapter, ArrayHashContext, ArrayHashMap, Iter,
};

use crate::shell::EnvStr;

Expand All @@ -15,37 +17,56 @@ type EnvMapInner = ArrayHashMap<EnvStr, EnvStr, EnvMapContext>;
#[derive(Default)]
struct EnvMapContext;

impl ArrayHashContext<EnvStr> for EnvMapContext {
fn hash(&self, s: &EnvStr) -> u32 {
impl EnvMapContext {
#[inline]
fn hash_bytes(s: &[u8]) -> u32 {
#[cfg(windows)]
{
return <array_hash_map::CaseInsensitiveAsciiStringContext as ArrayHashContext<[u8]>>::hash(
&array_hash_map::CaseInsensitiveAsciiStringContext::default(),
s.slice(),
s,
);
}
#[cfg(not(windows))]
{
array_hash_map::hash_string(s.slice())
array_hash_map::hash_string(s)
}
}

fn eql(&self, a: &EnvStr, b: &EnvStr, _b_index: usize) -> bool {
#[inline]
fn eql_bytes(a: &[u8], b: &[u8]) -> bool {
#[cfg(windows)]
{
// Must be length-checked: "PATH" must NOT match "PATHEXT".
return bun_core::strings::eql_case_insensitive_asciii_check_length(
a.slice(),
b.slice(),
);
return bun_core::strings::eql_case_insensitive_asciii_check_length(a, b);
}
#[cfg(not(windows))]
{
a.slice() == b.slice()
a == b
}
}
}

impl ArrayHashContext<EnvStr> for EnvMapContext {
fn hash(&self, s: &EnvStr) -> u32 {
Self::hash_bytes(s.slice())
}

fn eql(&self, a: &EnvStr, b: &EnvStr, _b_index: usize) -> bool {
Self::eql_bytes(a.slice(), b.slice())
}
}

impl ArrayHashAdapter<[u8], EnvStr> for EnvMapContext {
fn hash(&self, key: &[u8]) -> u32 {
Self::hash_bytes(key)
}

fn eql(&self, a: &[u8], b: &EnvStr, _b_index: usize) -> bool {
Self::eql_bytes(a, b.slice())
}
}

impl EnvMap {
pub(crate) fn init() -> EnvMap {
EnvMap {
Expand All @@ -71,16 +92,8 @@ impl EnvMap {
}
}

/// NOTE: This will `.ref()` value, so you should `defer value.deref()` it
/// before handing it to this function!!!
pub(crate) fn insert(&mut self, key: EnvStr, val: EnvStr) {
let result = self.map.get_or_put(key).expect("OOM");
if !result.found_existing {
key.ref_();
} else {
result.value_ptr.deref();
}
val.ref_();
*result.value_ptr = val;
}

Expand All @@ -96,39 +109,14 @@ impl EnvMap {
self.map.ensure_total_capacity(new_capacity).expect("OOM");
}

/// NOTE: Make sure you deref the string when done!
pub fn get(&self, key: EnvStr) -> Option<EnvStr> {
let val = *self.map.get(&key)?;
val.ref_();
Some(val)
pub fn get(&self, key: &[u8]) -> Option<&EnvStr> {
let i = self.map.get_index_adapted(key, &EnvMapContext)?;
Some(&self.map.values()[i])
}

pub(crate) fn clone(&self) -> EnvMap {
let new = EnvMap {
EnvMap {
map: self.map.clone().expect("OOM"),
};
new.ref_strings();
new
}

fn ref_strings(&self) {
for (key, value) in self.map.keys().iter().zip(self.map.values()) {
key.ref_();
value.ref_();
}
}

fn deref_strings(&self) {
for (key, value) in self.map.keys().iter().zip(self.map.values()) {
key.deref();
value.deref();
}
}
}

impl Drop for EnvMap {
fn drop(&mut self) {
self.deref_strings();
// map storage freed by its own Drop
}
}
Loading
Loading