Merge v3-t15: shared compile cache, precompile, /bin/*.wasm and /src/sausage copies (T15)

host/src/main.rs

@@ -5,6 +5,9 @@ use std::path::PathBuf;
 #[derive(Parser)]
 #[command(version, about)]
 struct Cli {
+    /// Shared host-user compile cache (otherwise SAUSAGE_CACHE or the XDG cache folder).
+    #[arg(long, global = true)]
+    cache: Option<PathBuf>,
     #[command(subcommand)]
     command: Command,
 }
@@ -29,7 +32,11 @@ enum Command {
 
 #[tokio::main]
 async fn main() -> anyhow::Result<()> {
-    match Cli::parse().command {
+    let cli = Cli::parse();
+    if let Some(cache) = cli.cache {
+        sausage::process::configure_cache(cache)?;
+    }
+    match cli.command {
         Command::Run { world, userland } => {
             let bundle =
                 userland.or_else(|| std::env::var_os("SAUSAGE_USERLAND").map(PathBuf::from));

host/src/process/cache.rs

@@ -0,0 +1,236 @@
+//! Shares compiled programs and single-flight compilation across worlds in one host process.
+use super::{Compiled, State, adapter, bindings};
+use std::{
+    collections::HashMap,
+    path::{Path, PathBuf},
+    sync::{Arc, Mutex, OnceLock},
+};
+use tokio::sync::watch;
+type Compilation = watch::Receiver<Option<Result<Compiled, String>>>;
+use wasmtime::{
+    Engine,
+    component::{Component, Linker},
+};
+use wasmtime_wasi::p2::bindings::CommandPre;
+
+static DIRECTORY: OnceLock<PathBuf> = OnceLock::new();
+static SHARED: Mutex<Option<Arc<Compiler>>> = Mutex::new(None);
+
+/// Select the host user's cache before opening a world.
+pub fn configure(directory: PathBuf) -> anyhow::Result<()> {
+    DIRECTORY
+        .set(directory)
+        .map_err(|_| anyhow::anyhow!("cache directory already configured"))
+}
+fn directory() -> anyhow::Result<PathBuf> {
+    if let Some(dir) = DIRECTORY.get() {
+        return Ok(dir.clone());
+    }
+    if let Some(dir) = std::env::var_os("SAUSAGE_CACHE") {
+        return Ok(dir.into());
+    }
+    if let Some(dir) = std::env::var_os("XDG_CACHE_HOME") {
+        return Ok(PathBuf::from(dir).join("sausage"));
+    }
+    Ok(
+        PathBuf::from(std::env::var_os("HOME").ok_or_else(|| anyhow::anyhow!("HOME is not set"))?)
+            .join(".cache/sausage"),
+    )
+}
+pub(super) fn shared() -> anyhow::Result<Arc<Compiler>> {
+    let mut shared = SHARED.lock().unwrap();
+    if shared.is_none() {
+        *shared = Some(Compiler::new(&directory()?)?);
+    }
+    Ok(shared.as_ref().unwrap().clone())
+}
+pub(super) struct Compiler {
+    pub engine: Engine,
+    programs: Mutex<HashMap<Vec<u8>, Compilation>>,
+    #[cfg(test)]
+    before_compile: Mutex<Option<Box<dyn FnOnce() + Send>>>,
+    #[cfg(test)]
+    compilations: std::sync::atomic::AtomicUsize,
+}
+impl Compiler {
+    pub fn new(directory: &Path) -> anyhow::Result<Arc<Self>> {
+        let _ = rustls::crypto::aws_lc_rs::default_provider().install_default();
+        let mut config = wasmtime::Config::new();
+        config.epoch_interruption(true);
+        // Keep Cranelift's default optimization: compiles are cached per content, and guests run
+        // many times. OptLevel::None saved only 12.7% of compile time (T15) but slows every run.
+        let mut cache = wasmtime::CacheConfig::new();
+        cache.with_directory(directory);
+        config.cache(Some(wasmtime::Cache::new(cache)?));
+        Ok(Arc::new(Self {
+            engine: Engine::new(&config)?,
+            programs: Mutex::new(HashMap::new()),
+            #[cfg(test)]
+            before_compile: Mutex::new(None),
+            #[cfg(test)]
+            compilations: std::sync::atomic::AtomicUsize::new(0),
+        }))
+    }
+    pub async fn get(self: &Arc<Self>, bytes: Vec<u8>) -> anyhow::Result<Compiled> {
+        let mut result = self
+            .programs
+            .lock()
+            .unwrap()
+            .entry(bytes.clone())
+            .or_insert_with(|| {
+                let (tx, rx) = watch::channel(None);
+                let compiler = self.clone();
+                // The job owns completion even if a command or a precompile waiter is cancelled.
+                tokio::spawn(async move {
+                    let result =
+                        tokio::task::spawn_blocking(move || compiler.compile(&bytes)).await;
+                    tx.send_replace(Some(
+                        result
+                            .map_err(anyhow::Error::from)
+                            .and_then(|r| r)
+                            .map_err(|e| format!("{e:#}")),
+                    ));
+                });
+                rx
+            })
+            .clone();
+        loop {
+            if let Some(result) = result.borrow().clone() {
+                return result.map_err(|e| anyhow::anyhow!(e));
+            }
+            result.changed().await?;
+        }
+    }
+    fn compile(&self, bytes: &[u8]) -> anyhow::Result<Compiled> {
+        #[cfg(test)]
+        {
+            self.compilations
+                .fetch_add(1, std::sync::atomic::Ordering::Relaxed);
+            if let Some(hook) = self.before_compile.lock().unwrap().take() {
+                hook();
+            }
+        }
+        if bytes.get(..8) == Some(b"\0asm\x01\0\0\0") {
+            let module = wasmtime::Module::new(&self.engine, bytes)?;
+            let mut linker = wasmtime::Linker::new(&self.engine);
+            sausage_vfs::preview1::add_to_linker(&mut linker, |s: &mut State| {
+                s.preview1.as_mut().unwrap()
+            })?;
+            Ok(Compiled::Module(linker.instantiate_pre(&module)?))
+        } else {
+            let component = Component::new(&self.engine, bytes)?;
+            let mut linker = Linker::new(&self.engine);
+            wasmtime_wasi::p2::add_to_linker_async(&mut linker)?;
+            wasmtime_wasi_http::p2::add_only_http_to_linker_async(&mut linker)?;
+            adapter::add_to_linker(&mut linker, |s: &mut State| adapter::View {
+                adapter: &mut s.fs,
+                table: &mut s.table,
+            })?;
+            bindings::sausage::process::host::add_to_linker::<_, wasmtime::component::HasSelf<_>>(
+                &mut linker,
+                |s| s,
+            )?;
+            Ok(Compiled::Component(CommandPre::new(
+                linker.instantiate_pre(&component)?,
+            )?))
+        }
+    }
+}
+/// Compile programs in both command folders without blocking world startup.
+pub(super) fn precompile(
+    compiler: Arc<Compiler>,
+    root: Arc<dyn sausage_vfs::Layer>,
+) -> tokio::task::JoinHandle<anyhow::Result<usize>> {
+    tokio::spawn(async move {
+        let mut tasks = tokio::task::JoinSet::new();
+        // Leave CPU and file bandwidth for commands during startup.
+        let slots = Arc::new(tokio::sync::Semaphore::new(2));
+        for folder in ["/bin", "/usr/local/bin"] {
+            let entries = match root.list(folder).await {
+                Ok(entries) => entries,
+                Err(e) if e.kind() == std::io::ErrorKind::NotFound => continue,
+                Err(e) => return Err(e.into()),
+            };
+            for entry in entries.into_iter().filter(|entry| !entry.directory) {
+                let root = root.clone();
+                let compiler = compiler.clone();
+                let path = format!("{folder}/{}", entry.name);
+                let slots = slots.clone();
+                tasks.spawn(async move {
+                    let _permit = slots.acquire().await?;
+                    let file = root
+                        .open(
+                            &path,
+                            sausage_vfs::Open {
+                                follow: true,
+                                ..Default::default()
+                            },
+                        )
+                        .await?;
+                    let bytes = file.read(0, file.stat().await?.size as usize).await?;
+                    if bytes.get(..4) != Some(b"\0asm") {
+                        return Ok::<_, anyhow::Error>(false);
+                    }
+                    compiler.get(bytes).await?;
+                    Ok(true)
+                });
+            }
+        }
+        let mut count = 0;
+        while let Some(result) = tasks.join_next().await {
+            if result?? {
+                count += 1;
+            }
+        }
+        Ok(count)
+    })
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::exec::{Command, Exec, Options};
+    use std::{future::Future, sync::atomic::Ordering};
+
+    #[tokio::test(flavor = "multi_thread")]
+    async fn command_joins_precompile_even_after_waiter_cancellation() {
+        let temp = tempfile::tempdir().unwrap();
+        let compiler = Compiler::new(&temp.path().join("cache")).unwrap();
+        let world = crate::world::World::open(&temp.path().join("world")).unwrap();
+        let bytes =
+            wat::parse_str(r#"(module (memory (export "memory") 1) (func (export "_start")))"#)
+                .unwrap();
+        std::fs::create_dir_all(temp.path().join("world/world/bin")).unwrap();
+        std::fs::write(temp.path().join("world/world/bin/program.wasm"), &bytes).unwrap();
+        let (started_tx, started_rx) = tokio::sync::oneshot::channel();
+        let (finish_tx, finish_rx) = std::sync::mpsc::channel();
+        *compiler.before_compile.lock().unwrap() = Some(Box::new(move || {
+            started_tx.send(()).unwrap();
+            finish_rx.recv().unwrap();
+        }));
+        // Startup returns while the compiler remains blocked behind the test gate.
+        let runtime = super::super::Runtime::with_compiler(world, compiler.clone()).unwrap();
+        started_rx.await.unwrap();
+        let background = compiler.clone();
+        let precompile = tokio::spawn(async move { background.get(bytes).await });
+        let mut output = |_: crate::exec::Stream, _: &[u8]| {};
+        let command = runtime.exec(
+            Command::Argv(vec!["/bin/program".into()]),
+            Options::default(),
+            &mut output,
+        );
+        tokio::pin!(command);
+        let pending = std::future::poll_fn(|cx| {
+            std::task::Poll::Ready(command.as_mut().poll(cx).is_pending())
+        })
+        .await;
+        assert!(pending);
+        assert_eq!(compiler.compilations.load(Ordering::Relaxed), 1);
+        precompile.abort();
+        let _ = precompile.await;
+        finish_tx.send(()).unwrap();
+        assert_eq!(command.await.unwrap(), 0);
+        assert_eq!(compiler.compilations.load(Ordering::Relaxed), 1);
+        assert!(!temp.path().join("world/host/compile-cache").exists());
+    }
+}

host/src/process/mod.rs

@@ -5,7 +5,7 @@ use crate::{
 };
 use sausage_vfs::{
     Mounts,
-    adapter::{self, Adapter, View},
+    adapter::{self, Adapter},
 };
 use std::{
     collections::HashMap,
@@ -22,7 +22,7 @@ use tokio::{
 };
 use wasmtime::{
     Engine, Store,
-    component::{Component, Linker, Resource, ResourceTable},
+    component::{Resource, ResourceTable},
 };
 use wasmtime_wasi::{
     WasiCtx, WasiCtxBuilder, WasiCtxView, WasiView,
@@ -45,6 +45,8 @@ mod bindings {
     });
 }
 use bindings::sausage::process::host::{Host, HostChild, Request, Stdio};
+mod cache;
+pub use cache::configure as configure_cache;
 mod services;
 mod stdio;
 pub use services::supervise;
@@ -77,7 +79,8 @@ struct Entry {
 pub struct Runtime {
     world: World,
     engine: Engine,
-    cache: Mutex<HashMap<Vec<u8>, Compiled>>,
+    compiler: Arc<cache::Compiler>,
+    cache: Mutex<std::collections::HashSet<Vec<u8>>>,
     entries: Mutex<HashMap<u32, Entry>>,
     next: AtomicU32,
     alive: Arc<AtomicBool>,
@@ -138,14 +141,14 @@ impl Runtime {
         Self::new(World::open(path)?)
     }
     pub fn new(world: World) -> anyhow::Result<Arc<Self>> {
-        // Deno and WASI HTTP enable different providers. Select one before HTTPS creates a client.
-        let _ = rustls::crypto::aws_lc_rs::default_provider().install_default();
-        let mut config = wasmtime::Config::new();
-        config.epoch_interruption(true);
-        let mut cache = wasmtime::CacheConfig::new();
-        cache.with_directory(world.host_dir.join("compile-cache"));
-        config.cache(Some(wasmtime::Cache::new(cache)?));
-        let engine = Engine::new(&config)?;
+        Self::with_compiler(world, cache::shared()?)
+    }
+    /// Open an independent engine using an explicit shared disk cache.
+    pub fn new_with_cache(world: World, directory: &Path) -> anyhow::Result<Arc<Self>> {
+        Self::with_compiler(world, cache::Compiler::new(directory)?)
+    }
+    fn with_compiler(world: World, compiler: Arc<cache::Compiler>) -> anyhow::Result<Arc<Self>> {
+        let engine = compiler.engine.clone();
         let alive = Arc::new(AtomicBool::new(true));
         let ticking = alive.clone();
         let clock = engine.clone();
@@ -155,19 +158,57 @@ impl Runtime {
                 clock.increment_epoch();
             }
         });
-        Ok(Arc::new(Self {
+        let runtime = Arc::new(Self {
             world,
             engine,
             alive,
-            cache: Mutex::new(HashMap::new()),
+            compiler,
+            cache: Mutex::new(Default::default()),
             entries: Mutex::new(HashMap::new()),
             next: AtomicU32::new(1),
-        }))
+        });
+        cache::precompile(runtime.compiler.clone(), runtime.world.root.clone());
+        Ok(runtime)
     }
     /// Number of distinct program contents compiled by this runtime.
     pub fn compiled_programs(&self) -> usize {
         self.cache.lock().unwrap().len()
     }
+    async fn resolve_program(
+        &self,
+        cwd: &str,
+        name: &str,
+        env: &[(String, String)],
+    ) -> anyhow::Result<String> {
+        let paths = if name.contains('/') {
+            vec![sausage_vfs::resolve(cwd, name)]
+        } else {
+            let path = env
+                .iter()
+                .find(|(key, _)| key == "PATH")
+                .map(|(_, value)| value.as_str())
+                .unwrap_or("/usr/local/bin:/bin");
+            path.split(':')
+                .map(|folder| {
+                    sausage_vfs::resolve(
+                        cwd,
+                        &std::path::Path::new(folder).join(name).to_string_lossy(),
+                    )
+                })
+                .collect()
+        };
+        for path in paths {
+            for candidate in [path.clone(), format!("{path}.wasm")] {
+                match self.world.root.stat(&candidate, true).await {
+                    Ok(stat) if !stat.directory => return Ok(candidate),
+                    Ok(_) => {}
+                    Err(e) if e.kind() == std::io::ErrorKind::NotFound => {}
+                    Err(e) => return Err(e.into()),
+                }
+            }
+        }
+        anyhow::bail!("program not found: {name}")
+    }
     fn kill_command(&self, command: u32) {
         for entry in self.entries.lock().unwrap().values() {
             if entry.command == command {
@@ -222,7 +263,9 @@ impl Runtime {
         request.cwd = sausage_vfs::resolve("/", &request.cwd);
         request.env.retain(|(name, _)| name != "PWD");
         request.env.push(("PWD".into(), request.cwd.clone()));
-        let executable = sausage_vfs::resolve(&request.cwd, &request.program);
+        let executable = self
+            .resolve_program(&request.cwd, &request.program, &request.env)
+            .await?;
         let file = self
             .world
             .root
@@ -237,38 +280,8 @@ impl Runtime {
         let bytes = file
             .read(0, usize::try_from(file.stat().await?.size)?)
             .await?;
-        let pre = {
-            let mut cache = self.cache.lock().unwrap();
-            if let Some(pre) = cache.get(&bytes) {
-                pre.clone()
-            } else {
-                let pre = if bytes.get(..8) == Some(b"\0asm\x01\0\0\0") {
-                    let module = wasmtime::Module::new(&self.engine, &bytes)?;
-                    let mut linker = wasmtime::Linker::new(&self.engine);
-                    sausage_vfs::preview1::add_to_linker(&mut linker, |s: &mut State| {
-                        s.preview1.as_mut().unwrap()
-                    })?;
-                    Compiled::Module(linker.instantiate_pre(&module)?)
-                } else {
-                    let component = Component::new(&self.engine, &bytes)?;
-                    let mut linker = Linker::new(&self.engine);
-                    wasmtime_wasi::p2::add_to_linker_async(&mut linker)?;
-                    // The library linker includes exit-with-code. Only Wasmtime's CLI gates it.
-                    wasmtime_wasi_http::p2::add_only_http_to_linker_async(&mut linker)?;
-                    adapter::add_to_linker(&mut linker, |s: &mut State| View {
-                        adapter: &mut s.fs,
-                        table: &mut s.table,
-                    })?;
-                    bindings::sausage::process::host::add_to_linker::<
-                        _,
-                        wasmtime::component::HasSelf<_>,
-                    >(&mut linker, |s| s)?;
-                    Compiled::Component(CommandPre::new(linker.instantiate_pre(&component)?)?)
-                };
-                cache.insert(bytes, pre.clone());
-                pre
-            }
-        };
+        let pre = self.compiler.get(bytes.clone()).await?;
+        self.cache.lock().unwrap().insert(bytes);
         let pid = self.next.fetch_add(1, Ordering::Relaxed);
         let command = if command == 0 { pid } else { command };
         let group = match request.group {

host/src/world.rs

@@ -19,17 +19,37 @@ pub struct World {
     pub processes: crate::process::Processes,
 }
 
+fn copy_tree(source: &Path, destination: &Path) -> std::io::Result<()> {
+    std::fs::create_dir_all(destination)?;
+    for entry in std::fs::read_dir(source)? {
+        let entry = entry?;
+        let target = destination.join(entry.file_name());
+        if entry.file_type()?.is_dir() {
+            copy_tree(&entry.path(), &target)?;
+        } else {
+            std::fs::copy(entry.path(), target)?;
+        }
+    }
+    Ok(())
+}
+
 impl World {
     /// Open (and create if missing) the world directory `dir`, including `/proc` and `/run` mounts.
     pub fn open(dir: &Path) -> anyhow::Result<Self> {
         Self::open_with_userland(dir, None)
     }
 
-    /// Mount an optional host bundle read-only at `/bin` (D43).
+    /// Copy an optional bundle into a new world. Existing worlds retain their programs.
     pub fn open_with_userland(dir: &Path, userland: Option<&Path>) -> anyhow::Result<Self> {
         let root_dir = dir.join("world");
         let host_dir = dir.join("host");
+        let fresh = !root_dir.exists();
         std::fs::create_dir_all(root_dir.join("usr/local/bin"))?;
+        if fresh && let Some(bundle) = userland {
+            for folder in ["bin", "src"] {
+                copy_tree(&bundle.join(folder), &root_dir.join(folder))?;
+            }
+        }
         std::fs::create_dir_all(&host_dir)?;
         let mut mounts = Mounts::default();
         mounts.mount(
@@ -55,17 +75,11 @@ impl World {
             }
             Ok(text.into_bytes())
         });
-        if let Some(bundle) = userland {
-            mounts.mount(
-                "/bin",
-                Arc::new(Host::new(bundle, Process::default())?),
-                true,
-            );
-        }
         mounts.mount("/proc", Arc::new(proc), true);
         mounts.mount("/run", Arc::new(Memory::default()), false);
+        let root: Arc<dyn Layer> = Arc::new(mounts);
         Ok(World {
-            root: Arc::new(mounts),
+            root,
             host_dir,
             processes,
         })