surma/sausage
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge main (SDK, agent) into wip-layout

Browse Source
This commit is contained in:
surma
2026-10-15 11:18:09 +01:00
parent 0a8e8c2e3c 7f76eab161
commit bb734a5c74
22 changed files with 607 additions and 518 deletions
Download Patch File Download Diff File Expand all files Collapse all files
examples/bench/Cargo.toml
+3 -1
View File
@@ -4,4 +4,6 @@ version = "0.1.0"
edition = "2024"
[dependencies]
serde_json = "1"
sausage-actor = { version = "0.1", path = "../../sdk" }
[workspace]
examples/bench/src/main.rs
+32 -39
View File
@@ -1,45 +1,38 @@
//! {"type":"bench","target":"echo","n":1000}: ping-pong with the target, one
//! message at a time, and report the round-trip times.
use serde_json::{Value, json};
use std::io::{BufRead, Write};
//! Ping-pong with a target and report the round-trip times.
use sausage_actor::{json, messages, reply, send};
use std::time::Instant;
fn main() {
let mut out = std::io::stdout().lock();
let mut lines = std::io::stdin().lock().lines();
while let Some(Ok(line)) = lines.next() {
let Ok(msg) = serde_json::from_str::<Value>(&line) else { continue };
match msg["body"]["type"].as_str() {
Some("shutdown") => break,
Some("bench") => {
let target = msg["body"]["target"].as_str().unwrap_or("echo").to_string();
let n = msg["body"]["n"].as_u64().unwrap_or(1000);
let mut times = Vec::new();
let started = Instant::now();
for i in 0..n {
let t = Instant::now();
writeln!(out, "{}", json!({"to": target, "id": i, "body": {"type": "ping", "i": i}})).unwrap();
out.flush().unwrap();
// Wait for the reply.
loop {
let Some(Ok(l)) = lines.next() else { return };
let Ok(r) = serde_json::from_str::<Value>(&l) else { continue };
if r["reply_to"].as_str().is_some_and(|s| s.ends_with(&format!("/{i}"))) {
break;
}
}
times.push(t.elapsed().as_micros() as u64);
}
let total = started.elapsed().as_secs_f64();
times.sort();
let p = |q: f64| times[((times.len() as f64 - 1.0) * q) as usize];
let reply = json!({"type": "bench_result", "n": n, "total_s": total,
"msgs_per_s": (n as f64 / total).round(),
"p50_us": p(0.5), "p90_us": p(0.9), "p99_us": p(0.99), "max_us": p(1.0)});
writeln!(out, "{}", json!({"to": msg["from"], "reply_to": msg["id"], "body": reply})).unwrap();
out.flush().unwrap();
}
_ => {}
let mut inbox = messages();
while let Some(message) = inbox.next() {
if message.kind() != "bench" {
continue;
}
let target = message.body["target"].as_str().unwrap_or("echo");
let n = message.body["n"].as_u64().unwrap_or(1000);
let mut times = Vec::new();
let started = Instant::now();
for i in 0..n {
let t = Instant::now();
let id = send(target, json!({"type": "ping", "i": i}));
loop {
let Some(reply) = inbox.next() else { return };
if reply.reply_to.as_deref() == Some(id.as_str()) {
break;
}
}
times.push(t.elapsed().as_micros() as u64);
}
let total = started.elapsed().as_secs_f64();
times.sort();
let p = |q: f64| times[((times.len() as f64 - 1.0) * q) as usize];
let _ = reply(
&message,
json!({
"type": "bench_result", "n": n, "total_s": total,
"msgs_per_s": (n as f64 / total).round(),
"p50_us": p(0.5), "p90_us": p(0.9), "p99_us": p(0.99), "max_us": p(1.0)
}),
);
}
}
examples/crasher/Cargo.toml
+2
View File
@@ -4,3 +4,5 @@ version = "0.1.0"
edition = "2024"
[dependencies]
[workspace]
examples/echo/Cargo.toml
+3 -1
View File
@@ -4,4 +4,6 @@ version = "0.1.0"
edition = "2024"
[dependencies]
serde_json = "1"
sausage-actor = { version = "0.1", path = "../../sdk" }
[workspace]
examples/echo/src/main.rs
+3 -11
View File
@@ -1,15 +1,7 @@
use serde_json::{json, Value};
use std::io::{BufRead, Write};
use sausage_actor::{json, messages, reply};
fn main() {
let mut out = std::io::stdout().lock();
for line in std::io::stdin().lock().lines() {
let Ok(msg) = serde_json::from_str::<Value>(&line.unwrap()) else { continue };
if msg["body"]["type"] == "shutdown" {
break;
}
let env = json!({"to": msg["from"], "reply_to": msg["id"], "body": {"type": "echo", "got": msg["body"]}});
writeln!(out, "{env}").unwrap();
out.flush().unwrap();
for message in messages() {
let _ = reply(&message, json!({"type": "echo", "got": message.body}));
}
}
examples/fetch/Cargo.toml
+3 -1
View File
@@ -4,5 +4,7 @@ version = "0.1.0"
edition = "2024"
[dependencies]
serde_json = "1"
sausage-actor = { version = "0.1", path = "../../sdk", features = ["async"] }
wstd = "0.6"
[workspace]
examples/fetch/src/main.rs
+37 -57
View File
@@ -1,77 +1,57 @@
//! Fetches URLs over HTTP(S) and writes the body to /public/fetch/<n>.
//! Reads stdin and runs requests at the same time: while a fetch runs,
//! `ping` still answers at once.
use serde_json::{Value, json};
use std::cell::RefCell;
use std::rc::Rc;
//! Fetch URLs over HTTP while answering pings.
use sausage_actor::{Inbox, Message, json, me, reply};
use wstd::http::{Body, Client, Request};
use wstd::io::{AsyncRead, AsyncWrite};
type Out = Rc<RefCell<Vec<String>>>;
async fn flush(out: &Out) {
let lines: Vec<String> = out.borrow_mut().drain(..).collect();
let mut stdout = wstd::io::stdout();
for l in lines {
let _ = stdout.write_all(format!("{l}\n").as_bytes()).await;
}
let _ = stdout.flush().await;
}
async fn fetch(n: u64, msg: Value, out: Out) {
let url = msg["body"]["url"].as_str().unwrap_or_default().to_string();
async fn fetch(n: u64, message: Message) {
let url = message.body["url"].as_str().unwrap_or_default();
let started = wstd::time::Instant::now();
let result = async {
let req = Request::get(&url).body(Body::empty()).map_err(|e| e.to_string())?;
let resp = Client::new().send(req).await.map_err(|e| format!("{e:?}"))?;
let status = resp.status().as_u16();
let bytes = resp.into_body().contents().await.map_err(|e| format!("{e:?}"))?.to_vec();
let request = Request::get(url)
.body(Body::empty())
.map_err(|e| e.to_string())?;
let response = Client::new()
.send(request)
.await
.map_err(|e| format!("{e:?}"))?;
let status = response.status().as_u16();
let bytes = response
.into_body()
.contents()
.await
.map_err(|e| format!("{e:?}"))?
.to_vec();
std::fs::create_dir_all("/public/fetch").map_err(|e| e.to_string())?;
let path = format!("/public/fetch/{n}");
std::fs::write(&path, &bytes).map_err(|e| e.to_string())?;
Ok::<_, String>((status, bytes.len(), path))
std::fs::write(path, &bytes).map_err(|e| e.to_string())?;
Ok::<_, String>((status, bytes.len()))
}
.await;
let ms = started.elapsed().as_millis();
let body = match result {
Ok((status, size, path)) => {
let id = std::env::var("SAUSAGE_ACTOR").unwrap_or("fetch".into());
json!({"type": "fetched", "status": status, "bytes": size, "path": path.replacen("/public", &format!("/actor/{id}/public"), 1), "ms": ms})
Ok((status, size)) => {
let id = me();
let actor = if id.is_empty() { "fetch" } else { &id };
json!({"type": "fetched", "status": status, "bytes": size, "path": format!("/actor/{actor}/public/fetch/{n}"), "ms": ms})
}
Err(e) => json!({"type": "error", "reason": e, "ms": ms}),
Err(reason) => json!({"type": "error", "reason": reason, "ms": ms}),
};
out.borrow_mut().push(json!({"to": msg["from"], "reply_to": msg["id"], "body": body}).to_string());
flush(&out).await;
let _ = reply(&message, body);
}
fn main() {
wstd::runtime::block_on(async {
let out: Out = Rc::new(RefCell::new(Vec::new()));
let mut stdin = wstd::io::stdin();
let mut buf = Vec::new();
let mut chunk = vec![0u8; 8192];
let mut n = 0u64;
loop {
let Ok(len) = stdin.read(&mut chunk).await else { break };
if len == 0 {
break;
}
buf.extend_from_slice(&chunk[..len]);
while let Some(pos) = buf.iter().position(|b| *b == b'\n') {
let line: Vec<u8> = buf.drain(..=pos).collect();
let Ok(msg) = serde_json::from_slice::<Value>(&line) else { continue };
match msg["body"]["type"].as_str() {
Some("shutdown") => return,
Some("fetch") => {
n += 1;
wstd::runtime::spawn(fetch(n, msg, out.clone())).detach();
}
Some("ping") => {
out.borrow_mut().push(json!({"to": msg["from"], "reply_to": msg["id"], "body": {"type": "pong"}}).to_string());
flush(&out).await;
}
_ => {}
let inbox = Inbox::start();
let mut n = 0;
while let Some(message) = inbox.recv().await {
match message.kind() {
"fetch" => {
n += 1;
wstd::runtime::spawn(fetch(n, message)).detach();
}
"ping" => {
let _ = reply(&message, json!({"type": "pong"}));
}
_ => {}
}
}
});
examples/fstest/Cargo.toml
+3 -1
View File
@@ -4,4 +4,6 @@ version = "0.1.0"
edition = "2024"
[dependencies]
serde_json = "1"
sausage-actor = { version = "0.1", path = "../../sdk" }
[workspace]
examples/fstest/src/main.rs
+24 -33
View File
@@ -1,47 +1,38 @@
//! Filesystem probe for the spike. Keeps files open across messages.
use serde_json::{Value, json};
use std::collections::HashMap;
use std::io::{BufRead, Write};
//! Filesystem probe. Keeps files open across messages.
use sausage_actor::{Value, json, messages, reply};
use std::{collections::HashMap, io::Write};
fn main() {
let mut open: HashMap<String, std::fs::File> = HashMap::new();
let mut out = std::io::stdout().lock();
for line in std::io::stdin().lock().lines() {
let Ok(line) = line else { break };
let Ok(msg) = serde_json::from_str::<Value>(&line) else { continue };
let b = &msg["body"];
let path = b["path"].as_str().unwrap_or_default().to_string();
let r = |res: std::io::Result<Value>| match res {
Ok(v) => v,
Err(e) => json!({"type": "error", "error": e.to_string()}),
};
let reply = match b["type"].as_str() {
Some("shutdown") => break,
Some("write") => r(std::fs::write(&path, b["text"].as_str().unwrap_or_default()).map(|_| json!({"type": "ok"}))),
Some("open") => r((|| {
let mut f = std::fs::File::create(&path)?;
f.write_all(b["text"].as_str().unwrap_or_default().as_bytes())?;
open.insert(path.clone(), f);
let mut open = HashMap::<String, std::fs::File>::new();
for msg in messages() {
let b = &msg.body;
let path = b["path"].as_str().unwrap_or_default();
let r = |result: std::io::Result<Value>| result.unwrap_or_else(|e| json!({"type": "error", "error": e.to_string()}));
let response = match msg.kind() {
"write" => r(std::fs::write(path, b["text"].as_str().unwrap_or_default()).map(|_| json!({"type": "ok"}))),
"open" => r((|| {
let mut file = std::fs::File::create(path)?;
file.write_all(b["text"].as_str().unwrap_or_default().as_bytes())?;
open.insert(path.into(), file);
Ok(json!({"type": "ok", "open": true}))
})()),
Some("append") => r((|| {
let mut f = std::fs::OpenOptions::new().append(true).open(&path)?;
f.write_all(b["text"].as_str().unwrap_or_default().as_bytes())?;
open.insert(path.clone(), f);
"append" => r((|| {
let mut file = std::fs::OpenOptions::new().append(true).open(path)?;
file.write_all(b["text"].as_str().unwrap_or_default().as_bytes())?;
open.insert(path.into(), file);
Ok(json!({"type": "ok", "open": true}))
})()),
Some("close") => json!({"type": "ok", "closed": open.remove(&path).is_some()}),
Some("read") => r(std::fs::read_to_string(&path).map(|t| json!({"type": "content", "text": t}))),
Some("list") => r(std::fs::read_dir(&path).map(|rd| {
"close" => json!({"type": "ok", "closed": open.remove(path).is_some()}),
"read" => r(std::fs::read_to_string(path).map(|text| json!({"type": "content", "text": text}))),
"list" => r(std::fs::read_dir(path).map(|rd| {
let mut names: Vec<String> = rd.filter_map(|e| e.ok()).map(|e| e.file_name().to_string_lossy().to_string()).collect();
names.sort();
json!({"type": "entries", "names": names})
})),
Some("remove") => r(std::fs::remove_file(&path).map(|_| json!({"type": "ok"}))),
Some("crash") => panic!("asked to crash"),
"remove" => r(std::fs::remove_file(path).map(|_| json!({"type": "ok"}))),
"crash" => panic!("asked to crash"),
_ => json!({"type": "error", "error": "unknown message"}),
};
writeln!(out, "{}", json!({"to": msg["from"], "reply_to": msg["id"], "body": reply})).unwrap();
out.flush().unwrap();
let _ = reply(&msg, response);
}
}
examples/notes-v2/Cargo.toml
+3
View File
@@ -4,4 +4,7 @@ version = "0.1.0"
edition = "2024"
[dependencies]
sausage-actor = { version = "0.1", path = "../../sdk" }
serde_json = "1"
[workspace]
examples/notes-v2/src/main.rs
+11 -19
View File
@@ -1,7 +1,5 @@
//! Keeps a list of notes in /data/notes.json and publishes them as
//! /public/notes.txt. Version 2: adds `count` and no longer crashes.
use serde_json::{Value, json};
use std::io::{BufRead, Write};
//! Version 2 adds a note count and fixes the crash command.
use sausage_actor::{json, messages, reply};
const DATA: &str = "/data/notes.json";
@@ -13,27 +11,21 @@ fn save(notes: &[String]) {
fn main() {
let mut notes: Vec<String> = std::fs::read_to_string(DATA)
.ok()
.and_then(|s| serde_json::from_str(&s).ok())
.and_then(|text| serde_json::from_str(&text).ok())
.unwrap_or_default();
let mut out = std::io::stdout().lock();
for line in std::io::stdin().lock().lines() {
let Ok(line) = line else { break };
let Ok(msg) = serde_json::from_str::<Value>(&line) else { continue };
let body = &msg["body"];
let reply = match body["type"].as_str() {
Some("shutdown") => break,
Some("add") => {
for message in messages() {
let body = &message.body;
let response = match message.kind() {
"add" => {
notes.push(body["text"].as_str().unwrap_or_default().to_string());
save(&notes);
json!({"type": "added", "count": notes.len()})
}
Some("list") => json!({"type": "notes", "items": notes}),
Some("count") => json!({"type": "count", "count": notes.len(), "version": 2}),
// Version 2 fixes the crash.
Some("crash") => json!({"type": "not_crashing", "version": 2}),
"list" => json!({"type": "notes", "items": notes}),
"count" => json!({"type": "count", "count": notes.len(), "version": 2}),
"crash" => json!({"type": "not_crashing", "version": 2}),
_ => json!({"type": "error", "reason": "unknown message"}),
};
writeln!(out, "{}", json!({"to": msg["from"], "reply_to": msg["id"], "body": reply})).unwrap();
out.flush().unwrap();
let _ = reply(&message, response);
}
}
examples/notes/Cargo.toml
+3
View File
@@ -4,4 +4,7 @@ version = "0.1.0"
edition = "2024"
[dependencies]
sausage-actor = { version = "0.1", path = "../../sdk" }
serde_json = "1"
[workspace]
examples/notes/src/main.rs
+10 -17
View File
@@ -1,7 +1,5 @@
//! Keeps a list of notes in /data/notes.json and publishes them as
//! /public/notes.txt.
use serde_json::{Value, json};
use std::io::{BufRead, Write};
//! Keep notes in `/data/notes.json` and publish them at `/public/notes.txt`.
use sausage_actor::{json, messages, reply};
const DATA: &str = "/data/notes.json";
@@ -13,25 +11,20 @@ fn save(notes: &[String]) {
fn main() {
let mut notes: Vec<String> = std::fs::read_to_string(DATA)
.ok()
.and_then(|s| serde_json::from_str(&s).ok())
.and_then(|text| serde_json::from_str(&text).ok())
.unwrap_or_default();
let mut out = std::io::stdout().lock();
for line in std::io::stdin().lock().lines() {
let Ok(line) = line else { break };
let Ok(msg) = serde_json::from_str::<Value>(&line) else { continue };
let body = &msg["body"];
let reply = match body["type"].as_str() {
Some("shutdown") => break,
Some("add") => {
for message in messages() {
let body = &message.body;
let response = match message.kind() {
"add" => {
notes.push(body["text"].as_str().unwrap_or_default().to_string());
save(&notes);
json!({"type": "added", "count": notes.len()})
}
Some("list") => json!({"type": "notes", "items": notes}),
Some("crash") => panic!("asked to crash"),
"list" => json!({"type": "notes", "items": notes}),
"crash" => panic!("asked to crash"),
_ => json!({"type": "error", "reason": "unknown message"}),
};
writeln!(out, "{}", json!({"to": msg["from"], "reply_to": msg["id"], "body": reply})).unwrap();
out.flush().unwrap();
let _ = reply(&message, response);
}
}
examples/ticker/Cargo.toml
+3 -1
View File
@@ -4,4 +4,6 @@ version = "0.1.0"
edition = "2024"
[dependencies]
serde_json = "1"
sausage-actor = { version = "0.1", path = "../../sdk" }
[workspace]
examples/ticker/src/main.rs
+7 -10
View File
@@ -1,16 +1,13 @@
//! Sends {"type":"add","text":"tick N"} to `notes` every 50 ms, N = 1..=200.
//! Does not read stdin, so the broker kills it at shutdown; that is fine.
use std::io::Write;
//! Send 200 notes to `notes`, 50 ms apart, then stay alive.
use sausage_actor::{json, send};
use std::time::Duration;
fn main() {
let mut out = std::io::stdout().lock();
for n in 1..=200 {
let msg = serde_json::json!({"to": "notes", "body": {"type": "add", "text": format!("tick {n}")}});
writeln!(out, "{msg}").unwrap();
out.flush().unwrap();
std::thread::sleep(std::time::Duration::from_millis(50));
let _ = send("notes", json!({"type": "add", "text": format!("tick {n}")}));
std::thread::sleep(Duration::from_millis(50));
}
// Stay alive without doing anything.
loop {
std::thread::sleep(std::time::Duration::from_secs(3600));
std::thread::sleep(Duration::from_secs(3600));
}
}