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
12 changes: 10 additions & 2 deletions src/bundler/LinkerContext.rs
Original file line number Diff line number Diff line change
Expand Up @@ -586,8 +586,16 @@ impl<'a> LinkerContext<'a> {
callback: SourceMapDataTask::run_quoted_source_contents,
},
};
batch.push(ThreadPoolLib::Batch::from(&raw mut line_offset.thread_task));
second_batch.push(ThreadPoolLib::Batch::from(&raw mut quoted.thread_task));
let line_offset = std::ptr::from_mut(line_offset);
let quoted = std::ptr::from_mut(quoted);
// SAFETY: field projections of the two live elements; the callbacks
// recover the whole `SourceMapDataTask` from these pointers.
unsafe {
batch.push(ThreadPoolLib::Batch::from(
&raw mut (*line_offset).thread_task,
));
second_batch.push(ThreadPoolLib::Batch::from(&raw mut (*quoted).thread_task));
}
}

// line offsets block sooner and are faster to compute, so we should schedule those first
Expand Down
34 changes: 18 additions & 16 deletions src/bundler/linker_context/generateChunksInParallel.rs
Original file line number Diff line number Diff line change
Expand Up @@ -98,7 +98,6 @@ pub(crate) fn generate_chunks_in_parallel<const IS_DEV_SERVER: bool>(

debug!(" START {} prepare CSS ast (total count)", total_count);

let mut batch = ThreadPoolLib::Batch::default();
let mut tasks: Vec<PrepareCssAstTask> = Vec::with_capacity(total_count);
for chunk in chunks.iter_mut() {
if chunk.content.is_css() {
Expand All @@ -112,12 +111,18 @@ pub(crate) fn generate_chunks_in_parallel<const IS_DEV_SERVER: bool>(
// (raw ptr is invariant); `.cast()` erases the inner `'a` to satisfy it.
linker: std::ptr::from_mut::<LinkerContext>(c).cast(),
});
// Capacity pre-reserved → push never reallocates → ptr stays stable.
let task = tasks.last_mut().unwrap();
batch.push(ThreadPoolLib::Batch::from(&raw mut task.task));
}
}
debug_assert_eq!(tasks.len(), total_count);
let tasks_ptr = tasks.as_mut_ptr();
let mut batch = ThreadPoolLib::Batch::default();
for i in 0..tasks.len() {
// SAFETY: `i < tasks.len()`; the callback recovers the whole
// `PrepareCssAstTask` from this element pointer.
batch.push(ThreadPoolLib::Batch::from(unsafe {
&raw mut (*tasks_ptr.add(i)).task
}));
}
// SAFETY: `parse_graph` is the `BundleV2.graph` backref (valid for
// the link step); `pool` is the arena-allocated bundler ThreadPool.
let worker_pool = c.worker_pool();
Expand Down Expand Up @@ -176,10 +181,7 @@ pub(crate) fn generate_chunks_in_parallel<const IS_DEV_SERVER: bool>(
debug_assert_eq!(chunks.len(), chunk_contexts.len());

debug!(" START {} compiling part ranges", total_count);
// Pre-reserved to `total_count` so pushes never reallocate; the
// batch holds raw pointers into this buffer.
let mut combined_part_ranges: Vec<PendingPartRange> = Vec::with_capacity(total_count);
let mut batch = ThreadPoolLib::Batch::default();
for (chunk, chunk_ctx) in chunks.iter_mut().zip(chunk_contexts.iter_mut()) {
match &chunk.content {
crate::chunk::Content::Javascript(js) => {
Expand All @@ -200,9 +202,6 @@ pub(crate) fn generate_chunks_in_parallel<const IS_DEV_SERVER: bool>(
bun_ptr::detach_lifetime_ref::<GenerateChunkCtx>(chunk_ctx)
},
});
batch.push(ThreadPoolLib::Batch::from(
&raw mut combined_part_ranges.last_mut().unwrap().task,
));
}
}
crate::chunk::Content::Css(css) => {
Expand All @@ -223,9 +222,6 @@ pub(crate) fn generate_chunks_in_parallel<const IS_DEV_SERVER: bool>(
bun_ptr::detach_lifetime_ref::<GenerateChunkCtx>(chunk_ctx)
},
});
batch.push(ThreadPoolLib::Batch::from(
&raw mut combined_part_ranges.last_mut().unwrap().task,
));
}
}
crate::chunk::Content::Html => {
Expand All @@ -245,13 +241,19 @@ pub(crate) fn generate_chunks_in_parallel<const IS_DEV_SERVER: bool>(
bun_ptr::detach_lifetime_ref::<GenerateChunkCtx>(chunk_ctx)
},
});
batch.push(ThreadPoolLib::Batch::from(
&raw mut combined_part_ranges.last_mut().unwrap().task,
));
}
}
}
debug_assert_eq!(combined_part_ranges.len(), total_count);
let part_ranges_ptr = combined_part_ranges.as_mut_ptr();
let mut batch = ThreadPoolLib::Batch::default();
for i in 0..combined_part_ranges.len() {
// SAFETY: `i < combined_part_ranges.len()`; the callbacks recover
// the whole `PendingPartRange` from this element pointer.
batch.push(ThreadPoolLib::Batch::from(unsafe {
&raw mut (*part_ranges_ptr.add(i)).task
}));
}
// SAFETY: `parse_graph` is the `BundleV2.graph` backref (valid for
// the link step); `pool` is the arena-allocated bundler ThreadPool.
let worker_pool = c.worker_pool();
Expand Down
9 changes: 7 additions & 2 deletions src/http/AsyncHTTP.rs
Original file line number Diff line number Diff line change
Expand Up @@ -388,7 +388,9 @@ pub fn preconnect(url: URL<'static>, is_url_owned: bool) {
));
async_http.client.flags.is_preconnect_only = true;

crate::HTTPThread::schedule(Batch::from(core::ptr::addr_of_mut!(async_http.task)));
let mut batch = Batch::default();
async_http.schedule(&mut batch);
crate::HTTPThread::schedule(batch);
}
}

Expand Down Expand Up @@ -527,7 +529,10 @@ impl<'a> AsyncHTTP<'a> {
}

pub fn schedule(&mut self, batch: &mut Batch) {
batch.push(Batch::from(core::ptr::addr_of_mut!(self.task)));
let this = core::ptr::from_mut(self);
// SAFETY: field projection of `self`; `HTTPThread::schedule` recovers
// the whole `AsyncHTTP` from this pointer.
batch.push(Batch::from(unsafe { &raw mut (*this).task }));
}
}

Expand Down
8 changes: 5 additions & 3 deletions src/install/PackageManager/runTasks.rs
Original file line number Diff line number Diff line change
Expand Up @@ -1604,12 +1604,14 @@ pub fn flush_patch_task_queue(this: &mut PackageManager) {
while let Some(patch_task) = this.patch_task_fifo.read_item() {
// SAFETY: fifo stores live `*mut PatchTask` pushed by
// `enqueue_patch_task`; exclusive ownership transferred here.
let pt = unsafe { &mut *patch_task };
pt.schedule(if matches!(pt.callback, PatchTaskCallback::Apply(_)) {
let is_apply = unsafe { (*patch_task).callback.is_apply() };
let batch = if is_apply {
&mut this.patch_apply_batch
} else {
&mut this.patch_calc_hash_batch
});
};
// SAFETY: as above.
unsafe { PatchTask::schedule(patch_task, batch) };
}
}

Expand Down
5 changes: 4 additions & 1 deletion src/install/isolated_install/Installer.rs
Original file line number Diff line number Diff line change
Expand Up @@ -166,9 +166,12 @@ impl<'a> Installer<'a> {
));

task.result = Result::None;
let task = std::ptr::from_mut(task);
// SAFETY: field projection of the live `tasks[entry_id]` slot;
// `Task::callback` recovers the whole `Task` from this pointer.
manager
.thread_pool
.schedule(thread_pool::Batch::from(&raw mut task.task));
.schedule(thread_pool::Batch::from(unsafe { &raw mut (*task).task }));
}

pub(crate) fn on_package_extracted(&mut self, task_id: crate::package_manager_task::Id) {
Expand Down
8 changes: 6 additions & 2 deletions src/install/patch_install.rs
Original file line number Diff line number Diff line change
Expand Up @@ -707,8 +707,12 @@ impl PatchTask {
Some(hasher.final_())
}

pub(crate) fn schedule(&mut self, batch: &mut Batch) {
batch.push(Batch::from(&raw mut self.task));
/// # Safety
/// `this` must be the live heap `PatchTask` handed out by `enqueue_patch_task`.
Comment thread
robobun marked this conversation as resolved.
Resolved
pub(crate) unsafe fn schedule(this: *mut Self, batch: &mut Batch) {
// SAFETY: field projection of the live task; `run_from_thread_pool`
// recovers the whole task from this pointer.
batch.push(Batch::from(unsafe { &raw mut (*this).task }));
}

pub(crate) fn new_calc_patch_hash(
Expand Down
9 changes: 5 additions & 4 deletions src/threading/ThreadPool.rs
Original file line number Diff line number Diff line change
Expand Up @@ -554,10 +554,11 @@ impl ThreadPool {
ctx: bun_ptr::BackRef::new(&wait_context),
});
}
// Push to the Vec first (no realloc: capacity reserved), then take
// stable addresses.
for runner_task in tasks.iter_mut() {
batch.push(Batch::from(ptr::addr_of_mut!(runner_task.task)));
let runner_tasks = tasks.as_mut_ptr();
for i in 0..tasks.len() {
// SAFETY: `i < tasks.len()`; `call` recovers the whole `RunnerTask`
// from this element pointer.
batch.push(Batch::from(unsafe { &raw mut (*runner_tasks.add(i)).task }));
}
self.schedule(batch);
self.wait_for_all();
Expand Down
Loading
Loading