diff --git a/.gitignore b/.gitignore index 0d4a6a2..1c1d1dd 100644 --- a/.gitignore +++ b/.gitignore @@ -3,3 +3,6 @@ sim.db sim.db-shm sim.db-wal +sim-shard-*.db +sim-shard-*.db-shm +sim-shard-*.db-wal diff --git a/CHANGELOG.md b/CHANGELOG.md index 6aa6fbe..62deb9a 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -14,10 +14,14 @@ Notable changes to this project are documented here. - Carts are actors that add carrying capacity, wear out and can be bought at the market. - Merchants carry their market's surplus to markets on other shards and bring back bargains. - People earn and spend coins; workers buy tools, meals and food for the village. +- The server runs as a gateway that starts `SIM_SHARDS` (default 3) shard processes, steps them in lockstep and shows them to the UI as one world. Shards exchange entities and messages directly over WebSockets. +- Every shard keeps its own database, configured with `SHARD_DATABASE_URL` (default `sqlite://sim-shard-{shard}.db?mode=rwc`). +- The Picard console reaches the actors of the other shards as `@shard-N` (`peers`, `@shard-1 actors`, `send @shard-1 PATH JSON`); `/api/actors/ws?shard=N` connects to shard `N` directly. ### Changed - People, rabbits, wolves, and partnerships are Picard actors that own their state; the world only reaches them through messages. Actors are named `person-`, `rabbit-`, `wolf-`, and `partnership-` instead of `entity-`. - Rabbits killed by hunters no longer move or breed later in the same tick. -- Update Picard to `054449c`. +- Update Picard to `f88b9cb`. - The default world is larger: 192×128 tiles with 180 people, 1200 rabbits, 80 wolves, and twice the food and houses. +- `DATABASE_URL` is replaced by `SHARD_DATABASE_URL`; existing `sim.db` snapshots are not migrated. diff --git a/Cargo.lock b/Cargo.lock index a5e7dc2..5fa78ac 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1939,7 +1939,7 @@ checksum = "9b4f627cb1b25917193a259e49bdad08f671f8d9708acfd5fe0a8c1455d87220" [[package]] name = "picard" version = "0.4.2" -source = "git+https://github.com/picard-rs/picard?rev=054449c9ab5a634c24fca81bcd4f9b20bbdc60b7#054449c9ab5a634c24fca81bcd4f9b20bbdc60b7" +source = "git+https://github.com/picard-rs/picard?rev=f88b9cbf078512f726c4c47dbce842fda878388e#f88b9cbf078512f726c4c47dbce842fda878388e" dependencies = [ "better_default", "chrono", @@ -1968,7 +1968,7 @@ dependencies = [ [[package]] name = "picard-macros" version = "0.2.0" -source = "git+https://github.com/picard-rs/picard?rev=054449c9ab5a634c24fca81bcd4f9b20bbdc60b7#054449c9ab5a634c24fca81bcd4f9b20bbdc60b7" +source = "git+https://github.com/picard-rs/picard?rev=f88b9cbf078512f726c4c47dbce842fda878388e#f88b9cbf078512f726c4c47dbce842fda878388e" dependencies = [ "proc-macro2", "quote", diff --git a/Cargo.toml b/Cargo.toml index 641cb8a..015d86e 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -25,7 +25,7 @@ serde = { version = "1", features = ["derive"] } serde_json = "1" sqlx = { version = "0.8", default-features = false, features = ["runtime-tokio-rustls", "any", "sqlite", "postgres"] } rand = "0.8" -picard = { git = "https://github.com/picard-rs/picard", rev = "054449c9ab5a634c24fca81bcd4f9b20bbdc60b7", features = [ +picard = { git = "https://github.com/picard-rs/picard", rev = "f88b9cbf078512f726c4c47dbce842fda878388e", features = [ "remoting", "stats", "tracing", diff --git a/src/cluster.rs b/src/cluster.rs new file mode 100644 index 0000000..133e7fc --- /dev/null +++ b/src/cluster.rs @@ -0,0 +1,6 @@ +//! Runs the world as several shard processes behind one gateway. + +pub mod gateway; +pub mod protocol; +pub mod shard; +pub mod transport; diff --git a/src/cluster/gateway.rs b/src/cluster/gateway.rs new file mode 100644 index 0000000..d574257 --- /dev/null +++ b/src/cluster/gateway.rs @@ -0,0 +1,590 @@ +//! The gateway: starts the shard nodes, steps them in lockstep and presents their parts of the +//! world to the UI as one world. + +use super::protocol::{Control, ControlReply, Peer}; +use super::shard::LISTENING; +use crate::world::{EntityKey, Quote, Shard, regenerated_seed}; +use serde_json::{Value, json}; +use std::io::{BufRead, BufReader}; +use std::net::TcpStream; +use std::process::{Child, Command, Stdio}; +use std::sync::{Arc, Mutex}; +use std::time::{Duration, Instant}; +use tokio::sync::{mpsc, oneshot, watch}; +use tungstenite::stream::MaybeTlsStream; +use tungstenite::{Message, WebSocket}; +use uuid::Uuid; + +const FRAME: Duration = Duration::from_millis(100); +const SAVE_INTERVAL: Duration = Duration::from_secs(1); +const SHARD_TIMEOUT: Duration = Duration::from_secs(120); + +pub enum Msg { + Edit(String), + Script(String, oneshot::Sender>), + #[cfg(feature = "console")] + PicardConsole { + request: String, + shard: usize, + reply: oneshot::Sender, + }, + Restart { + regenerate: bool, + reply: oneshot::Sender, + }, +} + +pub struct GatewayOptions { + pub shards: usize, + pub seed: u64, + /// Database URL of each shard; `{shard}` is replaced by the shard's index. + pub database_template: String, +} + +/// Where the gateway publishes the unified world for the HTTP handlers. +pub struct Outputs { + pub terrain: Arc>, + pub script: Arc>, + pub state: watch::Sender>, +} + +/// Runs the cluster until the process exits; a failing shard takes the whole cluster down. +pub fn conduct( + options: &GatewayOptions, + outputs: &Outputs, + mut messages: mpsc::UnboundedReceiver, +) { + let result = Cluster::start(options).and_then(|mut cluster| { + let result = cluster.run(outputs, &mut messages); + drop(cluster); + result + }); + if let Err(error) = result { + tracing::error!(%error, "the cluster stopped"); + std::process::exit(1); + } +} + +struct ShardProcess { + child: Child, + address: String, +} + +impl ShardProcess { + fn start(index: usize, options: &GatewayOptions) -> Result { + let executable = std::env::current_exe().map_err(|e| e.to_string())?; + let database = options + .database_template + .replace("{shard}", &index.to_string()); + let mut child = Command::new(executable) + .args(["shard", "--attached"]) + .args(["--index", &index.to_string()]) + .args(["--count", &options.shards.to_string()]) + .args(["--seed", &options.seed.to_string()]) + .args(["--database", &database]) + .stdin(Stdio::piped()) + .stdout(Stdio::piped()) + .stderr(Stdio::inherit()) + .spawn() + .map_err(|e| format!("failed to start shard {index}: {e}"))?; + let stdout = child.stdout.take().expect("piped shard stdout"); + let mut lines = BufReader::new(stdout).lines(); + let address = loop { + match lines.next() { + Some(Ok(line)) => match line.strip_prefix(LISTENING) { + Some(address) => break address.trim().to_owned(), + None => println!("[shard {index}] {line}"), + }, + _ => { + let _ = child.kill(); + return Err(format!("shard {index} exited before it was ready")); + } + } + }; + std::thread::spawn(move || { + for line in lines.map_while(Result::ok) { + println!("[shard {index}] {line}"); + } + }); + Ok(Self { child, address }) + } +} + +impl Drop for ShardProcess { + fn drop(&mut self) { + let _ = self.child.kill(); + let _ = self.child.wait(); + } +} + +type Socket = WebSocket>; + +/// The gateway's `/control` connection to one shard; commands are answered in order. +struct ShardLink { + index: usize, + socket: Socket, +} + +impl ShardLink { + fn connect(index: usize, address: &str) -> Result { + let url = format!("ws://{address}/control"); + let (socket, _) = tungstenite::connect(&url) + .map_err(|e| format!("failed to connect to shard {index} at {url}: {e}"))?; + if let MaybeTlsStream::Plain(stream) = socket.get_ref() { + let _ = stream.set_nodelay(true); + let _ = stream.set_read_timeout(Some(SHARD_TIMEOUT)); + } + Ok(Self { index, socket }) + } + + fn send(&mut self, control: &Control) -> Result<(), String> { + let frame = serde_json::to_string(control).map_err(|e| e.to_string())?; + self.socket + .send(Message::Text(frame.into())) + .map_err(|e| format!("lost shard {}: {e}", self.index)) + } + + fn receive(&mut self) -> Result { + loop { + match self.socket.read() { + Ok(Message::Text(text)) => { + return match serde_json::from_str(text.as_str()) { + Ok(ControlReply::Failed(error)) => { + Err(format!("shard {} failed: {error}", self.index)) + } + Ok(reply) => Ok(reply), + Err(error) => Err(format!( + "unreadable reply from shard {}: {error}", + self.index + )), + }; + } + Ok(Message::Ping(_) | Message::Pong(_)) => {} + Ok(_) => return Err(format!("shard {} closed the connection", self.index)), + Err(error) => return Err(format!("lost shard {}: {error}", self.index)), + } + } + } + + fn call(&mut self, control: &Control) -> Result { + self.send(control)?; + self.receive() + } +} + +struct Ready { + system_id: Uuid, + seed: u64, + tick: u64, + restored: bool, +} + +fn ready(reply: ControlReply) -> Result { + match reply { + ControlReply::Ready { + system_id, + seed, + tick, + restored, + } => Ok(Ready { + system_id, + seed, + tick, + restored, + }), + other => Err(unexpected(&other)), + } +} + +fn unexpected(reply: &ControlReply) -> String { + format!("unexpected reply from a shard: {reply:?}") +} + +struct Cluster { + links: Vec, + processes: Vec, + seed: u64, + /// Directory changes to publish with the next step. + directory: Vec<(EntityKey, Option)>, + board: Vec, + paused: bool, + speed: u64, + width: usize, + terrain_rev: Option, +} + +impl Cluster { + fn start(options: &GatewayOptions) -> Result { + let processes = (0..options.shards) + .map(|index| ShardProcess::start(index, options)) + .collect::, _>>()?; + let links = processes + .iter() + .enumerate() + .map(|(index, process)| ShardLink::connect(index, &process.address)) + .collect::, _>>()?; + let mut cluster = Self { + links, + processes, + seed: options.seed, + directory: Vec::new(), + board: Vec::new(), + paused: false, + speed: 1, + width: 0, + terrain_rev: None, + }; + let readies = cluster + .broadcast(|_| Control::Hello)? + .into_iter() + .map(ready) + .collect::, _>>()?; + let consistent = readies + .windows(2) + .all(|pair| pair[0].tick == pair[1].tick && pair[0].seed == pair[1].seed); + let restored = readies.iter().filter(|ready| ready.restored).count(); + if consistent && (restored == 0 || restored == readies.len()) { + cluster.seed = readies[0].seed; + cluster.connect_peers(&readies)?; + } else { + tracing::warn!("the shards' saved states don't match; starting a new world"); + cluster.restart(options.seed)?; + } + Ok(cluster) + } + + /// Sends one command to every shard before waiting for any answer, so they work in parallel. + fn broadcast( + &mut self, + control: impl Fn(usize) -> Control, + ) -> Result, String> { + for link in &mut self.links { + link.send(&control(link.index))?; + } + self.links.iter_mut().map(ShardLink::receive).collect() + } + + fn broadcast_done(&mut self, control: impl Fn(usize) -> Control) -> Result<(), String> { + for reply in self.broadcast(control)? { + if !matches!(reply, ControlReply::Done) { + return Err(unexpected(&reply)); + } + } + Ok(()) + } + + fn restart(&mut self, seed: u64) -> Result<(), String> { + let readies = self + .broadcast(|_| Control::Restart { seed })? + .into_iter() + .map(ready) + .collect::, _>>()?; + self.seed = seed; + self.terrain_rev = None; + self.connect_peers(&readies) + } + + /// Tells every shard where the others are and publishes where every entity lives. + fn connect_peers(&mut self, readies: &[Ready]) -> Result<(), String> { + let peers: Vec = readies + .iter() + .zip(&self.processes) + .map(|(ready, process)| Peer { + address: process.address.clone(), + system_id: ready.system_id, + }) + .collect(); + self.broadcast_done(|_| Control::Peers(peers.clone()))?; + self.directory.clear(); + for (index, reply) in self.broadcast(|_| Control::Roster)?.into_iter().enumerate() { + let ControlReply::Roster(keys) = reply else { + return Err(unexpected(&reply)); + }; + self.directory + .extend(keys.into_iter().map(|key| (key, Some(index)))); + } + self.board.clear(); + Ok(()) + } + + fn step(&mut self) -> Result<(), String> { + let directory = std::mem::take(&mut self.directory); + let board = std::mem::take(&mut self.board); + let replies = self.broadcast(|_| Control::Step { + directory: directory.clone(), + board: board.clone(), + })?; + let mut gone = Vec::new(); + for (index, reply) in replies.into_iter().enumerate() { + let ControlReply::Stepped { report, quotes } = reply else { + return Err(unexpected(&reply)); + }; + self.board.extend(quotes); + self.directory + .extend(report.moved.into_iter().map(|(key, to)| (key, Some(to)))); + self.directory + .extend(report.arrived.into_iter().map(|key| (key, Some(index)))); + gone.extend(report.left.into_iter().map(|key| (key, None))); + } + // Departures last, so that an entity that moved and died in the same step stays gone. + self.directory.extend(gone); + Ok(()) + } + + fn run( + &mut self, + outputs: &Outputs, + messages: &mut mpsc::UnboundedReceiver, + ) -> Result<(), String> { + let mut last_saved = Instant::now(); + self.publish(outputs)?; + loop { + let frame_started = Instant::now(); + while let Ok(message) = messages.try_recv() { + self.handle(message, outputs)?; + } + if !self.paused { + for _ in 0..self.speed { + self.step()?; + } + } + self.publish(outputs)?; + if last_saved.elapsed() >= SAVE_INTERVAL { + self.broadcast_done(|_| Control::Save)?; + last_saved = Instant::now(); + } + std::thread::sleep(FRAME.saturating_sub(frame_started.elapsed())); + } + } + + fn handle(&mut self, message: Msg, outputs: &Outputs) -> Result<(), String> { + match message { + Msg::Edit(command) => { + for (index, command) in route_edit(&command, self.links.len()) { + let reply = self.links[index].call(&Control::Edit(command))?; + if !matches!(reply, ControlReply::Done) { + return Err(unexpected(&reply)); + } + } + } + Msg::Script(source, reply) => { + let mut result = Ok(()); + for answer in self.broadcast(|_| Control::Script(source.clone()))? { + let ControlReply::Script(answer) = answer else { + return Err(unexpected(&answer)); + }; + if result.is_ok() { + result = answer; + } + } + if result.is_ok() { + *outputs.script.lock().expect("script lock") = source; + } + let _ = reply.send(result); + } + #[cfg(feature = "console")] + Msg::PicardConsole { + request, + shard, + reply, + } => { + let response = match self.links.get_mut(shard) { + Some(link) => match link.call(&Control::Console(request))? { + ControlReply::Console(response) => response, + other => return Err(unexpected(&other)), + }, + None => json!({ + "id": null, + "error": {"code": "invalid_request", "message": "no such shard"} + }) + .to_string(), + }; + let _ = reply.send(response); + } + Msg::Restart { regenerate, reply } => { + let seed = if regenerate { + regenerated_seed(self.seed) + } else { + self.seed + }; + self.restart(seed)?; + let _ = reply.send(seed); + } + } + Ok(()) + } + + /// Merges the shards' views into one world for the UI. + fn publish(&mut self, outputs: &Outputs) -> Result<(), String> { + let mut states = Vec::with_capacity(self.links.len()); + for reply in self.broadcast(|_| Control::View { terrain: false })? { + let ControlReply::View { state, .. } = reply else { + return Err(unexpected(&reply)); + }; + states.push(serde_json::from_str::(&state).map_err(|e| e.to_string())?); + } + let first = &states[0]; + self.paused = first["paused"].as_bool().unwrap_or(false); + self.speed = first["speed"].as_u64().unwrap_or(1); + let rev = first["rev"].as_u64(); + if rev != self.terrain_rev { + let reply = self.links[0].call(&Control::View { terrain: true })?; + let ControlReply::View { + terrain: Some(terrain), + .. + } = reply + else { + return Err(unexpected(&reply)); + }; + let parsed: Value = serde_json::from_str(&terrain).map_err(|e| e.to_string())?; + self.width = parsed["w"] + .as_u64() + .and_then(|w| usize::try_from(w).ok()) + .unwrap_or(0); + *outputs.terrain.lock().expect("terrain lock") = terrain; + self.terrain_rev = rev; + } + let merged = merge_views(&states, self.width); + outputs.state.send_replace(Arc::from(merged.to_string())); + Ok(()) + } +} + +/// Which shards an edit command goes to: people are found through shard 0's router, stock is +/// shared out, and everything else (including terrain, pause and speed) applies everywhere. +fn route_edit(command: &str, shards: usize) -> Vec<(usize, String)> { + let words: Vec<&str> = command.split_whitespace().collect(); + match words.as_slice() { + ["person", ..] => vec![(0, command.to_owned())], + ["stock", name, value] => { + let total = value + .parse::() + .map_or(0, |v| usize::try_from(v.max(0)).unwrap_or(usize::MAX)); + (0..shards) + .filter_map(|index| Shard::new(index, shards)) + .map(|shard| (shard.index, format!("stock {name} {}", shard.share(total)))) + .collect() + } + _ => (0..shards) + .map(|index| (index, command.to_owned())) + .collect(), + } +} + +/// Combines the shards' state JSON: settings come from shard 0, entities are concatenated, +/// totals are summed, and `homes` and `shards` tell the UI where each shard lives. +fn merge_views(states: &[Value], width: usize) -> Value { + let Some(Value::Object(first)) = states.first() else { + return json!({}); + }; + let mut merged = first.clone(); + for (key, value) in &mut merged { + match key.as_str() { + "home" | "log" => {} + "stock" | "actor_stats" => { + if let Value::Object(totals) = value { + for (name, total) in totals.iter_mut() { + let sum: u64 = states + .iter() + .filter_map(|state| state[key.as_str()][name.as_str()].as_u64()) + .sum(); + *total = sum.into(); + } + } + } + _ => { + if let Value::Array(items) = value { + for state in &states[1..] { + if let Some(more) = state[key.as_str()].as_array() { + items.extend(more.iter().cloned()); + } + } + } + } + } + } + let limit = states + .iter() + .filter_map(|state| state["log"].as_array().map(Vec::len)) + .max() + .unwrap_or(0); + let mut log: Vec = states + .iter() + .filter_map(|state| state["log"].as_array()) + .flatten() + .cloned() + .collect(); + log.sort_by_key(|entry| entry.as_str().and_then(log_day).unwrap_or(0)); + let log = log.split_off(log.len().saturating_sub(limit)); + merged.insert("log".into(), Value::Array(log)); + merged.insert( + "homes".into(), + states.iter().map(|state| state["home"].clone()).collect(), + ); + let layout = Shard::new(0, states.len()).unwrap_or(Shard::SINGLE); + merged.insert( + "shards".into(), + (0..states.len()) + .map(|index| { + let columns = layout.columns_of(index, width); + json!([columns.start, columns.end]) + }) + .collect(), + ); + Value::Object(merged) +} + +fn log_day(entry: &str) -> Option { + entry.strip_prefix("[day ")?.split(']').next()?.parse().ok() +} + +#[cfg(test)] +mod tests { + use super::{merge_views, route_edit}; + use serde_json::json; + + #[test] + fn views_merge_into_one_world() { + let left = json!({ + "paused": false, "speed": 2, "tick": 10, "home": [5, 5], + "stock": {"food": 3, "wood": 1}, + "actor_stats": {"live_actors": 4}, + "people": [[1, 5, 5]], "markets": [[1, 6, 6]], + "log": ["[day 1] a", "[day 3] c"] + }); + let right = json!({ + "paused": true, "speed": 9, "tick": 10, "home": [50, 5], + "stock": {"food": 4, "wood": 2}, + "actor_stats": {"live_actors": 6}, + "people": [[2, 50, 5], [4, 51, 5]], "markets": [], + "log": ["[day 2] b"] + }); + let merged = merge_views(&[left, right], 96); + assert_eq!(merged["speed"], 2); + assert_eq!(merged["paused"], false); + assert_eq!(merged["stock"], json!({"food": 7, "wood": 3})); + assert_eq!(merged["actor_stats"]["live_actors"], 10); + assert_eq!(merged["people"].as_array().unwrap().len(), 3); + assert_eq!(merged["markets"].as_array().unwrap().len(), 1); + assert_eq!(merged["log"], json!(["[day 2] b", "[day 3] c"])); + assert_eq!(merged["homes"], json!([[5, 5], [50, 5]])); + assert_eq!(merged["shards"], json!([[0, 48], [48, 96]])); + } + + #[test] + fn edits_go_where_they_apply() { + assert_eq!( + route_edit("person 4 coins 9", 3), + vec![(0, "person 4 coins 9".into())] + ); + assert_eq!( + route_edit("stock food 10", 3), + vec![ + (0, "stock food 4".into()), + (1, "stock food 3".into()), + (2, "stock food 3".into()) + ] + ); + assert_eq!(route_edit("pause 1", 2).len(), 2); + } +} diff --git a/src/cluster/protocol.rs b/src/cluster/protocol.rs new file mode 100644 index 0000000..47ae335 --- /dev/null +++ b/src/cluster/protocol.rs @@ -0,0 +1,64 @@ +//! Messages between the gateway and its shard nodes on the `/control` WebSocket. + +use crate::world::{EntityKey, Quote, StepReport}; +use serde::{Deserialize, Serialize}; +use uuid::Uuid; + +/// Where a shard node accepts entity messages from the other shards. +#[derive(Clone, Debug, Deserialize, Serialize, PartialEq, Eq)] +pub struct Peer { + /// The `host:port` of the shard node, which serves `/remote` and `/console`. + pub address: String, + /// The current actor system of the shard, which remote requests are addressed to. + pub system_id: Uuid, +} + +#[derive(Debug, Deserialize, Serialize)] +pub enum Control { + /// Asks a freshly connected shard what it is running. + Hello, + /// Connects the shard to the other shards; sent again whenever any of them restarts. + Peers(Vec), + /// Runs one step after applying the gateway's directory changes and market board. + Step { + directory: Vec<(EntityKey, Option)>, + board: Vec, + }, + /// Asks for the shard's part of the world, and for the terrain when `terrain` is set. + View { + terrain: bool, + }, + /// Takes in admitted entities and persists a snapshot. + Save, + /// Every entity the shard simulates, for the gateway's directory. + Roster, + Edit(String), + Script(String), + Restart { + seed: u64, + }, + Console(String), +} + +#[derive(Debug, Deserialize, Serialize)] +pub enum ControlReply { + Ready { + system_id: Uuid, + seed: u64, + tick: u64, + restored: bool, + }, + Done, + Stepped { + report: StepReport, + quotes: Vec, + }, + View { + state: String, + terrain: Option, + }, + Roster(Vec), + Script(Result<(), String>), + Console(String), + Failed(String), +} diff --git a/src/cluster/shard.rs b/src/cluster/shard.rs new file mode 100644 index 0000000..036feff --- /dev/null +++ b/src/cluster/shard.rs @@ -0,0 +1,330 @@ +//! A shard node: simulates one strip of the world, answers the gateway on `/control` and takes +//! entity messages from the other shards on `/remote`. + +use super::protocol::{Control, ControlReply, Peer}; +use super::transport::{self, ClusterTransport}; +use crate::storage::Storage; +use crate::world::{Router, Shard, SimScript, World}; +use axum::extract::State; +use axum::extract::ws::{Message, WebSocket, WebSocketUpgrade}; +use axum::response::Response; +use axum::routing::get; +use std::io::Write; +use std::sync::{Arc, RwLock}; +use std::time::Duration; +use tokio::sync::{mpsc, oneshot, watch}; + +/// The line a shard prints to stdout once it accepts connections, followed by its address. +pub const LISTENING: &str = "SHARD_LISTENING"; + +pub struct ShardOptions { + pub shard: Shard, + pub seed: u64, + pub database_url: String, + pub listen: String, + pub script: SimScript, + /// Exit when stdin closes, so that a shard never outlives the gateway that started it. + pub attached: bool, +} + +type Command = (Control, oneshot::Sender); +type CurrentRouter = Arc>>>; +#[cfg(feature = "console")] +type CurrentConsole = Arc>>; + +/// What the request handlers need from the current world, which changes on restart. +#[derive(Clone, Default)] +struct Current { + router: CurrentRouter, + #[cfg(feature = "console")] + console: CurrentConsole, +} + +impl Current { + fn publish(&self, world: &World) { + *self.router.write().expect("router lock") = Some(Arc::clone(world.router())); + #[cfg(feature = "console")] + { + *self.console.write().expect("console lock") = Some(world.console()); + } + } +} + +#[derive(Clone)] +struct Node { + commands: mpsc::UnboundedSender, + current: Current, +} + +pub async fn run(options: ShardOptions) { + let ShardOptions { + shard, + seed, + database_url, + listen, + script, + attached, + } = options; + let storage = Storage::connect(&database_url).await.unwrap_or_else(|e| { + panic!( + "failed to connect to the database of shard {}: {e}", + shard.index + ) + }); + let saved = storage + .load() + .await + .unwrap_or_else(|e| panic!("failed to load the snapshot of shard {}: {e}", shard.index)); + let (saves, snapshots) = watch::channel(None::); + tokio::spawn(save_snapshots(storage, snapshots)); + + let current = Current::default(); + let (commands, receiver) = mpsc::unbounded_channel(); + { + let current = current.clone(); + std::thread::Builder::new() + .name(format!("shard-{}", shard.index)) + .spawn(move || { + let (world, restored) = open_world(saved.as_deref(), seed, script, shard); + current.publish(&world); + run_world(world, restored, receiver, ¤t, &saves); + }) + .expect("failed to start the simulation thread"); + } + if attached { + std::thread::spawn(|| { + let _ = std::io::copy(&mut std::io::stdin(), &mut std::io::sink()); + std::process::exit(0); + }); + } + + let app = axum::Router::new() + .route("/control", get(control)) + .route("/remote", get(remote)); + #[cfg(feature = "console")] + let app = app.route("/console", get(console_socket)); + let app = app.with_state(Node { commands, current }); + let listener = tokio::net::TcpListener::bind(&listen) + .await + .unwrap_or_else(|e| panic!("shard {} failed to bind {listen}: {e}", shard.index)); + let address = listener.local_addr().expect("shard listener address"); + { + let mut stdout = std::io::stdout().lock(); + let _ = writeln!(stdout, "{LISTENING} {address}"); + let _ = stdout.flush(); + } + tracing::info!(shard = shard.index, %address, "shard listening"); + axum::serve(listener, app) + .await + .expect("shard server failed"); +} + +/// Restores the shard's saved world, or creates a new one when there is none or it doesn't fit. +fn open_world(saved: Option<&str>, seed: u64, script: SimScript, shard: Shard) -> (World, bool) { + match saved + .map(|snapshot| World::from_snapshot_with_script(snapshot, seed, script.clone(), shard)) + { + Some(Ok(world)) => (world, true), + Some(Err(error)) => { + tracing::warn!(shard = shard.index, %error, "discarding an unusable snapshot"); + (World::new_shard(seed, script, shard), false) + } + None => (World::new_shard(seed, script, shard), false), + } +} + +async fn save_snapshots(storage: Storage, mut snapshots: watch::Receiver>) { + while snapshots.changed().await.is_ok() { + let Some(snapshot) = snapshots.borrow_and_update().clone() else { + continue; + }; + if let Err(error) = storage.save(&snapshot).await { + tracing::error!(%error, "failed to persist the shard snapshot"); + } + } +} + +/// The Rune VM is not Send, so the world lives on this thread and only acts on gateway commands. +fn run_world( + mut world: World, + mut restored: bool, + mut commands: mpsc::UnboundedReceiver, + current: &Current, + saves: &watch::Sender>, +) { + while let Some((control, reply)) = commands.blocking_recv() { + let answer = match control { + Control::Hello => ready(&world, restored), + Control::Peers(peers) => { + connect(&world, peers); + ControlReply::Done + } + Control::Step { directory, board } => { + world.router().apply_directory(&directory); + world.set_board(board); + world.step(); + world.router().flush(); + ControlReply::Stepped { + report: world.take_report(), + quotes: world.quotes(), + } + } + Control::View { terrain } => ControlReply::View { + state: world.state_json(), + terrain: terrain.then(|| world.terrain_json()), + }, + Control::Save => { + world.admit_arrivals(); + match world.snapshot_json() { + Ok(snapshot) => { + saves.send_replace(Some(snapshot)); + ControlReply::Done + } + Err(error) => ControlReply::Failed(error.to_string()), + } + } + Control::Roster => ControlReply::Roster(world.roster()), + Control::Edit(command) => { + world.apply(&command); + ControlReply::Done + } + Control::Script(source) => ControlReply::Script(world.set_script(&source)), + Control::Restart { seed } => { + world = world.restarted(seed); + restored = false; + current.publish(&world); + ready(&world, restored) + } + Control::Console(request) => console(&world, &request), + }; + let _ = reply.send(answer); + } +} + +fn ready(world: &World, restored: bool) -> ControlReply { + ControlReply::Ready { + system_id: world.router().system_id(), + seed: world.seed(), + tick: world.tick(), + restored, + } +} + +fn connect(world: &World, peers: Vec) { + let this = world.shard().index; + let transport = ClusterTransport::new(this, peers.clone()); + world.router().set_remote(Arc::clone(&transport) as _); + #[cfg(feature = "console")] + for (index, peer) in peers + .into_iter() + .enumerate() + .filter(|&(index, _)| index != this) + { + let transport = Arc::clone(&transport); + world.register_console_peer( + &format!("shard-{index}"), + Arc::new(move |path, message| { + crate::world::RemoteSender::send(&*transport, index, path, message); + Ok(()) + }), + Arc::new(move |request| transport::console_query(&peer.address, request)), + ); + } +} + +#[cfg(feature = "console")] +fn console(world: &World, request: &str) -> ControlReply { + ControlReply::Console(world.console().handle_text(request)) +} + +#[cfg(not(feature = "console"))] +fn console(_: &World, _: &str) -> ControlReply { + ControlReply::Failed("this build has no Picard console".into()) +} + +async fn control(upgrade: WebSocketUpgrade, State(node): State) -> Response { + upgrade + .max_message_size(1 << 30) + .on_upgrade(move |socket| serve_control(socket, node)) +} + +async fn serve_control(mut socket: WebSocket, node: Node) { + while let Some(Ok(message)) = socket.recv().await { + let Message::Text(text) = message else { + continue; + }; + let answer = match serde_json::from_str::(text.as_str()) { + Ok(control) => { + let (reply, result) = oneshot::channel(); + if node.commands.send((control, reply)).is_err() { + return; + } + result + .await + .unwrap_or_else(|_| ControlReply::Failed("the simulation stopped".into())) + } + Err(error) => ControlReply::Failed(format!("unreadable command: {error}")), + }; + let Ok(frame) = serde_json::to_string(&answer) else { + return; + }; + if socket.send(Message::Text(frame.into())).await.is_err() { + return; + } + } +} + +async fn remote(upgrade: WebSocketUpgrade, State(node): State) -> Response { + upgrade + .max_message_size(1 << 30) + .on_upgrade(move |socket| serve_remote(socket, node)) +} + +async fn serve_remote(mut socket: WebSocket, node: Node) { + while let Some(Ok(message)) = socket.recv().await { + let Message::Text(text) = message else { + continue; + }; + let router = node.current.router.read().expect("router lock").clone(); + let Some(router) = router else { + return; + }; + let answer = tokio::task::spawn_blocking(move || { + transport::receive(&router, "shard", text.as_str()) + }); + let Ok(Ok(frame)) = tokio::time::timeout(Duration::from_secs(30), answer).await else { + return; + }; + if socket.send(Message::Text(frame.into())).await.is_err() { + return; + } + } +} + +#[cfg(feature = "console")] +async fn console_socket(upgrade: WebSocketUpgrade, State(node): State) -> Response { + upgrade.on_upgrade(move |socket| serve_console(socket, node)) +} + +/// Answers console requests forwarded by another shard's console; it never waits for the +/// simulation thread, which may itself be waiting for this answer. +#[cfg(feature = "console")] +async fn serve_console(mut socket: WebSocket, node: Node) { + while let Some(Ok(message)) = socket.recv().await { + let Message::Text(text) = message else { + continue; + }; + let console = node.current.console.read().expect("console lock").clone(); + let Some(console) = console else { + return; + }; + let Ok(response) = + tokio::task::spawn_blocking(move || console.handle_text(text.as_str())).await + else { + return; + }; + if socket.send(Message::Text(response.into())).await.is_err() { + return; + } + } +} diff --git a/src/cluster/transport.rs b/src/cluster/transport.rs new file mode 100644 index 0000000..a9c41e8 --- /dev/null +++ b/src/cluster/transport.rs @@ -0,0 +1,255 @@ +//! Delivers entity messages between shard nodes over persistent WebSockets. +//! +//! Every peer gets one connection and one worker thread. Messages are queued in order and sent +//! in batches; the receiving shard answers every batch with one response per message, so +//! [`RemoteSender::flush`] can wait until everything sent so far has reached its mailbox. + +use super::protocol::Peer; +use crate::world::{RemoteSender, Router}; +use picard::remoting::{ActorAddress, RemoteRequest, RemoteResponse}; +use std::collections::VecDeque; +use std::net::TcpStream; +use std::sync::{Arc, Condvar, Mutex}; +use std::thread::JoinHandle; +use std::time::{Duration, Instant}; +use tungstenite::stream::MaybeTlsStream; +use tungstenite::{Message, WebSocket}; + +const MAX_BATCH: usize = 512; +const FLUSH_TIMEOUT: Duration = Duration::from_secs(10); +const CONNECT_ATTEMPTS: u32 = 50; + +#[derive(Default)] +struct Queue { + waiting: VecDeque, + /// Messages queued or sent but not yet answered. + unanswered: usize, + closed: bool, +} + +struct Link { + queue: Mutex, + changed: Condvar, +} + +impl Link { + fn lock(&self) -> std::sync::MutexGuard<'_, Queue> { + self.queue.lock().expect("transport queue lock") + } +} + +struct Connection { + peer: Peer, + link: Arc, + worker: Option>, +} + +pub struct ClusterTransport { + connections: Vec>, +} + +impl ClusterTransport { + /// Connects lazily to every peer except `this` shard. + pub fn new(this: usize, peers: Vec) -> Arc { + let connections = peers + .into_iter() + .enumerate() + .map(|(index, peer)| { + (index != this).then(|| { + let link = Arc::new(Link { + queue: Mutex::default(), + changed: Condvar::new(), + }); + let worker = { + let (link, url) = + (Arc::clone(&link), format!("ws://{}/remote", peer.address)); + std::thread::Builder::new() + .name(format!("shard-link-{index}")) + .spawn(move || run_link(&url, &link)) + .expect("failed to start a shard link") + }; + Connection { + peer, + link, + worker: Some(worker), + } + }) + }) + .collect(); + Arc::new(Self { connections }) + } +} + +impl RemoteSender for ClusterTransport { + fn send(&self, shard: usize, path: &str, payload: serde_json::Value) { + let Some(Some(connection)) = self.connections.get(shard) else { + tracing::warn!(shard, path, "dropped a message for an unknown shard"); + return; + }; + let destination = ActorAddress::local(connection.peer.system_id, path); + let mut queue = connection.link.lock(); + queue + .waiting + .push_back(RemoteRequest::new(destination, None, payload)); + queue.unanswered += 1; + connection.link.changed.notify_all(); + } + + fn flush(&self) { + let deadline = Instant::now() + FLUSH_TIMEOUT; + for (shard, connection) in self.connections.iter().enumerate() { + let Some(connection) = connection else { + continue; + }; + let mut queue = connection.link.lock(); + while queue.unanswered > 0 && !queue.closed { + let left = deadline.saturating_duration_since(Instant::now()); + if left.is_zero() { + tracing::warn!( + shard, + unanswered = queue.unanswered, + "shard flush timed out" + ); + break; + } + queue = connection + .link + .changed + .wait_timeout(queue, left) + .expect("transport queue lock") + .0; + } + } + } +} + +impl Drop for ClusterTransport { + fn drop(&mut self) { + for connection in self.connections.iter_mut().flatten() { + connection.link.lock().closed = true; + connection.link.changed.notify_all(); + if let Some(worker) = connection.worker.take() { + let _ = worker.join(); + } + } + } +} + +type Socket = WebSocket>; + +fn connect(url: &str) -> Option { + for attempt in 0..CONNECT_ATTEMPTS { + match tungstenite::connect(url) { + Ok((socket, _)) => { + if let MaybeTlsStream::Plain(stream) = socket.get_ref() { + let _ = stream.set_nodelay(true); + } + return Some(socket); + } + Err(error) if attempt + 1 == CONNECT_ATTEMPTS => { + tracing::error!(url, %error, "failed to connect to a shard"); + } + Err(_) => std::thread::sleep(Duration::from_millis(100)), + } + } + None +} + +/// Sends queued messages to one peer until the transport is dropped. +fn run_link(url: &str, link: &Link) { + let mut socket: Option = None; + loop { + let batch: Vec = { + let mut queue = link.lock(); + while queue.waiting.is_empty() && !queue.closed { + queue = link.changed.wait(queue).expect("transport queue lock"); + } + if queue.closed { + if let Some(mut socket) = socket { + let _ = socket.close(None); + } + return; + } + let count = queue.waiting.len().min(MAX_BATCH); + queue.waiting.drain(..count).collect() + }; + let sent = batch.len(); + if socket.is_none() { + socket = connect(url); + } + let delivered = socket + .as_mut() + .is_some_and(|socket| exchange(socket, &batch)); + if !delivered { + tracing::warn!(url, dropped = sent, "lost messages to a shard"); + socket = None; + } + let mut queue = link.lock(); + queue.unanswered = queue.unanswered.saturating_sub(sent); + link.changed.notify_all(); + } +} + +fn exchange(socket: &mut Socket, batch: &[RemoteRequest]) -> bool { + let Ok(frame) = serde_json::to_string(batch) else { + return false; + }; + if socket.send(Message::Text(frame.into())).is_err() { + return false; + } + loop { + match socket.read() { + Ok(Message::Text(text)) => { + let Ok(responses) = serde_json::from_str::>(&text) else { + return false; + }; + for response in responses.iter().filter(|response| !response.accepted) { + tracing::debug!(error = ?response.error, "a shard rejected a message"); + } + return true; + } + Ok(Message::Ping(_) | Message::Pong(_)) => {} + _ => return false, + } + } +} + +/// Answers one batch of messages from another shard. +pub fn receive(router: &Router, peer: &str, frame: &str) -> String { + let responses: Vec = match serde_json::from_str::>(frame) { + Ok(requests) => requests + .into_iter() + .map(|request| router.handle(peer, request)) + .collect(), + Err(error) => { + tracing::warn!(%error, "received an unreadable batch from a shard"); + Vec::new() + } + }; + serde_json::to_string(&responses).unwrap_or_else(|_| "[]".into()) +} + +/// Runs one Picard console request on the shard at `address` and returns its JSON response. +#[cfg(feature = "console")] +pub fn console_query(address: &str, request: &str) -> Result { + let url = format!("ws://{address}/console"); + let (mut socket, _) = + tungstenite::connect(&url).map_err(|e| format!("cannot reach {url}: {e}"))?; + if let MaybeTlsStream::Plain(stream) = socket.get_ref() { + let _ = stream.set_read_timeout(Some(FLUSH_TIMEOUT)); + } + socket + .send(Message::Text(request.to_owned().into())) + .map_err(|e| e.to_string())?; + loop { + match socket.read() { + Ok(Message::Text(text)) => { + let _ = socket.close(None); + return Ok(text.as_str().to_owned()); + } + Ok(Message::Ping(_) | Message::Pong(_)) => {} + Ok(_) => return Err("the shard closed the console".into()), + Err(error) => return Err(error.to_string()), + } + } +} diff --git a/src/main.rs b/src/main.rs index 4328979..d7b6c6f 100644 --- a/src/main.rs +++ b/src/main.rs @@ -1,3 +1,4 @@ +mod cluster; mod storage; mod world; @@ -8,12 +9,13 @@ use axum::http::{StatusCode, header}; use axum::middleware::{self, Next}; use axum::response::{Html, IntoResponse, Response}; use axum::routing::{get, post}; +use cluster::gateway::{GatewayOptions, Msg, Outputs}; +use cluster::shard::ShardOptions; use include_dir::{Dir, include_dir}; use std::sync::{Arc, Mutex}; -use std::time::{Duration, Instant}; -use storage::Storage; +use std::time::Duration; use tokio::sync::{mpsc, oneshot, watch}; -use world::{Shard, SimScript, World}; +use world::{Shard, SimScript}; // The UI bundle that build.rs produced from ui/. static UI: Dir = include_dir!("$OUT_DIR/ui-pkg"); @@ -26,17 +28,6 @@ try { const m = await import('/pkg/ui.js'); await m.default({ module_or_path: '/ catch (e) { document.body.textContent = 'Failed to load the UI: ' + e; } "#; -enum Msg { - Edit(String), - Script(String, oneshot::Sender>), - #[cfg(feature = "console")] - PicardConsole(String, oneshot::Sender), - Restart { - regenerate: bool, - reply: oneshot::Sender, - }, -} - #[derive(Clone)] struct App { terrain: Arc>, @@ -45,95 +36,6 @@ struct App { tx: mpsc::UnboundedSender, } -struct SimStart { - seed: u64, - script: SimScript, - snapshot: Option, -} - -/// The Rune VM is not Send, so the world lives on this thread and publishes JSON snapshots. -fn run_sim( - SimStart { - seed, - script: sim_script, - snapshot, - }: SimStart, - mut rx: mpsc::UnboundedReceiver, - terrain: &Mutex, - script: &Mutex, - state_tx: &watch::Sender>, - snapshot_tx: &watch::Sender>, -) { - let mut world = match snapshot { - Some(snapshot) => { - World::from_snapshot_with_script(&snapshot, seed, sim_script, Shard::SINGLE) - .unwrap_or_else(|e| panic!("failed to restore simulation snapshot: {e}")) - } - None => World::new_shard(seed, sim_script, Shard::SINGLE), - }; - *script.lock().unwrap() = world.script().to_string(); - let mut rev = u32::MAX; - let mut last_saved = Instant::now(); - snapshot_tx.send_replace(Some( - world - .snapshot_json() - .expect("failed to serialize initial simulation snapshot"), - )); - loop { - while let Ok(msg) = rx.try_recv() { - match msg { - Msg::Edit(cmd) => world.apply(&cmd), - Msg::Script(src, reply) => { - let r = world.set_script(&src); - if r.is_ok() { - *script.lock().unwrap() = src; - } - let _ = reply.send(r); - } - #[cfg(feature = "console")] - Msg::PicardConsole(request, reply) => { - let _ = reply.send(world.handle_picard_console_text(&request)); - } - Msg::Restart { regenerate, reply } => { - let seed = if regenerate { - world.regenerated_seed() - } else { - world.seed() - }; - world = world.restarted(seed); - *script.lock().unwrap() = world.script().to_string(); - rev = u32::MAX; - snapshot_tx.send_replace(Some( - world - .snapshot_json() - .expect("failed to serialize restarted simulation"), - )); - let _ = reply.send(seed); - } - } - } - if !world.paused { - for _ in 0..world.speed { - world.step(); - } - } - if world.terrain_rev != rev { - rev = world.terrain_rev; - *terrain.lock().unwrap() = world.terrain_json(); - } - state_tx.send_replace(Arc::from(world.state_json())); - if last_saved.elapsed() >= Duration::from_secs(1) { - snapshot_tx.send_replace(Some( - world - .snapshot_json() - .expect("failed to serialize simulation snapshot"), - )); - last_saved = Instant::now(); - } - std::thread::sleep(Duration::from_millis(100)); - } -} - /// Rejects cross-site requests (including WebSocket upgrades), which could otherwise edit the world. async fn same_origin(req: Request, next: Next) -> Response { let h = req.headers(); @@ -224,19 +126,34 @@ async fn ws(upgrade: WebSocketUpgrade, State(app): State) -> Response { } #[cfg(feature = "console")] -async fn picard_ws(upgrade: WebSocketUpgrade, State(app): State) -> Response { - upgrade.on_upgrade(move |socket| picard_client(socket, app)) +#[derive(serde::Deserialize)] +struct ConsoleTarget { + #[serde(default)] + shard: usize, } #[cfg(feature = "console")] -async fn picard_client(mut socket: WebSocket, app: App) { +async fn picard_ws( + upgrade: WebSocketUpgrade, + axum::extract::Query(target): axum::extract::Query, + State(app): State, +) -> Response { + upgrade.on_upgrade(move |socket| picard_client(socket, app, target.shard)) +} + +#[cfg(feature = "console")] +async fn picard_client(mut socket: WebSocket, app: App, shard: usize) { while let Some(message) = socket.recv().await { match message { Ok(Message::Text(text)) if text.len() <= picard::console::MAX_REQUEST_BYTES => { let (reply, result) = oneshot::channel(); if app .tx - .send(Msg::PicardConsole(text.to_string(), reply)) + .send(Msg::PicardConsole { + request: text.to_string(), + shard, + reply, + }) .is_err() { return; @@ -298,78 +215,122 @@ async fn client(mut socket: WebSocket, app: App) { } } -#[tokio::main] -async fn main() { - let filter = tracing_subscriber::EnvFilter::try_from_default_env() - .unwrap_or_else(|_| tracing_subscriber::EnvFilter::new("picard=info")); - tracing_subscriber::fmt() - .with_env_filter(filter) - .with_target(false) - .init(); - - let seed = std::env::args() - .nth(1) - .and_then(|s| s.parse().ok()) - .unwrap_or(42u64); +fn load_script() -> (String, SimScript) { let config_path = std::env::var("SIM_CONFIG").unwrap_or_else(|_| "config.rn".into()); let config_source = std::fs::read_to_string(&config_path) .unwrap_or_else(|e| panic!("failed to read simulation config at {config_path}: {e}")); let sim_script = SimScript::load(&config_source) .unwrap_or_else(|e| panic!("failed to load simulation config at {config_path}: {e}")); - let addr = std::env::var("SIM_ADDR").unwrap_or_else(|_| "127.0.0.1:7878".into()); - let database_url = - std::env::var("DATABASE_URL").unwrap_or_else(|_| "sqlite://sim.db?mode=rwc".into()); - let storage = Storage::connect(&database_url) - .await - .expect("failed to connect to simulation database"); - let saved_snapshot = storage - .load() - .await - .expect("failed to load simulation snapshot"); - let (snapshot_tx, snapshot_rx) = watch::channel(None::); - let snapshot_storage = storage.clone(); - tokio::spawn(async move { - let mut snapshots = snapshot_rx; - let mut interval = tokio::time::interval(Duration::from_secs(1)); - interval.tick().await; - loop { - tokio::select! { - _ = interval.tick() => {} - changed = snapshots.changed() => { - if changed.is_err() { - break; - } - } - } - let Some(snapshot) = snapshots.borrow_and_update().clone() else { - continue; - }; - if let Err(error) = snapshot_storage.save(&snapshot).await { - eprintln!("failed to persist simulation snapshot: {error}"); - } + (config_source, sim_script) +} + +/// Parses `shard --index I --count N --seed S --database URL [--listen ADDR] [--attached]`. +fn shard_options(args: &[String], script: SimScript) -> ShardOptions { + let mut index = None; + let mut count = None; + let mut seed = 42; + let mut database_url = None; + let mut listen = "127.0.0.1:0".to_owned(); + let mut attached = false; + let mut args = args.iter(); + while let Some(flag) = args.next() { + if flag == "--attached" { + attached = true; + continue; } + let value = args + .next() + .unwrap_or_else(|| panic!("missing value for {flag}")); + let number = || { + value + .parse() + .unwrap_or_else(|_| panic!("invalid value for {flag}: {value}")) + }; + match flag.as_str() { + "--index" => index = Some(number()), + "--count" => count = Some(number()), + "--seed" => { + seed = value + .parse() + .unwrap_or_else(|_| panic!("invalid seed: {value}")) + } + "--database" => database_url = Some(value.clone()), + "--listen" => listen.clone_from(value), + _ => panic!("unknown shard option {flag}"), + } + } + let shard = Shard::new( + index.expect("--index is required"), + count.expect("--count is required"), + ) + .expect("--index must be less than --count"); + ShardOptions { + shard, + seed, + database_url: database_url.expect("--database is required"), + listen, + script, + attached, + } +} + +#[tokio::main] +async fn main() { + let args: Vec = std::env::args().skip(1).collect(); + let is_shard = args.first().is_some_and(|mode| mode == "shard"); + let filter = tracing_subscriber::EnvFilter::try_from_default_env() + .unwrap_or_else(|_| tracing_subscriber::EnvFilter::new("picard=info,sim=info")); + // Shards keep stdout for announcing their address to the gateway. + tracing_subscriber::fmt() + .with_env_filter(filter) + .with_target(false) + .with_writer(std::io::stderr) + .init(); + + let (config_source, sim_script) = load_script(); + if is_shard { + cluster::shard::run(shard_options(&args[1..], sim_script)).await; + return; + } + + let seed = args.first().and_then(|s| s.parse().ok()).unwrap_or(42u64); + let shards = std::env::var("SIM_SHARDS").ok().map_or(3, |count| { + count + .parse::() + .ok() + .filter(|&count| count > 0) + .unwrap_or_else(|| panic!("SIM_SHARDS must be a positive number, not {count}")) }); + let database_template = std::env::var("SHARD_DATABASE_URL") + .unwrap_or_else(|_| "sqlite://sim-shard-{shard}.db?mode=rwc".into()); + assert!( + shards == 1 || database_template.contains("{shard}"), + "SHARD_DATABASE_URL must contain {{shard}} so that every shard gets its own database" + ); + let addr = std::env::var("SIM_ADDR").unwrap_or_else(|_| "127.0.0.1:7878".into()); + let listener = tokio::net::TcpListener::bind(&addr) + .await + .expect("failed to bind"); let terrain_buf = Arc::new(Mutex::new(String::new())); let script = Arc::new(Mutex::new(config_source)); let (tx, rx) = mpsc::unbounded_channel(); let (state_tx, state_rx) = watch::channel(Arc::::from("{}")); { - let (terrain_buf, script) = (Arc::clone(&terrain_buf), Arc::clone(&script)); - std::thread::spawn(move || { - run_sim( - SimStart { - seed, - script: sim_script, - snapshot: saved_snapshot, - }, - rx, - &terrain_buf, - &script, - &state_tx, - &snapshot_tx, - ); - }); + let outputs = Outputs { + terrain: Arc::clone(&terrain_buf), + script: Arc::clone(&script), + state: state_tx, + }; + let options = GatewayOptions { + shards, + seed, + database_template, + }; + std::thread::Builder::new() + .name("gateway".into()) + .spawn(move || cluster::gateway::conduct(&options, &outputs, rx)) + .expect("failed to start the gateway"); } let app = App { @@ -395,9 +356,6 @@ async fn main() { .layer(middleware::from_fn(same_origin)) .with_state(app); - let listener = tokio::net::TcpListener::bind(&addr) - .await - .expect("failed to bind"); - println!("Simulation running at http://{addr}"); + println!("Simulation running at http://{addr} with {shards} shards"); axum::serve(listener, router).await.unwrap(); } diff --git a/src/world.rs b/src/world.rs index 0fd7d0b..221f336 100644 --- a/src/world.rs +++ b/src/world.rs @@ -1,9 +1,5 @@ mod actors; mod economy; -#[allow( - dead_code, - reason = "the cluster gateway and shard nodes use the remote side" -)] mod router; mod work; @@ -11,10 +7,11 @@ use actors::{ Activity, AnimalId, AnimalMsg, AnimalTurn, Area, BirthCost, Goods, Mailbox, Needs, PartnershipMsg, PersonEdit, PersonMsg, ReplyBody, RequestId, }; -use economy::{Cart, CartMsg, Job, Market, MarketMsg, Quote, Workplace, WorkplaceMsg}; +pub use economy::Quote; +use economy::{Cart, CartMsg, Job, Market, MarketMsg, Workplace, WorkplaceMsg}; use picard::actors::{ActorRef, ActorSystem, Sender, Tell}; use router::{Arrival, LocalActor}; -pub use router::{EntityKey, Kind, Router, Shard}; +pub use router::{EntityKey, Kind, RemoteSender, Router, Shard}; use rune::runtime::RuntimeContext; use rune::termcolor::Buffer; use rune::{Any, Context, Diagnostics, Module, Source, Sources, Unit, Vm}; @@ -29,6 +26,15 @@ pub const W: usize = 96; pub const H: usize = 64; pub const DEFAULT_SCRIPT: &str = include_str!("../config.rn"); +/// A random seed for a new world that differs from `current`. +pub fn regenerated_seed(current: u64) -> u64 { + let mut seed = rand::random::(); + if seed == current { + seed = seed.wrapping_add(1); + } + seed +} + #[derive(Clone, Copy, Debug, Deserialize, Serialize, rune::FromValue)] pub struct StageDuration { pub min_days: u32, @@ -681,6 +687,7 @@ fn config_accessor_source() -> String { /// A compiled simulation script: `pub const` defaults plus `pub fn` behaviours. #[derive(Clone)] pub struct SimScript { + #[cfg_attr(not(test), expect(dead_code, reason = "tests read the source back"))] source: String, config: SimulationConfig, runtime: Arc, @@ -737,6 +744,7 @@ impl SimScript { }) } + #[cfg(test)] pub fn source(&self) -> &str { &self.source } @@ -2862,14 +2870,6 @@ impl World { world } - pub fn regenerated_seed(&self) -> u64 { - let mut seed = rand::random::(); - if seed == self.seed { - seed = seed.wrapping_add(1); - } - seed - } - pub fn seed(&self) -> u64 { self.seed } @@ -2910,6 +2910,7 @@ impl World { }) } + #[cfg(test)] pub fn script(&self) -> &str { self.script.source() } @@ -3498,14 +3499,34 @@ impl World { } #[cfg(feature = "console")] - pub fn handle_picard_console_text(&self, request: &str) -> String { - self.actor_system.console().handle_text(request) + pub fn console(&self) -> picard::console::Console { + self.actor_system.console() + } + + /// Lets the Picard console reach the actors of another shard as `@name`. + #[cfg(feature = "console")] + pub fn register_console_peer( + &self, + name: &str, + sender: picard::console::ConsolePeerSender, + query: picard::console::ConsolePeerQuery, + ) { + self.actor_system.register_console_peer(name, sender); + self.actor_system.register_console_peer_query(name, query); } } /// The interface the cluster uses to drive a shard. -#[allow(dead_code, reason = "the cluster gateway and shard nodes use these")] impl World { + pub fn tick(&self) -> u64 { + self.tick + } + + /// Takes in entities handed over by other shards, so that a snapshot includes them. + pub fn admit_arrivals(&mut self) { + self.receive_arrivals(); + } + pub fn shard(&self) -> Shard { self.shard } @@ -4441,7 +4462,7 @@ mod tests { #[test] fn regenerated_seed_differs_from_current_seed() { let world = World::new(42); - assert_ne!(world.regenerated_seed(), world.seed()); + assert_ne!(super::regenerated_seed(world.seed()), world.seed()); } #[test] diff --git a/tests/cluster.rs b/tests/cluster.rs new file mode 100644 index 0000000..275226a --- /dev/null +++ b/tests/cluster.rs @@ -0,0 +1,86 @@ +//! Runs the real binary as a gateway with two shard processes and checks that the UI sees one +//! world spanning both shards. + +use std::io::{Read, Write}; +use std::net::{TcpListener, TcpStream}; +use std::process::{Child, Command, Stdio}; +use std::time::{Duration, Instant}; + +struct Gateway(Child); + +impl Drop for Gateway { + fn drop(&mut self) { + let _ = self.0.kill(); + let _ = self.0.wait(); + } +} + +fn get(address: &str, path: &str) -> Option { + let mut stream = TcpStream::connect(address).ok()?; + stream.set_read_timeout(Some(Duration::from_secs(5))).ok()?; + write!( + stream, + "GET {path} HTTP/1.1\r\nHost: {address}\r\nConnection: close\r\n\r\n" + ) + .ok()?; + let mut response = String::new(); + stream.read_to_string(&mut response).ok()?; + let (_, body) = response.split_once("\r\n\r\n")?; + serde_json::from_str(body).ok() +} + +#[test] +fn gateway_presents_the_shards_as_one_world() { + let address = { + let probe = TcpListener::bind("127.0.0.1:0").unwrap(); + probe.local_addr().unwrap().to_string() + }; + let directory = std::env::temp_dir().join(format!("sim-cluster-test-{}", std::process::id())); + std::fs::create_dir_all(&directory).unwrap(); + let database = format!( + "sqlite://{}/shard-{{shard}}.db?mode=rwc", + directory.display() + ); + let _gateway = Gateway( + Command::new(env!("CARGO_BIN_EXE_sim")) + .arg("5") + .env("SIM_SHARDS", "2") + .env("SIM_ADDR", &address) + .env("SHARD_DATABASE_URL", &database) + .env( + "SIM_CONFIG", + concat!(env!("CARGO_MANIFEST_DIR"), "/config.rn"), + ) + .env("RUST_LOG", "warn") + .stdout(Stdio::null()) + .spawn() + .unwrap(), + ); + + let deadline = Instant::now() + Duration::from_secs(120); + let mut first_tick = None; + loop { + assert!(Instant::now() < deadline, "the cluster never got going"); + std::thread::sleep(Duration::from_millis(500)); + let Some(state) = get(&address, "/api/state") else { + continue; + }; + let Some(tick) = state["tick"].as_u64() else { + continue; + }; + let start = *first_tick.get_or_insert(tick); + let shards = state["shards"].as_array().unwrap(); + assert_eq!(shards.len(), 2); + let border = shards[1][0].as_u64().unwrap(); + let people = state["people"].as_array().unwrap(); + let west = people.iter().any(|p| p[1].as_u64() < Some(border)); + let east = people.iter().any(|p| p[1].as_u64() >= Some(border)); + if tick > start + 5 && west && east { + assert_eq!(state["homes"].as_array().unwrap().len(), 2); + break; + } + } + let terrain = get(&address, "/api/terrain").unwrap(); + assert!(terrain["w"].as_u64().unwrap() > 0); + let _ = std::fs::remove_dir_all(&directory); +}