parallel_algorithms_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
*/
 
#include "../../core/thread_pool_aspects.hpp"
 
#include <algorithm>
#include <memory>
#include <numeric>
 
namespace jmmcg { namespace LIBJMMCG_VER_NAMESPACE { namespace ppd { namespace private_ {
 
namespace alg_wk_wrap {
 
   template<class Conts, typename CtrPred> inline void
   count_if_reduce<Conts, CtrPred>::process() {
      const typename operation_type::result_type::value_type intermediate=std::count_if(beg, end, fn.input().pred);
      fn.get_results()+=intermediate;
   }
 
   template<class Conts, typename Fn> inline void
   accumulate_reduce<Conts, Fn>::process() {
      const typename operation_type::result_type::value_type intermediate=std::accumulate(beg, end, fn.input().init.get(), fn.input().binop);
      fn.get_results().apply(intermediate, fn.input().binop);
   }
 
   template<class Conts, typename CtrPred> inline constexpr
   find_if_reduce<Conts, CtrPred>::find_if_reduce(in_iterator const &b, in_iterator const &e, operation_type &w)
   : beg(b), end(e), fn(w) {
   }
 
   template<class Conts, typename CtrPred> inline void
   find_if_reduce<Conts, CtrPred>::process() {
      if (!fn.get_results()) {
         const typename operation_type::result_type::value_type intermediate=std::find_if(beg, end, fn.input().pred)!=end;
         if (intermediate) {
            fn.get_results()=true;
         }
      }
   }
 
   template<class Conts, typename Fn>
   class max_element_reduce<Conts, Fn>::max : public std::binary_function<typename container_type::value_type, typename container_type::value_type, typename container_type::value_type> {
   public:
      typedef std::binary_function<typename container_type::value_type, typename container_type::value_type, typename container_type::value_type> base_t;
      typedef typename base_t::first_argument_type first_argument_type;
      typedef typename base_t::second_argument_type second_argument_type;
      typedef typename base_t::result_type result_type;
 
      explicit constexpr max(operation_type &w) noexcept(true) FORCE_INLINE
      : fn(w) {}
 
      result_type __fastcall operator()(first_argument_type lhs, second_argument_type rhs) const FORCE_INLINE {
         return std::max(lhs, rhs, fn.input().binop);
      }
 
   private:
      operation_type &fn;
   };
 
   template<class Conts, typename Fn> inline constexpr
   max_element_reduce<Conts, Fn>::max_element_reduce(in_iterator const &b, in_iterator const &e, operation_type &w)
   : beg(b), end(e), fn(w) {
   }
 
   template<class Conts, typename Fn> inline void
   max_element_reduce<Conts, Fn>::process() {
      const in_iterator max_in_subrange(std::max_element(beg, end, fn.input().binop));
      assert(max_in_subrange!=end);
      const typename operation_type::result_type::value_type intermediate=*max_in_subrange;
      fn.get_results().apply(intermediate, max(fn));
   }
 
   template<class Conts, typename Fn>
   class min_element_reduce<Conts, Fn>::min : public std::binary_function<typename container_type::value_type, typename container_type::value_type, typename container_type::value_type> {
   public:
      typedef std::binary_function<typename container_type::value_type, typename container_type::value_type, typename container_type::value_type> base_t;
      typedef typename base_t::first_argument_type first_argument_type;
      typedef typename base_t::second_argument_type second_argument_type;
      typedef typename base_t::result_type result_type;
 
      explicit constexpr min(operation_type &w) noexcept(true) FORCE_INLINE
      : fn(w) {}
 
      result_type __fastcall operator()(first_argument_type lhs, second_argument_type rhs) const FORCE_INLINE {
         return std::min(lhs, rhs, fn.input().binop);
      }
 
   private:
      operation_type &fn;
   };
 
   template<class Conts, typename Fn> inline constexpr
   min_element_reduce<Conts, Fn>::min_element_reduce(in_iterator const &b, in_iterator const &e, operation_type &w)
   : beg(b), end(e), fn(w) {
   }
 
   template<class Conts, typename Fn> inline void
   min_element_reduce<Conts, Fn>::process() {
      const in_iterator min_in_subrange(std::min_element(beg, end, fn.input().binop));
      assert(min_in_subrange!=end);
      const typename operation_type::result_type::value_type intermediate=*min_in_subrange;
      fn.get_results().apply(intermediate, min(fn));
   }
 
   template<class Conts, typename Fn> inline constexpr
   reverse_reduce<Conts, Fn>::reverse_reduce(in_iterator const &bs, in_iterator const &es, operation_type const &w)
   : beg_subrange(bs), end_subrange(es), fn(w), cont_size(std::distance(fn.input().cont_beg(), fn.input().cont_end())) {
   }
 
   template<class Conts, typename Fn> inline void
   reverse_reduce<Conts, Fn>::process() const {
      for (in_iterator lhs(beg_subrange); lhs!=end_subrange; ++lhs) {
   // fn.input().cont_beg() ... beg_subrange ... lhs ... end_subrange ... middle ... rhs ... fn.input().cont_end()
         const typename std::iterator_traits<in_iterator>::difference_type dist_from_beg=std::distance(fn.input().cont_beg(), lhs);
         const in_iterator rhs(std::next(fn.input().cont_beg(), cont_size-dist_from_beg-1));
         fn.input().binop(lhs, rhs);
      }
   }
 
   template<typename Conts, class UniOp> inline void
   fill_n_reduce<Conts, UniOp>::process() const {
      std::fill(beg, end, val.input().value);
   }
 
   template<typename Conts, class UniOp> inline void
   fill_reduce<Conts, UniOp>::process() const {
      std::fill(beg, end, val.input().value);
   }
 
   template<class Conts, typename Pred> inline void
   swap_ranges_reduce<Conts, Pred>::process() {
      auto const &op=fn.input().op;
      for (; begin1!=end1; ++begin1, ++begin2) {
         if (op.operator()(*begin1, *begin2)) {
            std::iter_swap(begin1, begin2);
         }
      }
   }
 
   template<class Comp, class TPB>
   template<class CoreWk> inline void
   merge_work_type<Comp, TPB>::resize_output(CoreWk &wk) noexcept(false) {
      wk.resize_output(
         wk.work_complete()->containers().input1.size()+wk.work_complete()->containers().input2.size()
      );
   }
 
   template<direction Dir, class out_iterator, class Closure> inline bool
   swap_pred<Dir, out_iterator, Closure>::operator()(typename out_iterator::value_type const &lhs, typename out_iterator::value_type const &rhs) const noexcept(noexcept(std::declval<typename Closure::argument_type>().comp(lhs, rhs))) {
      return (dir==direction::ascending && !arg.comp(lhs, rhs)) || (dir==direction::descending && arg.comp(lhs, rhs));
   }
 
   template<class Iter, class operation_type, direction LHSDir, direction RHSDir, class Dummy> inline void merge_final_sorter<Iter, operation_type, LHSDir, RHSDir, Dummy>::process(Dummy const &, out_iterator const begin, out_iterator const end, operation_type const &fn) noexcept(false) {
      typedef std::reverse_iterator<out_iterator> rev_in;
 
      const out_sz_t size_of_portion=std::distance(begin, end);
      const out_sz_t half_size_of_portion=size_of_portion>>1;
      const out_iterator middle(std::next(begin, half_size_of_portion));
      std::vector<typename out_iterator::value_type> out_colln;
      out_colln.reserve(size_of_portion);
   // TODO testing performance...   std::merge(begin, middle, rev_in(end), rev_in(middle), begin, arg3_type(swapper_t(fn)));
      std::merge(begin, middle, rev_in(end), rev_in(middle), std::back_inserter(out_colln), arg3_type(swapper_t(fn)));
      if (rhs_dir==direction::ascending) {
         std::move(out_colln.begin(), out_colln.end(), begin);
      } else {
         std::move(rev_in(out_colln.end()), rev_in(out_colln.begin()), begin);
      }
   }
 
   template<class Conts, typename Comp>
   template<direction LHSDir, direction RHSDir, template<class, class, direction, direction, class> class FinalSort>
   class batchers_bitonic_merge_reduce<Conts, Comp>::merge {
   public:
      static constexpr direction lhs_dir=LHSDir;   ///< The general direction of the comparator.
      static constexpr direction rhs_dir=RHSDir;   ///< The direction that the rhs of the sort should be flipped, because for merges the sequence is bitonic, so the rhs sub-ranges need to be flipped.
      typedef swap_pred<lhs_dir, out_iterator, operation_type> swapper_t;
      typedef boost::function<void (out_iterator, out_iterator, std::binary_negate<swapper_t>)> sort_fn_t;
 
      /**
         To assist in allowing compile-time computation of the algorithmic order of the threading model.
      */
      static constexpr ppd::generic_traits::memory_access_modes memory_access_mode=thread_pool_type::template swap_ranges_t<typename containers_type::output_t::container_type, swapper_t>::memory_access_mode;
 
      constexpr merge(out_iterator b, out_iterator e, operation_type const &f, sort_fn_t const &sfn, cliques::element_type const cl) noexcept(true) FORCE_INLINE
      : sorter(sfn), begin(b), end(e), fn(f), clique(cl) {
      }
      virtual ~merge() FORCE_INLINE {}
 
      /**
         \todo JMG: this needs extending to non-even sized collections...
      */
      void __fastcall process() const {
         typedef merge<lhs_dir, lhs_dir, FinalSort> lhs_sub_merge;
         typedef merge<lhs_dir, lhs_dir==direction::ascending ? direction::descending : direction::ascending, FinalSort> rhs_sub_merge;
 
         const out_sz_t size_of_portion=std::distance(begin, end);
         if (
            size_of_portion>3
            && (
               thread_pool_type::size_mode==pool_traits::size_mode_t::infinite
               || clique<fn.input().pool.pool_size()
            )
         ) {
            const out_sz_t half_size_of_portion=size_of_portion>>1;
            const out_iterator middle(std::next(begin, half_size_of_portion));
            const out_iterator ak(middle);
            assert(begin!=middle);
            assert(middle!=end);
            auto const &shuffle=fn.input().pool<<joinable(this, shuffle_str)<<cliques(clique)<<fn.input().pool.template swap_ranges<typename containers_type::output_t::container_type>(begin, middle, ak, swapper_t(fn)); // O(half_size_of_portion/p+log(p))
            *shuffle;
            auto const &merge_lhs=fn.input().pool<<joinable(this, lhs_merge_str)<<lhs_sub_merge(begin, middle, fn, sorter, clique<<1); // O(log(half_size_of_portion))
            auto const &merge_rhs=fn.input().pool<<joinable(this, rhs_merge_str)<<rhs_sub_merge(middle, end, fn, sorter, clique<<1);   // O(log(half_size_of_portion))
            *merge_lhs;
            *merge_rhs;
         } else {
            typedef FinalSort<out_iterator, operation_type, lhs_dir, rhs_dir, sort_fn_t> final_sort;
            final_sort::process(sorter, begin, end, fn);
         }
      }
 
      constexpr bool __fastcall operator<(merge const &) const noexcept(true) FORCE_INLINE {
         return true;
      }
 
   private:
      sort_fn_t const sorter;
      out_iterator const begin;
      out_iterator const end;
      operation_type const &fn;
      cliques::element_type const clique;
   };
 
   template<class Conts, typename Comp> inline void
   batchers_bitonic_merge_reduce<Conts, Comp>::combine() {
      typedef std::reverse_iterator<typename containers_type::input2_t::iterator> rev_in2;
 
      assert((conts.input1.size()+conts.input2.size())==conts.output.size());
 
      auto const &first=fn.input().pool<<joinable(this, combine1_str)<<cliques(clique<<1)<<fn.input().pool.template copy<typename containers_type::input1_t::container_type, typename containers_type::output_t::container_type>(conts.input1.begin(), conts.input1.end(), conts.output.begin());
      const out_iterator middle(std::next(conts.output.begin(), conts.input1.size()));
      auto const &second=fn.input().pool<<joinable(this, combine2_str)<<cliques(clique<<1)<<fn.input().pool.template copy<typename containers_type::input2_t::container_type, typename containers_type::output_t::container_type>(rev_in2(conts.input2.end()), rev_in2(conts.input2.begin()), middle);
      *first;
      *second;
   }
 
   template<class Conts, typename Comp> inline
   batchers_bitonic_merge_reduce<Conts, Comp>::batchers_bitonic_merge_reduce(containers_type &c, operation_type const &w, cliques::element_type const cl) noexcept(true)
   : conts(c), fn(w), clique(cl) {
   }
 
   template<class Conts, typename Comp> inline void
   batchers_bitonic_merge_reduce<Conts, Comp>::process() {
      typedef typename thread_pool_type::template create_direct<init_merger_t> init_merger_creator_t;
 
      if (conts.input1.size()!=conts.input2.size()) {
         info::function fun(
            __LINE__,
            __PRETTY_FUNCTION__,
            typeid(&batchers_bitonic_merge_reduce::process),
            info::function::argument(_T("containers_type"), exception_type::thread_traits::demangle_name(typeid(containers_type)))
         );
         fun.add_args(
            info::function::argument(_T("conts.input1.size()"), conts.input1.size()),
            info::function::argument(_T("conts.input2.size()"), conts.input2.size())
         );
         throw exception_type(_T("Both input collections must be the same size."), std::move(fun), info::revision(_T(LIBJMMCG_VERSION_NUMBER)));
      }
      if ((conts.input1.size()+conts.input2.size())%2) {
         info::function fun(
            __LINE__,
            __PRETTY_FUNCTION__,
            typeid(&batchers_bitonic_merge_reduce::process),
            info::function::argument(_T("containers_type"), exception_type::thread_traits::demangle_name(typeid(containers_type)))
         );
         fun.add_args(
            info::function::argument(_T("conts.input1.size()"), conts.input1.size()),
            info::function::argument(_T("conts.input2.size()"), conts.input2.size())
         );
         throw exception_type(_T("Both input collections must have sizes that are a power of two."), std::move(fun), info::revision(_T(LIBJMMCG_VERSION_NUMBER)));
      }
      combine();
      auto sorter=[](auto b, auto e, auto s) {
         return std::stable_sort<typename sort_fn_t::arg1_type, typename sort_fn_t::arg3_type>(b, e, s);
      };
      typename init_merger_creator_t::closure_t all(init_merger_t(conts.output.begin(), conts.output.end(), fn, sorter, clique), typename pool_traits_type::thread_wk_elem_type::cfg_details_type::params(fn.input().pool.cfg(), this, "init_merger"));
      all.update_edge(thread_pool_type::cfg_type::sequential_edge_annotation);
   // TODO Set start & end times of processing op, so that time-order of nodes in DFG can be seen.
      all.process();
   }
 
   template<typename Conts, class Comp>
   template<direction dir>
   class bitonic_sort_reduce<Conts, Comp>::sort {
   public:
      /**
         Make sure we use the correct algorithm in the merge operation to ensure that the sub-collection is correctly sorted.
 
         \see sort_final_sorter
      */
      typedef typename merge_t::template merge<dir, dir, sort_final_sorter> merge_in_dir_t;
      typedef typename merge_in_dir_t::sort_fn_t merge_sort_fn_t;
 
      /**
         To assist in allowing compile-time computation of the algorithmic order of the threading model.
      */
      static constexpr ppd::generic_traits::memory_access_modes memory_access_mode=merge_in_dir_t::memory_access_mode;
 
      /**
         Complexity: O(log^2(O(m_sfn)/p+log(p)))
      */
      constexpr sort(in_iterator b, in_iterator e, operation_type const &f, cliques::element_type const cl) noexcept(true) FORCE_INLINE
      : begin(b), end(e), fn(f), clique(cl) {
      }
      virtual ~sort() noexcept(true) FORCE_INLINE {}
 
      /**
         \todo JMG: this needs extending to non-even sized collections...
 
         If std::stable_sort() is used then on average this is O(n.log(n)), with enough memory.
      */
      void __fastcall process() const {
         typedef typename operation_type::argument_type::thread_pool_type::joinable joinable;
 
         const typename in_iterator::difference_type size_of_portion=std::distance(begin, end);
         auto merge_sorter=[](auto b, auto e, auto s) {
            return std::stable_sort<typename merge_sort_fn_t::arg1_type, typename merge_sort_fn_t::arg3_type>(b, e, s);
         };
         if (
            size_of_portion>3
            && (
               thread_pool_type::size_mode==pool_traits::size_mode_t::infinite
               || clique<fn.input().pool.pool_size()
            )
         ) {
            const typename in_iterator::difference_type half_size_of_portion=size_of_portion>>1;
            const in_iterator middle(std::next(begin, half_size_of_portion));
            auto const &sort_lhs_ascending=fn.input().pool<<joinable(this, ascending_lhs_str)<<sort<direction::ascending>(begin, middle, fn, clique<<1/* TODO ,  merge_sorter*/);   // O(s(half_size_of_portion))
            auto const &sort_rhs_descending=fn.input().pool<<joinable(this, descending_rhs_str)<<sort<direction::descending>(middle, end, fn, clique<<1/* TODO,  merge_sorter*/);   // O(s(half_size_of_portion))
            *sort_lhs_ascending;
            *sort_rhs_descending;
            auto const &bitonic_merge_all=fn.input().pool<<joinable(this, merge_str)<<merge_in_dir_t(begin, end, fn,  merge_sorter, clique); // O(m(half_size_of_portion))
            *bitonic_merge_all;
         } else {
            merge_sorter(begin, end, typename merge_sort_fn_t::arg3_type(swap_pred<dir, in_iterator, operation_type>(fn)));
         }
      }
 
      constexpr bool __fastcall operator<(sort const &) const noexcept(true) FORCE_INLINE {
         return true;
      }
 
   private:
      in_iterator const begin;
      in_iterator const end;
      operation_type const &fn;
      cliques::element_type const clique;
   };
 
   template<typename Conts, class Comp> inline constexpr
   bitonic_sort_reduce<Conts, Comp>::bitonic_sort_reduce(containers_type &c, operation_type const &op, cliques::element_type const cl) noexcept(true)
   : cont(c), fn(op), clique(cl) {
   }
 
   template<typename Conts, class Comp> inline void
   bitonic_sort_reduce<Conts, Comp>::process() const {
      typedef typename thread_pool_type::template create_direct<init_sorter_t> init_sort_creator_t;
 
      typename init_sort_creator_t::closure_t all(init_sorter_t(cont.input1.begin(), cont.input1.end(), fn, clique), typename pool_traits_type::thread_wk_elem_type::cfg_details_type::params(fn.input().pool.cfg(), this, "init_sorter"));
      all.update_edge(thread_pool_type::cfg_type::sequential_edge_annotation);
   // TODO Set start & end times of processing op, so that time-order of nodes in DFG can be seen.
      all.process();
   }
 
}
 
template<class ArgT, class UniFn, class PT>
struct unary_fun_work_type<ArgT, UniFn, PT>::arg_int_work_type {
   typedef typename unary_fun_work_type::result_type result_type;
 
   /**
      To assist in allowing compile-time computation of the algorithmic order of the threading model.
   */
   static constexpr ppd::generic_traits::memory_access_modes memory_access_mode=ppd::generic_traits::memory_access_modes::crew_memory_access;
 
   explicit constexpr __stdcall arg_int_work_type(argument_type &&a) FORCE_INLINE
   : arg(std::forward<argument_type>(a)) {
   }
   void __fastcall process(result_type &r) FORCE_INLINE {
      arg.process(r.result);
   }
 
   bool __fastcall operator<(arg_int_work_type const &rhs) const FORCE_INLINE {
      return arg<rhs.arg;
   }
   template<class Arg1> constexpr bool __fastcall FORCE_INLINE
   operator<(Arg1 const &) const noexcept(true) {
      return true;
   }
 
private:
   argument_type arg;
};
 
template<class ArgT, class UniFn, class PT>
struct unary_fun_work_type<ArgT, UniFn, PT>::arg_context_t : public sp_counter_type<long, typename pool_type::os_traits::lock_traits> {
   typedef typename pool_type::joinable joinable;
   typedef typename pool_type::template execution_context_stack<arg_int_work_type> execution_context;
 
   /**
      To assist in allowing compile-time computation of the algorithmic order of the threading model.
   */
   static constexpr ppd::generic_traits::memory_access_modes memory_access_mode=(
      execution_context::memory_access_mode==ppd::generic_traits::memory_access_modes::crew_memory_access
      ? ppd::generic_traits::memory_access_modes::crew_memory_access
      : ppd::generic_traits::memory_access_modes::erew_memory_access
   );
 
   const execution_context arg;
 
   __stdcall arg_context_t(argument_type &&a, pool_type &pool) FORCE_INLINE
   : arg(pool<<joinable(this, arg_str)<<arg_int_work_type(std::forward<argument_type>(a))) {
   }
   ~arg_context_t() noexcept(true) FORCE_INLINE {}
};
 
template<class ArgT, class UniFn, class PT> inline
unary_fun_work_type<ArgT, UniFn, PT>::unary_fun_work_type(argument_type &&a, operation_type const &o, pool_type &pool) noexcept(false)
: op(o), arg_cxt(new arg_context_t(std::forward<argument_type>(a), pool)) {
}
 
template<class ArgT, class UniFn, class PT> inline void
unary_fun_work_type<ArgT, UniFn, PT>::process(result_type &r) {
   assert(dynamic_cast<arg_context_t *>(arg_cxt.get().get()));
   r.result=op.operator()(arg_cxt->arg->result);
}
 
template<class ArgT, class UniFn, class PT> inline bool
unary_fun_work_type<ArgT, UniFn, PT>::operator<(unary_fun_work_type const &rhs) const noexcept(true) {
   return *arg_cxt->arg.wk_queue_item()<*rhs.arg_cxt->arg.wk_queue_item();
}
 
template<class ArgT1, class ArgT2, class BinFn, class PT>
template<class Arg>
struct binary_fun_work_type<ArgT1, ArgT2, BinFn, PT>::arg_int_work_type {
   typedef typename binary_fun_work_type::result_type result_type;
   typedef Arg argument_type;
 
   /**
      To assist in allowing compile-time computation of the algorithmic order of the threading model.
   */
   static constexpr ppd::generic_traits::memory_access_modes memory_access_mode=ppd::generic_traits::memory_access_modes::crew_memory_access;
 
   explicit constexpr __stdcall arg_int_work_type(argument_type &&a) FORCE_INLINE
   : arg(std::forward<argument_type>(a)) {
   }
   void __fastcall process(result_type &r) FORCE_INLINE {
      arg.process(r.result);
   }
 
   bool __fastcall operator<(arg_int_work_type const &rhs) const FORCE_INLINE {
      return arg<rhs.arg;
   }
   template<class Arg1> constexpr bool __fastcall FORCE_INLINE
   operator<(Arg1 const &) const noexcept(true) {
      return true;
   }
 
private:
   argument_type arg;
};
 
template<class ArgT1, class ArgT2, class BinFn, class PT>
struct binary_fun_work_type<ArgT1, ArgT2, BinFn, PT>::arg_contexts_t : public sp_counter_type<long, typename pool_type::os_traits::lock_traits> {
   typedef typename pool_type::joinable first_joinable;
   typedef typename pool_type::template execution_context_stack<arg_int_work_type<first_argument_type>> first_execution_context;
   typedef typename pool_type::joinable second_joinable;
   typedef typename pool_type::template execution_context_stack<arg_int_work_type<second_argument_type>> second_execution_context;
 
   /**
      To assist in allowing compile-time computation of the algorithmic order of the threading model.
   */
   static constexpr ppd::generic_traits::memory_access_modes memory_access_mode=(
      first_execution_context::memory_access_mode==ppd::generic_traits::memory_access_modes::crew_memory_access
      && second_execution_context::memory_access_mode==ppd::generic_traits::memory_access_modes::crew_memory_access
      ? ppd::generic_traits::memory_access_modes::crew_memory_access
      : ppd::generic_traits::memory_access_modes::erew_memory_access
   );
 
   const first_execution_context first_arg;
   const second_execution_context second_arg;
 
   __stdcall arg_contexts_t(first_argument_type &&lhs, second_argument_type &&rhs, pool_type &pool)
   : first_arg(pool<<first_joinable(this, lhs_str)<<arg_int_work_type<first_argument_type>(std::forward<first_argument_type>(lhs))),
      second_arg(pool<<second_joinable(this, rhs_str)<<arg_int_work_type<second_argument_type>(std::forward<second_argument_type>(rhs))) {
   }
   ~arg_contexts_t() noexcept(true) FORCE_INLINE {}
};
 
template<class ArgT1, class ArgT2, class BinFn, class PT> inline
binary_fun_work_type<ArgT1, ArgT2, BinFn, PT>::binary_fun_work_type(first_argument_type &&lhs, second_argument_type &&rhs, operation_type const &o, pool_type &pool) noexcept(false)
: op(o), arg_cxts(new arg_contexts_t(std::forward<first_argument_type>(lhs), std::forward<second_argument_type>(rhs), pool)) {
}
 
template<class ArgT1, class ArgT2, class BinFn, class PT> inline void
binary_fun_work_type<ArgT1, ArgT2, BinFn, PT>::process(result_type &r) {
   assert(dynamic_cast<arg_contexts_t *>(arg_cxts.get().get()));
   r.result=op.operator()(arg_cxts->first_arg->result, arg_cxts->second_arg->result);
}
 
template<class ArgT1, class ArgT2, class BinFn, class PT> inline bool
binary_fun_work_type<ArgT1, ArgT2, BinFn, PT>::operator<(binary_fun_work_type const &rhs) const noexcept(true) {
   return *arg_cxts->first_arg.wk_queue_item()<*rhs.arg_cxts->second_arg.wk_queue_item() || (*arg_cxts->first_arg.wk_queue_item()==*rhs.arg_cxts->first_arg.wk_queue_item() && *arg_cxts->second_arg.wk_queue_item()<*rhs.arg_cxts->second_arg.wk_queue_item());
}
 
} } } }