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); + } +}