VCluster.hpp

/*
 * Vcluster.hpp
 *
 *  Created on: Feb 8, 2016
 *      Author: Pietro Incardona
 */
 
#ifndef VCLUSTER_HPP
#define VCLUSTER_HPP
 
#include <signal.h>
 
#include "VCluster_base.hpp"
#include "VCluster_meta_function.hpp"
#include "util/math_util_complex.hpp"
#include "memory/mem_conf.hpp"
#include "util/cuda_util.hpp"
 
#ifdef CUDA_GPU
extern CudaMemory mem_tmp;
 
#ifndef MAX_NUMER_OF_PROPERTIES
#define MAX_NUMER_OF_PROPERTIES 20
#endif
 
extern CudaMemory exp_tmp;
extern CudaMemory exp_tmp2[MAX_NUMER_OF_PROPERTIES];
 
extern CudaMemory rem_tmp;
extern CudaMemory rem_tmp2[MAX_NUMER_OF_PROPERTIES];
 
#endif
 
extern size_t NBX_cnt;
 
void bt_sighandler(int sig, siginfo_t * info, void * ctx);
 
template<typename InternalMemory = HeapMemory>
class Vcluster: public Vcluster_base<InternalMemory>
{
    // Internal memory
    ExtPreAlloc<HeapMemory> * mem[NQUEUE];
 
    // Buffer that store the received bytes
    openfpm::vector<size_t> sz_recv_byte[NQUEUE];
 
    // The sending buffer used by semantic calls
    openfpm::vector<const void *> send_buf;
    openfpm::vector<size_t> send_sz_byte;
    openfpm::vector<size_t> prc_send_;
 
    unsigned int NBX_prc_scnt = 0;
    unsigned int NBX_prc_pcnt = 0;
 
 
    // Internal Heap memory
    HeapMemory * pmem[NQUEUE];
 
    template<typename Memory>
    struct base_info
    {
        openfpm::vector_fr<BMemory<Memory>> * recv_buf;
        openfpm::vector<size_t> * prc;
        openfpm::vector<size_t> * sz;
        openfpm::vector<size_t> * tags;
 
        size_t opt;
 
        base_info()
        {}
 
        base_info(openfpm::vector_fr<BMemory<Memory>> * recv_buf, openfpm::vector<size_t> & prc, openfpm::vector<size_t> & sz, openfpm::vector<size_t> & tags,size_t opt)
        :recv_buf(recv_buf),prc(&prc),sz(&sz),tags(&tags),opt(opt)
        {}
 
        void set(openfpm::vector_fr<BMemory<Memory>> * recv_buf, openfpm::vector<size_t> & prc, openfpm::vector<size_t> & sz, openfpm::vector<size_t> & tags,size_t opt)
        {
            this->recv_buf = recv_buf;
            this->prc = &prc;
            this->sz = &sz;
            this->tags = &tags;
            this->opt = opt;
        }
    };
 
    // Internal temporaty buffer
    base_info<InternalMemory> NBX_prc_bi[NQUEUE];
 
    typedef Vcluster_base<InternalMemory> self_base;
 
    template<typename T>
    struct index_gen {};
 
    template<int ... prp>
    struct index_gen<index_tuple<prp...>>
    {
        template<typename op,
            typename T,
            typename S,
            template <typename> class layout_base = memory_traits_lin>
        inline static void process_recv(
            Vcluster & vcl, S & recv,
            openfpm::vector<size_t> * sz_recv,
            openfpm::vector<size_t> * sz_recv_byte,
            op & op_param,size_t opt
        ) {
            if (opt == MPI_GPU_DIRECT && !std::is_same<InternalMemory,CudaMemory>::value)
            {
                // In order to have this option activated InternalMemory must be  CudaMemory
 
                std::cout << __FILE__ << ":" << __LINE__ << " error: in order to have MPI_GPU_DIRECT VCluster must use CudaMemory internally, the most probable" <<
                                                            " cause of this problem is that you are using MPI_GPU_DIRECT option with a non-GPU data-structure" << std::endl;
            }
 
            vcl.process_receive_buffer_with_prp<op,T,S,layout_base,prp...>(recv,sz_recv,sz_recv_byte,op_param,opt);
        }
    };
 
    template<typename op, typename T, typename S, template <typename> class layout_base>
    void prepare_send_buffer(
        openfpm::vector<T> & send,
        S & recv,
        openfpm::vector<size_t> & prc_send,
        openfpm::vector<size_t> & prc_recv,
        openfpm::vector<size_t> & sz_recv,
        size_t opt
    ) {
        sz_recv_byte[NBX_prc_scnt].resize(sz_recv.size());
 
        // Reset the receive buffer
        reset_recv_buf();
 
#ifdef SE_CLASS1
 
        if (send.size() != prc_send.size())
            std::cerr << __FILE__ << ":" << __LINE__ << " Error, the number of processor involved \"prc.size()\" must match the number of sending buffers \"send.size()\" " << std::endl;
 
#endif
 
        // Prepare the sending buffer
        send_buf.resize(0);
        send_sz_byte.resize(0);
        prc_send_.resize(0);
 
        size_t tot_size = 0;
 
        for (size_t i = 0; i < send.size() ; i++)
        {
            size_t req = 0;
 
            //Pack requesting
            pack_unpack_cond_with_prp<has_max_prop<T, has_value_type_ofp<T>::value>::value,op, T, S, layout_base>::packingRequest(send.get(i), req, send_sz_byte);
            tot_size += req;
        }
 
        pack_unpack_cond_with_prp_inte_lin<T>::construct_prc(prc_send,prc_send_);
 
 
        pmem[NBX_prc_scnt] = new HeapMemory;
 
        mem[NBX_prc_scnt] = new ExtPreAlloc<HeapMemory>(tot_size,*pmem[NBX_prc_scnt]);
        mem[NBX_prc_scnt]->incRef();
 
        for (size_t i = 0; i < send.size() ; i++)
        {
            //Packing
 
            Pack_stat sts;
 
            pack_unpack_cond_with_prp<has_max_prop<T, has_value_type_ofp<T>::value>::value, op, T, S, layout_base>::packing(*mem[NBX_prc_scnt], send.get(i), sts, send_buf,opt);
        }
 
        // receive information
        NBX_prc_bi[NBX_prc_scnt].set(&this->recv_buf[NBX_prc_scnt],prc_recv,sz_recv_byte[NBX_prc_scnt],this->tags[NBX_prc_scnt],opt);
 
        // Send and recv multiple messages
        if (opt & RECEIVE_KNOWN)
        {
            // We we are passing the number of element but not the byte, calculate the byte
            if (opt & KNOWN_ELEMENT_OR_BYTE)
            {
                // We know the number of element convert to byte (ONLY if it is possible)
                if (has_pack_gen<typename T::value_type>::value == false && is_vector<T>::value == true)
                {
                    for (size_t i = 0 ; i < sz_recv.size() ; i++)
                    {sz_recv_byte[NBX_prc_scnt].get(i) = sz_recv.get(i) * sizeof(typename T::value_type);}
                }
                else
                {
#ifndef DISABLE_ALL_RTTI
                    std::cout << __FILE__ << ":" << __LINE__ << " Error " << demangle(typeid(T).name()) << " the type does not work with the option or NO_CHANGE_ELEMENTS" << std::endl;
#endif
                }
 
                self_base::sendrecvMultipleMessagesNBXAsync(prc_send.size(),(size_t *)send_sz_byte.getPointer(),(size_t *)prc_send.getPointer(),(void **)send_buf.getPointer(),
                                            prc_recv.size(),(size_t *)prc_recv.getPointer(),(size_t *)sz_recv_byte[NBX_prc_scnt].getPointer(),msg_alloc_known,(void *)&NBX_prc_bi);
            }
            else
            {
                self_base::sendrecvMultipleMessagesNBXAsync(prc_send.size(),(size_t *)send_sz_byte.getPointer(),(size_t *)prc_send.getPointer(),(void **)send_buf.getPointer(),
                                            prc_recv.size(),(size_t *)prc_recv.getPointer(),msg_alloc_known,(void *)&NBX_prc_bi);
                sz_recv_byte[NBX_prc_scnt] = self_base::sz_recv_tmp;
            }
        }
        else
        {
            self_base::tags[NBX_prc_scnt].clear();
            prc_recv.clear();
            self_base::sendrecvMultipleMessagesNBXAsync(prc_send_.size(),(size_t *)send_sz_byte.getPointer(),(size_t *)prc_send_.getPointer(),(void **)send_buf.getPointer(),msg_alloc,(void *)&NBX_prc_bi[NBX_prc_scnt]);
        }
    }
 
 
    void reset_recv_buf()
    {
        for (size_t i = 0 ; i < self_base::recv_buf[NBX_prc_scnt].size() ; i++)
        {self_base::recv_buf[NBX_prc_scnt].get(i).resize(0);}
 
        self_base::recv_buf[NBX_prc_scnt].resize(0);
    }
 
    static void * msg_alloc(size_t msg_i ,size_t total_msg, size_t total_p, size_t i, size_t ri, size_t tag, void * ptr)
    {
        base_info<InternalMemory> & rinfo = *(base_info<InternalMemory> *)ptr;
 
        if (rinfo.recv_buf == NULL)
        {
            std::cerr << __FILE__ << ":" << __LINE__ << " Internal error this processor is not suppose to receive\n";
            return NULL;
        }
 
        rinfo.recv_buf->resize(ri+1);
 
        rinfo.recv_buf->get(ri).resize(msg_i);
 
        // Receive info
        rinfo.prc->add(i);
        rinfo.sz->add(msg_i);
        rinfo.tags->add(tag);
 
        // return the pointer
 
        // If we have GPU direct activated use directly the cuda buffer
        if (rinfo.opt & MPI_GPU_DIRECT)
        {
#if defined(MPIX_CUDA_AWARE_SUPPORT) && MPIX_CUDA_AWARE_SUPPORT
            return rinfo.recv_buf->last().getDevicePointer();
#else
            return rinfo.recv_buf->last().getPointer();
#endif
        }
 
        return rinfo.recv_buf->last().getPointer();
    }
 
 
    static void * msg_alloc_known(size_t msg_i ,size_t total_msg, size_t total_p, size_t i, size_t ri, size_t tag, void * ptr)
    {
        base_info<InternalMemory> & rinfo = *(base_info<InternalMemory> *)ptr;
 
        if (rinfo.recv_buf == NULL)
        {
            std::cerr << __FILE__ << ":" << __LINE__ << " Internal error this processor is not suppose to receive\n";
            return NULL;
        }
 
        rinfo.recv_buf->resize(ri+1);
 
        rinfo.recv_buf->get(ri).resize(msg_i);
 
        // return the pointer
        return rinfo.recv_buf->last().getPointer();
    }
    
    template<typename op, typename T, typename S, template <typename> class layout_base ,unsigned int ... prp >
    void process_receive_buffer_with_prp(
        S & recv,
        openfpm::vector<size_t> * sz,
        openfpm::vector<size_t> * sz_byte,
        op & op_param,
        size_t opt
    ) {
        if (sz != NULL)
        {sz->resize(self_base::recv_buf[NBX_prc_pcnt].size());}
 
        pack_unpack_cond_with_prp<has_max_prop<T, has_value_type_ofp<T>::value>::value,op, T, S, layout_base, prp... >::unpacking(recv, self_base::recv_buf[NBX_prc_pcnt], sz, sz_byte, op_param,opt);
    }
 
    public:
 
    Vcluster(int *argc, char ***argv, MPI_Comm ext_comm = MPI_COMM_WORLD)
    :Vcluster_base<InternalMemory>(argc,argv, ext_comm)
    {
    }
 
    template<typename T, typename S, template <typename> class layout_base=memory_traits_lin> bool SGather(T & send, S & recv,size_t root)
    {
        openfpm::vector<size_t> prc;
        openfpm::vector<size_t> sz;
 
        return SGather<T,S,layout_base>(send,recv,prc,sz,root);
    }
 
    template<size_t index, size_t N> struct MetaFuncOrd {
       enum { value = index };
    };
 
    template<typename T,
        typename S,
        template <typename> class layout_base = memory_traits_lin>
    bool SGather(
        T & send,
        S & recv,
        openfpm::vector<size_t> & prc,
        openfpm::vector<size_t> & sz,
    size_t root)
    {
#ifdef SE_CLASS1
        if (&send == (T *)&recv)
        {std::cerr << "Error: " << __FILE__ << ":" << __LINE__ << " using SGather in general the sending object and the receiving object must be different" << std::endl;}
#endif
 
        // Reset the receive buffer
        reset_recv_buf();
 
        // If we are on master collect the information
        if (self_base::getProcessUnitID() == root)
        {
            // send buffer (master does not send anything) so send req and send_buf
            // remain buffer with size 0
            openfpm::vector<size_t> send_req;
 
            self_base::tags[NBX_prc_scnt].clear();
 
            // receive information
            base_info<InternalMemory> bi(&this->recv_buf[NBX_prc_scnt],prc,sz,this->tags[NBX_prc_scnt],0);
 
            // Send and recv multiple messages
            self_base::sendrecvMultipleMessagesNBX(send_req.size(),NULL,NULL,NULL,msg_alloc,&bi);
 
            // we generate the list of the properties to unpack
            typedef typename ::generate_indexes<int, has_max_prop<T, has_value_type_ofp<T>::value>::number, MetaFuncOrd>::result ind_prop_to_pack;
 
            // operation object
            op_ssend_recv_add<void> opa;
 
            // Reorder the buffer
            reorder_buffer(prc,self_base::tags[NBX_prc_scnt],sz);
 
            index_gen<ind_prop_to_pack>::template process_recv<op_ssend_recv_add<void>,T,S,layout_base>(*this,recv,&sz,NULL,opa,0);
 
            recv.add(send);
            prc.add(root);
            sz.add(send.size());
        }
        else
        {
            // send buffer (master does not send anything) so send req and send_buf
            // remain buffer with size 0
            openfpm::vector<size_t> send_prc;
            openfpm::vector<size_t> send_prc_;
            send_prc.add(root);
 
            openfpm::vector<size_t> sz;
 
            openfpm::vector<const void *> send_buf;
                
            //Pack requesting
 
            size_t tot_size = 0;
 
            pack_unpack_cond_with_prp<has_max_prop<T, has_value_type_ofp<T>::value>::value,op_ssend_recv_add<void>, T, S, layout_base>::packingRequest(send, tot_size, sz);
 
            HeapMemory pmem;
 
            ExtPreAlloc<HeapMemory> & mem = *(new ExtPreAlloc<HeapMemory>(tot_size,pmem));
            mem.incRef();
 
            //Packing
 
            Pack_stat sts;
            
            pack_unpack_cond_with_prp<has_max_prop<T, has_value_type_ofp<T>::value>::value,op_ssend_recv_add<void>, T, S, layout_base>::packing(mem, send, sts, send_buf);
 
            pack_unpack_cond_with_prp_inte_lin<T>::construct_prc(send_prc,send_prc_);
 
            self_base::tags[NBX_prc_scnt].clear();
 
            // receive information
            base_info<InternalMemory> bi(NULL,prc,sz,self_base::tags[NBX_prc_scnt],0);
 
            // Send and recv multiple messages
            self_base::sendrecvMultipleMessagesNBX(send_prc_.size(),(size_t *)sz.getPointer(),(size_t *)send_prc_.getPointer(),(void **)send_buf.getPointer(),msg_alloc,(void *)&bi,NONE);
 
            mem.decRef();
            delete &mem;
        }
        
        return true;
    }
 
    void barrier()
    {
        MPI_Barrier(this->getMPIComm());
    }
 
    template<typename T, typename S, template <typename> class layout_base=memory_traits_lin>
    bool SScatter(T & send, S & recv, openfpm::vector<size_t> & prc, openfpm::vector<size_t> & sz, size_t root)
    {
        // Reset the receive buffer
        reset_recv_buf();
 
        // If we are on master scatter the information
        if (self_base::getProcessUnitID() == root)
        {
            // Prepare the sending buffer
            openfpm::vector<const void *> send_buf;
 
 
            openfpm::vector<size_t> sz_byte;
            sz_byte.resize(sz.size());
 
            size_t ptr = 0;
 
            for (size_t i = 0; i < sz.size() ; i++)
            {
                send_buf.add((char *)send.getPointer() + sizeof(typename T::value_type)*ptr );
                sz_byte.get(i) = sz.get(i) * sizeof(typename T::value_type);
                ptr += sz.get(i);
            }
 
            self_base::tags[NBX_prc_scnt].clear();
 
            // receive information
            base_info<InternalMemory> bi(&this->recv_buf[NBX_prc_scnt],prc,sz,this->tags[NBX_prc_scnt],0);
 
            // Send and recv multiple messages
            self_base::sendrecvMultipleMessagesNBX(prc.size(),(size_t *)sz_byte.getPointer(),(size_t *)prc.getPointer(),(void **)send_buf.getPointer(),msg_alloc,(void *)&bi);
 
            // we generate the list of the properties to pack
            typedef typename ::generate_indexes<int, has_max_prop<T, has_value_type_ofp<T>::value>::number, MetaFuncOrd>::result ind_prop_to_pack;
 
            // operation object
            op_ssend_recv_add<void> opa;
 
            index_gen<ind_prop_to_pack>::template process_recv<op_ssend_recv_add<void>,T,S,layout_base>(*this,recv,NULL,NULL,opa,0);
        }
        else
        {
            // The non-root receive
            openfpm::vector<size_t> send_req;
 
            self_base::tags[NBX_prc_scnt].clear();
 
            // receive information
            base_info<InternalMemory> bi(&this->recv_buf[NBX_prc_scnt],prc,sz,this->tags[NBX_prc_scnt],0);
 
            // Send and recv multiple messages
            self_base::sendrecvMultipleMessagesNBX(send_req.size(),NULL,NULL,NULL,msg_alloc,&bi);
 
            // we generate the list of the properties to pack
            typedef typename ::generate_indexes<int, has_max_prop<T, has_value_type_ofp<T>::value>::number, MetaFuncOrd>::result ind_prop_to_pack;
 
            // operation object
            op_ssend_recv_add<void> opa;
 
            index_gen<ind_prop_to_pack>::template process_recv<op_ssend_recv_add<void>,T,S,layout_base>(*this,recv,NULL,NULL,opa,0);
        }
 
        return true;
    }
    
    void reorder_buffer(openfpm::vector<size_t> & prc, const openfpm::vector<size_t> & tags, openfpm::vector<size_t> & sz_recv)
    {
 
        struct recv_buff_reorder
        {
            size_t proc;
 
            size_t tag;
 
            size_t pos;
 
            recv_buff_reorder()
            :proc(0),tag(0),pos(0)
            {};
 
            bool operator<(const recv_buff_reorder & rd) const
            {
                if (proc == rd.proc)
                {return tag < rd.tag;}
 
                return (proc < rd.proc);
            }
        };
 
        openfpm::vector<recv_buff_reorder> rcv;
 
        rcv.resize(self_base::recv_buf[NBX_prc_pcnt].size());
 
        for (size_t i = 0 ; i < rcv.size() ; i++)
        {
            rcv.get(i).proc = prc.get(i);
            if (i < tags.size())
            {rcv.get(i).tag = tags.get(i);}
            else
            {rcv.get(i).tag = (unsigned int)-1;}
            rcv.get(i).pos = i;
        }
 
        // we sort based on processor
        rcv.sort();
 
        openfpm::vector_fr<BMemory<InternalMemory>> recv_ord;
        recv_ord.resize(rcv.size());
 
        openfpm::vector<size_t> prc_ord;
        prc_ord.resize(rcv.size());
 
        openfpm::vector<size_t> sz_recv_ord;
        sz_recv_ord.resize(rcv.size());
 
        // Now we reorder rcv
        for (size_t i = 0 ; i < rcv.size() ; i++)
        {
            recv_ord.get(i).swap(self_base::recv_buf[NBX_prc_pcnt].get(rcv.get(i).pos));
            prc_ord.get(i) = rcv.get(i).proc;
            sz_recv_ord.get(i) = sz_recv.get(rcv.get(i).pos);
        }
 
        // move rcv into recv
        // Now we swap back to recv_buf in an ordered way
        for (size_t i = 0 ; i < rcv.size() ; i++)
        {
            self_base::recv_buf[NBX_prc_pcnt].get(i).swap(recv_ord.get(i));
        }
 
        prc.swap(prc_ord);
        sz_recv.swap(sz_recv_ord);
 
        // reorder prc_recv and recv_sz
    }
 
    template<typename T,
        typename S,
        template <typename> class layout_base = memory_traits_lin>
    bool SSendRecv(
        openfpm::vector<T> & send,
        S & recv,
        openfpm::vector<size_t> & prc_send,
        openfpm::vector<size_t> & prc_recv,
        openfpm::vector<size_t> & sz_recv,
        size_t opt = NONE)
    {
        prepare_send_buffer<op_ssend_recv_add<void>,T,S,layout_base>(send,recv,prc_send,prc_recv,sz_recv,opt);
 
        self_base::sendrecvMultipleMessagesNBXWait();
 
        // Reorder the buffer
        reorder_buffer(prc_recv,self_base::tags[NBX_prc_scnt],sz_recv_byte[NBX_prc_scnt]);
 
        mem[NBX_prc_scnt]->decRef();
        delete mem[NBX_prc_scnt];
        delete pmem[NBX_prc_scnt];
 
        // we generate the list of the properties to pack
        typedef typename ::generate_indexes<int, has_max_prop<T, has_value_type_ofp<T>::value>::number, MetaFuncOrd>::result ind_prop_to_pack;
 
        op_ssend_recv_add<void> opa;
 
        index_gen<ind_prop_to_pack>::template process_recv<op_ssend_recv_add<void>,T,S,layout_base>(*this,recv,&sz_recv,NULL,opa,opt);
 
        return true;
    }
 
    template<typename T,
        typename S,
        template <typename> class layout_base = memory_traits_lin>
    bool SSendRecvAsync(
        openfpm::vector<T> & send,
        S & recv,
        openfpm::vector<size_t> & prc_send,
        openfpm::vector<size_t> & prc_recv,
        openfpm::vector<size_t> & sz_recv,
        size_t opt = NONE
    ) {
        prepare_send_buffer<op_ssend_recv_add<void>,T,S,layout_base>(send,recv,prc_send,prc_recv,sz_recv,opt);
 
        NBX_prc_scnt++;
 
        return true;
    }
 
    template<typename T, typename S, template <typename> class layout_base, int ... prp>
    bool SSendRecvP(
        openfpm::vector<T> & send,
        S & recv,
        openfpm::vector<size_t> & prc_send,
        openfpm::vector<size_t> & prc_recv,
        openfpm::vector<size_t> & sz_recv,
        openfpm::vector<size_t> & sz_recv_byte_out,
        size_t opt = NONE
    ) {
        prepare_send_buffer<op_ssend_recv_add<void>,T,S,layout_base>(send,recv,prc_send,prc_recv,sz_recv,opt);
 
        self_base::sendrecvMultipleMessagesNBXWait();
 
        // Reorder the buffer
        reorder_buffer(prc_recv,self_base::tags[NBX_prc_scnt],sz_recv_byte[NBX_prc_scnt]);
 
        mem[NBX_prc_scnt]->decRef();
        delete mem[NBX_prc_scnt];
        delete pmem[NBX_prc_scnt];
 
        // operation object
        op_ssend_recv_add<void> opa;
 
        // process the received information
        process_receive_buffer_with_prp<op_ssend_recv_add<void>,T,S,layout_base,prp...>(recv,&sz_recv,&sz_recv_byte_out,opa,opt);
 
        return true;
    }
 
    template<typename T, typename S, template <typename> class layout_base, int ... prp>
    bool SSendRecvPAsync(
        openfpm::vector<T> & send,
        S & recv,
        openfpm::vector<size_t> & prc_send,
        openfpm::vector<size_t> & prc_recv,
        openfpm::vector<size_t> & sz_recv,
        openfpm::vector<size_t> & sz_recv_byte_out,
        size_t opt = NONE
    ) {
        prepare_send_buffer<op_ssend_recv_add<void>,T,S,layout_base>(send,recv,prc_send,prc_recv,sz_recv,opt);
 
        NBX_prc_scnt++;
 
        return true;
    }
 
    template<typename T, typename S, template <typename> class layout_base, int ... prp>
    bool SSendRecvP(
        openfpm::vector<T> & send,
        S & recv,
        openfpm::vector<size_t> & prc_send,
        openfpm::vector<size_t> & prc_recv,
        openfpm::vector<size_t> & sz_recv,
        size_t opt = NONE
    ) {
        prepare_send_buffer<op_ssend_recv_add<void>,T,S,layout_base>(send,recv,prc_send,prc_recv,sz_recv,opt);
 
        self_base::sendrecvMultipleMessagesNBXWait();
 
        // Reorder the buffer
        reorder_buffer(prc_recv,self_base::tags[NBX_prc_scnt],sz_recv_byte[NBX_prc_scnt]);
 
        mem[NBX_prc_scnt]->decRef();
        delete mem[NBX_prc_scnt];
        delete pmem[NBX_prc_scnt];
 
        // operation object
        op_ssend_recv_add<void> opa;
 
        // process the received information
        process_receive_buffer_with_prp<op_ssend_recv_add<void>,T,S,layout_base,prp...>(recv,&sz_recv,NULL,opa,opt);
 
        return true;
    }
 
    template<typename T, typename S, template <typename> class layout_base, int ... prp>
    bool SSendRecvPAsync(
        openfpm::vector<T> & send,
        S & recv,
        openfpm::vector<size_t> & prc_send,
        openfpm::vector<size_t> & prc_recv,
        openfpm::vector<size_t> & sz_recv,
        size_t opt = NONE
    ) {
        prepare_send_buffer<op_ssend_recv_add<void>,T,S,layout_base>(send,recv,prc_send,prc_recv,sz_recv,opt);
 
        NBX_prc_scnt++;
 
        return true;
    }
 
    template<typename op,
        typename T,
        typename S,
        template <typename> class layout_base,
        int ... prp>
    bool SSendRecvP_op(
        openfpm::vector<T> & send,
        S & recv,
        openfpm::vector<size_t> & prc_send,
        op & op_param,
        openfpm::vector<size_t> & prc_recv,
        openfpm::vector<size_t> & recv_sz,
        size_t opt = NONE
    ) {
        prepare_send_buffer<op,T,S,layout_base>(send,recv,prc_send,prc_recv,recv_sz,opt);
 
        self_base::sendrecvMultipleMessagesNBXWait();
 
        // Reorder the buffer
        reorder_buffer(prc_recv,self_base::tags[NBX_prc_scnt],sz_recv_byte[NBX_prc_scnt]);
 
        mem[NBX_prc_scnt]->decRef();
        delete mem[NBX_prc_scnt];
        delete pmem[NBX_prc_scnt];
 
        // process the received information
        process_receive_buffer_with_prp<op,T,S,layout_base,prp...>(recv,NULL,NULL,op_param,opt);
 
        return true;
    }
 
    template<typename op,
        typename T,
        typename S,
        template <typename> class layout_base,
        int ... prp>
    bool SSendRecvP_opAsync(
        openfpm::vector<T> & send,
        S & recv,
        openfpm::vector<size_t> & prc_send,
        op & op_param,
        openfpm::vector<size_t> & prc_recv,
        openfpm::vector<size_t> & recv_sz,
        size_t opt = NONE
    ) {
        prepare_send_buffer<op,T,S,layout_base>(send,recv,prc_send,prc_recv,recv_sz,opt);
 
        NBX_prc_scnt++;
 
        return true;
    }
 
    template<typename T,
        typename S,
        template <typename> class layout_base = memory_traits_lin>
    bool SSendRecvWait(
        openfpm::vector<T> & send,
        S & recv,
        openfpm::vector<size_t> & prc_send,
        openfpm::vector<size_t> & prc_recv,
        openfpm::vector<size_t> & sz_recv,
        size_t opt = NONE
    ) {
        self_base::sendrecvMultipleMessagesNBXWait();
 
        // Reorder the buffer
        reorder_buffer(prc_recv,self_base::tags[NBX_prc_pcnt],sz_recv_byte[NBX_prc_pcnt]);
 
        mem[NBX_prc_pcnt]->decRef();
        delete mem[NBX_prc_pcnt];
        delete pmem[NBX_prc_pcnt];
 
        // we generate the list of the properties to pack
        typedef typename ::generate_indexes<int, has_max_prop<T, has_value_type_ofp<T>::value>::number, MetaFuncOrd>::result ind_prop_to_pack;
 
        op_ssend_recv_add<void> opa;
 
        index_gen<ind_prop_to_pack>::template process_recv<op_ssend_recv_add<void>,T,S,layout_base>(*this,recv,&sz_recv,NULL,opa,opt);
 
        NBX_prc_pcnt++;
        if (NBX_prc_scnt == NBX_prc_pcnt)
        {
            NBX_prc_scnt = 0;
            NBX_prc_pcnt = 0;
        }
 
        return true;
    }
 
    template<typename T, typename S, template <typename> class layout_base, int ... prp>
    bool SSendRecvPWait(
        openfpm::vector<T> & send,
        S & recv,
        openfpm::vector<size_t> & prc_send,
        openfpm::vector<size_t> & prc_recv,
        openfpm::vector<size_t> & sz_recv,
        openfpm::vector<size_t> & sz_recv_byte_out,
        size_t opt = NONE
    ) {
        self_base::sendrecvMultipleMessagesNBXWait();
 
        // Reorder the buffer
        reorder_buffer(prc_recv,self_base::tags[NBX_prc_pcnt],sz_recv_byte[NBX_prc_pcnt]);
 
        mem[NBX_prc_pcnt]->decRef();
        delete mem[NBX_prc_pcnt];
        delete pmem[NBX_prc_pcnt];
 
        // operation object
        op_ssend_recv_add<void> opa;
 
        // process the received information
        process_receive_buffer_with_prp<op_ssend_recv_add<void>,T,S,layout_base,prp...>(recv,&sz_recv,&sz_recv_byte_out,opa,opt);
 
        NBX_prc_pcnt++;
        if (NBX_prc_scnt == NBX_prc_pcnt)
        {
            NBX_prc_scnt = 0;
            NBX_prc_pcnt = 0;
        }
 
        return true;
    }
 
    template<typename T, typename S, template <typename> class layout_base, int ... prp>
    bool SSendRecvPWait(
        openfpm::vector<T> & send,
        S & recv,
        openfpm::vector<size_t> & prc_send,
        openfpm::vector<size_t> & prc_recv,
        openfpm::vector<size_t> & sz_recv,
        size_t opt = NONE
    ) {
        self_base::sendrecvMultipleMessagesNBXWait();
 
        // Reorder the buffer
        reorder_buffer(prc_recv,self_base::tags[NBX_prc_pcnt],sz_recv_byte[NBX_prc_pcnt]);
 
        mem[NBX_prc_pcnt]->decRef();
        delete mem[NBX_prc_pcnt];
        delete pmem[NBX_prc_pcnt];
 
        // operation object
        op_ssend_recv_add<void> opa;
 
        // process the received information
        process_receive_buffer_with_prp<op_ssend_recv_add<void>,T,S,layout_base,prp...>(recv,&sz_recv,NULL,opa,opt);
 
        NBX_prc_pcnt++;
        if (NBX_prc_scnt == NBX_prc_pcnt)
        {
            NBX_prc_scnt = 0;
            NBX_prc_pcnt = 0;
        }
 
        return true;
    }
 
    template<typename op,
        typename T,
        typename S,
        template <typename> class layout_base,
        int ... prp>
    bool SSendRecvP_opWait(
        openfpm::vector<T> & send,
        S & recv,
        openfpm::vector<size_t> & prc_send,
        op & op_param,
        openfpm::vector<size_t> & prc_recv,
        openfpm::vector<size_t> & recv_sz,
        size_t opt = NONE
    ) {
        self_base::sendrecvMultipleMessagesNBXWait();
 
        // Reorder the buffer
        reorder_buffer(prc_recv,self_base::tags[NBX_prc_pcnt],sz_recv_byte[NBX_prc_pcnt]);
 
        mem[NBX_prc_pcnt]->decRef();
        delete mem[NBX_prc_pcnt];
        delete pmem[NBX_prc_pcnt];
 
        // process the received information
        process_receive_buffer_with_prp<op,T,S,layout_base,prp...>(recv,NULL,NULL,op_param,opt);
 
        NBX_prc_pcnt++;
        if (NBX_prc_scnt == NBX_prc_pcnt)
        {
            NBX_prc_scnt = 0;
            NBX_prc_pcnt = 0;
        }
 
        return true;
    }
 
};
 
// Function to initialize the global VCluster //
 
extern Vcluster<> * global_v_cluster_private_heap;
extern Vcluster<CudaMemory> * global_v_cluster_private_cuda;
 
static inline void init_global_v_cluster_private(int *argc, char ***argv, MPI_Comm ext_comm)
{
    if (global_v_cluster_private_heap == NULL)
    {global_v_cluster_private_heap = new Vcluster<>(argc,argv,ext_comm);}
 
    if (global_v_cluster_private_cuda == NULL)
    {global_v_cluster_private_cuda = new Vcluster<CudaMemory>(argc,argv,ext_comm);}
}
 
static inline void delete_global_v_cluster_private()
{
    delete global_v_cluster_private_heap;
    delete global_v_cluster_private_cuda;
}
 
template<typename Memory>
struct get_vcl
{
    static Vcluster<Memory> & get()
    {
        return *global_v_cluster_private_heap;
    }
};
 
template<>
struct get_vcl<CudaMemory>
{
    static Vcluster<CudaMemory> & get()
    {
        return *global_v_cluster_private_cuda;
    }
};
 
template<typename Memory = HeapMemory>
static inline Vcluster<Memory> & create_vcluster()
{
    if (global_v_cluster_private_heap == NULL)
    {std::cerr << __FILE__ << ":" << __LINE__ << " Error you must call openfpm_init before using any distributed data structures";}
 
    return get_vcl<Memory>::get();
}
 
 
 
static inline bool is_openfpm_init()
{
    return ofp_initialized;
}
 
 
void openfpm_init_vcl(int *argc, char ***argv, MPI_Comm ext_comm);
 
size_t openfpm_vcluster_compilation_mask();
 
void openfpm_finalize();
 
static void openfpm_init(int *argc, char ***argv, MPI_Comm ext_comm=MPI_COMM_WORLD)
{
    if (ofp_initialized)
    {
        return;
    }
    openfpm_init_vcl(argc,argv, ext_comm);
 
    size_t compiler_mask = CUDA_ON_BACKEND;
 
    init_wrappers();
 
    if (compiler_mask != openfpm_vcluster_compilation_mask() || compiler_mask != openfpm_ofpmmemory_compilation_mask())
    {
        std::cout << __FILE__ << ":" << __LINE__ << " Error: the program has been compiled with CUDA_ON_BACKEND: " << compiler_mask << " but libvcluster has been compiled with CUDA_ON_BACKEND: " <<
                                                    openfpm_vcluster_compilation_mask() << ", and libofpmmemory has been compiled with CUDA_ON_BACKEND: " << openfpm_ofpmmemory_compilation_mask() <<
                                                    " recompile the library with the right CUDA_ON_BACKEND" << std::endl;
    }
}
 
#endif