Merge branch 'v3' into v3-t93
This commit is contained in:
+3
-20
@@ -52,27 +52,10 @@ async fn main() -> anyhow::Result<()> {
|
||||
"Copy tools failed: default tools unavailable. Set SAUSAGE_USERLAND or pass --userland."
|
||||
)
|
||||
})?;
|
||||
use std::io::{IsTerminal, Write};
|
||||
let terminal = std::io::stderr().is_terminal();
|
||||
let mut progress_line = false;
|
||||
let initialized = sausage::world::World::init_with_progress(&world, &bundle, |line| {
|
||||
if terminal && line.starts_with("Precompile tools ") {
|
||||
eprint!("\r\x1b[2K{line}");
|
||||
let _ = std::io::stderr().flush();
|
||||
progress_line = true;
|
||||
} else {
|
||||
if progress_line {
|
||||
eprintln!();
|
||||
progress_line = false;
|
||||
}
|
||||
eprintln!("{line}");
|
||||
}
|
||||
sausage::world::World::init_with_progress(&world, &bundle, |line| {
|
||||
eprintln!("{line}");
|
||||
})
|
||||
.await;
|
||||
if progress_line {
|
||||
eprintln!();
|
||||
}
|
||||
initialized?;
|
||||
.await?;
|
||||
eprintln!("Next steps:");
|
||||
eprintln!(
|
||||
" Edit {} to configure providers and API keys.",
|
||||
|
||||
+58
-136
@@ -1,4 +1,4 @@
|
||||
//! Shares compiled programs and single-flight compilation across worlds in one host process.
|
||||
//! Compiles programs on first start and shares cached results across worlds in one host process.
|
||||
use super::{Compiled, State, adapter, bindings};
|
||||
use std::{
|
||||
collections::HashMap,
|
||||
@@ -130,98 +130,6 @@ impl Compiler {
|
||||
}
|
||||
}
|
||||
}
|
||||
/// Finish precompilation during init and report each bundled tool, excluding executable aliases.
|
||||
pub async fn precompile_with_progress(
|
||||
root: &dyn sausage_vfs::Layer,
|
||||
mut progress: impl FnMut(usize, usize) + Send,
|
||||
) -> anyhow::Result<()> {
|
||||
use anyhow::Context;
|
||||
let compiler = shared()?;
|
||||
let mut paths = Vec::new();
|
||||
for folder in ["/bin", "/usr/local/bin"] {
|
||||
for entry in root.list(folder).await? {
|
||||
if !entry.directory && entry.name.ends_with(".wasm") {
|
||||
paths.push(format!("{folder}/{}", entry.name));
|
||||
}
|
||||
}
|
||||
}
|
||||
paths.sort_by_key(|path| priority(path.rsplit('/').next().unwrap()));
|
||||
progress(0, paths.len());
|
||||
for (index, path) in paths.iter().enumerate() {
|
||||
let file = root
|
||||
.open(path, Default::default())
|
||||
.await
|
||||
.with_context(|| format!("Precompile {path} failed"))?;
|
||||
let bytes = file.read(0, file.stat().await?.size as usize).await?;
|
||||
compiler
|
||||
.get(bytes)
|
||||
.await
|
||||
.with_context(|| format!("Precompile {path} failed"))?;
|
||||
progress(index + 1, paths.len());
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn priority(name: &str) -> usize {
|
||||
[
|
||||
"sh", "git", "rg", "grep", "ls", "cat", "sed", "find", "cargo",
|
||||
]
|
||||
.iter()
|
||||
.position(|tool| *tool == name.strip_suffix(".wasm").unwrap_or(name))
|
||||
.unwrap_or(9)
|
||||
}
|
||||
/// Compile common tools first 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 mut 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()),
|
||||
};
|
||||
entries.sort_by_key(|entry| priority(&entry.name));
|
||||
for entry in entries.into_iter().filter(|entry| !entry.directory) {
|
||||
// Reserve before spawning so task scheduling cannot reorder admission.
|
||||
let permit = slots.clone().acquire_owned().await?;
|
||||
let root = root.clone();
|
||||
let compiler = compiler.clone();
|
||||
let path = format!("{folder}/{}", entry.name);
|
||||
tasks.spawn(async move {
|
||||
let _permit = permit;
|
||||
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::*;
|
||||
@@ -229,48 +137,55 @@ mod tests {
|
||||
use std::{future::Future, sync::atomic::Ordering};
|
||||
|
||||
#[tokio::test]
|
||||
async fn init_progress_counts_tools_not_aliases() -> anyhow::Result<()> {
|
||||
async fn startup_is_lazy_and_identical_programs_share_first_compilation() -> anyhow::Result<()>
|
||||
{
|
||||
use sausage_vfs::Layer;
|
||||
let dir = tempfile::tempdir()?;
|
||||
let world = crate::world::World::open(dir.path())?;
|
||||
let bin = dir.path().join("world/bin");
|
||||
std::fs::create_dir_all(&bin)?;
|
||||
let mut progress = Vec::new();
|
||||
precompile_with_progress(world.root.as_ref(), |done, total| {
|
||||
progress.push((done, total))
|
||||
})
|
||||
.await?;
|
||||
assert_eq!(progress, [(0, 0)]);
|
||||
let compiler = Compiler::new(&dir.path().join("cache"))?;
|
||||
let mut world = crate::world::World::open(dir.path())?;
|
||||
let root = Arc::new(sausage_vfs::Memory::default());
|
||||
root.mkdir("/bin").await?;
|
||||
let bytes =
|
||||
wat::parse_str(r#"(module (memory (export "memory") 1) (func (export "_start")))"#)?;
|
||||
std::fs::write(bin.join("one.wasm"), &bytes)?;
|
||||
std::fs::write(dir.path().join("world/usr/local/bin/two.wasm"), &bytes)?;
|
||||
std::fs::write(bin.join("README"), "Not a program")?;
|
||||
std::os::unix::fs::symlink("one.wasm", bin.join("one"))?;
|
||||
progress.clear();
|
||||
precompile_with_progress(world.root.as_ref(), |done, total| {
|
||||
progress.push((done, total))
|
||||
})
|
||||
.await?;
|
||||
assert_eq!(progress, [(0, 2), (1, 2), (2, 2)]);
|
||||
for (name, content) in [
|
||||
("one", bytes.clone()),
|
||||
("two", bytes),
|
||||
("unused", b"\0asm invalid".to_vec()),
|
||||
] {
|
||||
root.open(
|
||||
&format!("/bin/{name}.wasm"),
|
||||
sausage_vfs::Open {
|
||||
write: true,
|
||||
create: true,
|
||||
..Default::default()
|
||||
},
|
||||
)
|
||||
.await?
|
||||
.write(0, content)
|
||||
.await?;
|
||||
}
|
||||
world.root = root;
|
||||
let runtime =
|
||||
super::super::Runtime::with_compiler(world, compiler.clone(), Default::default())?;
|
||||
tokio::task::yield_now().await;
|
||||
assert!(compiler.programs.lock().unwrap().is_empty());
|
||||
assert_eq!(compiler.compilations.load(Ordering::Relaxed), 0);
|
||||
for program in ["/bin/one", "/bin/one", "/bin/two"] {
|
||||
assert_eq!(
|
||||
runtime
|
||||
.exec(
|
||||
Command::Argv(vec![program.into()]),
|
||||
Options::default(),
|
||||
&mut |_, _| {}
|
||||
)
|
||||
.await?,
|
||||
0
|
||||
);
|
||||
assert_eq!(compiler.compilations.load(Ordering::Relaxed), 1);
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn common_tools_precede_other_programs() {
|
||||
let mut names = [
|
||||
"z", "cat.wasm", "cargo", "sh.wasm", "git.wasm", "rg", "ls", "grep", "sed", "find",
|
||||
"arch",
|
||||
];
|
||||
names.sort_by_key(|name| priority(name));
|
||||
assert_eq!(
|
||||
names,
|
||||
[
|
||||
"sh.wasm", "git.wasm", "rg", "grep", "ls", "cat.wasm", "sed", "find", "cargo", "z",
|
||||
"arch"
|
||||
]
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test(flavor = "multi_thread")]
|
||||
async fn cold_compilation_does_not_spend_command_timeout() -> anyhow::Result<()> {
|
||||
use std::time::Duration;
|
||||
@@ -290,7 +205,6 @@ mod tests {
|
||||
let _ = finish_rx.recv();
|
||||
}));
|
||||
let runtime = super::super::Runtime::with_compiler(world, compiler, Default::default())?;
|
||||
started_rx.await?;
|
||||
let command_runtime = runtime.clone();
|
||||
let mut command = tokio::spawn(async move {
|
||||
command_runtime
|
||||
@@ -304,6 +218,7 @@ mod tests {
|
||||
)
|
||||
.await
|
||||
});
|
||||
started_rx.await?;
|
||||
let pending = tokio::time::timeout(Duration::from_millis(2100), &mut command).await;
|
||||
// Always release the blocking compiler, even when the old deadline fails this assertion.
|
||||
finish_tx.send(())?;
|
||||
@@ -456,7 +371,7 @@ mod tests {
|
||||
}
|
||||
|
||||
#[tokio::test(flavor = "multi_thread")]
|
||||
async fn command_joins_precompile_even_after_waiter_cancellation() {
|
||||
async fn concurrent_first_starts_share_compilation_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();
|
||||
@@ -471,13 +386,20 @@ mod tests {
|
||||
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(), Default::default())
|
||||
.unwrap();
|
||||
let first_runtime = runtime.clone();
|
||||
let first = tokio::spawn(async move {
|
||||
first_runtime
|
||||
.exec(
|
||||
Command::Argv(vec!["/bin/program".into()]),
|
||||
Options::default(),
|
||||
&mut |_, _| {},
|
||||
)
|
||||
.await
|
||||
});
|
||||
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()]),
|
||||
@@ -491,8 +413,8 @@ mod tests {
|
||||
.await;
|
||||
assert!(pending);
|
||||
assert_eq!(compiler.compilations.load(Ordering::Relaxed), 1);
|
||||
precompile.abort();
|
||||
let _ = precompile.await;
|
||||
first.abort();
|
||||
let _ = first.await;
|
||||
finish_tx.send(()).unwrap();
|
||||
assert_eq!(command.await.unwrap(), 0);
|
||||
assert_eq!(compiler.compilations.load(Ordering::Relaxed), 1);
|
||||
|
||||
@@ -55,7 +55,7 @@ pub mod net;
|
||||
pub fn default_script() -> Arc<dyn crate::cred::script::Script> {
|
||||
Arc::new(crate::cred::quickjs::QuickJs::default())
|
||||
}
|
||||
pub use cache::{configure as configure_cache, precompile_with_progress};
|
||||
pub use cache::configure as configure_cache;
|
||||
mod services;
|
||||
mod watcher;
|
||||
pub use watcher::Watcher;
|
||||
@@ -252,7 +252,6 @@ impl Runtime {
|
||||
entries: Mutex::new(HashMap::new()),
|
||||
next: AtomicU32::new(1),
|
||||
});
|
||||
cache::precompile(runtime.compiler.clone(), runtime.world.root.clone());
|
||||
Ok(runtime)
|
||||
}
|
||||
/// Access the shared network policy for explicit host-port rules.
|
||||
|
||||
@@ -62,11 +62,6 @@ impl World {
|
||||
);
|
||||
let world = Self::open_with_progress(dir, Some(bundle), &mut progress)
|
||||
.context("Create world folder and seed files failed")?;
|
||||
crate::process::precompile_with_progress(world.root.as_ref(), |done, total| {
|
||||
progress(&format!("Precompile tools {done}/{total}"));
|
||||
})
|
||||
.await
|
||||
.context("Precompile tools failed")?;
|
||||
progress("Create the first commit in /src/sausage");
|
||||
let exec = crate::process::exec(Self {
|
||||
root: world.root.clone(),
|
||||
|
||||
@@ -54,6 +54,7 @@ async fn init_seeds_config_tools_and_one_source_commit() -> anyhow::Result<()> {
|
||||
target.join("wasm32-wasip2/release/git.wasm"),
|
||||
bundle.join("bin/git.wasm"),
|
||||
)?;
|
||||
write(&bundle, "bin/unused.wasm", "not a component")?;
|
||||
write(&bundle, "src/sausage/README.md", "World source")?;
|
||||
write(
|
||||
&bundle,
|
||||
@@ -83,15 +84,13 @@ async fn init_seeds_config_tools_and_one_source_commit() -> anyhow::Result<()> {
|
||||
lines,
|
||||
[
|
||||
format!("Create world folder: {}", destination.display()),
|
||||
"Copy tools: 1".into(),
|
||||
"Copy tools: 2".into(),
|
||||
"Create /src/sausage".into(),
|
||||
"Seed /etc and /srv/www".into(),
|
||||
format!(
|
||||
"Write config: {}",
|
||||
destination.join("world/etc/sausage/config.yaml").display()
|
||||
),
|
||||
"Precompile tools 0/1".into(),
|
||||
"Precompile tools 1/1".into(),
|
||||
"Create the first commit in /src/sausage".into(),
|
||||
"Next steps:".into(),
|
||||
format!(
|
||||
@@ -149,10 +148,10 @@ fn init_errors_name_the_failed_step() -> anyhow::Result<()> {
|
||||
let bundle = dir.path().join("bundle");
|
||||
for (name, expected) in [
|
||||
("missing", "Copy tools failed"),
|
||||
("invalid", "Precompile tools failed"),
|
||||
("invalid", "Create source repository failed"),
|
||||
] {
|
||||
if name == "invalid" {
|
||||
write(&bundle, "bin/invalid.wasm", "not a component")?;
|
||||
write(&bundle, "bin/git.wasm", "not a component")?;
|
||||
write(&bundle, "src/sausage/README.md", "Source")?;
|
||||
}
|
||||
let output = std::process::Command::new(env!("CARGO_BIN_EXE_sausage"))
|
||||
@@ -166,9 +165,7 @@ fn init_errors_name_the_failed_step() -> anyhow::Result<()> {
|
||||
let stderr = String::from_utf8(output.stderr)?;
|
||||
assert!(stderr.contains(expected), "{stderr}");
|
||||
assert!(!stderr.contains("Next steps:"), "{stderr}");
|
||||
if name == "invalid" {
|
||||
assert!(stderr.contains("/bin/invalid.wasm"), "{stderr}");
|
||||
}
|
||||
assert!(!stderr.contains("Precompile"), "{stderr}");
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user