Skip to content

s3: abort in-flight tasks at VM teardown by request id instead of reading task.http #38353

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
robobun wants to merge 1 commit into
base: main
Choose a base branch
from
Open

s3: abort in-flight tasks at VM teardown by request id instead of reading task.http #38353

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
48 changes: 14 additions & 34 deletions src/runtime/webcore/s3/client.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -295,8 +295,9 @@ pub(crate) fn list_objects(
let headers = bun_http::Headers::from_pico_http_headers(result.headers());

let task_ptr = bun_core::heap::into_raw(Box::new(S3HttpSimpleTask {
// Written below via `MaybeUninit::write` before any read.
// Both written by `S3HttpSimpleTask::schedule` below.
http: core::mem::MaybeUninit::uninit(),
async_http_id: 0,
sign_result: result,
callback_context,
callback: s3_simple_request::Callback::ListObjects(callback),
Expand Down Expand Up @@ -343,7 +344,7 @@ pub(crate) fn list_objects(
// JS thread (request setup): read options from the current VM.
let vm = VirtualMachine::get();

task.http.write(bun_http::AsyncHTTP::init(
let http = bun_http::AsyncHTTP::init(
bun_http::Method::GET,
url,
task.headers.entries.clone().expect("OOM"),
Expand All @@ -364,19 +365,10 @@ pub(crate) fn list_objects(
signals: Some(task.signal_store.to()),
..Default::default()
},
));

// queue http request
bun_http::http_thread::init(&Default::default());
let mut batch = bun_threading::thread_pool::Batch::default();
// SAFETY: `http` was initialised by `task.http.write(...)` immediately above.
unsafe { task.http.assume_init_mut() }.schedule(&mut batch);
// Out on the HTTP thread until its final callback: the VM aborts it at
// teardown (registry) and waits for it (embedded work).
task.loop_handle.embedded_work_scheduled();
crate::jsc_hooks::ActiveHandle::S3Request(core::ptr::NonNull::new(task_ptr).expect("task"))
.register();
bun_http::HTTPThread::schedule(batch);
);
// SAFETY: `task_ptr` was allocated above and has not been handed to anything yet
// (`task` is not used past this point); `http`'s callback context is `task_ptr`.
unsafe { S3HttpSimpleTask::schedule(task_ptr, http) };
Ok(())
}

Expand Down Expand Up @@ -1227,8 +1219,9 @@ fn download_stream(
};
let task_ptr = bun_core::heap::into_raw(S3HttpDownloadStreamingTask::new(
S3HttpDownloadStreamingTask {
// `http: undefined` — fully overwritten by `task.http.write(AsyncHTTP::init(...))` below.
// Both written by `S3HttpDownloadStreamingTask::schedule` below.
http: core::mem::MaybeUninit::uninit(),
async_http_id: 0,
sign_result: result,
proxy_url: owned_proxy,
callback_context: NonNull::new(callback_context.cast::<()>())
Expand All @@ -1252,7 +1245,6 @@ fn download_stream(
crate::webcore::s3::download_stream::State::default().0,
),
concurrent_task: Default::default(),
async_http_id: 0,
},
));
// SAFETY: just allocated via heap::alloc, non-null; lifetime owned by HTTP callback
Expand Down Expand Up @@ -1283,7 +1275,7 @@ fn download_stream(
let verbose = vm.get_verbose_fetch();
let reject_unauthorized = vm.get_tls_reject_unauthorized();

task.http.write(bun_http::AsyncHTTP::init(
let http = bun_http::AsyncHTTP::init(
bun_http::Method::GET,
url,
task.headers.entries.clone().expect("OOM"),
Expand All @@ -1304,22 +1296,10 @@ fn download_stream(
reject_unauthorized: Some(reject_unauthorized),
..Default::default()
},
));
// SAFETY: `http` was initialised by `task.http.write(...)` immediately above.
let http = unsafe { task.http.assume_init_mut() };
task.async_http_id = http.async_http_id;
// enable streaming
http.enable_response_body_streaming();
// queue http request
bun_http::http_thread::init(&Default::default());
let mut batch = bun_threading::thread_pool::Batch::default();
http.schedule(&mut batch);
// Out on the HTTP thread until its final callback: the VM aborts it at
// teardown (registry) and waits for it (embedded work).
task.loop_handle.embedded_work_scheduled();
crate::jsc_hooks::ActiveHandle::S3Download(core::ptr::NonNull::new(task_ptr).expect("task"))
.register();
bun_http::HTTPThread::schedule(batch);
);
// SAFETY: `task_ptr` was allocated above and has not been handed to anything yet
// (`task` is not used past this point); `http`'s callback context is `task_ptr`.
unsafe { S3HttpDownloadStreamingTask::schedule(task_ptr, http) };
task_ptr
}

Expand Down
46 changes: 38 additions & 8 deletions src/runtime/webcore/s3/download_stream.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -12,12 +12,18 @@ use bun_s3_signing::error::S3Error;

use crate::webcore::s3::xml_response;
use bun_threading::Mutex;
use bun_threading::thread_pool::Batch;

bun_core::declare_scope!(S3, hidden);

pub struct S3HttpDownloadStreamingTask {
// `MaybeUninit` because `AsyncHTTP` contains non-null references, so
// `mem::zeroed()` can't be used here (mirrors `S3HttpSimpleTask`).
//
// The HTTP thread's from `schedule` until the final callback hands the task
// back (`update_state` overwrites it on every callback); JS-thread code
// reaches the in-flight request through `async_http_id` instead. Enforced by
// test/internal/source-lints/s3-task-http-field.test.ts.
pub(crate) http: core::mem::MaybeUninit<AsyncHTTP<'static>>,
/// How the HTTP thread reaches the VM to deliver chunks.
pub(crate) loop_handle: bun_jsc::LoopHandle,
Expand All @@ -39,9 +45,8 @@ pub struct S3HttpDownloadStreamingTask {

pub(crate) concurrent_task: ConcurrentTask,
pub(crate) proxy_url: Box<[u8]>,
/// Captured once on the main thread before the request is queued so the cancel
/// path can call `schedule_shutdown_by_id` without dereferencing `http` (which
/// `update_state` overwrites on the HTTP thread under `mutex`).
/// Set by `schedule` before the hand-off; what the cancel and VM-teardown
/// paths pass to `schedule_shutdown_by_id` instead of reading `http`.
pub(crate) async_http_id: u32,
}

Expand All @@ -59,6 +64,29 @@ impl S3HttpDownloadStreamingTask {
Box::new(init)
}

/// Stores `http` in the task and hands the request to the HTTP thread (see
/// the `http` field for what that gives up).
///
/// # Safety
/// `this` is a live task from `Self::new` that nothing else references yet;
/// `http`'s callback context is `this`. JS thread.
pub(crate) unsafe fn schedule(this: *mut Self, mut http: AsyncHTTP<'static>) {
http.enable_response_body_streaming();
bun_http::http_thread::init(&Default::default());
let mut batch = Batch::default();
// SAFETY: fn contract; statement-scoped accesses. The `&mut AsyncHTTP`
// from `write` ends with the statement that queues its task node.
unsafe {
(*this).async_http_id = http.async_http_id;
(*this).http.write(http).schedule(&mut batch);
// Out on the HTTP thread until its final callback: the VM aborts it
// at teardown (registry) and waits for it (embedded work).
(*this).loop_handle.embedded_work_scheduled();
}
crate::jsc_hooks::ActiveHandle::S3Download(NonNull::new(this).expect("task")).register();
bun_http::HTTPThread::schedule(batch);
}

pub(crate) fn get_state(&self) -> State {
State(self.state.load(Ordering::Acquire))
}
Expand Down Expand Up @@ -205,8 +233,8 @@ impl S3HttpDownloadStreamingTask {
// SAFETY: `async_http` points to a live AsyncHTTP owned by the HTTP thread; a
// bitwise read+write copies its current state into `self.http` without running
// destructors (the HTTP thread retains ownership of the source until the request
// completes). `self.http` was previously initialised in
// `client::download_stream`.
// completes). `self.http` was initialised in `Self::schedule`, and nothing reads
// it on the JS thread while the request is in flight.
unsafe { core::ptr::write(self.http.as_mut_ptr(), core::ptr::read(async_http)) };
}
wait_until_done
Expand Down Expand Up @@ -349,15 +377,17 @@ impl S3HttpDownloadStreamingTask {
/// # Safety
/// `this` is live (registered ⇒ not yet freed by `on_response`); JS thread.
pub(crate) unsafe fn stop_for_vm_teardown(this: *mut Self) {
// SAFETY: fn contract; `http` is initialised before the task is registered.
// SAFETY: fn contract. Registered ⇒ in flight ⇒ the HTTP thread may be writing
// `http` right now; only the atomic abort flag and the schedule-time id are read.
unsafe {
(*this).signal_store.aborted.store(true, Ordering::Relaxed);
bun_http::http_thread().schedule_shutdown((*this).http.assume_init_ref());
bun_http::http_thread().schedule_shutdown_by_id((*this).async_http_id);
}
}

fn release_portable(&mut self) {
// SAFETY: `http` is always initialised before the task is scheduled / dropped.
// SAFETY: `http` was initialised by `Self::schedule`, and a task is only dropped
// once the final callback has handed it back, so the HTTP thread is done with it.
let http = unsafe { self.http.assume_init_mut() };
http.clear_data();
http.request_headers = Default::default();
Expand Down
73 changes: 48 additions & 25 deletions src/runtime/webcore/s3/simple_request.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -112,9 +112,17 @@ pub struct S3HttpSimpleTask {
// `http.clear_data()`, never a full destructor, and `http_callback` does a no-drop bitwise
// overwrite. Wrapping in `MaybeUninit` makes both possible: write-without-
// drop on assignment, and `clear_data()`-only in `Drop`. Invariant: `http` is initialised by
// `execute_simple_s3_request` before the task pointer escapes, so every later access (in
// `http_callback` / `Drop`) may `assume_init`.
// `Self::schedule` before the task pointer escapes, so `http_callback` / `Drop` may
// `assume_init`.
//
// The HTTP thread's from `schedule` until the final callback hands the task back
// (`stage_http_result` overwrites it on every callback); JS-thread code reaches the
// in-flight request through `async_http_id` instead. Enforced by
// test/internal/source-lints/s3-task-http-field.test.ts.
pub(crate) http: core::mem::MaybeUninit<AsyncHTTP<'static>>,
/// Set by `schedule` before the hand-off; what `stop_for_vm_teardown` passes to
/// `schedule_shutdown_by_id` instead of reading `http`.
pub(crate) async_http_id: u32,
/// How the HTTP thread reaches the VM to deliver the response.
pub(crate) loop_handle: bun_jsc::LoopHandle,
pub(crate) sign_result: SignResult,
Expand Down Expand Up @@ -212,6 +220,29 @@ impl S3HttpSimpleTask {
bun_core::heap::into_raw(Box::new(init))
}

/// Stores `http` in the task and hands the request to the HTTP thread (see
/// the `http` field for what that gives up).
///
/// # Safety
/// `this` is a live task from `Self::new` that nothing else references yet;
/// `http`'s callback context is `this`. JS thread.
pub(crate) unsafe fn schedule(this: *mut Self, http: AsyncHTTP<'static>) {
bun_http::http_thread::init(&Default::default());
let mut batch = thread_pool::Batch::default();
// SAFETY: fn contract; statement-scoped accesses. The `&mut AsyncHTTP`
// from `write` ends with the statement that queues its task node.
unsafe {
(*this).async_http_id = http.async_http_id;
(*this).http.write(http).schedule(&mut batch);
// Out on the HTTP thread until its final callback: the VM aborts it
// at teardown (registry) and waits for it (embedded work).
(*this).loop_handle.embedded_work_scheduled();
}
crate::jsc_hooks::ActiveHandle::S3Request(core::ptr::NonNull::new(this).expect("task"))
.register();
bun_http::HTTPThread::schedule(batch);
}

fn error_with_body(&self, error_type: ErrorType) -> JsTerminatedResult<()> {
let mut code: &[u8] = b"UnknownError";
let mut message: &[u8] = b"an unexpected error has occurred";
Expand Down Expand Up @@ -405,7 +436,8 @@ impl S3HttpSimpleTask {
// conceptually transfers here; the http-thread side must free only its outer
// allocation (TrivialDeinit).
// SAFETY: `async_http` is a valid live pointer for the duration of this callback;
// `self.http` was previously initialised in `execute_simple_s3_request`.
// `self.http` was initialised in `Self::schedule`, and nothing reads it on the JS
// thread while the request is in flight.
unsafe { core::ptr::write(self.http.as_mut_ptr(), core::ptr::read(async_http)) };
}

Expand Down Expand Up @@ -477,16 +509,17 @@ impl S3HttpSimpleTask {
/// # Safety
/// `this` is live (registered ⇒ its response has not run); JS thread.
pub(crate) unsafe fn stop_for_vm_teardown(this: *mut Self) {
// SAFETY: fn contract; `http` is initialised before the task is registered.
// SAFETY: fn contract. Registered ⇒ in flight ⇒ the HTTP thread may be writing
// `http` right now; only the atomic abort flag and the schedule-time id are read.
unsafe {
(*this).signal_store.aborted.store(true, Ordering::Relaxed);
bun_http::http_thread().schedule_shutdown((*this).http.assume_init_ref());
bun_http::http_thread().schedule_shutdown_by_id((*this).async_http_id);
}
}

fn release_portable(&mut self) {
// SAFETY: `http` is always initialised before the task pointer escapes (see
// `execute_simple_s3_request`).
// SAFETY: `http` was initialised by `Self::schedule`, and a task is only dropped
// once the final callback has handed it back, so the HTTP thread is done with it.
let http = unsafe { self.http.assume_init_mut() };
http.clear_data();
http.request_headers = Default::default();
Expand Down Expand Up @@ -631,8 +664,9 @@ pub(crate) fn execute_simple_s3_request(
));
let proxy = options.proxy_url.unwrap_or(b"");
let task_ptr = S3HttpSimpleTask::new(S3HttpSimpleTask {
// written below via `MaybeUninit::write` before any read.
// Both written by `S3HttpSimpleTask::schedule` below.
http: core::mem::MaybeUninit::uninit(),
async_http_id: 0,
sign_result: result,
callback_context,
callback,
Expand All @@ -650,8 +684,8 @@ pub(crate) fn execute_simple_s3_request(
poll_ref,
signal_store: Default::default(),
});
// SAFETY: `task_ptr` is a freshly heap-allocated pointer; shared reads only until
// the scoped exclusive `http` writes below.
// SAFETY: `task_ptr` is a freshly heap-allocated pointer; shared reads only, all of
// them before the exclusive accesses through `task_ptr` below.
let task = unsafe { &*task_ptr };
// SAFETY: lifetime extension — `url`, `headers_buf`, and `proxy_url` borrow from
// heap-allocated fields of `*task` (sign_result.url / headers.buf / proxy_url) which the task
Expand Down Expand Up @@ -702,20 +736,9 @@ pub(crate) fn execute_simple_s3_request(
..Default::default()
},
);
// SAFETY: `task_ptr` is still the sole pointer (the HTTP thread only sees it after
// `schedule` below); scoped exclusive write of the `http` field.
unsafe { (*task_ptr).http.write(async_http) };
// queue http request
bun_http::http_thread::init(&Default::default());
let mut batch = thread_pool::Batch::default();
// SAFETY: `http` was initialised immediately above; scoped exclusive access.
unsafe { (*task_ptr).http.assume_init_mut() }.schedule(&mut batch);
// Out on the HTTP thread until its final callback: the VM aborts it at
// teardown (registry) and waits for it (embedded work).
// SAFETY: as above.
unsafe { (*task_ptr).loop_handle.embedded_work_scheduled() };
crate::jsc_hooks::ActiveHandle::S3Request(core::ptr::NonNull::new(task_ptr).expect("task"))
.register();
bun_http::HTTPThread::schedule(batch);
// SAFETY: `task_ptr` is still the sole pointer (the HTTP thread only sees it once
// `schedule` queues it) and `task` is not used past this point; `async_http`'s callback
// context is `task_ptr`.
unsafe { S3HttpSimpleTask::schedule(task_ptr, async_http) };
Ok(())
}
Loading