The runner as a library, with the WASM tier in process, for the phone (refs oxygenhub-android#8) 5cf1e8aefbb6

//! OxygenFlow's runner and tiers as a library: the `flow` command is built

//! on it, and so is the Android app's runner (oxygenhub-android#8), which

//! runs the WASM tier in its own process (docs/handover.md §10).

pub mod cache;

pub mod cpio;

pub mod egress;

pub mod facts;

/// The job-starting code `flow-init` uses in a microVM, shared with the

/// syscall tier.

#[path = "../../flow-init/src/job.rs"]

pub mod guest;

pub mod home;

pub mod images;

pub mod metrics;

pub mod microvm;

pub mod outputs;

pub mod runner;

pub mod swarm;

pub mod syscall;

pub mod wait;

pub mod wasm;

pub mod workspace;

- **The runner as a library, for the phone (oxygenhub-android#8):** `flow` is now a library and the command built on it, and the whole runner builds for Android (`aarch64-linux-android`); `runner::serve_embedded` and `runner::stop` run it inside an app and drain it. There the WASM tier runs modules in a thread rather than a child process (also with `FLOW_WASM_IN_PROCESS`), streaming their output from pipes, stopping them at the time limit or when the run is cancelled through wasmtime's epoch, and reporting the most memory they asked for. Two portability fixes on the way (the seccomp filter's BPF opcodes and an ioctl request type).

        // `as _`: the request's type differs between libcs (Android's is c_int).

        libc::ioctl(fd, libc::SIOCGIFFLAGS as _, &mut req) == 0 && {

            libc::ioctl(fd, libc::SIOCSIFFLAGS as _, &req) == 0

[lib]

name = "flow"

path = "src/lib.rs"

use flow::{facts, home, images, microvm, runner, syscall, wait, wasm, workspace};

    drain_on_stop();

    serve_until_drained(home)

}

/// The runner as a library, as the Android app runs it (oxygenhub-android#8):

/// no signal handlers; [`stop`] drains it, and this returns once the jobs

/// it holds are done and reported. It may be started again afterwards.

pub fn serve_embedded(home: &Home) -> Result<()> {

    STOPS.store(0, std::sync::atomic::Ordering::SeqCst);

    serve_until_drained(home)

}

/// Ask a runner started with [`serve_embedded`] to drain: claim nothing

/// more, finish and report what it holds, then return (oxygenflow#76's

/// drain, for a phone taken off charge or off the allowed network).

pub fn stop() {

    STOPS.fetch_add(1, std::sync::atomic::Ordering::SeqCst);

}

fn serve_until_drained(home: &Home) -> Result<()> {

        // An embedded runner starts afresh, and stop() drains it the same way.

        STOPS.store(0, std::sync::atomic::Ordering::SeqCst);

        stop();

        assert!(stopping());

        STOPS.store(0, std::sync::atomic::Ordering::SeqCst);

/// Classic BPF opcodes (linux/bpf_common.h), fixed by the kernel's ABI;

/// spelled out because libc does not export them for every target

/// (Android, where the runner is built for the WASM tier only).

const BPF_LD: u16 = 0x00;

const BPF_W: u16 = 0x00;

const BPF_ABS: u16 = 0x20;

const BPF_JMP: u16 = 0x05;

const BPF_JEQ: u16 = 0x10;

const BPF_K: u16 = 0x00;

const BPF_RET: u16 = 0x06;

    const LD: u16 = BPF_LD | BPF_W | BPF_ABS;

    const JEQ: u16 = BPF_JMP | BPF_JEQ | BPF_K;

    const RET: u16 = BPF_RET | BPF_K;

}

#[cfg(all(test, not(target_os = "android")))]

mod tests {

    /// The opcodes written out above are the ABI's, as glibc's libc has them.

    #[test]

    fn bpf_opcodes_are_the_kernels() {

        use super::*;

        assert_eq!(

            [BPF_LD, BPF_W, BPF_ABS, BPF_JMP, BPF_JEQ, BPF_K, BPF_RET],

            [libc::BPF_LD, libc::BPF_W, libc::BPF_ABS, libc::BPF_JMP, libc::BPF_JEQ, libc::BPF_K, libc::BPF_RET].map(|v| v as u16)

        );

    }

    if in_process() {

        return run_in_thread(&spec, environment, timeout, sink);

    }

}

/// Whether modules run in a thread of this process rather than in a child

/// `flow __wasm`: on Android, where the runner is a library inside an app

/// that cannot start itself as a command (oxygenhub-android#8), and

/// anywhere `FLOW_WASM_IN_PROCESS` is set (how the tests check this way).

pub fn in_process() -> bool {

    cfg!(target_os = "android") || std::env::var_os("FLOW_WASM_IN_PROCESS").is_some()

/// The most of a module's output kept before it is read, per stream.

const PIPE_BYTES: usize = 64 << 20;

/// Run the spec's modules in a thread of this process, streaming their

/// output as it comes: the pipes are read every 100 ms, and a time limit

/// or a cancelled run interrupts the module through wasmtime's epoch.

fn run_in_thread(spec: &Spec, environment: Environment, timeout: Duration, sink: &mut dyn Sink) -> Result<Outcome> {

    use wasmtime_wasi::p2::pipe::MemoryOutputPipe;

    let mut config = wasmtime::Config::new();

    config.epoch_interruption(true);

    let engine = ww(wasmtime::Engine::new(&config), || "start wasmtime".into())?;

    let (stdout, stderr) = (MemoryOutputPipe::new(PIPE_BYTES), MemoryOutputPipe::new(PIPE_BYTES));

    let (notes_tx, notes) = mpsc::channel::<String>();

    let io = Io::Pipes { stdout: stdout.clone(), stderr: stderr.clone(), notes: notes_tx };

    let started = Instant::now();

    let t = |s: &Instant| s.elapsed().as_millis() as u64;

    let deadline = started + timeout;

    let (mut timed_out, mut cancelled) = (false, false);

    // Bytes of each stream already passed on, and the part line not ended yet.

    let mut said = [(0usize, Vec::<u8>::new()), (0usize, Vec::<u8>::new())];

    let mut pass_on = |sink: &mut dyn Sink, last: bool| {

        for (i, pipe) in [&stdout, &stderr].into_iter().enumerate() {

            let all = pipe.contents();

            let (offset, partial) = &mut said[i];

            partial.extend_from_slice(&all[*offset..]);

            *offset = all.len();

            while let Some(n) = partial.iter().position(|b| *b == b'\n') {

                let line: Vec<u8> = partial.drain(..=n).collect();

                let text = String::from_utf8_lossy(&line[..line.len() - 1]).into_owned();

                sink.record(Record { t: t(&started), stream: if i == 0 { "out" } else { "err" }.into(), text });

            }

            if last && !partial.is_empty() {

                let text = String::from_utf8_lossy(partial).into_owned();

                partial.clear();

                sink.record(Record { t: t(&started), stream: if i == 0 { "out" } else { "err" }.into(), text });

            }

        }

        while let Ok(note) = notes.try_recv() {

            sink.record(Record { t: t(&started), stream: "flow".into(), text: note });

        }

    };

    let ran = std::thread::scope(|scope| {

        let worker = scope.spawn(|| run_modules(&engine, spec, &io));

        while !worker.is_finished() {

            std::thread::sleep(Duration::from_millis(100));

            pass_on(sink, false);

            if !timed_out && !cancelled && sink.cancelled() {

                cancelled = true;

                engine.increment_epoch();

                sink.record(Record { t: t(&started), stream: "flow".into(), text: "stopped: the run was cancelled".into() });

            }

            if Instant::now() >= deadline && !timed_out && !cancelled {

                timed_out = true;

                engine.increment_epoch();

                sink.record(Record {

                    t: t(&started),

                    stream: "flow".into(),

                    text: format!("stopped at the time limit of {} s", timeout.as_secs()),

                });

            }

        }

        worker.join().unwrap_or_else(|_| Err(anyhow::anyhow!("the WASM worker panicked")))

    });

    pass_on(sink, true);

    let (exit_code, peak) = match ran {

        Ok((code, peak)) => (Some(code), Some(peak)),

        Err(e) => {

            sink.record(Record { t: t(&started), stream: "flow".into(), text: format!("the WASM tier could not run the job: {e:#}") });

            (Some(125), None)

        }

    };

    Ok(Outcome {

        exit_code: if timed_out || cancelled { None } else { exit_code },

        duration_ms: t(&started),

        environment,

        timed_out,

        cancelled,

        peak_memory_mib: peak.map(|b| (b as u64).div_ceil(1 << 20)),

        pruned: Vec::new(),

    })

}

    let ran = (|| {

        let spec: Spec = serde_json::from_slice(&std::fs::read(spec_path)?).context("read the spec")?;

        let engine = ww(wasmtime::Engine::new(&wasmtime::Config::new()), || "start wasmtime".into())?;

        run_modules(&engine, &spec, &Io::Inherit)

    })();

    match ran {

        Ok((code, _)) => code,

}

/// Where a module's output and the tier's notes go: inherited by the

/// helper process (notes on stderr, marked), or pipes the runner reads.

enum Io {

    Inherit,

    Pipes {

        stdout: wasmtime_wasi::p2::pipe::MemoryOutputPipe,

        stderr: wasmtime_wasi::p2::pipe::MemoryOutputPipe,

        notes: mpsc::Sender<String>,

    },

}

impl Io {

    fn note(&self, text: String) {

        match self {

            Io::Inherit => eprintln!("{NOTE}{text}"),

            Io::Pipes { notes, .. } => {

                let _ = notes.send(text);

            }

        }

    }

/// The sandbox's memory limit, and the most a module asked for within it.

struct Limits {

    memory: usize,

    peak: usize,

}

impl wasmtime::ResourceLimiter for Limits {

    fn memory_growing(&mut self, _current: usize, desired: usize, maximum: Option<usize>) -> wasmtime::Result<bool> {

        let ok = desired <= self.memory && maximum.is_none_or(|m| desired <= m);

        if ok {

            self.peak = self.peak.max(desired);

        }

        Ok(ok)

    }

    fn table_growing(&mut self, _current: usize, desired: usize, maximum: Option<usize>) -> wasmtime::Result<bool> {

        Ok(maximum.is_none_or(|m| desired <= m))

    }

}

    limits: Limits,

/// Run each module in turn: the first non-zero exit code after all ran,

/// and the most memory any of them held, in bytes.

fn run_modules(engine: &wasmtime::Engine, spec: &Spec, io: &Io) -> Result<(i32, usize)> {

    let mut linker: wasmtime::Linker<State> = wasmtime::Linker::new(engine);

    let (mut first_failure, mut peak) = (0, 0);

            io.note(format!("running {path}"));

        let (code, held) = run_module(engine, &linker, spec, path, io)?;

        peak = peak.max(held);

        if spec.modules.len() > 1 {

            io.note(format!("{path}: exit {code}"));

        }

    Ok((first_failure, peak))

fn run_module(engine: &wasmtime::Engine, linker: &wasmtime::Linker<State>, spec: &Spec, path: &str, io: &Io) -> Result<(i32, usize)> {

    match io {

        Io::Inherit => {

            wasi.inherit_stdout().inherit_stderr();

        }

        Io::Pipes { stdout, stderr, .. } => {

            wasi.stdout(stdout.clone()).stderr(stderr.clone());

        }

    }

    wasi.args(&argv).envs(&spec.env).env("PWD", "/src").allow_blocking_current_thread(true);

    let limits = Limits { memory: spec.memory_mib as usize * 1024 * 1024, peak: 0 };

    // With epoch interruption on (in process), one tick from now traps:

    // the runner ticks only to stop the module.

    store.set_epoch_deadline(1);

    let code = match start.call(&mut store, ()) {

                io.note(format!("{path} trapped: {e:#}"));

    };

    Ok((code, store.data().limits.peak))

    /// A spec running `text` (WAT) as the only module, with 64 MiB.

    fn spec_of(dir: &Path, text: &str) -> Spec {

        std::fs::write(dir.join("m.wasm"), wat::parse_str(text).unwrap()).unwrap();

        Spec { workspace: dir.to_path_buf(), modules: vec!["m.wasm".into()], args: vec![], env: vec![], memory_mib: 64, preopens: vec![] }

    }

    fn env() -> Environment {

        Environment { runtime: "wasm".into(), parts: vec![], rootfs: String::new(), argv: vec![], env: vec![] }

    }

    /// Records, and whether to say the run was cancelled.

    struct Collect(Vec<Record>, bool);

    impl Sink for Collect {

        fn record(&mut self, r: Record) {

            self.0.push(r);

        }

        fn cancelled(&mut self) -> bool {

            self.1

        }

    }

    const PRINT: &str = r#"(module

      (import "wasi_snapshot_preview1" "fd_write" (func $w (param i32 i32 i32 i32) (result i32)))

      (memory (export "memory") 2)

      (data (i32.const 16) "line one\nline two")

      (func (export "_start")

        (i32.store (i32.const 0) (i32.const 16)) (i32.store (i32.const 4) (i32.const 17))

        (drop (call $w (i32.const 1) (i32.const 0) (i32.const 1) (i32.const 8)))))"#;

    const SPIN: &str = r#"(module (func (export "_start") (loop $l (br $l))))"#;

    #[test]

    fn in_process_output_streams_as_lines() {

        let dir = tempfile::tempdir().unwrap();

        let mut sink = Collect(vec![], false);

        let o = run_in_thread(&spec_of(dir.path(), PRINT), env(), Duration::from_secs(30), &mut sink).unwrap();

        assert_eq!(o.exit_code, Some(0));

        let out: Vec<&str> = sink.0.iter().filter(|r| r.stream == "out").map(|r| r.text.as_str()).collect();

        assert_eq!(out, ["line one", "line two"], "a last line without its newline still comes");

        assert_eq!(o.peak_memory_mib, Some(1), "two 64 KiB pages, in whole MiB");

    }

    #[test]

    fn in_process_stops_at_the_time_limit_and_when_cancelled() {

        let dir = tempfile::tempdir().unwrap();

        let mut sink = Collect(vec![], false);

        let started = Instant::now();

        let o = run_in_thread(&spec_of(dir.path(), SPIN), env(), Duration::from_millis(300), &mut sink).unwrap();

        assert!(o.timed_out && o.exit_code.is_none(), "{:?}", sink.0);

        assert!(started.elapsed() < Duration::from_secs(5), "the endless loop was interrupted");

        let mut sink = Collect(vec![], true);

        let o = run_in_thread(&spec_of(dir.path(), SPIN), env(), Duration::from_secs(30), &mut sink).unwrap();

        assert!(o.cancelled && o.exit_code.is_none());

        assert!(sink.0.iter().any(|r| r.text == "stopped: the run was cancelled"));

    }

    confined(false);

}

/// The same in a thread of the runner, as on Android, where the runner is

/// a library in an app that cannot start itself as a command.

#[test]

fn wasi_modules_run_confined_in_process() {

    confined(true);

}

fn confined(in_process: bool) {

    let mut flow = Command::new(env!("CARGO_BIN_EXE_flow"));

    if in_process {

        flow.env("FLOW_WASM_IN_PROCESS", "1");

    }

    let run = flow.arg("--home").arg(&home).arg("run").current_dir(&work).output().unwrap();