[network] use jthread and use std::vector for packet list instead of std::list
All checks were successful
eden-license / license-header (pull_request) Successful in 27s

Signed-off-by: lizzie <lizzie@eden-emu.dev>
This commit is contained in:
lizzie 2025-08-13 10:16:49 +01:00 committed by crueter
parent 57fbdd516e
commit d56e5dd74f
4 changed files with 221 additions and 247 deletions

View file

@ -19,9 +19,6 @@
namespace Core { namespace Core {
// Time between room is announced to web_service
static constexpr std::chrono::seconds announce_time_interval(15);
AnnounceMultiplayerSession::AnnounceMultiplayerSession() { AnnounceMultiplayerSession::AnnounceMultiplayerSession() {
#ifdef ENABLE_WEB_SERVICE #ifdef ENABLE_WEB_SERVICE
backend = std::make_unique<WebService::RoomJson>(Settings::values.web_api_url.GetValue(), backend = std::make_unique<WebService::RoomJson>(Settings::values.web_api_url.GetValue(),
@ -53,18 +50,58 @@ WebService::WebResult AnnounceMultiplayerSession::Register() {
} }
void AnnounceMultiplayerSession::Start() { void AnnounceMultiplayerSession::Start() {
if (announce_multiplayer_thread) { if (announce_multiplayer_thread.has_value()) {
Stop(); Stop();
} }
shutdown_event.Reset(); announce_multiplayer_thread.emplace([&](std::stop_token stoken) {
announce_multiplayer_thread = // Invokes all current bound error callbacks.
std::make_unique<std::thread>(&AnnounceMultiplayerSession::AnnounceMultiplayerLoop, this); const auto ErrorCallback = [this](WebService::WebResult result) {
std::lock_guard lock(callback_mutex);
for (auto callback : error_callbacks)
(*callback)(result);
};
if (!registered) {
WebService::WebResult result = Register();
if (result.result_code != WebService::WebResult::Code::Success) {
ErrorCallback(result);
return;
}
}
// Time between room is announced to web_service
std::chrono::seconds const announce_timeslice(15);
auto update_time = std::chrono::steady_clock::now();
std::future<WebService::WebResult> future;
while (!shutdown_event.WaitUntil(update_time)) {
update_time = std::chrono::steady_clock::now() + announce_timeslice;
auto room = Network::GetRoom().lock();
if (!room) {
break;
}
if (room->GetState() != Network::Room::State::Open) {
break;
}
UpdateBackendData(room);
WebService::WebResult result = backend->Update();
if (result.result_code != WebService::WebResult::Code::Success) {
ErrorCallback(result);
}
if (result.result_string == "404") {
registered = false;
// Needs to register the room again
WebService::WebResult register_result = Register();
if (register_result.result_code != WebService::WebResult::Code::Success) {
ErrorCallback(register_result);
}
}
}
});
} }
void AnnounceMultiplayerSession::Stop() { void AnnounceMultiplayerSession::Stop() {
if (announce_multiplayer_thread) { if (announce_multiplayer_thread.has_value()) {
shutdown_event.Set(); shutdown_event.Set();
announce_multiplayer_thread->join();
announce_multiplayer_thread.reset(); announce_multiplayer_thread.reset();
backend->Delete(); backend->Delete();
registered = false; registered = false;
@ -101,58 +138,10 @@ void AnnounceMultiplayerSession::UpdateBackendData(std::shared_ptr<Network::Room
} }
} }
void AnnounceMultiplayerSession::AnnounceMultiplayerLoop() {
// Invokes all current bound error callbacks.
const auto ErrorCallback = [this](WebService::WebResult result) {
std::lock_guard lock(callback_mutex);
for (auto callback : error_callbacks) {
(*callback)(result);
}
};
if (!registered) {
WebService::WebResult result = Register();
if (result.result_code != WebService::WebResult::Code::Success) {
ErrorCallback(result);
return;
}
}
auto update_time = std::chrono::steady_clock::now();
std::future<WebService::WebResult> future;
while (!shutdown_event.WaitUntil(update_time)) {
update_time += announce_time_interval;
auto room = Network::GetRoom().lock();
if (!room) {
break;
}
if (room->GetState() != Network::Room::State::Open) {
break;
}
UpdateBackendData(room);
WebService::WebResult result = backend->Update();
if (result.result_code != WebService::WebResult::Code::Success) {
ErrorCallback(result);
}
if (result.result_string == "404") {
registered = false;
// Needs to register the room again
WebService::WebResult register_result = Register();
if (register_result.result_code != WebService::WebResult::Code::Success) {
ErrorCallback(register_result);
}
}
}
}
AnnounceMultiplayerRoom::RoomList AnnounceMultiplayerSession::GetRoomList() { AnnounceMultiplayerRoom::RoomList AnnounceMultiplayerSession::GetRoomList() {
return backend->GetRoomList(); return backend->GetRoomList();
} }
bool AnnounceMultiplayerSession::IsRunning() const {
return announce_multiplayer_thread != nullptr;
}
void AnnounceMultiplayerSession::UpdateCredentials() { void AnnounceMultiplayerSession::UpdateCredentials() {
ASSERT_MSG(!IsRunning(), "Credentials can only be updated when session is not running"); ASSERT_MSG(!IsRunning(), "Credentials can only be updated when session is not running");
#ifdef ENABLE_WEB_SERVICE #ifdef ENABLE_WEB_SERVICE

View file

@ -1,8 +1,7 @@
// SPDX-FileCopyrightText: Copyright 2017 Citra Emulator Project
// SPDX-License-Identifier: GPL-2.0-or-later
// SPDX-FileCopyrightText: Copyright 2025 Eden Emulator Project // SPDX-FileCopyrightText: Copyright 2025 Eden Emulator Project
// SPDX-License-Identifier: GPL-3.0-or-later // SPDX-License-Identifier: GPL-3.0-or-later
// SPDX-FileCopyrightText: Copyright 2017 Citra Emulator Project
// SPDX-License-Identifier: GPL-2.0-or-later
#pragma once #pragma once
@ -73,7 +72,9 @@ public:
/** /**
* Whether the announce session is still running * Whether the announce session is still running
*/ */
bool IsRunning() const; [[nodiscard]] bool IsRunning() const {
return announce_multiplayer_thread.has_value();
}
/** /**
* Recreates the backend, updating the credentials. * Recreates the backend, updating the credentials.
@ -83,16 +84,13 @@ public:
private: private:
void UpdateBackendData(std::shared_ptr<Network::Room> room); void UpdateBackendData(std::shared_ptr<Network::Room> room);
void AnnounceMultiplayerLoop();
Common::Event shutdown_event; Common::Event shutdown_event;
std::mutex callback_mutex;
std::set<CallbackHandle> error_callbacks; std::set<CallbackHandle> error_callbacks;
std::unique_ptr<std::thread> announce_multiplayer_thread; std::optional<std::jthread> announce_multiplayer_thread;
/// Backend interface that logs fields /// Backend interface that logs fields
std::unique_ptr<AnnounceMultiplayerRoom::Backend> backend; std::unique_ptr<AnnounceMultiplayerRoom::Backend> backend;
std::mutex callback_mutex;
std::atomic_bool registered = false; ///< Whether the room has been registered std::atomic_bool registered = false; ///< Whether the room has been registered
}; };

View file

@ -1,6 +1,5 @@
// SPDX-FileCopyrightText: Copyright 2025 Eden Emulator Project // SPDX-FileCopyrightText: Copyright 2025 Eden Emulator Project
// SPDX-License-Identifier: GPL-3.0-or-later // SPDX-License-Identifier: GPL-3.0-or-later
// SPDX-FileCopyrightText: Copyright 2017 Citra Emulator Project // SPDX-FileCopyrightText: Copyright 2017 Citra Emulator Project
// SPDX-License-Identifier: GPL-2.0-or-later // SPDX-License-Identifier: GPL-2.0-or-later
@ -54,13 +53,11 @@ public:
RoomImpl() : random_gen(std::random_device()()) {} RoomImpl() : random_gen(std::random_device()()) {}
/// Thread that receives and dispatches network packets /// Thread that receives and dispatches network packets
std::unique_ptr<std::thread> room_thread; std::optional<std::jthread> room_thread;
/// Verification backend of the room /// Verification backend of the room
std::unique_ptr<VerifyUser::Backend> verify_backend; std::unique_ptr<VerifyUser::Backend> verify_backend;
/// Thread function that will receive and dispatch messages until the room is destroyed.
void ServerLoop();
void StartLoop(); void StartLoop();
/** /**
@ -240,59 +237,57 @@ public:
}; };
// RoomImpl // RoomImpl
void Room::RoomImpl::ServerLoop() { void Room::RoomImpl::StartLoop() {
while (state != State::Closed) { room_thread.emplace([&](std::stop_token stoken) {
ENetEvent event; while (state != State::Closed) {
if (enet_host_service(server, &event, 5) > 0) { ENetEvent event;
switch (event.type) { if (enet_host_service(server, &event, 5) > 0) {
case ENET_EVENT_TYPE_RECEIVE: switch (event.type) {
switch (event.packet->data[0]) { case ENET_EVENT_TYPE_RECEIVE:
case IdJoinRequest: switch (event.packet->data[0]) {
HandleJoinRequest(&event); case IdJoinRequest:
HandleJoinRequest(&event);
break;
case IdSetGameInfo:
HandleGameInfoPacket(&event);
break;
case IdProxyPacket:
HandleProxyPacket(&event);
break;
case IdLdnPacket:
HandleLdnPacket(&event);
break;
case IdChatMessage:
HandleChatPacket(&event);
break;
// Moderation
case IdModKick:
HandleModKickPacket(&event);
break;
case IdModBan:
HandleModBanPacket(&event);
break;
case IdModUnban:
HandleModUnbanPacket(&event);
break;
case IdModGetBanList:
HandleModGetBanListPacket(&event);
break;
}
enet_packet_destroy(event.packet);
break; break;
case IdSetGameInfo: case ENET_EVENT_TYPE_DISCONNECT:
HandleGameInfoPacket(&event); HandleClientDisconnection(event.peer);
break; break;
case IdProxyPacket: case ENET_EVENT_TYPE_NONE:
HandleProxyPacket(&event); case ENET_EVENT_TYPE_CONNECT:
break;
case IdLdnPacket:
HandleLdnPacket(&event);
break;
case IdChatMessage:
HandleChatPacket(&event);
break;
// Moderation
case IdModKick:
HandleModKickPacket(&event);
break;
case IdModBan:
HandleModBanPacket(&event);
break;
case IdModUnban:
HandleModUnbanPacket(&event);
break;
case IdModGetBanList:
HandleModGetBanListPacket(&event);
break; break;
} }
enet_packet_destroy(event.packet);
break;
case ENET_EVENT_TYPE_DISCONNECT:
HandleClientDisconnection(event.peer);
break;
case ENET_EVENT_TYPE_NONE:
case ENET_EVENT_TYPE_CONNECT:
break;
} }
} }
} // Close the connection to all members:
// Close the connection to all members: SendCloseMessage();
SendCloseMessage(); });
}
void Room::RoomImpl::StartLoop() {
room_thread = std::make_unique<std::thread>(&Room::RoomImpl::ServerLoop, this);
} }
void Room::RoomImpl::HandleJoinRequest(const ENetEvent* event) { void Room::RoomImpl::HandleJoinRequest(const ENetEvent* event) {
@ -1132,7 +1127,6 @@ void Room::SetVerifyUID(const std::string& uid) {
void Room::Destroy() { void Room::Destroy() {
room_impl->state = State::Closed; room_impl->state = State::Closed;
room_impl->room_thread->join();
room_impl->room_thread.reset(); room_impl->room_thread.reset();
if (room_impl->server) { if (room_impl->server) {

View file

@ -1,3 +1,5 @@
// SPDX-FileCopyrightText: Copyright 2025 Eden Emulator Project
// SPDX-License-Identifier: GPL-3.0-or-later
// SPDX-FileCopyrightText: Copyright 2017 Citra Emulator Project // SPDX-FileCopyrightText: Copyright 2017 Citra Emulator Project
// SPDX-License-Identifier: GPL-2.0-or-later // SPDX-License-Identifier: GPL-2.0-or-later
@ -18,6 +20,21 @@ constexpr u32 ConnectionTimeoutMs = 5000;
class RoomMember::RoomMemberImpl { class RoomMember::RoomMemberImpl {
public: public:
void SetState(const State new_state) noexcept {
if (state != new_state) {
state = new_state;
Invoke<State>(state);
}
}
void SetError(const Error new_error) noexcept {
Invoke<Error>(new_error);
}
[[nodiscard]] bool IsConnected() const noexcept {
return state == State::Joining || state == State::Joined || state == State::Moderator;
}
ENetHost* client = nullptr; ///< ENet network interface. ENetHost* client = nullptr; ///< ENet network interface.
ENetPeer* server = nullptr; ///< The server peer the client is connected to ENetPeer* server = nullptr; ///< The server peer the client is connected to
@ -30,9 +47,6 @@ public:
GameInfo current_game_info; GameInfo current_game_info;
std::atomic<State> state{State::Idle}; ///< Current state of the RoomMember. std::atomic<State> state{State::Idle}; ///< Current state of the RoomMember.
void SetState(const State new_state);
void SetError(const Error new_error);
bool IsConnected() const;
std::string nickname; ///< The nickname of this member. std::string nickname; ///< The nickname of this member.
@ -43,9 +57,9 @@ public:
std::mutex network_mutex; ///< Mutex that controls access to the `client` variable. std::mutex network_mutex; ///< Mutex that controls access to the `client` variable.
/// Thread that receives and dispatches network packets /// Thread that receives and dispatches network packets
std::unique_ptr<std::thread> loop_thread; std::optional<std::jthread> loop_thread;
std::mutex send_list_mutex; ///< Mutex that controls access to the `send_list` variable. std::mutex send_list_mutex; ///< Mutex that controls access to the `send_list` variable.
std::list<Packet> send_list; ///< A list that stores all packets to send the async std::vector<Packet> send_list; ///< A list that stores all packets to send the async
template <typename T> template <typename T>
using CallbackSet = std::set<CallbackHandle<T>>; using CallbackSet = std::set<CallbackHandle<T>>;
@ -68,8 +82,6 @@ public:
}; };
Callbacks callbacks; ///< All CallbackSets to all events Callbacks callbacks; ///< All CallbackSets to all events
void MemberLoop();
void StartLoop(); void StartLoop();
/** /**
@ -146,134 +158,117 @@ public:
}; };
// RoomMemberImpl // RoomMemberImpl
void RoomMember::RoomMemberImpl::SetState(const State new_state) { void RoomMember::RoomMemberImpl::StartLoop() {
if (state != new_state) { loop_thread.emplace([&](std::stop_token stoken) {
state = new_state; // Receive packets while the connection is open
Invoke<State>(state); while (IsConnected()) {
} std::lock_guard lock(network_mutex);
} ENetEvent event;
if (enet_host_service(client, &event, 5) > 0) {
void RoomMember::RoomMemberImpl::SetError(const Error new_error) { switch (event.type) {
Invoke<Error>(new_error); case ENET_EVENT_TYPE_RECEIVE:
} switch (event.packet->data[0]) {
case IdProxyPacket:
bool RoomMember::RoomMemberImpl::IsConnected() const { HandleProxyPackets(&event);
return state == State::Joining || state == State::Joined || state == State::Moderator; break;
} case IdLdnPacket:
HandleLdnPackets(&event);
void RoomMember::RoomMemberImpl::MemberLoop() { break;
// Receive packets while the connection is open case IdChatMessage:
while (IsConnected()) { HandleChatPacket(&event);
std::lock_guard lock(network_mutex); break;
ENetEvent event; case IdStatusMessage:
if (enet_host_service(client, &event, 5) > 0) { HandleStatusMessagePacket(&event);
switch (event.type) { break;
case ENET_EVENT_TYPE_RECEIVE: case IdRoomInformation:
switch (event.packet->data[0]) { HandleRoomInformationPacket(&event);
case IdProxyPacket: break;
HandleProxyPackets(&event); case IdJoinSuccess:
case IdJoinSuccessAsMod:
// The join request was successful, we are now in the room.
// If we joined successfully, there must be at least one client in the room: us.
ASSERT_MSG(member_information.size() > 0,
"We have not yet received member information.");
HandleJoinPacket(&event); // Get the MAC Address for the client
if (event.packet->data[0] == IdJoinSuccessAsMod) {
SetState(State::Moderator);
} else {
SetState(State::Joined);
}
break;
case IdModBanListResponse:
HandleModBanListResponsePacket(&event);
break;
case IdRoomIsFull:
SetState(State::Idle);
SetError(Error::RoomIsFull);
break;
case IdNameCollision:
SetState(State::Idle);
SetError(Error::NameCollision);
break;
case IdIpCollision:
SetState(State::Idle);
SetError(Error::IpCollision);
break;
case IdVersionMismatch:
SetState(State::Idle);
SetError(Error::WrongVersion);
break;
case IdWrongPassword:
SetState(State::Idle);
SetError(Error::WrongPassword);
break;
case IdCloseRoom:
SetState(State::Idle);
SetError(Error::LostConnection);
break;
case IdHostKicked:
SetState(State::Idle);
SetError(Error::HostKicked);
break;
case IdHostBanned:
SetState(State::Idle);
SetError(Error::HostBanned);
break;
case IdModPermissionDenied:
SetError(Error::PermissionDenied);
break;
case IdModNoSuchUser:
SetError(Error::NoSuchUser);
break;
}
enet_packet_destroy(event.packet);
break; break;
case IdLdnPacket: case ENET_EVENT_TYPE_DISCONNECT:
HandleLdnPackets(&event); if (state == State::Joined || state == State::Moderator) {
break; SetState(State::Idle);
case IdChatMessage: SetError(Error::LostConnection);
HandleChatPacket(&event);
break;
case IdStatusMessage:
HandleStatusMessagePacket(&event);
break;
case IdRoomInformation:
HandleRoomInformationPacket(&event);
break;
case IdJoinSuccess:
case IdJoinSuccessAsMod:
// The join request was successful, we are now in the room.
// If we joined successfully, there must be at least one client in the room: us.
ASSERT_MSG(member_information.size() > 0,
"We have not yet received member information.");
HandleJoinPacket(&event); // Get the MAC Address for the client
if (event.packet->data[0] == IdJoinSuccessAsMod) {
SetState(State::Moderator);
} else {
SetState(State::Joined);
} }
break; break;
case IdModBanListResponse: case ENET_EVENT_TYPE_NONE:
HandleModBanListResponsePacket(&event);
break; break;
case IdRoomIsFull: case ENET_EVENT_TYPE_CONNECT:
SetState(State::Idle); // The ENET_EVENT_TYPE_CONNECT event can not possibly happen here because we're
SetError(Error::RoomIsFull); // already connected
break; ASSERT_MSG(false, "Received unexpected connect event while already connected");
case IdNameCollision:
SetState(State::Idle);
SetError(Error::NameCollision);
break;
case IdIpCollision:
SetState(State::Idle);
SetError(Error::IpCollision);
break;
case IdVersionMismatch:
SetState(State::Idle);
SetError(Error::WrongVersion);
break;
case IdWrongPassword:
SetState(State::Idle);
SetError(Error::WrongPassword);
break;
case IdCloseRoom:
SetState(State::Idle);
SetError(Error::LostConnection);
break;
case IdHostKicked:
SetState(State::Idle);
SetError(Error::HostKicked);
break;
case IdHostBanned:
SetState(State::Idle);
SetError(Error::HostBanned);
break;
case IdModPermissionDenied:
SetError(Error::PermissionDenied);
break;
case IdModNoSuchUser:
SetError(Error::NoSuchUser);
break; break;
} }
enet_packet_destroy(event.packet);
break;
case ENET_EVENT_TYPE_DISCONNECT:
if (state == State::Joined || state == State::Moderator) {
SetState(State::Idle);
SetError(Error::LostConnection);
}
break;
case ENET_EVENT_TYPE_NONE:
break;
case ENET_EVENT_TYPE_CONNECT:
// The ENET_EVENT_TYPE_CONNECT event can not possibly happen here because we're
// already connected
ASSERT_MSG(false, "Received unexpected connect event while already connected");
break;
} }
std::vector<Packet> packets;
{
std::lock_guard send_lock(send_list_mutex);
packets.swap(send_list);
}
for (auto const& packet : packets) {
ENetPacket* enetPacket = enet_packet_create(packet.GetData(), packet.GetDataSize(),
ENET_PACKET_FLAG_RELIABLE);
enet_peer_send(server, 0, enetPacket);
}
enet_host_flush(client);
} }
std::list<Packet> packets; Disconnect();
{ });
std::lock_guard send_lock(send_list_mutex);
packets.swap(send_list);
}
for (const auto& packet : packets) {
ENetPacket* enetPacket = enet_packet_create(packet.GetData(), packet.GetDataSize(),
ENET_PACKET_FLAG_RELIABLE);
enet_peer_send(server, 0, enetPacket);
}
enet_host_flush(client);
}
Disconnect();
};
void RoomMember::RoomMemberImpl::StartLoop() {
loop_thread = std::make_unique<std::thread>(&RoomMember::RoomMemberImpl::MemberLoop, this);
} }
void RoomMember::RoomMemberImpl::Send(Packet&& packet) { void RoomMember::RoomMemberImpl::Send(Packet&& packet) {
@ -747,9 +742,7 @@ void RoomMember::Unbind(CallbackHandle<T> handle) {
void RoomMember::Leave() { void RoomMember::Leave() {
room_member_impl->SetState(State::Idle); room_member_impl->SetState(State::Idle);
room_member_impl->loop_thread->join();
room_member_impl->loop_thread.reset(); room_member_impl->loop_thread.reset();
enet_host_destroy(room_member_impl->client); enet_host_destroy(room_member_impl->client);
room_member_impl->client = nullptr; room_member_impl->client = nullptr;
} }