500 Internal Server Error

"; store.data_mut().guest_profiler = Some(Arc::new(GuestProfiler::new_component( store.engine(), module_name, interval, component, std::iter::empty(), )?)); fn sample( mut store: StoreContextMut, f: impl FnOnce(&mut GuestProfiler, StoreContext), ) { let mut profiler = store.data_mut().guest_profiler.take().unwrap(); f( Arc::get_mut(&mut profiler).expect("profiling doesn't support threads yet"), store.as_context(), ); store.data_mut().guest_profiler = Some(profiler); } // Hostcall entry/exit, etc. store.call_hook(|store, kind| { sample(store, |profiler, store| profiler.call_hook(store, kind)); Ok(()) }); store.epoch_deadline_callback(move |store| { sample(store, |profiler, store| { profiler.sample(store, std::time::Duration::ZERO) }); Ok(UpdateDeadline::Continue(1)) }); store.set_epoch_deadline(1); let write_profile = Box::new(move |mut store: StoreContextMut| { let profiler = Arc::try_unwrap(store.data_mut().guest_profiler.take().unwrap()) .expect("profiling doesn't support threads yet"); if let Err(e) = std::fs::File::create(&path) .map_err(wasmtime::Error::new) .and_then(|output| profiler.finish(std::io::BufWriter::new(output))) { eprintln!("failed writing profile at {path}: {e:#}"); } else { eprintln!(); eprintln!("Profile written to: {path}"); eprintln!("View this profile at https://profiler.firefox.com/."); } }); Ok(write_profile) } /// Build a minimal error response with an empty body. fn error_response(status: StatusCode) -> hyper::Response> { Response::builder() .status(status) .body( http_body_util::Empty::new() .map_err(|_| unreachable!()) .boxed_unsync(), ) .unwrap() } /// Debuggee body for `wasmtime serve -g`: instantiate the HTTP component /// once, then handle requests sequentially on a single store. #[cfg(feature = "debug")] async fn debug_serve_body( store: &mut Store, proxy_pre: wasmtime_wasi_http::p2::bindings::ProxyPre, addr: SocketAddr, ) -> Result<()> { use hyper::server::conn::http1; use wasmtime_wasi_http::p2::bindings::http::types::Scheme; use wasmtime_wasi_http::p2::body::HyperOutgoingBody; type P2Response = std::result::Result< hyper::Response, wasmtime_wasi_http::p2::bindings::http::types::ErrorCode, >; let engine_clone = store.engine().clone(); let _epoch_thread = std::thread::spawn(move || { loop { std::thread::sleep(Duration::from_millis(1)); engine_clone.increment_epoch(); } }); store.epoch_deadline_async_yield_and_update(1); // Instantiate the HTTP component once. let proxy = proxy_pre.instantiate_async(&mut *store).await?; // Bind the TCP listener. let socket = match addr { SocketAddr::V4(_) => tokio::net::TcpSocket::new_v4()?, SocketAddr::V6(_) => tokio::net::TcpSocket::new_v6()?, }; socket.set_reuseaddr(!cfg!(windows))?; socket.bind(addr)?; let listener = socket.listen(100)?; eprintln!("Serving HTTP on http://{}/", listener.local_addr()?); // Accept loop: handle one connection at a time, requests sequentially. loop { let (stream, _) = listener.accept().await?; stream.set_nodelay(true)?; let stream = TokioIo::new(stream); // Channel to bridge hyper's service_fn with our sequential // request processing on the single store. type RespBody = hyper::Response>; let (req_tx, mut req_rx) = tokio::sync::mpsc::channel::<( hyper::Request, tokio::sync::oneshot::Sender>, )>(1); let serve_conn = http1::Builder::new().keep_alive(true).serve_connection( stream, hyper::service::service_fn(move |req| { let req_tx = req_tx.clone(); async move { let (resp_tx, resp_rx) = tokio::sync::oneshot::channel(); if req_tx.send((req, resp_tx)).await.is_err() { return Ok::<_, Infallible>(error_response( StatusCode::SERVICE_UNAVAILABLE, )); } resp_rx .await .unwrap_or(Ok(error_response(StatusCode::SERVICE_UNAVAILABLE))) } }), ); tokio::pin!(serve_conn); loop { tokio::select! { result = &mut serve_conn => { if let Err(e) = result { eprintln!("connection error: {e:?}"); } break; } msg = req_rx.recv() => { let Some((req, resp_tx)) = msg else { break }; let (p2_tx, p2_rx) = tokio::sync::oneshot::channel::(); let wasi_req = store .data_mut() .http() .new_incoming_request(Scheme::Http, req); let wasi_out = wasi_req.and_then(|_req| { let out = store.data_mut().http().new_response_outparam(p2_tx); out.map(|out| (_req, out)) }); let (wasi_req, wasi_out) = match wasi_out { Ok(pair) => pair, Err(e) => { eprintln!("error creating WASI request: {e:?}"); let _ = resp_tx.send(Ok(error_response( StatusCode::INTERNAL_SERVER_ERROR, ))); continue; } }; if let Err(e) = proxy .wasi_http_incoming_handler() .call_handle(&mut *store, wasi_req, wasi_out) .await { eprintln!("handler error: {e:?}"); } let resp = match p2_rx.await { Ok(Ok(resp)) => resp.map(|body| { body.map_err(|e| e.into()).boxed_unsync() }), Ok(Err(e)) => { eprintln!("component error: {e:?}"); error_response(StatusCode::INTERNAL_SERVER_ERROR) } Err(_) => error_response(StatusCode::INTERNAL_SERVER_ERROR), }; let _ = resp_tx.send(Ok(resp)); } } } } } type Request = hyper::Request; async fn handle_request( handler: ProxyHandler, req: Request, ) -> Result>> { use tokio::sync::oneshot; let req_id = handler.next_req_id(); log::info!( "Request {req_id} handling {} to {}", req.method(), req.uri() ); // Here we must declare different channel types for p2 and p3 since p2's // `WasiHttpView::new_response_outparam` expects a specific kind of sender // that uses `p2::http::types::ErrorCode`, and we don't want to have to // convert from the p3 `ErrorCode` to the p2 one, only to convert again to // `wasmtime::Error`. type P2Response = Result< hyper::Response, p2::http::types::ErrorCode, >; type P3Response = hyper::Response>; enum Sender { P2(oneshot::Sender), P3(oneshot::Sender), } enum Receiver { P2(oneshot::Receiver), P3(oneshot::Receiver), } let (tx, rx) = match handler.instance_pre() { ProxyPre::P2(_) => { let (tx, rx) = oneshot::channel(); (Sender::P2(tx), Receiver::P2(rx)) } ProxyPre::P3(_) => { let (tx, rx) = oneshot::channel(); (Sender::P3(tx), Receiver::P3(rx)) } }; handler.spawn( if handler.state().max_instance_reuse_count() == 1 { Some(req_id) } else { None }, Box::new(move |store, proxy| { Box::pin( async move { match proxy { Proxy::P2(proxy) => { let Sender::P2(tx) = tx else { unreachable!() }; let (req, out) = store.with(move |mut store| { let req = store .data_mut() .http() .new_incoming_request(p2::http::types::Scheme::Http, req)?; let out = store.data_mut().http().new_response_outparam(tx)?; wasmtime::error::Ok((req, out)) })?; proxy .wasi_http_incoming_handler() .call_handle(store, req, out) .await } Proxy::P3(proxy) => { use wasmtime_wasi_http::p3::bindings::http::types::{ ErrorCode, Request, }; let Sender::P3(tx) = tx else { unreachable!() }; let (req, body) = req.into_parts(); let body = body.map_err(ErrorCode::from_hyper_request_error); let req = http::Request::from_parts(req, body); let (request, request_io_result) = Request::from_http(req); let res = proxy.handle(store, request).await??; let res = store .with(|mut store| res.into_http(&mut store, request_io_result))?; // With the guest response now transformed into a // host-compatible response layer one more wrapper // around the body. This layer is solely responsible // for dropping a channel half on destruction, and // this enables waiting here until the body is // consumed by waiting for this destruction to // happen. let (resp_body_tx, resp_body_rx) = oneshot::channel(); let res = res.map(|body| { let body = body.map_err(|e| e.into()); P3BodyWrapper { _tx: resp_body_tx, body, } .boxed_unsync() }); // If `wasmtime serve` is waiting on this response // and actually got it then wait for the body to // finish, otherwise it's thrown away so skip that // step. if tx.send(res).is_ok() { _ = resp_body_rx.await; } Ok(()) } } } .map(move |result| { if let Err(error) = result { eprintln!("[{req_id}] :: {error:?}"); } }), ) }), ); return Ok(match rx { Receiver::P2(rx) => rx .await .context("guest never invoked `response-outparam::set` method")? .map_err(|e| wasmtime::Error::from(e))? .map(|body| body.map_err(|e| e.into()).boxed_unsync()), Receiver::P3(rx) => rx.await?, }); // Forwarding implementation of `Body` to an inner `B` with the sole purpose // of carrying `_tx` to its destruction. struct P3BodyWrapper { body: B, _tx: oneshot::Sender<()>, } impl Body for P3BodyWrapper { type Data = B::Data; type Error = B::Error; fn poll_frame( mut self: Pin<&mut Self>, cx: &mut Context<'_>, ) -> Poll, Self::Error>>> { Pin::new(&mut self.body).poll_frame(cx) } fn is_end_stream(&self) -> bool { self.body.is_end_stream() } fn size_hint(&self) -> SizeHint { self.body.size_hint() } } } #[derive(Clone)] enum Output { Stdout, Stderr, } impl Output { fn write_all(&self, buf: &[u8]) -> io::Result<()> { use std::io::Write; match self { Output::Stdout => std::io::stdout().write_all(buf), Output::Stderr => std::io::stderr().write_all(buf), } } } #[derive(Clone)] struct LogStream { output: Output, state: Arc, } struct LogStreamState { prefix: String, needs_prefix_on_next_write: AtomicBool, } impl LogStream { fn new(prefix: String, output: Output) -> LogStream { LogStream { output, state: Arc::new(LogStreamState { prefix, needs_prefix_on_next_write: AtomicBool::new(true), }), } } fn write_all(&mut self, mut bytes: &[u8]) -> io::Result<()> { while !bytes.is_empty() { if self .state .needs_prefix_on_next_write .load(Ordering::Relaxed) { self.output.write_all(self.state.prefix.as_bytes())?; self.state .needs_prefix_on_next_write .store(false, Ordering::Relaxed); } match bytes.iter().position(|b| *b == b'\n') { Some(i) => { let (a, b) = bytes.split_at(i + 1); bytes = b; self.output.write_all(a)?; self.state .needs_prefix_on_next_write .store(true, Ordering::Relaxed); } None => { self.output.write_all(bytes)?; break; } } } Ok(()) } } impl wasmtime_wasi::cli::StdoutStream for LogStream { fn p2_stream(&self) -> Box { Box::new(self.clone()) } fn async_stream(&self) -> Box { Box::new(self.clone()) } } impl wasmtime_wasi::cli::IsTerminal for LogStream { fn is_terminal(&self) -> bool { match &self.output { Output::Stdout => std::io::stdout().is_terminal(), Output::Stderr => std::io::stderr().is_terminal(), } } } impl wasmtime_wasi::p2::OutputStream for LogStream { fn write(&mut self, bytes: bytes::Bytes) -> StreamResult<()> { self.write_all(&bytes) .map_err(|e| StreamError::LastOperationFailed(e.into()))?; Ok(()) } fn flush(&mut self) -> StreamResult<()> { Ok(()) } fn check_write(&mut self) -> StreamResult { Ok(1024 * 1024) } } #[async_trait::async_trait] impl wasmtime_wasi::p2::Pollable for LogStream { async fn ready(&mut self) {} } impl AsyncWrite for LogStream { fn poll_write( mut self: Pin<&mut Self>, _cx: &mut Context<'_>, buf: &[u8], ) -> Poll> { Poll::Ready(self.write_all(buf).map(|_| buf.len())) } fn poll_flush(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll> { Poll::Ready(Ok(())) } fn poll_shutdown(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll> { Poll::Ready(Ok(())) } } /// The pooling allocator is tailor made for the `wasmtime serve` use case, so /// try to use it when we can. The main cost of the pooling allocator, however, /// is the virtual memory required to run it. Not all systems support the same /// amount of virtual memory, for example some aarch64 and riscv64 configuration /// only support 39 bits of virtual address space. /// /// The pooling allocator, by default, will request 1000 linear memories each /// sized at 6G per linear memory. This is 6T of virtual memory which ends up /// being about 42 bits of the address space. This exceeds the 39 bit limit of /// some systems, so there the pooling allocator will fail by default. /// /// This function attempts to dynamically determine the hint for the pooling /// allocator. This returns `Some(true)` if the pooling allocator should be used /// by default, or `None` or an error otherwise. /// /// The method for testing this is to allocate a 0-sized 64-bit linear memory /// with a maximum size that's N bits large where we force all memories to be /// static. This should attempt to acquire N bits of the virtual address space. /// If successful that should mean that the pooling allocator is OK to use, but /// if it fails then the pooling allocator is not used and the normal mmap-based /// implementation is used instead. fn use_pooling_allocator_by_default() -> Result> { use wasmtime::{Config, Memory, MemoryType}; const BITS_TO_TEST: u32 = 42; let mut config = Config::new(); config.wasm_memory64(true); config.memory_reservation(1 << BITS_TO_TEST); let engine = Engine::new(&config)?; let mut store = Store::new(&engine, ()); // NB: the maximum size is in wasm pages to take out the 16-bits of wasm // page size here from the maximum size. let ty = MemoryType::new64(0, Some(1 << (BITS_TO_TEST - 16))); if Memory::new(&mut store, ty).is_ok() { Ok(Some(true)) } else { Ok(None) } }