后台服务协程化

render_2D/render_2D/axis/Axis_State_Strategy.hpp

@@ -284,6 +284,7 @@ public:
         });
     }
 protected:
+    using Render_Base::Render_Base;
     template <class Implementation, class State, class... Args>
     Axis_State_Strategy(With_Attached_Impl<Implementation, State> implementation,
                         Args&&... args)

web_server/app/Asio_WebSocket_Scene_Loop.h

@@ -0,0 +1,82 @@
+#pragma once
+#include "Web_Event.h"
+#include "Web_Plot_Session.h"
+#include <asio/any_io_executor.hpp>
+#include <asio/awaitable.hpp>
+#include <asio/co_spawn.hpp>
+#include <asio/error_code.hpp>
+#include <asio/experimental/concurrent_channel.hpp>
+#include <asio/redirect_error.hpp>
+#include <asio/system_error.hpp>
+#include <asio/use_awaitable.hpp>
+#include <atomic>
+#include <cstddef>
+#include <exception>
+#include <functional>
+#include <memory>
+#include <utility>
+namespace renderive::web {
+template <class Scene_Session>
+class Asio_WebSocket_Scene_Loop final : public std::enable_shared_from_this<Asio_WebSocket_Scene_Loop<Scene_Session>> {
+public:
+    enum class Submit_Result {
+        none,
+        closed,
+        queue_full
+    };
+    using Response_Handler = std::function<void(Web_Response)>;
+    using Failure_Handler = std::function<void(std::exception_ptr)>;
+    [[nodiscard]] static std::shared_ptr<Asio_WebSocket_Scene_Loop> create(asio::any_io_executor executor, Response_Handler response_handler, Failure_Handler failure_handler) {
+        auto loop = std::shared_ptr<Asio_WebSocket_Scene_Loop>(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<void> {
+            co_await self->run();
+        }, [self](std::exception_ptr exception) {
+            if (exception)
+                self->failure_handler_(exception);
+        });
+    }
+    asio::awaitable<void> 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<void(asio::error_code, Web_Event)> events_;
+    std::atomic_bool stopped_{};
+    Response_Handler response_handler_;
+    Failure_Handler failure_handler_;
+};
+}

web_server/app/Gallery_Plot_Session.cpp

@@ -8,7 +8,6 @@
 #include <chrono>
 #include <cmath>
 #include <cstdint>
-#include <mutex>
 #include <optional>
 #include <string>
 #include <string_view>
@@ -185,11 +184,7 @@ struct Gallery_Plot_Session::Impl {
         scene->set_client_metrics(performance);
         return true;
     }
-    [[nodiscard]] std::unique_lock<std::mutex> acquire_foreground_lock() {
-        return std::unique_lock<std::mutex>(mutex);
-    }
     std::unique_ptr<Gallery_Scene_Interface> scene;
-    std::mutex mutex;
     std::uint64_t session_id{};
     std::optional<Web_Response> handle_gallery(const Gallery_Request& request) {
         if (request.kind == Gallery_Request_Kind::Catalog)
@@ -302,18 +297,13 @@ struct Gallery_Plot_Session::Impl {
     }
     std::optional<Web_Response> handle_frame_request() {
         const auto request_started = std::chrono::steady_clock::now();
-        std::optional<std::string> 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<Web_Response> handle(const Web_Event& event) {
         if (std::holds_alternative<Frame_Request>(event))
             return handle_frame_request();
-        auto lock = acquire_foreground_lock();
         return std::visit(
             [this](const auto& value) -> std::optional<Web_Response> {
                 using T = std::decay_t<decltype(value)>;
@@ -348,7 +337,7 @@ struct Gallery_Plot_Session::Impl {
             event);
     }
 };
-Gallery_Plot_Session::Gallery_Plot_Session() : impl_(std::make_shared<Impl>()) {}
+Gallery_Plot_Session::Gallery_Plot_Session() : impl_(std::make_unique<Impl>()) {}
 Gallery_Plot_Session::~Gallery_Plot_Session() = default;
 std::optional<Web_Response> Gallery_Plot_Session::handle(const Web_Event& event) {
     return impl_->handle(event);

web_server/app/Gallery_Plot_Session.h

@@ -18,7 +18,7 @@ public:
 
 private:
     struct Impl;
-    std::shared_ptr<Impl> impl_;
+    std::unique_ptr<Impl> impl_;
 };
 
 } // namespace renderive::web

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 <renderive/error/Error_Policy.hpp>
-
 #include <trantor/utils/Logger.h>
-
-#include <exception>
+#include <chrono>
 #include <memory>
-
+#include <utility>
 namespace renderive::web {
-
-void Gallery_WebSocket_Controller::handleNewConnection(
-    const drogon::HttpRequestPtr&,
-    const drogon::WebSocketConnectionPtr& connection) {
-    connection->setContext(std::make_shared<Gallery_Plot_Session>());
-    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<Gallery_Plot_Session>;
+void send_response(const std::weak_ptr<drogon::WebSocketConnection>& 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<drogon::WebSocketConnection>& weak_connection, std::exception_ptr exception) {
+    static_cast<void>(::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<asio::thread_pool> scene_pool) : scene_pool_(std::move(scene_pool)) {}
+void Gallery_WebSocket_Controller::handleNewConnection(const drogon::HttpRequestPtr&, const drogon::WebSocketConnectionPtr& connection) {
+    const std::weak_ptr<drogon::WebSocketConnection> 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<Gallery_Plot_Session>();
+    const auto session = connection->getContext<Scene_Loop>();
     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<void>(::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<Scene_Loop>())
+        session->stop();
     connection->clearContext();
 }
-
-} // namespace renderive::web
+}

web_server/app/Gallery_WebSocket_Controller.h

@@ -1,14 +1,18 @@
 #pragma once
+#include <asio/thread_pool.hpp>
 #include <drogon/WebSocketController.h>
+#include <memory>
 namespace renderive::web {
-class Gallery_WebSocket_Controller final
-: public drogon::WebSocketController<Gallery_WebSocket_Controller, false> {
+class Gallery_WebSocket_Controller final : public drogon::WebSocketController<Gallery_WebSocket_Controller, false> {
 public:
+    explicit Gallery_WebSocket_Controller(std::shared_ptr<asio::thread_pool> 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<asio::thread_pool> scene_pool_;
 };
-} // namespace renderive::web
+}

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 <renderive/error/Error_Policy.hpp>
-
 #include <trantor/utils/Logger.h>
-
-#include <exception>
+#include <chrono>
 #include <memory>
-
+#include <utility>
 namespace renderive::web {
-
-void Renderive_WebSocket_Controller::handleNewConnection(
-    const drogon::HttpRequestPtr&,
-    const drogon::WebSocketConnectionPtr& connection) {
-    connection->setContext(std::make_shared<Web_Plot_Session>());
+namespace {
+using Scene_Loop = Asio_WebSocket_Scene_Loop<Web_Plot_Session>;
+void send_response(const std::weak_ptr<drogon::WebSocketConnection>& 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<drogon::WebSocketConnection>& weak_connection, std::exception_ptr exception) {
+    static_cast<void>(::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<asio::thread_pool> scene_pool) : scene_pool_(std::move(scene_pool)) {}
+void Renderive_WebSocket_Controller::handleNewConnection(const drogon::HttpRequestPtr&, const drogon::WebSocketConnectionPtr& connection) {
+    const std::weak_ptr<drogon::WebSocketConnection> 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<Web_Plot_Session>();
+    const auto session = connection->getContext<Scene_Loop>();
     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<void>(::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<Scene_Loop>())
+        session->stop();
     connection->clearContext();
 }
-
-} // namespace renderive::web
+}

web_server/app/Renderive_WebSocket_Controller.h

@@ -1,22 +1,18 @@
 #pragma once
-
+#include <asio/thread_pool.hpp>
 #include <drogon/WebSocketController.h>
-
+#include <memory>
 namespace renderive::web {
-
-class Renderive_WebSocket_Controller final
-    : public drogon::WebSocketController<Renderive_WebSocket_Controller, false> {
+class Renderive_WebSocket_Controller final : public drogon::WebSocketController<Renderive_WebSocket_Controller, false> {
 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<asio::thread_pool> 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<asio::thread_pool> scene_pool_;
 };
-
-} // namespace renderive::web
+}

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 <asio/thread_pool.hpp>
 #include <drogon/drogon.h>
 #include <algorithm>
 #include <filesystem>
@@ -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<Renderive_WebSocket_Controller>();
-    const auto gallery_controller = std::make_shared<Gallery_WebSocket_Controller>();
     const auto hardware_threads = std::max(2U, std::thread::hardware_concurrency());
+    auto scene_pool = std::make_shared<asio::thread_pool>(std::min(8U, hardware_threads));
+    const auto controller = std::make_shared<Renderive_WebSocket_Controller>(scene_pool);
+    const auto gallery_controller = std::make_shared<Gallery_WebSocket_Controller>(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

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