worldgop.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_WORLDGOP_H__INCLUDED
#define MADNESS_WORLD_WORLDGOP_H__INCLUDED



#include <madness/world/worldtypes.h>
#include <madness/world/bufar.h>
#include <madness/world/worldfwd.h>
#include <madness/world/deferred_cleanup.h>
#include <madness/world/worldtask.h>
#include <madness/world/group.h>
#include <madness/world/dist_cache.h>


namespace madness {

    // Forward declarations
    class World;
    class WorldAmInterface;
    class WorldTaskQueue;
    namespace detail {

        class DeferredCleanup;

    }  // namespace detail

    template <typename T>
    struct WorldSumOp {
        inline T operator()(const T& a, const T& b) const {
            return a+b;
        }
    };

    template <typename T>
    struct WorldMultOp {
        inline T operator()(const T& a, const T& b) const {
            return a*b;
        }
    };

    template <typename T>
    struct WorldMaxOp {
        inline T operator()(const T& a, const T& b) const {
            return a>b? a : b;
        }
    };

    template <typename T>
    struct WorldAbsMaxOp {
        inline T operator()(const T& a, const T& b) const {
            return abs(a)>abs(b)? abs(a) : abs(b);
        }
    };

    template <typename T>
    struct WorldMinOp {
        inline T operator()(const T& a, const T& b) const {
            return a<b? a : b;
        }
    };

    template <typename T>
    struct WorldAbsMinOp {
        inline T operator()(const T& a, const T& b) const {
            return abs(a)<abs(b)? abs(a) : abs(b);
        }
    };

    template <typename T>
    struct WorldBitAndOp {
        inline T operator()(const T& a, const T& b) const {
            return a & b;
        }
    };

    template <typename T>
    struct WorldBitOrOp {
        inline T operator()(const T& a, const T& b) const {
            return a | b;
        }
    };

    template <typename T>
    struct WorldBitXorOp {
        inline T operator()(const T& a, const T& b) const {
            return a ^ b;
        }
    };

    template <typename T>
    struct WorldLogicAndOp {
        inline T operator()(const T& a, const T& b) const {
            return a && b;
        }
    };

    template <typename T>
    struct WorldLogicOrOp {
        inline T operator()(const T& a, const T& b) const {
            return a || b;
        }
    };



    class WorldGopInterface {
    private:
        World& world_; 
        std::shared_ptr<detail::DeferredCleanup> deferred_; 
        bool debug_; 

        friend class detail::DeferredCleanup;

        // Message tags
        struct PointToPointTag { };
        struct LazySyncTag { };
        struct GroupLazySyncTag { };
        struct BcastTag { };
        struct GroupBcastTag { };
        struct ReduceTag { };
        struct GroupReduceTag { };
        struct AllReduceTag { };
        struct GroupAllReduceTag { };



        template <typename keyT, typename valueT>
        class DelayedSend : public CallbackInterface {
        private:
            World& world_; 
            const ProcessID dest_; 
            const keyT key_; 
            Future<valueT> value_; 

            // Not allowed
            DelayedSend(const DelayedSend<keyT, valueT>&);
            DelayedSend<keyT, valueT>& operator=(const DelayedSend<keyT, valueT>&);

        public:


            DelayedSend(World& world, const ProcessID dest,
                    const keyT& key, const Future<valueT>& value) :
                        world_(world), dest_(dest), key_(key), value_(value)
        { }

            virtual ~DelayedSend() { }


            virtual void notify() {
                MADNESS_ASSERT(value_.probe());
                world_.gop.send_internal(dest_, key_, value_.get());
                delete this;
            }
        }; // class DelayedSend


        template <typename valueT, typename keyT>
        static Future<valueT> recv_internal(const keyT& key) {
            return detail::DistCache<keyT>::template get_cache_value<valueT>(key);
        }


        template <typename keyT, typename valueT>
        typename disable_if<is_future<valueT> >::type
        send_internal(const ProcessID dest, const keyT& key, const valueT& value) const {
            typedef detail::DistCache<keyT> dist_cache;

            if(world_.rank() == dest) {
                // When dest is this process, skip the task and set the future immediately.
                dist_cache::set_cache_value(key, value);
            } else {
                // Spawn a remote task to set the value
                world_.taskq.add(dest, dist_cache::template set_cache_value<valueT>, key,
                        value, TaskAttributes::hipri());
            }
        }


        template <typename keyT, typename valueT>
        void send_internal(ProcessID dest, const keyT& key, const Future<valueT>& value) const {
            typedef detail::DistCache<keyT> dist_cache;

            if(world_.rank() == dest) {
                dist_cache::set_cache_value(key, value);
            } else {
                // The destination is not this node, so send it to the destination.
                if(value.probe()) {
                    // Spawn a remote task to set the value
                    world_.taskq.add(dest, dist_cache::template set_cache_value<valueT>, key,
                            value.get(), TaskAttributes::hipri());
                } else {
                    // The future is not ready, so create a callback object that will
                    // send value to the destination node when it is ready.
                    DelayedSend<keyT, valueT>* delayed_send_callback =
                            new DelayedSend<keyT, valueT>(world_, dest, key, value);
                    const_cast<Future<valueT>&>(value).register_callback(delayed_send_callback);

                }
            }
        }


        template <typename keyT>
        void lazy_sync_parent(const ProcessID parent, const keyT& key,
                const ProcessID, const ProcessID) const
        {
            send_internal(parent, key, key.proc());
        }


        template <typename keyT, typename opT>
        void lazy_sync_children(const ProcessID child0, const ProcessID child1,
                const keyT& key, opT& op, const ProcessID) const
        {
            // Signal children to execute the operation.
            if(child0 != -1)
                send_internal(child0, key, 1);
            if(child1 != -1)
                send_internal(child1, key, 1);

            // Execute the operation on this process.
            op();
        }


        template <typename tagT, typename keyT, typename opT>
        void lazy_sync_internal(const ProcessID parent, const ProcessID child0,
                const ProcessID child1, const keyT& key, const opT& op) const {
            typedef ProcessKey<keyT, tagT> key_type;

            // Get signals from parent and children.
            madness::Future<ProcessID> child0_signal = (child0 != -1 ?
                    recv_internal<ProcessID>(key_type(key, child0)) :
                    madness::Future<ProcessID>(-1));
            madness::Future<ProcessID> child1_signal = (child1 != -1 ?
                    recv_internal<ProcessID>(key_type(key, child1)) :
                    madness::Future<ProcessID>(-1));
            madness::Future<ProcessID> parent_signal = (parent != -1 ?
                    recv_internal<ProcessID>(key_type(key, parent)) :
                    madness::Future<ProcessID>(-1));

            // Construct the task that notifies children to run the operation
            key_type my_key(key, world_.rank());
            typedef void (WorldGopInterface::*lazy_sync_childrenT)(const ProcessID, const ProcessID,
                const key_type&, opT&, const ProcessID) const;
            world_.taskq.add(*this, lazy_sync_childrenT(& WorldGopInterface::template lazy_sync_children<key_type, opT>),
                    child0_signal, child1_signal, my_key, op, parent_signal,
                    TaskAttributes::hipri());

            // Send signal to parent
            if(parent != -1) {
                if(child0_signal.probe() && child1_signal.probe())
                    send_internal(parent, my_key, world_.rank());
                else {
                    typedef void (WorldGopInterface::*lazy_sync_parentT)(const ProcessID, const key_type&,
                        const ProcessID, const ProcessID) const;
                    world_.taskq.add(*this, lazy_sync_parentT(& WorldGopInterface::template lazy_sync_parent<key_type>),
                            parent, my_key, child0_signal, child1_signal,
                            TaskAttributes::hipri());
                }
            }
        }


        template <typename keyT, typename valueT, typename taskfnT>
        static void bcast_handler(const AmArg& arg) {
            // Deserialize message arguments
            taskfnT taskfn;
            keyT key;
            valueT value;
            ProcessID root;

            arg & taskfn & key & value & root;

            // Add task to queue
            arg.get_world()->taskq.add(arg.get_world()->gop, taskfn, key,
                    value, root, TaskAttributes::hipri());
        }

        template <typename keyT, typename valueT, typename taskfnT>
        static void group_bcast_handler(const AmArg& arg) {
            // Deserialize message arguments
            taskfnT taskfn;
            keyT key;
            valueT value;
            ProcessID group_root;
            DistributedID group_key;

            arg & taskfn & key & value & group_root & group_key;

            // Get the local group
            const Future<Group> group = Group::get_group(group_key);

            // Add task to queue
            arg.get_world()->taskq.add(arg.get_world()->gop, taskfn, key, value,
                    group_root, group, TaskAttributes::hipri());
        }



        template <typename keyT, typename valueT>
        void bcast_task(const keyT& key, const valueT& value, const ProcessID root) const {
            typedef void (WorldGopInterface::*taskfnT)(const keyT&, const valueT&,
                    const ProcessID) const;

            // Compute binary tree data
            ProcessID parent = -1, child0 = -1, child1 = -1;
            world_.mpi.binary_tree_info(root, parent, child0, child1);

            // Set the local data, except on the root process
            if(parent != -1)
                detail::DistCache<keyT>::set_cache_value(key, value);

            // Precompute send checks
            const bool send0 = (child0 != -1);
            const bool send1 = (child1 != -1);

            // Send active messages to children
            if(send0 || send1) { // Check that this process has children in the binary tree

                // Get handler function and arguments
                void (*handler)(const AmArg&) =
                        & WorldGopInterface::template bcast_handler<keyT, valueT, taskfnT>;
                AmArg* const args = new_am_arg(
                        & WorldGopInterface::template bcast_task<keyT, valueT>,
                        key, value, root);

                // Send active message to children
                // Note: Only the last message is "managed" so the args buffer
                // can be reused.
                if(send0)
                    world_.am.send(child0, handler, args, RMI::ATTR_ORDERED, !send1);
                if(send1)
                    world_.am.send(child1, handler, args, RMI::ATTR_ORDERED, true);
            }
        }

        template <typename keyT, typename valueT>
        void group_bcast_task(const keyT& key, const valueT& value,
                const ProcessID group_root, const Group& group) const
        {
            typedef void (WorldGopInterface::*taskfnT)(const keyT&, const valueT&,
                    const ProcessID, const Group&) const;

            // Set the local data, except on the root process
            ProcessID parent = -1, child0 = -1, child1 = -1;
            group.make_tree(group_root, parent, child0, child1);

            // Set the local data
            if(parent != -1) {
                detail::DistCache<keyT>::set_cache_value(key, value);
                group.remote_update();
            }

            // Precompute send checks
            const bool send0 = (child0 != -1);
            const bool send1 = (child1 != -1);

            if(send0 || send1) { // Check that this process has children in the binary tree

                // Get handler function and arguments
                void (*handler)(const AmArg&) =
                        & WorldGopInterface::template group_bcast_handler<keyT, valueT, taskfnT>;
                AmArg* const args = new_am_arg(
                        & WorldGopInterface::template group_bcast_task<keyT, valueT>,
                        key, value, group_root, group.id());

                // Send active message to children
                if(send0)
                    world_.am.send(child0, handler, args, RMI::ATTR_ORDERED, !send1);
                if(send1)
                    world_.am.send(child1, handler, args, RMI::ATTR_ORDERED, true);
            }
        }


        template <typename tagT, typename keyT, typename valueT>
        void bcast_internal(const keyT& key, Future<valueT>& value, const ProcessID root) const {
            MADNESS_ASSERT((root >= 0) && (root < world_.size()));
            MADNESS_ASSERT((world_.rank() == root) || (! value.probe()));

            // Add operation tag to key
            typedef TaggedKey<keyT, tagT> key_type;
            const key_type tagged_key(key);

            if(world_.size() > 1) { // Do nothing for the trivial case
                if(world_.rank() == root) {
                    // This process owns the data to be broadcast.

                    // Spawn remote tasks that will set the local cache for this
                    // broadcast on other nodes.
                    if(value.probe())
                        // The value is ready so send it now
                        bcast_task(tagged_key, value.get(), root);
                    else {
                        // The value is not ready so spawn a task to send the
                        // data when it is ready.
                        typedef void (WorldGopInterface::*bcast_taskT)(const key_type&,
                                const valueT&, const ProcessID root) const;
                        world_.taskq.add(*this, bcast_taskT(& WorldGopInterface::template bcast_task<key_type, valueT>),
                                tagged_key, value, root, TaskAttributes::hipri());
                    }
                } else {
                    MADNESS_ASSERT(! value.probe());

                    // Get the broadcast value from local cache
                    detail::DistCache<key_type>::get_cache_value(tagged_key, value);
                }
            }
        }


        template <typename tagT, typename keyT, typename valueT>
        void bcast_internal(const keyT& key, Future<valueT>& value,
                const ProcessID group_root, const Group& group) const
        {
            // Construct the internal broadcast key
            typedef TaggedKey<keyT, tagT> key_type;
            const key_type tagged_key(key);

            if(group.rank() == group_root) {
                // This process owns the data to be broadcast.
                if(value.probe())
                    group_bcast_task(tagged_key, value.get(), group_root, group);
                else {
                    typedef void (WorldGopInterface::*group_bcast_taskT)(const key_type&, const valueT&,
                        const ProcessID, const Group&) const;
                    world_.taskq.add(this, group_bcast_taskT(& WorldGopInterface::template group_bcast_task<key_type, valueT>),
                            tagged_key, value, group_root, group,
                            TaskAttributes::hipri());
                }
            } else {
                MADNESS_ASSERT(! value.probe());

                // This is not the root process, so retrieve the broadcast data
                detail::DistCache<key_type>::get_cache_value(tagged_key, value);

                // Increment local use counter for group
                group.local_update();
            }
        }

        template <typename valueT, typename opT>
        static typename detail::result_of<opT>::type
        reduce_task(const valueT& value, const opT& op) {
            typename detail::result_of<opT>::type result = op();
            op(result, value);
            return result;
        }

        template <typename opT>
        static typename detail::result_of<opT>::type
        reduce_result_task(const std::vector<Future<typename detail::result_of<opT>::type> >& results,
                const opT& op)
        {
            MADNESS_ASSERT(results.size() != 0ul);
            Future<typename detail::result_of<opT>::type> result = results.front();
            for(std::size_t i = 1ul; i < results.size(); ++i)
                op(result.get(), results[i].get());
            return result.get();
        }


        template <typename tagT, typename keyT, typename valueT, typename opT>
        Future<typename detail::result_of<opT>::type>
        reduce_internal(const ProcessID parent, const ProcessID child0,
                const ProcessID child1, const ProcessID root, const keyT& key,
                const valueT& value, const opT& op)
        {
            // Create tagged key
            typedef ProcessKey<keyT, tagT> key_type;
            typedef typename detail::result_of<opT>::type result_type;
            typedef typename remove_future<valueT>::type value_type;
            std::vector<Future<result_type> > results;
            results.reserve(3);

            // Add local data to vector of values to reduce
            results.push_back(world_.taskq.add(WorldGopInterface::template reduce_task<valueT, opT>,
                    value, op, TaskAttributes::hipri()));

            // Reduce child data
            if(child0 != -1)
                results.push_back(recv_internal<result_type>(key_type(key, child0)));
            if(child1 != -1)
                results.push_back(recv_internal<result_type>(key_type(key, child1)));

            // Submit the local reduction task
            Future<result_type> local_result =
                    world_.taskq.add(WorldGopInterface::template reduce_result_task<opT>,
                            results, op, TaskAttributes::hipri());

            // Send reduced value to parent or, if this is the root process, set the
            // result future.
            if(parent == -1)
                return local_result;
            else
                send_internal(parent, key_type(key, world_.rank()), local_result);

            return Future<result_type>::default_initializer();
        }


    public:

        // In the World constructor can ONLY rely on MPI and MPI being initialized
        WorldGopInterface(World& world) :
            world_(world), deferred_(new detail::DeferredCleanup()), debug_(false)
        { }

        ~WorldGopInterface() {
            deferred_->destroy(true);
            deferred_->do_cleanup();
        }


        bool set_debug(bool value) {
            bool status = debug_;
            debug_ = value;
            return status;
        }

        void barrier() {
            long i = world_.rank();
            sum(i);
            if (i != world_.size()*(world_.size()-1)/2) error("bad value after sum in barrier");
        }



        void fence();



        void broadcast(void* buf, size_t nbyte, ProcessID root, bool dowork = true);



        template <typename T>
        inline void broadcast(T* buf, size_t nelem, ProcessID root) {
            broadcast((void *) buf, nelem*sizeof(T), root);
        }

        template <typename T>
        void broadcast(T& t) {
            broadcast(&t, 1, 0);
        }

        template <typename T>
        void broadcast(T& t, ProcessID root) {
            broadcast(&t, 1, root);
        }


        template <typename objT>
        void broadcast_serializable(objT& obj, ProcessID root) {
            size_t BUFLEN;
            if (world_.rank() == root) {
                archive::BufferOutputArchive count;
                count & obj;
                BUFLEN = count.size();
            }
            broadcast(BUFLEN, root);

            unsigned char* buf = new unsigned char[BUFLEN];
            if (world_.rank() == root) {
                archive::BufferOutputArchive ar(buf,BUFLEN);
                ar & obj;
            }
            broadcast(buf, BUFLEN, root);
            if (world_.rank() != root) {
                archive::BufferInputArchive ar(buf,BUFLEN);
                ar & obj;
            }
            delete [] buf;
        }


        template <typename T, class opT>
        void reduce(T* buf, size_t nelem, opT op) {
            SafeMPI::Request req0, req1;
            ProcessID parent, child0, child1;
            world_.mpi.binary_tree_info(0, parent, child0, child1);
            Tag gsum_tag = world_.mpi.unique_tag();

            T* buf0 = new T[nelem];
            T* buf1 = new T[nelem];

            if (child0 != -1) req0 = world_.mpi.Irecv(buf0, nelem*sizeof(T), MPI_BYTE, child0, gsum_tag);
            if (child1 != -1) req1 = world_.mpi.Irecv(buf1, nelem*sizeof(T), MPI_BYTE, child1, gsum_tag);

            if (child0 != -1) {
                World::await(req0);
                for (long i=0; i<(long)nelem; ++i) buf[i] = op(buf[i],buf0[i]);
            }
            if (child1 != -1) {
                World::await(req1);
                for (long i=0; i<(long)nelem; ++i) buf[i] = op(buf[i],buf1[i]);
            }

            delete [] buf0;
            delete [] buf1;

            if (parent != -1) {
                req0 = world_.mpi.Isend(buf, nelem*sizeof(T), MPI_BYTE, parent, gsum_tag);
                World::await(req0);
            }

            broadcast(buf, nelem, 0);
        }

        template <typename T>
        inline void sum(T* buf, size_t nelem) {
            reduce< T, WorldSumOp<T> >(buf, nelem, WorldSumOp<T>());
        }

        template <typename T>
        inline void min(T* buf, size_t nelem) {
            reduce< T, WorldMinOp<T> >(buf, nelem, WorldMinOp<T>());
        }

        template <typename T>
        inline void max(T* buf, size_t nelem) {
            reduce< T, WorldMaxOp<T> >(buf, nelem, WorldMaxOp<T>());
        }

        template <typename T>
        inline void absmin(T* buf, size_t nelem) {
            reduce< T, WorldAbsMinOp<T> >(buf, nelem, WorldAbsMinOp<T>());
        }

        template <typename T>
        inline void absmax(T* buf, size_t nelem) {
            reduce< T, WorldAbsMaxOp<T> >(buf, nelem, WorldAbsMaxOp<T>());
        }

        template <typename T>
        inline void product(T* buf, size_t nelem) {
            reduce< T, WorldMultOp<T> >(buf, nelem, WorldMultOp<T>());
        }

        template <typename T>
        inline void bit_and(T* buf, size_t nelem) {
            reduce< T, WorldBitAndOp<T> >(buf, nelem, WorldBitAndOp<T>());
        }

        template <typename T>
        inline void bit_or(T* buf, size_t nelem) {
            reduce< T, WorldBitOrOp<T> >(buf, nelem, WorldBitOrOp<T>());
        }

        template <typename T>
        inline void bit_xor(T* buf, size_t nelem) {
            reduce< T, WorldBitXorOp<T> >(buf, nelem, WorldBitXorOp<T>());
        }

        template <typename T>
        inline void logic_and(T* buf, size_t nelem) {
            reduce< T, WorldLogicAndOp<T> >(buf, nelem, WorldLogicAndOp<T>());
        }

        template <typename T>
        inline void logic_or(T* buf, size_t nelem) {
            reduce< T, WorldLogicOrOp<T> >(buf, nelem, WorldLogicOrOp<T>());
        }

        template <typename T>
        void sum(T& a) {
            sum(&a, 1);
        }

        template <typename T>
        void max(T& a) {
            max(&a, 1);
        }

        template <typename T>
        void min(T& a) {
            min(&a, 1);
        }

        template <typename T>
        std::vector<T> concat0(const std::vector<T>& v, size_t bufsz=1024*1024) {
            SafeMPI::Request req0, req1;
            ProcessID parent, child0, child1;
            world_.mpi.binary_tree_info(0, parent, child0, child1);
            Tag gsum_tag = world_.mpi.unique_tag();

            unsigned char* buf0 = new unsigned char[bufsz];
            unsigned char* buf1 = new unsigned char[bufsz];

            if (child0 != -1) req0 = world_.mpi.Irecv(buf0, bufsz, MPI_BYTE, child0, gsum_tag);
            if (child1 != -1) req1 = world_.mpi.Irecv(buf1, bufsz, MPI_BYTE, child1, gsum_tag);

            std::vector<T> left, right;
            if (child0 != -1) {
                World::await(req0);
                archive::BufferInputArchive ar(buf0, bufsz);
                ar & left;
            }
            if (child1 != -1) {
                World::await(req1);
                archive::BufferInputArchive ar(buf1, bufsz);
                ar & right;
                for (unsigned int i=0; i<right.size(); ++i) left.push_back(right[i]);
            }

            for (unsigned int i=0; i<v.size(); ++i) left.push_back(v[i]);

            if (parent != -1) {
                archive::BufferOutputArchive ar(buf0, bufsz);
                ar & left;
                req0 = world_.mpi.Isend(buf0, ar.size(), MPI_BYTE, parent, gsum_tag);
                World::await(req0);
            }

            delete [] buf0;
            delete [] buf1;

            if (parent == -1) return left;
            else return std::vector<T>();
        }


        template <typename valueT, typename keyT>
        static Future<valueT> recv(const ProcessID source, const keyT& key) {
            return recv_internal<valueT>(ProcessKey<keyT, PointToPointTag>(key, source));
        }


        template <typename keyT, typename valueT>
        void send(const ProcessID dest, const keyT& key, const valueT& value) const {
            send_internal(dest, ProcessKey<keyT, PointToPointTag>(key, world_.rank()), value);
        }


        template <typename keyT, typename opT>
        void lazy_sync(const keyT& key, const opT& op) const {
            if(world_.size() > 1) { // Do nothing for the trivial case
                // Get the binary tree data
                Hash<keyT> hasher;
                const ProcessID root = hasher(key) % world_.size();
                ProcessID parent = -1, child0 = -1, child1 = -1;
                world_.mpi.binary_tree_info(root, parent, child0, child1);

                lazy_sync_internal<LazySyncTag>(parent, child0, child1, key, op);
            } else {
                typedef void (WorldGopInterface::*lazy_sync_childrenT)(const ProcessID, const ProcessID,
                    const keyT&, opT&, const ProcessID) const;
                // There is only one process, so run the sync operation now.
                world_.taskq.add(*this, lazy_sync_childrenT(& WorldGopInterface::template lazy_sync_children<keyT, opT>),
                        -1, -1, key, op, -1, TaskAttributes::hipri());
            }
        }



        template <typename keyT, typename opT>
        void lazy_sync(const keyT& key, const opT& op, const Group& group) const {
            MADNESS_ASSERT(! group.empty());
            MADNESS_ASSERT(group.get_world().id() == world_.id());

            if(group.size() > 1) { // Do nothing for the trivial case
                // Get the binary tree data
                Hash<keyT> hasher;
                const ProcessID group_root = hasher(key) % group.size();
                ProcessID parent = -1, child0 = -1, child1 = -1;
                group.make_tree(group_root, parent, child0, child1);

                lazy_sync_internal<GroupLazySyncTag>(parent, child0, child1, key, op);
            } else {
                typedef void (WorldGopInterface::*lazy_sync_children)(const ProcessID, const ProcessID,
                        const keyT&, opT&, const ProcessID) const;
                world_.taskq.add(*this, lazy_sync_children(& WorldGopInterface::template lazy_sync_children<keyT, opT>),
                        -1, -1, key, op, -1, TaskAttributes::hipri());
            }
        }


        template <typename keyT, typename valueT>
        void bcast(const keyT& key, Future<valueT>& value, const ProcessID root) const {
            MADNESS_ASSERT((root >= 0) && (root < world_.size()));
            MADNESS_ASSERT((world_.rank() == root) || (! value.probe()));

            if(world_.size() > 1) // Do nothing for the trivial case
                bcast_internal<BcastTag>(key, value, root);
        }


        template <typename keyT, typename valueT>
        void bcast(const keyT& key, Future<valueT>& value,
                const ProcessID group_root, const Group& group) const
        {
            MADNESS_ASSERT(! group.empty());
            MADNESS_ASSERT(group.get_world().id() == world_.id());
            MADNESS_ASSERT((group_root >= 0) && (group_root < group.size()));
            MADNESS_ASSERT((group.rank() == group_root) || (! value.probe()));

            if(group.size() > 1) // Do nothing for the trivial case
                bcast_internal<GroupBcastTag>(key, value, group_root, group);
        }


        template <typename keyT, typename valueT, typename opT>
        Future<typename detail::result_of<opT>::type>
        reduce(const keyT& key, const valueT& value, const opT& op, const ProcessID root) {
            MADNESS_ASSERT((root >= 0) && (root < world_.size()));

            // Get the binary tree data
            ProcessID parent = -1, child0 = -1, child1 = -1;
            world_.mpi.binary_tree_info(root, parent, child0, child1);

            return reduce_internal<ReduceTag>(parent, child0, child1, root, key,
                    value, op);
        }


        template <typename keyT, typename valueT, typename opT>
        Future<typename detail::result_of<opT>::type>
        reduce(const keyT& key, const valueT& value, const opT& op,
                const ProcessID group_root, const Group& group)
        {
            MADNESS_ASSERT(! group.empty());
            MADNESS_ASSERT(group.get_world().id() == world_.id());
            MADNESS_ASSERT((group_root >= 0) && (group_root < group.size()));

            // Get the binary tree data
            ProcessID parent = -1, child0 = -1, child1 = -1;
            group.make_tree(group_root, parent, child0, child1);

            return reduce_internal<ReduceTag>(parent, child0, child1, group_root,
                    key, value, op);
        }


        template <typename keyT, typename valueT, typename opT>
        Future<typename detail::result_of<opT>::type>
        all_reduce(const keyT& key, const valueT& value, const opT& op) {
            // Compute the parent and child processes of this process in a binary tree.
            Hash<keyT> hasher;
            const ProcessID root = hasher(key) % world_.size();
            ProcessID parent = -1, child0 = -1, child1 = -1;
            world_.mpi.binary_tree_info(root, parent, child0, child1);

            // Reduce the data
            Future<typename detail::result_of<opT>::type> reduce_result =
                    reduce_internal<AllReduceTag>(parent, child0, child1, root,
                            key, value, op);

            if(world_.rank() != root)
                reduce_result = Future<typename detail::result_of<opT>::type>();

            // Broadcast the result of the reduction to all processes
            bcast_internal<AllReduceTag>(key, reduce_result, root);

            return reduce_result;
        }


        template <typename keyT, typename valueT, typename opT>
        Future<typename detail::result_of<opT>::type>
        all_reduce(const keyT& key, const valueT& value, const opT& op, const Group& group) {
            MADNESS_ASSERT(! group.empty());
            MADNESS_ASSERT(group.get_world().id() == world_.id());

            // Compute the parent and child processes of this process in a binary tree.
            Hash<keyT> hasher;
            const ProcessID group_root = hasher(key) % group.size();
            ProcessID parent = -1, child0 = -1, child1 = -1;
            group.make_tree(group_root, parent, child0, child1);

            // Reduce the data
            Future<typename detail::result_of<opT>::type> reduce_result =
                    reduce_internal<GroupAllReduceTag>(parent, child0, child1,
                            group_root, key, value, op);


            if(group.rank() != group_root)
                reduce_result = Future<typename detail::result_of<opT>::type>();

            // Broadcast the result of the reduction to all processes in the group
            bcast_internal<GroupAllReduceTag>(key, reduce_result, 0, group);

            return reduce_result;
        }
    }; // class WorldGopInterface

} // namespace madness

#endif // MADNESS_WORLD_WORLDGOP_H__INCLUDED