500 Internal Server Error

"; store.data_mut().guest_profiler = Some(Arc::new(GuestProfiler::new_component( 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(()) }); let start = Instant::now(); let timeout = cmd.run.common.wasm.timeout; store.epoch_deadline_callback(move |store| { sample(store, |profiler, store| { profiler.sample(store, std::time::Duration::ZERO) }); // Originally epoch counting was used here; this is problematic in // a lot of cases due to there being a lot of time (e.g. in hostcalls) // when we are not expected to get sample hits. if let Some(timeout) = timeout { if start.elapsed() > timeout { bail!("Timeout expired"); } } Ok(UpdateDeadline::Continue(1)) }); store.set_epoch_deadline(1); let engine = store.engine().clone(); let epoch_thread = Some(EpochThread::spawn(interval, engine)); let write_profile = Box::new(move |store: &mut Store| { 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(anyhow::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, epoch_thread)) } struct ProxyHandlerInner { cmd: ServeCommand, engine: Engine, instance_pre: ProxyPre, next_id: AtomicU64, } impl ProxyHandlerInner { fn next_req_id(&self) -> u64 { self.next_id.fetch_add(1, Ordering::Relaxed) } } #[derive(Clone)] struct ProxyHandler(Arc); impl ProxyHandler { fn new(cmd: ServeCommand, engine: Engine, instance_pre: ProxyPre) -> Self { Self(Arc::new(ProxyHandlerInner { cmd, engine, instance_pre, next_id: AtomicU64::from(0), })) } } type Request = hyper::Request; async fn handle_request( ProxyHandler(inner): ProxyHandler, req: Request, component: Component, ) -> Result> { let (sender, receiver) = tokio::sync::oneshot::channel(); let req_id = inner.next_req_id(); log::info!( "Request {req_id} handling {} to {}", req.method(), req.uri() ); let mut store = inner.cmd.new_store(&inner.engine, req_id)?; let req = store.data_mut().new_incoming_request(Scheme::Http, req)?; let out = store.data_mut().new_response_outparam(sender)?; let proxy = inner.instance_pre.instantiate_async(&mut store).await?; let comp = component.clone(); let task = tokio::task::spawn(async move { let (write_profile, epoch_thread) = setup_epoch_handler(&inner.cmd, &mut store, comp)?; if let Err(e) = proxy .wasi_http_incoming_handler() .call_handle(&mut store, req, out) .await { log::error!("[{req_id}] :: {e:?}"); return Err(e); } write_profile(&mut store); drop(epoch_thread); Ok(()) }); let result = match receiver.await { Ok(Ok(resp)) => Ok(resp), Ok(Err(e)) => Err(e.into()), Err(_) => { // An error in the receiver (`RecvError`) only indicates that the // task exited before a response was sent (i.e., the sender was // dropped); it does not describe the underlying cause of failure. // Instead we retrieve and propagate the error from inside the task // which should more clearly tell the user what went wrong. Note // that we assume the task has already exited at this point so the // `await` should resolve immediately. let e = match task.await { Ok(Ok(())) => { bail!("guest never invoked `response-outparam::set` method") } Ok(Err(e)) => e, Err(e) => e.into(), }; Err(e.context("guest never invoked `response-outparam::set` method")) } }; result } #[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) } }