active_group/pub.cpp

Go to the documentation of this file.

/*
  SObjectizer 5.
*/
 
#include <so_5/disp/active_group/pub.hpp>
 
#include <so_5/send_functions.hpp>
 
#include <so_5/details/rollback_on_exception.hpp>
 
#include <so_5/disp/reuse/actual_work_thread_factory_to_use.hpp>
#include <so_5/disp/reuse/data_source_prefix_helpers.hpp>
#include <so_5/disp/reuse/make_actual_dispatcher.hpp>
 
#include <so_5/disp/reuse/work_thread/work_thread.hpp>
 
#include <so_5/stats/repository.hpp>
#include <so_5/stats/messages.hpp>
#include <so_5/stats/std_names.hpp>
 
#include <map>
#include <mutex>
#include <algorithm>
 
namespace so_5
{
 
namespace disp
{
 
namespace active_group
{
 
namespace impl
{
 
namespace work_thread = so_5::disp::reuse::work_thread;
namespace stats = so_5::stats;
 
namespace
{
 
/*!
 * \brief Just a helper function for consequetive call to shutdown and wait.
 *
 * \since
 * v.5.5.4
 */
template< class T >
void
shutdown_and_wait( T & w )
  {
    w.shutdown();
    w.wait();
  }
 
void
send_thread_activity_stats(
  const so_5::mbox_t &,
  const stats::prefix_t &,
  work_thread::work_thread_no_activity_tracking_t & )
  {
    /* Nothing to do */
  }
 
void
send_thread_activity_stats(
  const so_5::mbox_t & mbox,
  const stats::prefix_t & prefix,
  work_thread::work_thread_with_activity_tracking_t & wt )
  {
    so_5::send< stats::messages::work_thread_activity >(
        mbox,
        prefix,
        stats::suffixes::work_thread_activity(),
        wt.thread_id(),
        wt.take_activity_stats() );
  }
 
} /* anonymous */
 
//
// actual_dispatcher_iface_t
//
/*!
 * \brief An actual interface of active group dispatcher.
 *
 * \since
 * v.5.6.0
 */
class actual_dispatcher_iface_t : public basic_dispatcher_iface_t
  {
  public :
    /*!
     * \brief Create a new thread for a group if it necessary.
     *
     * If name \a group_name is unknown then a new work
     * thread is started. This thread is marked as it has one
     * working agent on it.
     *
     * If there already is a thread for \a group_name then the
     * counter of working agents is incremented.
     */
    virtual void
    allocate_thread_for_group( const std::string & group_name ) = 0;
 
    /*!
     * \brief Get the event_queue for the specified active group.
     *
     * It is expected that thread for the group is already
     * created by calling allocate_thread_for_group() method.
     */
    virtual so_5::event_queue_t *
    query_thread_for_group( const std::string & group_name ) noexcept = 0;
 
    /*!
     * \brief Release the thread for the specified active group.
     *
     * Method decrements the working agent count for the thread of
     * \a group_name. If there no more working agents left then
     * the event_queue and working thread for that group will be
     * destroyed.
     */
    virtual void
    release_thread_for_group( const std::string & group_name ) noexcept = 0;
  };
 
//
// actual_dispatcher_iface_shptr_t
//
using actual_dispatcher_iface_shptr_t =
    std::shared_ptr< actual_dispatcher_iface_t >;
 
//
// actual_binder_t
//
/*!
 * \brief Implementation of binder interface for %active_group dispatcher.
 *
 * \since
 * v.5.6.0
 */
class actual_binder_t final : public disp_binder_t
  {
    //! Dispatcher to be used.
    actual_dispatcher_iface_shptr_t m_disp;
    //! Name of group for new agents.
    const std::string m_group_name;
 
  public :
    actual_binder_t(
      actual_dispatcher_iface_shptr_t disp,
      nonempty_name_t group_name ) noexcept
      : m_disp{ std::move(disp) }
      , m_group_name{ group_name.giveout_value() }
      {}
 
    void
    preallocate_resources(
      agent_t & /*agent*/ ) override
      {
        m_disp->allocate_thread_for_group( m_group_name );
      }
 
    void
    undo_preallocation(
      agent_t & /*agent*/ ) noexcept override
      {
        m_disp->release_thread_for_group( m_group_name );
      }
 
    void
    bind(
      agent_t & agent ) noexcept override
      {
        auto queue = m_disp->query_thread_for_group( m_group_name );
        agent.so_bind_to_dispatcher( *queue );
      }
 
    void
    unbind(
      agent_t & /*agent*/ ) noexcept override
      {
        m_disp->release_thread_for_group( m_group_name );
      }
  };
 
//
// dispatcher_template_t
//
 
/*!
 * \brief Implementation of active object dispatcher in form of template class.
 */
template< typename Work_Thread >
class dispatcher_template_t final : public actual_dispatcher_iface_t
  {
  public:
    dispatcher_template_t(
      //! SObjectizer Environment to work in.
      outliving_reference_t< environment_t > env,
      //! Base part of data sources names.
      const std::string_view name_base,
      //! Dispatcher's parameters.
      disp_params_t params )
      : m_env{ env }
      , m_params{ std::move(params) }
      , m_data_source{
          outliving_mutable(m_env.get().stats_repository()),
          name_base,
          outliving_mutable( *this )
        }
      {}
 
    ~dispatcher_template_t() noexcept override
      {
        // All working threads should receive stop signal.
        for( auto & p: m_groups )
          p.second.m_thread->shutdown();
 
        // All working threads should be joined.
        for( auto & p: m_groups )
          p.second.m_thread->wait();
      }
 
    disp_binder_shptr_t
    binder( nonempty_name_t group_name ) override
      {
        return std::make_shared< actual_binder_t >(
            this->shared_from_this(),
            std::move(group_name) );
      }
 
    void
    allocate_thread_for_group( const std::string & group_name ) override
      {
        std::lock_guard< std::mutex > lock{ m_lock };
 
        auto it = m_groups.find( group_name );
 
        if( m_groups.end() == it )
          {
            // New thread should be created.
            auto thread = std::make_shared< Work_Thread >(
                acquire_work_thread( m_params, m_env.get() ),
                m_params.queue_params().lock_factory() );
 
            thread->start();
 
            so_5::details::do_with_rollback_on_exception(
                [&] {
                  m_groups.emplace(
                      group_name,
                      thread_with_refcounter_t{ thread, 1u } );
                },
                [&thread] { shutdown_and_wait( *thread ); } );
          }
        else
          {
            // Number of agents bound has to be incremented now.
            it->second.m_user_agent += 1u;
          }
      }
 
    so_5::event_queue_t *
    query_thread_for_group( const std::string & group_name ) noexcept override
      {
        std::lock_guard< std::mutex > lock{ m_lock };
 
        return m_groups.find( group_name )->second.m_thread->
            get_agent_binding();
      }
 
    void
    release_thread_for_group( const std::string & group_name ) noexcept override
      {
        auto thread = search_and_try_remove_group_from_map( group_name );
        if( thread )
          shutdown_and_wait( *thread );
      }
 
  private:
    friend class disp_data_source_t;
 
    //! An alias for shared pointer to work thread.
    using work_thread_shptr_t = std::shared_ptr< Work_Thread >;
 
    //! Auxiliary class for the working agent counting.
    struct thread_with_refcounter_t
      {
        work_thread_shptr_t m_thread;
        std::size_t m_user_agent;
      };
 
    //! Typedef for mapping from group names to a single thread
    //! dispatcher.
    using active_group_map_t =
        std::map< std::string, thread_with_refcounter_t >;
 
    /*!
     * \brief Data source for run-time monitoring of whole dispatcher.
     *
     * \since
     * v.5.5.4
     */
    class disp_data_source_t final : public stats::source_t
      {
        //! Dispatcher to work with.
        outliving_reference_t< dispatcher_template_t > m_dispatcher;
 
        //! Basic prefix for data sources.
        stats::prefix_t m_base_prefix;
 
      public :
        disp_data_source_t(
          const std::string_view name_base,
          outliving_reference_t< dispatcher_template_t > disp )
          : m_dispatcher{ disp }
          {
            using namespace so_5::disp::reuse;
 
            m_base_prefix = make_disp_prefix(
                "ag", // ao -- active_groups
                name_base,
                &m_dispatcher );
          }
 
        void
        distribute( const so_5::mbox_t & mbox ) override
          {
            auto & disp = m_dispatcher.get();
 
            std::lock_guard< std::mutex > lock{ disp.m_lock };
 
            so_5::send< stats::messages::quantity< std::size_t > >(
                mbox,
                m_base_prefix,
                stats::suffixes::disp_active_group_count(),
                disp.m_groups.size() );
 
            std::size_t agent_count = 0;
            for( const auto & p : disp.m_groups )
              {
                distribute_value_for_work_thread(
                    mbox,
                    p.first,
                    p.second );
 
                agent_count += p.second.m_user_agent;
              }
 
            so_5::send< stats::messages::quantity< std::size_t > >(
                mbox,
                m_base_prefix,
                stats::suffixes::agent_count(),
                agent_count );
          }
 
      private:
        void
        distribute_value_for_work_thread(
          const so_5::mbox_t & mbox,
          const std::string & group_name,
          const thread_with_refcounter_t & wt )
          {
            std::ostringstream ss;
            ss << m_base_prefix.c_str() << "/wt-" << group_name;
 
            const stats::prefix_t prefix{ ss.str() };
 
            so_5::send< stats::messages::quantity< std::size_t > >(
                mbox,
                prefix,
                stats::suffixes::agent_count(),
                wt.m_user_agent );
 
            so_5::send< stats::messages::quantity< std::size_t > >(
                mbox,
                prefix,
                stats::suffixes::work_thread_queue_size(),
                wt.m_thread->demands_count() );
 
            send_thread_activity_stats(
                mbox,
                prefix,
                *(wt.m_thread) );
          }
      };
 
    /*!
     * \brief SObjectizer Environment to work in.
     *
     * It is necessary for calling work_thread_factory.
     *
     * \since v.5.7.3
     */
    outliving_reference_t< environment_t > m_env;
 
    //! Parameters for the dispatcher.
    const disp_params_t m_params;
 
    //! A map of dispatchers for active groups.
    active_group_map_t m_groups;
 
    //! This object lock.
    std::mutex m_lock;
 
    /*!
     * \brief Data source for run-time monitoring.
     *
     * \since
     * v.5.5.4, v.5.6.0
     */
    stats::auto_registered_source_holder_t< disp_data_source_t >
        m_data_source;
 
    /*!
     * \brief Helper function for searching and erasing agent's
     * thread from map of active threads.
     *
     * \since
     * v.5.5.4
     *
     * \note Does all actions on locked object.
     *
     * \return nullptr if thread for the group is not found
     * or there are still some agents on it.
     */
    work_thread_shptr_t
    search_and_try_remove_group_from_map(
      const std::string & group_name ) noexcept
      {
        work_thread_shptr_t result;
 
        std::lock_guard< std::mutex > lock{ m_lock };
 
        auto it = m_groups.find( group_name );
 
        if( m_groups.end() != it && 0u == --(it->second.m_user_agent) )
          {
            result = it->second.m_thread;
            m_groups.erase( it );
          }
 
        return result;
      }
};
 
//
// dispatcher_handle_maker_t
//
class dispatcher_handle_maker_t
  {
  public :
    static dispatcher_handle_t
    make( actual_dispatcher_iface_shptr_t disp ) noexcept
      {
        return { std::move( disp ) };
      }
  };
 
} /* namespace impl */
 
//
// make_dispatcher
//
SO_5_FUNC dispatcher_handle_t
make_dispatcher(
  environment_t & env,
  const std::string_view data_sources_name_base,
  disp_params_t params )
  {
    using namespace so_5::disp::reuse;
 
    using dispatcher_no_activity_tracking_t =
        impl::dispatcher_template_t<
            work_thread::work_thread_no_activity_tracking_t >;
 
    using dispatcher_with_activity_tracking_t =
        impl::dispatcher_template_t<
            work_thread::work_thread_with_activity_tracking_t >;
 
    auto binder = so_5::disp::reuse::make_actual_dispatcher<
            impl::actual_dispatcher_iface_t,
            dispatcher_no_activity_tracking_t,
            dispatcher_with_activity_tracking_t >(
        outliving_mutable(env),
        data_sources_name_base,
        std::move(params) );
 
    return impl::dispatcher_handle_maker_t::make( std::move(binder) );
  }
 
} /* namespace active_group */
 
} /* namespace disp */
 
} /* namespace so_5 */