diff --git a/Extension/Multiplayer/Session/room.h b/Extension/Multiplayer/Session/room.h index 2b8655d1..f1688390 100644 --- a/Extension/Multiplayer/Session/room.h +++ b/Extension/Multiplayer/Session/room.h @@ -160,11 +160,17 @@ struct JoinBackoff { }); } }; +// How much one peer may send in a second. `when` is when the packet arrived (the transport's +// TransportMessage::arrived), not when it is read: a receiver that could not read for a few +// seconds (a server whose console held it up, a game loading a map) reads everything that +// arrived meanwhile at once, and counted by reading time every peer looked like a flood and +// was dropped. Packets are not always read in arrival order (Steam's lanes are read one after +// another): one from before the second being counted is counted in it. struct ReceiveBudget { std::uint64_t since{}, bytes{}, packets{}; - bool accept(std::uint64_t now, std::size_t size, unsigned sources = 1) { - if (now - since >= 1000000) { - since = now; + bool accept(std::uint64_t when, std::size_t size, unsigned sources = 1) { + if (when > since && when - since >= 1000000) { + since = when; bytes = packets = 0; } bytes += size; diff --git a/Extension/Multiplayer/Session/session_receive.cpp b/Extension/Multiplayer/Session/session_receive.cpp index dbc59e16..379363eb 100644 --- a/Extension/Multiplayer/Session/session_receive.cpp +++ b/Extension/Multiplayer/Session/session_receive.cpp @@ -334,7 +334,7 @@ void networking(Session &s, const NativeFrame &local, std::uint64_t now) { if (!link) continue; const bool direct_link = s.mode == Mode::join && message.peer != s.host_id; - if (!link->budget.accept(now, message.bytes.size(), + if (!link->budget.accept(message.arrived ? message.arrived : now, message.bytes.size(), s.mode == Mode::join && !direct_link ? max_remote_players : 1U)) { disconnect(s, message.peer, "Peer exceeded the multiplayer packet limit."); if (s.mode == Mode::off) diff --git a/Extension/Multiplayer/Steam/steam_transport.cpp b/Extension/Multiplayer/Steam/steam_transport.cpp index 4b2d4b8a..16ccf320 100644 --- a/Extension/Multiplayer/Steam/steam_transport.cpp +++ b/Extension/Multiplayer/Steam/steam_transport.cpp @@ -696,7 +696,8 @@ std::vector SteamTransport::receive() { continue; } const auto *data = static_cast(message->m_pData); - result.push_back({id, {data, data + message->m_cbSize}}); + result.push_back({id, {data, data + message->m_cbSize}, + message->m_usecTimeReceived > 0 ? static_cast(message->m_usecTimeReceived) : 0}); ++p.state.received; p.state.received_bytes += static_cast(message->m_cbSize); } diff --git a/Extension/Multiplayer/Steam/steam_transport.h b/Extension/Multiplayer/Steam/steam_transport.h index 8ab05b04..d035b136 100644 --- a/Extension/Multiplayer/Steam/steam_transport.h +++ b/Extension/Multiplayer/Steam/steam_transport.h @@ -16,6 +16,10 @@ struct TransportPeer { struct TransportMessage { std::uint64_t peer{}; std::vector bytes; + // When Steam received it, in microseconds on a clock of its own (only differences mean + // anything); 0 when the transport does not say. A receiver that was kept from reading for a + // while gets everything that arrived meanwhile in one go: this is what tells that from a flood. + std::uint64_t arrived{}; }; // One message of SteamTransport::send_batch, with send()'s arguments; `sent` receives what // send() would have returned for it. diff --git a/Extension/Multiplayer/Test/mesh_tests.cpp b/Extension/Multiplayer/Test/mesh_tests.cpp index 42c4bf3b..77d05126 100644 --- a/Extension/Multiplayer/Test/mesh_tests.cpp +++ b/Extension/Multiplayer/Test/mesh_tests.cpp @@ -89,6 +89,8 @@ struct SimulatedNetwork { static inline std::uint64_t hold_control_to{}; static inline std::array lane_sends{}; static inline std::map queues; + // What a message's arrival time is (TransportMessage::arrived); 0: the transport does not say. + static inline std::uint64_t clock{}; static std::pair pair(std::uint64_t a, std::uint64_t b) { return {std::min(a, b), std::max(a, b)}; } @@ -200,7 +202,7 @@ bool SteamTransport::send(std::uint64_t id, std::span bytes, return true; if (!reliable && SimulatedNetwork::lose_unreliable && ++p.unreliable % 5 == 0) return true; - Impl::bus.at(id)->inbox.push_back({{p.state.local_id, {bytes.begin(), bytes.end()}}, lane}); + Impl::bus.at(id)->inbox.push_back({{p.state.local_id, {bytes.begin(), bytes.end()}, SimulatedNetwork::clock}, lane}); return true; } void SteamTransport::send_batch(std::span messages) { @@ -469,6 +471,8 @@ struct Simulation { NativeFrame local; std::vector positions; unsigned capacity, tps; + std::size_t held = static_cast(-1); // a node that is not run: held up, reading nothing + bool stamp{}; // messages carry when they arrived explicit Simulation(unsigned limit = max_players, unsigned rate = 20) : capacity(limit), tps(rate) { local.ready = true; local.pose.skater.resize(395); @@ -503,9 +507,10 @@ struct Simulation { void run(unsigned frames) { for (unsigned i = 0; i < frames; ++i) { now += network_tick_us; + SimulatedNetwork::clock = stamp ? now : 0; for (std::size_t n = 0; n < nodes.size(); ++n) { auto &s = *nodes[n]; - if (s.mode == Mode::off) + if (s.mode == Mode::off || n == held) continue; local.pose.root.position[0] = static_cast(now - 10000000) / 1000000; local.pose.root.position[2] = n < positions.size() ? positions[n] : static_cast(n); @@ -639,6 +644,56 @@ void party_checks() { check(!host.local_party && host.parties.parties().empty(), "A guest who left the lobby stayed in the host's party"); std::cout << "Lobby parties: nobody by default, invites, two parties, party chat, leaving and departures passed.\n"; } +// A host that is kept from reading for a few seconds (a server whose console held it up) then +// reads everything its guests sent meanwhile in one go. That is not a flood: nobody is dropped +// for it, while a real flood still is. +void stall_checks() { + { + // 16 s of a player's ordinary traffic read at once, counted by when it arrived. + ReceiveBudget backlog; + bool kept = true; + for (std::uint64_t i = 0; i < 16 * 90; ++i) kept = kept && backlog.accept(5000000 + i * (1000000 / 90), 300); + check(kept, "A backlog read after a stall was taken for a flood"); + ReceiveBudget flood; + bool refused = false; + for (std::uint64_t i = 0; i < 2000; ++i) refused = refused || !flood.accept(5000000 + i * 100, 300); + check(refused, "A flood was let through the packet limit"); + // Lanes are read one after another, so arrival times step back and forth. + ReceiveBudget lanes; + refused = false; + for (std::uint64_t i = 0; i < 2000; ++i) refused = refused || !lanes.accept(i % 2 ? 9000000 : 9500000, 300); + check(refused, "Packets read out of arrival order started the count again"); + } + Simulation sim(max_players, 120); + sim.stamp = true; + for (unsigned i = 0; i < 4; ++i) sim.add(); + sim.run(240); + auto &host = *sim.nodes[0]; + const auto connected = [&] { + unsigned count{}; + for (std::size_t n = 1; n < sim.nodes.size(); ++n) { + const auto *peer = find_peer(host, sim.nodes[n]->transport.status().local_id); + count += peer && peer->handshaken && sim.nodes[n]->mode == Mode::join; + } + return count; + }; + check(connected() == 3, "The stall fixture did not connect its guests"); + // Nine seconds unread: under the ten after which a silent player is given up on, and more + // than a second's allowance from every guest. + const auto sent = sim.nodes[1]->transport.status().sent; + sim.held = 0; + sim.run(static_cast(9000000 / network_tick_us)); + sim.held = static_cast(-1); + check(sim.nodes[1]->transport.status().sent - sent > 2ULL * multiplayer_tick_rates.back() + 80 + 32, + "The stall fixture did not queue more than the packet limit"); + sim.run(240); + if (host.mode != Mode::host || connected() != 3) + for (std::size_t n = 0; n < sim.nodes.size(); ++n) + std::cerr << " node " << n << " mode " << static_cast(sim.nodes[n]->mode) << ": " << sim.nodes[n]->status << "\n"; + check(host.mode == Mode::host && connected() == 3, "Guests were dropped after their host was held up for a few seconds"); + SimulatedNetwork::clock = 0; + std::cout << "Stalls: a held-up host keeps its guests; floods and out-of-order arrival are still limited.\n"; +} // What a host changes outside its physics tuning (the trainer's class values and trick // multipliers) reaches its guests while it sets everyone's physics: when it changes, to a // player who joins later, and from nobody but the host. @@ -1934,6 +1989,7 @@ int main(int argc, char **argv) { dingosdk::multiplayer::throwdown_routing_checks(); dingosdk::multiplayer::party_checks(); dingosdk::multiplayer::physics_extras_checks(); + dingosdk::multiplayer::stall_checks(); dingosdk::multiplayer::scoring_checks(); dingosdk::multiplayer::object_sync_checks(); dingosdk::multiplayer::session_controls_checks(); diff --git a/Server/main.cpp b/Server/main.cpp index 3b38e239..6cd5e27c 100644 --- a/Server/main.cpp +++ b/Server/main.cpp @@ -23,6 +23,7 @@ #endif #include #include +#include #include #include #include @@ -44,6 +45,57 @@ constexpr int restart_for_update = -2; std::string update_version; std::mutex log_mutex; std::ofstream log_file; +// The console is written by a thread of its own. Windows holds a console's output while text +// in its window is selected, and a pipe nobody reads fills up: whoever writes then waits. When +// that was the server loop, the server read nothing from its players until the console let go, +// and they were gone by then. The log file is still written where the line is made. +struct Console { + std::mutex mutex; + std::condition_variable more, drained; + std::deque lines; + std::size_t skipped{}; + bool writing{}, started{}; + void write(std::string line) { + std::lock_guard lock(mutex); + if (!started) { + started = true; + std::thread([this] { run(); }).detach(); + } + // A console that stays held must not grow this without end: the oldest lines go, and + // the file has them all. + if (lines.size() >= 4096) { + lines.pop_front(); + ++skipped; + } + lines.push_back(std::move(line)); + more.notify_one(); + } + void run() { + for (;;) { + std::unique_lock lock(mutex); + more.wait(lock, [&] { return !lines.empty(); }); + const auto line = std::move(lines.front()); + lines.pop_front(); + const auto missed = std::exchange(skipped, 0); + writing = true; + lock.unlock(); + if (missed) std::printf("(%zu earlier lines are only in ReSkateServer.log: the console was not taking output)\n", missed); + std::fputs(line.c_str(), stdout); + lock.lock(); + writing = false; + if (lines.empty()) drained.notify_all(); + } + } + // Waits until everything written so far is on screen, but not for a console that is held. + void flush(std::chrono::milliseconds limit) { + std::unique_lock lock(mutex); + drained.wait_for(lock, limit, [&] { return lines.empty() && !writing; }); + } +}; +Console &console() { + static auto *value = new Console; // outlives every thread that may still be writing at exit + return *value; +} void write_log(const std::string &text) { std::lock_guard lock(log_mutex); const auto now = std::time(nullptr); @@ -55,8 +107,8 @@ void write_log(const std::string &text) { #endif char stamp[32]{}; std::strftime(stamp, sizeof stamp, "%Y-%m-%d %H:%M:%S", &local); - std::printf("[%s] %s\n", stamp + 11, text.c_str()); if (log_file) log_file << '[' << stamp << "] " << text << std::endl; + console().write(std::string("[") + (stamp + 11) + "] " + text + "\n"); } std::atomic finished{}; #ifdef _WIN32 @@ -444,6 +496,7 @@ int wmain(int argc, wchar_t **argv) { write_log("Installing server update " + update_version + "..."); install_update(folder()); write_log("Server update " + update_version + " installed; starting it."); + console().flush(std::chrono::seconds(2)); if (relaunch()) { code = 0; break; @@ -462,9 +515,10 @@ int wmain(int argc, wchar_t **argv) { if (code != 0 && !stopping && own_window()) { auto &input = console_input(); input.take(); - std::printf("Press Enter to close.\n"); + console().write("Press Enter to close.\n"); while (!stopping && input.take().empty()) Sleep(50); } + console().flush(std::chrono::seconds(2)); return code; } #else @@ -478,6 +532,7 @@ int main(int argc, char **argv) { write_log("Installing server update " + update_version + "..."); install_update(folder()); write_log("Server update " + update_version + " installed; starting it."); + console().flush(std::chrono::seconds(2)); if (relaunch()) { code = 0; break; @@ -491,6 +546,7 @@ int main(int argc, char **argv) { } } finished = true; + console().flush(std::chrono::seconds(2)); return code; } #endif diff --git a/Server/server_host.cpp b/Server/server_host.cpp index 1a452eb5..5833df38 100644 --- a/Server/server_host.cpp +++ b/Server/server_host.cpp @@ -414,10 +414,12 @@ bool Host::accept_data(Guest &source, const Packet &p) { } return true; } -void Host::receive(std::uint64_t peer, std::span bytes) { +void Host::receive(std::uint64_t peer, std::span bytes, std::uint64_t received_at) { auto *link = find(peer); if (!link) return; - if (!link->budget.accept(now_, bytes.size(), 1U)) return drop(peer, "Peer exceeded the multiplayer packet limit."); + // Counted by arrival: after the server was held up, everyone's packets are read at once. + if (!link->budget.accept(received_at ? received_at : now_, bytes.size(), 1U)) + return drop(peer, "Peer exceeded the multiplayer packet limit."); bool missing_reference{}; const auto decoded = link->receiver.receive(bytes, missing_reference, world_); if (missing_reference) return; @@ -764,7 +766,7 @@ void Host::tick(std::uint64_t now) { } if (link.connected && !guest->connected_at) guest->connected_at = now_; } - for (const auto &message : transport_.receive()) receive(message.peer, message.bytes); + for (const auto &message : transport_.receive()) receive(message.peer, message.bytes, message.arrived); receive_cosmetics(); for (auto it = guests_.begin(); it != guests_.end();) { auto &g = *(it++)->second; diff --git a/Server/server_host.h b/Server/server_host.h index 198ebeb2..8ca26a4c 100644 --- a/Server/server_host.h +++ b/Server/server_host.h @@ -158,7 +158,8 @@ class Host { void change_map(std::string_view map); // a level name, level path or destination bool same_map(std::string_view asset) const { return map_hash(map_destination(asset)) == map_; } bool accept_data(Guest &source, const Packet &); - void receive(std::uint64_t peer, std::span bytes); + // `received_at`: the transport's arrival time for the message (TransportMessage::arrived). + void receive(std::uint64_t peer, std::span bytes, std::uint64_t received_at); void receive_cosmetics(); void sync_objects(); void apply_layers(); diff --git a/assets/launcher/background.jpg b/assets/launcher/background.jpg index 2c80aeba..1bea5f35 100644 Binary files a/assets/launcher/background.jpg and b/assets/launcher/background.jpg differ