Host und Gast im Kern
Der Host nimmt Gäste auf, macht aus ihren Aktionen Befehle für den Spieler ihrer Verbindung, begrenzt sie pro Takt und schickt jeden Takt an alle Gäste. Gäste rechnen jeden Takt nach, vergleichen die Prüfsummen und holen sich nach einer Abweichung die Welt des Hosts neu. Tests verbinden Host und Gäste im Speicher, auch über 2000 Takte mit Gehen und Wiederkommen. Co-Authored-By: Claude Haiku 4.5 <noreply@anthropic.com>
This commit is contained in:
Steven Wroblewski
@@ -0,0 +1,231 @@
|
||||
//! A guest's side of a shared world. The guest gets the host's world once
|
||||
//! and then runs every tick the host sends, with the same commands, so that
|
||||
//! both worlds stay the same. Checksums from the host show whether they do.
|
||||
|
||||
use std::collections::VecDeque;
|
||||
use std::fmt;
|
||||
|
||||
use crate::codec::DecodeError;
|
||||
use crate::command::Command;
|
||||
use crate::player::{clamp_name, PlayerId, PlayerKey};
|
||||
use crate::protocol::{Action, Refusal, ToGuest, ToHost};
|
||||
use crate::replay::Recording;
|
||||
use crate::rules::rules_id;
|
||||
use crate::snapshot::SnapshotError;
|
||||
use crate::world::{Event, World};
|
||||
|
||||
/// Reason a guest cannot go on with the session.
|
||||
#[derive(Clone, Debug, PartialEq, Eq)]
|
||||
pub enum GuestError {
|
||||
/// A message from the host could not be read.
|
||||
Unreadable(DecodeError),
|
||||
/// The world the host sent could not be restored.
|
||||
BadWorld(SnapshotError),
|
||||
/// The host sent a tick before it sent its world.
|
||||
TickBeforeWorld,
|
||||
}
|
||||
|
||||
impl fmt::Display for GuestError {
|
||||
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
|
||||
match self {
|
||||
GuestError::Unreadable(error) => write!(f, "unreadable message from the host: {error}"),
|
||||
GuestError::BadWorld(error) => write!(f, "the host's world is unusable: {error}"),
|
||||
GuestError::TickBeforeWorld => write!(f, "the host sent a tick before its world"),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl std::error::Error for GuestError {}
|
||||
|
||||
/// What a message from the host changed, apart from queued ticks.
|
||||
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
|
||||
pub enum Arrival {
|
||||
/// Nothing the shell has to react to right away.
|
||||
Nothing,
|
||||
/// The host's world arrived. `replaced` is true when it replaced a world
|
||||
/// that had diverged.
|
||||
World { replaced: bool },
|
||||
/// The host turned the guest away.
|
||||
Refused(Refusal),
|
||||
}
|
||||
|
||||
/// A tick from the host that has not run yet.
|
||||
struct PendingTick {
|
||||
tick: u64,
|
||||
commands: Vec<Command>,
|
||||
check: Option<u64>,
|
||||
}
|
||||
|
||||
/// The world the guest follows, with the recording of everything it ran.
|
||||
struct Following {
|
||||
world: World,
|
||||
recording: Recording,
|
||||
}
|
||||
|
||||
/// The guest side of a shared world.
|
||||
pub struct Guest {
|
||||
key: PlayerKey,
|
||||
following: Option<Following>,
|
||||
/// True after a divergence, until the host's world arrives again.
|
||||
diverged: bool,
|
||||
pending: VecDeque<PendingTick>,
|
||||
outbox: Vec<Vec<u8>>,
|
||||
diverged_recording: Option<Vec<u8>>,
|
||||
}
|
||||
|
||||
impl Guest {
|
||||
/// A guest that greets the host as the player `key` named `name`. The
|
||||
/// greeting waits in the outbox.
|
||||
pub fn new(key: PlayerKey, name: &str) -> Self {
|
||||
let hello = ToHost::Hello {
|
||||
rules: rules_id(),
|
||||
key,
|
||||
name: clamp_name(name),
|
||||
};
|
||||
Self {
|
||||
key,
|
||||
following: None,
|
||||
diverged: false,
|
||||
pending: VecDeque::new(),
|
||||
outbox: vec![hello.encode()],
|
||||
diverged_recording: None,
|
||||
}
|
||||
}
|
||||
|
||||
/// The world, once the host has sent it.
|
||||
pub fn world(&self) -> Option<&World> {
|
||||
self.following.as_ref().map(|following| &following.world)
|
||||
}
|
||||
|
||||
/// The recording of every tick this guest ran since the host's world
|
||||
/// last arrived.
|
||||
pub fn recording(&self) -> Option<&Recording> {
|
||||
self.following
|
||||
.as_ref()
|
||||
.map(|following| &following.recording)
|
||||
}
|
||||
|
||||
/// The guest's player, once its join has run.
|
||||
pub fn local_player(&self) -> Option<PlayerId> {
|
||||
let world = self.world()?;
|
||||
let id = world.find_player(&self.key)?;
|
||||
world.player(id)?.online.then_some(id)
|
||||
}
|
||||
|
||||
/// Handles a message from the host.
|
||||
pub fn receive(&mut self, bytes: &[u8]) -> Result<Arrival, GuestError> {
|
||||
match ToGuest::decode(bytes).map_err(GuestError::Unreadable)? {
|
||||
ToGuest::World { snapshot } => {
|
||||
let world = World::from_snapshot(&snapshot).map_err(GuestError::BadWorld)?;
|
||||
let replaced = self.diverged;
|
||||
if let Some(old) = self.following.take() {
|
||||
if replaced {
|
||||
self.diverged_recording = Some(old.recording.to_bytes(&old.world));
|
||||
}
|
||||
}
|
||||
// Ticks sent before this world are part of it already.
|
||||
let tick = world.tick();
|
||||
self.pending.retain(|pending| pending.tick >= tick);
|
||||
self.following = Some(Following {
|
||||
recording: Recording::start(&world),
|
||||
world,
|
||||
});
|
||||
self.diverged = false;
|
||||
Ok(Arrival::World { replaced })
|
||||
}
|
||||
ToGuest::Refused(refusal) => Ok(Arrival::Refused(refusal)),
|
||||
ToGuest::Tick {
|
||||
tick,
|
||||
commands,
|
||||
check,
|
||||
} => {
|
||||
if self.following.is_none() {
|
||||
return Err(GuestError::TickBeforeWorld);
|
||||
}
|
||||
self.pending.push_back(PendingTick {
|
||||
tick,
|
||||
commands,
|
||||
check,
|
||||
});
|
||||
Ok(Arrival::Nothing)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Ticks that arrived and have not run yet.
|
||||
pub fn pending_ticks(&self) -> usize {
|
||||
self.pending.len()
|
||||
}
|
||||
|
||||
/// Runs the oldest tick that arrived and returns what happened, or
|
||||
/// `None` if there is none to run. After a divergence no tick runs until
|
||||
/// the host's world arrives again.
|
||||
pub fn step(&mut self) -> Option<Vec<Event>> {
|
||||
if self.diverged {
|
||||
return None;
|
||||
}
|
||||
let following = self.following.as_mut()?;
|
||||
while let Some(next) = self.pending.pop_front() {
|
||||
let expected = following.world.tick();
|
||||
if next.tick < expected {
|
||||
continue;
|
||||
}
|
||||
if next.tick > expected {
|
||||
// A tick is missing: this world can no longer follow.
|
||||
self.diverge();
|
||||
return None;
|
||||
}
|
||||
let events = following
|
||||
.recording
|
||||
.step(&mut following.world, &next.commands);
|
||||
if next
|
||||
.check
|
||||
.is_some_and(|check| check != following.world.state_hash())
|
||||
{
|
||||
self.diverge();
|
||||
}
|
||||
return Some(events);
|
||||
}
|
||||
None
|
||||
}
|
||||
|
||||
/// Reports the current state to the host and stops running ticks until
|
||||
/// the host's world arrives again.
|
||||
fn diverge(&mut self) {
|
||||
let Some(world) = self.world() else {
|
||||
return;
|
||||
};
|
||||
let report = ToHost::Diverged {
|
||||
tick: world.tick(),
|
||||
hash: world.state_hash(),
|
||||
};
|
||||
self.outbox.push(report.encode());
|
||||
self.diverged = true;
|
||||
}
|
||||
|
||||
/// True after a divergence, until the host's world arrives again.
|
||||
pub fn has_diverged(&self) -> bool {
|
||||
self.diverged
|
||||
}
|
||||
|
||||
/// Asks the host to carry out an action of the guest's player. Does
|
||||
/// nothing before the player is in the world or while the world is
|
||||
/// diverged.
|
||||
pub fn act(&mut self, action: Action) {
|
||||
if self.diverged || self.local_player().is_none() {
|
||||
return;
|
||||
}
|
||||
self.outbox.push(ToHost::Act(action).encode());
|
||||
}
|
||||
|
||||
/// Messages for the host, in the order they were made.
|
||||
pub fn take_outbox(&mut self) -> Vec<Vec<u8>> {
|
||||
std::mem::take(&mut self.outbox)
|
||||
}
|
||||
|
||||
/// The encoded recording of the world that diverged last, kept so that
|
||||
/// the divergence can be replayed. Empty after it has been taken.
|
||||
pub fn take_diverged_recording(&mut self) -> Option<Vec<u8>> {
|
||||
self.diverged_recording.take()
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,323 @@
|
||||
//! The host's side of a shared world: the world, its recording and the
|
||||
//! guests connected to it.
|
||||
//!
|
||||
//! The host runs the ticks. Its own player's actions and the actions its
|
||||
//! guests send become commands of the next tick, and every tick goes out to
|
||||
//! every accepted guest. A solo game is a host without guests.
|
||||
|
||||
use std::collections::BTreeMap;
|
||||
|
||||
use crate::command::Command;
|
||||
use crate::player::{clamp_name, PlayerId, PlayerKey, MAX_KNOWN_PLAYERS, MAX_ONLINE_PLAYERS};
|
||||
use crate::protocol::{Action, Refusal, ToGuest, ToHost, CHECK_INTERVAL_TICKS};
|
||||
use crate::replay::Recording;
|
||||
use crate::rules::rules_id;
|
||||
use crate::world::{Event, World};
|
||||
|
||||
/// Identifies a connection. The shell chooses the numbers.
|
||||
pub type Peer = u32;
|
||||
|
||||
/// Most actions one player may take per tick. Further actions in the same
|
||||
/// tick are dropped, so that no guest can flood the world with commands.
|
||||
pub const MAX_ACTIONS_PER_TICK: u32 = 4;
|
||||
|
||||
/// A guest reported that its state differs from the host's checksum.
|
||||
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
|
||||
pub struct Divergence {
|
||||
pub peer: Peer,
|
||||
/// The guest's tick when it noticed.
|
||||
pub tick: u64,
|
||||
/// The hash of the guest's state at that tick.
|
||||
pub hash: u64,
|
||||
}
|
||||
|
||||
/// Where the connection of a guest stands.
|
||||
#[derive(Clone, Debug, PartialEq, Eq)]
|
||||
enum Link {
|
||||
/// Connected, but the guest has not greeted yet.
|
||||
Greeting,
|
||||
/// Accepted. The guest has the world, and its join runs in the next tick.
|
||||
Joining { key: PlayerKey },
|
||||
/// In the world as `player`. `actions` counts its actions in this tick.
|
||||
Playing { player: PlayerId, actions: u32 },
|
||||
}
|
||||
|
||||
/// The host of a shared world.
|
||||
pub struct Host {
|
||||
world: World,
|
||||
recording: Recording,
|
||||
rules: u64,
|
||||
local_key: Option<PlayerKey>,
|
||||
local_actions: u32,
|
||||
/// Commands for the next tick, in the order they arrived.
|
||||
inbox: Vec<Command>,
|
||||
links: BTreeMap<Peer, Link>,
|
||||
outbox: Vec<(Peer, Vec<u8>)>,
|
||||
closing: Vec<Peer>,
|
||||
divergences: Vec<Divergence>,
|
||||
}
|
||||
|
||||
impl Host {
|
||||
/// Hosts `world`, new or loaded. Nobody is in it yet.
|
||||
pub fn new(world: World) -> Self {
|
||||
Self {
|
||||
recording: Recording::start(&world),
|
||||
world,
|
||||
rules: rules_id(),
|
||||
local_key: None,
|
||||
local_actions: 0,
|
||||
inbox: Vec::new(),
|
||||
links: BTreeMap::new(),
|
||||
outbox: Vec::new(),
|
||||
closing: Vec::new(),
|
||||
divergences: Vec::new(),
|
||||
}
|
||||
}
|
||||
|
||||
pub fn world(&self) -> &World {
|
||||
&self.world
|
||||
}
|
||||
|
||||
/// The recording of everything that happened since the world was hosted.
|
||||
pub fn recording(&self) -> &Recording {
|
||||
&self.recording
|
||||
}
|
||||
|
||||
/// Queues the entry of the host's own player for the next tick.
|
||||
pub fn join_local(&mut self, key: PlayerKey, name: &str) {
|
||||
self.local_key = Some(key);
|
||||
self.inbox.push(Command::Join {
|
||||
key,
|
||||
name: clamp_name(name),
|
||||
});
|
||||
}
|
||||
|
||||
/// The host's own player, once it is in the world.
|
||||
pub fn local_player(&self) -> Option<PlayerId> {
|
||||
online_player(&self.world, &self.local_key?)
|
||||
}
|
||||
|
||||
/// Queues an action of the host's own player for the next tick. Does
|
||||
/// nothing before the player is in the world.
|
||||
pub fn act_local(&mut self, action: Action) {
|
||||
let Some(player) = self.local_player() else {
|
||||
return;
|
||||
};
|
||||
if self.local_actions < MAX_ACTIONS_PER_TICK {
|
||||
self.local_actions += 1;
|
||||
self.inbox.push(action.command(player));
|
||||
}
|
||||
}
|
||||
|
||||
/// A guest connected as `peer`. Its first message must be a greeting.
|
||||
pub fn connect(&mut self, peer: Peer) {
|
||||
self.links.entry(peer).or_insert(Link::Greeting);
|
||||
}
|
||||
|
||||
/// The connection `peer` closed. A guest in the world leaves it in the
|
||||
/// next tick; an accepted guest whose join has not run yet never enters.
|
||||
pub fn disconnect(&mut self, peer: Peer) {
|
||||
match self.links.remove(&peer) {
|
||||
Some(Link::Joining { key }) => self.inbox.retain(
|
||||
|command| !matches!(command, Command::Join { key: joining, .. } if *joining == key),
|
||||
),
|
||||
Some(Link::Playing { player, .. }) => self.inbox.push(Command::Leave { player }),
|
||||
Some(Link::Greeting) | None => {}
|
||||
}
|
||||
}
|
||||
|
||||
/// Handles a message that arrived from `peer`. A guest that breaks the
|
||||
/// protocol is closed; see [`Host::take_closing`].
|
||||
pub fn receive(&mut self, peer: Peer, bytes: &[u8]) {
|
||||
let Some(link) = self.links.get(&peer).cloned() else {
|
||||
return;
|
||||
};
|
||||
match (link, ToHost::decode(bytes)) {
|
||||
(Link::Greeting, Ok(ToHost::Hello { rules, key, name })) => {
|
||||
self.greet(peer, rules, key, &name);
|
||||
}
|
||||
(Link::Greeting, Ok(ToHost::ForeignHello { .. })) => {
|
||||
self.refuse(peer, Refusal::OtherProtocol);
|
||||
}
|
||||
(Link::Playing { player, actions }, Ok(ToHost::Act(action))) => {
|
||||
if actions < MAX_ACTIONS_PER_TICK {
|
||||
self.links.insert(
|
||||
peer,
|
||||
Link::Playing {
|
||||
player,
|
||||
actions: actions + 1,
|
||||
},
|
||||
);
|
||||
self.inbox.push(action.command(player));
|
||||
}
|
||||
}
|
||||
// The guest acted before its join ran. It cannot know its player
|
||||
// yet, so the action is dropped.
|
||||
(Link::Joining { .. }, Ok(ToHost::Act(_))) => {}
|
||||
(Link::Joining { .. } | Link::Playing { .. }, Ok(ToHost::Diverged { tick, hash })) => {
|
||||
self.divergences.push(Divergence { peer, tick, hash });
|
||||
self.send(
|
||||
peer,
|
||||
&ToGuest::World {
|
||||
snapshot: self.world.to_snapshot(),
|
||||
},
|
||||
);
|
||||
}
|
||||
_ => {
|
||||
self.disconnect(peer);
|
||||
self.closing.push(peer);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
fn greet(&mut self, peer: Peer, rules: u64, key: PlayerKey, name: &str) {
|
||||
if let Err(refusal) = self.admission(rules, &key) {
|
||||
self.refuse(peer, refusal);
|
||||
return;
|
||||
}
|
||||
self.send(
|
||||
peer,
|
||||
&ToGuest::World {
|
||||
snapshot: self.world.to_snapshot(),
|
||||
},
|
||||
);
|
||||
self.inbox.push(Command::Join {
|
||||
key,
|
||||
name: clamp_name(name),
|
||||
});
|
||||
self.links.insert(peer, Link::Joining { key });
|
||||
}
|
||||
|
||||
/// Whether a guest with `rules` and `key` may join now. Joins that are
|
||||
/// queued for the next tick count as if they had run.
|
||||
fn admission(&self, rules: u64, key: &PlayerKey) -> Result<(), Refusal> {
|
||||
if rules != self.rules {
|
||||
return Err(Refusal::OtherRules);
|
||||
}
|
||||
let queued: Vec<&PlayerKey> = self
|
||||
.inbox
|
||||
.iter()
|
||||
.filter_map(|command| match command {
|
||||
Command::Join { key, .. } => Some(key),
|
||||
_ => None,
|
||||
})
|
||||
.collect();
|
||||
if online_player(&self.world, key).is_some() || queued.contains(&key) {
|
||||
return Err(Refusal::AlreadyPlaying);
|
||||
}
|
||||
let online = self
|
||||
.world
|
||||
.players()
|
||||
.iter()
|
||||
.filter(|player| player.online)
|
||||
.count();
|
||||
let entering = queued
|
||||
.iter()
|
||||
.filter(|queued| online_player(&self.world, queued).is_none())
|
||||
.count();
|
||||
if online + entering >= MAX_ONLINE_PLAYERS {
|
||||
return Err(Refusal::WorldFull);
|
||||
}
|
||||
let new = queued
|
||||
.iter()
|
||||
.filter(|queued| self.world.find_player(queued).is_none())
|
||||
.count();
|
||||
if self.world.find_player(key).is_none()
|
||||
&& self.world.players().len() + new >= MAX_KNOWN_PLAYERS
|
||||
{
|
||||
return Err(Refusal::TooManyPlayers);
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn refuse(&mut self, peer: Peer, refusal: Refusal) {
|
||||
self.send(peer, &ToGuest::Refused(refusal));
|
||||
self.links.remove(&peer);
|
||||
self.closing.push(peer);
|
||||
}
|
||||
|
||||
fn send(&mut self, peer: Peer, message: &ToGuest) {
|
||||
self.outbox.push((peer, message.encode()));
|
||||
}
|
||||
|
||||
/// Runs the next tick with every queued command, sends it to every
|
||||
/// accepted guest and returns what happened.
|
||||
pub fn step(&mut self) -> Vec<Event> {
|
||||
let commands = std::mem::take(&mut self.inbox);
|
||||
let tick = self.world.tick();
|
||||
let events = self.recording.step(&mut self.world, &commands);
|
||||
let check = self
|
||||
.world
|
||||
.tick()
|
||||
.is_multiple_of(CHECK_INTERVAL_TICKS)
|
||||
.then(|| self.world.state_hash());
|
||||
let message = ToGuest::Tick {
|
||||
tick,
|
||||
commands,
|
||||
check,
|
||||
}
|
||||
.encode();
|
||||
for (peer, link) in &self.links {
|
||||
if *link != Link::Greeting {
|
||||
self.outbox.push((*peer, message.clone()));
|
||||
}
|
||||
}
|
||||
|
||||
self.local_actions = 0;
|
||||
let mut refused = Vec::new();
|
||||
for (peer, link) in self.links.iter_mut() {
|
||||
match link {
|
||||
Link::Joining { key } => match online_player(&self.world, key) {
|
||||
Some(player) => *link = Link::Playing { player, actions: 0 },
|
||||
None => refused.push(*peer),
|
||||
},
|
||||
Link::Playing { actions, .. } => *actions = 0,
|
||||
Link::Greeting => {}
|
||||
}
|
||||
}
|
||||
// The admission check keeps the world from refusing an accepted
|
||||
// guest. Should it happen anyway, the guest learns why.
|
||||
for peer in refused {
|
||||
self.refuse(peer, Refusal::WorldFull);
|
||||
}
|
||||
events
|
||||
}
|
||||
|
||||
/// Number of accepted guests, including those whose join has not run.
|
||||
pub fn guest_count(&self) -> usize {
|
||||
self.links
|
||||
.values()
|
||||
.filter(|link| **link != Link::Greeting)
|
||||
.count()
|
||||
}
|
||||
|
||||
/// The player of the guest connected as `peer`, once it is in the world.
|
||||
pub fn player_of(&self, peer: Peer) -> Option<PlayerId> {
|
||||
match self.links.get(&peer) {
|
||||
Some(Link::Playing { player, .. }) => Some(*player),
|
||||
_ => None,
|
||||
}
|
||||
}
|
||||
|
||||
/// Messages to send, as (peer, bytes), in the order they were made.
|
||||
pub fn take_outbox(&mut self) -> Vec<(Peer, Vec<u8>)> {
|
||||
std::mem::take(&mut self.outbox)
|
||||
}
|
||||
|
||||
/// Connections the host has given up. The shell closes them after it
|
||||
/// has sent the messages for them.
|
||||
pub fn take_closing(&mut self) -> Vec<Peer> {
|
||||
std::mem::take(&mut self.closing)
|
||||
}
|
||||
|
||||
/// Divergences the guests reported since the last call.
|
||||
pub fn take_divergences(&mut self) -> Vec<Divergence> {
|
||||
std::mem::take(&mut self.divergences)
|
||||
}
|
||||
}
|
||||
|
||||
/// The player with `key`, if it is online in `world`.
|
||||
fn online_player(world: &World, key: &PlayerKey) -> Option<PlayerId> {
|
||||
let id = world.find_player(key)?;
|
||||
world.player(id)?.online.then_some(id)
|
||||
}
|
||||
@@ -8,7 +8,9 @@
|
||||
|
||||
pub mod codec;
|
||||
pub mod command;
|
||||
pub mod guest;
|
||||
pub mod hash;
|
||||
pub mod host;
|
||||
pub mod inventory;
|
||||
pub mod map;
|
||||
pub mod mapgen;
|
||||
|
||||
Reference in New Issue
Block a user