diff --git a/CMakeLists.txt b/CMakeLists.txt index 20c4a13..a0e7729 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -212,6 +212,20 @@ if(DINGOSDK_BUILD_MULTIPLAYER_TESTS AND WIN32) add_executable(dingosdk_ea_service_block_tests Extension/Boot/Test/ea_service_block_tests.cpp) target_include_directories(dingosdk_ea_service_block_tests PRIVATE "${PROJECT_SOURCE_DIR}") add_test(NAME ea_service_block COMMAND dingosdk_ea_service_block_tests) + # The dedicated server's host under a full server of simulated players, over a transport + # that is not Steam's: what its passes cost, and a hash of what it sent (not run by ctest). + add_executable(dingosdk_server_load_tests Server/Test/server_load_tests.cpp Server/server_host.cpp Server/server_party.cpp + Extension/Multiplayer/Session/party_book.cpp Server/server_config.cpp + Server/server_update.cpp Server/server_release.cpp Launcher/updater.cpp + Server/global_bans.cpp Extension/Multiplayer/developer_identity.cpp + Extension/Multiplayer/Net/protocol.cpp + Extension/Multiplayer/Net/delta_codec.cpp Extension/Multiplayer/Net/wire_codec.cpp + Extension/Multiplayer/Remote/playback_buffers.cpp Extension/Multiplayer/Session/password.cpp + Server/server_activity.cpp Server/server_votes.cpp Server/server_commands.cpp Extension/Throwdowns/throwdown_wire.cpp) + target_include_directories(dingosdk_server_load_tests PRIVATE "${PROJECT_SOURCE_DIR}") + target_include_directories(dingosdk_server_load_tests SYSTEM PRIVATE "${PROJECT_SOURCE_DIR}/External/steam_networking") + target_link_libraries(dingosdk_server_load_tests PRIVATE dingosdk_launcher_support dingosdk_world_layer_scan dingosdk_json + dingosdk_lz4 dingosdk_zstd dingosdk_logging dingosdk_miniz dingosdk_word_filter dingosdk_https winhttp bcrypt winmm) endif() # The route and logging tests live in the local, untracked tests/ folder. option(DINGOSDK_BUILD_ROUTE_TESTS "Build native route cycle regression tests" OFF) diff --git a/Server/README.txt b/Server/README.txt index b4c1ddb..f7573b6 100644 --- a/Server/README.txt +++ b/Server/README.txt @@ -218,6 +218,14 @@ pack_ms How long a message to a player may wait to go in the same bandwidth and less CPU. It adds up to that long to when an update arrives. Voice is never held back. 0 sends every message at once, as before. Takes effect on restart. +threads How many threads share the sending of each pass (default 0: + one for each of the machine's processors but one, up to + 8; 1: a single thread, as before). Most of a full + server's work is building each player's update of everyone + else, which the threads do for several players at once. + What is sent is the same whatever the number. The server + says how many it uses when it starts. Takes effect on + restart. finger_distance Past this many metres (default 25) a player's fingers are not sent moving: they stay as they were, and move again when the player is nearer. Fingers are nearly half of every position diff --git a/Server/Test/server_config_tests.cpp b/Server/Test/server_config_tests.cpp index 794f73c..c0bdfe1 100644 --- a/Server/Test/server_config_tests.cpp +++ b/Server/Test/server_config_tests.cpp @@ -74,6 +74,7 @@ int run() { scaled.bone_scale_limit = 0.5f; check(config_error(scaled).find("bone_scale_limit") != std::string::npos, "A bone scale limit under 1 was accepted"); check(has("network.pack_ms") && config.pack_ms == 10, "pack_ms is not a new setting of 10"); + check(has("network.threads") && config.threads == 0, "threads is not a new setting of 0"); check(has("network.finger_distance") && config.finger_distance == 25, "finger_distance is not a new setting of 25"); check(has("network.use_steam_relay") && config.use_steam_relay, "use_steam_relay is not a new setting of true"); ServerConfig crowd; diff --git a/Server/Test/server_load_tests.cpp b/Server/Test/server_load_tests.cpp new file mode 100644 index 0000000..6bcb172 --- /dev/null +++ b/Server/Test/server_load_tests.cpp @@ -0,0 +1,295 @@ +// Drives the dedicated server's host with simulated players over a transport that is not +// Steam's, to see what a full server costs and that a change to how it sends leaves what it +// sends alone. Not a pass/fail test of behaviour: it prints how long the passes took and a +// hash of everything the server sent each player. +// +// dingosdk_server_load_tests [players=120] [seconds=20] [spacing in metres=6] [threads=0: the server's own choice] [hash=1] [stall for this many seconds at 12 s=0] +// +// The players stand in a grid and move a little every pose; each sends 20 poses a second the +// way a game does (pose_batch.h) and says which of the server's pose messages it read. +#include "Server/server_host.h" +#include "Extension/Multiplayer/Net/pose_batch.h" +#include "Extension/Multiplayer/Net/pose_codec.h" +#include "Extension/Multiplayer/Net/wire_codec.h" +#include "Extension/Multiplayer/Steam/steam_transport.h" +#include "Engine/Game/Build/supported_build.h" + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +namespace dingosdk::multiplayer { +namespace { +// What the fake transport holds: the simulated players and what goes each way. +struct Wire { + std::mutex mutex; // sends may come from several threads + TransportStatus status; + std::vector inbound; + struct Player { + std::uint64_t hash = 0xCBF29CE484222325ULL; // of the pose and sound messages sent them, in order + std::uint64_t messages{}, bytes{}; + std::vector batches; // pose messages to say were read + std::vector> acks; // the server's, of this player's own uploads + }; + std::map players; + std::map closed; + std::uint64_t now{}; + bool hashing = true; // off: a send costs next to nothing, to time the server alone +}; +Wire &wire() { + static Wire value; + return value; +} +} // namespace + +struct SteamTransport::Impl {}; +SteamTransport::SteamTransport() = default; +SteamTransport::~SteamTransport() = default; +bool SteamTransport::open() { return true; } +bool SteamTransport::open_game_server(void *) { return true; } +bool SteamTransport::host(unsigned) { + wire().status.ready = wire().status.hosting = true; + wire().status.local_id = 90071992547409920ULL + 1; + return true; +} +bool SteamTransport::join(std::uint64_t, std::uint32_t, std::uint16_t) { return false; } +bool SteamTransport::listen_direct(std::uint16_t) { return true; } +void SteamTransport::set_packing(unsigned) {} +std::vector SteamTransport::take_direct_notes() { return {}; } +bool SteamTransport::set_steam_debug(bool) { return false; } +bool SteamTransport::connect_peer(std::uint64_t) { return false; } +void SteamTransport::allow_peers(std::span) {} +bool SteamTransport::socket_test() { return true; } +void SteamTransport::stop() {} +void SteamTransport::disconnect(std::uint64_t id, const char *reason) { + auto &w = wire(); + std::lock_guard lock(w.mutex); + w.closed[id] = reason ? reason : ""; + std::erase_if(w.status.peers, [&](const TransportPeer &peer) { return peer.id == id; }); +} +void SteamTransport::poll() {} +bool SteamTransport::send(std::uint64_t id, std::span bytes, bool reliable, bool fresh, TrafficLane lane) { + auto &w = wire(); + std::lock_guard lock(w.mutex); + const auto found = w.players.find(id); + if (found == w.players.end()) return false; + auto &player = found->second; + ++player.messages; + player.bytes += bytes.size(); + // Pose and sound messages carry nothing of the session's random numbers: they hash the same + // from run to run. The rest (rosters, greetings) are only counted. + if (pose_batch::is_batch(bytes) && bytes.size() >= 8) { + std::uint32_t number{}; + for (unsigned i = 0; i < 4; ++i) number |= std::uint32_t{bytes[4 + i]} << (8 * i); + player.batches.push_back(number); + } + if (w.hashing && (pose_batch::is_batch(bytes) || lane == TrafficLane::gameplay)) { + const auto mix = [&](std::uint8_t byte) { player.hash = (player.hash ^ byte) * 0x100000001B3ULL; }; + for (const auto byte : bytes) mix(byte); + mix(static_cast(reliable)); + mix(static_cast(fresh)); + } + if (pose_batch::is_ack(bytes)) player.acks.emplace_back(bytes.begin(), bytes.end()); + return true; +} +void SteamTransport::send_batch(std::span messages) { + for (auto &message : messages) message.sent = send(message.id, message.bytes, message.reliable, message.fresh, message.lane); +} +std::vector SteamTransport::receive() { return std::exchange(wire().inbound, {}); } +std::string SteamTransport::name(std::uint64_t) { return {}; } +const TransportStatus &SteamTransport::status() const { return wire().status; } +std::string SteamTransport::take_closed(std::uint64_t) { return {}; } +std::string SteamTransport::link_report(std::uint64_t) { return {}; } +std::int64_t SteamTransport::pending(std::uint64_t) const { return 0; } +std::string SteamTransport::relay_status() const { return "OK"; } +bool SteamTransport::set_send_rate(int) { return true; } +int SteamTransport::send_rate() const { return 900 * 1024; } +std::vector SteamTransport::links() { return {}; } +bool SteamTransport::bind(void *, void *, void *) { return true; } +} // namespace dingosdk::multiplayer + +namespace { +using namespace dingosdk; +using namespace dingosdk::multiplayer; + +// One simulated player's game: where it stands, the poses it has sent and its upload stream. +struct Simulated { + std::uint64_t id{}, epoch{}; + std::array home{}; + std::uint32_t sequence{}; + std::uint64_t next_pose{}; + bool greeted{}; + struct Kept { + std::uint32_t sequence{}; + std::uint64_t time_us{}; + pose_codec::QuantPose pose; + }; + std::deque kept; + pose_batch::Sender upload; + pose_batch::Ack read; // the server's pose messages this player read +}; + +Pose pose_of(const Simulated &player, std::uint64_t now) { + const auto t = static_cast(now % 100000000ULL) / 1000000.f; + const auto phase = static_cast(player.id % 97); + Pose pose; + pose.root.position = {player.home[0] + std::sin(t * 0.7f + phase) * 2.f, player.home[1], player.home[2] + std::cos(t * 0.5f + phase) * 2.f}; + const auto turn = t * 0.3f + phase; + pose.root.rotation = {0.f, std::sin(turn * 0.5f), 0.f, std::cos(turn * 0.5f)}; + pose.skater.resize(pose_codec::finger_skeleton); + for (std::size_t bone = 0; bone < pose.skater.size(); ++bone) { + auto &b = pose.skater[bone]; + // A skater moves a fraction of its bones much from one pose to the next. + const auto angle = bone % 5 ? 0.1f * static_cast(bone % 13) + : std::sin(t * (1.f + static_cast(bone % 7) * 0.2f) + phase + static_cast(bone)) * 0.4f; + b.position = {static_cast(bone % 5) * 0.03f, static_cast(bone % 11) * 0.02f, 0.01f}; + b.rotation = {std::sin(angle * 0.5f), 0.f, 0.f, std::cos(angle * 0.5f)}; + } + if (pose.skater.size() > 1) pose.skater[1].position = pose.root.position; + pose.board.resize(16); + for (std::size_t bone = 0; bone < pose.board.size(); ++bone) { + const auto angle = std::sin(t * 2.f + phase + static_cast(bone)) * 0.2f; + pose.board[bone].rotation = {0.f, 0.f, std::sin(angle * 0.5f), std::cos(angle * 0.5f)}; + } + pose.board[0].position = pose.root.position; + return pose; +} +} // namespace + +int main(int argc, char **argv) { + const auto count = argc > 1 ? static_cast(std::atoi(argv[1])) : 120U; + const auto seconds = argc > 2 ? static_cast(std::atoi(argv[2])) : 20U; + const auto spacing = argc > 3 ? static_cast(std::atof(argv[3])) : 6.f; + server::ServerConfig config; + config.max_players = std::max(count, 2U); + config.activity_log = false; + const auto stall = argc > 6 ? static_cast(std::atoi(argv[6])) : 0U; + wire().hashing = argc <= 5 || std::atoi(argv[5]) != 0; + config.threads = argc > 4 ? static_cast(std::atoi(argv[4])) : 0U; + SteamTransport transport; + std::size_t lines{}; + server::Host host(config, transport, [&](const std::string &line) { + if (line.find("joined") == std::string::npos && lines++ < 40) std::printf("log: %s\n", line.c_str()); + }); + std::string error; + if (!host.start(error)) { + std::printf("The server did not start: %s\n", error.c_str()); + return 1; + } + const auto invite = parse_invite(host.invite()); + if (!invite) { + std::printf("No invite.\n"); + return 1; + } + const auto secret = invite->secret; + const auto map = map_hash(server::map_destination(config.map)); + auto &w = wire(); + std::vector players(count); + const auto side = static_cast(std::ceil(std::sqrt(static_cast(count)))); + for (unsigned index = 0; index < count; ++index) { + auto &player = players[index]; + player.id = 76561198000000001ULL + index; + player.epoch = 5000 + index; + player.home = {static_cast(index % side) * spacing, 10.f, static_cast(index / side) * spacing}; + w.players[player.id]; + w.status.peers.push_back({player.id, true}); + } + const auto header = [&](const Simulated &player, PacketKind kind, std::uint64_t now) { + Packet p; + p.kind = kind; + p.session = secret; + p.map = map; + p.world = 1; + p.epoch = player.epoch; + p.source = player.id; + p.time_us = now; + return p; + }; + using Clock = std::chrono::steady_clock; + const std::uint64_t start = 1000000000ULL, step = 5000, warm = 8000000; + double busy{}, longest{}; + std::uint64_t passes{}, bytes_at_warm{}; + const auto sent_bytes = [&] { + std::uint64_t total{}; + for (const auto &[id, player] : w.players) total += player.bytes; + return total; + }; + for (std::uint64_t now = start; now < start + std::uint64_t{seconds} * 1000000; now += step) { + w.now = now; + for (auto &player : players) { + if (w.closed.contains(player.id)) continue; + auto &seen = w.players[player.id]; + if (!player.greeted) { + auto hello = header(player, PacketKind::hello, now); + hello.build = supported_build::game_sha256_bytes; + hello.text = "Player " + std::to_string(player.id % 1000); + w.inbound.push_back({player.id, encode_wire(hello), now}); + player.greeted = true; + player.next_pose = now + 200000 + (player.id % 10) * 5000; + continue; + } + // What the server said it holds of this player's poses, and which of its messages they read. + for (const auto &ack : std::exchange(seen.acks, {})) + if (const auto read = pose_batch::Ack::read(ack)) player.upload.ack(*read); + if (!seen.batches.empty()) { + for (const auto number : std::exchange(seen.batches, {})) player.read.note(number); + const auto ack = player.read.bytes(); + w.inbound.push_back({player.id, {ack.begin(), ack.end()}, now}); + } + if (now < player.next_pose) continue; + player.next_pose += 50000; + auto p = header(player, PacketKind::pose, now); + p.sequence = ++player.sequence; + p.pose = pose_of(player, now); + player.kept.push_back({p.sequence, p.time_us, pose_codec::quantize(p.pose)}); + while (player.kept.size() > 64) player.kept.pop_front(); + const auto find = [&](std::uint32_t sequence) -> std::optional { + for (auto it = player.kept.rbegin(); it != player.kept.rend(); ++it) + if (it->sequence == sequence) return pose_batch::KeptView{it->sequence, it->time_us, &it->pose}; + return {}; + }; + const auto emit = [&](std::span message, bool) { + w.inbound.push_back({player.id, {message.begin(), message.end()}, now}); + return true; + }; + player.upload.begin(1, map); + player.upload.add(p.source, p.epoch, *find(p.sequence), find, false, false, pose_batch::Rate::full, now, emit); + player.upload.flush(emit); + } + // A stall: for a while the server does not get to run, and everything sent meanwhile waits for it. + if (stall && now - start >= 12000000 && now - start < 12000000 + std::uint64_t{stall} * 1000000) continue; + const auto began = Clock::now(); + host.tick(now); + const auto took = std::chrono::duration(Clock::now() - began).count(); + if (now - start == warm) bytes_at_warm = sent_bytes(); + if (now - start >= warm) { + busy += took; + longest = std::max(longest, took); + ++passes; + } + } + const auto measured = static_cast(seconds) - static_cast(warm) / 1000000.0; + std::uint64_t hash = 0xCBF29CE484222325ULL, messages{}; + for (const auto &[id, player] : w.players) { + hash = (hash ^ player.hash) * 0x100000001B3ULL; + messages += player.messages; + } + std::printf("%u players, %u s (the first %.0f s not timed), %zu left on\n", count, seconds, static_cast(warm) / 1000000.0, + w.status.peers.size()); + for (const auto &[id, reason] : w.closed) std::printf(" closed %llu: %s\n", static_cast(id), reason.c_str()); + if (measured > 0 && passes) + std::printf("busy %.1f%% of one core; %.2f ms a pass on average, longest %.1f ms; %.0f KB/s out\n", + busy / (measured * 10.0), busy / static_cast(passes), longest, + static_cast(sent_bytes() - bytes_at_warm) / 1024.0 / measured); + std::printf("%llu messages sent; pose and sound hash %016llx\n", static_cast(messages), + static_cast(hash)); + std::printf("%s\n", host.command("net").c_str()); + return 0; +} diff --git a/Server/server_config.cpp b/Server/server_config.cpp index 91b431a..a1990b6 100644 --- a/Server/server_config.cpp +++ b/Server/server_config.cpp @@ -122,6 +122,7 @@ Layout layout(const ServerConfig &c) { network.set("send_rate", c.send_rate); network.set("crowd_budget", c.crowd_budget); network.set("pack_ms", c.pack_ms); + network.set("threads", c.threads); network.set("finger_distance", c.finger_distance); auto &distances = network.section("distances"); distances.set("full_rate_return", c.distances.full_rate_return); @@ -279,6 +280,7 @@ ServerConfig load_config(const std::filesystem::path &file, std::vectoris_object()) { c.distances.full_rate_return = d->value("full_rate_return", c.distances.full_rate_return); @@ -403,6 +405,7 @@ std::string config_error(const ServerConfig &c) { if (c.bone_scale_limit != 0 && !(c.bone_scale_limit >= 1.f && c.bone_scale_limit <= 8.f)) return "bone_scale_limit must be 0 (no limit) or 1 to 8 (1: no resized body parts at all)."; if (c.pack_ms > 50) return "pack_ms must be 0 (off) to 50."; + if (c.threads > 32) return "threads must be 0 (one for each processor but one) to 32."; if (c.finger_distance > 10000) return "finger_distance must be 0 (fingers always sent) to 10000."; if (!valid_crowd_budget(c.crowd_budget)) return "crowd_budget must be 0 (no limit) or " + std::to_string(min_crowd_budget) + " to " + diff --git a/Server/server_config.h b/Server/server_config.h index 058b960..4824c18 100644 --- a/Server/server_config.h +++ b/Server/server_config.h @@ -53,6 +53,9 @@ struct ServerConfig { // milliseconds (0: each goes at once in a packet of its own). Fewer, fuller packets: // less sent for the same updates, and less work sending it. unsigned pack_ms = 10; + // How many threads share the sending of each pass, this one included (1: the one thread, + // as before 2.0.2). 0: one for each of the machine's processors but one, up to 8. + unsigned threads = 0; // Past this many metres a player's fingers are not sent moving (0: always). A skater's // forty finger bones turn in nearly every pose and are half of what a pose carries. unsigned finger_distance = 25; diff --git a/Server/server_host.cpp b/Server/server_host.cpp index eb9309b..b02f51c 100644 --- a/Server/server_host.cpp +++ b/Server/server_host.cpp @@ -160,6 +160,16 @@ bool Host::start(std::string &error) { password_ = config_.password.empty() ? std::nullopt : password_key(config_.password, secret_); voice_policy_ = {config_.voice_chat, 1}; apply_layers(); + // The threads that share each pass's sending with this one ("threads": this many in all; + // 0: one for each of the machine's processors but one, which is left to Steam's own). + { + const auto cores = std::max(1U, std::thread::hardware_concurrency()); + const auto wanted = config_.threads ? config_.threads : std::clamp(cores - 1, 1U, 8U); + workers_.reset(); + if (wanted > 1) workers_ = std::make_unique(wanted - 1); + log_("Threads: " + std::to_string(1 + (workers_ ? workers_->threads() : 0)) + " share the sending of each pass (\"threads\"; this machine has " + + std::to_string(cores) + " processors)."); + } running_ = true; roster_dirty_ = true; return true; @@ -285,67 +295,106 @@ Host::Guest::KeptPose *Host::keep_pose(Guest &from, const Packet &packet) { } // Sends each player the poses queued for them this pass (broadcast): their differences from a // pose of the same player the recipient is known to hold, packed several to a message. -void Host::flush_poses() { - for (auto &[id, guest] : guests_) { - auto &g = *guest; - // Which of their own pose messages arrived, once for all read this pass. - if (std::exchange(g.upload_ack_due, false)) { - const auto ack = g.upload_ack.bytes(); - if (transport_.send(id, ack, false, true, TrafficLane::gameplay)) - for (auto *counted : {&g.traffic, &traffic_}) counted->total.out[traffic_kind(PacketKind::pose)] += ack.size(); - } - // The sound of the skaters they can hear, in one message, reliably: each builds on the - // last, and many samples are a pulse that must not be lost. - if (!g.queued_sound.empty()) { - const auto sounds = std::exchange(g.queued_sound, {}); - if (g.handshaken && g.world_ready) { - g.sound_sender.begin(world_, map_); - for (const auto &q : sounds) - if (const auto *from = find(q.source)) g.sound_sender.add(q.source, from->member.epoch, q.sequence, q.time_us, *q.samples, now_); - if (g.sound_sender.pending()) { - const auto message = g.sound_sender.message(); - const bool went = transport_.send(id, message, true, true, traffic_lane(PacketKind::audio)); - if (went) - for (auto *counted : {&g.traffic, &traffic_}) counted->total.out[traffic_kind(PacketKind::audio)] += message.size(); - g.sound_sender.sent(went); - } +// What one player is sent of the pass: the ack of their own poses, the sound they hear and the +// poses of everyone they see. It reads the other players (the poses kept of them) and writes +// only this player's own state and `sent`, so the players are done on several threads at once +// (flush_poses); what is everyone's is added up afterwards from `sent`. +void Host::flush_player(std::uint64_t id, Guest &g, Flushed &sent) { + // One at a time into the transport, in this player's order. + const auto send = [&](std::span message, bool reliable, bool fresh, TrafficLane lane) { + std::lock_guard lock(send_mutex_); + return transport_.send(id, message, reliable, fresh, lane); + }; + const auto count = [&](PacketKind kind, std::size_t bytes) { + g.traffic.total.out[traffic_kind(kind)] += bytes; + sent.traffic.out[traffic_kind(kind)] += bytes; + }; + // Which of their own pose messages arrived, once for all read this pass. + if (std::exchange(g.upload_ack_due, false)) { + const auto ack = g.upload_ack.bytes(); + if (send(ack, false, true, TrafficLane::gameplay)) count(PacketKind::pose, ack.size()); + } + // The sound of the skaters they can hear, in one message, reliably: each builds on the + // last, and many samples are a pulse that must not be lost. + if (!g.queued_sound.empty()) { + const auto sounds = std::exchange(g.queued_sound, {}); + if (g.handshaken && g.world_ready) { + g.sound_sender.begin(world_, map_); + for (const auto &q : sounds) + if (const auto *from = find(q.source)) g.sound_sender.add(q.source, from->member.epoch, q.sequence, q.time_us, *q.samples, now_); + if (g.sound_sender.pending()) { + const auto message = g.sound_sender.message(); + const bool went = send(message, true, true, traffic_lane(PacketKind::audio)); + if (went) count(PacketKind::audio, message.size()); + g.sound_sender.sent(went); } } - if (g.queued_poses.empty()) continue; - const auto queued = std::exchange(g.queued_poses, {}); - if (!g.handshaken || !g.world_ready) continue; - const auto emit = [&](std::span message, bool reliable) { - if (!transport_.send(id, message, reliable, !reliable, TrafficLane::gameplay)) return false; - for (auto *counted : {&g.traffic, &traffic_}) counted->total.out[traffic_kind(PacketKind::pose)] += message.size(); - return true; + } + if (g.queued_poses.empty()) return; + const auto queued = std::exchange(g.queued_poses, {}); + if (!g.handshaken || !g.world_ready) return; + const auto emit = [&](std::span message, bool reliable) { + if (!send(message, reliable, !reliable, TrafficLane::gameplay)) return false; + count(PacketKind::pose, message.size()); + return true; + }; + g.pose_sender.begin(world_, map_); + for (const auto &q : queued) { + auto *from = find(q.source); + if (!from) continue; + const auto find_kept = [&](std::uint32_t sequence) -> std::optional { + for (auto it = from->kept_poses.rbegin(); it != from->kept_poses.rend(); ++it) + if (it->sequence == sequence) return pose_batch::KeptView{it->sequence, it->time_us, &it->pose}; + return {}; }; - g.pose_sender.begin(world_, map_); - for (const auto &q : queued) { - auto *from = find(q.source); - if (!from) continue; - const auto find_kept = [&](std::uint32_t sequence) -> std::optional { - for (auto it = from->kept_poses.rbegin(); it != from->kept_poses.rend(); ++it) - if (it->sequence == sequence) return pose_batch::KeptView{it->sequence, it->time_us, &it->pose}; - return {}; - }; - const auto pose = find_kept(q.sequence); - if (!pose) continue; - const auto added = g.pose_sender.add(q.source, from->member.epoch, *pose, find_kept, q.hold_fingers, q.collision, q.rate, now_, emit); - if (added.did == pose_batch::Sender::Did::whole) { - // Kept long enough for the ack of it to find it here. - for (auto &kept : from->kept_poses) - if (kept.sequence == q.sequence) kept.keep = 3; - for (auto *counted : {&g.traffic, &traffic_}) ++counted->total.snapshots; - ++pose_sizes_.whole_sent; - pose_sizes_.whole_sent_bytes += added.bytes; - } else if (added.did == pose_batch::Sender::Did::difference) { - const auto tier = std::min(q.tier, 3); - ++pose_sizes_.sent[tier]; - pose_sizes_.sent_bytes[tier] += added.bytes + 8; - pose_sizes_.held += q.hold_fingers; - } + const auto pose = find_kept(q.sequence); + if (!pose) continue; + const auto added = g.pose_sender.add(q.source, from->member.epoch, *pose, find_kept, q.hold_fingers, q.collision, q.rate, now_, emit); + if (added.did == pose_batch::Sender::Did::whole) { + // Kept long enough for the ack of it to find it here (marked once every player is done). + sent.whole.emplace_back(from, q.sequence); + ++g.traffic.total.snapshots; + ++sent.traffic.snapshots; + ++sent.sizes.whole_sent; + sent.sizes.whole_sent_bytes += added.bytes; + } else if (added.did == pose_batch::Sender::Did::difference) { + const auto tier = std::min(q.tier, 3); + ++sent.sizes.sent[tier]; + sent.sizes.sent_bytes[tier] += added.bytes + 8; + sent.sizes.held += q.hold_fingers; } - g.pose_sender.flush(emit); + } + g.pose_sender.flush(emit); +} +void Host::flush_poses() { + std::vector> players; + std::size_t poses{}; + for (auto &[id, guest] : guests_) { + if (!guest->upload_ack_due && guest->queued_sound.empty() && guest->queued_poses.empty()) continue; + players.emplace_back(id, guest.get()); + poses += guest->queued_poses.size(); + } + std::vector sent(players.size()); + const std::function job = [&](std::size_t index) { + flush_player(players[index].first, *players[index].second, sent[index]); + }; + // Waking the threads costs more than a pass with little in it. + if (workers_ && poses >= 256) workers_->run(players.size(), job); + else + for (std::size_t index = 0; index < players.size(); ++index) job(index); + for (const auto &done : sent) { + for (std::size_t kind = 0; kind < done.traffic.out.size(); ++kind) traffic_.total.out[kind] += done.traffic.out[kind]; + traffic_.total.snapshots += done.traffic.snapshots; + pose_sizes_.whole_sent += done.sizes.whole_sent; + pose_sizes_.whole_sent_bytes += done.sizes.whole_sent_bytes; + pose_sizes_.held += done.sizes.held; + for (std::size_t tier = 0; tier < done.sizes.sent.size(); ++tier) { + pose_sizes_.sent[tier] += done.sizes.sent[tier]; + pose_sizes_.sent_bytes[tier] += done.sizes.sent_bytes[tier]; + } + for (const auto &[from, sequence] : done.whole) + for (auto &kept : from->kept_poses) + if (kept.sequence == sequence) kept.keep = 3; } } // A player said which messages of poses they read in full: the poses in those are ones they hold. @@ -1101,10 +1150,47 @@ void Host::receive(std::uint64_t peer, std::span bytes, std: broadcast(p, false, true, p.source); return; } - if (accept_data(*link, p)) - broadcast(p, p.kind == PacketKind::audio && std::any_of(p.audio.begin(), p.audio.end(), - [](const auto &sample) { return sample.event; }), - true, p.source); + if (!accept_data(*link, p)) return; + // A pose waits until the pass has read everything (relay_poses). + if (p.kind == PacketKind::pose) { + link->relay_poses.push_back(p); + return; + } + // A contact's effect is only worth showing as it happens: with the server behind, a + // player's backlog of them is not passed on. + if (p.kind == PacketKind::effects && ++link->effects_pass > 2) { + ++shed_; + return; + } + broadcast(p, p.kind == PacketKind::audio && std::any_of(p.audio.begin(), p.audio.end(), + [](const auto &sample) { return sample.event; }), + true, p.source); +} +// The poses read this pass go on to everyone. A pass normally reads one or two of each player's. +// More means the server is behind (a stall, or more players than it can carry), and passing +// every one of them on is what kept it behind: each pass then had the whole of the last +// pass's arrivals to send, took as long again, and sent poses whole because the ones they +// would have built on had gone stale, which is more work still. So of a player's backlog only +// the newest goes on. Their skater is where they are now; the others see fewer poses of them +// for a moment, which their games already smooth over. +void Host::relay_poses() { + std::vector poses; + for (auto &[id, guest] : guests_) { + guest->effects_pass = 0; + if (guest->relay_poses.empty()) continue; + auto own = std::exchange(guest->relay_poses, {}); + const std::size_t first = own.size() > 3 ? own.size() - 1 : 0; + shed_ += first; + for (std::size_t index = first; index < own.size(); ++index) poses.push_back(std::move(own[index])); + } + for (const auto &pose : poses) broadcast(pose, false, true, pose.source); + if (shed_ != shed_logged_ && now_ - shed_log_at_ >= 30000000) { + log_("[network] The server is behind: " + std::to_string(shed_ - shed_logged_) + + " poses and effects that arrived late were not passed on. Players see each other at a lower rate until it catches up" + + (shed_log_at_ ? "" : "; if this keeps coming, the server has more players than its CPU can carry") + "."); + shed_logged_ = shed_; + shed_log_at_ = now_; + } } void Host::receive_cosmetics() { for (auto &[id, guest] : guests_) { @@ -1373,6 +1459,7 @@ std::string Host::network_report(std::string_view player, bool console) { text += std::string(i ? ", " : " ") + guest_name(*uploads[i].guest) + " " + number(uploads[i].total) + " KB/s (" + traffic_names[uploads[i].most] + " " + number(rate(uploads[i].guest->traffic, false, uploads[i].most)) + ")"; } + text += "\nshed while behind: " + std::to_string(shed_) + " poses and effects not passed on (since the server started)"; text += "\nsends: " + std::to_string(net.send_failures) + " failed, " + std::to_string(net.skipped) + " skipped, " + std::to_string(net.dropped) + " dropped (since the server started)"; // The connections doing worst first: a long queue, poor delivery, silence, then ping. @@ -1488,6 +1575,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, message.arrived); + relay_poses(); flush_poses(); receive_cosmetics(); introduce(); diff --git a/Server/server_host.h b/Server/server_host.h index cb6fbc7..90a6d6b 100644 --- a/Server/server_host.h +++ b/Server/server_host.h @@ -1,5 +1,6 @@ #pragma once #include "server_activity.h" +#include "worker_pool.h" #include "server_config.h" #include "speed_check.h" #include "Engine/Game/Multiplayer/chat_rate.h" @@ -12,6 +13,7 @@ #include "Extension/Multiplayer/Session/room.h" #include "Extension/Multiplayer/Steam/steam_transport.h" #include "Engine/Game/World/world_layers.h" +#include #include #include #include @@ -145,6 +147,10 @@ class Host { OutfitBudget outfit_budget; SoundBudget sound_budget; EffectBudget effect_budget; + // Their poses read this pass, passed on once all of the pass's messages are read + // (relay_poses), and how many of their effects packets were passed on in it. + std::vector relay_poses; + unsigned effects_pass{}; ChatRate chat_rate; ChatBudget admin_budget, throwdown_budget, party_budget; SpeedCheck speed; // how fast their game runs, from their pose timestamps @@ -253,9 +259,27 @@ class Host { std::array changed{}; // positions as floats, positions in mm, rotations, scales std::uint64_t bones{}; } pose_sizes_; + struct Flushed { + Traffic traffic; + PoseSizes sizes; + std::vector> whole; + }; + // The threads that share a pass's sending (none: this one does it all), and the lock + // each send into the transport is made under. + std::unique_ptr workers_; + std::mutex send_mutex_; void measure_pose(Guest &from, const Packet &packet); Guest::KeptPose *keep_pose(Guest &from, const Packet &packet); + // What sending one player their part of a pass added to the server's own counts, and the + // whole poses it sent (the player they are of, and which): see flush_player. + struct PoseSizes; + struct Flushed; + void flush_player(std::uint64_t id, Guest &g, Flushed &sent); void flush_poses(); + void relay_poses(); + // Poses and effects that arrived while the server was behind and were not passed on: in + // all, as of the last log line about it, and when that line was written. + std::uint64_t shed_{}, shed_logged_{}, shed_log_at_{}; void pose_ack(Guest &guest, const pose_batch::Ack &ack); std::uint64_t traffic_mark_{}, traffic_window_us_{}; // when `mark` was taken, and how long `last` covers void meet_later(Guest &guest); diff --git a/Server/worker_pool.h b/Server/worker_pool.h new file mode 100644 index 0000000..d3821f4 --- /dev/null +++ b/Server/worker_pool.h @@ -0,0 +1,95 @@ +#pragma once +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +namespace dingosdk::server { +// A few threads that share one batch of jobs with the thread that asks, which waits for the +// last of them. For the parts of the server's pass that are the same work once per player and +// touch nothing of each other's. +class WorkerPool { + public: + // `threads`: how many besides the caller's own; 0 runs everything on the caller's. + explicit WorkerPool(unsigned threads) { + try { + for (unsigned index = 0; index < threads; ++index) threads_.emplace_back([this] { work(); }); + } catch (const std::system_error &) {} // fewer than asked for, or none + } + ~WorkerPool() { + { + std::lock_guard lock(mutex_); + stopping_ = true; + } + wake_.notify_all(); + } + WorkerPool(const WorkerPool &) = delete; + WorkerPool &operator=(const WorkerPool &) = delete; + std::size_t threads() const noexcept { return threads_.size(); } + // job(0) ... job(count - 1), each once, on these threads and the caller's; back when all are + // done. A job must not throw. + void run(std::size_t count, const std::function &job) { + if (!count) return; + if (threads_.empty() || count == 1) { + for (std::size_t index = 0; index < count; ++index) job(index); + return; + } + // Each batch is its own: a thread still leaving the last one finds nothing in it to take. + const auto batch = std::make_shared(&job, count); + { + std::lock_guard lock(mutex_); + batch_ = batch; + ++number_; + } + wake_.notify_all(); + take(*batch); + std::unique_lock lock(mutex_); + done_.wait(lock, [&] { return !batch->left; }); + batch_.reset(); + } + + private: + struct Batch { + Batch(const std::function *work, std::size_t jobs) : job(work), count(jobs), left(jobs) {} + const std::function *job; + const std::size_t count; + std::atomic next{0}; + std::size_t left; // under the pool's mutex + }; + void take(Batch &batch) { + for (;;) { + const auto index = batch.next.fetch_add(1, std::memory_order_relaxed); + if (index >= batch.count) return; + (*batch.job)(index); + std::lock_guard lock(mutex_); + if (!--batch.left) done_.notify_all(); + } + } + void work() { + std::uint64_t seen{}; + for (;;) { + std::shared_ptr batch; + { + std::unique_lock lock(mutex_); + wake_.wait(lock, [&] { return stopping_ || number_ != seen; }); + if (stopping_) return; + seen = number_; + batch = batch_; + } + if (batch) take(*batch); + } + } + std::mutex mutex_; + std::condition_variable wake_, done_; + std::shared_ptr batch_; + std::uint64_t number_{}; + bool stopping_{}; + std::vector threads_; // last: joined before the rest goes +}; +} // namespace dingosdk::server