worldtask.h

Go to the documentation of this file.

/*
  This file is part of MADNESS.

  Copyright (C) 2007,2010 Oak Ridge National Laboratory

  This program is free software; you can redistribute it and/or modify
  it under the terms of the GNU General Public License as published by
  the Free Software Foundation; either version 2 of the License, or
  (at your option) any later version.

  This program 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 General Public License for more details.

  You should have received a copy of the GNU General Public License
  along with this program; if not, write to the Free Software
  Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA

  For more information please contact:

  Robert J. Harrison
  Oak Ridge National Laboratory
  One Bethel Valley Road
  P.O. Box 2008, MS-6367

  email: harrisonrj@ornl.gov
  tel:   865-241-3937
  fax:   865-572-0680


  $Id$
*/


#ifndef MADNESS_WORLD_WORLDTASK_H__INCLUDED
#define MADNESS_WORLD_WORLDTASK_H__INCLUDED


// To do:
// a) Redo this ontop of serializable tasks which will remote much of the clutter
//    due to multiple length argument lists.
// b) Stealing which pretty much presume a) has been done

#include <iostream>
#include <madness/world/nodefaults.h>
#include <madness/world/worldrange.h>
#include <madness/world/worldtime.h>
#include <madness/world/taskfn.h>

namespace madness {

    // Forward decls
    class World;
    class WorldTaskQueue;
    template <typename> struct TaskFunction;
    template <typename> struct TaskMemfun;

    namespace detail {

        template <typename ptrT, typename memfnT, typename resT>
        memfnT get_mem_func_ptr(const MemFuncWrapper<ptrT, memfnT, resT>&);
        template <typename, typename> class ForEachRootTask;

        template <typename refT, typename functionT>
        struct TaskHandlerInfo {
            refT ref;               
            functionT func;         
            TaskAttributes attr;    


            TaskHandlerInfo(const refT& ref, functionT func, const TaskAttributes& attr)
                    : ref(ref), func(func),attr(attr) {}
            TaskHandlerInfo() {}


            template <typename Archive>
            void serialize(const Archive& ar) {
                serialize_internal<functionT>(ar);
            }

        private:


            template <typename fnT>
            struct is_func_ptr {
                static const bool value =
                    (std::is_function<typename std::remove_pointer<fnT>::type >::value
                    || std::is_member_function_pointer<fnT>::value);
            };


            template <typename fnT, typename Archive>
            typename enable_if<is_func_ptr<fnT> >::type
            serialize_internal(const Archive& ar) {
                ar & ref & archive::wrap_opaque(func) & attr;
            }

            template <typename fnT, typename Archive>
            typename disable_if<is_func_ptr<fnT> >::type
            serialize_internal(const Archive& ar) {
                ar & ref & func & attr;
            }
        }; // struct TaskHandlerInfo

        template <typename fnT>
        struct function_enabler : public
            lazy_enable_if_c<
                function_traits<fnT>::value || has_result_type<fnT>::value,
                task_result_type<fnT> >
        { };

        template <typename memfnT>
        struct memfunc_enabler : public
            lazy_enable_if<
                std::is_member_function_pointer<memfnT>,
                task_result_type<memfnT> >
        { };

        template <typename T>
        struct DefaultInitPtr {
            static T init() { return NULL; }
        };

        template <typename T>
        struct DefaultInitPtr<std::shared_ptr<T> > {
            static std::shared_ptr<T> init() { return std::shared_ptr<T>(); }
        };

        template <typename ptrT, typename memfnT, typename resT>
        class MemFuncWrapper {
        private:
            ptrT ptr_;
            memfnT memfn_;

            friend memfnT get_mem_func_ptr<ptrT, memfnT, resT>(const MemFuncWrapper<ptrT, memfnT, resT>&);

        public:
            typedef MemFuncWrapper<ptrT, memfnT, void> MemFuncWrapper_;
            typedef resT result_type;
            typedef memfnT memfn_type;

            MemFuncWrapper() : ptr_(DefaultInitPtr<ptrT>::init()), memfn_() { }

            MemFuncWrapper(const MemFuncWrapper_& other) :
                ptr_(other.ptr_), memfn_(other.memfn_)
            { }

            MemFuncWrapper(ptrT ptr, memfnT memfn) :
                ptr_(ptr), memfn_(memfn)
            { }

            MemFuncWrapper_& operator=(const MemFuncWrapper_& other) {
                ptr_ = other.ptr_;
                memfn_ = other.memfn_;
                return *this;
            }

            resT operator()() const {
                MADNESS_ASSERT(ptr_);
                return ((*ptr_).*memfn_)();
            }

            template <typename a1T>
            resT operator()(a1T& a1) const {
                MADNESS_ASSERT(ptr_);
                return ((*ptr_).*memfn_)(a1);
            }

            template <typename a1T, typename a2T>
            resT operator()(a1T& a1, a2T& a2) const {
                MADNESS_ASSERT(ptr_);
                return ((*ptr_).*memfn_)(a1, a2);
            }

            template <typename a1T, typename a2T, typename a3T>
            resT operator()(a1T& a1, a2T& a2, a3T& a3) const {
                MADNESS_ASSERT(ptr_);
                return ((*ptr_).*memfn_)(a1, a2, a3);
            }

            template <typename a1T, typename a2T, typename a3T, typename a4T>
            resT operator()(a1T& a1, a2T& a2, a3T& a3, a4T& a4) const {
                MADNESS_ASSERT(ptr_);
                return ((*ptr_).*memfn_)(a1, a2, a3, a4);
            }

            template <typename a1T, typename a2T, typename a3T, typename a4T,
                typename a5T>
            resT operator()(a1T& a1, a2T& a2, a3T& a3, a4T& a4, a5T& a5) const {
                MADNESS_ASSERT(ptr_);
                return ((*ptr_).*memfn_)(a1, a2, a3, a4, a5);
            }

            template <typename a1T, typename a2T, typename a3T, typename a4T,
                typename a5T, typename a6T>
            resT operator()(a1T& a1, a2T& a2, a3T& a3, a4T& a4, a5T& a5, a6T& a6) const {
                MADNESS_ASSERT(ptr_);
                return ((*ptr_).*memfn_)(a1, a2, a3, a4, a5, a6);
            }

            template <typename a1T, typename a2T, typename a3T, typename a4T,
                typename a5T, typename a6T, typename a7T>
            resT operator()(a1T& a1, a2T& a2, a3T& a3, a4T& a4, a5T& a5, a6T& a6, a7T& a7) const {
                MADNESS_ASSERT(ptr_);
                return ((*ptr_).*memfn_)(a1, a2, a3, a4, a5, a6, a7);
            }

            template <typename a1T, typename a2T, typename a3T, typename a4T,
                typename a5T, typename a6T, typename a7T, typename a8T>
            resT operator()(a1T& a1, a2T& a2, a3T& a3, a4T& a4, a5T& a5, a6T& a6, a7T& a7, a8T& a8) const {
                MADNESS_ASSERT(ptr_);
                return ((*ptr_).*memfn_)(a1, a2, a3, a4, a5, a6, a7, a8);
            }

            template <typename a1T, typename a2T, typename a3T, typename a4T,
                typename a5T, typename a6T, typename a7T, typename a8T, typename a9T>
            resT operator()(a1T& a1, a2T& a2, a3T& a3, a4T& a4, a5T& a5, a6T& a6, a7T& a7, a8T& a8, a9T& a9) const {
                MADNESS_ASSERT(ptr_);
                return ((*ptr_).*memfn_)(a1, a2, a3, a4, a5, a6, a7, a8, a9);
            }

            template <typename Archive>
            void serialize(const Archive& ar) {
                ar & ptr_ & memfn_;
            }

            friend memfnT get_mem_func_ptr(const MemFuncWrapper_& wrapper) {
                return wrapper.memfn_;
            }
        }; // class MemFuncWrapper

        template <typename ptrT, typename memfnT>
        class MemFuncWrapper<ptrT, memfnT, void> {
        private:
            ptrT ptr_;
            memfnT memfn_;

            friend memfnT get_mem_func_ptr<ptrT, memfnT, void>(const MemFuncWrapper<ptrT, memfnT, void>&);

        public:
            typedef MemFuncWrapper<ptrT, memfnT, void> MemFuncWrapper_;
            typedef void result_type;
            typedef memfnT memfn_type;

            MemFuncWrapper() : ptr_(DefaultInitPtr<ptrT>::init()), memfn_() { }

            MemFuncWrapper(const MemFuncWrapper_& other) :
                ptr_(other.ptr_), memfn_(other.memfn_)
            { }

            MemFuncWrapper(ptrT ptr, memfnT memfn) :
                ptr_(ptr), memfn_(memfn)
            { }

            MemFuncWrapper_& operator=(const MemFuncWrapper_& other) {
                ptr_ = other.ptr_;
                memfn_ = other.memfn_;
                return *this;
            }

            void operator()() const {
                MADNESS_ASSERT(ptr_);
                ((*ptr_).*memfn_)();
            }

            template <typename a1T>
            void operator()(a1T& a1) const {
                MADNESS_ASSERT(ptr_);
                ((*ptr_).*memfn_)(a1);
            }

            template <typename a1T, typename a2T>
            void operator()(a1T& a1, a2T& a2) const {
                MADNESS_ASSERT(ptr_);
                ((*ptr_).*memfn_)(a1, a2);
            }

            template <typename a1T, typename a2T, typename a3T>
            void operator()(a1T& a1, a2T& a2, a3T& a3) const {
                MADNESS_ASSERT(ptr_);
                ((*ptr_).*memfn_)(a1, a2, a3);
            }

            template <typename a1T, typename a2T, typename a3T, typename a4T>
            void operator()(a1T& a1, a2T& a2, a3T& a3, a4T& a4) const {
                MADNESS_ASSERT(ptr_);
                ((*ptr_).*memfn_)(a1, a2, a3, a4);
            }

            template <typename a1T, typename a2T, typename a3T, typename a4T,
                typename a5T>
            void operator()(a1T& a1, a2T& a2, a3T& a3, a4T& a4, a5T& a5) const {
                MADNESS_ASSERT(ptr_);
                ((*ptr_).*memfn_)(a1, a2, a3, a4, a5);
            }

            template <typename a1T, typename a2T, typename a3T, typename a4T,
                typename a5T, typename a6T>
            void operator()(a1T& a1, a2T& a2, a3T& a3, a4T& a4, a5T& a5, a6T& a6) const {
                MADNESS_ASSERT(ptr_);
                ((*ptr_).*memfn_)(a1, a2, a3, a4, a5, a6);
            }

            template <typename a1T, typename a2T, typename a3T, typename a4T,
                typename a5T, typename a6T, typename a7T>
            void operator()(a1T& a1, a2T& a2, a3T& a3, a4T& a4, a5T& a5, a6T& a6, a7T& a7) const {
                MADNESS_ASSERT(ptr_);
                ((*ptr_).*memfn_)(a1, a2, a3, a4, a5, a6, a7);
            }

            template <typename a1T, typename a2T, typename a3T, typename a4T,
                typename a5T, typename a6T, typename a7T, typename a8T>
            void operator()(a1T& a1, a2T& a2, a3T& a3, a4T& a4, a5T& a5, a6T& a6, a7T& a7, a8T& a8) const {
                MADNESS_ASSERT(ptr_);
                ((*ptr_).*memfn_)(a1, a2, a3, a4, a5, a6, a7, a8);
            }

            template <typename a1T, typename a2T, typename a3T, typename a4T,
                typename a5T, typename a6T, typename a7T, typename a8T, typename a9T>
            void operator()(a1T& a1, a2T& a2, a3T& a3, a4T& a4, a5T& a5, a6T& a6, a7T& a7, a8T& a8, a9T& a9) const {
                MADNESS_ASSERT(ptr_);
                ((*ptr_).*memfn_)(a1, a2, a3, a4, a5, a6, a7, a8, a9);
            }

            template <typename Archive>
            void serialize(const Archive& ar) {
                ar & ptr_ & memfn_;
            }

        }; // class MemFuncWrapper<ptrT, memfnT, void>

        template <typename objT, typename memfnT>
        MemFuncWrapper<objT*, memfnT, typename result_of<memfnT>::type>
        wrap_mem_fn(objT& obj, memfnT memfn) {
            return MemFuncWrapper<objT*, memfnT,
                    typename memfunc_traits<memfnT>::result_type>(& obj, memfn);
        }

        template <typename objT, typename memfnT>
        MemFuncWrapper<objT*, memfnT, typename result_of<memfnT>::type>
        wrap_mem_fn(objT* obj, memfnT memfn) {
            return MemFuncWrapper<objT*, memfnT,
                    typename memfunc_traits<memfnT>::result_type>(obj, memfn);
        }

        template <typename objT, typename memfnT>
        MemFuncWrapper<std::shared_ptr<objT>, memfnT, typename result_of<memfnT>::type>
        wrap_mem_fn(const std::shared_ptr<objT>& obj, memfnT memfn) {
            return MemFuncWrapper<std::shared_ptr<objT>, memfnT,
                    typename memfunc_traits<memfnT>::result_type>(obj, memfn);
        }

        template <typename objT, typename memfnT>
        MemFuncWrapper<std::shared_ptr<objT>, memfnT, typename result_of<memfnT>::type>
        wrap_mem_fn(std::shared_ptr<objT>& obj, memfnT memfn) {
            return MemFuncWrapper<std::shared_ptr<objT>, memfnT,
                    typename memfunc_traits<memfnT>::result_type>(obj, memfn);
        }

        template <typename ptrT, typename memfnT, typename resT>
        memfnT get_mem_func_ptr(const MemFuncWrapper<ptrT, memfnT, resT>& wrapper) {
            return wrapper.memfn_;
        }

    }  // namespace detail


    class WorldTaskQueue : public CallbackInterface, private NO_DEFAULTS {
        friend class TaskInterface;
    private:
        World& world;              
        const ProcessID me;        
        AtomicInt nregistered;     

        void notify() { nregistered--; }

        // Used in reduce kernel
        template <typename resultT, typename opT>
        static resultT sum(const resultT& left, const resultT& right, const opT& op) {
            //std::cout << " REDUCE SUM " << left << " " << right << std::endl;
            return op(left,right);
        }

        template <typename taskT>
        static void spawn_remote_task_handler(const AmArg& arg) {
            MADNESS_ASSERT(taskT::arity <= 9u);

            // Get task info and arguments form active message

            detail::TaskHandlerInfo<typename taskT::futureT::remote_refT,
                    typename taskT::functionT> info;

            archive::BufferInputArchive input_arch = arg & info;

            // Construct task
            taskT* task = new taskT(typename taskT::futureT(info.ref),
                    info.func, info.attr, input_arch);

            // Add task to queue
            arg.get_world()->taskq.add(task);
        }

        template <typename T>
        inline const T& am_arg(const Future<T>& f) {
            MADNESS_ASSERT(f.probe());
            return f.get();
        }

        template <typename T> inline const T& am_arg(const T& t) { return t; }

        typedef detail::voidT voidT;

        template <typename taskT, typename fnT, typename a1T, typename a2T, typename a3T,
                typename a4T, typename a5T, typename a6T, typename a7T,
                typename a8T, typename a9T>
        inline typename taskT::futureT
        send_task(ProcessID where, fnT fn, const a1T& a1,
                const a2T& a2, const a3T& a3, const a4T& a4, const a5T& a5,
                const a6T& a6, const a7T& a7, const a8T& a8, const a9T& a9,
                const TaskAttributes& attr)
        {
            typename taskT::futureT result;
            typedef detail::TaskHandlerInfo<typename taskT::futureT::remote_refT, typename taskT::functionT> infoT;
            world.am.send(where, & WorldTaskQueue::template spawn_remote_task_handler<taskT>,
                    new_am_arg(infoT(result.remote_ref(world), fn, attr),
                    a1, a2, a3, a4, a5, a6, a7, a8, a9));

            return result;
        }


    public:
        WorldTaskQueue(World& world);

        size_t size() const { return nregistered; }



        void add(TaskInterface* t)  {
            nregistered++;

            t->set_info(&world, this);       // Stuff info

            if (t->ndep() == 0) {
                ThreadPool::add(t); // If no dependencies directly submit
            } else {
                // With dependencies must use the callback to avoid race condition
                t->register_submit_callback();
                //t->dec();
            }
        }

        template <typename fnT, typename a1T, typename a2T, typename a3T,
            typename a4T, typename a5T, typename a6T, typename a7T, typename a8T,
            typename a9T>
        typename TaskFn<fnT, a1T, a2T, a3T, a4T, a5T, a6T, a7T, a8T, a9T>::futureT
        add(TaskFn<fnT, a1T, a2T, a3T, a4T, a5T, a6T, a7T, a8T, a9T>* t) {
            typename TaskFn<fnT, a1T, a2T, a3T, a4T, a5T, a6T, a7T, a8T, a9T>::futureT
                res(t->result());
            add(static_cast<TaskInterface*>(t));
            return res;
        }


        template <typename resultT, typename rangeT, typename opT>
        Future<resultT> reduce(const rangeT& range, const opT& op) {
            if (range.size() <= range.get_chunksize()) {
                resultT sum = resultT();
                for (typename rangeT::iterator it=range.begin(); it != range.end(); ++it) sum = op(sum,op(it));
                return Future<resultT>(sum);
            } else {
                rangeT left = range;
                rangeT right(left,Split());

                Future<resultT>  leftsum = add(*this, &WorldTaskQueue::reduce<resultT,rangeT,opT>, left,  op);
                Future<resultT> rightsum = add(*this, &WorldTaskQueue::reduce<resultT,rangeT,opT>, right, op);
                return add(&WorldTaskQueue::sum<resultT,opT>, leftsum, rightsum, op);
            }
        }


        template <typename rangeT, typename opT>
        Future<bool> for_each(const rangeT& range, const opT& op) {
            detail::ForEachRootTask<rangeT, opT>* for_each_root =
                    new detail::ForEachRootTask<rangeT, opT>(world, range, op);
            Future<bool> result = for_each_root->result();
            add(for_each_root);
            return result;
        }



        template <typename fnT>
        typename detail::function_enabler<fnT>::type
        add(fnT fn, const TaskAttributes& attr=TaskAttributes()) {
            typedef TaskFn<fnT> taskT;
            return add(new taskT(typename taskT::futureT(),
                    fn, attr));
        }

        template <typename fnT, typename a1T>
        typename detail::function_enabler<fnT>::type
        add(fnT fn, const a1T& a1, const TaskAttributes& attr=TaskAttributes()) {
            typedef TaskFn<fnT, a1T> taskT;
            return add(new taskT(typename taskT::futureT(),
                    fn, a1, attr));
        }

        template <typename fnT, typename a1T, typename a2T>
        typename detail::function_enabler<fnT>::type
        add(fnT fn, const a1T& a1, const a2T& a2, const TaskAttributes& attr=TaskAttributes()) {
            typedef TaskFn<fnT, a1T, a2T> taskT;
            return add(new taskT(typename taskT::futureT(),
                    fn, a1, a2, attr));
        }

        template <typename fnT, typename a1T, typename a2T, typename a3T>
        typename detail::function_enabler<fnT>::type
        add(fnT fn, const a1T& a1, const a2T& a2, const a3T& a3,
            const TaskAttributes& attr=TaskAttributes())
        {
            typedef TaskFn<fnT, a1T, a2T, a3T> taskT;
            return add(new taskT(typename taskT::futureT(),
                    fn, a1, a2, a3, attr));
        }

        template <typename fnT, typename a1T, typename a2T, typename a3T, typename a4T>
        typename detail::function_enabler<fnT>::type
        add(fnT fn, const a1T& a1, const a2T& a2, const a3T& a3, const a4T& a4,
            const TaskAttributes& attr=TaskAttributes())
        {
            typedef TaskFn<fnT, a1T, a2T, a3T, a4T> taskT;
            return add(new taskT(typename taskT::futureT(),
                    fn, a1, a2, a3, a4, attr));
        }

        template <typename fnT, typename a1T, typename a2T, typename a3T, typename a4T,
            typename a5T>
        typename detail::function_enabler<fnT>::type
        add(fnT fn, const a1T& a1, const a2T& a2, const a3T& a3, const a4T& a4,
            const a5T& a5, const TaskAttributes& attr=TaskAttributes())
        {
            typedef TaskFn<fnT, a1T, a2T, a3T, a4T, a5T> taskT;
            return add(new taskT(typename taskT::futureT(),
                    fn, a1, a2, a3, a4, a5, attr));
        }

        template <typename fnT, typename a1T, typename a2T, typename a3T, typename a4T,
            typename a5T, typename a6T>
        typename detail::function_enabler<fnT>::type
        add(fnT fn, const a1T& a1, const a2T& a2, const a3T& a3, const a4T& a4,
            const a5T& a5, const a6T& a6, const TaskAttributes& attr=TaskAttributes())
        {
            typedef TaskFn<fnT, a1T, a2T, a3T, a4T, a5T, a6T> taskT;
            return add(new taskT(typename taskT::futureT(),
                    fn, a1, a2, a3, a4, a5, a6, attr));
        }

        template <typename fnT, typename a1T, typename a2T, typename a3T, typename a4T,
            typename a5T, typename a6T, typename a7T>
        typename detail::function_enabler<fnT>::type
        add(fnT fn, const a1T& a1, const a2T& a2, const a3T& a3, const a4T& a4,
            const a5T& a5, const a6T& a6, const a7T& a7,
            const TaskAttributes& attr=TaskAttributes())
        {
            typedef TaskFn<fnT, a1T, a2T, a3T, a4T, a5T, a6T, a7T> taskT;
            return add(new taskT(typename taskT::futureT(),
                    fn, a1, a2, a3, a4, a5, a6, a7, attr));
        }

        template <typename fnT, typename a1T, typename a2T, typename a3T, typename a4T,
            typename a5T, typename a6T, typename a7T, typename a8T>
        typename detail::function_enabler<fnT>::type
        add(fnT fn, const a1T& a1, const a2T& a2, const a3T& a3, const a4T& a4,
            const a5T& a5, const a6T& a6, const a7T& a7, const a8T& a8,
            const TaskAttributes& attr=TaskAttributes())
        {
            typedef TaskFn<fnT, a1T, a2T, a3T, a4T, a5T, a6T, a7T, a8T> taskT;
            return add(new taskT(typename taskT::futureT(),
                    fn, a1, a2, a3, a4, a5, a6, a7, a8, attr));
        }


        template <typename fnT, typename a1T, typename a2T, typename a3T, typename a4T,
            typename a5T, typename a6T, typename a7T, typename a8T, typename a9T>
        typename detail::function_enabler<fnT>::type
        add(fnT fn, const a1T& a1, const a2T& a2, const a3T& a3, const a4T& a4,
            const a5T& a5, const a6T& a6, const a7T& a7, const a8T& a8, const a9T& a9,
            const TaskAttributes& attr=TaskAttributes())
        {
            typedef TaskFn<fnT, a1T, a2T, a3T, a4T, a5T, a6T, a7T, a8T, a9T> taskT;
            return add(new taskT(typename taskT::futureT(),
                    fn, a1, a2, a3, a4, a5, a6, a7, a8, a9, attr));
        }



        template <typename fnT>
        typename detail::function_enabler<fnT>::type
        add(ProcessID dest, fnT fn, const TaskAttributes& attr=TaskAttributes())
        {
            typedef TaskFn<fnT> taskT;
            if(dest == me)
                return add(fn, attr);
            else
                return send_task<taskT>(dest, fn, voidT::value, voidT::value,
                        voidT::value, voidT::value, voidT::value, voidT::value,
                        voidT::value, voidT::value, voidT::value, attr);
        }



        template <typename fnT, typename a1T>
        typename detail::function_enabler<fnT>::type
        add(ProcessID dest, fnT fn, const a1T& a1, const TaskAttributes& attr=TaskAttributes())
        {
            typedef TaskFn<fnT, a1T> taskT;
            if(dest == me)
                return add(fn, a1, attr);
            else
                return send_task<taskT>(dest, fn, am_arg(a1), voidT::value,
                        voidT::value, voidT::value, voidT::value, voidT::value,
                        voidT::value, voidT::value, voidT::value, attr);
        }


        template <typename fnT, typename a1T, typename a2T>
        typename detail::function_enabler<fnT>::type
        add(ProcessID dest, fnT fn, const a1T& a1, const a2T& a2,
                const TaskAttributes& attr=TaskAttributes())
        {
            typedef TaskFn<fnT, a1T, a2T> taskT;
            if(dest == me)
                return add(fn, a1, a2, attr);
            else
                return send_task<taskT>(dest, fn, am_arg(a1), am_arg(a2),
                        voidT::value, voidT::value, voidT::value, voidT::value,
                        voidT::value, voidT::value, voidT::value, attr);
        }


        template <typename fnT, typename a1T, typename a2T, typename a3T>
        typename detail::function_enabler<fnT>::type
        add(ProcessID dest, fnT fn, const a1T& a1, const a2T& a2, const a3T& a3,
                const TaskAttributes& attr=TaskAttributes())
        {
            typedef TaskFn<fnT, a1T, a2T, a3T> taskT;
            if(dest == me)
                return add(fn, a1, a2, a3, attr);
            else
                return send_task<taskT>(dest, fn, am_arg(a1), am_arg(a2),
                        am_arg(a3), voidT::value, voidT::value, voidT::value,
                        voidT::value, voidT::value, voidT::value, attr);
        }


        template <typename fnT, typename a1T, typename a2T, typename a3T,
                typename a4T>
        typename detail::function_enabler<fnT>::type
        add(ProcessID dest, fnT fn, const a1T& a1, const a2T& a2, const a3T& a3,
                const a4T& a4, const TaskAttributes& attr=TaskAttributes())
        {
            typedef TaskFn<fnT, a1T, a2T, a3T, a4T> taskT;
            if(dest == me)
                return add(fn, a1, a2, a3, a4, attr);
            else
                return send_task<taskT>(dest, fn, am_arg(a1), am_arg(a2),
                        am_arg(a3), am_arg(a4), voidT::value, voidT::value,
                        voidT::value, voidT::value, voidT::value, attr);
        }


        template <typename fnT, typename a1T, typename a2T, typename a3T,
                typename a4T, typename a5T>
        typename detail::function_enabler<fnT>::type
        add(ProcessID dest, fnT fn, const a1T& a1, const a2T& a2, const a3T& a3,
                const a4T& a4, const a5T& a5, const TaskAttributes& attr=TaskAttributes())
        {
            typedef TaskFn<fnT, a1T, a2T, a3T, a4T, a5T> taskT;
            if(dest == me)
                return add(fn, a1, a2, a3, a4, a5, attr);
            else
                return send_task<taskT>(dest, fn, am_arg(a1), am_arg(a2),
                        am_arg(a3), am_arg(a4), am_arg(a5), voidT::value,
                        voidT::value, voidT::value, voidT::value, attr);
        }


        template <typename fnT, typename a1T, typename a2T, typename a3T,
                typename a4T, typename a5T, typename a6T>
        typename detail::function_enabler<fnT>::type
        add(ProcessID dest, fnT fn, const a1T& a1, const a2T& a2, const a3T& a3,
                const a4T& a4, const a5T& a5, const a6T& a6,
                const TaskAttributes& attr=TaskAttributes())
        {
            typedef TaskFn<fnT, a1T, a2T, a3T, a4T, a5T, a6T> taskT;
            if(dest == me)
                return add(fn, a1, a2, a3, a4, a5, a6, attr);
            else
                return send_task<taskT>(dest, fn, am_arg(a1), am_arg(a2),
                        am_arg(a3), am_arg(a4), am_arg(a5), am_arg(a6),
                        voidT::value, voidT::value, voidT::value, attr);
        }


        template <typename fnT, typename a1T, typename a2T, typename a3T,
                typename a4T, typename a5T, typename a6T, typename a7T>
        typename detail::function_enabler<fnT>::type
        add(ProcessID dest, fnT fn, const a1T& a1, const a2T& a2, const a3T& a3,
                const a4T& a4, const a5T& a5, const a6T& a6, const a7T& a7,
                const TaskAttributes& attr=TaskAttributes())
        {
            typedef TaskFn<fnT, a1T, a2T, a3T, a4T, a5T, a6T, a7T> taskT;
            if(dest == me)
                return add(fn, a1, a2, a3, a4, a5, a6, a7, attr);
            else
                return send_task<taskT>(dest, fn, am_arg(a1), am_arg(a2),
                        am_arg(a3), am_arg(a4), am_arg(a5), am_arg(a6),
                        am_arg(a7), voidT::value, voidT::value, attr);
        }


        template <typename fnT, typename a1T, typename a2T, typename a3T,
                typename a4T, typename a5T, typename a6T, typename a7T,
                typename a8T>
        typename detail::function_enabler<fnT>::type
        add(ProcessID dest, fnT fn, const a1T& a1, const a2T& a2, const a3T& a3,
                const a4T& a4, const a5T& a5, const a6T& a6, const a7T& a7,
                const a8T& a8, const TaskAttributes& attr=TaskAttributes())
        {
            typedef TaskFn<fnT, a1T, a2T, a3T, a4T, a5T, a6T, a7T, a8T> taskT;
            if(dest == me)
                return add(fn, a1, a2, a3, a4, a5, a6, a7, a8, attr);
            else
                return send_task<taskT>(dest, fn, am_arg(a1), am_arg(a2),
                        am_arg(a3), am_arg(a4), am_arg(a5), am_arg(a6),
                        am_arg(a7), am_arg(a8), voidT::value, attr);
        }


        template <typename fnT, typename a1T, typename a2T, typename a3T,
                typename a4T, typename a5T, typename a6T, typename a7T,
                typename a8T, typename a9T>
        typename detail::function_enabler<fnT>::type
        add(ProcessID dest, fnT fn, const a1T& a1, const a2T& a2, const a3T& a3,
                const a4T& a4, const a5T& a5, const a6T& a6, const a7T& a7,
                const a8T& a8, const a9T& a9, const TaskAttributes& attr=TaskAttributes())
        {
            typedef TaskFn<fnT, a1T, a2T, a3T, a4T, a5T, a6T, a7T, a8T, a9T> taskT;
            if(dest == me)
                return add(fn, a1, a2, a3, a4, a5, a6, a7, a8, a9, attr);
            else
                return send_task<taskT>(dest, fn, am_arg(a1), am_arg(a2),
                        am_arg(a3), am_arg(a4), am_arg(a5), am_arg(a6),
                        am_arg(a7), am_arg(a8), am_arg(a9), attr);
        }

        template <typename objT, typename memfnT>
        typename detail::memfunc_enabler<memfnT>::type
        add(objT& obj, memfnT memfun, const TaskAttributes& attr=TaskAttributes())
        { return add(detail::wrap_mem_fn(obj,memfun),attr); }

        template <typename objT, typename memfnT, typename a1T>
        typename detail::memfunc_enabler<memfnT>::type
        add(objT& obj, memfnT memfun, const a1T& a1,
                const TaskAttributes& attr=TaskAttributes())
        { return add(detail::wrap_mem_fn(obj,memfun),a1,attr); }

        template <typename objT, typename memfnT, typename a1T, typename a2T>
        typename detail::memfunc_enabler<memfnT>::type
        add(objT& obj, memfnT memfun, const a1T& a1, const a2T& a2,
                const TaskAttributes& attr=TaskAttributes())
        { return add(detail::wrap_mem_fn(obj,memfun),a1,a2,attr); }

        template <typename objT, typename memfnT, typename a1T, typename a2T,
            typename a3T>
        typename detail::memfunc_enabler<memfnT>::type
        add(objT& obj, memfnT memfun, const a1T& a1, const a2T& a2, const a3T& a3,
                const TaskAttributes& attr=TaskAttributes())
        { return add(detail::wrap_mem_fn(obj,memfun),a1,a2,a3,attr); }

        template <typename objT, typename memfnT, typename a1T, typename a2T,
            typename a3T, typename a4T>
        typename detail::memfunc_enabler<memfnT>::type
        add(objT& obj, memfnT memfun, const a1T& a1, const a2T& a2, const a3T& a3,
                const a4T& a4, const TaskAttributes& attr=TaskAttributes())
        { return add(detail::wrap_mem_fn(obj,memfun),a1,a2,a3,a4,attr); }

        template <typename objT, typename memfnT, typename a1T, typename a2T,
            typename a3T, typename a4T, typename a5T>
        typename detail::memfunc_enabler<memfnT>::type
        add(objT& obj, memfnT memfun, const a1T& a1, const a2T& a2, const a3T& a3,
                const a4T& a4, const a5T& a5, const TaskAttributes attr=TaskAttributes())
        { return add(detail::wrap_mem_fn(obj,memfun),a1,a2,a3,a4,a5,attr); }

        template <typename objT, typename memfnT, typename a1T, typename a2T,
            typename a3T, typename a4T, typename a5T, typename a6T>
        typename detail::memfunc_enabler<memfnT>::type
        add(objT& obj, memfnT memfun, const a1T& a1, const a2T& a2, const a3T& a3,
                const a4T& a4, const a5T& a5, const a6T& a6,
                const TaskAttributes& attr=TaskAttributes())
        { return add(detail::wrap_mem_fn(obj,memfun),a1,a2,a3,a4,a5,a6,attr); }

        template <typename objT, typename memfnT, typename a1T, typename a2T,
            typename a3T, typename a4T, typename a5T, typename a6T, typename a7T>
        typename detail::memfunc_enabler<memfnT>::type
        add(objT& obj, memfnT memfun, const a1T& a1, const a2T& a2, const a3T& a3,
                const a4T& a4, const a5T& a5, const a6T& a6, const a7T& a7,
                const TaskAttributes& attr=TaskAttributes())
        { return add(detail::wrap_mem_fn(obj,memfun),a1,a2,a3,a4,a5,a6,a7,attr); }

        template <typename objT, typename memfnT, typename a1T, typename a2T,
            typename a3T, typename a4T, typename a5T, typename a6T, typename a7T,
            typename a8T>
        typename detail::memfunc_enabler<memfnT>::type
        add(objT& obj, memfnT memfun, const a1T& a1, const a2T& a2, const a3T& a3,
                const a4T& a4, const a5T& a5, const a6T& a6, const a7T& a7,
                const a8T& a8, const TaskAttributes attr=TaskAttributes())
        { return add(detail::wrap_mem_fn(obj,memfun),a1,a2,a3,a4,a5,a6,a7,a8,attr); }

        template <typename objT, typename memfnT, typename a1T, typename a2T,
            typename a3T, typename a4T, typename a5T, typename a6T, typename a7T,
            typename a8T, typename a9T>
        typename detail::memfunc_enabler<memfnT>::type
        add(objT& obj, memfnT memfun, const a1T& a1, const a2T& a2, const a3T& a3,
                const a4T& a4, const a5T& a5, const a6T& a6, const a7T& a7,
                const a8T& a8, const a9T& a9, const TaskAttributes attr=TaskAttributes())
        { return add(detail::wrap_mem_fn(obj,memfun),a1,a2,a3,a4,a5,a6,a7,a8,a9,attr); }

        template <typename objT, typename memfnT>
        typename detail::memfunc_enabler<memfnT>::type
        add(objT* obj, memfnT memfun, const TaskAttributes& attr=TaskAttributes())
        { return add(detail::wrap_mem_fn(obj,memfun),attr); }

        template <typename objT, typename memfnT, typename a1T>
        typename detail::memfunc_enabler<memfnT>::type
        add(objT* obj, memfnT memfun, const a1T& a1,
                const TaskAttributes& attr=TaskAttributes())
        { return add(detail::wrap_mem_fn(obj,memfun),a1,attr); }

        template <typename objT, typename memfnT, typename a1T, typename a2T>
        typename detail::memfunc_enabler<memfnT>::type
        add(objT* obj, memfnT memfun, const a1T& a1, const a2T& a2,
                const TaskAttributes& attr=TaskAttributes())
        { return add(detail::wrap_mem_fn(obj,memfun),a1,a2,attr); }

        template <typename objT, typename memfnT, typename a1T, typename a2T,
            typename a3T>
        typename detail::memfunc_enabler<memfnT>::type
        add(objT* obj, memfnT memfun, const a1T& a1, const a2T& a2, const a3T& a3,
                const TaskAttributes& attr=TaskAttributes())
        { return add(detail::wrap_mem_fn(obj,memfun),a1,a2,a3,attr); }

        template <typename objT, typename memfnT, typename a1T, typename a2T,
            typename a3T, typename a4T>
        typename detail::memfunc_enabler<memfnT>::type
        add(objT* obj, memfnT memfun, const a1T& a1, const a2T& a2, const a3T& a3,
                const a4T& a4, const TaskAttributes& attr=TaskAttributes())
        { return add(detail::wrap_mem_fn(obj,memfun),a1,a2,a3,a4,attr); }


        template <typename objT, typename memfnT, typename a1T, typename a2T,
            typename a3T, typename a4T, typename a5T>
        typename detail::memfunc_enabler<memfnT>::type
        add(objT* obj, memfnT memfun, const a1T& a1, const a2T& a2, const a3T& a3,
                const a4T& a4, const a5T& a5, const TaskAttributes attr=TaskAttributes())
        { return add(detail::wrap_mem_fn(obj,memfun),a1,a2,a3,a4,a5,attr); }


        template <typename objT, typename memfnT, typename a1T, typename a2T,
            typename a3T, typename a4T, typename a5T, typename a6T>
        typename detail::memfunc_enabler<memfnT>::type
        add(objT* obj, memfnT memfun, const a1T& a1, const a2T& a2, const a3T& a3,
                const a4T& a4, const a5T& a5, const a6T& a6,
                const TaskAttributes& attr=TaskAttributes())
        { return add(detail::wrap_mem_fn(obj,memfun),a1,a2,a3,a4,a5,a6,attr); }

        template <typename objT, typename memfnT, typename a1T, typename a2T,
            typename a3T, typename a4T, typename a5T, typename a6T, typename a7T>
        typename detail::memfunc_enabler<memfnT>::type
        add(objT* obj, memfnT memfun, const a1T& a1, const a2T& a2, const a3T& a3,
                const a4T& a4, const a5T& a5, const a6T& a6, const a7T& a7,
                const TaskAttributes& attr=TaskAttributes())
        { return add(detail::wrap_mem_fn(obj,memfun),a1,a2,a3,a4,a5,a6,a7,attr); }

        template <typename objT, typename memfnT, typename a1T, typename a2T,
            typename a3T, typename a4T, typename a5T, typename a6T, typename a7T,
            typename a8T>
        typename detail::memfunc_enabler<memfnT>::type
        add(objT* obj, memfnT memfun, const a1T& a1, const a2T& a2, const a3T& a3,
                const a4T& a4, const a5T& a5, const a6T& a6, const a7T& a7,
                const a8T& a8, const TaskAttributes attr=TaskAttributes())
        { return add(detail::wrap_mem_fn(obj,memfun),a1,a2,a3,a4,a5,a6,a7,a8,attr); }

        template <typename objT, typename memfnT, typename a1T, typename a2T,
            typename a3T, typename a4T, typename a5T, typename a6T, typename a7T,
            typename a8T, typename a9T>
        typename detail::memfunc_enabler<memfnT>::type
        add(objT* obj, memfnT memfun, const a1T& a1, const a2T& a2,
                const a3T& a3, const a4T& a4, const a5T& a5, const a6T& a6,
                const a7T& a7, const a8T& a8, const a9T& a9,
                const TaskAttributes attr=TaskAttributes())
        { return add(detail::wrap_mem_fn(obj,memfun),a1,a2,a3,a4,a5,a6,a7,a8,a9,attr); }

    private:

        struct ProbeAllDone {
            WorldTaskQueue* tq;
            ProbeAllDone(WorldTaskQueue* tq) : tq(tq) {}
            bool operator()() const { return (tq->nregistered == 0); }
        };

    public:


        void fence()  {
            try {
                ThreadPool::await(ProbeAllDone(this), true);
            } catch(...) {
                printf("!!MADNESS ERROR: Exception thrown in WorldTaskQueue::fence() with %i pending task(s)\n", int(nregistered));
                throw;
            }
        }
    };

    namespace detail {


        template <typename rangeT, typename opT>
        class ForEachTask : public TaskInterface {
        private:
            rangeT range_; 
            opT op_; 
            ForEachRootTask<rangeT, opT>& root_; 

            // not allowed
            ForEachTask(const ForEachTask<rangeT, opT>&);
            ForEachTask& operator=(const ForEachTask<rangeT, opT>&);

        public:

            ForEachTask(const rangeT range, const opT& op, ForEachRootTask<rangeT, opT>& root) :
                TaskInterface(0, TaskAttributes::hipri()), range_(range), op_(op), root_(root)
            {
                // Increment the master task dependency counter for this task
                root_.inc();
            }

            virtual ~ForEachTask() { }

            virtual void run(const TaskThreadEnv&) {
                // Spawn leaf tasks and split range until it is less than chuncksize
                while(range_.size() > range_.get_chunksize()) {
                    rangeT right(range_,Split());
                    ForEachTask<rangeT,opT>* leaf = new ForEachTask<rangeT,opT>(right, op_, root_);
                    root_.world().taskq.add(leaf);
                }

                // Iterate over the remaining chunck of range and call op_ for each element
                int status = 0;
                for(typename rangeT::iterator it = range_.begin(); it != range_.end();  ++it)
                    if(op_(it))
                        ++status;

                // Notify the root task that this task is done give the status
                root_.complete(status);
            }

        private:

            virtual void get_id(std::pair<void*,unsigned short>& id) const {
                PoolTaskInterface::make_id(id, *this);
            }
        }; // class ForEachTask



        template <typename rangeT, typename opT>
        class ForEachRootTask : public TaskInterface {
        private:
            World& world_; 
            AtomicInt status_; 
            Future<bool> completion_status_; 

        public:


            ForEachRootTask(World& world, const rangeT range, const opT& op) :
                TaskInterface(0, TaskAttributes::hipri()), world_(world)
            {
                status_ = - (range.size());
                // Create the first for each task.
                world_.taskq.add(new ForEachTask<rangeT,opT>(range, op, *this));
            }

            virtual ~ForEachRootTask() { }


            World& world() const { return world_; }


            const Future<bool>& result() const { return completion_status_; }


            void complete(const int status) {
                status_ += status;
                DependencyInterface::notify();
            }


            virtual void run(const TaskThreadEnv&) { completion_status_.set(status_ == 0); }

        private:

            virtual void get_id(std::pair<void*,unsigned short>& id) const {
                PoolTaskInterface::make_id(id, *this);
            }
        }; // class ForEachRootTask


    }  // namespace detail

}


#endif // MADNESS_WORLD_WORLDTASK_H__INCLUDED