diff --git a/render_2D/render_2D/axis/Axis_State_Strategy.hpp b/render_2D/render_2D/axis/Axis_State_Strategy.hpp index 194142f..01bb587 100644 --- a/render_2D/render_2D/axis/Axis_State_Strategy.hpp +++ b/render_2D/render_2D/axis/Axis_State_Strategy.hpp @@ -284,6 +284,7 @@ public: }); } protected: + using Render_Base::Render_Base; template Axis_State_Strategy(With_Attached_Impl implementation, Args&&... args) diff --git a/web_server/app/Asio_WebSocket_Scene_Loop.h b/web_server/app/Asio_WebSocket_Scene_Loop.h new file mode 100644 index 0000000..57760ce --- /dev/null +++ b/web_server/app/Asio_WebSocket_Scene_Loop.h @@ -0,0 +1,82 @@ +#pragma once +#include "Web_Event.h" +#include "Web_Plot_Session.h" +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +namespace renderive::web { +template +class Asio_WebSocket_Scene_Loop final : public std::enable_shared_from_this> { +public: + enum class Submit_Result { + none, + closed, + queue_full + }; + using Response_Handler = std::function; + using Failure_Handler = std::function; + [[nodiscard]] static std::shared_ptr create(asio::any_io_executor executor, Response_Handler response_handler, Failure_Handler failure_handler) { + auto loop = std::shared_ptr(new Asio_WebSocket_Scene_Loop(std::move(executor), std::move(response_handler), std::move(failure_handler))); + loop->start(); + return loop; + } + [[nodiscard]] Submit_Result submit(Web_Event event) { + if (stopped_.load(std::memory_order_acquire)) + return Submit_Result::closed; + if (events_.try_send(asio::error_code{}, std::move(event))) + return Submit_Result::none; + return stopped_.load(std::memory_order_acquire) ? Submit_Result::closed : Submit_Result::queue_full; + } + void stop() { + if (stopped_.exchange(true, std::memory_order_acq_rel)) + return; + events_.reset(); + events_.close(); + } +private: + static constexpr std::size_t queue_capacity = 128; + Asio_WebSocket_Scene_Loop(asio::any_io_executor executor, Response_Handler response_handler, Failure_Handler failure_handler) + : events_(std::move(executor), queue_capacity), response_handler_(std::move(response_handler)), failure_handler_(std::move(failure_handler)) {} + void start() { + auto self = this->shared_from_this(); + asio::co_spawn(events_.get_executor(), [self]() -> asio::awaitable { + co_await self->run(); + }, [self](std::exception_ptr exception) { + if (exception) + self->failure_handler_(exception); + }); + } + asio::awaitable run() { + Scene_Session session; + for (;;) { + asio::error_code error; + auto event = co_await events_.async_receive(asio::redirect_error(asio::use_awaitable, error)); + if (error) { + if (stopped_.load(std::memory_order_acquire)) + co_return; + throw asio::system_error(error); + } + auto response = session.handle(event); + if (stopped_.load(std::memory_order_acquire)) + co_return; + if (response) + response_handler_(std::move(*response)); + } + } + asio::experimental::concurrent_channel events_; + std::atomic_bool stopped_{}; + Response_Handler response_handler_; + Failure_Handler failure_handler_; +}; +} diff --git a/web_server/app/Gallery_Plot_Session.cpp b/web_server/app/Gallery_Plot_Session.cpp index a89e78b..928c7e2 100644 --- a/web_server/app/Gallery_Plot_Session.cpp +++ b/web_server/app/Gallery_Plot_Session.cpp @@ -8,7 +8,6 @@ #include #include #include -#include #include #include #include @@ -185,11 +184,7 @@ struct Gallery_Plot_Session::Impl { scene->set_client_metrics(performance); return true; } - [[nodiscard]] std::unique_lock acquire_foreground_lock() { - return std::unique_lock(mutex); - } std::unique_ptr scene; - std::mutex mutex; std::uint64_t session_id{}; std::optional handle_gallery(const Gallery_Request& request) { if (request.kind == Gallery_Request_Kind::Catalog) @@ -302,18 +297,13 @@ struct Gallery_Plot_Session::Impl { } std::optional handle_frame_request() { const auto request_started = std::chrono::steady_clock::now(); - std::optional pixels; - { - auto lock = acquire_foreground_lock(); - if (!scene || !scene->request_frame()) - return std::nullopt; - const auto encode_started = std::chrono::steady_clock::now(); - pixels = scene->encode_latest_pixels(); - const auto encode_finished = std::chrono::steady_clock::now(); - if (pixels) - scene->record_pixel_response(request_started, encode_started, encode_finished, - pixels->size()); - } + if (!scene || !scene->request_frame()) + return std::nullopt; + const auto encode_started = std::chrono::steady_clock::now(); + auto pixels = scene->encode_latest_pixels(); + const auto encode_finished = std::chrono::steady_clock::now(); + if (pixels) + scene->record_pixel_response(request_started, encode_started, encode_finished, pixels->size()); if (!pixels) return std::nullopt; return Web_Response{Web_Response_Type::Pixels, std::move(*pixels)}; @@ -321,7 +311,6 @@ struct Gallery_Plot_Session::Impl { std::optional handle(const Web_Event& event) { if (std::holds_alternative(event)) return handle_frame_request(); - auto lock = acquire_foreground_lock(); return std::visit( [this](const auto& value) -> std::optional { using T = std::decay_t; @@ -348,7 +337,7 @@ struct Gallery_Plot_Session::Impl { event); } }; -Gallery_Plot_Session::Gallery_Plot_Session() : impl_(std::make_shared()) {} +Gallery_Plot_Session::Gallery_Plot_Session() : impl_(std::make_unique()) {} Gallery_Plot_Session::~Gallery_Plot_Session() = default; std::optional Gallery_Plot_Session::handle(const Web_Event& event) { return impl_->handle(event); diff --git a/web_server/app/Gallery_Plot_Session.h b/web_server/app/Gallery_Plot_Session.h index bb38961..11cc377 100644 --- a/web_server/app/Gallery_Plot_Session.h +++ b/web_server/app/Gallery_Plot_Session.h @@ -18,7 +18,7 @@ public: private: struct Impl; - std::shared_ptr impl_; + std::unique_ptr impl_; }; } // namespace renderive::web diff --git a/web_server/app/Gallery_WebSocket_Controller.cpp b/web_server/app/Gallery_WebSocket_Controller.cpp index 1db2aa2..9d9c101 100644 --- a/web_server/app/Gallery_WebSocket_Controller.cpp +++ b/web_server/app/Gallery_WebSocket_Controller.cpp @@ -1,78 +1,68 @@ #include "Gallery_WebSocket_Controller.h" - +#include "Asio_WebSocket_Scene_Loop.h" #include "Gallery_Plot_Session.h" #include "Web_Event_Adapter.h" #include - #include - -#include +#include #include - +#include namespace renderive::web { - -void Gallery_WebSocket_Controller::handleNewConnection( - const drogon::HttpRequestPtr&, - const drogon::WebSocketConnectionPtr& connection) { - connection->setContext(std::make_shared()); - connection->setPingMessage("renderive-gallery", std::chrono::seconds(20)); - LOG_INFO << "Renderive Gallery WebSocket connected: " - << connection->peerAddr().toIpPort(); -} - -void Gallery_WebSocket_Controller::handleNewMessage( - const drogon::WebSocketConnectionPtr& connection, - std::string&& message, - const drogon::WebSocketMessageType& type) { - if (type == drogon::WebSocketMessageType::Ping || - type == drogon::WebSocketMessageType::Pong || - type == drogon::WebSocketMessageType::Close) { +namespace { +using Scene_Loop = Asio_WebSocket_Scene_Loop; +void send_response(const std::weak_ptr& weak_connection, Web_Response response) { + const auto connection = weak_connection.lock(); + if (!connection || !connection->connected()) + return; + const auto message_type = response.type == Web_Response_Type::Pixels ? drogon::WebSocketMessageType::Binary : drogon::WebSocketMessageType::Text; + connection->send(response.payload.data(), response.payload.size(), message_type); +} +void fail_session(const std::weak_ptr& weak_connection, std::exception_ptr exception) { + static_cast(::renderive::error::capture("handling Gallery WebSocket scene loop", exception)); + LOG_ERROR << "Renderive Gallery WebSocket scene loop failed"; + if (const auto connection = weak_connection.lock(); connection && connection->connected()) + connection->shutdown(drogon::CloseCode::kUnexpectedCondition, "Renderive Gallery rendering failed"); +} +} +Gallery_WebSocket_Controller::Gallery_WebSocket_Controller(std::shared_ptr scene_pool) : scene_pool_(std::move(scene_pool)) {} +void Gallery_WebSocket_Controller::handleNewConnection(const drogon::HttpRequestPtr&, const drogon::WebSocketConnectionPtr& connection) { + const std::weak_ptr weak_connection = connection; + connection->setContext(Scene_Loop::create(scene_pool_->get_executor(), [weak_connection](Web_Response response) { + send_response(weak_connection, std::move(response)); + }, [weak_connection](std::exception_ptr exception) { + fail_session(weak_connection, exception); + })); + connection->setPingMessage("renderive-gallery", std::chrono::seconds(20)); + LOG_INFO << "Renderive Gallery WebSocket connected: " << connection->peerAddr().toIpPort(); +} +void Gallery_WebSocket_Controller::handleNewMessage(const drogon::WebSocketConnectionPtr& connection, std::string&& message, const drogon::WebSocketMessageType& type) { + if (type == drogon::WebSocketMessageType::Ping || type == drogon::WebSocketMessageType::Pong || type == drogon::WebSocketMessageType::Close) return; - } if (type != drogon::WebSocketMessageType::Text) { - connection->shutdown(drogon::CloseCode::kInvalidMessage, - "Renderive Gallery accepts text events only"); + connection->shutdown(drogon::CloseCode::kInvalidMessage, "Renderive Gallery accepts text events only"); return; } if (message.size() > 64 * 1024) { - connection->shutdown(drogon::CloseCode::kMessageTooBig, - "Renderive Gallery event is too large"); + connection->shutdown(drogon::CloseCode::kMessageTooBig, "Renderive Gallery event is too large"); return; } const auto event = Web_Event_Adapter::decode(message); if (!event) { - connection->shutdown(drogon::CloseCode::kWrongMessageContent, - "Invalid Renderive Gallery event"); + connection->shutdown(drogon::CloseCode::kWrongMessageContent, "Invalid Renderive Gallery event"); return; } - const auto session = connection->getContext(); + const auto session = connection->getContext(); if (!session) { - connection->shutdown(drogon::CloseCode::kUnexpectedCondition, - "Renderive Gallery session is unavailable"); + connection->shutdown(drogon::CloseCode::kUnexpectedCondition, "Renderive Gallery session is unavailable"); return; } - try { - if (auto response = session->handle(*event)) { - const auto message_type = response->type == Web_Response_Type::Pixels - ? drogon::WebSocketMessageType::Binary - : drogon::WebSocketMessageType::Text; - connection->send(response->payload.data(), response->payload.size(), - message_type); - } - } catch (...) { - static_cast(::renderive::error::capture( - "handling Gallery WebSocket request", std::current_exception())); - LOG_ERROR << "Renderive Gallery session failed"; - connection->shutdown(drogon::CloseCode::kUnexpectedCondition, - "Renderive Gallery rendering failed"); - } + if (session->submit(std::move(*event)) == Scene_Loop::Submit_Result::queue_full) + connection->shutdown(drogon::CloseCode::kViolation, "Renderive Gallery scene queue is full"); } - -void Gallery_WebSocket_Controller::handleConnectionClosed( - const drogon::WebSocketConnectionPtr& connection) { - LOG_INFO << "Renderive Gallery WebSocket closed: " - << connection->peerAddr().toIpPort(); +void Gallery_WebSocket_Controller::handleConnectionClosed(const drogon::WebSocketConnectionPtr& connection) { + LOG_INFO << "Renderive Gallery WebSocket closed: " << connection->peerAddr().toIpPort(); + if (const auto session = connection->getContext()) + session->stop(); connection->clearContext(); } - -} // namespace renderive::web +} diff --git a/web_server/app/Gallery_WebSocket_Controller.h b/web_server/app/Gallery_WebSocket_Controller.h index edf2658..a01c53d 100644 --- a/web_server/app/Gallery_WebSocket_Controller.h +++ b/web_server/app/Gallery_WebSocket_Controller.h @@ -1,14 +1,18 @@ #pragma once +#include #include +#include namespace renderive::web { -class Gallery_WebSocket_Controller final -: public drogon::WebSocketController { +class Gallery_WebSocket_Controller final : public drogon::WebSocketController { public: + explicit Gallery_WebSocket_Controller(std::shared_ptr scene_pool); void handleNewMessage(const drogon::WebSocketConnectionPtr& connection, std::string&& message, const drogon::WebSocketMessageType& type) override; void handleNewConnection(const drogon::HttpRequestPtr& request, const drogon::WebSocketConnectionPtr& connection) override; void handleConnectionClosed(const drogon::WebSocketConnectionPtr& connection) override; WS_PATH_LIST_BEGIN - WS_PATH_ADD("/renderive/gallery"); + WS_PATH_ADD("/renderive/gallery"); WS_PATH_LIST_END +private: + std::shared_ptr scene_pool_; }; -} // namespace renderive::web +} diff --git a/web_server/app/Renderive_WebSocket_Controller.cpp b/web_server/app/Renderive_WebSocket_Controller.cpp index 6f78d22..76ab33d 100644 --- a/web_server/app/Renderive_WebSocket_Controller.cpp +++ b/web_server/app/Renderive_WebSocket_Controller.cpp @@ -1,76 +1,68 @@ #include "Renderive_WebSocket_Controller.h" - +#include "Asio_WebSocket_Scene_Loop.h" #include "Web_Event_Adapter.h" #include "Web_Plot_Session.h" #include - #include - -#include +#include #include - +#include namespace renderive::web { - -void Renderive_WebSocket_Controller::handleNewConnection( - const drogon::HttpRequestPtr&, - const drogon::WebSocketConnectionPtr& connection) { - connection->setContext(std::make_shared()); +namespace { +using Scene_Loop = Asio_WebSocket_Scene_Loop; +void send_response(const std::weak_ptr& weak_connection, Web_Response response) { + const auto connection = weak_connection.lock(); + if (!connection || !connection->connected()) + return; + const auto message_type = response.type == Web_Response_Type::Pixels ? drogon::WebSocketMessageType::Binary : drogon::WebSocketMessageType::Text; + connection->send(response.payload.data(), response.payload.size(), message_type); +} +void fail_session(const std::weak_ptr& weak_connection, std::exception_ptr exception) { + static_cast(::renderive::error::capture("handling Renderive WebSocket scene loop", exception)); + LOG_ERROR << "Renderive WebSocket scene loop failed"; + if (const auto connection = weak_connection.lock(); connection && connection->connected()) + connection->shutdown(drogon::CloseCode::kUnexpectedCondition, "Renderive rendering failed"); +} +} +Renderive_WebSocket_Controller::Renderive_WebSocket_Controller(std::shared_ptr scene_pool) : scene_pool_(std::move(scene_pool)) {} +void Renderive_WebSocket_Controller::handleNewConnection(const drogon::HttpRequestPtr&, const drogon::WebSocketConnectionPtr& connection) { + const std::weak_ptr weak_connection = connection; + connection->setContext(Scene_Loop::create(scene_pool_->get_executor(), [weak_connection](Web_Response response) { + send_response(weak_connection, std::move(response)); + }, [weak_connection](std::exception_ptr exception) { + fail_session(weak_connection, exception); + })); connection->setPingMessage("renderive", std::chrono::seconds(20)); LOG_INFO << "Renderive WebSocket connected: " << connection->peerAddr().toIpPort(); } - -void Renderive_WebSocket_Controller::handleNewMessage( - const drogon::WebSocketConnectionPtr& connection, - std::string&& message, - const drogon::WebSocketMessageType& type) { - if (type == drogon::WebSocketMessageType::Ping || - type == drogon::WebSocketMessageType::Pong || - type == drogon::WebSocketMessageType::Close) { +void Renderive_WebSocket_Controller::handleNewMessage(const drogon::WebSocketConnectionPtr& connection, std::string&& message, const drogon::WebSocketMessageType& type) { + if (type == drogon::WebSocketMessageType::Ping || type == drogon::WebSocketMessageType::Pong || type == drogon::WebSocketMessageType::Close) return; - } if (type != drogon::WebSocketMessageType::Text) { - connection->shutdown(drogon::CloseCode::kInvalidMessage, - "Renderive accepts text events only"); + connection->shutdown(drogon::CloseCode::kInvalidMessage, "Renderive accepts text events only"); return; } if (message.size() > 16 * 1024) { - connection->shutdown(drogon::CloseCode::kMessageTooBig, - "Renderive event is too large"); + connection->shutdown(drogon::CloseCode::kMessageTooBig, "Renderive event is too large"); return; } const auto event = Web_Event_Adapter::decode(message); if (!event) { - connection->shutdown(drogon::CloseCode::kWrongMessageContent, - "Invalid Renderive event"); + connection->shutdown(drogon::CloseCode::kWrongMessageContent, "Invalid Renderive event"); return; } - const auto session = connection->getContext(); + const auto session = connection->getContext(); if (!session) { - connection->shutdown(drogon::CloseCode::kUnexpectedCondition, - "Renderive session is unavailable"); + connection->shutdown(drogon::CloseCode::kUnexpectedCondition, "Renderive session is unavailable"); return; } - try { - if (auto response = session->handle(*event)) { - const auto message_type = response->type == Web_Response_Type::Pixels - ? drogon::WebSocketMessageType::Binary - : drogon::WebSocketMessageType::Text; - connection->send(response->payload.data(), response->payload.size(), - message_type); - } - } catch (...) { - static_cast(::renderive::error::capture( - "handling Renderive WebSocket request", std::current_exception())); - LOG_ERROR << "Renderive WebSocket session failed"; - connection->shutdown(drogon::CloseCode::kUnexpectedCondition, - "Renderive rendering failed"); - } + if (session->submit(std::move(*event)) == Scene_Loop::Submit_Result::queue_full) + connection->shutdown(drogon::CloseCode::kViolation, "Renderive scene queue is full"); } - -void Renderive_WebSocket_Controller::handleConnectionClosed( - const drogon::WebSocketConnectionPtr& connection) { +void Renderive_WebSocket_Controller::handleConnectionClosed(const drogon::WebSocketConnectionPtr& connection) { LOG_INFO << "Renderive WebSocket closed: " << connection->peerAddr().toIpPort(); + if (const auto session = connection->getContext()) + session->stop(); connection->clearContext(); } - -} // namespace renderive::web +} diff --git a/web_server/app/Renderive_WebSocket_Controller.h b/web_server/app/Renderive_WebSocket_Controller.h index 68dad1a..c1ed8ba 100644 --- a/web_server/app/Renderive_WebSocket_Controller.h +++ b/web_server/app/Renderive_WebSocket_Controller.h @@ -1,22 +1,18 @@ #pragma once - +#include #include - +#include namespace renderive::web { - -class Renderive_WebSocket_Controller final - : public drogon::WebSocketController { +class Renderive_WebSocket_Controller final : public drogon::WebSocketController { public: - void handleNewMessage(const drogon::WebSocketConnectionPtr& connection, - std::string&& message, - const drogon::WebSocketMessageType& type) override; - void handleNewConnection(const drogon::HttpRequestPtr& request, - const drogon::WebSocketConnectionPtr& connection) override; + explicit Renderive_WebSocket_Controller(std::shared_ptr scene_pool); + void handleNewMessage(const drogon::WebSocketConnectionPtr& connection, std::string&& message, const drogon::WebSocketMessageType& type) override; + void handleNewConnection(const drogon::HttpRequestPtr& request, const drogon::WebSocketConnectionPtr& connection) override; void handleConnectionClosed(const drogon::WebSocketConnectionPtr& connection) override; - WS_PATH_LIST_BEGIN WS_PATH_ADD("/renderive"); WS_PATH_LIST_END +private: + std::shared_ptr scene_pool_; }; - -} // namespace renderive::web +} diff --git a/web_server/app/Web_Server.cpp b/web_server/app/Web_Server.cpp index f675a86..a2f43ef 100644 --- a/web_server/app/Web_Server.cpp +++ b/web_server/app/Web_Server.cpp @@ -2,6 +2,7 @@ #include "Gallery_WebSocket_Controller.h" #include "Renderive_WebSocket_Controller.h" #include "Web_Performance_Log.h" +#include #include #include #include @@ -18,9 +19,10 @@ int run_web_server(std::uint16_t port, const std::filesystem::path& asset_root) return 3; } initialize_web_performance_log(asset_root / "logs"); - const auto controller = std::make_shared(); - const auto gallery_controller = std::make_shared(); const auto hardware_threads = std::max(2U, std::thread::hardware_concurrency()); + auto scene_pool = std::make_shared(std::min(8U, hardware_threads)); + const auto controller = std::make_shared(scene_pool); + const auto gallery_controller = std::make_shared(scene_pool); std::cout << "Renderive WebSocket backend: ws://127.0.0.1:" << port << "/renderive\n" << "Renderive control gallery: ws://127.0.0.1:" << port @@ -46,6 +48,8 @@ int run_web_server(std::uint16_t port, const std::filesystem::path& asset_root) .setThreadNum(std::min(8U, hardware_threads)) .setIdleConnectionTimeout(90) .run(); + scene_pool->stop(); + scene_pool->join(); return 0; } } // namespace renderive::web diff --git a/web_server/third_party/asio/main.cmake b/web_server/third_party/asio/main.cmake index c525cae..4209690 100644 --- a/web_server/third_party/asio/main.cmake +++ b/web_server/third_party/asio/main.cmake @@ -13,6 +13,9 @@ target_compile_definitions(asio_interface INTERFACE ASIO_NO_TS_EXECUTORS ASIO_NO_DYNAMIC_BUFFER_V1 ) +if (WIN32) + target_compile_definitions(asio_interface INTERFACE _WIN32_WINNT=0x0602) +endif() # 这个 Linux kernel header 路径不要在 Windows 下加 if (UNIX AND NOT APPLE) @@ -40,6 +43,7 @@ endif() add_library(asio_object STATIC ${asio_srcs} ) +target_compile_features(asio_object PRIVATE cxx_std_20) target_link_libraries(asio_object PUBLIC asio_interface