binsorter.h

Go to the documentation of this file.

#include <madness/world/world.h>
#include <madness/world/worldobj.h>
#include <madness/world/worldexc.h>
#include <vector>

namespace madness {

    template <typename T, typename inserterT>
    class BinSorter : public WorldObject< BinSorter<T,inserterT> > {
        World* pworld;
        inserterT inserter;
        std::size_t bufsize;
        std::vector<T>* bins;
        
        void flush(int owner) {
            if (bins[owner].size())
                this->send(owner, &BinSorter<T,inserterT>::sorter,  bins[owner]);
            bins[owner].clear();
        }
        
        void sorter(const std::vector<T>& v) {
            for(typename std::vector<T>::const_iterator it = v.begin(); it != v.end(); ++it) {
                inserter(*it);
            }
        }
        
    public:

        BinSorter(World& world, inserterT inserter, int bufsize=0)
            : WorldObject<BinSorter>(world)
            , pworld(&world)
            , inserter(inserter)
            , bufsize(bufsize)
            , bins(new std::vector<T>[world.size()])
        {
            // bufsize ... max from AM buffer size is about 512K/sizeof(T)
            // bufsize ... max from total buffer use is about 1GB/sizeof(T)/P
            if (bufsize <= 0) this->bufsize = std::min((RMI::max_msg_len()-1024)/sizeof(T),(1<<30)/(world.size()*sizeof(T)));

            // for (int i=0; i<world.size(); i++) {
            //     bins[i].reserve(bufsize); // Not a good idea on large process counts unless truly all to all?
            // }

            //print("binsorter bufsize is", this->bufsize, this->bufsize*sizeof(T));
            WorldObject< BinSorter<T,inserterT> >::process_pending();
        }
        
        virtual ~BinSorter() {
            for (int i=0; i<pworld->size(); i++) {
                MADNESS_ASSERT(bins[i].size() == 0);
            }
            delete [] bins;
        }
        
        void finish() {
            for (int i=0; i<pworld->size(); i++) {
                flush(i);
            }
            pworld->gop.fence();
        }
        
        void insert(ProcessID p, const T& value) {
            bins[p].push_back(value);

            // More intelligent buffer management would look at total use and flush
            // largest buffers rather than using a fixed buffersize per process
            if (bins[p].size() >= bufsize) flush(p);
        }
    };
}