diff --git a/crates/gpui/examples/image/image.rs b/crates/gpui/examples/image/image.rs index 688f881bc6..2a885714c8 100644 --- a/crates/gpui/examples/image/image.rs +++ b/crates/gpui/examples/image/image.rs @@ -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()); diff --git a/crates/gpui/examples/image_gallery.rs b/crates/gpui/examples/image_gallery.rs index bc5cda396c..b4c2887755 100644 --- a/crates/gpui/examples/image_gallery.rs +++ b/crates/gpui/examples/image_gallery.rs @@ -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()); diff --git a/crates/gpui_platform/src/gpui_platform.rs b/crates/gpui_platform/src/gpui_platform.rs index 1d2fea90b4..09b66bb8fc 100644 --- a/crates/gpui_platform/src/gpui_platform.rs +++ b/crates/gpui_platform/src/gpui_platform.rs @@ -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. diff --git a/crates/gpui_web/src/dispatcher.rs b/crates/gpui_web/src/dispatcher.rs index 58b921dfbc..a00c81bf5f 100644 --- a/crates/gpui_web/src/dispatcher.rs +++ b/crates/gpui_web/src/dispatcher.rs @@ -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), // TODO-Wasm: Shouldn't these run on their own dedicated thread? RealtimeFunction(Box), } @@ -127,7 +140,6 @@ impl MainThreadMailbox { pub struct WebDispatcher { main_thread_id: std::thread::ThreadId, - browser_window: web_sys::Window, background_sender: PriorityQueueSender, main_thread_mailbox: Arc, supports_threads: bool, @@ -135,11 +147,6 @@ pub struct WebDispatcher { _background_threads: Vec>, } -// 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(); } } diff --git a/crates/gpui_web/src/http_client.rs b/crates/gpui_web/src/http_client.rs index f48746d088..6e585682a0 100644 --- a/crates/gpui_web/src/http_client.rs +++ b/crates/gpui_web/src/http_client.rs @@ -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, user_agent: Option, 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) -> Self { + pub(crate) fn new(dispatcher: Arc) -> Self { Self { - user_agent, + dispatcher, + user_agent: None, credentials: FetchCredentials::default(), } } + pub(crate) fn with_user_agent( + dispatcher: Arc, + user_agent: &str, + ) -> anyhow::Result { + 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 { - 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 { - 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); - -unsafe impl Send for AssertSend {} - -impl Future for AssertSend { - type Output = F::Output; - - fn poll(self: Pin<&mut Self>, cx: &mut std::task::Context<'_>) -> Poll { - // 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::() { - 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::() { - 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>, + credentials: FetchCredentials, +) -> anyhow::Result> { + 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::() { + 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::() { + 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>> { let mut buffer = Vec::new(); body.read_to_end(&mut buffer).await?; diff --git a/crates/gpui_web/src/platform.rs b/crates/gpui_web/src/platform.rs index bec68aaf6f..33af331e30 100644 --- a/crates/gpui_web/src/platform.rs +++ b/crates/gpui_web/src/platform.rs @@ -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, background_executor: BackgroundExecutor, foreground_executor: ForegroundExecutor, text_system: Arc, @@ -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::with_user_agent(self.dispatcher.clone(), user_agent) + } } impl Platform for WebPlatform {