From 1f55c35f2b3dea6965a40832e970b084a1d05306 Mon Sep 17 00:00:00 2001 From: Lion Kortlepel Date: Thu, 9 Apr 2026 18:54:21 +0200 Subject: [PATCH] fix ReadSocketWithTimeout to use dedicated io context polled on a new jthread jthread so it's cancellable --- include/TNetwork.h | 14 ++++ src/TNetwork.cpp | 172 +++++++++++++++++++++++++++------------------ 2 files changed, 118 insertions(+), 68 deletions(-) diff --git a/include/TNetwork.h b/include/TNetwork.h index 0d3df81..bd5ba58 100644 --- a/include/TNetwork.h +++ b/include/TNetwork.h @@ -28,6 +28,19 @@ struct TConnection; +class TIoPollThread { +public: + TIoPollThread(); + ~TIoPollThread(); + + boost::asio::io_context& IoCtx() noexcept { return mIoCtx; } + +private: + boost::asio::io_context mIoCtx; + boost::asio::executor_work_guard mWorkGuard; + std::jthread mThread; +}; + class TNetwork { public: TNetwork(TServer& Server, TPPSMonitor& PPSMonitor, TResourceManager& ResourceManager); @@ -63,6 +76,7 @@ private: std::thread mUDPThread; std::thread mTCPThread; std::mutex mOpenIDMutex; + TIoPollThread mIoCtxPoller; TConnectionLimiter mConnectionLimiter; std::vector UDPRcvFromClient(boost::asio::ip::udp::endpoint& ClientEndpoint); diff --git a/src/TNetwork.cpp b/src/TNetwork.cpp index 0ba9e76..c3bdc5e 100644 --- a/src/TNetwork.cpp +++ b/src/TNetwork.cpp @@ -28,12 +28,16 @@ #include #include #include +#include #include #include #include #include #include +#include +#include #include +#include #include #include #include @@ -57,47 +61,6 @@ static void CompressProperly(std::vector& Data) { Data = CombinedData; } -static boost::system::error_code ReadSocketWithTimeout(ip::tcp::socket& Socket, void* Buf, size_t Len, std::chrono::steady_clock::duration Timeout) { - boost::system::error_code Ec; - const bool WasNonBlocking = Socket.non_blocking(); - Socket.non_blocking(true, Ec); - if (Ec) { - return Ec; - } - - auto RestoreBlockingMode = [&]() { - boost::system::error_code IgnoreEc; - Socket.non_blocking(WasNonBlocking, IgnoreEc); - }; - - const auto Deadline = std::chrono::steady_clock::now() + Timeout; - auto* Data = static_cast(Buf); - size_t TotalRead = 0; - - while (TotalRead < Len) { - const size_t BytesRead = Socket.read_some(boost::asio::buffer(Data + TotalRead, Len - TotalRead), Ec); - if (!Ec) { - TotalRead += BytesRead; - continue; - } - - if (Ec == error::would_block || Ec == error::try_again) { - if (std::chrono::steady_clock::now() >= Deadline) { - RestoreBlockingMode(); - return error::timed_out; - } - std::this_thread::sleep_for(std::chrono::milliseconds(1)); - continue; - } - - RestoreBlockingMode(); - return Ec; - } - - RestoreBlockingMode(); - return Ec; -} - // for unit-tests, otherwise unused static bool OpenLoopbackSocketPair(io_context& IoCtx, ip::tcp::socket& ClientSocket, ip::tcp::socket& ServerSocket, boost::system::error_code& Ec) { ip::tcp::acceptor Acceptor(IoCtx); @@ -131,7 +94,8 @@ TNetwork::TNetwork(TServer& Server, TPPSMonitor& PPSMonitor, TResourceManager& R , mPPSMonitor(PPSMonitor) , mUDPSock(Server.IoCtx()) , mResourceManager(ResourceManager) - , mConnectionLimiter(MAX_CONCURRENT_CONNECTIONS, MAX_GLOBAL_CONNECTIONS){ + , mIoCtxPoller() + , mConnectionLimiter(MAX_CONCURRENT_CONNECTIONS, MAX_GLOBAL_CONNECTIONS) { Application::SetSubsystemStatus("TCPNetwork", Application::Status::Starting); Application::SetSubsystemStatus("UDPNetwork", Application::Status::Starting); Application::RegisterShutdownHandler([&] { @@ -809,49 +773,103 @@ void TNetwork::UpdatePlayer(TClient& Client) { //(void)Respond(Client, Packet, true); } -boost::system::error_code TNetwork::ReadWithTimeout(TConnection& Connection, void *Buf, size_t Len, std::chrono::steady_clock::duration Timeout) -{ - return ReadSocketWithTimeout(Connection.Socket, Buf, Len, Timeout); +static boost::system::error_code ReadSocketWithTimeout( + boost::asio::io_context& io, + boost::asio::ip::tcp::socket& socket, + void* buffer, + std::size_t length, + std::chrono::steady_clock::duration timeout); + +boost::system::error_code TNetwork::ReadWithTimeout(TConnection& Connection, void* Buf, size_t Len, std::chrono::steady_clock::duration Timeout) { + return ReadSocketWithTimeout(mIoCtxPoller.IoCtx(), Connection.Socket, Buf, Len, Timeout); +} + +static boost::system::error_code ReadSocketWithTimeout( + boost::asio::io_context& IoCtx, + boost::asio::ip::tcp::socket& Socket, + void* Buffer, + std::size_t Length, + std::chrono::steady_clock::duration Timeout) { + namespace asio = boost::asio; + using boost::system::error_code; + + struct TTimeoutState { + explicit TTimeoutState(asio::io_context& IoCtx) + : Timer(IoCtx) { } + + asio::steady_timer Timer; + std::promise> Promise; + std::atomic_bool Completed { false }; + }; + + auto State = std::make_shared(IoCtx); + auto Future = State->Promise.get_future(); + + asio::async_read( + Socket, + asio::buffer(Buffer, Length), + [State](error_code ec, std::size_t n) { + if (!State->Completed.exchange(true)) { + State->Timer.cancel(); + State->Promise.set_value({ ec, n }); + } + }); + + State->Timer.expires_after(Timeout); + State->Timer.async_wait( + [State, &Socket](error_code ec) { + if (ec == asio::error::operation_aborted) + return; + + if (!State->Completed.exchange(true)) { + error_code IgnoredEc; + Socket.cancel(IgnoredEc); + State->Promise.set_value({ asio::error::timed_out, 0 }); + } + }); + + auto [ec, NRead] = Future.get(); + return ec; } TEST_CASE("ReadSocketWithTimeout returns timed_out when peer sends no data") { - io_context IoCtx; + TIoPollThread TimerThread; boost::system::error_code Ec; - ip::tcp::socket ClientSocket(IoCtx); - ip::tcp::socket ServerSocket(IoCtx); - OpenLoopbackSocketPair(IoCtx, ClientSocket, ServerSocket, Ec); + ip::tcp::socket ClientSocket(TimerThread.IoCtx()); + ip::tcp::socket ServerSocket(TimerThread.IoCtx()); + OpenLoopbackSocketPair(TimerThread.IoCtx(), ClientSocket, ServerSocket, Ec); REQUIRE(!Ec); uint8_t ReadByte = 0; - const auto ReadEc = ReadSocketWithTimeout(ServerSocket, &ReadByte, 1, std::chrono::milliseconds(50)); + const auto ReadEc = ReadSocketWithTimeout(TimerThread.IoCtx(), ServerSocket, &ReadByte, 1, std::chrono::milliseconds(50)); CHECK(ReadEc == error::timed_out); } TEST_CASE("ReadSocketWithTimeout reads small payload") { - io_context IoCtx; + TIoPollThread TimerThread; boost::system::error_code Ec; - ip::tcp::socket ClientSocket(IoCtx); - ip::tcp::socket ServerSocket(IoCtx); - OpenLoopbackSocketPair(IoCtx, ClientSocket, ServerSocket, Ec); + ip::tcp::socket ClientSocket(TimerThread.IoCtx()); + ip::tcp::socket ServerSocket(TimerThread.IoCtx()); + OpenLoopbackSocketPair(TimerThread.IoCtx(), ClientSocket, ServerSocket, Ec); REQUIRE(!Ec); const std::array Sent { 'O', 'K' }; boost::asio::write(ClientSocket, boost::asio::buffer(Sent), Ec); REQUIRE(!Ec); - std::array Received {}; - const auto ReadEc = ReadSocketWithTimeout(ServerSocket, Received.data(), Received.size(), std::chrono::milliseconds(200)); + std::array Received { }; + const auto ReadEc = ReadSocketWithTimeout(TimerThread.IoCtx(), ServerSocket, Received.data(), Received.size(), std::chrono::milliseconds(200)); CHECK(!ReadEc); CHECK(Received == Sent); } TEST_CASE("ReadSocketWithTimeout reads large payload") { - io_context IoCtx; + TIoPollThread TimerThread; boost::system::error_code Ec; - ip::tcp::socket ClientSocket(IoCtx); - ip::tcp::socket ServerSocket(IoCtx); - OpenLoopbackSocketPair(IoCtx, ClientSocket, ServerSocket, Ec); + ip::tcp::socket ClientSocket(TimerThread.IoCtx()); + ip::tcp::socket ServerSocket(TimerThread.IoCtx()); + OpenLoopbackSocketPair(TimerThread.IoCtx(), ClientSocket, ServerSocket, Ec); REQUIRE(!Ec); constexpr size_t PacketSize = 2 * 1024 * 1024; @@ -859,27 +877,27 @@ TEST_CASE("ReadSocketWithTimeout reads large payload") { boost::asio::write(ClientSocket, boost::asio::buffer(Sent), Ec); REQUIRE(!Ec); std::vector Received(PacketSize); - const auto ReadEc = ReadSocketWithTimeout(ServerSocket, Received.data(), Received.size(), std::chrono::seconds(2)); + const auto ReadEc = ReadSocketWithTimeout(TimerThread.IoCtx(), ServerSocket, Received.data(), Received.size(), std::chrono::seconds(2)); CHECK(!ReadEc); CHECK(Received == Sent); } TEST_CASE("ReadSocketWithTimeout can timeout then retry successfully") { - io_context IoCtx; + TIoPollThread TimerThread; boost::system::error_code Ec; - ip::tcp::socket ClientSocket(IoCtx); - ip::tcp::socket ServerSocket(IoCtx); - OpenLoopbackSocketPair(IoCtx, ClientSocket, ServerSocket, Ec); + ip::tcp::socket ClientSocket(TimerThread.IoCtx()); + ip::tcp::socket ServerSocket(TimerThread.IoCtx()); + OpenLoopbackSocketPair(TimerThread.IoCtx(), ClientSocket, ServerSocket, Ec); REQUIRE(!Ec); uint8_t Received = 0; - CHECK(ReadSocketWithTimeout(ServerSocket, &Received, 1, std::chrono::milliseconds(20)) == error::timed_out); + CHECK(ReadSocketWithTimeout(TimerThread.get(), ServerSocket, &Received, 1, std::chrono::milliseconds(20)) == error::timed_out); const uint8_t Sent = 0x42; boost::asio::write(ClientSocket, boost::asio::buffer(&Sent, 1), Ec); REQUIRE(!Ec); - const auto ReadEc = ReadSocketWithTimeout(ServerSocket, &Received, 1, std::chrono::milliseconds(200)); + const auto ReadEc = ReadSocketWithTimeout(TimerThread.IoCtx(), ServerSocket, &Received, 1, std::chrono::milliseconds(200)); CHECK(!ReadEc); CHECK(Received == Sent); @@ -1047,9 +1065,9 @@ void TNetwork::SendFile(TClient& c, const std::string& UnsafeName) { #if defined(BEAMMP_LINUX) #include #include +#include #include #include -#include #endif void TNetwork::SendFileToClient(TClient& c, size_t Size, const std::string& Name) { TScopedTimer timer(fmt::format("Download of '{}' for client {}", Name, c.GetID())); @@ -1275,3 +1293,21 @@ std::vector TNetwork::UDPRcvFromClient(boost::asio::ip::udp::endpoint& beammp_assert(Rcv <= Ret.size()); return std::vector(Ret.begin(), Ret.begin() + Rcv); } + +TIoPollThread::TIoPollThread() + : mWorkGuard(boost::asio::make_work_guard(mIoCtx)) + , mThread([this](std::stop_token StopToken) { + while (!StopToken.stop_requested()) { + try { + mIoCtx.run(); + break; + } catch (...) { + mIoCtx.restart(); + } + } + }) { } + +TIoPollThread::~TIoPollThread() { + mWorkGuard.reset(); + mIoCtx.stop(); +}