connection_count_limiter.hpp

Go to the documentation of this file.

/*
 * RESTinio
 */
 
/*!
 * @file
 * @brief Stuff related to limits of active parallel connections.
 * @since v.0.6.12
 */
 
#pragma once
 
#include <restinio/null_mutex.hpp>
#include <restinio/default_strands.hpp>
 
#include <restinio/utils/tagged_scalar.hpp>
 
#include <any>
#include <cstdint>
#include <memory>
#include <mutex>
#include <utility>
 
namespace restinio
{
 
namespace connection_count_limits
{
 
//
// max_parallel_connections_t
//
struct max_parallel_connections_tag {};
 
/*!
 * @brief A kind of strict typedef for maximum count of active connections.
 *
 * @since v.0.6.12
 */
using max_parallel_connections_t = restinio::utils::tagged_scalar_t<
        std::size_t, max_parallel_connections_tag >;
 
//
// max_active_accepts_t
//
struct max_active_accepts_tag {};
 
/*!
 * @brief A kind of strict typedef for maximum count of active accepts.
 *
 * @since v.0.6.12
 */
using max_active_accepts_t = restinio::utils::tagged_scalar_t<
        std::size_t, max_active_accepts_tag >;
 
namespace impl
{
 
/*!
 * @brief An interface of acceptor to be used by connection count limiters.
 *
 * An instance of a connection count limiter will receive a reference to the
 * acceptor. The limiter has to call the acceptor and this interface declares
 * methods of the acceptor that will be invoked by the limiter.
 *
 * The assumed working scheme is:
 *
 * - the acceptor calls `accept_next` for the limiter;
 * - the limiter checks the possibility to call `accept()`. If it is possible,
 *   then the limiter calls `call_accept_now` back (right inside `accept_next`
 *   invocation). If it isn't possible, then the limiter stores the socket's
 *   slot index somewhere inside the limiter;
 * - sometime later the limiter calls `schedule_next_accept_attempt` for the
 *   acceptor. The acceptor then should perform a new call to `accept_next` in
 *   the appropriate worker context.
 *
 * @since v.0.6.12
 */
class acceptor_callback_iface_t
{
public:
    /*!
     * This method will be invoked by a limiter when there is a possibility to
     * call accept() right now.
     */
    virtual void
    call_accept_now(
        //! An index of socket's slot to be used for accept().
        std::size_t index ) noexcept = 0;
 
    /*!
     * This method will be invoked by a limiter when there is no possibility to
     * call accept() right now, but the next call to `accept_next` should be
     * scheduled as soon as possible in the appropriate worker context.
     * 
     * It is assumed that the acceptor will use asio::post() with a completion
     * handler that calls the `accept_next` method of the limiter.
     */
    virtual void
    schedule_next_accept_attempt(
        //! An index of socket's slot to be used for accept().
        std::size_t index ) noexcept = 0;
};
 
/*!
 * @brief Actual implementation of connection count limiter.
 *
 * @note
 * This is not Copyable nor Moveable type.
 *
 * @tparam Mutex_Type Type of mutex to be used for protection of limiter
 * object. It is expected to be std::mutex or null_mutex_t.
 *
 * @since v.0.6.12
 */
template< typename Mutex_Type >
class actual_limiter_t
{
    //! Lock object to be used.
    Mutex_Type m_lock;
 
    //! Mandatory pointer to the acceptor connected with this limiter.
    not_null_pointer_t< acceptor_callback_iface_t > m_acceptor;
 
    /*!
     * @brief The counter of active accept() operations.
     *
     * Incremented every time the acceptor_callback_iface_t::call_accept_now()
     * is invoked. Decremented in increment_parallel_connections().
     *
     * @attention
     * It seems to be a fragile scheme because if there won't be a call to
     * increment_parallel_connections() after the invocation of
     * acceptor_callback_iface_t::call_accept_now() the value of
     * m_active_accepts will be incorrect. But it is hard to invent a more
     * bulletproof solution and it seems that the missing call to
     * increment_parallel_connections() could be only on the shutdown of the
     * acceptor.
     */
    std::size_t m_active_accepts{ 0u };
 
    /*!
     * @brief The counter of active connections.
     *
     * This value is incremented in increment_parallel_connections()
     * and decremented in decrement_parallel_connections().
     */
    std::size_t m_connections{ 0u };
 
    //! The limit for parallel connections.
    const std::size_t m_max_parallel_connections;
 
    /*!
     * @brief The storage for holding pending socket's slots.
     *
     * @note
     * This storage is used as stack: new indexes are added to the
     * end and are got from the end of the vector (LIFO working scheme).
     *
     * @attention
     * The capacity for that storage is preallocated in the constructor
     * so we don't expect any allocations during the usage of
     * m_pending_indexes. This allows accept_next() method to be
     * noexcept. But this works only if max_pending_indexes passed
     * to the constructor is right.
     */
    std::vector< std::size_t > m_pending_indexes;
 
    [[nodiscard]]
    bool
    has_free_slots() const noexcept
    {
        return (m_active_accepts + m_connections) < m_max_parallel_connections;
    }
 
public:
    actual_limiter_t(
        not_null_pointer_t< acceptor_callback_iface_t > acceptor,
        max_parallel_connections_t max_parallel_connections,
        max_active_accepts_t max_pending_indexes )
        :   m_acceptor{ acceptor }
        ,   m_max_parallel_connections{ max_parallel_connections.value() }
    {
        m_pending_indexes.reserve( max_pending_indexes.value() );
    }
 
    actual_limiter_t( const actual_limiter_t & ) = delete;
    actual_limiter_t( actual_limiter_t && ) = delete;
 
    void
    increment_parallel_connections() noexcept
    {
        std::lock_guard< Mutex_Type > lock{ m_lock };
 
        // Expects that m_active_accepts is always greater than 0.
        --m_active_accepts;
 
        ++m_connections;
    }
 
    // Note: this method is noexcept because it can be called from
    // destructors.
    void
    decrement_parallel_connections() noexcept
    {
        // Decrement active connections under acquired lock.
        // If the count of connections drops below the limit and
        // there are some pending indexes then one of them will
        // be returned (wrapped into an optional).
        auto index_to_activate = [this]() -> std::optional<std::size_t> {
            std::lock_guard< Mutex_Type > lock{ m_lock };
 
            // Expects that m_connections is always greater than 0.
            --m_connections;
 
            if( has_free_slots() && !m_pending_indexes.empty() )
            {
                std::size_t pending_index = m_pending_indexes.back();
                m_pending_indexes.pop_back();
                return pending_index;
            }
            else
                return std::nullopt;
        }();
 
        if( index_to_activate )
        {
            m_acceptor->schedule_next_accept_attempt( *index_to_activate );
        }
    }
 
    /*!
     * This method either calls acceptor_callback_iface_t::call_accept_now() (in
     * that case m_active_accepts is incremented) or stores @a index into the
     * internal storage.
     */
    void
    accept_next( std::size_t index ) noexcept
    {
        // Perform all operations under acquired lock.
        // The result is a flag that tells can accept() be called right now.
        const bool accept_now = [this, index]() -> bool {
            std::lock_guard< Mutex_Type > lock{ m_lock };
 
            if( has_free_slots() )
            {
                ++m_active_accepts;
                return true;
            }
            else
            {
                m_pending_indexes.push_back( index );
                return false;
            }
        }();
 
        if( accept_now )
        {
            m_acceptor->call_accept_now( index );
        }
    }
};
 
} /* namespace impl */
 
/*!
 * @brief An implementation of connection count limiter for the case
 * when connection count is not limited.
 *
 * @since v.0.6.12
 */
class noop_connection_count_limiter_t
{
    not_null_pointer_t< connection_count_limits::impl::acceptor_callback_iface_t > m_acceptor;
 
public:
    noop_connection_count_limiter_t(
        not_null_pointer_t< connection_count_limits::impl::acceptor_callback_iface_t > acceptor,
        max_parallel_connections_t /*max_parallel_connections*/,
        max_active_accepts_t /*max_pending_indexes*/ )
        :   m_acceptor{ acceptor }
    {
    }
 
    void
    increment_parallel_connections() noexcept { /* Nothing to do */ }
 
    void
    decrement_parallel_connections() noexcept { /* Nothing to do */ }
 
    /*!
     * Calls acceptor_callback_iface_t::call_accept_now() directly.
     * The @a index is never stored anywhere.
     */
    void
    accept_next( std::size_t index ) noexcept
    {
        m_acceptor->call_accept_now( index );
    }
};
 
/*!
 * @brief Template class for connection count limiter for the case when
 * connection count limit is actually used.
 *
 * The actual implementation will be provided by specializations of
 * that class for specific Strand types.
 *
 * @since v.0.6.12
 */
template< typename Strand >
class connection_count_limiter_t;
 
/*!
 * @brief Implementation of connection count limiter for single-threading
 * mode.
 *
 * In single-threading mode there is no need to protect limiter from
 * access from different threads. So null_mutex_t is used.
 *
 * @since v.0.6.12
 */
template<>
class connection_count_limiter_t< noop_strand_t >
    :   public connection_count_limits::impl::actual_limiter_t< null_mutex_t >
{
    using base_t = connection_count_limits::impl::actual_limiter_t< null_mutex_t >;
 
public:
    using base_t::base_t;
};
 
/*!
 * @brief Implementation of connection count limiter for multi-threading
 * mode.
 *
 * In multi-threading mode std::mutex is used for the protection of
 * limiter object.
 *
 * @since v.0.6.12
 */
template<>
class connection_count_limiter_t< default_strand_t >
    :   public connection_count_limits::impl::actual_limiter_t< std::mutex >
{
    using base_t = connection_count_limits::impl::actual_limiter_t< std::mutex >;
 
public:
    using base_t::base_t;
};
 
/*!
 * @brief Helper type for controlling the lifetime of the connection.
 *
 * Connection count limiter should be informed when a new connection
 * created and when an existing connection is closed. An instance
 * of connection_lifetime_monitor_t should be used for that purpose:
 * a new instance of connection_lifetime_monitor_t should be created
 * and bound to a connection object. The constructor of
 * connection_lifetime_monitor_t will inform the limiter about
 * the creation of a new connection. The destructor of
 * connection_lifetime_monitor_t will inform the limiter about the
 * destruction of a connection.
 *
 * @note
 * This type is not Copyable but Movabale.
 *
 * @attention
 * The pointer to Count_Manager passed to the constructor should
 * remain valid the whole lifetime of connection_lifetime_monitor_t
 * instance.
 *
 * @since v.0.6.12
 */
template< typename Count_Manager >
class connection_lifetime_monitor_t
{
// FIXME: there should be a more efficient way that doesn't require std::any
// instance. The std::any is used for a quick fix for
// https://github.com/Stiffstream/restinio/issues/246
 
    /// Holder of std::shared_ptr<acceptor_t>
    ///
    /// The acceptor shouldn't be destroyed while this monitor object is alive.
    /// To ensure this a shared_ptr has to be stored inside monitor instance.
    /// But acceptor_t depends on Traits type that in not available here. For
    /// simplicity shared_ptr is wrapped into std::any.
    std::any m_acceptor;
 
    /// Pointer to manager object that counts live connections.
    ///
    /// @note
    /// It may become nullptr if the object is moved.
    Count_Manager * m_manager;
 
public:
    template< typename Acceptor_Type >
    connection_lifetime_monitor_t(
        Acceptor_Type & acceptor,
        not_null_pointer_t< Count_Manager > manager ) noexcept
        :   m_acceptor{ acceptor.shared_from_this() }
        ,   m_manager{ manager }
    {
        m_manager->increment_parallel_connections();
    }
 
    ~connection_lifetime_monitor_t()
    {
        if( m_manager )
            m_manager->decrement_parallel_connections();
    }
 
    connection_lifetime_monitor_t(
        const connection_lifetime_monitor_t & ) = delete;
 
    friend void
    swap(
        connection_lifetime_monitor_t & a,
        connection_lifetime_monitor_t & b ) noexcept
    {
        using std::swap;
        swap( a.m_acceptor, b.m_acceptor );
        swap( a.m_manager, b.m_manager );
    }
 
    connection_lifetime_monitor_t(
        connection_lifetime_monitor_t && other ) noexcept
        :   m_acceptor{
                std::exchange(
                        other.m_acceptor,
                        std::any{} )
            }
        ,   m_manager{ std::exchange( other.m_manager, nullptr ) }
    {}
 
    connection_lifetime_monitor_t &
    operator=( connection_lifetime_monitor_t && other ) noexcept
    {
        connection_lifetime_monitor_t tmp{ std::move(other) };
        swap( *this, tmp );
        return *this;
    }
 
    connection_lifetime_monitor_t &
    operator=( const connection_lifetime_monitor_t & ) = delete;
};
 
/*!
 * @brief Specialization of connection_lifetime_monitor for the case
 * when connection count limiter is not used at all.
 *
 * Holds nothing. Does nothing.
 *
 * @since v.0.6.12
 */
template<>
class connection_lifetime_monitor_t< noop_connection_count_limiter_t >
{
public:
    template< typename Acceptor_Type >
    connection_lifetime_monitor_t(
        Acceptor_Type & /* acceptor */,
        not_null_pointer_t< noop_connection_count_limiter_t > ) noexcept
    {}
};
 
} /* namespace connection_count_limits */
 
/*!
 * @brief A kind of metafunction that deduces actual types related
 * to connection count limiter in the dependecy of Traits.
 *
 * Deduces the following types:
 *
 * - limiter_t. The actual type of connection count limiter to be
 *   used in the RESTinio's server;
 * - lifetime_monitor_t. The actual type of connection_lifetime_monitor
 *   to be used with connection objects.
 *
 * @tparam Traits The type with traits for RESTinio's server.
 *
 * @since v.0.6.12
 */
template<
    typename Traits >
struct connection_count_limit_types
{
    using limiter_t = typename std::conditional
        <
            Traits::use_connection_count_limiter,
            connection_count_limits::connection_count_limiter_t<
                    typename Traits::strand_t >,
            connection_count_limits::noop_connection_count_limiter_t
        >::type;
 
    using lifetime_monitor_t =
            connection_count_limits::connection_lifetime_monitor_t< limiter_t >;
};
 
} /* namespace restinio */