| 1 | |
|
| 2 | |
|
| 3 | |
|
| 4 | |
|
| 5 | |
|
| 6 | |
|
| 7 | |
|
| 8 | |
|
| 9 | |
|
| 10 | |
|
| 11 | |
|
| 12 | |
|
| 13 | | use crate::microvm::{Outcome, Sink}; |
| 14 | | use crate::syscall::{lines, self_hash, FAILED}; |
| 15 | | use anyhow::{bail, Context, Result}; |
| 16 | | use flow_core::api::Record; |
| 17 | | use flow_core::{Environment, Job, MountSource, Sandbox}; |
| 18 | | use serde::{Deserialize, Serialize}; |
| 19 | | use std::path::{Path, PathBuf}; |
| 20 | | use std::sync::mpsc; |
| 21 | | use std::time::{Duration, Instant}; |
| 22 | |
|
| 23 | |
|
| 24 | | pub const WASMTIME: &str = "49.0.4"; |
| 25 | |
|
| 26 | |
|
| 27 | |
|
| 28 | |
|
| 29 | | pub const AVAILABLE: bool = cfg!(feature = "wasm"); |
| 30 | |
|
| 31 | |
|
| 32 | | const WITHOUT: &str = "this flow was built without the WASM tier (the `wasm` feature): it runs on systems wasmtime supports"; |
| 33 | |
|
| 34 | |
|
| 35 | |
|
| 36 | | const NOTE: &str = "\u{1e}N "; |
| 37 | |
|
| 38 | |
|
| 39 | | #[cfg(feature = "wasm")] |
| 40 | | const TRAPPED: i32 = 134; |
| 41 | |
|
| 42 | |
|
| 43 | | #[derive(Serialize, Deserialize)] |
| 44 | | struct Spec { |
| 45 | | |
| 46 | | workspace: PathBuf, |
| 47 | | |
| 48 | | modules: Vec<String>, |
| 49 | | |
| 50 | | args: Vec<String>, |
| 51 | | env: Vec<(String, String)>, |
| 52 | | memory_mib: u32, |
| 53 | | |
| 54 | | preopens: Vec<(PathBuf, String)>, |
| 55 | | } |
| 56 | |
|
| 57 | |
|
| 58 | |
|
| 59 | |
|
| 60 | 12× | pub fn environment(sandbox: &Sandbox, job: &Job, modules: &[(String, String)]) -> Environment { |
| 61 | | Environment { |
| 62 | 12× | runtime: "wasm".into(), |
| 63 | 12× | parts: vec![ |
| 64 | 12× | ("wasmtime".into(), WASMTIME.into()), |
| 65 | 12× | ("flow".into(), self_hash().to_string()), |
| 66 | 12× | ("memory_mib".into(), sandbox.memory_mib.to_string()), |
| 67 | | ] |
| 68 | 12× | .into_iter() |
| 69 | 14× | .chain(modules.iter().map(|(path, hash)| (format!("module {path}"), hash.clone()))) |
| 70 | 12× | .collect(), |
| 71 | 12× | rootfs: String::new(), |
| 72 | 12× | argv: job.exec.clone(), |
| 73 | 24× | env: job.env.iter().map(|(k, v)| (k.clone(), v.clone())).collect(), |
| 74 | | } |
| 75 | 12× | } |
| 76 | |
|
| 77 | |
|
| 78 | |
|
| 79 | |
|
| 80 | 18× | pub fn modules(workspace: &Path, program: &str) -> Result<Vec<String>> { |
| 81 | 18× | let rel = program.strip_prefix("/src/").unwrap_or(program).trim_start_matches("./"); |
| 82 | 46× | if rel.is_empty() || rel.starts_with('/') || rel.split('/').any(|p| p == "..") { |
| 83 | 2× | bail!("module {program:?}: give a path inside the job's files, such as target/wasm32-wasip1/debug/app.wasm"); |
| 84 | 16× | } |
| 85 | 16× | let (dir, last) = rel.rsplit_once('/').unwrap_or(("", rel)); |
| 86 | 16× | if dir.contains('*') { |
| 87 | 1× | bail!("module {program:?}: only the file name may hold `*`"); |
| 88 | 15× | } |
| 89 | 15× | if !last.contains('*') { |
| 90 | 11× | if !workspace.join(rel).is_file() { |
| 91 | 0× | bail!("module {rel} is not among the job's files (its scope or an earlier job's artifact)"); |
| 92 | 11× | } |
| 93 | 11× | return Ok(vec![rel.to_string()]); |
| 94 | 4× | } |
| 95 | 4× | let mut found: Vec<String> = std::fs::read_dir(workspace.join(dir)) |
| 96 | 4× | .with_context(|| format!("module {program:?}: the directory {dir} is not among the job's files"))? |
| 97 | 10× | .filter_map(|e| e.ok()) |
| 98 | 10× | .filter(|e| e.file_type().is_ok_and(|t| t.is_file())) |
| 99 | 10× | .filter_map(|e| e.file_name().into_string().ok()) |
| 100 | 10× | .filter(|name| wildcard(last, name)) |
| 101 | 6× | .map(|name| if dir.is_empty() { name } else { format!("{dir}/{name}") }) |
| 102 | 4× | .collect(); |
| 103 | 4× | found.sort(); |
| 104 | 4× | if found.is_empty() { |
| 105 | 1× | bail!("no module matches {program:?}"); |
| 106 | 3× | } |
| 107 | 3× | Ok(found) |
| 108 | 18× | } |
| 109 | |
|
| 110 | |
|
| 111 | 17× | fn wildcard(pattern: &str, name: &str) -> bool { |
| 112 | 17× | let parts: Vec<&str> = pattern.split('*').collect(); |
| 113 | 17× | let (first, last) = (parts[0], parts[parts.len() - 1]); |
| 114 | 17× | if parts.len() == 1 { |
| 115 | 2× | return pattern == name; |
| 116 | 15× | } |
| 117 | 15× | if !name.starts_with(first) || !name[first.len()..].ends_with(last) || name.len() < first.len() + last.len() { |
| 118 | 6× | return false; |
| 119 | 9× | } |
| 120 | 9× | let mut rest = &name[first.len()..name.len() - last.len()]; |
| 121 | 9× | for middle in &parts[1..parts.len() - 1] { |
| 122 | 1× | match rest.find(middle) { |
| 123 | 1× | Some(i) => rest = &rest[i + middle.len()..], |
| 124 | 0× | None => return false, |
| 125 | | } |
| 126 | | } |
| 127 | 9× | true |
| 128 | 17× | } |
| 129 | |
|
| 130 | |
|
| 131 | |
|
| 132 | |
|
| 133 | 12× | pub fn run(sandbox: &Sandbox, job: &Job, workspace: &Path, scratch: &Path, timeout: Duration, sink: &mut dyn Sink) -> Result<Outcome> { |
| 134 | 12× | if !AVAILABLE { |
| 135 | 0× | bail!("job {}: {WITHOUT}", job.name); |
| 136 | 12× | } |
| 137 | 12× | if !job.secrets.is_empty() || !job.identities.is_empty() { |
| 138 | 0× | bail!("job {}: the WASM tier takes no secrets or identities yet; run it in a microvm or syscall sandbox", job.name); |
| 139 | 12× | } |
| 140 | 12× | let Some(program) = job.exec.first() else { bail!("job {}: exec names no module", job.name) }; |
| 141 | 12× | let modules = modules(workspace, program)?; |
| 142 | 12× | let hashed: Vec<(String, String)> = |
| 143 | 14× | modules.iter().map(|m| Ok((m.clone(), crate::home::hash_file(&workspace.join(m))?))).collect::<Result<_>>()?; |
| 144 | 12× | let environment = environment(sandbox, job, &hashed); |
| 145 | 12× | let mut preopens = Vec::new(); |
| 146 | 12× | let mut disks = 0; |
| 147 | 12× | for (i, m) in job.mounts.iter().enumerate() { |
| 148 | 0× | let dir = match m.source { |
| 149 | | MountSource::Disk => { |
| 150 | 0× | disks += 1; |
| 151 | 0× | crate::syscall::disk_dir(scratch, disks - 1) |
| 152 | | } |
| 153 | 0× | MountSource::Tmpfs => scratch.join(format!("tmpfs-{i}")), |
| 154 | | }; |
| 155 | 0× | std::fs::create_dir_all(&dir)?; |
| 156 | 0× | preopens.push((dir, m.target.clone())); |
| 157 | | } |
| 158 | 12× | let spec = Spec { |
| 159 | 12× | workspace: workspace.to_path_buf(), |
| 160 | 12× | modules, |
| 161 | 12× | args: job.exec[1..].to_vec(), |
| 162 | 24× | env: job.env.iter().map(|(k, v)| (k.clone(), v.clone())).collect(), |
| 163 | 12× | memory_mib: sandbox.memory_mib, |
| 164 | 12× | preopens, |
| 165 | | }; |
| 166 | 12× | if in_process() { |
| 167 | 6× | return run_in_thread(&spec, environment, timeout, sink); |
| 168 | 6× | } |
| 169 | 6× | let spec_path = scratch.join("wasm.json"); |
| 170 | 6× | std::fs::write(&spec_path, serde_json::to_vec(&spec)?)?; |
| 171 | |
|
| 172 | 6× | let started = Instant::now(); |
| 173 | 6× | let mut child = std::process::Command::new(crate::tier::self_exe()) |
| 174 | 6× | .arg("__wasm") |
| 175 | 6× | .arg(&spec_path) |
| 176 | 6× | .env_clear() |
| 177 | 6× | .current_dir(scratch) |
| 178 | 6× | .stdin(std::process::Stdio::null()) |
| 179 | 6× | .stdout(std::process::Stdio::piped()) |
| 180 | 6× | .stderr(std::process::Stdio::piped()) |
| 181 | 6× | .spawn() |
| 182 | 6× | .context("start the WASM helper")?; |
| 183 | 6× | let (tx, rx) = mpsc::channel::<(bool, Vec<u8>)>(); |
| 184 | 6× | let out = child.stdout.take().unwrap(); |
| 185 | 6× | let err = child.stderr.take().unwrap(); |
| 186 | 6× | let tx2 = tx.clone(); |
| 187 | 6× | let readers = [std::thread::spawn(move || lines(out, true, tx)), std::thread::spawn(move || lines(err, false, tx2))]; |
| 188 | 15× | let t = |s: &Instant| s.elapsed().as_millis() as u64; |
| 189 | 6× | let (mut timed_out, mut cancelled, mut failed_to_start) = (false, false, false); |
| 190 | 6× | let mut peak_memory_mib = None; |
| 191 | 6× | let deadline = started + timeout; |
| 192 | | loop { |
| 193 | 15× | if let Some(m) = crate::facts::peak_rss_mib(child.id()) { |
| 194 | 14× | peak_memory_mib = Some(m); |
| 195 | 14× | } |
| 196 | 15× | match rx.recv_timeout(Duration::from_millis(200)) { |
| 197 | 9× | Ok((stdout, line)) => { |
| 198 | 9× | let text = String::from_utf8_lossy(&line).into_owned(); |
| 199 | 9× | let rec = |stream: &str, text: &str| Record { t: t(&started), stream: stream.into(), text: text.into() }; |
| 200 | 9× | if let Some(note) = text.strip_prefix(NOTE).filter(|_| !stdout) { |
| 201 | 5× | sink.record(rec("flow", note)); |
| 202 | 5× | } else if let Some(why) = text.strip_prefix(FAILED).filter(|_| !stdout) { |
| 203 | 0× | failed_to_start = true; |
| 204 | 0× | sink.record(rec("flow", &format!("the WASM tier could not run the job: {why}"))); |
| 205 | 0× | } else { |
| 206 | 4× | sink.record(rec(if stdout { "out" } else { "err" }, &text)); |
| 207 | | } |
| 208 | | } |
| 209 | 0× | Err(mpsc::RecvTimeoutError::Timeout) => {} |
| 210 | 6× | Err(mpsc::RecvTimeoutError::Disconnected) => break, |
| 211 | | } |
| 212 | 9× | if !timed_out && !cancelled && sink.cancelled() { |
| 213 | 0× | cancelled = true; |
| 214 | 0× | let _ = child.kill(); |
| 215 | 0× | sink.record(Record { t: t(&started), stream: "flow".into(), text: "stopped: the run was cancelled".into() }); |
| 216 | 9× | } |
| 217 | 9× | if Instant::now() >= deadline && !timed_out && !cancelled { |
| 218 | 0× | timed_out = true; |
| 219 | 0× | let _ = child.kill(); |
| 220 | 0× | sink.record(Record { |
| 221 | 0× | t: t(&started), |
| 222 | 0× | stream: "flow".into(), |
| 223 | 0× | text: format!("stopped at the time limit of {} s", timeout.as_secs()), |
| 224 | 0× | }); |
| 225 | 9× | } |
| 226 | | } |
| 227 | 6× | let status = child.wait()?; |
| 228 | 12× | for r in readers { |
| 229 | 12× | let _ = r.join(); |
| 230 | 12× | } |
| 231 | 6× | let _ = std::fs::remove_file(&spec_path); |
| 232 | 6× | let exit_code = status.code(); |
| 233 | 6× | if exit_code.is_none() && !timed_out && !cancelled && !failed_to_start { |
| 234 | 0× | sink.record(Record { |
| 235 | 0× | t: t(&started), |
| 236 | 0× | stream: "flow".into(), |
| 237 | 0× | text: format!("the WASM helper ended without an exit code ({status})"), |
| 238 | 0× | }); |
| 239 | 6× | } |
| 240 | | Ok(Outcome { |
| 241 | 6× | exit_code: if timed_out || cancelled { None } else { exit_code }, |
| 242 | 6× | duration_ms: t(&started), |
| 243 | 6× | environment, |
| 244 | 6× | timed_out, |
| 245 | 6× | cancelled, |
| 246 | 6× | peak_memory_mib, |
| 247 | 6× | pruned: Vec::new(), |
| 248 | | }) |
| 249 | 12× | } |
| 250 | |
|
| 251 | |
|
| 252 | |
|
| 253 | |
|
| 254 | |
|
| 255 | 12× | pub fn in_process() -> bool { |
| 256 | 12× | cfg!(target_os = "android") || std::env::var_os("FLOW_WASM_IN_PROCESS").is_some() |
| 257 | 12× | } |
| 258 | |
|
| 259 | | #[cfg(feature = "wasm")] |
| 260 | |
|
| 261 | | const PIPE_BYTES: usize = 64 << 20; |
| 262 | |
|
| 263 | | #[cfg(feature = "wasm")] |
| 264 | |
|
| 265 | |
|
| 266 | |
|
| 267 | 9× | fn run_in_thread(spec: &Spec, environment: Environment, timeout: Duration, sink: &mut dyn Sink) -> Result<Outcome> { |
| 268 | | use wasmtime_wasi::p2::pipe::MemoryOutputPipe; |
| 269 | 9× | let mut config = wasmtime::Config::new(); |
| 270 | 9× | config.epoch_interruption(true); |
| 271 | 9× | let engine = ww(wasmtime::Engine::new(&config), || "start wasmtime".into())?; |
| 272 | 9× | let (stdout, stderr) = (MemoryOutputPipe::new(PIPE_BYTES), MemoryOutputPipe::new(PIPE_BYTES)); |
| 273 | 9× | let (notes_tx, notes) = mpsc::channel::<String>(); |
| 274 | 9× | let io = Io::Pipes { stdout: stdout.clone(), stderr: stderr.clone(), notes: notes_tx }; |
| 275 | 9× | let started = Instant::now(); |
| 276 | 23× | let t = |s: &Instant| s.elapsed().as_millis() as u64; |
| 277 | 9× | let deadline = started + timeout; |
| 278 | 9× | let (mut timed_out, mut cancelled) = (false, false); |
| 279 | | |
| 280 | 9× | let mut said = [(0usize, Vec::<u8>::new()), (0usize, Vec::<u8>::new())]; |
| 281 | 22× | let mut pass_on = |sink: &mut dyn Sink, last: bool| { |
| 282 | 44× | for (i, pipe) in [&stdout, &stderr].into_iter().enumerate() { |
| 283 | 44× | let all = pipe.contents(); |
| 284 | 44× | let (offset, partial) = &mut said[i]; |
| 285 | 44× | partial.extend_from_slice(&all[*offset..]); |
| 286 | 44× | *offset = all.len(); |
| 287 | 59× | while let Some(n) = partial.iter().position(|b| *b == b'\n') { |
| 288 | 4× | let line: Vec<u8> = partial.drain(..=n).collect(); |
| 289 | 4× | let text = String::from_utf8_lossy(&line[..line.len() - 1]).into_owned(); |
| 290 | 4× | sink.record(Record { t: t(&started), stream: if i == 0 { "out" } else { "err" }.into(), text }); |
| 291 | | } |
| 292 | 44× | if last && !partial.is_empty() { |
| 293 | 1× | let text = String::from_utf8_lossy(partial).into_owned(); |
| 294 | 1× | partial.clear(); |
| 295 | 1× | sink.record(Record { t: t(&started), stream: if i == 0 { "out" } else { "err" }.into(), text }); |
| 296 | 43× | } |
| 297 | | } |
| 298 | 29× | while let Ok(note) = notes.try_recv() { |
| 299 | 7× | sink.record(Record { t: t(&started), stream: "flow".into(), text: note }); |
| 300 | 7× | } |
| 301 | 22× | }; |
| 302 | 9× | let ran = std::thread::scope(|scope| { |
| 303 | 9× | let worker = scope.spawn(|| run_modules(&engine, spec, &io)); |
| 304 | 22× | while !worker.is_finished() { |
| 305 | 13× | std::thread::sleep(Duration::from_millis(100)); |
| 306 | 13× | pass_on(sink, false); |
| 307 | | |
| 308 | | |
| 309 | | |
| 310 | | |
| 311 | 13× | if timed_out || cancelled { |
| 312 | 2× | engine.increment_epoch(); |
| 313 | 11× | } |
| 314 | 13× | if !timed_out && !cancelled && sink.cancelled() { |
| 315 | 1× | cancelled = true; |
| 316 | 1× | engine.increment_epoch(); |
| 317 | 1× | sink.record(Record { t: t(&started), stream: "flow".into(), text: "stopped: the run was cancelled".into() }); |
| 318 | 12× | } |
| 319 | 13× | if Instant::now() >= deadline && !timed_out && !cancelled { |
| 320 | 1× | timed_out = true; |
| 321 | 1× | engine.increment_epoch(); |
| 322 | 1× | sink.record(Record { |
| 323 | 1× | t: t(&started), |
| 324 | 1× | stream: "flow".into(), |
| 325 | 1× | text: format!("stopped at the time limit of {} s", timeout.as_secs()), |
| 326 | 1× | }); |
| 327 | 12× | } |
| 328 | | } |
| 329 | 9× | worker.join().unwrap_or_else(|_| Err(anyhow::anyhow!("the WASM worker panicked"))) |
| 330 | 9× | }); |
| 331 | 9× | pass_on(sink, true); |
| 332 | 9× | let (exit_code, peak) = match ran { |
| 333 | 9× | Ok((code, peak)) => (Some(code), Some(peak)), |
| 334 | 0× | Err(e) => { |
| 335 | 0× | sink.record(Record { t: t(&started), stream: "flow".into(), text: format!("the WASM tier could not run the job: {e:#}") }); |
| 336 | 0× | (Some(125), None) |
| 337 | | } |
| 338 | | }; |
| 339 | | Ok(Outcome { |
| 340 | 9× | exit_code: if timed_out || cancelled { None } else { exit_code }, |
| 341 | 9× | duration_ms: t(&started), |
| 342 | 9× | environment, |
| 343 | 9× | timed_out, |
| 344 | 9× | cancelled, |
| 345 | 9× | peak_memory_mib: peak.map(|b| (b as u64).div_ceil(1 << 20)), |
| 346 | 9× | pruned: Vec::new(), |
| 347 | | }) |
| 348 | 9× | } |
| 349 | |
|
| 350 | | #[cfg(feature = "wasm")] |
| 351 | |
|
| 352 | |
|
| 353 | |
|
| 354 | 0× | pub fn wasm_main(spec_path: &Path) -> i32 { |
| 355 | 0× | let ran = (|| { |
| 356 | 0× | let spec: Spec = serde_json::from_slice(&std::fs::read(spec_path)?).context("read the spec")?; |
| 357 | 0× | let engine = ww(wasmtime::Engine::new(&wasmtime::Config::new()), || "start wasmtime".into())?; |
| 358 | 0× | run_modules(&engine, &spec, &Io::Inherit) |
| 359 | | })(); |
| 360 | 0× | match ran { |
| 361 | 0× | Ok((code, _)) => code, |
| 362 | 0× | Err(e) => { |
| 363 | 0× | eprintln!("{FAILED}{e:#}"); |
| 364 | 0× | 125 |
| 365 | | } |
| 366 | | } |
| 367 | 0× | } |
| 368 | |
|
| 369 | |
|
| 370 | | #[cfg(not(feature = "wasm"))] |
| 371 | | pub fn wasm_main(_spec_path: &Path) -> i32 { |
| 372 | | eprintln!("{FAILED}{WITHOUT}"); |
| 373 | | 125 |
| 374 | | } |
| 375 | |
|
| 376 | | #[cfg(not(feature = "wasm"))] |
| 377 | | fn run_in_thread(_spec: &Spec, _environment: Environment, _timeout: Duration, _sink: &mut dyn Sink) -> Result<Outcome> { |
| 378 | | bail!(WITHOUT) |
| 379 | | } |
| 380 | |
|
| 381 | | #[cfg(feature = "wasm")] |
| 382 | |
|
| 383 | 68× | fn ww<T>(r: wasmtime::Result<T>, what: impl FnOnce() -> String) -> Result<T> { |
| 384 | 68× | r.map_err(|e| anyhow::anyhow!("{}: {e:#}", what())) |
| 385 | 68× | } |
| 386 | |
|
| 387 | | #[cfg(feature = "wasm")] |
| 388 | |
|
| 389 | |
|
| 390 | | enum Io { |
| 391 | | Inherit, |
| 392 | | Pipes { |
| 393 | | stdout: wasmtime_wasi::p2::pipe::MemoryOutputPipe, |
| 394 | | stderr: wasmtime_wasi::p2::pipe::MemoryOutputPipe, |
| 395 | | notes: mpsc::Sender<String>, |
| 396 | | }, |
| 397 | | } |
| 398 | |
|
| 399 | | #[cfg(feature = "wasm")] |
| 400 | | impl Io { |
| 401 | 7× | fn note(&self, text: String) { |
| 402 | 7× | match self { |
| 403 | 0× | Io::Inherit => eprintln!("{NOTE}{text}"), |
| 404 | 7× | Io::Pipes { notes, .. } => { |
| 405 | 7× | let _ = notes.send(text); |
| 406 | 7× | } |
| 407 | | } |
| 408 | 7× | } |
| 409 | | } |
| 410 | |
|
| 411 | | #[cfg(feature = "wasm")] |
| 412 | |
|
| 413 | | struct Limits { |
| 414 | | memory: usize, |
| 415 | | peak: usize, |
| 416 | | } |
| 417 | |
|
| 418 | | #[cfg(feature = "wasm")] |
| 419 | | impl wasmtime::ResourceLimiter for Limits { |
| 420 | 9× | fn memory_growing(&mut self, _current: usize, desired: usize, maximum: Option<usize>) -> wasmtime::Result<bool> { |
| 421 | 9× | let ok = desired <= self.memory && maximum.is_none_or(|m| desired <= m); |
| 422 | 9× | if ok { |
| 423 | 8× | self.peak = self.peak.max(desired); |
| 424 | 8× | } |
| 425 | 9× | Ok(ok) |
| 426 | 9× | } |
| 427 | |
|
| 428 | 0× | fn table_growing(&mut self, _current: usize, desired: usize, maximum: Option<usize>) -> wasmtime::Result<bool> { |
| 429 | 0× | Ok(maximum.is_none_or(|m| desired <= m)) |
| 430 | 0× | } |
| 431 | | } |
| 432 | |
|
| 433 | | #[cfg(feature = "wasm")] |
| 434 | | struct State { |
| 435 | | wasi: wasmtime_wasi::p1::WasiP1Ctx, |
| 436 | | limits: Limits, |
| 437 | | } |
| 438 | |
|
| 439 | | #[cfg(feature = "wasm")] |
| 440 | |
|
| 441 | |
|
| 442 | 9× | fn run_modules(engine: &wasmtime::Engine, spec: &Spec, io: &Io) -> Result<(i32, usize)> { |
| 443 | 9× | let mut linker: wasmtime::Linker<State> = wasmtime::Linker::new(engine); |
| 444 | 9× | ww(wasmtime_wasi::p1::add_to_linker_sync(&mut linker, |s: &mut State| &mut s.wasi), || "link WASI".into())?; |
| 445 | 9× | let (mut first_failure, mut peak) = (0, 0); |
| 446 | 10× | for path in &spec.modules { |
| 447 | 10× | if spec.modules.len() > 1 { |
| 448 | 2× | io.note(format!("running {path}")); |
| 449 | 8× | } |
| 450 | 10× | let (code, held) = run_module(engine, &linker, spec, path, io)?; |
| 451 | 10× | peak = peak.max(held); |
| 452 | 10× | if spec.modules.len() > 1 { |
| 453 | 2× | io.note(format!("{path}: exit {code}")); |
| 454 | 8× | } |
| 455 | 10× | if first_failure == 0 && code != 0 { |
| 456 | 4× | first_failure = code; |
| 457 | 6× | } |
| 458 | | } |
| 459 | 9× | Ok((first_failure, peak)) |
| 460 | 9× | } |
| 461 | |
|
| 462 | | #[cfg(feature = "wasm")] |
| 463 | 10× | fn run_module(engine: &wasmtime::Engine, linker: &wasmtime::Linker<State>, spec: &Spec, path: &str, io: &Io) -> Result<(i32, usize)> { |
| 464 | | use wasmtime_wasi::{FsPerms, WasiCtxBuilder}; |
| 465 | 10× | let module = ww(wasmtime::Module::from_file(engine, spec.workspace.join(path)), || format!("module {path} does not load"))?; |
| 466 | 10× | let name = path.rsplit('/').next().unwrap_or(path); |
| 467 | 10× | let mut argv = vec![name.to_string()]; |
| 468 | 10× | argv.extend(spec.args.iter().cloned()); |
| 469 | 10× | let mut wasi = WasiCtxBuilder::new(); |
| 470 | 10× | match io { |
| 471 | 0× | Io::Inherit => { |
| 472 | 0× | wasi.inherit_stdout().inherit_stderr(); |
| 473 | 0× | } |
| 474 | 10× | Io::Pipes { stdout, stderr, .. } => { |
| 475 | 10× | wasi.stdout(stdout.clone()).stderr(stderr.clone()); |
| 476 | 10× | } |
| 477 | | } |
| 478 | 10× | wasi.args(&argv).envs(&spec.env).env("PWD", "/src").allow_blocking_current_thread(true); |
| 479 | 20× | let open = |wasi: &mut WasiCtxBuilder, dir: &Path, at: &str| { |
| 480 | 20× | ww(wasi.preopened_dir(dir, at, FsPerms::ReadWrite).map(|_| ()), || format!("open {} as {at}", dir.display())) |
| 481 | 20× | }; |
| 482 | 10× | open(&mut wasi, &spec.workspace, "/src")?; |
| 483 | | |
| 484 | 10× | open(&mut wasi, &spec.workspace, ".")?; |
| 485 | 10× | for (dir, target) in &spec.preopens { |
| 486 | 0× | open(&mut wasi, dir, target)?; |
| 487 | | } |
| 488 | 10× | let limits = Limits { memory: spec.memory_mib as usize * 1024 * 1024, peak: 0 }; |
| 489 | 10× | let mut store = wasmtime::Store::new(engine, State { wasi: wasi.build_p1(), limits }); |
| 490 | 10× | store.limiter(|s| &mut s.limits); |
| 491 | | |
| 492 | | |
| 493 | 10× | store.set_epoch_deadline(1); |
| 494 | 10× | let instance = ww(linker.instantiate(&mut store, &module), || format!("module {path} cannot start here"))?; |
| 495 | 10× | let start = |
| 496 | 10× | ww(instance.get_typed_func::<(), ()>(&mut store, "_start"), || format!("module {path} has no _start: is it a WASI command?"))?; |
| 497 | 10× | let code = match start.call(&mut store, ()) { |
| 498 | 6× | Ok(()) => 0, |
| 499 | 4× | Err(e) => match e.downcast_ref::<wasmtime_wasi::I32Exit>() { |
| 500 | 1× | Some(exit) => exit.0, |
| 501 | | None => { |
| 502 | 3× | io.note(format!("{path} trapped: {e:#}")); |
| 503 | 3× | TRAPPED |
| 504 | | } |
| 505 | | }, |
| 506 | | }; |
| 507 | 10× | Ok((code, store.data().limits.peak)) |
| 508 | 10× | } |
| 509 | |
|
| 510 | | #[cfg(all(test, feature = "wasm"))] |
| 511 | | mod tests { |
| 512 | | use super::*; |
| 513 | |
|
| 514 | | |
| 515 | 3× | fn spec_of(dir: &Path, text: &str) -> Spec { |
| 516 | 3× | std::fs::write(dir.join("m.wasm"), wat::parse_str(text).unwrap()).unwrap(); |
| 517 | 3× | Spec { workspace: dir.to_path_buf(), modules: vec!["m.wasm".into()], args: vec![], env: vec![], memory_mib: 64, preopens: vec![] } |
| 518 | 3× | } |
| 519 | |
|
| 520 | 3× | fn env() -> Environment { |
| 521 | 3× | Environment { runtime: "wasm".into(), parts: vec![], rootfs: String::new(), argv: vec![], env: vec![] } |
| 522 | 3× | } |
| 523 | |
|
| 524 | | |
| 525 | | struct Collect(Vec<Record>, bool); |
| 526 | | impl Sink for Collect { |
| 527 | 6× | fn record(&mut self, r: Record) { |
| 528 | 6× | self.0.push(r); |
| 529 | 6× | } |
| 530 | 5× | fn cancelled(&mut self) -> bool { |
| 531 | 5× | self.1 |
| 532 | 5× | } |
| 533 | | } |
| 534 | |
|
| 535 | | const PRINT: &str = r#"(module |
| 536 | | (import "wasi_snapshot_preview1" "fd_write" (func $w (param i32 i32 i32 i32) (result i32))) |
| 537 | | (memory (export "memory") 2) |
| 538 | | (data (i32.const 16) "line one\nline two") |
| 539 | | (func (export "_start") |
| 540 | | (i32.store (i32.const 0) (i32.const 16)) (i32.store (i32.const 4) (i32.const 17)) |
| 541 | | (drop (call $w (i32.const 1) (i32.const 0) (i32.const 1) (i32.const 8)))))"#; |
| 542 | |
|
| 543 | | const SPIN: &str = r#"(module (func (export "_start") (loop $l (br $l))))"#; |
| 544 | |
|
| 545 | | #[test] |
| 546 | 1× | fn in_process_output_streams_as_lines() { |
| 547 | 1× | let dir = tempfile::tempdir().unwrap(); |
| 548 | 1× | let mut sink = Collect(vec![], false); |
| 549 | 1× | let o = run_in_thread(&spec_of(dir.path(), PRINT), env(), Duration::from_secs(30), &mut sink).unwrap(); |
| 550 | 1× | assert_eq!(o.exit_code, Some(0)); |
| 551 | 2× | let out: Vec<&str> = sink.0.iter().filter(|r| r.stream == "out").map(|r| r.text.as_str()).collect(); |
| 552 | 1× | assert_eq!(out, ["line one", "line two"], "a last line without its newline still comes"); |
| 553 | 1× | assert_eq!(o.peak_memory_mib, Some(1), "two 64 KiB pages, in whole MiB"); |
| 554 | 1× | } |
| 555 | |
|
| 556 | | #[test] |
| 557 | 1× | fn in_process_stops_at_the_time_limit_and_when_cancelled() { |
| 558 | 1× | let dir = tempfile::tempdir().unwrap(); |
| 559 | 1× | let mut sink = Collect(vec![], false); |
| 560 | 1× | let started = Instant::now(); |
| 561 | 1× | let o = run_in_thread(&spec_of(dir.path(), SPIN), env(), Duration::from_millis(300), &mut sink).unwrap(); |
| 562 | 1× | assert!(o.timed_out && o.exit_code.is_none(), "{:?}", sink.0); |
| 563 | 1× | assert!(started.elapsed() < Duration::from_secs(5), "the endless loop was interrupted"); |
| 564 | 1× | let mut sink = Collect(vec![], true); |
| 565 | 1× | let o = run_in_thread(&spec_of(dir.path(), SPIN), env(), Duration::from_secs(30), &mut sink).unwrap(); |
| 566 | 1× | assert!(o.cancelled && o.exit_code.is_none()); |
| 567 | 1× | assert!(sink.0.iter().any(|r| r.text == "stopped: the run was cancelled")); |
| 568 | 1× | } |
| 569 | |
|
| 570 | | #[test] |
| 571 | 1× | fn globs_match_names_with_stars() { |
| 572 | 1× | assert!(wildcard("*.wasm", "oxy_core-1a2b.wasm")); |
| 573 | 1× | assert!(wildcard("oxy_core-*.wasm", "oxy_core-1a2b.wasm")); |
| 574 | 1× | assert!(!wildcard("oxy_core-*.wasm", "oxy_core-1a2b.d")); |
| 575 | 1× | assert!(wildcard("a*b*c", "aXbYc") && !wildcard("a*b*c", "aXcYb")); |
| 576 | 1× | assert!(wildcard("plain.wasm", "plain.wasm") && !wildcard("plain.wasm", "other.wasm")); |
| 577 | 1× | } |
| 578 | |
|
| 579 | | #[test] |
| 580 | 1× | fn modules_are_files_of_the_job() { |
| 581 | 1× | let dir = tempfile::tempdir().unwrap(); |
| 582 | 1× | let deps = dir.path().join("target/wasm32-wasip1/debug/deps"); |
| 583 | 1× | std::fs::create_dir_all(&deps).unwrap(); |
| 584 | 3× | for f in ["b-2.wasm", "a-1.wasm", "a-1.d"] { |
| 585 | 3× | std::fs::write(deps.join(f), b"").unwrap(); |
| 586 | 3× | } |
| 587 | 1× | let ws = dir.path(); |
| 588 | 1× | assert_eq!( |
| 589 | 1× | modules(ws, "/src/target/wasm32-wasip1/debug/deps/*.wasm").unwrap(), |
| 590 | | ["target/wasm32-wasip1/debug/deps/a-1.wasm", "target/wasm32-wasip1/debug/deps/b-2.wasm"] |
| 591 | | ); |
| 592 | 1× | assert_eq!(modules(ws, "./target/wasm32-wasip1/debug/deps/b-2.wasm").unwrap(), ["target/wasm32-wasip1/debug/deps/b-2.wasm"]); |
| 593 | 1× | assert!(modules(ws, "target/wasm32-wasip1/debug/deps/*.so").unwrap_err().to_string().contains("no module matches")); |
| 594 | 1× | assert!(modules(ws, "../etc/passwd").is_err()); |
| 595 | 1× | assert!(modules(ws, "/etc/x.wasm").is_err()); |
| 596 | 1× | assert!(modules(ws, "target/*/debug/a.wasm").unwrap_err().to_string().contains("only the file name")); |
| 597 | 1× | } |
| 598 | | } |