pool_thread_impl.hpp

Go to the documentation of this file.

/******************************************************************************
** Copyright © 2004 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
*/
 
namespace jmmcg { namespace LIBJMMCG_VER_NAMESPACE { namespace ppd { namespace pool { namespace private_ { namespace thread_types {
 
template<class PTT> inline
steal<generic_traits::return_data::nonjoinable, thread_os_traits<generic_traits::api_type::posix_pthreads, heavyweight_threading>, PTT, pool_traits::work_distribution_mode_t::queue_model_t::pool_owns_queue>::
steal(exit_requested_type &exit_r, signalled_work_queue_type &work_q) noexcept(true)
: base_t(exit_r), batch(work_q) {
}
 
template<class PTT> inline
steal<generic_traits::return_data::nonjoinable, thread_os_traits<generic_traits::api_type::posix_pthreads, heavyweight_threading>, PTT, pool_traits::work_distribution_mode_t::queue_model_t::pool_owns_queue>::
~steal() noexcept(false) {
   base_t::wait_thread_exit();
   assert(batch.batch.batch_empty());
}
 
template<class PTT> inline
bool
steal<generic_traits::return_data::nonjoinable, thread_os_traits<generic_traits::api_type::posix_pthreads, heavyweight_threading>, PTT, pool_traits::work_distribution_mode_t::queue_model_t::pool_owns_queue>::
add_work_to_batch(typename signalled_work_queue_type::value_type &&wk) {
   return batch.batch.add_work_to_batch(std::forward<typename signalled_work_queue_type::value_type>(wk));
}
 
template<class PTT> inline
bool
steal<generic_traits::return_data::nonjoinable, thread_os_traits<generic_traits::api_type::posix_pthreads, heavyweight_threading>, PTT, pool_traits::work_distribution_mode_t::queue_model_t::pool_owns_queue>::
process_a_batch_item(typename os_traits::thread_exception const &exception_thrown_in_thr) {
   return batch.batch.process_a_batch_item();
}
 
template<class PTT> inline
typename steal<generic_traits::return_data::nonjoinable, thread_os_traits<generic_traits::api_type::posix_pthreads, heavyweight_threading>, PTT, pool_traits::work_distribution_mode_t::queue_model_t::pool_owns_queue>::thread_traits::api_params_type::states
steal<generic_traits::return_data::nonjoinable, thread_os_traits<generic_traits::api_type::posix_pthreads, heavyweight_threading>, PTT, pool_traits::work_distribution_mode_t::queue_model_t::pool_owns_queue>::
process() noexcept(false) {
   // Serialize access to the queue by all of the other threads in the pool, this means that only one job can be removed at a time from the work queue.
   for (;;) {
      typename exit_requested_type::lock_result_type lkd;
      {
         const typename os_traits::thread_traits::cancellability set;
         lkd=this->exit_requested_.lock();
      }
      if (lkd.second!=lock_traits::atom_set || lkd.first==exit_requested_type::states::exit_requested) {
         // Ensure the rest of the threads in the pool exit.
         this->exit_requested_.set(exit_requested_type::states::exit_requested);
         break;
      } else if (lkd.first==exit_requested_type::states::new_work_arrived) {
         // The counter in signalled_work_queue is one item too low, and will need to be re-added, but we'll do that closer to where we lock the queue, to reduce the chance of a horizontal thread slipping in and getting the work.
         batch.batch.process_a_batch(batch.signalled_work_queue);
         assert(batch.batch.batch_empty());
      }
   }
   return thread_traits::api_params_type::no_kernel_thread;
}
 
template<class PTT> inline
slave<generic_traits::return_data::nonjoinable, thread_os_traits<generic_traits::api_type::posix_pthreads, heavyweight_threading>, PTT>::
slave(exit_requested_type &exit_r, typename signalled_work_queue_type::value_type &&wk) noexcept(true)
: base_t(exit_r), some_work(std::forward<typename signalled_work_queue_type::value_type>(wk)), statistics_() {
}
 
template<class PTT> inline
slave<generic_traits::return_data::nonjoinable, thread_os_traits<generic_traits::api_type::posix_pthreads, heavyweight_threading>, PTT>::
~slave() noexcept(true) {
   base_t::wait_thread_exit();
}
 
template<class PTT> inline
typename slave<generic_traits::return_data::nonjoinable, thread_os_traits<generic_traits::api_type::posix_pthreads, heavyweight_threading>, PTT>::thread_traits::api_params_type::states
slave<generic_traits::return_data::nonjoinable, thread_os_traits<generic_traits::api_type::posix_pthreads, heavyweight_threading>, PTT>::
process() noexcept(false) {
   some_work->process_nonjoinable(signalled_work_queue_type::value_ret_type::value_type::value_type::cfg_type::vertical_edge_annotation);
   this->statistics_.processed_vertical_work();
   return thread_traits::api_params_type::no_kernel_thread;
}
 
template<class PTT> inline
steal<generic_traits::return_data::joinable, thread_os_traits<generic_traits::api_type::posix_pthreads, heavyweight_threading>, PTT, pool_traits::work_distribution_mode_t::queue_model_t::pool_owns_queue>::
steal(exit_requested_type &exit_r, signalled_work_queue_type &work_q) noexcept(true)
: base_t(exit_r), batch(work_q) {
}
 
template<class PTT> inline
steal<generic_traits::return_data::joinable, thread_os_traits<generic_traits::api_type::posix_pthreads, heavyweight_threading>, PTT, pool_traits::work_distribution_mode_t::queue_model_t::pool_owns_queue>::
~steal() noexcept(false) {
   base_t::wait_thread_exit();
   assert(batch.batch.batch_empty());
}
 
template<class PTT> inline
bool
steal<generic_traits::return_data::joinable, thread_os_traits<generic_traits::api_type::posix_pthreads, heavyweight_threading>, PTT, pool_traits::work_distribution_mode_t::queue_model_t::pool_owns_queue>::
add_work_to_batch(typename signalled_work_queue_type::value_type &&wk) {
   return batch.batch.add_work_to_batch(std::forward<typename signalled_work_queue_type::value_type>(wk));
}
 
template<class PTT> inline
bool
steal<generic_traits::return_data::joinable, thread_os_traits<generic_traits::api_type::posix_pthreads, heavyweight_threading>, PTT, pool_traits::work_distribution_mode_t::queue_model_t::pool_owns_queue>::
process_a_batch_item(typename os_traits::thread_exception const &exception_thrown_in_thr) {
   return batch.batch.process_a_batch_item();
}
 
template<class PTT> inline
typename steal<generic_traits::return_data::joinable, thread_os_traits<generic_traits::api_type::posix_pthreads, heavyweight_threading>, PTT, pool_traits::work_distribution_mode_t::queue_model_t::pool_owns_queue>::thread_traits::api_params_type::states
steal<generic_traits::return_data::joinable, thread_os_traits<generic_traits::api_type::posix_pthreads, heavyweight_threading>, PTT, pool_traits::work_distribution_mode_t::queue_model_t::pool_owns_queue>::
process() noexcept(false) {
   // Serialize access to the queue by all of the other threads in the pool, this means that only one job can be removed at a time from the work queue.
   for (;;) {
      typename exit_requested_type::lock_result_type lkd;
      {
         const typename os_traits::thread_traits::cancellability set;
         lkd=this->exit_requested_.lock();
      }
      assert(lkd.second==exit_requested_type::lock_result_type::second_type::atom_set);
      if (lkd.first==exit_requested_type::states::exit_requested) {
         // Ensure the rest of the threads in the pool exit.
         this->exit_requested_.set(exit_requested_type::states::exit_requested);
         break;
      } else if (lkd.first==exit_requested_type::states::new_work_arrived) {
         // The counter in signalled_work_queue is one item too low, and will need to be re-added, but we'll do that closer to where we lock the queue, to reduce the chance of a horizontal thread slipping in and getting the work.
         batch.batch.process_a_batch(batch.signalled_work_queue);
         assert(batch.batch.batch_empty());
      }
   }
   return thread_traits::api_params_type::no_kernel_thread;
}
 
template<class PTT> inline
steal<generic_traits::return_data::joinable, thread_os_traits<generic_traits::api_type::posix_pthreads, heavyweight_threading>, PTT, pool_traits::work_distribution_mode_t::queue_model_t::thread_owns_queue<pool_traits::work_distribution_mode_t::queue_model_t::stealing_mode_t::random>>::
steal(exit_requested_type &exit_r) noexcept(true)
: base_t(exit_r), work() {
}
 
template<class PTT> inline
steal<generic_traits::return_data::joinable, thread_os_traits<generic_traits::api_type::posix_pthreads, heavyweight_threading>, PTT, pool_traits::work_distribution_mode_t::queue_model_t::thread_owns_queue<pool_traits::work_distribution_mode_t::queue_model_t::stealing_mode_t::random>>::
steal(steal &&s) noexcept(true)
: base_t(s.exit_requested_), work(std::move(s.work)) {
}
 
template<class PTT> inline
steal<generic_traits::return_data::joinable, thread_os_traits<generic_traits::api_type::posix_pthreads, heavyweight_threading>, PTT, pool_traits::work_distribution_mode_t::queue_model_t::thread_owns_queue<pool_traits::work_distribution_mode_t::queue_model_t::stealing_mode_t::random>>::
~steal() noexcept(false) {
   base_t::wait_thread_exit();
}
 
template<class PTT> inline bool
steal<generic_traits::return_data::joinable, thread_os_traits<generic_traits::api_type::posix_pthreads, heavyweight_threading>, PTT, pool_traits::work_distribution_mode_t::queue_model_t::thread_owns_queue<pool_traits::work_distribution_mode_t::queue_model_t::stealing_mode_t::random>>::
push_front(typename signalled_work_queue_type::value_type &&wk) {
   work.batch.push_front(std::forward<typename signalled_work_queue_type::value_type>(wk));
   return true;
}
 
template<class PTT> inline bool
steal<generic_traits::return_data::joinable, thread_os_traits<generic_traits::api_type::posix_pthreads, heavyweight_threading>, PTT, pool_traits::work_distribution_mode_t::queue_model_t::thread_owns_queue<pool_traits::work_distribution_mode_t::queue_model_t::stealing_mode_t::random>>::
process_a_batch_item(typename os_traits::thread_exception const &exception_thrown_in_thread) {
// TODO
}
 
template<class PTT> inline
typename steal<generic_traits::return_data::joinable, thread_os_traits<generic_traits::api_type::posix_pthreads, heavyweight_threading>, PTT, pool_traits::work_distribution_mode_t::queue_model_t::thread_owns_queue<pool_traits::work_distribution_mode_t::queue_model_t::stealing_mode_t::random>>::thread_traits::api_params_type::states
steal<generic_traits::return_data::joinable, thread_os_traits<generic_traits::api_type::posix_pthreads, heavyweight_threading>, PTT, pool_traits::work_distribution_mode_t::queue_model_t::thread_owns_queue<pool_traits::work_distribution_mode_t::queue_model_t::stealing_mode_t::random>>::
process() noexcept(false) {
   using cfg_type=typename container_type::container_type::value_type::value_type::cfg_type;
 
   for (;;) {
      // Busy wait for work or exit flag....
      while (work.batch.empty()) {
         const typename exit_requested_type::lock_result_type lkd=this->exit_requested_.try_lock();
         assert(lkd.second==exit_requested_type::lock_result_type::second_type::atom_set);
         if (lkd.first==exit_requested_type::states::exit_requested) {
            // Ensure the rest of the threads in the pool exit.
            this->exit_requested_.set(exit_requested_type::states::exit_requested);
            return thread_traits::api_params_type::no_kernel_thread;
         }
      }
      typename container_type::container_type::value_type current_work(work.batch.pop_front_1_nochk_nosig());
      ppd::private_::eval_shared_deleter_t<typename container_type::container_type::value_type> wk(current_work);
      wk.process_the_work(std::bind(&statistics_type::processed_vertical_work, &work.statistics_), cfg_type::vertical_edge_annotation);
   }
   return thread_traits::api_params_type::no_kernel_thread;
}
 
template<class PTT> inline
slave<generic_traits::return_data::joinable, thread_os_traits<generic_traits::api_type::posix_pthreads, heavyweight_threading>, PTT>::
slave(exit_requested_type &exit_r, typename signalled_work_queue_type::value_type &&wk) noexcept(true)
: base_t(exit_r), some_work(std::forward<typename signalled_work_queue_type::value_type>(wk)), statistics_() {
}
 
template<class PTT> inline
slave<generic_traits::return_data::joinable, thread_os_traits<generic_traits::api_type::posix_pthreads, heavyweight_threading>, PTT>::
~slave() noexcept(true) {
   base_t::wait_thread_exit();
}
 
template<class PTT> inline
typename slave<generic_traits::return_data::joinable, thread_os_traits<generic_traits::api_type::posix_pthreads, heavyweight_threading>, PTT>::thread_traits::api_params_type::states
slave<generic_traits::return_data::joinable, thread_os_traits<generic_traits::api_type::posix_pthreads, heavyweight_threading>, PTT>::
process() noexcept(false) {
   class set_work_complete {
   public:
      explicit set_work_complete(typename signalled_work_queue_type::value_type::atomic_ptr_t &wc) noexcept(true)
      : work_ptr(wc) {
      }
      ~set_work_complete() noexcept(true) {
         work_ptr->work_complete()->set();
      }
 
   private:
      typename signalled_work_queue_type::value_type::atomic_ptr_t &work_ptr;
   };
 
   typename signalled_work_queue_type::value_type::atomic_ptr_t work_ptr(some_work.get());
   if (dynamic_cast<typename signalled_work_queue_type::value_type::no_ref_counting *>(work_ptr.get())) {
      some_work=typename signalled_work_queue_type::value_type();
   }
   if (work_ptr->result_traits()==generic_traits::return_data::joinable) {
      const set_work_complete setter(work_ptr);
      work_ptr->process_joinable(signalled_work_queue_type::value_ret_type::value_type::value_type::cfg_type::vertical_edge_annotation);
   } else {
      work_ptr->process_nonjoinable(signalled_work_queue_type::value_ret_type::value_type::value_type::cfg_type::vertical_edge_annotation);
   }
   this->statistics_.processed_vertical_work();
   return thread_traits::api_params_type::no_kernel_thread;
}
 
} } } } } }