Convert middleware/ and server/

* middleware/ uses DataMutex
* bap_listener uses DataMutex

Sunrise/src/middleware/content/packages/reader/package_handle_cache.cpp

@@ -4,6 +4,7 @@
 #include <cwchar>
 #include <limits>
 
+#include "core/threading/data_mutex.h"
 #include "handle_cache.h"
 
 namespace sunrise::middleware::content::packages::reader::handle_cache {
@@ -16,9 +17,12 @@ constexpr std::uint64_t kHashBasis = 14695981039346656037ULL;
 /** Standard 64-bit FNV-1a prime mixes each path character. */
 constexpr std::uint64_t kHashPrime = 1099511628211ULL;
 
-SRWLOCK g_lock{SRWLOCK_INIT};
-std::array<FileSlot, kSharedSlots> g_slots{};
-std::uint64_t g_useCounter{};
+struct CacheState {
+    std::array<FileSlot, kSharedSlots> slots{};
+    std::uint64_t useCounter{};
+};
+
+core::threading::DataMutex<CacheState> g_cache{};
 
 /** @param path Full package path. @return Its key. */
 [[nodiscard]] std::uint64_t path_hash(const Path& path) noexcept {
@@ -123,11 +127,11 @@ bool read(const Path& path, std::uint64_t offset, std::span<std::byte> output) n
     }
     // One lock covers the lookup and the read. The read is positioned on a file the next
     // caller may replace.
-    AcquireSRWLockExclusive(&g_lock);
-    const HANDLE file = acquire(g_slots, g_useCounter, path);
-    const bool complete = file != nullptr && read_positioned(file, offset, output);
-    ReleaseSRWLockExclusive(&g_lock);
-    return complete;
+    return g_cache.lock([&path, offset, output](CacheState& cache) {
+        const HANDLE file = acquire(cache.slots, cache.useCounter, path);
+        const bool complete = file != nullptr && read_positioned(file, offset, output);
+        return complete;
+    });
 }
 
 /** Reads an exact byte range through the files one reader keeps open. */
@@ -145,10 +149,10 @@ bool read(Scratch& scratch,
 
 /** Closes the shared files, which the build passes do when they finish. */
 void release() noexcept {
-    AcquireSRWLockExclusive(&g_lock);
-    close_slots(g_slots);
-    g_useCounter = 0;
-    ReleaseSRWLockExclusive(&g_lock);
+    g_cache.lock([](CacheState& cache) {
+        close_slots(cache.slots);
+        cache.useCounter = 0;
+    });
 }
 
 /** @param scratch Reader whose own files are closed. */

Sunrise/src/server/bap/bap_route.cpp

@@ -4,6 +4,8 @@
 #include <atomic>
 #include <cstdio>
 #include <limits>
+#include <mutex>
+#include <shared_mutex>
 #include <string_view>
 
 #include "../../core/logging/log.h"
@@ -15,6 +17,7 @@
 #include "activity_authority_query_owner.h"
 #include "activity_authority_reset_owner.h"
 #include "activity_mission_seed_lease.h"
+#include "core/threading/srw_lock.h"
 #include "encrypted/bap_connection_publication.h"
 #include "encrypted/push/activity/internal.h"
 #include "internal.h"
@@ -28,7 +31,7 @@ namespace layouts = state::build_data::scenarios;
 namespace roster_message = middleware::bap::activity_message::sensor_auth_update;
 namespace tables = middleware::content::packages::tables;
 
-SRWLOCK g_lock{SRWLOCK_INIT};
+core::threading::SrwLock g_lock{};
 std::array<Session, kSessionCount> g_sessions{};
 Scratch g_scratch{};
 std::uint64_t g_accountGeneration{};
@@ -847,7 +850,7 @@ std::size_t activity_link_count_locked(const state::activity::SessionBinding& bi
 bool consume(const client::network::BapRequest& request,
              client::network::BapResponse& response) noexcept {
     response = {};
-    AcquireSRWLockExclusive(&g_lock);
+    const std::lock_guard lock(g_lock);
     bool success = false;
     // Polls report whether they reached scratch.
     bool touchesScratch = request.event != client::network::BapEvent::poll;
@@ -870,15 +873,13 @@ bool consume(const client::network::BapRequest& request,
     if (touchesScratch) {
         SecureZeroMemory(&g_scratch, sizeof g_scratch);
     }
-    ReleaseSRWLockExclusive(&g_lock);
     return success;
 }
 
 /** Counts authenticated BAP links that currently own one exact activity generation. */
 std::size_t activity_link_count(const state::activity::SessionBinding& binding) noexcept {
-    AcquireSRWLockShared(&g_lock);
+    const std::shared_lock lock(g_lock);
     const std::size_t count = activity_link_count_locked(binding);
-    ReleaseSRWLockShared(&g_lock);
     return count;
 }
 
@@ -886,7 +887,7 @@ std::size_t activity_link_count(const state::activity::SessionBinding& binding)
 bool activity_link_view(const state::activity::SessionBinding& binding,
                         ActivityLinkView& output) noexcept {
     output = {};
-    AcquireSRWLockShared(&g_lock);
+    const std::shared_lock lock(g_lock);
     const Session* const session = unique_activity_link_locked(binding, output.matchingLinks);
     if (session != nullptr) {
         const auto region = selected_region_locked(*session);
@@ -902,7 +903,6 @@ bool activity_link_view(const state::activity::SessionBinding& binding,
         output.rosterReason = session->activityRosterReason;
         output.playerKey = encrypted::push::activity::published_player_key(*session);
     }
-    ReleaseSRWLockShared(&g_lock);
     return session != nullptr;
 }
 
@@ -911,7 +911,7 @@ ActivityMissionSeedLeaseStatus
 activity_mission_seed_available(const state::activity::SessionBinding& binding,
                                 std::uint32_t scenarioRow,
                                 std::uint64_t expectedGeneration) noexcept {
-    AcquireSRWLockExclusive(&g_lock);
+    const std::lock_guard lock(g_lock);
     Session* session = nullptr;
     std::size_t matchingLinks = 0;
     ActivityMissionSeedLeaseStatus status =
@@ -919,7 +919,6 @@ activity_mission_seed_available(const state::activity::SessionBinding& binding,
     if (status == ActivityMissionSeedLeaseStatus::ready && session->activityRosterStaged.staged) {
         status = ActivityMissionSeedLeaseStatus::outputBusy;
     }
-    ReleaseSRWLockExclusive(&g_lock);
     return status;
 }
 
@@ -930,7 +929,7 @@ activity_mission_seed_lease(const state::activity::SessionBinding& binding,
                             std::uint64_t expectedGeneration,
                             ActivityMissionSeedLeaseView& output) noexcept {
     output = {};
-    AcquireSRWLockExclusive(&g_lock);
+    const std::lock_guard lock(g_lock);
     Session* session = nullptr;
     ActivityMissionSeedLeaseStatus status = mission_seed_link_locked(
         binding, scenarioRow, expectedGeneration, session, output.matchingLinks);
@@ -940,7 +939,6 @@ activity_mission_seed_lease(const state::activity::SessionBinding& binding,
     if (status == ActivityMissionSeedLeaseStatus::ready) {
         read_mission_seed_lease(*session, output.matchingLinks, output);
     }
-    ReleaseSRWLockExclusive(&g_lock);
     return status;
 }
 
@@ -949,7 +947,7 @@ ActivityMissionSeedLeaseStatus
 select_activity_mission_seed(const state::activity::SessionBinding& binding,
                              const ActivityMissionSeedPlan& plan,
                              std::uint64_t expectedGeneration) noexcept {
-    AcquireSRWLockExclusive(&g_lock);
+    const std::lock_guard lock(g_lock);
     Session* session = nullptr;
     std::size_t matchingLinks = 0;
     ActivityMissionSeedLeaseStatus status = mission_seed_link_locked(
@@ -963,7 +961,6 @@ select_activity_mission_seed(const state::activity::SessionBinding& binding,
         if (lease.configured && same_mission_seed_plan(lease.plan, plan)) {
             // The script may select the plan the roster adopted by default. That is a selection.
             lease.scriptSelected = true;
-            ReleaseSRWLockExclusive(&g_lock);
             return ActivityMissionSeedLeaseStatus::ready;
         }
         if (lease.configured && lease.revision == (std::numeric_limits<std::uint64_t>::max)()) {
@@ -983,7 +980,6 @@ select_activity_mission_seed(const state::activity::SessionBinding& binding,
             }
             if (!regionKnown) {
                 if (lease.registeredRegionCount >= lease.registeredRegions.size()) {
-                    ReleaseSRWLockExclusive(&g_lock);
                     return ActivityMissionSeedLeaseStatus::refused;
                 }
                 lease.registeredRegions[lease.registeredRegionCount++] = plan.effectiveRegion;
@@ -1002,7 +998,6 @@ select_activity_mission_seed(const state::activity::SessionBinding& binding,
             lease.scriptSelected = true;
         }
     }
-    ReleaseSRWLockExclusive(&g_lock);
     return status;
 }
 
@@ -1011,13 +1006,12 @@ bool activity_type23_override_available(const state::activity::SessionBinding& b
                                         const activity::host::ScriptableTarget& target,
                                         std::int32_t expectedRegion,
                                         std::uint64_t expectedGeneration) noexcept {
-    AcquireSRWLockShared(&g_lock);
+    const std::shared_lock lock(g_lock);
     std::size_t linkCount = 0;
     const Session* const session = unique_activity_link_locked(binding, linkCount);
     const bool available =
         session != nullptr
         && canonical_type23_available_locked(*session, target, expectedRegion, expectedGeneration);
-    ReleaseSRWLockShared(&g_lock);
     return available;
 }
 
@@ -1027,7 +1021,7 @@ bool current_activity_link_view(std::int32_t localSliceSet,
     output = {};
     const Session* only = nullptr;
     const Session* matched = nullptr;
-    AcquireSRWLockShared(&g_lock);
+    const std::shared_lock lock(g_lock);
     for (const Session& session : g_sessions) {
         if (session.id == 0 || !session.authenticated
             || session.activity.role == ActivityClientRole::none
@@ -1053,7 +1047,6 @@ bool current_activity_link_view(std::int32_t localSliceSet,
         output.effectiveRegion = selected_region_locked(*selected).index;
         output.publicTarget = selected->activity.role == ActivityClientRole::publicTarget;
     }
-    ReleaseSRWLockShared(&g_lock);
     return selected != nullptr;
 }
 
@@ -1061,7 +1054,7 @@ bool current_activity_link_view(std::int32_t localSliceSet,
 bool activity_replication_view(const state::activity::SessionBinding& binding,
                                ActivityReplicationView& output) noexcept {
     output = {};
-    AcquireSRWLockShared(&g_lock);
+    const std::shared_lock lock(g_lock);
     std::size_t count = 0;
     const Session* const session = unique_activity_link_locked(binding, count);
     const bool ready =
@@ -1075,7 +1068,6 @@ bool activity_replication_view(const state::activity::SessionBinding& binding,
         output.memberId = session->activityMemberKey;
         output.replicationEpoch = session->activity.replicationEpoch;
     }
-    ReleaseSRWLockShared(&g_lock);
     return ready;
 }
 
@@ -1086,7 +1078,7 @@ bool activity_replication_view_for_session(std::uint64_t activitySessionId,
     if (activitySessionId == 0) {
         return false;
     }
-    AcquireSRWLockShared(&g_lock);
+    const std::shared_lock lock(g_lock);
     const Session* selected = nullptr;
     std::size_t count = 0;
     for (const Session& session : g_sessions) {
@@ -1110,7 +1102,6 @@ bool activity_replication_view_for_session(std::uint64_t activitySessionId,
         output.memberId = selected->activityMemberKey;
         output.replicationEpoch = selected->activity.replicationEpoch;
     }
-    ReleaseSRWLockShared(&g_lock);
     return count == 1;
 }
 
@@ -1121,7 +1112,7 @@ bool activity_replication_view_for_group(std::uint64_t groupSessionId,
     if (groupSessionId == 0) {
         return false;
     }
-    AcquireSRWLockShared(&g_lock);
+    const std::shared_lock lock(g_lock);
     const Session* selected = nullptr;
     std::size_t count = 0;
     for (const Session& session : g_sessions) {
@@ -1143,7 +1134,6 @@ bool activity_replication_view_for_group(std::uint64_t groupSessionId,
         output.memberId = selected->activityMemberKey;
         output.replicationEpoch = selected->activity.replicationEpoch;
     }
-    ReleaseSRWLockShared(&g_lock);
     return count == 1;
 }
 
@@ -1151,7 +1141,7 @@ bool activity_replication_view_for_group(std::uint64_t groupSessionId,
 bool request_replication_epoch(const state::activity::SessionBinding& binding,
                                std::uint64_t expectedGeneration,
                                std::uint8_t generation) noexcept {
-    AcquireSRWLockExclusive(&g_lock);
+    const std::lock_guard lock(g_lock);
     std::size_t count = 0;
     Session* const session = unique_mutable_activity_link_locked(binding, count);
     bool queued = session != nullptr && expectedGeneration != 0
@@ -1169,7 +1159,6 @@ bool request_replication_epoch(const state::activity::SessionBinding& binding,
             session->activityKeepaliveDueTick = 0;
         }
     }
-    ReleaseSRWLockExclusive(&g_lock);
     return queued;
 }
 
@@ -1179,7 +1168,7 @@ request_activity_authority_query(const state::activity::SessionBinding& binding,
                                  std::uint64_t expectedGeneration,
                                  std::int32_t& correlation) noexcept {
     correlation = -1;
-    AcquireSRWLockExclusive(&g_lock);
+    const std::lock_guard lock(g_lock);
     std::size_t linkCount = 0;
     Session* const session = unique_mutable_activity_link_locked(binding, linkCount);
     ActivityAuthorityQueryStatus status = ActivityAuthorityQueryStatus::noActivityLink;
@@ -1189,7 +1178,6 @@ request_activity_authority_query(const state::activity::SessionBinding& binding,
                            session->activityAuthorityQuery, expectedGeneration, correlation)
                      : ActivityAuthorityQueryStatus::staleActivityClient;
     }
-    ReleaseSRWLockExclusive(&g_lock);
     return status;
 }
 
@@ -1199,7 +1187,7 @@ activity_authority_query_snapshot(const state::activity::SessionBinding& binding
                                   std::uint64_t expectedGeneration,
                                   ActivityAuthorityQuerySnapshot& output) noexcept {
     output = {};
-    AcquireSRWLockShared(&g_lock);
+    const std::shared_lock lock(g_lock);
     std::size_t linkCount = 0;
     const Session* const session = unique_activity_link_locked(binding, linkCount);
     ActivityAuthorityQueryStatus status = ActivityAuthorityQueryStatus::noActivityLink;
@@ -1209,7 +1197,6 @@ activity_authority_query_snapshot(const state::activity::SessionBinding& binding
                            session->activityAuthorityQuery, expectedGeneration, output)
                      : ActivityAuthorityQueryStatus::staleActivityClient;
     }
-    ReleaseSRWLockShared(&g_lock);
     return status;
 }
 
@@ -1219,7 +1206,7 @@ request_activity_authority_reset(const state::activity::SessionBinding& binding,
                                  std::uint64_t expectedGeneration,
                                  std::int32_t& correlation) noexcept {
     correlation = -1;
-    AcquireSRWLockExclusive(&g_lock);
+    const std::lock_guard lock(g_lock);
     std::size_t linkCount = 0;
     Session* const session = unique_mutable_activity_link_locked(binding, linkCount);
     ActivityAuthorityResetStatus status = ActivityAuthorityResetStatus::noActivityLink;
@@ -1229,7 +1216,6 @@ request_activity_authority_reset(const state::activity::SessionBinding& binding,
                            session->activityAuthorityReset, expectedGeneration, correlation)
                      : ActivityAuthorityResetStatus::staleActivityClient;
     }
-    ReleaseSRWLockExclusive(&g_lock);
     return status;
 }
 
@@ -1239,7 +1225,7 @@ activity_authority_reset_snapshot(const state::activity::SessionBinding& binding
                                   std::uint64_t expectedGeneration,
                                   ActivityAuthorityResetSnapshot& output) noexcept {
     output = {};
-    AcquireSRWLockShared(&g_lock);
+    const std::shared_lock lock(g_lock);
     std::size_t linkCount = 0;
     const Session* const session = unique_activity_link_locked(binding, linkCount);
     ActivityAuthorityResetStatus status = ActivityAuthorityResetStatus::noActivityLink;
@@ -1249,7 +1235,6 @@ activity_authority_reset_snapshot(const state::activity::SessionBinding& binding
                            session->activityAuthorityReset, expectedGeneration, output)
                      : ActivityAuthorityResetStatus::staleActivityClient;
     }
-    ReleaseSRWLockShared(&g_lock);
     return status;
 }
 
@@ -1263,7 +1248,7 @@ bool request_activity_type23_override(
     std::int32_t expectedRegion,
     std::uint64_t expectedGeneration,
     const activity::host::ScriptableOutputReservation* reservation) noexcept {
-    AcquireSRWLockExclusive(&g_lock);
+    const std::lock_guard lock(g_lock);
     std::size_t linkCount = 0;
     const Session* const session = unique_activity_link_locked(binding, linkCount);
     const bool queued =
@@ -1271,7 +1256,6 @@ bool request_activity_type23_override(
         && canonical_type23_available_locked(*session, target, expectedRegion, expectedGeneration)
         && activity::host::request_type23_override(
             binding, target, channel, value, snap, expectedGeneration, reservation);
-    ReleaseSRWLockExclusive(&g_lock);
     return queued;
 }
 
@@ -1282,14 +1266,13 @@ bool request_activity_lifetime_override(
     std::int32_t expectedRegion,
     std::uint64_t expectedGeneration,
     const activity::host::ScriptableOutputReservation* reservation) noexcept {
-    AcquireSRWLockExclusive(&g_lock);
+    const std::lock_guard lock(g_lock);
     std::size_t linkCount = 0;
     const Session* const session = unique_activity_link_locked(binding, linkCount);
     const bool queued = session != nullptr
                         && lifetime_available_locked(*session, expectedRegion, expectedGeneration)
                         && activity::host::request_lifetime_override(
                             binding, lifetimeState, expectedGeneration, reservation);
-    ReleaseSRWLockExclusive(&g_lock);
     return queued;
 }
 
@@ -1306,7 +1289,7 @@ bool request_activity_state_local_type23_override(
     std::uint32_t scenarioRow,
     std::uint32_t stateRow,
     const activity::host::ScriptableOutputReservation* reservation) noexcept {
-    AcquireSRWLockExclusive(&g_lock);
+    const std::lock_guard lock(g_lock);
     std::size_t linkCount = 0;
     const Session* const session = unique_activity_link_locked(binding, linkCount);
     const bool queued =
@@ -1326,7 +1309,6 @@ bool request_activity_state_local_type23_override(
                                                                snap,
                                                                expectedGeneration,
                                                                reservation);
-    ReleaseSRWLockExclusive(&g_lock);
     return queued;
 }
 
@@ -1342,7 +1324,7 @@ bool request_activity_sdk_auth_override(
     std::uint32_t scenarioRow,
     std::uint32_t stateRow,
     const activity::host::ScriptableOutputReservation* reservation) noexcept {
-    AcquireSRWLockExclusive(&g_lock);
+    const std::lock_guard lock(g_lock);
     std::size_t linkCount = 0;
     const Session* const session = unique_activity_link_locked(binding, linkCount);
     const bool available =
@@ -1367,7 +1349,6 @@ bool request_activity_sdk_auth_override(
                                                                      bitCount,
                                                                      expectedGeneration,
                                                                      reservation);
-    ReleaseSRWLockExclusive(&g_lock);
     return queued;
 }
 
@@ -1377,13 +1358,12 @@ bool request_activity_type31_override(
     const activity::host::ScriptableTarget& target,
     std::int32_t expectedRegion,
     const activity::host::ScriptableOutputReservation* reservation) noexcept {
-    AcquireSRWLockExclusive(&g_lock);
+    const std::lock_guard lock(g_lock);
     std::size_t linkCount = 0;
     const Session* const session = unique_activity_link_locked(binding, linkCount);
     const bool queued = expectedRegion >= 0 && session != nullptr
                         && selected_region_locked(*session).index == expectedRegion
                         && activity::host::request_type31_override(binding, target, reservation);
-    ReleaseSRWLockExclusive(&g_lock);
     return queued;
 }
 
@@ -1397,7 +1377,7 @@ bool request_activity_state_local_type31_override(
     std::uint32_t,
     std::uint32_t,
     const activity::host::ScriptableOutputReservation* reservation) noexcept {
-    AcquireSRWLockExclusive(&g_lock);
+    const std::lock_guard lock(g_lock);
     std::size_t linkCount = 0;
     const Session* const session = unique_activity_link_locked(binding, linkCount);
     const encrypted::push::activity::EffectiveRegion region =
@@ -1411,7 +1391,6 @@ bool request_activity_state_local_type31_override(
         && valid_state_local_type31_target(target, stateLocalRosterGroup)
         && activity::host::request_state_local_type31_override(
             binding, target, stateLocalRosterGroup, expectedGeneration, reservation);
-    ReleaseSRWLockExclusive(&g_lock);
     return queued;
 }
 
@@ -1425,7 +1404,7 @@ bool request_activity_state_local_sequence_override(
     std::uint32_t,
     std::uint32_t,
     const activity::host::ScriptableOutputReservation* reservation) noexcept {
-    AcquireSRWLockExclusive(&g_lock);
+    const std::lock_guard lock(g_lock);
     std::size_t linkCount = 0;
     const Session* const session = unique_activity_link_locked(binding, linkCount);
     const encrypted::push::activity::EffectiveRegion region =
@@ -1441,7 +1420,6 @@ bool request_activity_state_local_sequence_override(
         && target.authSchema == middleware::bap::activity_message::scriptable_auth::kType5Schema
         && activity::host::request_state_local_sequence_override(
             binding, target, stateLocalRosterGroup, expectedGeneration, reservation);
-    ReleaseSRWLockExclusive(&g_lock);
     return queued;
 }
 
@@ -1456,7 +1434,7 @@ bool request_activity_state_local_cinematic_override(
     std::uint32_t,
     std::uint32_t,
     const activity::host::ScriptableOutputReservation* reservation) noexcept {
-    AcquireSRWLockExclusive(&g_lock);
+    const std::lock_guard lock(g_lock);
     std::size_t linkCount = 0;
     const Session* const session = unique_activity_link_locked(binding, linkCount);
     const encrypted::push::activity::EffectiveRegion region =
@@ -1472,7 +1450,6 @@ bool request_activity_state_local_cinematic_override(
         && target.authSchema == middleware::bap::activity_message::scriptable_auth::kType6Schema
         && activity::host::request_state_local_cinematic_override(
             binding, target, stateLocalRosterGroup, active, expectedGeneration, reservation);
-    ReleaseSRWLockExclusive(&g_lock);
     return queued;
 }
 
@@ -1487,7 +1464,7 @@ bool request_activity_state_local_performance_override(
     std::uint32_t,
     std::uint32_t,
     const activity::host::ScriptableOutputReservation* reservation) noexcept {
-    AcquireSRWLockExclusive(&g_lock);
+    const std::lock_guard lock(g_lock);
     std::size_t linkCount = 0;
     const Session* const session = unique_activity_link_locked(binding, linkCount);
     const encrypted::push::activity::EffectiveRegion region =
@@ -1503,7 +1480,6 @@ bool request_activity_state_local_performance_override(
         && target.authSchema == middleware::bap::activity_message::scriptable_auth::kType42Schema
         && activity::host::request_state_local_performance_override(
             binding, target, stateLocalRosterGroup, stateNameHash, expectedGeneration, reservation);
-    ReleaseSRWLockExclusive(&g_lock);
     return queued;
 }
 
@@ -1517,7 +1493,7 @@ bool request_activity_state_local_authored_scene_override(
     std::uint32_t,
     std::uint32_t,
     const activity::host::ScriptableOutputReservation* reservation) noexcept {
-    AcquireSRWLockExclusive(&g_lock);
+    const std::lock_guard lock(g_lock);
     std::size_t linkCount = 0;
     const Session* const session = unique_activity_link_locked(binding, linkCount);
     const encrypted::push::activity::EffectiveRegion region =
@@ -1531,7 +1507,6 @@ bool request_activity_state_local_authored_scene_override(
         && valid_state_local_authored_scene_target(target, stateLocalRosterGroup)
         && activity::host::request_state_local_authored_scene_override(
             binding, target, stateLocalRosterGroup, expectedGeneration, reservation);
-    ReleaseSRWLockExclusive(&g_lock);
     return queued;
 }
 
@@ -1547,7 +1522,7 @@ bool request_activity_state_local_dialogue_override(
     std::uint32_t,
     std::uint32_t,
     const activity::host::ScriptableOutputReservation* reservation) noexcept {
-    AcquireSRWLockExclusive(&g_lock);
+    const std::lock_guard lock(g_lock);
     std::size_t linkCount = 0;
     const Session* const session = unique_activity_link_locked(binding, linkCount);
     const encrypted::push::activity::EffectiveRegion region =
@@ -1568,7 +1543,6 @@ bool request_activity_state_local_dialogue_override(
                                                                  authoredCueCount,
                                                                  expectedGeneration,
                                                                  reservation);
-    ReleaseSRWLockExclusive(&g_lock);
     return queued;
 }
 
@@ -1582,7 +1556,7 @@ bool request_activity_state_local_objective_reset(
     std::uint32_t,
     std::uint32_t,
     const activity::host::ScriptableOutputReservation* reservation) noexcept {
-    AcquireSRWLockExclusive(&g_lock);
+    const std::lock_guard lock(g_lock);
     std::size_t linkCount = 0;
     const Session* const session = unique_activity_link_locked(binding, linkCount);
     const encrypted::push::activity::EffectiveRegion region =
@@ -1598,7 +1572,6 @@ bool request_activity_state_local_objective_reset(
         && target.authSchema == middleware::bap::activity_message::scriptable_auth::kType3Schema
         && activity::host::request_state_local_objective_reset(
             binding, target, stateLocalRosterGroup, expectedGeneration, reservation);
-    ReleaseSRWLockExclusive(&g_lock);
     return queued;
 }
 
@@ -1612,7 +1585,7 @@ bool request_activity_state_local_task_override(
     std::uint32_t,
     std::uint32_t,
     const activity::host::ScriptableOutputReservation* reservation) noexcept {
-    AcquireSRWLockExclusive(&g_lock);
+    const std::lock_guard lock(g_lock);
     std::size_t linkCount = 0;
     const Session* const session = unique_activity_link_locked(binding, linkCount);
     const encrypted::push::activity::EffectiveRegion region =
@@ -1628,7 +1601,6 @@ bool request_activity_state_local_task_override(
         && target.authSchema == middleware::bap::activity_message::scriptable_auth::kType38Schema
         && activity::host::request_state_local_task_override(
             binding, target, stateLocalRosterGroup, expectedGeneration, reservation);
-    ReleaseSRWLockExclusive(&g_lock);
     return queued;
 }
 
@@ -1644,7 +1616,7 @@ bool request_activity_squad_override(
     std::uint64_t expectedGeneration,
     const activity::host::ScriptableOutputReservation* reservation,
     std::array<std::int8_t, 4> authoredProfile) noexcept {
-    AcquireSRWLockExclusive(&g_lock);
+    const std::lock_guard lock(g_lock);
     std::size_t linkCount = 0;
     const Session* const session = unique_activity_link_locked(binding, linkCount);
     const encrypted::push::activity::EffectiveRegion region =
@@ -1664,25 +1636,22 @@ bool request_activity_squad_override(
                                                                   nameHash,
                                                                   reservation,
                                                                   authoredProfile);
-    ReleaseSRWLockExclusive(&g_lock);
     return queued;
 }
 
 /** Cancels one exact typed override revision while excluding activity-link publication. */
 bool cancel_activity_scriptable_override(const state::activity::SessionBinding& binding,
                                          std::uint64_t expectedRevision) noexcept {
-    AcquireSRWLockExclusive(&g_lock);
+    const std::lock_guard lock(g_lock);
     const bool canceled =
         activity::host::cancel_pending_scriptable_override(binding, expectedRevision);
-    ReleaseSRWLockExclusive(&g_lock);
     return canceled;
 }
 
 /** Cancels a pending raw incident while excluding activity-link publication. */
 bool cancel_activity_host_incident(const state::activity::SessionBinding& binding) noexcept {
-    AcquireSRWLockExclusive(&g_lock);
+    const std::lock_guard lock(g_lock);
     const bool canceled = server::activity::host::cancel_pending_incident(binding);
-    ReleaseSRWLockExclusive(&g_lock);
     return canceled;
 }
 
@@ -1691,21 +1660,20 @@ bool cancel_activity_host_incident(const state::activity::SessionBinding& bindin
 bool session_channel(std::uint32_t connectionId,
                      std::array<std::byte, state::kBapNonceSize>& sendNonce,
                      std::array<std::byte, state::kAesKeySize>& sessionKey) noexcept {
-    AcquireSRWLockShared(&g_lock);
+    const std::shared_lock lock(g_lock);
     const Session* const session = session_for(connectionId);
     const bool armed = session != nullptr && session->authenticated;
     if (armed) {
         sendNonce = session->sendNonce;
         sessionKey = session->sessionKey;
     }
-    ReleaseSRWLockShared(&g_lock);
     return armed;
 }
 #endif
 
 /** Securely erases every connection-owned nonce and transform buffer. */
 void shutdown() noexcept {
-    AcquireSRWLockExclusive(&g_lock);
+    const std::lock_guard lock(g_lock);
     for (auto& session : g_sessions) {
         if (session.id != 0
             && session.matchmakingContext.generation != state::matchmaking::kInvalidGeneration) {
@@ -1719,7 +1687,6 @@ void shutdown() noexcept {
     SecureZeroMemory(g_sessions.data(), sizeof g_sessions);
     SecureZeroMemory(&g_scratch, sizeof g_scratch);
     g_accountGeneration = 0;
-    ReleaseSRWLockExclusive(&g_lock);
 }
 
 } // namespace sunrise::server::bap

Sunrise/src/server/transport/bap_listener.cpp

@@ -8,15 +8,14 @@
 #include "../../core/logging/log.h"
 #include "../../core/settings/settings.h"
 #include "../activity/host_runtime.h"
+#include "core/threading/data_mutex.h"
 #include "internal.h"
 
 namespace sunrise::server::transport {
 
-Listener g_listener;
-
 namespace {
 
-SRWLOCK g_listenerLock{SRWLOCK_INIT};
+core::threading::DataMutex<Listener> g_listener;
 
 /** Makes one socket nonblocking. @return True when it can no longer block its caller. */
 [[nodiscard]] bool make_nonblocking(SOCKET socket) noexcept {
@@ -25,21 +24,21 @@ SRWLOCK g_listenerLock{SRWLOCK_INIT};
 }
 
 /** @return Index of the first unused peer slot, or the slot count when all are taken. */
-[[nodiscard]] std::size_t free_slot() noexcept {
-    for (std::size_t slot = 0; slot < g_listener.peers.size(); ++slot) {
-        if (g_listener.peers[slot].socket == INVALID_SOCKET) {
+[[nodiscard]] std::size_t free_slot(const Listener& listener) noexcept {
+    for (std::size_t slot = 0; slot < listener.peers.size(); ++slot) {
+        if (listener.peers[slot].socket == INVALID_SOCKET) {
             return slot;
         }
     }
-    return g_listener.peers.size();
+    return listener.peers.size();
 }
 
 /**
  * Takes one waiting connection into a free slot and opens its Server session.
  * @param slot Peer slot already checked to be free.
  */
-void accept_peer(std::size_t slot) noexcept {
-    const SOCKET accepted = accept(g_listener.acceptor, nullptr, nullptr);
+void accept_peer(Listener& listener, std::size_t slot) noexcept {
+    const SOCKET accepted = accept(listener.acceptor, nullptr, nullptr);
     if (accepted == INVALID_SOCKET) {
         return;
     }
@@ -47,24 +46,24 @@ void accept_peer(std::size_t slot) noexcept {
         closesocket(accepted);
         return;
     }
-    Peer& peer = g_listener.peers[slot];
+    Peer& peer = listener.peers[slot];
     peer.socket = accepted;
     peer.streamSize = 0;
     peer.outputOffset = 0;
     peer.outputSize = 0;
+    peer.connectionId = connection_id(slot + 1);
+
     std::array<char, core::log::kLineCapacity> line{};
-    const int written = std::snprintf(line.data(),
-                                      line.size(),
-                                      "ev=transport stage=accept result=ok conn=%u",
-                                      connection_id(slot));
+    const int written = std::snprintf(
+        line.data(), line.size(), "ev=transport stage=accept result=ok conn=%u", peer.connectionId);
     if (written > 0) {
         const std::size_t length = static_cast<std::size_t>(written) < line.size()
                                        ? static_cast<std::size_t>(written)
                                        : line.size() - 1;
         core::log::write(core::log::Channel::server, core::log::Level::info, {line.data(), length});
     }
-    if (!offer(slot, client::network::BapEvent::open, {})) {
-        close_peer(slot);
+    if (!offer(peer, client::network::BapEvent::open, {})) {
+        close_peer(peer);
     }
 }
 
@@ -72,8 +71,7 @@ void accept_peer(std::size_t slot) noexcept {
  * Reads at most once from one readable peer.
  * @param slot Live peer slot reported readable.
  */
-void receive_peer(std::size_t slot) noexcept {
-    Peer& peer = g_listener.peers[slot];
+void receive_peer(Peer& peer) noexcept {
     const std::size_t free = kStreamCapacity - peer.streamSize;
     if (free == 0) {
         return;
@@ -87,7 +85,7 @@ void receive_peer(std::size_t slot) noexcept {
         return;
     }
     if (received == 0 || WSAGetLastError() != WSAEWOULDBLOCK) {
-        close_peer(slot);
+        close_peer(peer);
     }
 }
 
@@ -96,8 +94,7 @@ void receive_peer(std::size_t slot) noexcept {
  * @param slot Live peer slot.
  * @return True while the peer remains usable.
  */
-[[nodiscard]] bool flush_peer(std::size_t slot) noexcept {
-    Peer& peer = g_listener.peers[slot];
+[[nodiscard]] bool flush_peer(Peer& peer) noexcept {
     if (peer.outputSize == 0) {
         return true;
     }
@@ -120,56 +117,51 @@ void receive_peer(std::size_t slot) noexcept {
  * @param pollDue True on a poll tick.
  */
 void service_peer(
-    std::size_t slot, fd_set& readable, fd_set& writable, bool wasPending, bool pollDue) noexcept {
-    Peer& peer = g_listener.peers[slot];
+    Peer& peer, fd_set& readable, fd_set& writable, bool wasPending, bool pollDue) noexcept {
     bool sent = false;
     if (wasPending && FD_ISSET(peer.socket, &writable)) {
         sent = true;
-        if (!flush_peer(slot)) {
-            close_peer(slot);
+        if (!flush_peer(peer)) {
+            close_peer(peer);
             return;
         }
     }
     if (peer.socket != INVALID_SOCKET && FD_ISSET(peer.socket, &readable)) {
-        receive_peer(slot);
+        receive_peer(peer);
     }
     if (peer.socket == INVALID_SOCKET) {
         return;
     }
-    if (peer.outputSize == 0 && !drain_stream(slot)) {
-        close_peer(slot);
+    if (peer.outputSize == 0 && !drain_stream(peer)) {
+        close_peer(peer);
         return;
     }
-    if (pollDue && peer.outputSize == 0 && !offer(slot, client::network::BapEvent::poll, {})) {
-        close_peer(slot);
+    if (pollDue && peer.outputSize == 0 && !offer(peer, client::network::BapEvent::poll, {})) {
+        close_peer(peer);
         return;
     }
-    if (!wasPending && !sent && peer.outputSize != 0 && !flush_peer(slot)) {
-        close_peer(slot);
+    if (!wasPending && !sent && peer.outputSize != 0 && !flush_peer(peer)) {
+        close_peer(peer);
     }
 }
 
 } // namespace
 
 /** Starts the nonblocking loopback listener on one port. */
-bool initialize_on_port(std::uint16_t port) noexcept {
-    AcquireSRWLockExclusive(&g_listenerLock);
-    if (g_listener.active) {
-        ReleaseSRWLockExclusive(&g_listenerLock);
+bool initialize_on_port(Listener& listener, std::uint16_t port) noexcept {
+    if (listener.active) {
         return true;
     }
     // This DLL initializes before the game touches Winsock, so the listener starts it itself.
     WSADATA winsock{};
     if (WSAStartup(MAKEWORD(2, 2), &winsock) != 0) {
-        ReleaseSRWLockExclusive(&g_listenerLock);
         return false;
     }
-    g_listener.winsockOwned = true;
-    g_listener.acceptor = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP);
-    if (g_listener.acceptor == INVALID_SOCKET) {
+    listener.winsockOwned = true;
+    listener.acceptor = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP);
+    if (listener.acceptor == INVALID_SOCKET) {
         WSACleanup();
-        g_listener.winsockOwned = false;
-        ReleaseSRWLockExclusive(&g_listenerLock);
+        listener.winsockOwned = false;
         return false;
     }
     sockaddr_in address{};
@@ -177,24 +169,23 @@ bool initialize_on_port(std::uint16_t port) noexcept {
     address.sin_port = htons(port);
     address.sin_addr.s_addr = htonl(INADDR_LOOPBACK);
     BOOL reuse = TRUE;
-    (void)setsockopt(g_listener.acceptor,
+    (void)setsockopt(listener.acceptor,
                      SOL_SOCKET,
                      SO_REUSEADDR,
                      reinterpret_cast<const char*>(&reuse),
                      sizeof reuse);
-    if (!make_nonblocking(g_listener.acceptor)
-        || bind(g_listener.acceptor, reinterpret_cast<const sockaddr*>(&address), sizeof address)
+    if (!make_nonblocking(listener.acceptor)
+        || bind(listener.acceptor, reinterpret_cast<const sockaddr*>(&address), sizeof address)
                == SOCKET_ERROR
-        || listen(g_listener.acceptor, static_cast<int>(g_listener.peers.size())) == SOCKET_ERROR) {
-        closesocket(g_listener.acceptor);
-        g_listener.acceptor = INVALID_SOCKET;
+        || listen(listener.acceptor, static_cast<int>(listener.peers.size())) == SOCKET_ERROR) {
+        closesocket(listener.acceptor);
+        listener.acceptor = INVALID_SOCKET;
         WSACleanup();
-        g_listener.winsockOwned = false;
-        ReleaseSRWLockExclusive(&g_listenerLock);
+        listener.winsockOwned = false;
         return false;
     }
-    g_listener.active = true;
-    g_listener.nextPollTick = 0;
+    listener.active = true;
+    listener.nextPollTick = 0;
     std::array<char, 64> line{};
     const int written = std::snprintf(line.data(),
                                       line.size(),
@@ -205,110 +196,107 @@ bool initialize_on_port(std::uint16_t port) noexcept {
                          core::log::Level::info,
                          {line.data(), static_cast<std::size_t>(written)});
     }
-    ReleaseSRWLockExclusive(&g_listenerLock);
     return true;
 }
 
 /** Starts the nonblocking listener on the configured BAP port. */
 bool initialize() noexcept {
-    return initialize_on_port(core::settings::get().server.bapPort);
+    return g_listener.lock([](Listener& listener) {
+        return initialize_on_port(listener, core::settings::get().server.bapPort);
+    });
 }
 
 /** Runs one bounded listener slice on the caller thread. @param now Monotonic tick count. */
 void service(std::uint64_t now) noexcept {
-    if (!TryAcquireSRWLockExclusive(&g_listenerLock)) {
-        return;
-    }
-    if (!g_listener.active) {
-        ReleaseSRWLockExclusive(&g_listenerLock);
-        return;
-    }
+    g_listener.try_lock([now](Listener& listener) {
+        if (!listener.active) {
+            return;
+        }
 
-    fd_set readable;
-    fd_set writable;
-    FD_ZERO(&readable);
-    FD_ZERO(&writable);
-    std::array<bool, client::network::kBapConnectionCount> wasPending{};
-    const std::size_t accepting = free_slot();
-    // With no free slot the acceptor is left out of the set, so a connect waits in the backlog
-    // with no handshake and no other symptom. Report the edge.
-    const bool full = accepting == g_listener.peers.size();
-    if (full != g_listener.slotsFull) {
-        g_listener.slotsFull = full;
-        std::array<char, core::log::kLineCapacity> line{};
-        const int written = std::snprintf(line.data(),
-                                          line.size(),
-                                          "ev=transport stage=accept result=%s slots=%zu",
-                                          full ? "full" : "free",
-                                          g_listener.peers.size());
-        if (written > 0) {
-            core::log::write(core::log::Channel::server,
-                             full ? core::log::Level::warn : core::log::Level::info,
-                             {line.data(), static_cast<std::size_t>(written)});
+        fd_set readable;
+        fd_set writable;
+        FD_ZERO(&readable);
+        FD_ZERO(&writable);
+        std::array<bool, client::network::kBapConnectionCount> wasPending{};
+        const std::size_t accepting = free_slot(listener);
+        // With no free slot the acceptor is left out of the set, so a connect waits in the backlog
+        // with no handshake and no other symptom. Report the edge.
+        const bool full = accepting == listener.peers.size();
+        if (full != listener.slotsFull) {
+            listener.slotsFull = full;
+            std::array<char, core::log::kLineCapacity> line{};
+            const int written = std::snprintf(line.data(),
+                                              line.size(),
+                                              "ev=transport stage=accept result=%s slots=%zu",
+                                              full ? "full" : "free",
+                                              listener.peers.size());
+            if (written > 0) {
+                core::log::write(core::log::Channel::server,
+                                 full ? core::log::Level::warn : core::log::Level::info,
+                                 {line.data(), static_cast<std::size_t>(written)});
+            }
         }
-    }
-    if (!full) {
-        FD_SET(g_listener.acceptor, &readable);
-    }
-    for (std::size_t slot = 0; slot < g_listener.peers.size(); ++slot) {
-        const Peer& peer = g_listener.peers[slot];
-        if (peer.socket == INVALID_SOCKET) {
-            continue;
+        if (!full) {
+            FD_SET(listener.acceptor, &readable);
         }
-        if (peer.streamSize < kStreamCapacity) {
-            FD_SET(peer.socket, &readable);
+        for (std::size_t slot = 0; slot < listener.peers.size(); ++slot) {
+            const Peer& peer = listener.peers[slot];
+            if (peer.socket == INVALID_SOCKET) {
+                continue;
+            }
+            if (peer.streamSize < kStreamCapacity) {
+                FD_SET(peer.socket, &readable);
+            }
+            if (peer.outputSize != 0) {
+                FD_SET(peer.socket, &writable);
+                wasPending[slot] = true;
+            }
         }
-        if (peer.outputSize != 0) {
-            FD_SET(peer.socket, &writable);
-            wasPending[slot] = true;
+        timeval timeout{};
+        if (select(0, &readable, &writable, nullptr, &timeout) == SOCKET_ERROR) {
+            return;
         }
-    }
-    timeval timeout{};
-    if (select(0, &readable, &writable, nullptr, &timeout) == SOCKET_ERROR) {
-        ReleaseSRWLockExclusive(&g_listenerLock);
-        return;
-    }
 
-    const bool timedPoll = g_listener.nextPollTick == 0 || now >= g_listener.nextPollTick;
-    if (timedPoll) {
-        g_listener.nextPollTick = now + static_cast<std::uint64_t>(kServiceIntervalMs);
-    }
-    // The poll is what lets a committed answer out. Holding one for the rest of the
-    // interval costs every queued mission action a full interval of its own.
-    const bool pollDue = timedPoll || activity::host::any_output_pending();
-    if (!full && FD_ISSET(g_listener.acceptor, &readable)) {
-        accept_peer(accepting);
-    }
-    for (std::size_t slot = 0; slot < g_listener.peers.size(); ++slot) {
-        if (g_listener.peers[slot].socket != INVALID_SOCKET) {
-            service_peer(slot, readable, writable, wasPending[slot], pollDue);
+        const bool timedPoll = listener.nextPollTick == 0 || now >= listener.nextPollTick;
+        if (timedPoll) {
+            listener.nextPollTick = now + static_cast<std::uint64_t>(kServiceIntervalMs);
         }
-    }
-    ReleaseSRWLockExclusive(&g_listenerLock);
+        // The poll is what lets a committed answer out. Holding one for the rest of the
+        // interval costs every queued mission action a full interval of its own.
+        const bool pollDue = timedPoll || activity::host::any_output_pending();
+        if (!full && FD_ISSET(listener.acceptor, &readable)) {
+            accept_peer(listener, accepting);
+        }
+        for (std::size_t slot = 0; slot < listener.peers.size(); ++slot) {
+            Peer& peer = listener.peers[slot];
+            if (peer.socket != INVALID_SOCKET) {
+                service_peer(peer, readable, writable, wasPending[slot], pollDue);
+            }
+        }
+    });
 }
 
 /** Closes every socket owned by the listener. */
 void shutdown() noexcept {
-    AcquireSRWLockExclusive(&g_listenerLock);
-    if (!g_listener.active) {
-        ReleaseSRWLockExclusive(&g_listenerLock);
-        return;
-    }
-    g_listener.active = false;
-    if (g_listener.acceptor != INVALID_SOCKET) {
-        closesocket(g_listener.acceptor);
-        g_listener.acceptor = INVALID_SOCKET;
-    }
-    for (std::size_t slot = 0; slot < g_listener.peers.size(); ++slot) {
-        close_peer(slot);
-    }
-    g_listener.nextPollTick = 0;
-    g_listener.slotsFull = false;
-    if (g_listener.winsockOwned) {
-        WSACleanup();
-        g_listener.winsockOwned = false;
-    }
-    ReleaseSRWLockExclusive(&g_listenerLock);
+    g_listener.lock([](Listener& listener) {
+        if (!listener.active) {
+            return;
+        }
+        listener.active = false;
+        if (listener.acceptor != INVALID_SOCKET) {
+            closesocket(listener.acceptor);
+            listener.acceptor = INVALID_SOCKET;
+        }
+        for (Peer& peer : listener.peers) {
+            close_peer(peer);
+        }
+        listener.nextPollTick = 0;
+        listener.slotsFull = false;
+        if (listener.winsockOwned) {
+            WSACleanup();
+            listener.winsockOwned = false;
+        }
+    });
 }
 
 } // namespace sunrise::server::transport

Sunrise/src/server/transport/bap_peer_session.cpp

@@ -27,15 +27,12 @@ namespace {
 } // namespace
 
 /** Offers one event to the Server and stages whatever it produces. */
-bool offer(std::size_t slot,
-           client::network::BapEvent event,
-           std::span<const std::byte> frame) noexcept {
-    Peer& peer = g_listener.peers[slot];
+bool offer(Peer& peer, client::network::BapEvent event, std::span<const std::byte> frame) noexcept {
     if (peer.outputSize != 0) {
         return false;
     }
     client::network::BapResponse response{};
-    const client::network::BapRequest request{event, connection_id(slot), frame, peer.output};
+    const client::network::BapRequest request{event, peer.connectionId, frame, peer.output};
     if (!bap::consume(request, response) || response.size == 0) {
         return true;
     }
@@ -48,8 +45,7 @@ bool offer(std::size_t slot,
 }
 
 /** Removes and offers at most one complete frame from one peer's stream. */
-bool drain_stream(std::size_t slot) noexcept {
-    Peer& peer = g_listener.peers[slot];
+bool drain_stream(Peer& peer) noexcept {
     if (peer.outputSize != 0) {
         return true;
     }
@@ -66,7 +62,7 @@ bool drain_stream(std::size_t slot) noexcept {
     const int count = std::snprintf(line.data(),
                                     line.size(),
                                     "ev=transport stage=frame conn=%u type=%u bytes=%zu",
-                                    connection_id(slot),
+                                    peer.connectionId,
                                     static_cast<unsigned>(pending[1]),
                                     total);
     if (count > 0) {
@@ -76,7 +72,7 @@ bool drain_stream(std::size_t slot) noexcept {
         core::log::write(
             core::log::Channel::server, core::log::Level::debug, {line.data(), length});
     }
-    if (!offer(slot, client::network::BapEvent::frame, pending.first(total))) {
+    if (!offer(peer, client::network::BapEvent::frame, pending.first(total))) {
         return false;
     }
     peer.streamSize -= total;
@@ -101,17 +97,17 @@ bool advance_output(Peer& peer, std::size_t sent) noexcept {
 }
 
 /** Closes one peer and reports its session end to the Server. */
-void close_peer(std::size_t slot) noexcept {
-    Peer& peer = g_listener.peers[slot];
+void close_peer(Peer& peer) noexcept {
     if (peer.socket == INVALID_SOCKET) {
         return;
     }
     client::network::BapResponse response{};
     const client::network::BapRequest request{
-        client::network::BapEvent::close, connection_id(slot), {}, {}};
+        client::network::BapEvent::close, peer.connectionId, {}, {}};
     (void)bap::consume(request, response);
     closesocket(peer.socket);
     peer.socket = INVALID_SOCKET;
+    peer.connectionId = 0;
     peer.streamSize = 0;
     peer.outputOffset = 0;
     peer.outputSize = 0;

Sunrise/src/server/transport/internal.h

@@ -22,6 +22,7 @@ inline constexpr std::size_t kStreamCapacity = client::network::kBapFrameCapacit
 /** One accepted connection with bounded ingress and committed egress storage. */
 struct Peer {
     SOCKET socket{INVALID_SOCKET};
+    std::uint32_t connectionId{};
     std::size_t streamSize{};
     std::array<std::byte, kStreamCapacity> stream{};
     std::size_t outputOffset{};
@@ -40,8 +41,6 @@ struct Listener {
     std::array<Peer, client::network::kBapConnectionCount> peers{};
 };
 
-extern Listener g_listener;
-
 /** A peer slot answers on the connection id the Server indexes its sessions by. */
 [[nodiscard]] constexpr std::uint32_t connection_id(std::size_t slot) noexcept {
     return static_cast<std::uint32_t>(slot + 1);
@@ -53,13 +52,13 @@ extern Listener g_listener;
  * @return True when the response metadata fits the peer buffer.
  */
 [[nodiscard]] bool
-offer(std::size_t slot, client::network::BapEvent event, std::span<const std::byte> frame) noexcept;
+offer(Peer& peer, client::network::BapEvent event, std::span<const std::byte> frame) noexcept;
 
 /**
  * Removes and offers at most one whole frame from one peer's stream.
  * @return True while the buffered prefix is valid.
  */
-[[nodiscard]] bool drain_stream(std::size_t slot) noexcept;
+[[nodiscard]] bool drain_stream(Peer& peer) noexcept;
 
 /**
  * Advances one committed output by an accepted send count.
@@ -70,9 +69,6 @@ offer(std::size_t slot, client::network::BapEvent event, std::span<const std::by
 [[nodiscard]] bool advance_output(Peer& peer, std::size_t sent) noexcept;
 
 /** Closes one peer and reports its session end to the Server. */
-void close_peer(std::size_t slot) noexcept;
-
-/** @param port Host-order loopback port. Zero picks an ephemeral port. */
-[[nodiscard]] bool initialize_on_port(std::uint16_t port) noexcept;
+void close_peer(Peer& peer) noexcept;
 
 } // namespace sunrise::server::transport