diff --git a/application/application.cpp b/application/application.cpp index a99a4662..b9d89ac0 100644 --- a/application/application.cpp +++ b/application/application.cpp @@ -461,6 +461,15 @@ void eu07_application::request_vehicle_leave(network::NetworkEntityId const Enti m_network->request_leave(Entity); } +namespace { + +// how many steps in a row may disagree with the authority before we ask to be corrected, +// and how long to leave the correction alone before asking again +uint64_t const NETWORK_RESYNC_THRESHOLD = 120; +uint64_t const NETWORK_RESYNC_COOLDOWN = 600; + +} // namespace + void eu07_application::network_scenario_loaded() { if (m_network) @@ -595,6 +604,16 @@ int eu07_application::run() logtype::net); } Global.desync = (float)m_statemismatches; + + // a short disagreement rides itself out; a lasting one will not, + // and calls for the authoritative state rather than a restart + if ((m_statemismatches >= NETWORK_RESYNC_THRESHOLD) && + (m_lastresynctick == 0 || authoritative.tick > m_lastresynctick + NETWORK_RESYNC_COOLDOWN)) + { + m_lastresynctick = authoritative.tick; + WriteLog("net: resync requested at tick " + std::to_string(authoritative.tick), logtype::net); + m_network->request_resync(authoritative.tick, statehash); + } } else { diff --git a/application/application.h b/application/application.h index 5010a83b..bf6d69ad 100644 --- a/application/application.h +++ b/application/application.h @@ -125,6 +125,8 @@ private: bool m_screenshot_queued = false; // consecutive authoritative steps whose state digest did not match ours std::uint64_t m_statemismatches { 0 }; + // tick at which we last asked the server to put us back in line + std::uint64_t m_lastresynctick { 0 }; modeptr_array m_modes { nullptr }; // collection of available application behaviour modes mode_stack m_modestack; // current behaviour mode diff --git a/network/manager.cpp b/network/manager.cpp index 662f5ab3..f4da38a9 100644 --- a/network/manager.cpp +++ b/network/manager.cpp @@ -88,6 +88,7 @@ void network::server_manager::publish_vehicle_list() void network::server_manager::update_crews() { + Crews.expire_absences(); Crews.reconcile(); if (--crew_publish_countdown <= 0) { @@ -157,6 +158,12 @@ void network::manager::notify_scenario_loaded() client->send_ready(); } +void network::manager::request_resync(uint64_t tick, uint64_t state_hash) +{ + if (client) + client->send_resync_request(tick, state_hash); +} + void network::manager::update() { for (auto &backend : backend_list()) diff --git a/network/manager.h b/network/manager.h index ab91affa..414153bb 100644 --- a/network/manager.h +++ b/network/manager.h @@ -54,5 +54,8 @@ namespace network // the local scenario is loaded; a client asks the server for a snapshot of the // world as it stands right now void notify_scenario_loaded(); + + // our world drifted away from the authoritative one; ask to be corrected + void request_resync(uint64_t tick, uint64_t state_hash); }; } diff --git a/network/message.cpp b/network/message.cpp index d7975611..98c78170 100644 --- a/network/message.cpp +++ b/network/message.cpp @@ -7,6 +7,7 @@ void network::client_hello::serialize(std::ostream &stream) const sn_utils::ls_int32(stream, version); sn_utils::ls_uint32(stream, start_packet); sn_utils::s_str(stream, app_version); + sn_utils::ls_uint64(stream, session_token); } void network::client_hello::deserialize(std::istream &stream) @@ -14,6 +15,7 @@ void network::client_hello::deserialize(std::istream &stream) version = sn_utils::ld_int32(stream); start_packet = sn_utils::ld_uint32(stream); app_version = sn_utils::d_str(stream); + session_token = sn_utils::ld_uint64(stream); } void network::server_reject::serialize(std::ostream &stream) const @@ -34,6 +36,7 @@ void network::server_hello::serialize(std::ostream &stream) const sn_utils::s_str(stream, scenario); sn_utils::s_str(stream, app_version); sn_utils::ls_uint32(stream, peer_id); + sn_utils::ls_uint64(stream, session_token); } void network::server_hello::deserialize(std::istream &stream) @@ -44,6 +47,7 @@ void network::server_hello::deserialize(std::istream &stream) scenario = sn_utils::d_str(stream); app_version = sn_utils::d_str(stream); peer_id = sn_utils::ld_uint32(stream); + session_token = sn_utils::ld_uint64(stream); } void network::vehicle_list::serialize(std::ostream &stream) const @@ -106,6 +110,18 @@ void network::snapshot::deserialize(std::istream &stream) stream.read(blob.data(), size); } +void network::request_resync::serialize(std::ostream &stream) const +{ + sn_utils::ls_uint64(stream, tick); + sn_utils::ls_uint64(stream, state_hash); +} + +void network::request_resync::deserialize(std::istream &stream) +{ + tick = sn_utils::ld_uint64(stream); + state_hash = sn_utils::ld_uint64(stream); +} + void network::claim_vehicle::serialize(std::ostream &stream) const { sn_utils::ls_uint32(stream, entity_id); @@ -277,6 +293,8 @@ std::shared_ptr network::deserialize_message(std::istream &str msg = std::make_shared(); else if (type == message::SNAPSHOT) msg = std::make_shared(); + else if (type == message::REQUEST_RESYNC) + msg = std::make_shared(); if (!msg) { // unknown message type; hand back a marker the peer handlers treat as a protocol error diff --git a/network/message.h b/network/message.h index cca3bdc7..6a28db8b 100644 --- a/network/message.h +++ b/network/message.h @@ -23,6 +23,7 @@ struct message CREW_UPDATE, CLIENT_READY, SNAPSHOT, + REQUEST_RESYNC, TYPE_MAX }; @@ -44,6 +45,10 @@ struct client_hello : public message uint32_t start_packet; // build identification of the connecting simulator, informational only std::string app_version; + // identity handed out by this server on an earlier connection, echoed back so that a + // player who dropped out gets their seat and their peer id back. it is a reconnect + // token, not a credential - it says "this is the same session", nothing more + uint64_t session_token{ 0 }; }; // sent by the server instead of SERVER_HELLO when the client cannot join; @@ -70,6 +75,8 @@ struct server_hello : public message std::string app_version; // identity assigned to the joining peer for the rest of the session uint32_t peer_id{ PEER_NONE }; + // hand this back on a later connection to be recognised as the same participant + uint64_t session_token{ 0 }; virtual void serialize(std::ostream &stream) const override; virtual void deserialize(std::istream &stream) override; @@ -111,6 +118,19 @@ struct snapshot : public message virtual void deserialize(std::istream &stream) override; }; +// a peer telling the server its world has drifted too far to carry on, and asking to be +// put back in line +struct request_resync : public message +{ + request_resync() : message(REQUEST_RESYNC) {} + + uint64_t tick{ 0 }; + uint64_t state_hash{ 0 }; + + virtual void serialize(std::ostream &stream) const override; + virtual void deserialize(std::istream &stream) override; +}; + // a peer asking the server to be put on the crew of a vehicle struct claim_vehicle : public message { diff --git a/network/network.cpp b/network/network.cpp index 6ae3886d..0ef5de18 100644 --- a/network/network.cpp +++ b/network/network.cpp @@ -53,6 +53,7 @@ void network::connection::connected() msg.version = EU07_NETWORK_VERSION; msg.start_packet = packet_counter; msg.app_version = Global.asVersion; + msg.session_token = Global.network_session_token; send_message(msg); } } @@ -93,9 +94,10 @@ void network::server::prune_clients() if ((*it)->state == connection::DEAD) { if ((*it)->peer_id != PEER_NONE) { WriteLog("net: peer " + std::to_string((*it)->peer_id) + " disconnected", logtype::net); - // a vehicle keeps running for whoever is left on board; it only falls back - // to the AI when the crew becomes empty, which Crews decides on its own - Crews.drop_peer((*it)->peer_id); + // the seat is held for a while in case the peer comes straight back. a + // vehicle keeps running for whoever is left on board either way; it only + // falls back to the AI once the crew really is empty + Crews.suspend_peer((*it)->peer_id); } it = clients.erase(it); continue; @@ -169,7 +171,8 @@ void network::server::handle_message(std::shared_ptr conn, const mes reply.config = 0; // TODO: pass bitfield with state of relevant setting switches reply.scenario = Global.SceneryFile; reply.app_version = Global.asVersion; - reply.peer_id = allocate_peer_id(); + reply.session_token = cmd.session_token; + reply.peer_id = resolve_peer_identity(reply.session_token); conn->peer_id = reply.peer_id; // the peer now goes and loads the scenario. it gets the state of the world as a // snapshot when it reports back, instead of replaying the session frame by frame @@ -195,6 +198,11 @@ void network::server::handle_message(std::shared_ptr conn, const mes return; } + // the roster first, so that the crew section of the snapshot resolves to names + vehicle_list roster; + roster.vehicles = Entities.entries(); + conn->send_message(roster); + snapshot reply; reply.tick = Global.simulation_tick; reply.blob = take_snapshot(); @@ -209,6 +217,17 @@ void network::server::handle_message(std::shared_ptr conn, const mes // and it still has to learn who is where Crews.mark_all_pending(); } + else if (msg.type == message::REQUEST_RESYNC) { + const auto& cmd = dynamic_cast(msg); + + WriteLog("net: resync requested by peer " + std::to_string(conn->peer_id) + " at tick " + + std::to_string(cmd.tick) + " (its digest " + std::to_string(cmd.state_hash) + ")", logtype::net); + + snapshot reply; + reply.tick = Global.simulation_tick; + reply.blob = take_snapshot(); + conn->send_message(reply); + } else if (msg.type == message::CLAIM_VEHICLE) { const auto& cmd = dynamic_cast(msg); @@ -355,6 +374,17 @@ void network::client::send_ready() WriteLog("net: scenario loaded, asking the server for a snapshot", logtype::net); } +void network::client::send_resync_request(uint64_t tick, uint64_t state_hash) +{ + if (!conn || conn->state != connection::ACTIVE) + return; + + request_resync msg; + msg.tick = tick; + msg.state_hash = state_hash; + conn->send_message(msg); +} + void network::client::send_claim(NetworkEntityId entity_id) { if (!conn || conn->state != connection::ACTIVE) @@ -419,6 +449,11 @@ void network::client::handle_message(std::shared_ptr conn, const mes WriteLog("net: assigned peer id " + std::to_string(cmd.peer_id), logtype::net); } + if (cmd.session_token != 0) { + // kept for the next connection, so that a dropout does not cost us our seat + Global.network_session_token = cmd.session_token; + } + Global.network_status.clear(); WriteLog("net: accept received", logtype::net); @@ -438,6 +473,15 @@ void network::client::handle_message(std::shared_ptr conn, const mes if (apply_snapshot(cmd.blob)) { Global.network_snapshot_applied = true; + + // the roster may say we are on a crew already - after a reconnect, or because + // the claim was granted while we were still loading. walk into that cab + auto const own = Crews.vehicle_of(Global.network_peer_id); + if (own != ENTITY_NONE) { + auto const name = Entities.name_of(own); + if (!name.empty()) + Global.network_pending_vehicle = name; + } } else { ErrorLog("net: could not apply the snapshot handed to us", logtype::net); diff --git a/network/network.h b/network/network.h index f302f98c..1d3603dc 100644 --- a/network/network.h +++ b/network/network.h @@ -124,6 +124,8 @@ namespace network void send_commands(command_queue::commands_map commands); // tells the server the scenario is loaded and a snapshot can be applied void send_ready(); + // our world has drifted too far to carry on; ask to be put back in line + void send_resync_request(uint64_t tick, uint64_t state_hash); // lobby requests; the server is the one that decides void send_claim(NetworkEntityId entity_id); void send_leave(NetworkEntityId entity_id); diff --git a/network/session.cpp b/network/session.cpp index 46a8fea7..80118fca 100644 --- a/network/session.cpp +++ b/network/session.cpp @@ -184,6 +184,73 @@ bool crew_registry::leave(PeerId Peer, NetworkEntityId Id) return true; } +PeerId resolve_peer_identity(uint64_t &Token) +{ + static std::unordered_map known; + + if (Token != 0) + { + auto const lookup = known.find(Token); + if (lookup != known.end()) + { + WriteLog("net: peer " + std::to_string(lookup->second) + " recognised, reconnecting", logtype::net); + Crews.resume_peer(lookup->second); + return lookup->second; + } + } + + auto const peer = allocate_peer_id(); + + // only has to be hard to collide with, not hard to guess: it decides who is who + // across a dropped connection, nothing more + Token = ((uint64_t)Global.local_random_engine() << 32) ^ (uint64_t)Global.local_random_engine() ^ ((uint64_t)peer << 8); + if (Token == 0) + Token = peer; + + known.emplace(Token, peer); + + return peer; +} + +void crew_registry::suspend_peer(PeerId Peer) +{ + if (vehicle_of(Peer) == ENTITY_NONE) + return; + + m_absent[Peer] = Timer::GetTime() + RECONNECT_GRACE_SECONDS; + + WriteLog("net: peer " + std::to_string(Peer) + " lost its connection, holding its seat for " + std::to_string((int)RECONNECT_GRACE_SECONDS) + " s", logtype::net); +} + +void crew_registry::resume_peer(PeerId Peer) +{ + m_absent.erase(Peer); +} + +bool crew_registry::is_absent(PeerId Peer) const +{ + return m_absent.find(Peer) != m_absent.end(); +} + +void crew_registry::expire_absences() +{ + auto const now = Timer::GetTime(); + + for (auto it = m_absent.begin(); it != m_absent.end();) + { + if (now < it->second) + { + ++it; + continue; + } + + WriteLog("net: peer " + std::to_string(it->first) + " did not come back, giving up its seat", logtype::net); + auto const peer = it->first; + it = m_absent.erase(it); + drop_peer(peer); + } +} + void crew_registry::drop_peer(PeerId Peer) { for (auto &pair : m_vehicles) diff --git a/network/session.h b/network/session.h index fa6472df..51058c14 100644 --- a/network/session.h +++ b/network/session.h @@ -55,8 +55,17 @@ public: claim_result claim(PeerId Peer, NetworkEntityId Id); // returns true when the peer really was part of the crew bool leave(PeerId Peer, NetworkEntityId Id); - // removes a peer from every crew it belonged to, e.g. after a disconnect + // removes a peer from every crew it belonged to void drop_peer(PeerId Peer); + // the peer lost its connection. its seat is held for a while, so that a short dropout + // does not empty a cab and hand a vehicle back to the AI under the other driver's nose + void suspend_peer(PeerId Peer); + // the peer made it back in time and keeps everything it had + void resume_peer(PeerId Peer); + // actually removes the peers whose grace period ran out + void expire_absences(); + // true while the peer is gone but still holding its seat + bool is_absent(PeerId Peer) const; std::vector crew_of(NetworkEntityId Id) const; bool is_member(PeerId Peer, NetworkEntityId Id) const; @@ -84,10 +93,21 @@ private: std::unordered_map m_vehicles; std::vector m_pending; + // peers that dropped out, with the moment their seat stops being held + std::unordered_map m_absent; }; extern crew_registry Crews; +// how long a disconnected peer keeps its place in the crew +constexpr double RECONNECT_GRACE_SECONDS = 60.0; + +// resolves the identity of a connecting peer. an unknown or empty token means somebody +// new and gets a fresh peer id together with a token of their own; a token this server +// handed out before brings back the peer id that went with it, and with it the seat in +// whatever crew the peer was on +PeerId resolve_peer_identity(uint64_t &Token); + // what a peer is allowed to ask the simulation for enum class command_verdict { diff --git a/utilities/Globals.h b/utilities/Globals.h index 22e87639..6e8927fe 100644 --- a/utilities/Globals.h +++ b/utilities/Globals.h @@ -368,6 +368,8 @@ struct global_settings { uint64_t simulation_tick = 0; // set once a joining client has taken over the state of the world from the server bool network_snapshot_applied = false; + // identity handed out by the server, presented again when reconnecting + uint64_t network_session_token = 0; std::unordered_map trainset_overrides;