diff --git a/application/application.cpp b/application/application.cpp index 90ae44d0..13d8afbf 100644 --- a/application/application.cpp +++ b/application/application.cpp @@ -498,8 +498,6 @@ int eu07_application::run() // // trivia: being client and server is possible - double frameStartTime = Timer::GetTime(); - if (m_modes[m_modestack.top()]->is_command_processor()) { // active mode is doing real calculations (e.g. drivermode) @@ -527,22 +525,29 @@ int eu07_application::run() // if we're slave if (m_network && m_network->client) { - // fetch frame info from network layer, - authoritative = m_network->client->get_next_delta(MAX_NETWORK_PER_FRAME - loop_remaining); - - // use delta and commands received from master - Timer::set_delta_override(authoritative.dt); - add_to_dequemap(commands_to_exec, authoritative.commands); + // take everything the authority has sent since the last frame. the + // commands are carried out at once; the world itself runs on our own + // clock rather than replaying the server's frame times, which is what + // made it run at the wrong speed whenever the two machines drew at + // different rates, and put every buffered frame between the player and + // their own controls + authoritative = m_network->client->take_pending(commands_to_exec); // the authority owns the timeline, so we take its step number as ours if (authoritative.valid) Global.simulation_tick = authoritative.tick; + // what we do to the train we run ourselves happens now, not after a + // round trip. it still goes to the server, so that everybody else sees + // it; the echo of it is dropped when it comes back + command_queue::commands_map predicted; + network::collect_predictable(local_commands, predicted); + add_to_dequemap(commands_to_exec, predicted); + // and send our local commands to master m_network->client->send_commands(local_commands); - if (!authoritative.valid) - loop_remaining = -1; + loop_remaining = -1; } // if we're master else @@ -603,25 +608,10 @@ int eu07_application::run() } } - // set total delta for rendering code - double totalDelta = Timer::GetTime() - frameStartTime; - Timer::set_delta_override(totalDelta); } } - if (!loop_remaining) - { - // loop break forced by counter - float received = m_network->client->get_frame_counter(); - float awaiting = m_network->client->get_awaiting_frames(); - - // TODO: don't meddle with mode progresbar - m_modes[m_modestack.top()]->set_progress(100.0f * (received - awaiting) / received); - } - else - { - m_modes[m_modestack.top()]->set_progress(0.0f, 0.0f); - } + m_modes[m_modestack.top()]->set_progress(0.0f, 0.0f); } else { diff --git a/application/driveruilayer.cpp b/application/driveruilayer.cpp index 70de30ee..67cfbfa0 100644 --- a/application/driveruilayer.cpp +++ b/application/driveruilayer.cpp @@ -282,7 +282,7 @@ void driver_ui::render_() ImGui::OpenPopup(popupheader); } - if (Global.network_position_error > 1.0f) + if (Global.network_position_error > 0.5f) { ImGui::SetNextWindowSize(ImVec2(-1, -1)); if (ImGui::Begin("network", nullptr, ImGuiWindowFlags_NoCollapse)) diff --git a/network/manager.cpp b/network/manager.cpp index a988bca0..4624463d 100644 --- a/network/manager.cpp +++ b/network/manager.cpp @@ -32,9 +32,17 @@ command_queue::commands_map network::server_manager::pop_commands() void network::server_manager::push_delta(double render_dt, double dt, uint64_t tick, uint64_t state_hash, const command_queue::commands_map &commands) { - if (dt == 0.0 && commands.empty()) + if (dt == 0.0) return; + // a client no longer replays these frame by frame, so an empty one is only worth + // sending now and then, to carry the tick and the digest + if (commands.empty() && (--heartbeat_countdown > 0)) + return; + + if (commands.empty()) + heartbeat_countdown = HEARTBEAT_INTERVAL_FRAMES; + frame_info msg; msg.render_dt = render_dt; msg.dt = dt; @@ -203,8 +211,12 @@ void network::manager::update() servers->publish_state(); } - if (client) + if (client) { client->update(); + + if (Global.simulation_loaded) + client->publish_state(); + } } void network::manager::create_server(const std::string &backend, const std::string &conf) diff --git a/network/manager.h b/network/manager.h index 4ce057ed..d389c755 100644 --- a/network/manager.h +++ b/network/manager.h @@ -42,6 +42,10 @@ namespace network int crew_publish_countdown = 0; int state_countdown = 0; int state_updates = 0; + int heartbeat_countdown = 0; + + // how many frames with nothing in them may pass before one is sent anyway + static const int HEARTBEAT_INTERVAL_FRAMES = 10; }; class manager diff --git a/network/network.cpp b/network/network.cpp index 1f2ec880..5398c978 100644 --- a/network/network.cpp +++ b/network/network.cpp @@ -218,6 +218,13 @@ 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::SNAPSHOT) { + const auto& cmd = dynamic_cast(msg); + + // a peer telling us what its own trains are doing. the filter is what keeps it to + // its own: it cannot shove anybody else's stock around + apply_snapshot(cmd.blob, snapshot_mode::correction, conn->peer_id); + } else if (msg.type == message::REQUEST_RESYNC) { const auto& cmd = dynamic_cast(msg); @@ -321,57 +328,39 @@ void network::client::update() } // client -network::frame_delta network::client::get_next_delta(int counter) +network::frame_delta network::client::take_pending(command_queue::commands_map &commands) { - auto now = std::chrono::high_resolution_clock::now(); - if (counter == 1) { - frame_time = now - last_frame; - last_frame = now; - } + frame_delta delta; - if (delta_queue.empty()) { - // buffer underflow - return frame_delta(); - } + while (!delta_queue.empty()) { + auto const &entry = delta_queue.front(); + for (auto const &kv : entry.commands) { + auto lookup = commands.end(); - float size = delta_queue.size() - consume_counter; - const auto& entry = delta_queue.front(); - float mult = entry.render_dt / std::chrono::duration_cast>(frame_time).count(); + for (auto const &data : kv.second) { + // our own commands for a train we run ourselves were carried out the + // moment we issued them; applying the echo would do it a second time + if ((data.source == Global.network_peer_id) && is_predictable(kv.first)) + continue; - if (counter == 1 && size < MAX_BUFFER_SIZE * 2.0f) { - last_target = last_target * TARGET_MIX + - (std::min(TARGET_MIN + jitteriness * JITTERINESS_MULTIPIER, MAX_BUFFER_SIZE)) * (1.0f - TARGET_MIX); - float diff = size - last_target; - jitteriness = std::max(jitteriness * JITTERINESS_MIX, std::abs(diff)); + if (lookup == commands.end()) + lookup = commands.emplace(kv.first, command_queue::commanddata_sequence()).first; - float speed = 1.0f + diff * CONSUME_MULTIPIER; - - consume_counter += speed; - } - - float last_rcv_diff = std::chrono::duration_cast>(now - last_rcv).count(); - - if (size > MAX_BUFFER_SIZE || consume_counter > mult || last_rcv_diff > 1.0f) { - if (consume_counter > mult) { - consume_counter = std::clamp(consume_counter - mult, -MAX_BUFFER_SIZE, MAX_BUFFER_SIZE); + lookup->second.emplace_back(data); + } } - frame_delta delta; delta.dt = entry.dt; delta.tick = entry.tick; delta.state_hash = entry.state_hash; delta.state_hash_version = entry.state_hash_version; - delta.commands = entry.commands; delta.valid = true; delta_queue.pop(); - - return delta; - } else { - // nothing to push - return frame_delta(); } + + return delta; } void network::client::send_commands(command_queue::commands_map commands) @@ -399,6 +388,30 @@ void network::client::send_ready() WriteLog("net: scenario loaded, asking the server for a snapshot", logtype::net); } +void network::client::publish_state() +{ + if (!conn || conn->state != connection::ACTIVE) + return; + + if (--state_countdown > 0) + return; + + state_countdown = STATE_INTERVAL_FRAMES; + + bool const full = ((state_updates++ % STATE_FULL_EVERY) == 0); + + auto const blob = take_snapshot(full, true); + if (blob.empty()) + return; + + snapshot msg; + msg.tick = Global.simulation_tick; + msg.mode = 1; // a correction, on top of what the receiver is already running + msg.blob = blob; + + conn->send_message(msg); +} + void network::client::send_resync_request(uint64_t tick, uint64_t state_hash) { if (!conn || conn->state != connection::ACTIVE) diff --git a/network/network.h b/network/network.h index 436cd74b..62a12b50 100644 --- a/network/network.h +++ b/network/network.h @@ -101,41 +101,40 @@ namespace network size_t reconnect_delay = 0; const size_t RECONNECT_DELAY_FRAMES = 60; - // the queue of authoritative frames waiting to be played out. every frame sitting - // in here is a frame of delay between pressing a key and seeing it happen, so it - // is kept as short as the connection allows rather than a comfortable second deep - const float MAX_BUFFER_SIZE = 12.0f; - const float JITTERINESS_MIX = 0.998f; - const float TARGET_MIN = 1.0f; - const float TARGET_MIX = 0.98f; - const float JITTERINESS_MULTIPIER = 2.0f; - const float CONSUME_MULTIPIER = 0.05f; + // authoritative frames that have arrived and not been handed to the simulation yet. + // they are taken in full every render: nothing is held back on purpose. the old + // design played them out one per frame at the server's pace, which meant the world + // ran at the wrong speed whenever the two machines drew at different rates, and + // every frame kept in reserve was a frame of delay on the player's own controls std::queue delta_queue; // how far out of place a correction found us, in a row. a full resync is only // worth asking for when the routine stream is not catching up on its own int bad_corrections = 0; uint64_t last_resync_tick = 0; + int state_countdown = 0; + int state_updates = 0; + static constexpr int STATE_INTERVAL_FRAMES = 6; + static constexpr int STATE_FULL_EVERY = 100; static constexpr double RESYNC_POSITION_ERROR = 25.0; static constexpr int RESYNC_BAD_CORRECTIONS = 5; // a correction needs time to take hold; asking again before it has is pointless static constexpr uint64_t RESYNC_COOLDOWN_TICKS = 600; - float last_target = 2.0f; - float jitteriness = 1.0f; - float consume_counter = 0.0f; - std::chrono::high_resolution_clock::time_point last_rcv; - std::chrono::high_resolution_clock::time_point last_frame; - std::chrono::high_resolution_clock::duration frame_time; public: void update(); - frame_delta get_next_delta(int counter); + // hands over everything the authority has sent since the last call, merging the + // commands into the map. returns what the newest of those frames said about itself + frame_delta take_pending(command_queue::commands_map &commands); void send_commands(command_queue::commands_map commands); // tells the server the scenario is loaded and a snapshot can be applied void send_ready(); + // publishes what our own trains are doing, so that the rest of the session can see + // them. we run their physics, so this is the authority on them + void publish_state(); // 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 diff --git a/network/session.cpp b/network/session.cpp index 0022657b..f22eab61 100644 --- a/network/session.cpp +++ b/network/session.cpp @@ -179,6 +179,61 @@ bool may_control(PeerId Peer, NetworkEntityId Entity) return Entities.consist_of(seat) == train; } +PeerId simulation_owner(NetworkEntityId Entity) +{ + auto const train = Entities.consist_of(Entity); + if (train == ENTITY_NONE) + return PEER_HOST; + + // the crew of the lowest numbered car decides, so that every peer works it out the + // same way without having to be told + PeerId owner{PEER_NONE}; + NetworkEntityId lowest{ENTITY_NONE}; + + for (auto const id : Entities.consist_members(train)) + { + auto const crew = Crews.crew_of(id); + if (crew.empty()) + continue; + + if (lowest == ENTITY_NONE || id < lowest) + { + lowest = id; + owner = crew.front(); + } + } + + return owner != PEER_NONE ? owner : PEER_HOST; +} + +bool is_locally_simulated(NetworkEntityId Entity) +{ + auto const self = (Global.network_peer_id != PEER_NONE ? Global.network_peer_id : PEER_HOST); + return simulation_owner(Entity) == self; +} + +bool is_predictable(uint32_t Recipient) +{ + if ((command_target)(Recipient & ~0xffff) != command_target::vehicle) + return false; + + auto const entity = (NetworkEntityId)(Recipient & 0xffff); + return is_locally_simulated(entity) && may_control(Global.network_peer_id, entity); +} + +void collect_predictable(command_queue::commands_map const &Commands, command_queue::commands_map &Predicted) +{ + for (auto const &kv : Commands) + { + if (!is_predictable(kv.first)) + continue; + + auto lookup = Predicted.emplace(kv.first, command_queue::commanddata_sequence()); + for (auto const &data : kv.second) + lookup.first->second.emplace_back(data); + } +} + vehicle_crew &crew_registry::entry(NetworkEntityId Id) { auto lookup = m_vehicles.find(Id); diff --git a/network/session.h b/network/session.h index 73878318..42e5b17a 100644 --- a/network/session.h +++ b/network/session.h @@ -148,6 +148,19 @@ command_verdict validate_command(PeerId Peer, user_command Command, uint32_t Rec // set they are sitting in, which is what makes the cab switches of a multiple unit work bool may_control(PeerId Peer, NetworkEntityId Entity); +// whose machine actually runs the physics of a given train. the first person aboard does, +// so their own train moves smoothly for them and only its result travels to the others; +// a train with nobody in it is run by the host +PeerId simulation_owner(NetworkEntityId Entity); +// true when this peer is the one running it, and therefore must not have it corrected +bool is_locally_simulated(NetworkEntityId Entity); +// true for a command this peer may carry out at once instead of waiting for the round +// trip: it addresses a train this peer both runs and is allowed to work +bool is_predictable(uint32_t Recipient); +// copies out of Commands everything that qualifies, leaving the original alone - it still +// has to reach the server so that everybody else sees it +void collect_predictable(command_queue::commands_map const &Commands, command_queue::commands_map &Predicted); + // walking into a cab with the in-game key is the same request as pressing the button in // the lobby, so it goes through the crew registry rather than being treated as a command // that reaches the whole world. returns the vehicle it resolved to, if any diff --git a/network/snapshot.cpp b/network/snapshot.cpp index c2c3482c..ca4e0ecc 100644 --- a/network/snapshot.cpp +++ b/network/snapshot.cpp @@ -33,7 +33,7 @@ uint32_t const SNAPSHOT_MAGIC{0x4e535545}; // 'EUSN' // a joining peer is placed exactly; a running one is left to its own physics for small // errors, because nudging a vehicle every fraction of a second is worse than the error double const REPOSITION_TOLERANCE_JOIN{0.5}; -double const REPOSITION_TOLERANCE_CORRECTION{2.5}; +double const REPOSITION_TOLERANCE_CORRECTION{1.5}; // --------------------------------------------------------------------------- // what the authority last sent, so that a routine correction only carries changes @@ -355,41 +355,20 @@ vehicle_state read_from(TDynamicObject const &Vehicle) return state; } -// true for the train the local player is working. its controls are driven by the commands -// its crew issue, which every peer already receives, so the correction must keep its hands -// off them - otherwise it keeps snapping a lever back a fraction of a second after the -// player moved it, and the control looks broken -bool is_own_train(std::string const &Name) -{ - auto const seat = network::Crews.vehicle_of(Global.network_peer_id); - if (seat == network::ENTITY_NONE) - return false; - - auto const train = network::Entities.consist_of(network::Entities.id_of(Name)); - return (train != network::ENTITY_NONE) && (network::Entities.consist_of(seat) == train); -} - // applies everything except where the vehicle is; position is dealt with per consist, -// because moving one vehicle of a coupled set on its own tears the couplers apart +// because moving one vehicle of a coupled set on its own tears the couplers apart. +// only ever called for vehicles somebody else is running void apply_controls(TDynamicObject &Vehicle, vehicle_state const &State) { auto &mover = *Vehicle.MoverParameters; - // what the physics produced is always taken from the authority; what a person set is - // not, on the train that person is working - bool const ours = is_own_train(State.name); - mover.V = State.velocity; mover.Vel = State.velocitykmh; - - if (!ours) - { - mover.DirActive = State.diractive; - mover.MainCtrlPos = State.mainctrl; - mover.ScndCtrlPos = State.scndctrl; - mover.BrakeCtrlPos = State.brakectrl; - mover.LocalBrakePosA = State.localbrake; - } + mover.DirActive = State.diractive; + mover.MainCtrlPos = State.mainctrl; + mover.ScndCtrlPos = State.scndctrl; + mover.BrakeCtrlPos = State.brakectrl; + mover.LocalBrakePosA = State.localbrake; mover.PipePress = State.pipepress; mover.BrakePress = State.brakepress; @@ -398,14 +377,6 @@ void apply_controls(TDynamicObject &Vehicle, vehicle_state const &State) mover.Compressor = State.compressor; mover.CompressedVolume = State.compressedvolume; - if (ours) - { - // the rest of it is what the crew set, and their commands reach us on their own - if (Vehicle.Mechanik != nullptr && Vehicle.Mechanik->AIControllFlag != State.aiactive) - Vehicle.Mechanik->TakeControl(State.aiactive); - return; - } - // the appliances go through the vehicle's own switches wherever it has them, so that // whatever they drag along stays consistent. range local: every vehicle carries its // own state in the update, there is no need to push it down the consist twice @@ -470,7 +441,7 @@ void reposition(TDynamicObject &Vehicle, vehicle_state const &State) Vehicle.place_on_track(track, nose, false); } -std::string write_vehicles(bool const Full) +std::string write_vehicles(bool const Full, bool const OwnedOnly) { std::ostringstream entries; uint32_t count{0}; @@ -480,6 +451,9 @@ std::string write_vehicles(bool const Full) if (vehicle == nullptr || vehicle->MoverParameters == nullptr) continue; + if (OwnedOnly && !network::is_locally_simulated(network::Entities.id_of(vehicle->name()))) + continue; + auto const state = read_from(*vehicle); std::ostringstream packed; @@ -516,7 +490,7 @@ std::string write_vehicles(bool const Full) return body.str(); } -void read_vehicles(std::istream &Stream, network::snapshot_mode const Mode, network::snapshot_result &Result) +void read_vehicles(std::istream &Stream, network::snapshot_mode const Mode, network::PeerId const Owner, network::snapshot_result &Result) { auto const count = sn_utils::ld_uint32(Stream); double const tolerance = (Mode == network::snapshot_mode::join ? REPOSITION_TOLERANCE_JOIN : REPOSITION_TOLERANCE_CORRECTION); @@ -532,6 +506,18 @@ void read_vehicles(std::istream &Stream, network::snapshot_mode const Mode, netw if (vehicle == nullptr || vehicle->MoverParameters == nullptr) continue; + auto const entity = network::Entities.id_of(state.name); + + // a train this peer runs is not corrected by anybody: its own physics is what the + // rest of the session is being told about, and overwriting it would be the thing + // that makes the owner's ride stutter + if ((Mode == network::snapshot_mode::correction) && network::is_locally_simulated(entity)) + continue; + + // and a peer may only move what it is entitled to move + if ((Owner != network::PEER_NONE) && (network::simulation_owner(entity) != Owner)) + continue; + apply_controls(*vehicle, state); states.emplace(vehicle, state); @@ -793,10 +779,28 @@ void network::reset_snapshot_history() g_lastswitches.clear(); } -std::string network::take_snapshot(bool const Full) +std::string network::take_snapshot(bool const Full, bool const OwnedOnly) { + auto const vehicles = write_vehicles(Full, OwnedOnly); + + if (OwnedOnly) + { + // what a peer says about its own trains, and nothing else: the clock, the crews, + // the memory cells and the switches all belong to the authority + if (vehicles.empty()) + return std::string(); + + std::ostringstream owned; + sn_utils::ls_uint32(owned, SNAPSHOT_MAGIC); + sn_utils::ls_uint32(owned, SNAPSHOT_VERSION); + sn_utils::ls_uint64(owned, Global.simulation_tick); + sn_utils::ls_uint32(owned, 1); + write_chunk(owned, SNAPSHOT_VEHICLES, vehicles); + + return owned.str(); + } + auto const session = write_session(Full); - auto const vehicles = write_vehicles(Full); auto const crews = write_crews(); auto const memcells = write_memcells(Full); auto const switches = write_switches(Full); @@ -839,7 +843,7 @@ std::string network::take_snapshot(bool const Full) return blob; } -network::snapshot_result network::apply_snapshot(std::string const &Blob, snapshot_mode const Mode) +network::snapshot_result network::apply_snapshot(std::string const &Blob, snapshot_mode const Mode, PeerId const Owner) { snapshot_result result; @@ -880,19 +884,23 @@ network::snapshot_result network::apply_snapshot(std::string const &Blob, snapsh switch (id) { case SNAPSHOT_SESSION: - read_session(chunk, Mode); + if (Owner == PEER_NONE) + read_session(chunk, Mode); break; case SNAPSHOT_VEHICLES: - read_vehicles(chunk, Mode, result); + read_vehicles(chunk, Mode, Owner, result); break; case SNAPSHOT_CREWS: - read_crews(chunk); + if (Owner == PEER_NONE) + read_crews(chunk); break; case SNAPSHOT_MEMCELLS: - read_memcells(chunk); + if (Owner == PEER_NONE) + read_memcells(chunk); break; case SNAPSHOT_SWITCHES: - read_switches(chunk); + if (Owner == PEER_NONE) + read_switches(chunk); break; default: // a section this build knows nothing about; its length is right there, so it diff --git a/network/snapshot.h b/network/snapshot.h index 1d7b0f3e..b68ee163 100644 --- a/network/snapshot.h +++ b/network/snapshot.h @@ -12,6 +12,8 @@ http://mozilla.org/MPL/2.0/. #include #include +#include "network/entities.h" + namespace network { @@ -58,10 +60,14 @@ struct snapshot_result // a full snapshot carries everything and is what a joining peer gets; otherwise only what // has changed since the last one goes out, which is what keeps the routine correction // stream small -std::string take_snapshot(bool Full); +// OwnedOnly limits it to the trains this peer runs itself and leaves out everything that +// is not a vehicle; that is what a client sends upstream about its own train +std::string take_snapshot(bool Full, bool OwnedOnly = false); -// restores state handed to us by the authority -snapshot_result apply_snapshot(std::string const &Blob, snapshot_mode Mode); +// restores state handed to us. vehicles this peer runs itself are never touched - their +// physics is the authority, not the other way round. Owner, when given, limits the update +// to the trains that peer is entitled to move, so a client cannot shove anybody else's +snapshot_result apply_snapshot(std::string const &Blob, snapshot_mode Mode, PeerId Owner = PEER_NONE); // forgets what was last sent, so that the next delta comes out complete void reset_snapshot_history();