fixed_threads_container.hpp

Go to the documentation of this file.

#ifndef LIBJMMCG_CORE_PRIVATE_FIXED_THREADS_CONTAINER_HPP
#define LIBJMMCG_CORE_PRIVATE_FIXED_THREADS_CONTAINER_HPP
 
/******************************************************************************
** Copyright © 2012 by J.M.McGuiness, coder@hussar.me.uk
**
** This library is free software; you can redistribute it and/or
** modify it under the terms of the GNU Lesser General Public
** License as published by the Free Software Foundation; either
** version 2.1 of the License, or (at your option) any later version.
**
** This library is distributed in the hope that it will be useful,
** but WITHOUT ANY WARRANTY; without even the implied warranty of
** MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
** Lesser General Public License for more details.
**
** You should have received a copy of the GNU Lesser General Public
** License along with this library; if not, write to the Free Software
** Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
*/
 
#include "../../core/config.h"
#include "../../core/trace.hpp"
 
#include <algorithm>
#include <cassert>
#include <memory>
#include <unordered_map>
 
namespace jmmcg { namespace LIBJMMCG_VER_NAMESPACE { namespace ppd { namespace private_ {
 
/// This is a low-level collection of pool_thread objects, that is created by one other thread.
/**
   This class is a bit crafty: it creates two collections: one to contain the pool_threads, the other a hash map into that pool, to speed looking up into the pool. The former pool carefully in-place constructs the pool_threads, so that their layout is orderly, hopefully mimicing the architectural layout of cores. hence enhancing cache locality, etc.
 
   \see pool_thread
*/
template<
   class PTT,
   class ThrdT
>
class fixed_pool_of_threads {
   using internal_container_type=std::vector<ThrdT>;
 
public:
   using pool_traits_type=PTT;
   using os_traits=typename pool_traits_type::os_traits;
   using thread_traits=typename os_traits::thread_traits;
   using container_type=std::unordered_map<
      typename thread_traits::api_params_type::tid_type, ///< Note that this should be a perfect hash.
      typename internal_container_type::iterator
   >;
   using thread_type=typename internal_container_type::value_type;
   using value_type=thread_type;
   using key_type=typename container_type::key_type;
   using iterator=typename container_type::iterator;
   using const_iterator=typename container_type::const_iterator;
   using size_type=typename container_type::size_type;
   using exit_requested_type=typename PTT::template pool_thread_queue_details<typename thread_type::queue_model>::exit_requested_type;
   using have_work_type=typename PTT::template pool_thread_queue_details<typename thread_type::queue_model>::have_work_type;
 
   fixed_pool_of_threads(size_type const sz, exit_requested_type &exit_requested, typename internal_container_type::value_type::signalled_work_queue_type &signalled_work_queue) noexcept(false) {
      pool.reserve(sz);
      lookup_table.reserve(sz);
      lookup_table.rehash(sz);
      for (size_type i=0; i<sz; ++i) {
         pool.emplace_back(std::ref(exit_requested), std::ref(signalled_work_queue));
         auto a_thread=std::next(pool.begin(), pool.size()-1);
         a_thread->create_running();
         typename thread_traits::api_params_type::processor_mask_type mask(i);
         try {
            a_thread->kernel_affinity(mask);
         } catch (typename thread_type::exception_type const &ex) {
            // It's not a nightmare if we can't set the affinity.
         }
         const auto tid=a_thread->params().id;
         lookup_table.insert(typename container_type::value_type(tid, a_thread));
      }
      assert(pool.size()==lookup_table.size());
      assert(size()==sz);
   }
 
   fixed_pool_of_threads(size_type const sz, exit_requested_type &exit_requested) noexcept(false) {
      pool.reserve(sz);
      lookup_table.reserve(sz);
      lookup_table.rehash(sz);
      for (size_type i=0; i<sz; ++i) {
         pool.emplace_back(std::ref(exit_requested));
         auto a_thread=std::next(pool.begin(), pool.size()-1);
         a_thread->create_running();
         typename thread_traits::api_params_type::processor_mask_type mask(i);
         try {
            a_thread->kernel_affinity(mask);
         } catch (typename thread_type::exception_type const &ex) {
            // It's not a nightmare if we can't set the affinity.
         }
         const auto tid=a_thread->params().id;
         lookup_table.insert(typename container_type::value_type(tid, a_thread));
      }
      assert(pool.size()==lookup_table.size());
      assert(size()==sz);
   }
 
   fixed_pool_of_threads(fixed_pool_of_threads const &)=delete;
   fixed_pool_of_threads &operator=(fixed_pool_of_threads const &)=delete;
   fixed_pool_of_threads(fixed_pool_of_threads &&)=default;
   fixed_pool_of_threads &operator=(fixed_pool_of_threads &&)=default;
 
   thread_type const &first_thread() const noexcept(true) FORCE_INLINE {
      return *pool.begin();
   }
   thread_type &first_thread() noexcept(true) FORCE_INLINE {
      return *pool.begin();
   }
 
   const_iterator end() const noexcept(true) FORCE_INLINE {
      return lookup_table.end();
   }
 
   const_iterator find(key_type key) const noexcept(true) {
      return lookup_table.find(key);
   }
 
   iterator find(key_type key) noexcept(true) FORCE_INLINE {
      return lookup_table.find(key);
   }
 
   bool __fastcall empty() const noexcept(true) FORCE_INLINE {
      assert(lookup_table.empty()==pool.empty());
      return pool.empty();
   }
 
   size_type __fastcall size() const noexcept(true) FORCE_INLINE {
      const size_type sz=lookup_table.size();
      [[maybe_unused]] const size_type currently_running=num_running();
      assert(sz>=currently_running);
      return sz;
   }
 
   void clear() noexcept(false) FORCE_INLINE {
      // Note that there may be inactive threads in the infinite threads pool, that haven't yet been purged.
      assert(lookup_table.size()>=num_running());
      lookup_table.clear();
      pool.clear();
   }
 
private:
   template<class S> friend
   class wrkr_accumulate_across_threads;
   template<class S> friend
   class mstr_accumulate_across_threads;
   internal_container_type pool;
   container_type lookup_table;
 
   internal_container_type const &__fastcall colln() const noexcept(true) FORCE_INLINE {
      return pool;
   }
 
   size_type num_running() const noexcept(true) FORCE_INLINE {
      assert(pool.size()==lookup_table.size());
      return std::accumulate(
         pool.begin(),
         pool.end(),
         size_type(),
         [](size_type const s, typename internal_container_type::const_iterator::value_type const &v) {
            return s+static_cast<size_type>(v.is_running());
         }
      );
   }
};
 
} } } }
 
#endif