TLA  Line data    Source code

       1                 : //
       2                 : // Copyright (c) 2026 Steve Gerbino
       3                 : //
       4                 : // Distributed under the Boost Software License, Version 1.0. (See accompanying
       5                 : // file LICENSE_1_0.txt or copy at http://www.boost.org/LICENSE_1_0.txt)
       6                 : //
       7                 : // Official repository: https://github.com/cppalliance/corosio
       8                 : //
       9                 : 
      10                 : #ifndef BOOST_COROSIO_NATIVE_DETAIL_REACTOR_REACTOR_SOCKET_SERVICE_HPP
      11                 : #define BOOST_COROSIO_NATIVE_DETAIL_REACTOR_REACTOR_SOCKET_SERVICE_HPP
      12                 : 
      13                 : #include <boost/corosio/io/io_object.hpp>
      14                 : #include <boost/corosio/detail/scheduler_op.hpp>
      15                 : #include <boost/corosio/native/detail/reactor/reactor_service_state.hpp>
      16                 : #include <boost/capy/ex/execution_context.hpp>
      17                 : 
      18                 : #include <memory>
      19                 : #include <mutex>
      20                 : 
      21                 : namespace boost::corosio::detail {
      22                 : 
      23                 : /** CRTP base for reactor-backed socket/datagram service implementations.
      24                 : 
      25                 :     Provides the shared construct/destroy/shutdown/close/post/work
      26                 :     logic that is identical across all reactor backends and socket
      27                 :     types. Derived classes add only protocol-specific open/bind.
      28                 : 
      29                 :     @tparam Derived     The concrete service type (CRTP).
      30                 :     @tparam ServiceBase The abstract service base (tcp_service
      31                 :                         or udp_service).
      32                 :     @tparam Scheduler   The backend's scheduler type.
      33                 :     @tparam Impl        The backend's socket/datagram impl type.
      34                 : */
      35                 : template<class Derived, class ServiceBase, class Scheduler, class Impl>
      36                 : class reactor_socket_service : public ServiceBase
      37                 : {
      38                 :     friend Derived;
      39                 :     using state_type = reactor_service_state<Scheduler, Impl>;
      40                 : 
      41 HIT         932 :     explicit reactor_socket_service(capy::execution_context& ctx)
      42             932 :         : state_(
      43                 :               std::make_unique<state_type>(
      44             932 :                   ctx.template use_service<Scheduler>()))
      45                 :     {
      46             932 :     }
      47                 : 
      48                 : public:
      49             932 :     ~reactor_socket_service() override = default;
      50                 : 
      51             932 :     void shutdown() override
      52                 :     {
      53             932 :         std::lock_guard lock(state_->mutex_);
      54                 : 
      55             932 :         while (auto* impl = state_->impl_list_.pop_front())
      56                 :         {
      57 MIS           0 :             static_cast<Derived*>(this)->pre_shutdown(impl);
      58               0 :             impl->close_socket();
      59                 :         }
      60                 : 
      61                 :         // Don't clear impl_ptrs_ here. The scheduler shuts down after us
      62                 :         // and drains completed_ops_, calling destroy() on each queued op.
      63                 :         // Letting ~state_ release the ptrs (during service destruction,
      64                 :         // after scheduler shutdown) keeps every impl alive until all ops
      65                 :         // have been drained.
      66 HIT         932 :     }
      67                 : 
      68           13621 :     io_object::implementation* construct() override
      69                 :     {
      70           13621 :         auto impl = std::make_shared<Impl>(static_cast<Derived&>(*this));
      71           13621 :         auto* raw = impl.get();
      72                 : 
      73                 :         {
      74           13621 :             std::lock_guard lock(state_->mutex_);
      75           13621 :             state_->impl_ptrs_.emplace(raw, std::move(impl));
      76           13621 :             state_->impl_list_.push_back(raw);
      77           13621 :         }
      78                 : 
      79           13621 :         return raw;
      80           13621 :     }
      81                 : 
      82           13621 :     void destroy(io_object::implementation* impl) override
      83                 :     {
      84           13621 :         auto* typed = static_cast<Impl*>(impl);
      85           13621 :         static_cast<Derived*>(this)->pre_destroy(typed);
      86           13621 :         typed->close_socket();
      87           13621 :         std::lock_guard lock(state_->mutex_);
      88           13621 :         state_->impl_list_.remove(typed);
      89           13621 :         state_->impl_ptrs_.erase(typed);
      90           13621 :     }
      91                 : 
      92           22699 :     void close(io_object::handle& h) override
      93                 :     {
      94           22699 :         static_cast<Impl*>(h.get())->close_socket();
      95           22699 :     }
      96                 : 
      97          484656 :     Scheduler& scheduler() const noexcept
      98                 :     {
      99          484656 :         return state_->sched_;
     100                 :     }
     101                 : 
     102           88791 :     void post(scheduler_op* op)
     103                 :     {
     104           88791 :         state_->sched_.post(op);
     105           88791 :     }
     106                 : 
     107            4887 :     void work_started() noexcept
     108                 :     {
     109            4887 :         state_->sched_.work_started();
     110            4887 :     }
     111                 : 
     112             298 :     void work_finished() noexcept
     113                 :     {
     114             298 :         state_->sched_.work_finished();
     115             298 :     }
     116                 : 
     117                 : protected:
     118                 :     // Override in derived to add pre-close logic (e.g. kqueue linger reset)
     119 MIS           0 :     void pre_shutdown(Impl*) noexcept {}
     120 HIT       13621 :     void pre_destroy(Impl*) noexcept {}
     121                 : 
     122                 :     std::unique_ptr<state_type> state_;
     123                 : 
     124                 : private:
     125                 :     reactor_socket_service(reactor_socket_service const&)            = delete;
     126                 :     reactor_socket_service& operator=(reactor_socket_service const&) = delete;
     127                 : };
     128                 : 
     129                 : } // namespace boost::corosio::detail
     130                 : 
     131                 : #endif // BOOST_COROSIO_NATIVE_DETAIL_REACTOR_REACTOR_SOCKET_SERVICE_HPP