From bc58855ff5ee37088f37a368707cec489ad517a3 Mon Sep 17 00:00:00 2001 From: Steve Biedermann Date: Mon, 12 Oct 2026 19:20:30 +0200 Subject: [PATCH] feat: shard-aware world with transparent actor transfers Add a router that delivers actor messages locally or to the shard that currently hosts the entity. People and animals that leave their shard's columns depart with their full state, forward messages while in flight and are re-admitted on the destination shard. Enlarge the default world. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- CHANGELOG.md | 2 + config.rn | 28 +- src/main.rs | 10 +- src/world.rs | 708 +++++++++++++++++++++++++++++++++++++------- src/world/actors.rs | 71 ++++- src/world/router.rs | 557 ++++++++++++++++++++++++++++++++++ 6 files changed, 1253 insertions(+), 123 deletions(-) create mode 100644 src/world/router.rs diff --git a/CHANGELOG.md b/CHANGELOG.md index ecebd76..ceb9dff 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,9 +7,11 @@ Notable changes to this project are documented here. ### Added - Expose the Picard actor console over the `/api/actors/ws` WebSocket endpoint. +- The world can be split into shards that each own a strip of columns. People and animals that walk across a shard border are transferred with their inventory and stay reachable at the same address. ### 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`. +- The default world is larger: 192×128 tiles with 180 people, 1200 rabbits, 80 wolves, and twice the food and houses. diff --git a/config.rn b/config.rn index 7a891c7..ed81dc2 100644 --- a/config.rn +++ b/config.rn @@ -2,8 +2,8 @@ // `pub fn` items are entity behaviours. Every constant below is required. // World -pub const WIDTH = 96; -pub const HEIGHT = 64; +pub const WIDTH = 192; +pub const HEIGHT = 128; // Calendar pub const TICKS_PER_DAY = 1800; @@ -21,20 +21,20 @@ pub const OLDER_ADULT = #{ min_days: 10, max_days: 20, fertile: true }; pub const ELDERLY = #{ min_days: 4, max_days: 10, fertile: false }; // Initial world -pub const INITIAL_POPULATION = 90; -pub const INITIAL_YOUNG_ADULTS = 30; -pub const INITIAL_RABBITS = 600; -pub const INITIAL_WOLVES = 40; -pub const INITIAL_FOOD = 600; +pub const INITIAL_POPULATION = 180; +pub const INITIAL_YOUNG_ADULTS = 60; +pub const INITIAL_RABBITS = 1200; +pub const INITIAL_WOLVES = 80; +pub const INITIAL_FOOD = 1200; pub const INITIAL_PERSON_HUNGER = 10.0; pub const INITIAL_ANIMAL_ENERGY = 70.0; pub const ANIMAL_ENERGY_DEATH_THRESHOLD = 0.0; pub const STARTING_TICK_FRACTION = 4; // Population and housing -pub const POPULATION_CAPACITY = 12; +pub const POPULATION_CAPACITY = 30; pub const HOUSE_CAPACITY = 3; -pub const MAX_HOUSES = 10; +pub const MAX_HOUSES = 20; // Daylight pub const DAYLIGHT_NIGHT_THRESHOLD = 0.2; @@ -92,9 +92,9 @@ pub const STONE_REGROW_AMOUNT = 6; // Animal spawning pub const ANIMAL_POPULATION_INTERVAL = 50; -pub const RABBIT_SPAWN_CAP = 15; -pub const WOLF_SPAWN_CAP = 2; -pub const WOLF_SPAWN_RABBIT_THRESHOLD = 40; +pub const RABBIT_SPAWN_CAP = 30; +pub const WOLF_SPAWN_CAP = 4; +pub const WOLF_SPAWN_RABBIT_THRESHOLD = 80; pub const WOLF_MIN_HOME_DISTANCE = 15; // Rabbits @@ -103,7 +103,7 @@ pub const RABBIT_ENERGY_REGEN = 0.2; pub const RABBIT_ENERGY_MAX = 100.0; pub const RABBIT_WOLF_DETECTION_RANGE = 4; pub const RABBIT_BREED_ENERGY = 80.0; -pub const RABBIT_MAX_POPULATION = 120; +pub const RABBIT_MAX_POPULATION = 300; pub const RABBIT_BREED_CHANCE = 0.01; pub const RABBIT_BREED_ENERGY_COST = 30.0; pub const RABBIT_NEWBORN_ENERGY = 50.0; @@ -115,7 +115,7 @@ pub const WOLF_HUNT_SUCCESS_CHANCE = 0.4; pub const WOLF_HUNT_ENERGY_GAIN = 50.0; pub const WOLF_ENERGY_MAX = 100.0; pub const WOLF_BREED_ENERGY = 90.0; -pub const WOLF_MAX_POPULATION = 10; +pub const WOLF_MAX_POPULATION = 24; pub const WOLF_BREED_CHANCE = 0.02; pub const WOLF_BREED_ENERGY_COST = 40.0; pub const WOLF_NEWBORN_ENERGY = 50.0; diff --git a/src/main.rs b/src/main.rs index d4a30c3..4328979 100644 --- a/src/main.rs +++ b/src/main.rs @@ -13,7 +13,7 @@ use std::sync::{Arc, Mutex}; use std::time::{Duration, Instant}; use storage::Storage; use tokio::sync::{mpsc, oneshot, watch}; -use world::{SimScript, World}; +use world::{Shard, SimScript, World}; // The UI bundle that build.rs produced from ui/. static UI: Dir = include_dir!("$OUT_DIR/ui-pkg"); @@ -65,9 +65,11 @@ fn run_sim( snapshot_tx: &watch::Sender>, ) { let mut world = match snapshot { - Some(snapshot) => World::from_snapshot_with_script(&snapshot, seed, sim_script) - .unwrap_or_else(|e| panic!("failed to restore simulation snapshot: {e}")), - None => World::new_with_script(seed, sim_script), + 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; diff --git a/src/world.rs b/src/world.rs index 5b7f1eb..fa89ac2 100644 --- a/src/world.rs +++ b/src/world.rs @@ -1,10 +1,17 @@ mod actors; +#[allow( + dead_code, + reason = "the cluster gateway and shard nodes use the remote side" +)] +mod router; use actors::{ - Activity, AnimalActor, AnimalId, AnimalMsg, AnimalTurn, Area, BirthCost, Goods, Mailbox, Needs, - PartnershipActor, PartnershipMsg, PersonActor, PersonEdit, PersonMsg, ReplyBody, RequestId, + Activity, AnimalId, AnimalMsg, AnimalTurn, Area, BirthCost, Goods, Mailbox, Needs, + PartnershipMsg, PersonEdit, PersonMsg, ReplyBody, RequestId, }; -use picard::actors::{ActorRef, ActorRefFactory, ActorReference, ActorSystem, Sender, Tell}; +use picard::actors::{ActorRef, ActorSystem, Sender, Tell}; +use router::{Arrival, LocalActor}; +pub use router::{EntityKey, Kind, Router, Shard}; use rune::runtime::RuntimeContext; use rune::termcolor::Buffer; use rune::{Any, Context, Diagnostics, Module, Source, Sources, Unit, Vm}; @@ -12,6 +19,7 @@ use serde::{Deserialize, Serialize}; use std::cmp::Reverse; use std::collections::{BTreeMap, BinaryHeap, HashSet}; use std::fmt::Write; +use std::ops::Range; use std::sync::{Arc, OnceLock}; pub const W: usize = 96; @@ -916,6 +924,47 @@ enum Species { Wolf, } +impl Species { + fn kind(self) -> Kind { + match self { + Species::Rabbit => Kind::Rabbit, + Species::Wolf => Kind::Wolf, + } + } +} + +/// How long a departed entity's actor keeps forwarding messages to the shard it moved to. +const FORWARDING_TICKS: u64 = 100; + +/// An entity that moved to another shard and whose old actor still forwards its messages. +struct Departure { + key: EntityKey, + shard: usize, + tick: u64, + actor: LocalActor, +} + +/// What changed in a shard during its steps, for the gateway to maintain the cluster directory. +#[derive(Clone, Debug, Default, Deserialize, Serialize)] +pub struct StepReport { + /// Entities handed to another shard. + pub moved: Vec<(EntityKey, usize)>, + /// Entities that were born on, spawned on or moved into this shard. + pub arrived: Vec, + /// Entities that died or were removed. + pub left: Vec, +} + +/// Reusable A* buffers; `visited` marks which entries belong to the current search. +#[derive(Default)] +struct PathScratch { + search: u32, + visited: Vec, + costs: Vec, + previous: Vec, + frontier: BinaryHeap>, +} + /// The entity state observed from the actors at the start of a step, kept up to date with /// their replies while the step runs. It is a read model only; the actors own the state. #[derive(Default)] @@ -952,6 +1001,11 @@ pub struct World { next_partnership_id: u32, mailbox: Mailbox, actor_system: ActorSystem, + shard: Shard, + router: Arc, + departed: Vec, + report: StepReport, + paths: PathScratch, houses: Vec<(usize, usize)>, home: (usize, usize), food: u32, @@ -1003,6 +1057,10 @@ struct WorldSnapshot { partnerships: Vec, #[serde(default = "default_next_partnership_id")] next_partnership_id: u32, + #[serde(default)] + next_animal_id: Option, + #[serde(default)] + shard: Option, } fn default_next_partnership_id() -> u32 { @@ -1041,26 +1099,37 @@ fn sample_stage_duration(calendar: &CalendarSettings, stage: LifeStage, rng: &mu + u32::try_from(rng.below(range as usize)).expect("sample is below a u32 range") } -/// Drops partnerships with a missing member and clears partner links without a partnership. -fn repair_partnerships(people: &mut [Person], partnerships: &mut Vec) { - let alive: HashSet = people.iter().map(|person| person.id).collect(); - partnerships.retain(|partnership| { - alive.contains(&partnership.first_person_id) - && alive.contains(&partnership.second_person_id) +/// Drops partnerships with a member that is not `alive` and clears the partner links of +/// `people` in them; returns the dropped partnerships. With `clear_orphans`, which needs every +/// partnership to be known, links without any partnership are cleared too. +fn repair_partnerships( + people: &mut [Person], + partnerships: &mut Vec, + alive: impl Fn(u32) -> bool, + clear_orphans: bool, +) -> Vec { + let (kept, dissolved): (Vec<_>, Vec<_>) = partnerships.drain(..).partition(|partnership| { + alive(partnership.first_person_id) && alive(partnership.second_person_id) }); - let partnered: HashSet<(u32, u32)> = partnerships - .iter() - .map(|partnership| (partnership.first_person_id, partnership.second_person_id)) - .flat_map(|(first, second)| [(first, second), (second, first)]) - .collect(); + *partnerships = kept; + let pairs = |partnerships: &[Partnership]| -> HashSet<(u32, u32)> { + partnerships + .iter() + .map(|partnership| (partnership.first_person_id, partnership.second_person_id)) + .flat_map(|(first, second)| [(first, second), (second, first)]) + .collect() + }; + let partnered = pairs(partnerships); + let broken = pairs(&dissolved); for person in people { - if person - .partner_id - .is_some_and(|partner| !partnered.contains(&(person.id, partner))) - { + if person.partner_id.is_some_and(|partner| { + broken.contains(&(person.id, partner)) + || (clear_orphans && !partnered.contains(&(person.id, partner))) + }) { person.partner_id = None; } } + dissolved } /// Collects one reply per actor and sorts them by key, so the simulation stays deterministic @@ -1123,9 +1192,10 @@ impl World { } fn populate(&mut self) { - for i in 0..self.config.initial_population { + let young_adults = self.shard.share(self.config.initial_young_adults); + for i in 0..self.shard.share(self.config.initial_population) { let mut person = self.new_person(self.home); - let stage = if i < self.config.initial_young_adults { + let stage = if i < young_adults { LifeStage::YoungAdult } else { LifeStage::Adult @@ -1133,13 +1203,13 @@ impl World { self.make_grown_up(&mut person, stage); self.spawn_person_actor(person); } - for _ in 0..self.config.initial_rabbits { + for _ in 0..self.shard.share(self.config.initial_rabbits) { if let Some(p) = self.random_land() { let energy = self.config.initial_animal_energy; self.spawn_animal(Species::Rabbit, p, energy); } } - for _ in 0..self.config.initial_wolves { + for _ in 0..self.shard.share(self.config.initial_wolves) { if let Some(p) = self.random_land() && cheb(p, self.home) > self.config.wolf_min_home_distance { @@ -1175,10 +1245,17 @@ impl World { person.age_ticks = age_days * self.calendar.ticks_per_day; } - fn empty(seed: u64, script: SimScript, tiles: Vec, res: Vec<(u8, u16)>) -> World { + fn empty( + seed: u64, + script: SimScript, + tiles: Vec, + res: Vec<(u8, u16)>, + shard: Shard, + ) -> World { let config = script.config.clone(); let (width, height) = (config.width, config.height); let actor_system = ActorSystem::new().expect("failed to start Picard actor system"); + let first_id = shard.first_id(); World { seed, width, @@ -1189,19 +1266,25 @@ impl World { rabbit_actors: BTreeMap::new(), wolf_actors: BTreeMap::new(), partnership_actors: BTreeMap::new(), - next_animal_id: 1, - next_partnership_id: 1, + next_animal_id: first_id, + next_partnership_id: u32::try_from(first_id).expect("shard counts are small"), mailbox: Mailbox::new(&actor_system), + router: Router::new(shard, &actor_system), actor_system, + shard, + departed: Vec::new(), + report: StepReport::default(), + paths: PathScratch::default(), houses: vec![], home: (width / 2, height / 2), - food: config.initial_food, + food: u32::try_from(shard.share(config.initial_food as usize)) + .expect("a share is at most the total"), wood: 0, stone: 0, tools: 0, rng: Rng(seed.wrapping_mul(0x9E37_79B9_7F4A_7C15) | 1), tick: (config.calendar.ticks_per_day / config.starting_tick_fraction).max(1), - next_id: 1, + next_id: u32::try_from(first_id).expect("shard counts are small"), log: vec![], vm: script.vm(), script, @@ -1213,14 +1296,22 @@ impl World { } } + #[cfg(test)] pub fn new_with_script(seed: u64, script: SimScript) -> World { + Self::new_shard(seed, script, Shard::SINGLE) + } + + /// Creates the part of the world owned by `shard`. Every shard generates the same terrain + /// and resources from `seed`, then keeps only the resources in its own columns. + pub fn new_shard(seed: u64, script: SimScript, shard: Shard) -> World { let tiles = Self::generate_terrain(seed, &script.config); let res = vec![(0, 0); tiles.len()]; - let mut w = Self::empty(seed, script, tiles, res); + let mut w = Self::empty(seed, script, tiles, res, shard); let (width, height) = (w.width, w.height); + let centre = w.region_centre(); w.home = w .find_land(GRASS) - .unwrap_or_else(|| w.find_land(SAND).unwrap_or((width / 2, height / 2))); + .unwrap_or_else(|| w.find_land(SAND).unwrap_or(centre)); for i in 0..width * height { let r = w.rng.f(); w.res[i] = match w.tiles[i] { @@ -1234,19 +1325,68 @@ impl World { _ => (0, 0), }; } + let columns = w.columns(); + for (i, res) in w.res.iter_mut().enumerate() { + if !columns.contains(&(i % width)) { + *res = (0, 0); + } + } let (hx, hy) = w.home; w.res[hy * width + hx] = (0, 0); + w.rng.0 ^= (shard.index as u64).wrapping_mul(0xD1B5_4A32_D192_ED03); + w.rng.0 |= 1; w.populate(); w } + /// The columns of the map this shard simulates. + fn columns(&self) -> Range { + self.shard.columns(self.width) + } + + fn owns(&self, (x, _): (usize, usize)) -> bool { + self.columns().contains(&x) + } + + fn region_centre(&self) -> (usize, usize) { + let columns = self.columns(); + (columns.start.midpoint(columns.end), self.height / 2) + } + + fn take_person_id(&mut self) -> u32 { + let id = self.next_id; + self.next_id += u32::try_from(self.shard.id_stride()).expect("shard counts are small"); + id + } + + fn take_partnership_id(&mut self) -> u32 { + let id = self.next_partnership_id; + self.next_partnership_id += + u32::try_from(self.shard.id_stride()).expect("shard counts are small"); + id + } + + /// Records an entity appearing on this shard, unless it is the only one. + fn report_arrival(&mut self, key: EntityKey) { + if self.shard.count > 1 { + self.report.arrived.push(key); + } + } + + fn report_departure(&mut self, key: EntityKey) { + if self.shard.count > 1 { + self.report.left.push(key); + } + } + fn spawn_person_actor(&mut self, person: Person) { let id = person.id; let actor = self - .actor_system - .actor_of_args::(&format!("person-{id}"), person) + .router + .spawn_person(person) .expect("failed to create Picard person actor"); self.person_actors.insert(id, actor); + self.report_arrival(EntityKey::person(id)); } fn animal_actors(&self, species: Species) -> &BTreeMap> { @@ -1268,16 +1408,13 @@ impl World { fn spawn_animal_actor(&mut self, species: Species, animal: Animal) -> AnimalId { let id = self.next_animal_id; - self.next_animal_id += 1; - let kind = match species { - Species::Rabbit => "rabbit", - Species::Wolf => "wolf", - }; + self.next_animal_id += self.shard.id_stride(); let actor = self - .actor_system - .actor_of_args::(&format!("{kind}-{id}"), (id, animal)) + .router + .spawn_animal(species.kind(), id, animal) .expect("failed to create Picard animal actor"); self.animal_actors_mut(species).insert(id, actor); + self.report_arrival(EntityKey::new(species.kind(), id)); id } @@ -1299,27 +1436,124 @@ impl World { fn spawn_partnership_actor(&mut self, partnership: Partnership) { let id = partnership.id; let actor = self - .actor_system - .actor_of_args::(&format!("partnership-{id}"), partnership) + .router + .spawn_partnership(partnership) .expect("failed to create Picard partnership actor"); self.partnership_actors.insert(id, actor); } fn retire_person(&mut self, id: u32) { if let Some(actor) = self.person_actors.remove(&id) { - actor.stop(); + let key = EntityKey::person(id); + self.router.unregister(key); + LocalActor::Person(actor).stop(); + self.report_departure(key); } } fn retire_animal(&mut self, species: Species, id: AnimalId) { if let Some(actor) = self.animal_actors_mut(species).remove(&id) { - actor.stop(); + let key = EntityKey::new(species.kind(), id); + self.router.unregister(key); + LocalActor::Animal(actor).stop(); + self.report_departure(key); } } fn retire_partnership(&mut self, id: u32) { if let Some(actor) = self.partnership_actors.remove(&id) { - actor.stop(); + self.router + .unregister(EntityKey::new(Kind::Partnership, u64::from(id))); + LocalActor::Partnership(actor).stop(); + } + } + + /// Takes in the entities other shards handed over since the last step. + fn receive_arrivals(&mut self) { + for arrival in self.router.take_arrivals() { + let key = match arrival { + Arrival::Person(person, actor) => { + self.person_actors.insert(person.id, actor); + self.say(&format!("Person #{} arrived from afar", person.id)); + EntityKey::person(person.id) + } + Arrival::Animal(kind, id, actor) => { + let species = if kind == Kind::Wolf { + Species::Wolf + } else { + Species::Rabbit + }; + self.animal_actors_mut(species).insert(id, actor); + EntityKey::new(kind, id) + } + }; + self.departed.retain(|departure| departure.key != key); + self.report_arrival(key); + } + } + + /// Stops the actors of entities that left long enough ago for every shard to know where + /// they went; the router keeps forwarding their messages from then on. + fn retire_departures(&mut self) { + let tick = self.tick; + let (expired, waiting): (Vec<_>, Vec<_>) = self + .departed + .drain(..) + .partition(|departure| tick.saturating_sub(departure.tick) >= FORWARDING_TICKS); + self.departed = waiting; + for departure in expired { + if self.router.retire_forwarder(departure.key, departure.shard) { + departure.actor.stop(); + } + } + } + + /// Hands every entity standing outside this shard's columns to the shard that owns its + /// position. The entities' actors pass their whole state on themselves. + fn transfer_out(&mut self, census: &Census) { + if self.shard.count == 1 { + return; + } + let request = self.mailbox.request(); + let mut leaving = Vec::new(); + for person in &census.people { + if !self.owns(person.position()) + && let Some(actor) = self.person_actors.remove(&person.id) + { + let shard = self.shard.owner(person.x, self.width); + actor.tell(PersonMsg::Depart(request, shard), self.mailbox.reply_to()); + leaving.push(( + EntityKey::person(person.id), + shard, + LocalActor::Person(actor), + )); + } + } + for species in [Species::Rabbit, Species::Wolf] { + let animals = match species { + Species::Rabbit => &census.rabbits, + Species::Wolf => &census.wolves, + }; + for (id, animal) in animals { + if !self.owns(animal.position()) + && let Some(actor) = self.animal_actors_mut(species).remove(id) + { + let shard = self.shard.owner(animal.x, self.width); + actor.tell(AnimalMsg::Depart(request, shard), self.mailbox.reply_to()); + let key = EntityKey::new(species.kind(), *id); + leaving.push((key, shard, LocalActor::Animal(actor))); + } + } + } + self.mailbox.collect(request, leaving.len()); + for (key, shard, actor) in leaving { + self.report.moved.push((key, shard)); + self.departed.push(Departure { + key, + shard, + tick: self.tick, + actor, + }); } } @@ -1387,9 +1621,7 @@ impl World { } fn tell_person(&self, id: u32, message: PersonMsg) { - if let Some(actor) = self.person_actors.get(&id) { - actor.tell(message, None); - } + self.router.tell(EntityKey::person(id), message); } /// Sends a request to one actor and waits for its reply. @@ -1403,10 +1635,10 @@ impl World { } fn find_land(&self, kind: u8) -> Option<(usize, usize)> { - let c = (self.width / 2, self.height / 2); + let c = self.region_centre(); let mut best: Option<(usize, usize)> = None; for y in 0..self.height { - for x in 0..self.width { + for x in self.columns() { if self.tiles[y * self.width + x] == kind && best.is_none_or(|b| cheb((x, y), c) < cheb(b, c)) { @@ -1418,8 +1650,10 @@ impl World { } fn random_land(&mut self) -> Option<(usize, usize)> { + let columns = self.columns(); for _ in 0..self.config.random_land_attempts { - let (x, y) = (self.rng.below(self.width), self.rng.below(self.height)); + let x = columns.start + self.rng.below(columns.len()); + let y = self.rng.below(self.height); if self.walkable((x, y)) { return Some((x, y)); } @@ -1429,8 +1663,8 @@ impl World { fn new_person(&mut self, (x, y): (usize, usize)) -> Person { let stage_duration_days = self.sample_stage_duration(LifeStage::Toddler); - let person = Person { - id: self.next_id, + Person { + id: self.take_person_id(), x, y, hunger: self.config.initial_person_hunger, @@ -1445,9 +1679,7 @@ impl World { stage_duration_days, partner_id: None, rng: self.rng.next() | 1, - }; - self.next_id += 1; - person + } } fn spawn_person_at(&mut self, position: (usize, usize)) -> Person { @@ -1539,45 +1771,72 @@ impl World { return from; } + let mut paths = std::mem::take(&mut self.paths); + let step = self.find_path_step(&mut paths, from, to); + self.paths = paths; + step.unwrap_or_else(|| self.wander(from.0, from.1)) + } + + /// A* search from `from` to `to`; returns the first step of a shortest path. + fn find_path_step( + &self, + paths: &mut PathScratch, + from: (usize, usize), + to: (usize, usize), + ) -> Option<(usize, usize)> { + let tiles = self.width * self.height; + if paths.visited.len() != tiles { + *paths = PathScratch { + visited: vec![0; tiles], + costs: vec![0; tiles], + previous: vec![0; tiles], + ..PathScratch::default() + }; + } + paths.search = paths.search.wrapping_add(1); + if paths.search == 0 { + paths.visited.fill(0); + paths.search = 1; + } + let search = paths.search; let start = from.1 * self.width + from.0; let target = to.1 * self.width + to.0; - let mut frontier = BinaryHeap::from([Reverse((cheb(from, to), start))]); - let mut costs = vec![usize::MAX; self.width * self.height]; - let mut previous = vec![usize::MAX; self.width * self.height]; - costs[start] = 0; - previous[start] = start; + paths.frontier.clear(); + paths.frontier.push(Reverse((cheb(from, to), start))); + paths.visited[start] = search; + paths.costs[start] = 0; + paths.previous[start] = start; - while let Some(Reverse((_, current))) = frontier.pop() { + while let Some(Reverse((_, current))) = paths.frontier.pop() { if current == target { let mut next = target; - while previous[next] != start { - next = previous[next]; + while paths.previous[next] != start { + next = paths.previous[next]; } - return (next % self.width, next / self.width); + return Some((next % self.width, next / self.width)); } let position = (current % self.width, current / self.width); for dy in -1..=1 { for dx in -1..=1 { - if (dx, dy) == (0, 0) { - continue; - } let Some((x, y)) = self.walk_to(position, dx, dy) else { continue; }; let neighbor = y * self.width + x; - let cost = costs[current] + 1; - if cost >= costs[neighbor] { + let cost = paths.costs[current] + 1; + if paths.visited[neighbor] == search && cost >= paths.costs[neighbor] { continue; } - costs[neighbor] = cost; - previous[neighbor] = current; - frontier.push(Reverse((cost + cheb((x, y), to), neighbor))); + paths.visited[neighbor] = search; + paths.costs[neighbor] = cost; + paths.previous[neighbor] = current; + paths + .frontier + .push(Reverse((cost + cheb((x, y), to), neighbor))); } } } - - self.wander(from.0, from.1) + None } fn nearest_res(&self, from: (usize, usize), kind: u8, r: usize) -> Option<(usize, usize)> { @@ -1619,6 +1878,8 @@ impl World { pub fn step(&mut self) { self.tick += 1; + self.receive_arrivals(); + self.retire_departures(); self.regrow(); let mut census = Census { people: self.advance_people(), @@ -1631,10 +1892,12 @@ impl World { } else { self.tick_wolves(&mut census); } - if self.retire_dead(&mut census) { + let new_day = self.tick.is_multiple_of(self.calendar.ticks_per_day); + if self.retire_dead(&mut census) || (self.shard.count > 1 && new_day) { self.prune_partnerships(&mut census.people); } self.population(&mut census); + self.transfer_out(&census); } /// Ages every person by one tick and returns them, logging stage changes and deaths. @@ -1697,12 +1960,19 @@ impl World { /// Dissolves partnerships that lost a member and tells their survivors. fn prune_partnerships(&mut self, people: &mut [Person]) { let mut partnerships = self.observe_partnerships(); - let before: HashSet = partnerships.iter().map(|p| p.id).collect(); let partners_before: Vec> = people.iter().map(|p| p.partner_id).collect(); - repair_partnerships(people, &mut partnerships); - let kept: HashSet = partnerships.iter().map(|p| p.id).collect(); - for id in before.difference(&kept) { - self.retire_partnership(*id); + let local: HashSet = people.iter().map(|p| p.id).collect(); + let router = Arc::clone(&self.router); + let alive = |id: u32| local.contains(&id) || router.knows(EntityKey::person(id)); + let dissolved = + repair_partnerships(people, &mut partnerships, alive, self.shard.count == 1); + for partnership in dissolved { + self.retire_partnership(partnership.id); + for member in [partnership.first_person_id, partnership.second_person_id] { + if !local.contains(&member) && router.knows(EntityKey::person(member)) { + self.tell_person(member, PersonMsg::SetPartner(None)); + } + } } for (person, partner_before) in people.iter().zip(partners_before) { if person.partner_id != partner_before { @@ -1712,8 +1982,10 @@ impl World { } fn regrow(&mut self) { + let columns = self.columns(); for _ in 0..self.config.resource_regrow_attempts { - let i = self.rng.below(self.width * self.height); + let tile = self.rng.below(columns.len() * self.height); + let i = (tile / columns.len()) * self.width + columns.start + tile % columns.len(); if self.res[i].0 != 0 { continue; } @@ -1747,14 +2019,14 @@ impl World { .is_multiple_of(self.config.animal_population_interval) { let energy = self.config.initial_animal_energy; - if census.rabbits.len() < self.config.rabbit_spawn_cap + if census.rabbits.len() < self.shard.share(self.config.rabbit_spawn_cap) && let Some(p) = self.random_land() { let rabbit = self.spawn_animal(Species::Rabbit, p, energy); census.rabbits.push(rabbit); } - if census.wolves.len() < self.config.wolf_spawn_cap - && census.rabbits.len() > self.config.wolf_spawn_rabbit_threshold + if census.wolves.len() < self.shard.share(self.config.wolf_spawn_cap) + && census.rabbits.len() > self.shard.share(self.config.wolf_spawn_rabbit_threshold) && let Some(p) = self.random_land() && cheb(p, self.home) > self.config.wolf_min_home_distance { @@ -1796,14 +2068,14 @@ impl World { people[second].partner_id = Some(first_id); self.tell_person(first_id, PersonMsg::SetPartner(Some(second_id))); self.tell_person(second_id, PersonMsg::SetPartner(Some(first_id))); + let id = self.take_partnership_id(); self.spawn_partnership_actor(Partnership { - id: self.next_partnership_id, + id, first_person_id: first_id, second_person_id: second_id, formed_tick: self.tick, last_birth_tick: self.tick / ticks_per_week * ticks_per_week, }); - self.next_partnership_id += 1; self.say(&format!( "People #{first_id} and #{second_id} became partners" )); @@ -1866,7 +2138,8 @@ impl World { } fn population_capacity(&self) -> usize { - self.config.population_capacity + self.houses.len() * self.config.house_capacity + self.shard.share(self.config.population_capacity) + + self.houses.len() * self.config.house_capacity } fn decide(&mut self, func: &str, c: Ctx) -> String { @@ -2247,7 +2520,7 @@ impl World { Species::Rabbit, census, breeders, - self.config.rabbit_max_population, + self.shard.share(self.config.rabbit_max_population), cost, self.config.rabbit_newborn_energy, ); @@ -2305,7 +2578,7 @@ impl World { Species::Wolf, census, breeders, - self.config.wolf_max_population, + self.shard.share(self.config.wolf_max_population), cost, self.config.wolf_newborn_energy, ); @@ -2322,7 +2595,7 @@ impl World { } pub fn restarted(&self, seed: u64) -> Self { - let mut world = Self::new_with_script(seed, self.script.clone()); + let mut world = Self::new_shard(seed, self.script.clone(), self.shard); world.speed = self.speed; world.paused = self.paused; world.calendar = self.calendar.clone(); @@ -2371,6 +2644,8 @@ impl World { calendar: self.calendar.clone(), partnerships: self.observe_partnerships(), next_partnership_id: self.next_partnership_id, + next_animal_id: Some(self.next_animal_id), + shard: Some(self.shard), }) } @@ -2380,15 +2655,16 @@ impl World { #[cfg(test)] pub fn from_snapshot(snapshot: &str, fallback_seed: u64) -> Result { - Self::from_snapshot_with_script(snapshot, fallback_seed, default_script()) + Self::from_snapshot_with_script(snapshot, fallback_seed, default_script(), Shard::SINGLE) } - /// Restores world state from a snapshot; `script` (normally `config.rn` + /// Restores world state from a snapshot of `shard`; `script` (normally `config.rn` /// on disk) supplies the behaviours and tuning constants. pub fn from_snapshot_with_script( snapshot: &str, fallback_seed: u64, script: SimScript, + shard: Shard, ) -> Result { let config = &script.config; let snapshot_value: serde_json::Value = @@ -2427,15 +2703,26 @@ impl World { { return Err("world snapshot has invalid terrain dimensions".into()); } + let saved_shard = saved.shard.unwrap_or(Shard::SINGLE); + if saved_shard != shard { + return Err(format!( + "world snapshot belongs to shard {} of {}, not {} of {}", + saved_shard.index, saved_shard.count, shard.index, shard.count + )); + } let mut world = Self::empty( saved.seed.unwrap_or(fallback_seed), script, saved.tiles, saved.res, + shard, ); world.width = saved.width; world.height = saved.height; world.next_partnership_id = saved.next_partnership_id; + if let Some(next_animal_id) = saved.next_animal_id { + world.next_animal_id = next_animal_id; + } world.houses = saved.houses; world.home = saved.home; world.food = saved.food; @@ -2463,7 +2750,16 @@ impl World { } } let mut partnerships = saved.partnerships; - repair_partnerships(&mut people, &mut partnerships); + // Other shards are unknown until the cluster directory arrives; partnerships with + // members elsewhere are checked by the daily pruning instead. + let local: HashSet = people.iter().map(|person| person.id).collect(); + let single = shard.count == 1; + repair_partnerships( + &mut people, + &mut partnerships, + |id| !single || local.contains(&id), + single, + ); for person in people { world.spawn_person_actor(person); } @@ -2616,6 +2912,9 @@ impl World { let amt = clamp_to(amt, 1, i64::from(self.config.max_resource_edit)); for y in y0..=y1 { for x in x0..=x1 { + if !self.owns((x, y)) { + continue; + } let i = y * self.width + x; let free = self.walkable((x, y)) && (x, y) != self.home && !self.houses.contains(&(x, y)); @@ -2635,7 +2934,7 @@ impl World { let mut left = self.config.manual_spawn_limit; for y in y0..=y1 { for x in x0..=x1 { - if left == 0 || !self.walkable((x, y)) { + if left == 0 || !self.walkable((x, y)) || !self.owns((x, y)) { continue; } left -= 1; @@ -2663,6 +2962,7 @@ impl World { for y in y0..=y1 { for x in x0..=x1 { if self.houses.len() < self.config.max_houses + && self.owns((x, y)) && self.walkable((x, y)) && self.res[y * self.width + x].0 == 0 && (x, y) != self.home @@ -2921,6 +3221,37 @@ impl World { } } +/// 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 shard(&self) -> Shard { + self.shard + } + + pub fn router(&self) -> &Arc { + &self.router + } + + /// Returns and resets what changed since the last report. + pub fn take_report(&mut self) -> StepReport { + std::mem::take(&mut self.report) + } + + /// Every person and animal currently simulated by this shard. + pub fn roster(&self) -> Vec { + let people = self.person_actors.keys().map(|&id| EntityKey::person(id)); + let rabbits = self + .rabbit_actors + .keys() + .map(|&id| EntityKey::new(Kind::Rabbit, id)); + let wolves = self + .wolf_actors + .keys() + .map(|&id| EntityKey::new(Kind::Wolf, id)); + people.chain(rabbits).chain(wolves).collect() + } +} + impl Drop for World { fn drop(&mut self) { // Shutdown starts eagerly; Drop must not block waiting for completion. @@ -2930,8 +3261,81 @@ impl Drop for World { #[cfg(test)] mod tests { - use super::actors::{Activity, AnimalTurn, Goods, PersonMsg, ReplyBody}; - use super::{Animal, DEFAULT_SCRIPT, H, LifeStage, SimScript, W, World, cheb, default_config}; + use super::actors::{Activity, AnimalTurn, Goods, PersonEdit, PersonMsg, ReplyBody}; + use super::{ + Animal, DEFAULT_SCRIPT, EntityKey, FORWARDING_TICKS, H, LifeStage, Router, Shard, + SimScript, W, World, cheb, default_config, + }; + use crate::world::router::RemoteSender; + use picard::remoting::{ActorAddress, RemoteRequest}; + use std::sync::{Arc, Mutex, Weak}; + use std::time::{Duration, Instant}; + + /// Delivers remote messages by calling the receiving shard's router directly. + #[derive(Default)] + struct Loopback { + routers: Mutex>>, + } + + impl RemoteSender for Loopback { + fn send(&self, shard: usize, path: &str, payload: serde_json::Value) { + let router = self.routers.lock().unwrap()[shard].upgrade().unwrap(); + let destination = ActorAddress::local(router.system_id(), path); + let response = + router.handle("loopback", RemoteRequest::new(destination, None, payload)); + assert!(response.accepted, "{path} rejected: {:?}", response.error); + } + + fn flush(&self) {} + } + + fn connected_shards(count: usize) -> Vec { + let loopback = Arc::new(Loopback::default()); + let shards: Vec = (0..count) + .map(|index| { + let shard = Shard::new(index, count).unwrap(); + World::new_shard(42, SimScript::load(DEFAULT_SCRIPT).unwrap(), shard) + }) + .collect(); + for world in &shards { + loopback + .routers + .lock() + .unwrap() + .push(Arc::downgrade(world.router())); + world.router().set_remote(loopback.clone()); + } + shards + } + + /// Does what the gateway does after every step: publishes where entities went. + fn publish_directory(shards: &mut [World]) { + let mut changes = Vec::new(); + for (index, world) in shards.iter_mut().enumerate() { + let report = world.take_report(); + changes.extend(report.moved.into_iter().map(|(key, to)| (key, Some(to)))); + changes.extend(report.arrived.into_iter().map(|key| (key, Some(index)))); + changes.extend(report.left.into_iter().map(|key| (key, None))); + } + for world in shards { + world.router().apply_directory(&changes); + } + } + + fn wait_for_person(world: &World, id: u32, done: impl Fn(&super::Person) -> bool) { + let deadline = Instant::now() + Duration::from_secs(5); + loop { + let people = world.observe_people(); + if people.iter().any(|person| person.id == id && done(person)) { + return; + } + assert!( + Instant::now() < deadline, + "person #{id} never got the message" + ); + std::thread::sleep(Duration::from_millis(10)); + } + } /// Replaces the value of `pub const {name}` in `source`, whatever its current value is. fn with_constant(source: &str, name: &str, value: &str) -> String { @@ -2953,6 +3357,31 @@ mod tests { .join("\n") } + /// The default script at the size of the original single village, for long-running tests. + fn village_script() -> SimScript { + let source = [ + ("WIDTH", "96"), + ("HEIGHT", "64"), + ("INITIAL_POPULATION", "90"), + ("INITIAL_YOUNG_ADULTS", "30"), + ("INITIAL_RABBITS", "600"), + ("INITIAL_WOLVES", "40"), + ("INITIAL_FOOD", "600"), + ("POPULATION_CAPACITY", "12"), + ("MAX_HOUSES", "10"), + ("RABBIT_SPAWN_CAP", "15"), + ("WOLF_SPAWN_CAP", "2"), + ("WOLF_SPAWN_RABBIT_THRESHOLD", "40"), + ("RABBIT_MAX_POPULATION", "120"), + ("WOLF_MAX_POPULATION", "10"), + ] + .into_iter() + .fold(DEFAULT_SCRIPT.to_string(), |source, (name, value)| { + with_constant(&source, name, value) + }); + SimScript::load(&source).unwrap() + } + fn is_fertile(world: &World, stage: LifeStage) -> bool { world .calendar @@ -2966,6 +3395,82 @@ mod tests { state } + #[test] + fn shards_split_resources_and_entities_without_overlap() { + let single = World::new(42); + let shards = connected_shards(3); + let width = single.width; + for (index, res) in single.res.iter().enumerate() { + let owners = shards.iter().filter(|w| w.res[index].0 != 0).count(); + assert!(owners <= 1, "tile {index} has resources on several shards"); + let position = (index % width, index / width); + if res.0 != 0 && position != single.home { + let expected = &shards[Shard::new(0, 3).unwrap().owner(position.0, width)]; + assert!(position == expected.home || expected.res[index] == *res); + } + } + let mut ids: Vec = shards.iter().flat_map(World::roster).collect(); + let total = ids.len(); + ids.sort_unstable(); + ids.dedup(); + assert_eq!(ids.len(), total, "entity ids must be unique across shards"); + let people: usize = shards.iter().map(|w| w.person_actors.len()).sum(); + assert_eq!(people, single.person_actors.len()); + for world in &shards { + assert!(world.owns(world.home)); + } + } + + #[test] + fn people_cross_shards_with_their_inventory_and_stay_reachable() { + let mut shards = connected_shards(2); + publish_directory(&mut shards); + let id = *shards[0].person_actors.keys().next().unwrap(); + let destination = shards[1].home; + shards[0].tell_person(id, PersonMsg::Edit(PersonEdit::Wood(7))); + shards[0].tell_person(id, PersonMsg::MoveTo(destination)); + shards[0].step(); + assert!(!shards[0].person_actors.contains_key(&id)); + let report = shards[0].take_report(); + assert_eq!(report.moved, vec![(EntityKey::person(id), 1)]); + + shards[1].receive_arrivals(); + assert!(shards[1].person_actors.contains_key(&id)); + let arrived = shards[1] + .observe_people() + .into_iter() + .find(|person| person.id == id) + .unwrap(); + assert_eq!(arrived.wood, 7); + assert!(shards[1].owns(arrived.position())); + + // The old actor forwards messages sent to the person on the shard they left. + shards[0].tell_person(id, PersonMsg::Edit(PersonEdit::Stone(3))); + wait_for_person(&shards[1], id, |person| person.stone == 3); + + // Once it stops, the router forwards them itself until the directory catches up. + shards[0].tick += FORWARDING_TICKS; + shards[0].retire_departures(); + assert!(shards[0].departed.is_empty()); + shards[0].tell_person(id, PersonMsg::Edit(PersonEdit::Stone(4))); + wait_for_person(&shards[1], id, |person| person.stone == 4); + + publish_directory(&mut shards); + shards[0].tell_person(id, PersonMsg::Edit(PersonEdit::Stone(5))); + wait_for_person(&shards[1], id, |person| person.stone == 5); + + // Walking back resumes life on the first shard. + let back = shards[0].home; + shards[1].tell_person(id, PersonMsg::MoveTo(back)); + shards[1].step(); + shards[0].receive_arrivals(); + assert!(shards[0].person_actors.contains_key(&id)); + assert!(!shards[1].person_actors.contains_key(&id)); + publish_directory(&mut shards); + shards[1].tell_person(id, PersonMsg::Edit(PersonEdit::Stone(6))); + wait_for_person(&shards[0], id, |person| person.stone == 6); + } + #[test] fn new_people_progress_through_every_stage_and_die() { let mut world = World::new(42); @@ -3090,7 +3595,7 @@ mod tests { #[test] fn people_can_pathfind_back_home_around_blocked_terrain() { - let mut world = World::new(42); + let mut world = World::new_with_script(42, village_script()); let start = (59, 48); let mut position = start; @@ -3107,7 +3612,7 @@ mod tests { #[test] fn village_survives_until_its_first_weekly_birth_check() { - let mut world = World::new(42); + let mut world = World::new_with_script(42, village_script()); let first_week_tick = world.calendar.ticks_per_day * u64::from(world.calendar.days_per_week); let mut daily = Vec::new(); @@ -3441,8 +3946,13 @@ mod tests { default_config().max_simulation_speed ); - let restored = - World::from_snapshot_with_script(&world.snapshot_json().unwrap(), 42, script).unwrap(); + let restored = World::from_snapshot_with_script( + &world.snapshot_json().unwrap(), + 42, + script, + Shard::SINGLE, + ) + .unwrap(); assert_eq!((restored.width, restored.height), (48, 32)); assert_eq!(simulation_state(&restored), simulation_state(&world)); } diff --git a/src/world/actors.rs b/src/world/actors.rs index 8b3c717..3f916f7 100644 --- a/src/world/actors.rs +++ b/src/world/actors.rs @@ -4,6 +4,7 @@ //! sends them the effects of world-level decisions (movement, gathering, hunting, births) and //! lets them apply entity-local rules (ageing, hunger, eating, energy, dying) themselves. +use super::router::{Admission, EntityKey, Router}; use super::{Animal, CalendarSettings, LifeStage, Partnership, Person, Rng, sample_stage_duration}; use picard::actors::{ Actor, ActorFactoryArgs, ActorRef, ActorRefFactory, ActorSystem, Context as ActorContext, @@ -12,6 +13,7 @@ use picard::actors::{ use serde::{Deserialize, Serialize}; use std::cell::{Cell, RefCell}; use std::collections::HashMap; +use std::sync::Arc; use std::sync::mpsc::{Receiver, Sender as MpscSender, channel}; use std::time::Duration; @@ -93,6 +95,11 @@ pub(super) enum PersonMsg { Edit(PersonEdit), SetPartner(Option), MoveTo((usize, usize)), + /// Hands the person to `shard`, whose region they walked into; this actor then forwards + /// every message to them until it is stopped. + Depart(RequestId, usize), + /// The person walked back into this shard while this actor was still forwarding. + Readmit(Person), } /// One animal tick: the world supplies movement and opportunities, the animal its metabolism. @@ -122,6 +129,8 @@ pub(super) enum AnimalMsg { /// A successful attack; only the first one kills the animal. Hunted(RequestId), SetEnergyWithin(Area, f32), + Depart(RequestId, usize), + Readmit(Animal), } #[derive(Clone, Debug, Deserialize, Serialize)] @@ -149,6 +158,7 @@ pub(super) enum ReplyBody { Hunted(AnimalId, bool), Partnership(Partnership), BirthDue(Partnership, bool), + Departed, } #[derive(Clone, Debug, Deserialize, Serialize)] @@ -166,15 +176,22 @@ fn reply(sender: Sender, request: RequestId, body: ReplyBody) { pub(super) struct PersonActor { person: Person, + router: Arc, + /// The shard this person moved to; set while this actor only forwards messages. + forward_to: Option, } -impl ActorFactoryArgs for PersonActor { - fn create_args(mut person: Person) -> Self { +impl ActorFactoryArgs<(Person, Arc)> for PersonActor { + fn create_args((mut person, router): (Person, Arc)) -> Self { if person.rng == 0 { // Snapshots from before people had their own random stream. person.rng = u64::from(person.id).wrapping_mul(0x9E37_79B9_7F4A_7C15) | 1; } - Self { person } + Self { + person, + router, + forward_to: None, + } } } @@ -254,6 +271,16 @@ impl Actor for PersonActor { type Msg = PersonMsg; fn recv(&mut self, _ctx: &ActorContext, msg: Self::Msg, sender: Sender) { + if let Some(shard) = self.forward_to { + if let PersonMsg::Readmit(person) = msg { + self.person = person; + self.forward_to = None; + } else { + let key = EntityKey::person(self.person.id); + self.router.forward(shard, key, &msg); + } + return; + } match msg { PersonMsg::Observe(request) => { reply(sender, request, ReplyBody::Person(self.person.clone())); @@ -277,18 +304,34 @@ impl Actor for PersonActor { PersonMsg::Edit(edit) => self.edit(edit), PersonMsg::SetPartner(partner_id) => self.person.partner_id = partner_id, PersonMsg::MoveTo(position) => (self.person.x, self.person.y) = position, + PersonMsg::Depart(request, shard) => { + self.router + .admit_at(shard, Admission::Person(self.person.clone())); + self.forward_to = Some(shard); + reply(sender, request, ReplyBody::Departed); + } + PersonMsg::Readmit(person) => self.person = person, } } } pub(super) struct AnimalActor { + key: EntityKey, id: AnimalId, animal: Animal, + router: Arc, + forward_to: Option, } -impl ActorFactoryArgs<(AnimalId, Animal)> for AnimalActor { - fn create_args((id, animal): (AnimalId, Animal)) -> Self { - Self { id, animal } +impl ActorFactoryArgs<(EntityKey, Animal, Arc)> for AnimalActor { + fn create_args((key, animal, router): (EntityKey, Animal, Arc)) -> Self { + Self { + key, + id: key.id, + animal, + router, + forward_to: None, + } } } @@ -330,6 +373,15 @@ impl Actor for AnimalActor { type Msg = AnimalMsg; fn recv(&mut self, _ctx: &ActorContext, msg: Self::Msg, sender: Sender) { + if let Some(shard) = self.forward_to { + if let AnimalMsg::Readmit(animal) = msg { + self.animal = animal; + self.forward_to = None; + } else { + self.router.forward(shard, self.key, &msg); + } + return; + } match msg { AnimalMsg::Observe(request) => { reply( @@ -360,6 +412,13 @@ impl Actor for AnimalActor { self.animal.energy = energy; } } + AnimalMsg::Depart(request, shard) => { + let admission = Admission::Animal(self.key.kind, self.id, self.animal.clone()); + self.router.admit_at(shard, admission); + self.forward_to = Some(shard); + reply(sender, request, ReplyBody::Departed); + } + AnimalMsg::Readmit(animal) => self.animal = animal, } } } diff --git a/src/world/router.rs b/src/world/router.rs new file mode 100644 index 0000000..1c50538 --- /dev/null +++ b/src/world/router.rs @@ -0,0 +1,557 @@ +//! Location-transparent addressing of entity actors. +//! +//! Actors never talk to each other through raw actor references. They send messages to an +//! [`EntityKey`] through the [`Router`], which delivers them to the local actor or, when the +//! entity lives on another shard, serialises them and sends them over Picard remoting. Entities +//! that leave a shard's region are handed to the shard that owns their new position as an +//! [`Admission`]; the departed actor keeps forwarding messages until every shard knows where the +//! entity went. + +use super::actors::{ + AnimalActor, AnimalId, AnimalMsg, PartnershipActor, PartnershipMsg, PersonActor, PersonMsg, +}; +use super::{Animal, Partnership, Person}; +use picard::actors::{ActorRef, ActorRefFactory, ActorReference, ActorSystem, CreateError, Tell}; +use picard::remoting::{RemoteError, RemoteHandler, RemoteRequest, RemoteResponse}; +use serde::{Deserialize, Serialize}; +use std::collections::HashMap; +use std::ops::Range; +use std::sync::{Arc, Mutex, RwLock, Weak}; +use std::time::Duration; + +/// Remote path that accepts entities moving into this shard's region. +pub const ADMIT_PATH: &str = "/shard/admit"; + +#[derive(Clone, Copy, Debug, Deserialize, Serialize, PartialEq, Eq, Hash, PartialOrd, Ord)] +#[serde(rename_all = "snake_case")] +pub enum Kind { + Person, + Rabbit, + Wolf, + Partnership, +} + +impl Kind { + const ALL: [Kind; 4] = [Kind::Person, Kind::Rabbit, Kind::Wolf, Kind::Partnership]; + + fn name(self) -> &'static str { + match self { + Kind::Person => "person", + Kind::Rabbit => "rabbit", + Kind::Wolf => "wolf", + Kind::Partnership => "partnership", + } + } +} + +/// Cluster-wide identity of an entity actor; ids are unique per kind across all shards. +#[derive(Clone, Copy, Debug, Deserialize, Serialize, PartialEq, Eq, Hash, PartialOrd, Ord)] +pub struct EntityKey { + pub kind: Kind, + pub id: u64, +} + +impl EntityKey { + pub fn new(kind: Kind, id: u64) -> Self { + Self { kind, id } + } + + pub fn person(id: u32) -> Self { + Self::new(Kind::Person, u64::from(id)) + } + + /// The actor name, which is also the last segment of its path. + pub fn name(self) -> String { + format!("{}-{}", self.kind.name(), self.id) + } + + pub fn path(self) -> String { + format!("/user/{}", self.name()) + } + + pub fn from_path(path: &str) -> Option { + let (kind, id) = path.strip_prefix("/user/")?.rsplit_once('-')?; + let kind = Kind::ALL.into_iter().find(|k| k.name() == kind)?; + Some(Self::new(kind, id.parse().ok()?)) + } +} + +/// This process's part of the world: shard `index` of `count` owns a vertical strip of columns. +#[derive(Clone, Copy, Debug, Deserialize, Serialize, PartialEq, Eq)] +pub struct Shard { + pub index: usize, + pub count: usize, +} + +impl Shard { + pub const SINGLE: Shard = Shard { index: 0, count: 1 }; + + pub fn new(index: usize, count: usize) -> Option { + (count > 0 && index < count).then_some(Self { index, count }) + } + + /// The columns of shard `index` in a world `width` tiles wide. + pub fn columns_of(&self, index: usize, width: usize) -> Range { + width * index / self.count..width * (index + 1) / self.count + } + + pub fn columns(&self, width: usize) -> Range { + self.columns_of(self.index, width) + } + + /// The shard owning column `x`. + pub fn owner(&self, x: usize, width: usize) -> usize { + (0..self.count) + .find(|&index| self.columns_of(index, width).contains(&x)) + .unwrap_or(self.count - 1) + } + + /// First id of this shard's id sequences; ids advance by [`Shard::count`] so that every + /// shard hands out distinct ids without coordination. + pub fn first_id(&self) -> u64 { + self.index as u64 + 1 + } + + pub fn id_stride(&self) -> u64 { + self.count as u64 + } + + /// This shard's share of a cluster-wide `total`. + pub fn share(&self, total: usize) -> usize { + total / self.count + usize::from(self.index < total % self.count) + } +} + +/// Delivers serialised messages to other shards; implemented by the cluster transport. +pub trait RemoteSender: Send + Sync { + /// Queues `payload` for the actor at `path` on shard `shard`. Messages to one shard are + /// delivered in the order they were queued. + fn send(&self, shard: usize, path: &str, payload: serde_json::Value); + + /// Waits until every queued message has been accepted by its receiving shard. + fn flush(&self); +} + +/// The live actor behind a local entity. +#[derive(Clone)] +pub(super) enum LocalActor { + Person(ActorRef), + Animal(ActorRef), + Partnership(ActorRef), +} + +impl LocalActor { + pub(super) fn stop(self) { + match self { + LocalActor::Person(actor) => actor.stop(), + LocalActor::Animal(actor) => actor.stop(), + LocalActor::Partnership(actor) => actor.stop(), + } + } +} + +/// A message type that can be routed to the entities whose actors accept it. +pub(super) trait Routed: picard::Message + Clone { + fn local(actor: &LocalActor) -> Option<&ActorRef>; +} + +impl Routed for PersonMsg { + fn local(actor: &LocalActor) -> Option<&ActorRef> { + match actor { + LocalActor::Person(actor) => Some(actor), + _ => None, + } + } +} + +impl Routed for AnimalMsg { + fn local(actor: &LocalActor) -> Option<&ActorRef> { + match actor { + LocalActor::Animal(actor) => Some(actor), + _ => None, + } + } +} + +impl Routed for PartnershipMsg { + fn local(actor: &LocalActor) -> Option<&ActorRef> { + match actor { + LocalActor::Partnership(actor) => Some(actor), + _ => None, + } + } +} + +/// An entity's complete state, including its inventory, moving into another shard's region. +#[derive(Clone, Debug, Deserialize, Serialize)] +pub(super) enum Admission { + Person(Person), + Animal(Kind, AnimalId, Animal), +} + +impl Admission { + fn key(&self) -> EntityKey { + match self { + Admission::Person(person) => EntityKey::person(person.id), + Admission::Animal(kind, id, _) => EntityKey::new(*kind, *id), + } + } +} + +/// An admitted entity, waiting for the world to pick it up at the start of its next step. +pub(super) enum Arrival { + Person(Person, ActorRef), + Animal(Kind, AnimalId, ActorRef), +} + +impl Arrival { + fn key(&self) -> EntityKey { + match self { + Arrival::Person(person, _) => EntityKey::person(person.id), + Arrival::Animal(kind, id, ..) => EntityKey::new(*kind, *id), + } + } +} + +pub struct Router { + shard: Shard, + system: ActorSystem, + this: Weak, + dispatch: RemoteHandler, + local: RwLock>, + /// Shard of every entity elsewhere in the cluster, as last published by the gateway. + directory: RwLock>, + /// Where entities that left this shard went, until the directory catches up. + forwarding: RwLock>, + remote: RwLock>>, + arrivals: Mutex>, +} + +impl Router { + pub fn new(shard: Shard, system: &ActorSystem) -> Arc { + Arc::new_cyclic(|this| Self { + shard, + system: system.clone(), + this: this.clone(), + dispatch: system.remote_handler(), + local: RwLock::default(), + directory: RwLock::default(), + forwarding: RwLock::default(), + remote: RwLock::default(), + arrivals: Mutex::default(), + }) + } + + /// Identity of this shard's actor system, which remote requests must be addressed to. + pub fn system_id(&self) -> uuid::Uuid { + self.system.id() + } + + pub fn set_remote(&self, remote: Arc) { + *self.remote.write().expect("router lock") = Some(remote); + } + + fn arc(&self) -> Arc { + self.this.upgrade().expect("the router outlives its actors") + } + + pub(super) fn register(&self, key: EntityKey, actor: LocalActor) { + self.forwarding.write().expect("router lock").remove(&key); + self.local.write().expect("router lock").insert(key, actor); + } + + pub(super) fn unregister(&self, key: EntityKey) { + self.local.write().expect("router lock").remove(&key); + } + + /// Remembers that `key` moved to `shard` once its forwarding actor is gone. + pub(super) fn set_forwarding(&self, key: EntityKey, shard: usize) { + self.forwarding + .write() + .expect("router lock") + .insert(key, shard); + } + + /// Applies directory changes published by the gateway; `None` means the entity is gone. + pub fn apply_directory(&self, changes: &[(EntityKey, Option)]) { + let mut directory = self.directory.write().expect("router lock"); + let mut forwarding = self.forwarding.write().expect("router lock"); + for &(key, shard) in changes { + forwarding.remove(&key); + match shard { + Some(shard) if shard != self.shard.index => { + directory.insert(key, shard); + } + _ => { + directory.remove(&key); + } + } + } + } + + /// Whether the entity exists anywhere in the cluster, as far as this shard knows. + pub fn knows(&self, key: EntityKey) -> bool { + self.local.read().expect("router lock").contains_key(&key) + || self.remote_location(key).is_some() + } + + fn remote_location(&self, key: EntityKey) -> Option { + let forwarded = self + .forwarding + .read() + .expect("router lock") + .get(&key) + .copied(); + forwarded + .or_else(|| { + self.directory + .read() + .expect("router lock") + .get(&key) + .copied() + }) + .filter(|&shard| shard != self.shard.index) + } + + /// Delivers `message` to the entity, wherever in the cluster it is. + pub(super) fn tell(&self, key: EntityKey, message: M) { + let local = self + .local + .read() + .expect("router lock") + .get(&key) + .and_then(M::local) + .cloned(); + if let Some(actor) = local { + actor.tell(message, None); + } else if let Some(shard) = self.remote_location(key) { + self.forward(shard, key, &message); + } else { + tracing::debug!(entity = %key.name(), "dropped a message for an unknown entity"); + } + } + + /// Sends `message` to the entity on `shard`, bypassing the local registry. + pub(super) fn forward(&self, shard: usize, key: EntityKey, message: &M) { + match serde_json::to_value(message) { + Ok(payload) => self.send_remote(shard, &key.path(), payload), + Err(error) => tracing::warn!(%error, "failed to serialise a routed message"), + } + } + + fn send_remote(&self, shard: usize, path: &str, payload: serde_json::Value) { + let remote = self.remote.read().expect("router lock").clone(); + match remote { + Some(remote) => remote.send(shard, path, payload), + None => tracing::debug!(path, "dropped a message for a shard without a connection"), + } + } + + /// Hands an entity to the shard owning its new position. + pub(super) fn admit_at(&self, shard: usize, admission: Admission) { + if shard == self.shard.index { + self.admit(admission); + return; + } + match serde_json::to_value(&admission) { + Ok(payload) => self.send_remote(shard, ADMIT_PATH, payload), + Err(error) => tracing::warn!(%error, "failed to serialise an admission"), + } + } + + /// Takes in an entity: an actor still forwarding for it resumes, otherwise a new one starts. + pub(super) fn admit(&self, admission: Admission) { + // Holding the arrivals lock keeps `retire_forwarder` from stopping a resumed actor. + let mut arrivals = self.arrivals.lock().expect("router lock"); + let key = admission.key(); + let existing = self.local.read().expect("router lock").get(&key).cloned(); + let arrival = match (admission, existing) { + (Admission::Person(person), Some(LocalActor::Person(actor))) => { + actor.tell(PersonMsg::Readmit(person.clone()), None); + Arrival::Person(person, actor) + } + (Admission::Person(person), _) => { + let Some(actor) = self.spawn_person(person.clone()) else { + return; + }; + Arrival::Person(person, actor) + } + (Admission::Animal(kind, id, animal), Some(LocalActor::Animal(actor))) => { + actor.tell(AnimalMsg::Readmit(animal), None); + Arrival::Animal(kind, id, actor) + } + (Admission::Animal(kind, id, animal), _) => { + let Some(actor) = self.spawn_animal(kind, id, animal) else { + return; + }; + Arrival::Animal(kind, id, actor) + } + }; + self.forwarding.write().expect("router lock").remove(&key); + arrivals.push(arrival); + } + + /// Replaces the forwarding actor of an entity that moved to `shard` by a forwarding entry; + /// returns `false` when the entity has come back in the meantime and must keep its actor. + pub(super) fn retire_forwarder(&self, key: EntityKey, shard: usize) -> bool { + let arrivals = self.arrivals.lock().expect("router lock"); + let returned = arrivals.iter().any(|arrival| arrival.key() == key); + if !returned { + self.unregister(key); + self.set_forwarding(key, shard); + } + !returned + } + + pub(super) fn take_arrivals(&self) -> Vec { + std::mem::take(&mut *self.arrivals.lock().expect("router lock")) + } + + /// Starts an actor, retrying while a stopped actor with the same name is still shutting down. + fn start( + &self, + key: EntityKey, + create: impl Fn(&str) -> Result, CreateError>, + ) -> Option> { + let name = key.name(); + let mut attempts = 0; + loop { + match create(&name) { + Ok(actor) => return Some(actor), + Err(CreateError::AlreadyExists(_)) if attempts < 200 => { + attempts += 1; + std::thread::sleep(Duration::from_millis(5)); + } + Err(error) => { + tracing::error!(entity = %name, %error, "failed to start an entity actor"); + return None; + } + } + } + } + + pub(super) fn spawn_person(&self, person: Person) -> Option> { + let key = EntityKey::person(person.id); + let actor = self.start(key, |name| { + self.system + .actor_of_args::(name, (person.clone(), self.arc())) + })?; + self.register(key, LocalActor::Person(actor.clone())); + Some(actor) + } + + pub(super) fn spawn_animal( + &self, + kind: Kind, + id: AnimalId, + animal: Animal, + ) -> Option> { + let key = EntityKey::new(kind, id); + let actor = self.start(key, |name| { + self.system + .actor_of_args::(name, (key, animal.clone(), self.arc())) + })?; + self.register(key, LocalActor::Animal(actor.clone())); + Some(actor) + } + + pub(super) fn spawn_partnership( + &self, + partnership: Partnership, + ) -> Option> { + let key = EntityKey::new(Kind::Partnership, u64::from(partnership.id)); + let actor = self.start(key, |name| { + self.system + .actor_of_args::(name, partnership.clone()) + })?; + self.register(key, LocalActor::Partnership(actor.clone())); + Some(actor) + } + + /// Answers a remoting request from another shard: admissions are taken in directly, entity + /// messages go to Picard's remote dispatch, and messages for entities that are no longer + /// here are forwarded to where they went. + pub fn handle(&self, peer: &str, request: RemoteRequest) -> RemoteResponse { + let id = request.id; + if request.destination.system_id != self.system.id() { + return RemoteResponse::rejected( + id, + RemoteError::Rejected("destination system ID does not match this shard".into()), + ); + } + if request.destination.path == ADMIT_PATH { + return match serde_json::from_value::(request.payload) { + Ok(admission) => { + self.admit(admission); + RemoteResponse::accepted(id) + } + Err(error) => { + RemoteResponse::rejected(id, RemoteError::InvalidMessage(error.to_string())) + } + }; + } + let path = request.destination.path.clone(); + let payload = request.payload.clone(); + let response = (self.dispatch)(peer, request); + if !response.accepted + && matches!( + response.error, + Some(RemoteError::ActorNotFound(_) | RemoteError::Rejected(_)) + ) + && let Some(key) = EntityKey::from_path(&path) + && let Some(shard) = self.remote_location(key) + { + self.send_remote(shard, &path, payload); + return RemoteResponse::accepted(id); + } + response + } + + /// Waits until every message queued for other shards has been accepted. + pub fn flush(&self) { + let remote = self.remote.read().expect("router lock").clone(); + if let Some(remote) = remote { + remote.flush(); + } + } +} + +#[cfg(test)] +mod tests { + use super::{EntityKey, Kind, Shard}; + + #[test] + fn shards_partition_the_columns_and_hand_out_distinct_ids() { + for (width, count) in [(192, 3), (10, 3), (97, 4), (8, 1)] { + let mut owners = Vec::new(); + for index in 0..count { + let shard = Shard::new(index, count).unwrap(); + for x in shard.columns(width) { + assert_eq!(shard.owner(x, width), index); + owners.push(x); + } + } + assert_eq!(owners, (0..width).collect::>()); + } + let ids: Vec = (0..3) + .flat_map(|index| { + let shard = Shard::new(index, 3).unwrap(); + (0..4).map(move |n| shard.first_id() + n * shard.id_stride()) + }) + .collect(); + let mut unique = ids.clone(); + unique.sort_unstable(); + unique.dedup(); + assert_eq!(unique.len(), ids.len()); + assert_eq!(Shard::new(1, 3).unwrap().share(10), 3); + assert_eq!(Shard::new(0, 3).unwrap().share(10), 4); + } + + #[test] + fn entity_keys_round_trip_through_actor_paths() { + let key = EntityKey::new(Kind::Partnership, 42); + assert_eq!(key.path(), "/user/partnership-42"); + assert_eq!(EntityKey::from_path(&key.path()), Some(key)); + assert_eq!(EntityKey::from_path("/user/world-inbox"), None); + } +}