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/cudify/cudify.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)
    :Vcluster_base<InternalMemory>(argc,argv)
    {
    }

    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(MPI_COMM_WORLD);
    }

    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)
{
    if (global_v_cluster_private_heap == NULL)
    {global_v_cluster_private_heap = new Vcluster<>(argc,argv);}

    if (global_v_cluster_private_cuda == NULL)
    {global_v_cluster_private_cuda = new Vcluster<CudaMemory>(argc,argv);}
}

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);

size_t openfpm_vcluster_compilation_mask();

void openfpm_finalize();

static void openfpm_init(int *argc, char ***argv)
{
    openfpm_init_vcl(argc,argv);

    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