wasmer_wasix/runners/wcgi/
runner.rs

1use std::{net::SocketAddr, sync::Arc};
2
3use super::super::Body;
4use anyhow::{Context, Error};
5use futures::{StreamExt, stream::FuturesUnordered};
6use http::{Request, Response};
7use tower::ServiceBuilder;
8use tower_http::{catch_panic::CatchPanicLayer, cors::CorsLayer, trace::TraceLayer};
9use wcgi_host::CgiDialect;
10use webc::metadata::{
11    Command,
12    annotations::{Wasi, Wcgi},
13};
14
15use crate::{
16    Runtime, WasiEnvBuilder,
17    bin_factory::BinaryPackage,
18    capabilities::Capabilities,
19    runners::{
20        MappedDirectory,
21        wasi_common::CommonWasiOptions,
22        wcgi::handler::{Handler, SharedState},
23    },
24    runtime::task_manager::VirtualTaskManagerExt,
25};
26
27use super::Callbacks;
28
29#[derive(Debug)]
30pub struct WcgiRunner {
31    config: Config,
32}
33
34impl WcgiRunner {
35    pub fn new<C>(callbacks: C) -> Self
36    where
37        C: Callbacks,
38    {
39        Self {
40            config: Config::new(callbacks),
41        }
42    }
43
44    pub fn config(&mut self) -> &mut Config {
45        &mut self.config
46    }
47
48    #[tracing::instrument(skip_all)]
49    pub(crate) fn prepare_handler(
50        &mut self,
51        command_name: &str,
52        pkg: &BinaryPackage,
53        propagate_stderr: bool,
54        default_dialect: CgiDialect,
55        runtime: Arc<dyn Runtime + Send + Sync>,
56    ) -> Result<Handler, Error> {
57        let cmd = pkg
58            .get_command(command_name)
59            .with_context(|| format!("The package doesn't contain a \"{command_name}\" command"))?;
60        let metadata = cmd.metadata();
61        let wasi = metadata
62            .annotation("wasi")?
63            .unwrap_or_else(|| Wasi::new(command_name));
64
65        let module = runtime.load_command_module_sync(cmd)?;
66
67        let Wcgi { dialect, .. } = metadata.annotation("wcgi")?.unwrap_or_default();
68        let dialect = match dialect {
69            Some(d) => d.parse().context("Unable to parse the CGI dialect")?,
70            None => default_dialect,
71        };
72
73        let container_fs = Arc::clone(&pkg.webc_fs);
74
75        let wasi_common = self.config.wasi.clone();
76        let rt = Arc::clone(&runtime);
77        let setup_builder = move |builder: &mut WasiEnvBuilder| {
78            wasi_common.prepare_webc_env(builder, Some(Arc::clone(&container_fs)), &wasi, None)?;
79            builder.set_runtime(Arc::clone(&rt));
80            Ok(())
81        };
82
83        let shared = SharedState {
84            module,
85            module_hash: pkg.hash(),
86            dialect,
87            propagate_stderr,
88            program_name: command_name.to_string(),
89            setup_builder: Arc::new(setup_builder),
90            callbacks: Arc::clone(&self.config.callbacks),
91            runtime,
92        };
93
94        Ok(Handler::new(Arc::new(shared)))
95    }
96
97    pub(crate) fn run_command_with_handler<S>(
98        &mut self,
99        handler: S,
100        runtime: Arc<dyn Runtime + Send + Sync>,
101    ) -> Result<(), Error>
102    where
103        S: tower::Service<
104                Request<hyper::body::Incoming>,
105                Response = http::Response<Body>,
106                Error = anyhow::Error,
107                Future = std::pin::Pin<
108                    Box<dyn futures::Future<Output = Result<Response<Body>, Error>> + Send>,
109                >,
110            >,
111        S: Clone + Send + Sync + 'static,
112    {
113        let service = ServiceBuilder::new()
114            .layer(
115                TraceLayer::new_for_http()
116                    .make_span_with(|request: &Request<hyper::body::Incoming>| {
117                        tracing::info_span!(
118                            "request",
119                            method = %request.method(),
120                            uri = %request.uri(),
121                            status_code = tracing::field::Empty,
122                        )
123                    })
124                    .on_response(super::super::response_tracing::OnResponseTracer),
125            )
126            .layer(CatchPanicLayer::new())
127            .layer(CorsLayer::permissive())
128            .service(handler);
129
130        let address = self.config.addr;
131        tracing::info!(%address, "Starting the server");
132
133        let callbacks = Arc::clone(&self.config.callbacks);
134        runtime.task_manager().spawn_and_block_on(async move {
135            let (mut shutdown, abort_handle) =
136                futures::future::abortable(futures::future::pending::<()>());
137
138            callbacks.started(abort_handle);
139
140            let listener = tokio::net::TcpListener::bind(&address).await?;
141            let graceful = hyper_util::server::graceful::GracefulShutdown::new();
142
143            let http = hyper::server::conn::http1::Builder::new();
144
145            let mut futs = FuturesUnordered::new();
146
147            loop {
148                tokio::select! {
149                    Ok((stream, _addr)) = listener.accept() => {
150                        let io = hyper_util::rt::tokio::TokioIo::new(stream);
151                        let service = hyper_util::service::TowerToHyperService::new(service.clone());
152                        let conn = http.serve_connection(io, service);
153                        // watch this connection
154                        let fut = graceful.watch(conn);
155                        futs.push(async move {
156                            if let Err(e) = fut.await {
157                                eprintln!("Error serving connection: {e:?}");
158                            }
159                        });
160                    },
161
162                    _ = futs.next() => {}
163
164                    _ = &mut shutdown => {
165                        eprintln!("graceful shutdown signal received");
166                        // stop the accept loop
167                        break;
168                    }
169                }
170            }
171
172            Ok::<_, anyhow::Error>(())
173        })??;
174
175        Ok(())
176    }
177}
178
179impl crate::runners::Runner for WcgiRunner {
180    fn can_run_command(command: &Command) -> Result<bool, Error> {
181        Ok(command
182            .runner
183            .starts_with(webc::metadata::annotations::WCGI_RUNNER_URI))
184    }
185
186    fn run_command(
187        &mut self,
188        command_name: &str,
189        pkg: &BinaryPackage,
190        runtime: Arc<dyn Runtime + Send + Sync>,
191    ) -> Result<(), Error> {
192        let handler = self.prepare_handler(
193            command_name,
194            pkg,
195            false,
196            CgiDialect::Rfc3875,
197            Arc::clone(&runtime),
198        )?;
199        self.run_command_with_handler(handler, runtime)
200    }
201}
202
203#[derive(Debug)]
204pub struct Config {
205    pub(crate) wasi: CommonWasiOptions,
206    pub(crate) addr: SocketAddr,
207    pub(crate) callbacks: Arc<dyn Callbacks>,
208}
209
210impl Config {
211    pub fn addr(&mut self, addr: SocketAddr) -> &mut Self {
212        self.addr = addr;
213        self
214    }
215
216    /// Add an argument to the WASI executable's command-line arguments.
217    pub fn arg(&mut self, arg: impl Into<String>) -> &mut Self {
218        self.wasi.args.push(arg.into());
219        self
220    }
221
222    /// Add multiple arguments to the WASI executable's command-line arguments.
223    pub fn args<A, S>(&mut self, args: A) -> &mut Self
224    where
225        A: IntoIterator<Item = S>,
226        S: Into<String>,
227    {
228        self.wasi.args.extend(args.into_iter().map(|s| s.into()));
229        self
230    }
231
232    /// Expose an environment variable to the guest.
233    pub fn env(&mut self, name: impl Into<String>, value: impl Into<String>) -> &mut Self {
234        self.wasi.env.insert(name.into(), value.into());
235        self
236    }
237
238    /// Expose multiple environment variables to the guest.
239    pub fn envs<I, K, V>(&mut self, variables: I) -> &mut Self
240    where
241        I: IntoIterator<Item = (K, V)>,
242        K: Into<String>,
243        V: Into<String>,
244    {
245        self.wasi
246            .env
247            .extend(variables.into_iter().map(|(k, v)| (k.into(), v.into())));
248        self
249    }
250
251    /// Forward all of the host's environment variables to the guest.
252    pub fn forward_host_env(&mut self) -> &mut Self {
253        self.wasi.forward_host_env = true;
254        self
255    }
256
257    pub fn map_directory(&mut self, dir: MappedDirectory) -> &mut Self {
258        self.wasi.mounts.push(dir.into());
259        self
260    }
261
262    pub fn map_directories(
263        &mut self,
264        mappings: impl IntoIterator<Item = MappedDirectory>,
265    ) -> &mut Self {
266        for mapping in mappings {
267            self.map_directory(mapping);
268        }
269        self
270    }
271
272    /// Set callbacks that will be triggered at various points in the runner's
273    /// lifecycle.
274    pub fn callbacks(&mut self, callbacks: impl Callbacks + 'static) -> &mut Self {
275        self.callbacks = Arc::new(callbacks);
276        self
277    }
278
279    /// Add a package that should be available to the instance at runtime.
280    pub fn inject_package(&mut self, pkg: BinaryPackage) -> &mut Self {
281        self.wasi.injected_packages.push(pkg);
282        self
283    }
284
285    /// Add packages that should be available to the instance at runtime.
286    pub fn inject_packages(
287        &mut self,
288        packages: impl IntoIterator<Item = BinaryPackage>,
289    ) -> &mut Self {
290        self.wasi.injected_packages.extend(packages);
291        self
292    }
293
294    pub fn capabilities(&mut self) -> &mut Capabilities {
295        &mut self.wasi.capabilities
296    }
297
298    #[cfg(feature = "journal")]
299    pub fn add_snapshot_trigger(&mut self, on: crate::journal::SnapshotTrigger) {
300        self.wasi.snapshot_on.push(on);
301    }
302
303    #[cfg(feature = "journal")]
304    pub fn add_default_snapshot_triggers(&mut self) -> &mut Self {
305        for on in crate::journal::DEFAULT_SNAPSHOT_TRIGGERS {
306            if !self.has_snapshot_trigger(on) {
307                self.add_snapshot_trigger(on);
308            }
309        }
310        self
311    }
312
313    #[cfg(feature = "journal")]
314    pub fn has_snapshot_trigger(&self, on: crate::journal::SnapshotTrigger) -> bool {
315        self.wasi.snapshot_on.contains(&on)
316    }
317
318    #[cfg(feature = "journal")]
319    pub fn with_snapshot_interval(&mut self, period: std::time::Duration) -> &mut Self {
320        if !self.has_snapshot_trigger(crate::journal::SnapshotTrigger::PeriodicInterval) {
321            self.add_snapshot_trigger(crate::journal::SnapshotTrigger::PeriodicInterval);
322        }
323        self.wasi.snapshot_interval.replace(period);
324        self
325    }
326
327    #[cfg(feature = "journal")]
328    pub fn with_stop_running_after_snapshot(&mut self, stop_running: bool) {
329        self.wasi.stop_running_after_snapshot = stop_running;
330    }
331
332    #[cfg(feature = "journal")]
333    pub fn add_read_only_journal(
334        &mut self,
335        journal: Arc<crate::journal::DynReadableJournal>,
336    ) -> &mut Self {
337        self.wasi.read_only_journals.push(journal);
338        self
339    }
340
341    #[cfg(feature = "journal")]
342    pub fn add_writable_journal(&mut self, journal: Arc<crate::journal::DynJournal>) -> &mut Self {
343        self.wasi.writable_journals.push(journal);
344        self
345    }
346}
347
348impl Config {
349    pub fn new<C>(callbacks: C) -> Self
350    where
351        C: Callbacks,
352    {
353        Self {
354            addr: ([127, 0, 0, 1], 8000).into(),
355            wasi: CommonWasiOptions::default(),
356            callbacks: Arc::new(callbacks),
357        }
358    }
359}
360
361#[cfg(test)]
362mod tests {
363    use super::*;
364
365    #[test]
366    fn send_and_sync() {
367        fn assert_send<T: Send>() {}
368        fn assert_sync<T: Sync>() {}
369
370        assert_send::<WcgiRunner>();
371        assert_sync::<WcgiRunner>();
372    }
373}
wasmer_wasix/runners/wcgi/runner.rs

wasmer_wasix/runners/wcgi/
runner.rs
