yenru0/zed
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

gpui_web: Support Fetch requests from background workers (#61782)

Browse Source 
Browser Fetch APIs must run on the main thread, while GPUI image loading
can invoke its HTTP client from background WASM workers. This routes
`FetchHttpClient` requests through the web dispatcher's main-thread
mailbox, installs the client by default for web applications, and
removes thread-affine browser state from the safely movable dispatcher.

Release Notes:
- N/A
This commit is contained in:
Gaauwe Rombouts authored and GitHub committed 2026-07-28 13:32:52 +00:00
1 parent dbd8acd97b
commit 182babebf0
6 files changed
+198 -195
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

crates/gpui/examples/image/image.rs
-10
View File
@@ -164,16 +164,6 @@ fn run_example() {
let http_client = ReqwestClient::user_agent("gpui example").unwrap();
cx.set_http_client(Arc::new(http_client));
}
#[cfg(target_family = "wasm")]
{
// Safety: the web examples run single-threaded; the client is
// created and used exclusively on the main thread.
let http_client = unsafe {
gpui_web::FetchHttpClient::with_user_agent("gpui example")
.expect("failed to create FetchHttpClient")
};
cx.set_http_client(Arc::new(http_client));
}
cx.activate(true);
cx.on_action(|_: &Quit, cx| cx.quit());
crates/gpui/examples/image_gallery.rs
-10
View File
@@ -259,16 +259,6 @@ fn run_example() {
let http_client = ReqwestClient::user_agent("gpui example").unwrap();
cx.set_http_client(Arc::new(http_client));
}
#[cfg(target_family = "wasm")]
{
// Safety: the web examples run single-threaded; the client is
// created and used exclusively on the main thread.
let http_client = unsafe {
gpui_web::FetchHttpClient::with_user_agent("gpui example")
.expect("failed to create FetchHttpClient")
};
cx.set_http_client(Arc::new(http_client));
}
cx.activate(true);
cx.on_action(|_: &Quit, cx| cx.quit());
crates/gpui_platform/src/gpui_platform.rs
+11 -1
View File
@@ -11,6 +11,14 @@ pub fn background_executor() -> gpui::BackgroundExecutor {
}
pub fn application() -> gpui::Application {
#[cfg(target_family = "wasm")]
{
let platform = Rc::new(gpui_web::WebPlatform::new(true));
let http_client = std::sync::Arc::new(platform.fetch_http_client());
gpui::Application::with_platform(platform).with_http_client(http_client)
}
#[cfg(not(target_family = "wasm"))]
gpui::Application::with_platform(current_platform(false))
}
@@ -21,7 +29,9 @@ pub fn headless() -> gpui::Application {
/// Unlike `application`, this function returns a single-threaded web application.
#[cfg(target_family = "wasm")]
pub fn single_threaded_web() -> gpui::Application {
gpui::Application::with_platform(Rc::new(gpui_web::WebPlatform::new(false)))
let platform = Rc::new(gpui_web::WebPlatform::new(false));
let http_client = std::sync::Arc::new(platform.fetch_http_client());
gpui::Application::with_platform(platform).with_http_client(http_client)
}
/// Initializes panic hooks and logging for the web platform.
crates/gpui_web/src/dispatcher.rs
+39 -15
View File
@@ -21,12 +21,25 @@ fn shared_memory_supported() -> bool {
has_shared_array_buffer && has_atomics && is_shared_buffer
}
fn wait_async_supported() -> bool {
let global = js_sys::global();
let Ok(atomics) = js_sys::Reflect::get(&global, &JsValue::from_str("Atomics")) else {
return false;
};
let Ok(wait_async) = js_sys::Reflect::get(&atomics, &JsValue::from_str("waitAsync")) else {
return false;
};
wait_async.is_function()
}
enum MainThreadItem {
Runnable(RunnableVariant),
Delayed {
runnable: RunnableVariant,
millis: i32,
},
Function(Box<dyn FnOnce() + Send>),
// TODO-Wasm: Shouldn't these run on their own dedicated thread?
RealtimeFunction(Box<dyn FnOnce() + Send>),
}
@@ -127,7 +140,6 @@ impl MainThreadMailbox {
pub struct WebDispatcher {
main_thread_id: std::thread::ThreadId,
browser_window: web_sys::Window,
background_sender: PriorityQueueSender<RunnableVariant>,
main_thread_mailbox: Arc<MainThreadMailbox>,
supports_threads: bool,
@@ -135,11 +147,6 @@ pub struct WebDispatcher {
_background_threads: Vec<wasm_thread::JoinHandle<()>>,
}
// Safety: `web_sys::Window` is only accessed from the main thread
// All other fields are `Send + Sync` by construction.
unsafe impl Send for WebDispatcher {}
unsafe impl Sync for WebDispatcher {}
impl WebDispatcher {
pub fn new(browser_window: web_sys::Window, allow_threads: bool) -> Self {
#[cfg(feature = "multithreaded")]
@@ -149,14 +156,16 @@ impl WebDispatcher {
let main_thread_mailbox = Arc::new(MainThreadMailbox::new());
let supports_threads =
cfg!(feature = "multithreaded") && allow_threads && shared_memory_supported();
let supports_threads = cfg!(feature = "multithreaded")
&& allow_threads
&& shared_memory_supported()
&& wait_async_supported();
if supports_threads {
main_thread_mailbox.run_waker_loop(browser_window.clone());
} else if cfg!(feature = "multithreaded") && allow_threads {
log::warn!(
"SharedArrayBuffer not available; falling back to single-threaded dispatcher"
"Required WebAssembly threading APIs are unavailable; falling back to single-threaded dispatcher"
);
}
@@ -197,7 +206,6 @@ impl WebDispatcher {
Self {
main_thread_id: std::thread::current().id(),
browser_window,
background_sender,
main_thread_mailbox,
supports_threads,
@@ -209,6 +217,19 @@ impl WebDispatcher {
fn on_main_thread(&self) -> bool {
std::thread::current().id() == self.main_thread_id
}
pub(crate) fn dispatch_function_on_main_thread(
&self,
function: impl FnOnce() + Send + 'static,
) {
if self.on_main_thread() {
let callback = Closure::once_into_js(function);
browser_window().queue_microtask(callback.unchecked_ref());
} else {
self.main_thread_mailbox
.post(Priority::High, MainThreadItem::Function(Box::new(function)));
}
}
}
impl PlatformDispatcher for WebDispatcher {
@@ -235,7 +256,7 @@ impl PlatformDispatcher for WebDispatcher {
fn dispatch_on_main_thread(&self, runnable: RunnableVariant, priority: Priority) {
if self.on_main_thread() {
schedule_runnable(&self.browser_window, runnable, priority);
schedule_runnable(&browser_window(), runnable, priority);
} else {
self.main_thread_mailbox
.post(priority, MainThreadItem::Runnable(runnable));
@@ -248,7 +269,7 @@ impl PlatformDispatcher for WebDispatcher {
let callback = Closure::once_into_js(move || {
runnable.run();
});
self.browser_window
browser_window()
.set_timeout_with_callback_and_timeout_and_arguments_0(
callback.unchecked_ref(),
millis,
@@ -265,8 +286,7 @@ impl PlatformDispatcher for WebDispatcher {
let callback = Closure::once_into_js(move || {
function();
});
self.browser_window
.queue_microtask(callback.unchecked_ref());
browser_window().queue_microtask(callback.unchecked_ref());
} else {
self.main_thread_mailbox
.post(Priority::High, MainThreadItem::RealtimeFunction(function));
@@ -278,6 +298,10 @@ impl PlatformDispatcher for WebDispatcher {
}
}
fn browser_window() -> web_sys::Window {
web_sys::window().expect("must be running in a browser window context")
}
fn execute_on_main_thread(window: &web_sys::Window, item: MainThreadItem) {
match item {
MainThreadItem::Runnable(runnable) => {
@@ -294,7 +318,7 @@ fn execute_on_main_thread(window: &web_sys::Window, item: MainThreadItem) {
)
.ok();
}
MainThreadItem::RealtimeFunction(function) => {
MainThreadItem::Function(function) | MainThreadItem::RealtimeFunction(function) => {
function();
}
}
crates/gpui_web/src/http_client.rs
+131 -158
View File
@@ -1,9 +1,8 @@
use anyhow::anyhow;
use futures::AsyncReadExt as _;
use crate::WebDispatcher;
use anyhow::{Context as _, anyhow};
use futures::{AsyncReadExt as _, channel::oneshot};
use http_client::{AsyncBody, HttpClient, RedirectPolicy};
use std::future::Future;
use std::pin::Pin;
use std::task::Poll;
use std::sync::Arc;
use wasm_bindgen::JsCast as _;
use wasm_bindgen::prelude::*;
@@ -14,6 +13,7 @@ extern "C" {
}
pub struct FetchHttpClient {
dispatcher: Arc<WebDispatcher>,
user_agent: Option<http_client::http::header::HeaderValue>,
credentials: FetchCredentials,
}
@@ -31,79 +31,33 @@ pub enum FetchCredentials {
}
impl FetchHttpClient {
// Kept private so that all construction goes through the constructors
// below, which are `unsafe` when background threads are enabled.
fn build(user_agent: Option<http_client::http::header::HeaderValue>) -> Self {
pub(crate) fn new(dispatcher: Arc<WebDispatcher>) -> Self {
Self {
user_agent,
dispatcher,
user_agent: None,
credentials: FetchCredentials::default(),
}
}
pub(crate) fn with_user_agent(
dispatcher: Arc<WebDispatcher>,
user_agent: &str,
) -> anyhow::Result<Self> {
Ok(Self {
dispatcher,
user_agent: Some(http_client::http::header::HeaderValue::from_str(
user_agent,
)?),
credentials: FetchCredentials::default(),
})
}
pub fn with_credentials(mut self, credentials: FetchCredentials) -> Self {
self.credentials = credentials;
self
}
}
#[cfg(feature = "multithreaded")]
impl FetchHttpClient {
/// # Safety
///
/// The caller must ensure the client is only used from the main thread:
/// the futures returned by [`HttpClient::send`] call browser APIs that do
/// not exist on worker threads, so they must never be polled from a task
/// spawned on the background executor.
pub unsafe fn new() -> Self {
Self::build(None)
}
/// # Safety
///
/// The caller must ensure the client is only used from the main thread:
/// the futures returned by [`HttpClient::send`] call browser APIs that do
/// not exist on worker threads, so they must never be polled from a task
/// spawned on the background executor.
pub unsafe fn with_user_agent(user_agent: &str) -> anyhow::Result<Self> {
Ok(Self::build(Some(
http_client::http::header::HeaderValue::from_str(user_agent)?,
)))
}
}
#[cfg(not(feature = "multithreaded"))]
impl FetchHttpClient {
pub fn new() -> Self {
Self::build(None)
}
pub fn with_user_agent(user_agent: &str) -> anyhow::Result<Self> {
Ok(Self::build(Some(
http_client::http::header::HeaderValue::from_str(user_agent)?,
)))
}
}
/// Wraps a `!Send` future to satisfy the `Send` bound on `BoxFuture`.
///
/// Safety: only valid because the `FetchHttpClient` constructors confine the
/// client to the main thread (via their `unsafe` contract when
/// `multithreaded` is enabled, or by the absence of threads when it is not),
/// so the wrapped future is never actually sent across threads.
struct AssertSend<F>(F);
unsafe impl<F> Send for AssertSend<F> {}
impl<F: Future> Future for AssertSend<F> {
type Output = F::Output;
fn poll(self: Pin<&mut Self>, cx: &mut std::task::Context<'_>) -> Poll<Self::Output> {
// Safety: pin projection for a single-field newtype wrapper.
let inner = unsafe { self.map_unchecked_mut(|this| &mut this.0) };
inner.poll(cx)
}
}
impl HttpClient for FetchHttpClient {
fn user_agent(&self) -> Option<&http_client::http::header::HeaderValue> {
self.user_agent.as_ref()
@@ -120,105 +74,124 @@ impl HttpClient for FetchHttpClient {
{
let (parts, body) = req.into_parts();
let credentials = self.credentials;
let dispatcher = self.dispatcher.clone();
Box::pin(AssertSend(async move {
Box::pin(async move {
let body_bytes = read_body_to_bytes(body).await?;
let (sender, receiver) = oneshot::channel();
let init = web_sys::RequestInit::new();
init.set_method(parts.method.as_str());
init.set_credentials(match credentials {
FetchCredentials::Omit => web_sys::RequestCredentials::Omit,
FetchCredentials::SameOrigin => web_sys::RequestCredentials::SameOrigin,
FetchCredentials::Include => web_sys::RequestCredentials::Include,
dispatcher.dispatch_function_on_main_thread(move || {
wasm_bindgen_futures::spawn_local(async move {
let result = fetch(parts, body_bytes, credentials).await;
if sender.send(result).is_err() {
log::debug!("fetch response receiver was dropped");
}
});
});
if let Some(redirect_policy) = parts.extensions.get::<RedirectPolicy>() {
match redirect_policy {
RedirectPolicy::NoFollow => {
init.set_redirect(web_sys::RequestRedirect::Manual);
}
RedirectPolicy::FollowLimit(_) | RedirectPolicy::FollowAll => {
init.set_redirect(web_sys::RequestRedirect::Follow);
}
}
}
if let Some(ref bytes) = body_bytes {
let uint8array = js_sys::Uint8Array::from(bytes.as_slice());
init.set_body(uint8array.as_ref());
}
let url = parts.uri.to_string();
let request = web_sys::Request::new_with_str_and_init(&url, &init)
.map_err(|error| anyhow!("failed to create fetch Request: {error:?}"))?;
let request_headers = request.headers();
for (name, value) in &parts.headers {
let value_str = value
.to_str()
.map_err(|_| anyhow!("non-ASCII header value for {name}"))?;
request_headers
.set(name.as_str(), value_str)
.map_err(|error| anyhow!("failed to set header {name}: {error:?}"))?;
}
let promise = global_fetch(&request)
.map_err(|error| anyhow!("fetch threw an error: {error:?}"))?;
let response_value = wasm_bindgen_futures::JsFuture::from(promise)
.await
.map_err(|error| anyhow!("fetch failed: {error:?}"))?;
let web_response: web_sys::Response = response_value
.dyn_into()
.map_err(|error| anyhow!("fetch result is not a Response: {error:?}"))?;
let status = web_response.status();
let mut builder = http_client::http::Response::builder().status(status);
// `Headers` is a JS iterable yielding `[name, value]` pairs.
// `js_sys::Array::from` calls `Array.from()` which accepts any iterable.
let header_pairs = js_sys::Array::from(&web_response.headers());
for index in 0..header_pairs.length() {
match header_pairs.get(index).dyn_into::<js_sys::Array>() {
Ok(pair) => match (pair.get(0).as_string(), pair.get(1).as_string()) {
(Some(name), Some(value)) => {
builder = builder.header(name, value);
}
(name, value) => {
log::warn!(
"skipping response header at index {index}: \
name={name:?}, value={value:?}"
);
}
},
Err(entry) => {
log::warn!("skipping non-array header entry at index {index}: {entry:?}");
}
}
}
// The entire response body is eagerly buffered into memory via
// `arrayBuffer()`. The Fetch API does not expose a synchronous
// streaming interface; streaming would require `ReadableStream`
// interop which is significantly more complex.
let body_promise = web_response
.array_buffer()
.map_err(|error| anyhow!("failed to initiate response body read: {error:?}"))?;
let body_value = wasm_bindgen_futures::JsFuture::from(body_promise)
.await
.map_err(|error| anyhow!("failed to read response body: {error:?}"))?;
let array_buffer: js_sys::ArrayBuffer = body_value
.dyn_into()
.map_err(|error| anyhow!("response body is not an ArrayBuffer: {error:?}"))?;
let response_bytes = js_sys::Uint8Array::new(&array_buffer).to_vec();
builder
.body(AsyncBody::from(response_bytes))
.map_err(|error| anyhow!(error))
}))
receiver.await.context("browser fetch task was canceled")?
})
}
}
async fn fetch(
parts: http_client::http::request::Parts,
body_bytes: Option<Vec<u8>>,
credentials: FetchCredentials,
) -> anyhow::Result<http_client::http::Response<AsyncBody>> {
let init = web_sys::RequestInit::new();
init.set_method(parts.method.as_str());
init.set_credentials(match credentials {
FetchCredentials::Omit => web_sys::RequestCredentials::Omit,
FetchCredentials::SameOrigin => web_sys::RequestCredentials::SameOrigin,
FetchCredentials::Include => web_sys::RequestCredentials::Include,
});
if let Some(redirect_policy) = parts.extensions.get::<RedirectPolicy>() {
match redirect_policy {
RedirectPolicy::NoFollow => {
init.set_redirect(web_sys::RequestRedirect::Manual);
}
RedirectPolicy::FollowLimit(_) | RedirectPolicy::FollowAll => {
init.set_redirect(web_sys::RequestRedirect::Follow);
}
}
}
if let Some(ref bytes) = body_bytes {
let uint8array = js_sys::Uint8Array::from(bytes.as_slice());
init.set_body(uint8array.as_ref());
}
let url = parts.uri.to_string();
let request = web_sys::Request::new_with_str_and_init(&url, &init)
.map_err(|error| anyhow!("failed to create fetch Request: {error:?}"))?;
let request_headers = request.headers();
for (name, value) in &parts.headers {
let value_str = value
.to_str()
.map_err(|_| anyhow!("non-ASCII header value for {name}"))?;
request_headers
.set(name.as_str(), value_str)
.map_err(|error| anyhow!("failed to set header {name}: {error:?}"))?;
}
let promise =
global_fetch(&request).map_err(|error| anyhow!("fetch threw an error: {error:?}"))?;
let response_value = wasm_bindgen_futures::JsFuture::from(promise)
.await
.map_err(|error| anyhow!("fetch failed: {error:?}"))?;
let web_response: web_sys::Response = response_value
.dyn_into()
.map_err(|error| anyhow!("fetch result is not a Response: {error:?}"))?;
let status = web_response.status();
let mut builder = http_client::http::Response::builder().status(status);
// `Headers` is a JS iterable yielding `[name, value]` pairs.
// `js_sys::Array::from` calls `Array.from()` which accepts any iterable.
let header_pairs = js_sys::Array::from(&web_response.headers());
for index in 0..header_pairs.length() {
match header_pairs.get(index).dyn_into::<js_sys::Array>() {
Ok(pair) => match (pair.get(0).as_string(), pair.get(1).as_string()) {
(Some(name), Some(value)) => {
builder = builder.header(name, value);
}
(name, value) => {
log::warn!(
"skipping response header at index {index}: \
name={name:?}, value={value:?}"
);
}
},
Err(entry) => {
log::warn!("skipping non-array header entry at index {index}: {entry:?}");
}
}
}
// The entire response body is eagerly buffered into memory via
// `arrayBuffer()`. The Fetch API does not expose a synchronous
// streaming interface; streaming would require `ReadableStream`
// interop which is significantly more complex.
let body_promise = web_response
.array_buffer()
.map_err(|error| anyhow!("failed to initiate response body read: {error:?}"))?;
let body_value = wasm_bindgen_futures::JsFuture::from(body_promise)
.await
.map_err(|error| anyhow!("failed to read response body: {error:?}"))?;
let array_buffer: js_sys::ArrayBuffer = body_value
.dyn_into()
.map_err(|error| anyhow!("response body is not an ArrayBuffer: {error:?}"))?;
let response_bytes = js_sys::Uint8Array::new(&array_buffer).to_vec();
builder
.body(AsyncBody::from(response_bytes))
.map_err(|error| anyhow!(error))
}
async fn read_body_to_bytes(mut body: AsyncBody) -> anyhow::Result<Option<Vec<u8>>> {
let mut buffer = Vec::new();
body.read_to_end(&mut buffer).await?;
crates/gpui_web/src/platform.rs
+17 -1
View File
@@ -1,6 +1,7 @@
use crate::dispatcher::WebDispatcher;
use crate::display::WebDisplay;
use crate::events::EventListenerHandle;
use crate::http_client::FetchHttpClient;
use crate::keyboard::WebKeyboardLayout;
use crate::window::WebWindow;
use anyhow::Result;
@@ -34,6 +35,7 @@ static BUNDLED_FONTS: &[&[u8]] = &[
pub struct WebPlatform {
browser_window: web_sys::Window,
dispatcher: Arc<WebDispatcher>,
background_executor: BackgroundExecutor,
foreground_executor: ForegroundExecutor,
text_system: Arc<dyn PlatformTextSystem>,
@@ -67,7 +69,7 @@ impl WebPlatform {
allow_multi_threading,
));
let background_executor = BackgroundExecutor::new(dispatcher.clone());
let foreground_executor = ForegroundExecutor::new(dispatcher);
let foreground_executor = ForegroundExecutor::new(dispatcher.clone());
let text_system = Arc::new(gpui_wgpu::CosmicTextSystem::new_without_system_fonts(
"IBM Plex Sans",
));
@@ -92,6 +94,7 @@ impl WebPlatform {
Self {
browser_window,
dispatcher,
background_executor,
foreground_executor,
text_system,
@@ -104,6 +107,19 @@ impl WebPlatform {
_cursor_restore_listeners: cursor_restore_listeners,
}
}
/// Returns an HTTP client that runs browser Fetch operations on this platform's main thread.
pub fn fetch_http_client(&self) -> FetchHttpClient {
FetchHttpClient::new(self.dispatcher.clone())
}
/// Returns a browser Fetch HTTP client with the given reported user agent.
pub fn fetch_http_client_with_user_agent(
&self,
user_agent: &str,
) -> anyhow::Result<FetchHttpClient> {
FetchHttpClient::with_user_agent(self.dispatcher.clone(), user_agent)
}
}
impl Platform for WebPlatform {