Server: shed stale poses when behind, and send each pass on several threads

A full server (about 120 players) slowed down at random until everyone lagged:
one pass a second, each taking over six seconds, 115% busy, and whole poses
going out by the thousand.

Why it stuck. After any stall the next pass had everything that arrived
meanwhile to pass on, to every player, so it took longer still. By then the
poses each update would have built on had gone stale, so poses went out whole,
ten times the size and more work again. Each pass carried the whole of the
last one's arrivals and the server never caught up.

Shedding. A pass normally reads one or two poses from each player. When more
than three of one player's are waiting, only the newest is passed on. Contact
effects are held to two packets a player a pass. Each pass then has a bounded
amount to do whatever the backlog, so the server catches up; players see the
others at a lower rate for a moment. The log says when it happens (at most
every 30 s) and `net` has a "shed while behind" count.

Threads. Building each player's update of everyone else was 80% of a pass. It
reads the other players and writes only that player's own state, so it is now
done for several players at once (worker_pool.h); the server's own counts and
the marks on whole poses are added up afterwards, and sends into the transport
are made one at a time. New setting network.threads: 0 (default) is one thread
for each processor but one, up to 8; 1 is the single thread as before. The
server says how many it uses when it starts.

Load test. Server/Test/server_load_tests.cpp drives the host with simulated
players over a transport that is not Steam's (built with the multiplayer tests
on Windows, not run by ctest). With 120 players at 20 TPS and 27 MB/s out, the
loop's share of one core went from about 50% on one thread to 26% on four and
22% on eight, and every player was sent byte for byte the same on 1, 2, 4 and
8 threads. A simulated 3 s stall shed about 7,000 poses, took one 169 ms pass
to recover and lost nobody.

The test's transport has none of Steam's send cost, so a real server's loop
sits higher than these figures. Receiving, choosing who is sent which pose and
the sends themselves still run on one thread.

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)

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

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;

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 <algorithm>
+#include <chrono>
+#include <cmath>
+#include <cstdio>
+#include <cstdlib>
+#include <deque>
+#include <map>
+#include <mutex>
+#include <string>
+#include <vector>
+
+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<TransportMessage> 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<std::uint32_t> batches; // pose messages to say were read
+        std::vector<std::vector<std::uint8_t>> acks; // the server's, of this player's own uploads
+    };
+    std::map<std::uint64_t, Player> players;
+    std::map<std::uint64_t, std::string> 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<std::string> 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<const Member>) {}
+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<const std::uint8_t> 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<std::uint8_t>(reliable));
+        mix(static_cast<std::uint8_t>(fresh));
+    }
+    if (pose_batch::is_ack(bytes)) player.acks.emplace_back(bytes.begin(), bytes.end());
+    return true;
+}
+void SteamTransport::send_batch(std::span<TransportSend> messages) {
+    for (auto &message : messages) message.sent = send(message.id, message.bytes, message.reliable, message.fresh, message.lane);
+}
+std::vector<TransportMessage> 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<TransportLink> 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<float, 3> 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> 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<float>(now % 100000000ULL) / 1000000.f;
+    const auto phase = static_cast<float>(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<float>(bone % 13)
+                                    : std::sin(t * (1.f + static_cast<float>(bone % 7) * 0.2f) + phase + static_cast<float>(bone)) * 0.4f;
+        b.position = {static_cast<float>(bone % 5) * 0.03f, static_cast<float>(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<float>(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<unsigned>(std::atoi(argv[1])) : 120U;
+    const auto seconds = argc > 2 ? static_cast<unsigned>(std::atoi(argv[2])) : 20U;
+    const auto spacing = argc > 3 ? static_cast<float>(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<unsigned>(std::atoi(argv[6])) : 0U;
+    wire().hashing = argc <= 5 || std::atoi(argv[5]) != 0;
+    config.threads = argc > 4 ? static_cast<unsigned>(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<Simulated> players(count);
+    const auto side = static_cast<unsigned>(std::ceil(std::sqrt(static_cast<double>(count))));
+    for (unsigned index = 0; index < count; ++index) {
+        auto &player = players[index];
+        player.id = 76561198000000001ULL + index;
+        player.epoch = 5000 + index;
+        player.home = {static_cast<float>(index % side) * spacing, 10.f, static_cast<float>(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<pose_batch::KeptView> {
+                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<const std::uint8_t> 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<double, std::milli>(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<double>(seconds) - static_cast<double>(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<double>(warm) / 1000000.0,
+                w.status.peers.size());
+    for (const auto &[id, reason] : w.closed) std::printf("  closed %llu: %s\n", static_cast<unsigned long long>(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<double>(passes), longest,
+                    static_cast<double>(sent_bytes() - bytes_at_warm) / 1024.0 / measured);
+    std::printf("%llu messages sent; pose and sound hash %016llx\n", static_cast<unsigned long long>(messages),
+                static_cast<unsigned long long>(hash));
+    std::printf("%s\n", host.command("net").c_str());
+    return 0;
+}

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::vector<std::str
     c.send_rate = get("network", "send_rate", c.send_rate);
     c.crowd_budget = get("network", "crowd_budget", c.crowd_budget);
     c.pack_ms = get("network", "pack_ms", c.pack_ms);
+    c.threads = get("network", "threads", c.threads);
     c.finger_distance = get("network", "finger_distance", c.finger_distance);
     if (const auto *d = find("network", "distances"); d && d->is_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 " +

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;

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<WorkerPool>(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<const std::uint8_t> 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<const std::uint8_t> 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<const std::uint8_t> 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<pose_batch::KeptView> {
+            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<pose_batch::KeptView> {
-                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<std::size_t>(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<std::size_t>(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<std::pair<std::uint64_t, Guest *>> 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<Flushed> sent(players.size());
+    const std::function<void(std::size_t)> 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<const std::uint8_t> 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<Packet> 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();

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 <mutex>
 #include <unordered_map>
 #include <functional>
 #include <map>
@@ -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<Packet> 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, 4> changed{}; // positions as floats, positions in mm, rotations, scales
         std::uint64_t bones{};
     } pose_sizes_;
+    struct Flushed {
+        Traffic traffic;
+        PoseSizes sizes;
+        std::vector<std::pair<Guest *, std::uint32_t>> 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<WorkerPool> 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);

Server/worker_pool.h

@@ -0,0 +1,95 @@
+#pragma once
+#include <atomic>
+#include <condition_variable>
+#include <cstddef>
+#include <cstdint>
+#include <functional>
+#include <memory>
+#include <mutex>
+#include <system_error>
+#include <thread>
+#include <vector>
+
+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<void(std::size_t)> &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<Batch>(&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<void(std::size_t)> *work, std::size_t jobs) : job(work), count(jobs), left(jobs) {}
+        const std::function<void(std::size_t)> *job;
+        const std::size_t count;
+        std::atomic<std::size_t> 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> 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> batch_;
+    std::uint64_t number_{};
+    bool stopping_{};
+    std::vector<std::jthread> threads_; // last: joined before the rest goes
+};
+} // namespace dingosdk::server