VCluster_base.hpp

#ifndef VCLUSTER_BASE_HPP_
#define VCLUSTER_BASE_HPP_


#include "util/cuda_util.hpp"
#ifdef OPENMPI
#include <mpi.h>
#include <mpi-ext.h>
#else
#include <mpi.h>
#endif
#include "MPI_wrapper/MPI_util.hpp"
#include "Vector/map_vector.hpp"
#include "MPI_wrapper/MPI_IallreduceW.hpp"
#include "MPI_wrapper/MPI_IrecvW.hpp"
#include "MPI_wrapper/MPI_IsendW.hpp"
#include "MPI_wrapper/MPI_IAllGather.hpp"
#include "MPI_wrapper/MPI_IBcastW.hpp"
#include <exception>
#include "Vector/map_vector.hpp"
#ifdef DEBUG
#include "util/check_no_pointers.hpp"
#include "util/util_debug.hpp"
#endif
#include "util/Vcluster_log.hpp"
#include "memory/BHeapMemory.hpp"
#include "Packer_Unpacker/has_max_prop.hpp"
#include "data_type/aggregate.hpp"
#include "util/cuda/ofp_context.hxx"

#ifdef HAVE_PETSC
#include <petscvec.h>
#endif

extern double time_spent;

enum NBX_Type
{
    NBX_UNACTIVE,
    NBX_UNKNOWN,
    NBX_KNOWN,
    NBX_KNOWN_PRC
};

constexpr int MSG_LENGTH = 1024;
constexpr int MSG_SEND_RECV = 1025;
constexpr int SEND_SPARSE = 8192;
constexpr int NONE = 1;
constexpr int NEED_ALL_SIZE = 2;

constexpr int SERIVCE_MESSAGE_TAG = 16384;
constexpr int SEND_RECV_BASE = 4096;
constexpr int GATHER_BASE = 24576;

constexpr int RECEIVE_KNOWN = 4;
constexpr int KNOWN_ELEMENT_OR_BYTE = 8;
constexpr int MPI_GPU_DIRECT = 16;

constexpr int NQUEUE = 4;

// number of vcluster instances
extern size_t n_vcluster;
// Global MPI initialization
extern bool global_mpi_init;
// initialization flag
extern bool ofp_initialized;
extern size_t tot_sent;
extern size_t tot_recv;


extern size_t NBX_cnt;

template<typename T> void assign(T * ptr1, T * ptr2)
{
    *ptr1 = *ptr2;
};


union red
{
    char c;
    unsigned char uc;
    short s;
    unsigned short us;
    int i;
    unsigned int ui;
    float f;
    double d;
};

template<typename InternalMemory>
class Vcluster_base
{
    Vcluster_log log;

    openfpm::vector<size_t> proc_com;

    openfpm::vector<int> map_scatter;

    openfpm::vector<MPI_Request> req;

    openfpm::vector<MPI_Status> stat;

    std::vector<int> post_exe;

    mgpu::ofp_context_t * context;

    // Single objects

    int m_size;
    int m_rank;

    int numPE = 1;


    NBX_Type NBX_active[NQUEUE];

    size_t rid[NQUEUE];

    int NBX_prc_qcnt = -1;

    bool NBX_prc_reached_bar_req[NQUEUE];


    int NBX_prc_cnt_base = 0;
    size_t NBX_prc_n_send[NQUEUE];
    size_t * NBX_prc_prc[NQUEUE];
    void ** NBX_prc_ptr[NQUEUE];
    size_t * NBX_prc_sz[NQUEUE];
    size_t NBX_prc_n_recv[NQUEUE];
    void * (* NBX_prc_msg_alloc[NQUEUE])(size_t,size_t,size_t,size_t,size_t,size_t,void *);
    size_t * NBX_prc_prc_recv[NQUEUE];
    void * NBX_prc_ptr_arg[NQUEUE];


    std::vector<red> r;

    openfpm::vector<void *> ptr_send[NQUEUE];

    openfpm::vector<size_t> sz_send[NQUEUE];

    MPI_Request bar_req;

    MPI_Status bar_stat;

    Vcluster_base & operator=(const Vcluster_base &)    {return *this;};

    int shmrank;

    int nbx_cycle;

    Vcluster_base(const Vcluster_base &)
    {};

    void queue_all_sends(size_t n_send , size_t sz[],
                         size_t prc[], void * ptr[])
    {
        if (stat.size() != 0 || (req.size() != 0 && NBX_prc_qcnt == 0))
        {std::cerr << "Error: " << __FILE__ << ":" << __LINE__ << " this function must be called when no other requests are in progress. Please remember that if you use function like max(),sum(),send(),recv() check that you did not miss to call the function execute() \n";}


        if (NBX_prc_qcnt == 0)
        {
            stat.clear();
            req.clear();
            // Do MPI_Issend
        }

        for (size_t i = 0 ; i < n_send ; i++)
        {
            if (sz[i] != 0)
            {
                req.add();

#ifdef SE_CLASS2
                check_valid(ptr[i],sz[i]);
#endif

                tot_sent += sz[i];

//              std::cout << "TAG: " << SEND_SPARSE + (NBX_cnt + NBX_prc_qcnt)*131072 + i << "   " << NBX_cnt << "   "  << NBX_prc_qcnt << "  " << " rank: " << rank() << "   " << NBX_prc_cnt_base << "  nbx_cycle: " << nbx_cycle << std::endl;

                MPI_SAFE_CALL(MPI_Issend(ptr[i], sz[i], MPI_BYTE, prc[i], SEND_SPARSE + (NBX_cnt + NBX_prc_qcnt)*131072 + i, MPI_COMM_WORLD,&req.last()));
                log.logSend(prc[i]);
            }
        }
    }

protected:

    openfpm::vector_fr<BMemory<InternalMemory>> recv_buf[NQUEUE];

    openfpm::vector<size_t> tags[NQUEUE];

public:

    // Finalize the MPI program
    ~Vcluster_base()
    {
#ifdef SE_CLASS2
        check_delete(this);
#endif
        n_vcluster--;

        // if there are no other vcluster instances finalize
        if (n_vcluster == 0)
        {
            int already_finalised;

            MPI_Finalized(&already_finalised);
            if (!already_finalised)
            {
                if (MPI_Finalize() != 0)
                {
                    std::cerr << __FILE__ << ":" << __LINE__  << " MPI_Finalize FAILED \n";
                }
            }
        }

        delete context;
    }

    Vcluster_base(int *argc, char ***argv)
    {
        // reset NBX_Active

        for (unsigned int i = 0 ; i < NQUEUE ; i++)
        {
            NBX_active[i] = NBX_Type::NBX_UNACTIVE;
            rid[i] = 0;
        }

#ifdef SE_CLASS2
        check_new(this,8,VCLUSTER_EVENT,PRJ_VCLUSTER);
#endif

        n_vcluster++;

        int already_initialised;
        MPI_Initialized(&already_initialised);

        // Check if MPI is already initialized
        if (!already_initialised)
        {
            MPI_Init(argc,argv);
        }

        // We try to get the local processors rank

        MPI_Comm shmcomm;
        MPI_Comm_split_type(MPI_COMM_WORLD, MPI_COMM_TYPE_SHARED, 0,
                            MPI_INFO_NULL, &shmcomm);

        MPI_Comm_rank(shmcomm, &shmrank);
        MPI_Comm_free(&shmcomm);

        // Get the total number of process
        // and the rank of this process

        MPI_Comm_size(MPI_COMM_WORLD, &m_size);
        MPI_Comm_rank(MPI_COMM_WORLD, &m_rank);

#ifdef SE_CLASS2
            process_v_cl = m_rank;
#endif

        // create and fill map scatter with one
        map_scatter.resize(m_size);

        for (size_t i = 0 ; i < map_scatter.size() ; i++)
        {
            map_scatter.get(i) = 1;
        }

        // open the log file
        log.openLog(m_rank);

        // Initialize bar_req
        bar_req = MPI_Request();
        bar_stat = MPI_Status();

#ifdef EXTERNAL_SET_GPU
                int dev;
                cudaGetDevice(&dev);
                context = new mgpu::ofp_context_t(mgpu::gpu_context_opt::no_print_props,dev);
#else
                context = new mgpu::ofp_context_t(mgpu::gpu_context_opt::no_print_props,shmrank);
#endif


#if defined(PRINT_RANK_TO_GPU) && defined(CUDA_GPU)

                char node_name[MPI_MAX_PROCESSOR_NAME];
                int len;

                MPI_Get_processor_name(node_name,&len);

                std::cout << "Rank: " << m_rank << " on host: " << node_name << " work on GPU: " << context->getDevice() << "/" << context->getNDevice() << std::endl;
#endif

                int flag;
                void *tag_ub_v;
                int tag_ub;

                MPI_Comm_get_attr(MPI_COMM_WORLD, MPI_TAG_UB, &tag_ub_v, &flag);
                tag_ub = *(int*)tag_ub_v;

                if (flag == true)
                {
                    nbx_cycle = (tag_ub - SEND_SPARSE - 131072 - NQUEUE*131072) / 131072;

                    if (nbx_cycle < NQUEUE*2)
                    {std::cerr << __FILE__ << ":" << __LINE__ << " Error MPI_TAG_UB is too small for OpenFPM" << std::endl;}
                }
                else
                {nbx_cycle = 2048;}
    }

#ifdef SE_CLASS1

    template<typename T> void checkType()
    {
        // if T is a primitive like int, long int, float, double, ... make sense
        // (pointers, l-references and r-references are not fundamentals)
        if (std::is_fundamental<T>::value == true)
        {return;}

        // if it is a pointer make no sense
        if (std::is_pointer<T>::value == true)
        {std::cerr << "Error: " << __FILE__ << ":" << __LINE__ << " the type " << demangle(typeid(T).name()) << " is a pointer, sending pointers values has no sense\n";}

        // if it is an l-value reference make no send
        if (std::is_lvalue_reference<T>::value == true)
        {std::cerr << "Error: " << __FILE__ << ":" << __LINE__ << " the type " << demangle(typeid(T).name()) << " is a pointer, sending pointers values has no sense\n";}

        // if it is an r-value reference make no send
        if (std::is_rvalue_reference<T>::value == true)
        {std::cerr << "Error: " << __FILE__ << ":" << __LINE__ << " the type " << demangle(typeid(T).name()) << " is a pointer, sending pointers values has no sense\n";}

        // ... if not, check that T has a method called noPointers
        switch (check_no_pointers<T>::value())
        {
            case PNP::UNKNOWN:
            {
                std::cerr << "Warning: " << __FILE__ << ":" << __LINE__ << " impossible to check the type " << demangle(typeid(T).name()) << " please consider to add a static method \"static bool noPointers()\" \n" ;
                break;
            }
            case PNP::POINTERS:
            {
                std::cerr << "Error: " << __FILE__ << ":" << __LINE__ << " the type " << demangle(typeid(T).name()) << " has pointers inside, sending pointers values has no sense\n";
                break;
            }
            default:
            {

            }
        }
    }

#endif

    mgpu::ofp_context_t & getmgpuContext(bool iw = true)
    {
        if (context == NULL && iw == true)
        {
            std::cout << __FILE__ << ":" << __LINE__ << " Warning: it seem that modern gpu context is not initialized."
                                                        "Either a compatible working cuda device has not been found, either openfpm_init has been called in a file that not compiled with NVCC" << std::endl;
        }

        return *context;
    }

    MPI_Comm getMPIComm()
    {
        return MPI_COMM_WORLD;
    }

    size_t getProcessingUnits()
    {
        return m_size*numPE;
    }

    size_t size()
    {
        return this->m_size*numPE;
    }

    void print_stats()
    {
#ifdef VCLUSTER_PERF_REPORT
        std::cout << "-- REPORT COMMUNICATIONS -- " << std::endl;

        std::cout << "Processor " << this->rank() << " sent: " << tot_sent << std::endl;
        std::cout << "Processor " << this->rank() << " received: " << tot_recv << std::endl;

        std::cout << "Processor " << this->rank() << " time spent: " << time_spent << std::endl;
        std::cout << "Processor " << this->rank() << " Bandwidth: S:"  << (double)tot_sent / time_spent * 1e-9 << "GB/s  R:" << (double)tot_recv / time_spent * 1e-9 << "GB/s" <<  std::endl;
#else

        std::cout << "Error to activate performance stats on VCluster enable VCLUSTER_PERF_REPORT" << std::endl;

#endif
    }

    void clear_stats()
    {
#ifdef VCLUSTER_PERF_REPORT

        tot_sent = 0;
        tot_recv = 0;

        time_spent = 0;
#else

        std::cout << "Error to activate performance stats on VCluster enable VCLUSTER_PERF_REPORT" << std::endl;

#endif
    }

    size_t getProcessUnitID()
    {
        return m_rank;
    }

    size_t rank()
    {
        return m_rank;
    }


    template<typename T> void sum(T & num)
    {
#ifdef SE_CLASS1
        checkType<T>();
#endif

        // reduce over MPI

        // Create one request
        req.add();

        // reduce
        MPI_IallreduceW<T>::reduce(num,MPI_SUM,req.last());
    }

    template<typename T> void max(T & num)
    {
#ifdef SE_CLASS1
        checkType<T>();
#endif
        // reduce over MPI

        // Create one request
        req.add();

        // reduce
        MPI_IallreduceW<T>::reduce(num,MPI_MAX,req.last());
    }

    template<typename T> void min(T & num)
    {
#ifdef SE_CLASS1
        checkType<T>();
#endif
        // reduce over MPI

        // Create one request
        req.add();

        // reduce
        MPI_IallreduceW<T>::reduce(num,MPI_MIN,req.last());
    }

    void progressCommunication()
    {
        MPI_Status stat_t;
        int stat = false;
        MPI_SAFE_CALL(MPI_Iprobe(MPI_ANY_SOURCE,MPI_ANY_TAG,MPI_COMM_WORLD,&stat,&stat_t));

        // If I have an incoming message and is related to this NBX communication
        if (stat == true)
        {
            unsigned int i = (stat_t.MPI_TAG - SEND_SPARSE) / 131072 - NBX_prc_cnt_base;

            if (i >= NQUEUE || NBX_active[i] == NBX_Type::NBX_UNACTIVE || NBX_active[i] == NBX_Type::NBX_KNOWN  || NBX_active[i] == NBX_Type::NBX_KNOWN_PRC)
            {return;}

            int msize;

            // Get the message tag and size
            MPI_SAFE_CALL(MPI_Get_count(&stat_t,MPI_BYTE,&msize));

            // Ok we check if the TAG come from one of our send TAG
            if (stat_t.MPI_TAG >= (int)(SEND_SPARSE + NBX_prc_cnt_base*131072) && stat_t.MPI_TAG < (int)(SEND_SPARSE + (NBX_prc_cnt_base + NBX_prc_qcnt + 1)*131072))
            {
                // Get the pointer to receive the message
                void * ptr = this->NBX_prc_msg_alloc[i](msize,0,0,stat_t.MPI_SOURCE,rid[i],stat_t.MPI_TAG,this->NBX_prc_ptr_arg[i]);

                // Log the receiving request
                log.logRecv(stat_t);

                rid[i]++;

                // Check the pointer
#ifdef SE_CLASS2
                check_valid(ptr,msize);
#endif
                tot_recv += msize;
                #ifdef VCLUSTER_GARBAGE_INJECTOR
                    #if defined (__NVCC__) && !defined(CUDA_ON_CPU)
                    cudaPointerAttributes cpa;
                    auto error = cudaPointerGetAttributes(&cpa,ptr);
                    if (error == cudaSuccess)
                    {
                        if(cpa.type == cudaMemoryTypeDevice)
                        {cudaMemset(ptr,0xFF,msize);}
                        else
                        {memset(ptr,0xFF,msize);}
                    }
                    #else
                    memset(ptr,0xFF,msize);
                    #endif
                #endif
                MPI_SAFE_CALL(MPI_Recv(ptr,msize,MPI_BYTE,stat_t.MPI_SOURCE,stat_t.MPI_TAG,MPI_COMM_WORLD,&stat_t));

#ifdef SE_CLASS2
                check_valid(ptr,msize);
#endif
            }
        }

        // Check the status of all the MPI_issend and call the barrier if finished

        for (unsigned int i = 0 ; i < NQUEUE ; i++)
        {
            if (i >= NQUEUE || NBX_active[i] == NBX_Type::NBX_UNACTIVE || NBX_active[i] == NBX_Type::NBX_KNOWN || NBX_active[i] == NBX_Type::NBX_KNOWN_PRC)
            {continue;}

            if (NBX_prc_reached_bar_req[i] == false)
            {
                int flag = false;
                if (req.size() != 0)
                {MPI_SAFE_CALL(MPI_Testall(req.size(),&req.get(0),&flag,MPI_STATUSES_IGNORE));}
                else
                {flag = true;}

                // If all send has been completed
                if (flag == true)
                {MPI_SAFE_CALL(MPI_Ibarrier(MPI_COMM_WORLD,&bar_req));NBX_prc_reached_bar_req[i] = true;}
            }
        }
    }

    template<typename T> void sendrecvMultipleMessagesNBX(openfpm::vector< size_t > & prc,
                                                          openfpm::vector< T > & data,
                                                          openfpm::vector< size_t > & prc_recv,
                                                          openfpm::vector< size_t > & recv_sz ,
                                                          void * (* msg_alloc)(size_t,size_t,size_t,size_t,size_t,size_t,void *),
                                                          void * ptr_arg,
                                                          long int opt=NONE)
    {
        #ifdef VCLUSTER_PERF_REPORT
        timer nbx_timer;
        nbx_timer.start();

        #endif

        // Allocate the buffers

        for (size_t i = 0 ; i < prc.size() ; i++)
        {send(prc.get(i),SEND_SPARSE + NBX_cnt*131072,data.get(i).getPointer(),data.get(i).size());}

        for (size_t i = 0 ; i < prc_recv.size() ; i++)
        {
            void * ptr_recv = msg_alloc(recv_sz.get(i),0,0,prc_recv.get(i),i,SEND_SPARSE + NBX_cnt*131072,ptr_arg);

            recv(prc_recv.get(i),SEND_SPARSE + NBX_cnt*131072,ptr_recv,recv_sz.get(i));
        }

        execute();

        // Circular counter
        NBX_cnt = (NBX_cnt + 1) % nbx_cycle;

        #ifdef VCLUSTER_PERF_REPORT
        nbx_timer.stop();
        time_spent += nbx_timer.getwct();
        #endif
    }

    template<typename T> void sendrecvMultipleMessagesNBXAsync(openfpm::vector< size_t > & prc,
                                                          openfpm::vector< T > & data,
                                                          openfpm::vector< size_t > & prc_recv,
                                                          openfpm::vector< size_t > & recv_sz ,
                                                          void * (* msg_alloc)(size_t,size_t,size_t,size_t,size_t,size_t,void *),
                                                          void * ptr_arg,
                                                          long int opt=NONE)
    {
        NBX_prc_qcnt++;
        if (NBX_prc_qcnt >= NQUEUE)
        {
            std::cout << __FILE__ << ":" << __LINE__ << " error you can queue at most " << NQUEUE << " asychronous communication functions " << std::endl;
            return;
        }

        // Allocate the buffers

        for (size_t i = 0 ; i < prc.size() ; i++)
        {send(prc.get(i),SEND_SPARSE + NBX_cnt*131072,data.get(i).getPointer(),data.get(i).size());}

        for (size_t i = 0 ; i < prc_recv.size() ; i++)
        {
            void * ptr_recv = msg_alloc(recv_sz.get(i),0,0,prc_recv.get(i),i,SEND_SPARSE + NBX_cnt*131072,ptr_arg);

            recv(prc_recv.get(i),SEND_SPARSE + NBX_cnt*131072,ptr_recv,recv_sz.get(i));
        }

        NBX_active[NBX_prc_qcnt] = NBX_Type::NBX_KNOWN;
        if (NBX_prc_qcnt == 0)
        {NBX_prc_cnt_base = NBX_cnt;}
    }

    template<typename T>
    void sendrecvMultipleMessagesNBX(openfpm::vector< size_t > & prc,
                                     openfpm::vector< T > & data,
                                     void * (* msg_alloc)(size_t,size_t,size_t,size_t,size_t,size_t,void *),
                                     void * ptr_arg, long int opt=NONE)
    {
#ifdef SE_CLASS1
        checkType<typename T::value_type>();
#endif

        // resize the pointer list
        ptr_send[NBX_prc_qcnt+1].resize(prc.size());
        sz_send[NBX_prc_qcnt+1].resize(prc.size());

        for (size_t i = 0 ; i < prc.size() ; i++)
        {
            ptr_send[NBX_prc_qcnt+1].get(i) = data.get(i).getPointer();
            sz_send[NBX_prc_qcnt+1].get(i) = data.get(i).size() * sizeof(typename T::value_type);
        }

        sendrecvMultipleMessagesNBX(prc.size(),(size_t *)sz_send[NBX_prc_qcnt+1].getPointer(),(size_t *)prc.getPointer(),(void **)ptr_send[NBX_prc_qcnt+1].getPointer(),msg_alloc,ptr_arg,opt);
    }

    template<typename T>
    void sendrecvMultipleMessagesNBXAsync(openfpm::vector< size_t > & prc,
                                     openfpm::vector< T > & data,
                                     void * (* msg_alloc)(size_t,size_t,size_t,size_t,size_t,size_t,void *),
                                     void * ptr_arg, long int opt=NONE)
    {
#ifdef SE_CLASS1
        checkType<typename T::value_type>();
#endif
        // resize the pointer list
        ptr_send[NBX_prc_qcnt+1].resize(prc.size());
        sz_send[NBX_prc_qcnt+1].resize(prc.size());

        for (size_t i = 0 ; i < prc.size() ; i++)
        {
            ptr_send[NBX_prc_qcnt+1].get(i) = data.get(i).getPointer();
            sz_send[NBX_prc_qcnt+1].get(i) = data.get(i).size() * sizeof(typename T::value_type);
        }

        sendrecvMultipleMessagesNBXAsync(prc.size(),(size_t *)sz_send[NBX_prc_qcnt+1].getPointer(),(size_t *)prc.getPointer(),(void **)ptr_send[NBX_prc_qcnt+1].getPointer(),msg_alloc,ptr_arg,opt);
    }

    void sendrecvMultipleMessagesNBX(size_t n_send , size_t sz[],
                                     size_t prc[] , void * ptr[],
                                     size_t n_recv, size_t prc_recv[] ,
                                     size_t sz_recv[] ,void * (* msg_alloc)(size_t,size_t,size_t,size_t,size_t, size_t,void *),
                                     void * ptr_arg, long int opt=NONE)
    {
        #ifdef VCLUSTER_PERF_REPORT
        timer nbx_timer;
        nbx_timer.start();

        #endif

        // Allocate the buffers

        for (size_t i = 0 ; i < n_send ; i++)
        {send(prc[i],SEND_SPARSE + NBX_cnt*131072,ptr[i],sz[i]);}

        for (size_t i = 0 ; i < n_recv ; i++)
        {
            void * ptr_recv = msg_alloc(sz_recv[i],0,0,prc_recv[i],i,SEND_SPARSE + NBX_cnt*131072,ptr_arg);

            recv(prc_recv[i],SEND_SPARSE + NBX_cnt*131072,ptr_recv,sz_recv[i]);
        }

        execute();

        // Circular counter
        NBX_cnt = (NBX_cnt + 1) % nbx_cycle;

        #ifdef VCLUSTER_PERF_REPORT
        nbx_timer.stop();
        time_spent += nbx_timer.getwct();
        #endif
    }

    void sendrecvMultipleMessagesNBXAsync(size_t n_send , size_t sz[],
                                     size_t prc[] , void * ptr[],
                                     size_t n_recv, size_t prc_recv[] ,
                                     size_t sz_recv[] ,void * (* msg_alloc)(size_t,size_t,size_t,size_t,size_t, size_t,void *),
                                     void * ptr_arg, long int opt=NONE)
    {
        NBX_prc_qcnt++;
        if (NBX_prc_qcnt >= NQUEUE)
        {
            std::cout << __FILE__ << ":" << __LINE__ << " error you can queue at most " << NQUEUE << " asychronous communication functions " << std::endl;
            return;
        }

        // Allocate the buffers

        for (size_t i = 0 ; i < n_send ; i++)
        {send(prc[i],SEND_SPARSE + NBX_cnt*131072,ptr[i],sz[i]);}

        for (size_t i = 0 ; i < n_recv ; i++)
        {
            void * ptr_recv = msg_alloc(sz_recv[i],0,0,prc_recv[i],i,SEND_SPARSE + NBX_cnt*131072,ptr_arg);

            recv(prc_recv[i],SEND_SPARSE + NBX_cnt*131072,ptr_recv,sz_recv[i]);
        }

        NBX_active[NBX_prc_qcnt] = NBX_Type::NBX_KNOWN;
        if (NBX_prc_qcnt == 0)
        {NBX_prc_cnt_base = NBX_cnt;}
    }

    openfpm::vector<size_t> sz_recv_tmp;

    void sendrecvMultipleMessagesNBX(size_t n_send , size_t sz[], size_t prc[] ,
                                     void * ptr[], size_t n_recv, size_t prc_recv[] ,
                                     void * (* msg_alloc)(size_t,size_t,size_t,size_t,size_t,size_t,void *),
                                     void * ptr_arg, long int opt=NONE)
    {
        #ifdef VCLUSTER_PERF_REPORT
        timer nbx_timer;
        nbx_timer.start();
        #endif

        sz_recv_tmp.resize(n_recv);

        // First we understand the receive size for each processor

        for (size_t i = 0 ; i < n_send ; i++)
        {send(prc[i],SEND_SPARSE + NBX_cnt*131072,&sz[i],sizeof(size_t));}

        for (size_t i = 0 ; i < n_recv ; i++)
        {recv(prc_recv[i],SEND_SPARSE + NBX_cnt*131072,&sz_recv_tmp.get(i),sizeof(size_t));}

        execute();

        // Circular counter
        NBX_cnt = (NBX_cnt + 1) % nbx_cycle;

        // Allocate the buffers

        for (size_t i = 0 ; i < n_send ; i++)
        {send(prc[i],SEND_SPARSE + NBX_cnt*131072,ptr[i],sz[i]);}

        for (size_t i = 0 ; i < n_recv ; i++)
        {
            void * ptr_recv = msg_alloc(sz_recv_tmp.get(i),0,0,prc_recv[i],i,0,ptr_arg);

            recv(prc_recv[i],SEND_SPARSE + NBX_cnt*131072,ptr_recv,sz_recv_tmp.get(i));
        }

        execute();

        // Circular counter
        NBX_cnt = (NBX_cnt + 1) % nbx_cycle;

        #ifdef VCLUSTER_PERF_REPORT
        nbx_timer.stop();
        time_spent += nbx_timer.getwct();
        #endif
    }

    void sendrecvMultipleMessagesNBXAsync(size_t n_send , size_t sz[], size_t prc[] ,
                                     void * ptr[], size_t n_recv, size_t prc_recv[] ,
                                     void * (* msg_alloc)(size_t,size_t,size_t,size_t,size_t,size_t,void *),
                                     void * ptr_arg, long int opt=NONE)
    {
        NBX_prc_qcnt++;
        if (NBX_prc_qcnt >= NQUEUE)
        {
            std::cout << __FILE__ << ":" << __LINE__ << " error you can queue at most " << NQUEUE << " asychronous communication functions " << std::endl;
            return;
        }

        sz_recv_tmp.resize(n_recv);

        // First we understand the receive size for each processor

        for (size_t i = 0 ; i < n_send ; i++)
        {send(prc[i],SEND_SPARSE + NBX_cnt*131072,&sz[i],sizeof(size_t));}

        for (size_t i = 0 ; i < n_recv ; i++)
        {recv(prc_recv[i],SEND_SPARSE + NBX_cnt*131072,&sz_recv_tmp.get(i),sizeof(size_t));}


        NBX_prc_n_send[NBX_prc_qcnt] = n_send;
        NBX_prc_prc[NBX_prc_qcnt] = prc;
        NBX_prc_ptr[NBX_prc_qcnt] = ptr;
        NBX_prc_sz[NBX_prc_qcnt] = sz;
        NBX_prc_n_recv[NBX_prc_qcnt] = n_recv;
        NBX_prc_prc_recv[NBX_prc_qcnt] = prc_recv;
        NBX_prc_msg_alloc[NBX_prc_qcnt] = msg_alloc;
        NBX_prc_ptr_arg[NBX_prc_qcnt] = ptr_arg;

        NBX_active[NBX_prc_qcnt] = NBX_Type::NBX_KNOWN_PRC;
        if (NBX_prc_qcnt == 0)
        {NBX_prc_cnt_base = NBX_cnt;}
    }

    void sendrecvMultipleMessagesNBX(size_t n_send , size_t sz[],
                                     size_t prc[] , void * ptr[],
                                     void * (* msg_alloc)(size_t,size_t,size_t,size_t,size_t,size_t,void *),
                                     void * ptr_arg, long int opt = NONE)
    {
        #ifdef VCLUSTER_PERF_REPORT
        timer nbx_timer;
        nbx_timer.start();

        #endif

        NBX_prc_qcnt++;
        if (NBX_prc_qcnt != 0)
        {
            std::cout << __FILE__ << ":" << __LINE__ << " error there are some asynchronous call running you have to complete them before go back to synchronous" << std::endl;
            return;
        }

        queue_all_sends(n_send,sz,prc,ptr);

        this->NBX_prc_ptr_arg[NBX_prc_qcnt] = ptr_arg;
        this->NBX_prc_msg_alloc[NBX_prc_qcnt] = msg_alloc;

        rid[NBX_prc_qcnt] = 0;
        int flag = false;

        NBX_prc_reached_bar_req[NBX_prc_qcnt] = false;
        NBX_active[NBX_prc_qcnt] = NBX_Type::NBX_UNKNOWN;
        NBX_prc_cnt_base = NBX_cnt;

        log.start(10);

        // Wait that all the send are acknowledge
        do
        {
            progressCommunication();

            // Check if all processor reached the async barrier
            if (NBX_prc_reached_bar_req[NBX_prc_qcnt])
            {MPI_SAFE_CALL(MPI_Test(&bar_req,&flag,&bar_stat))};

            // produce a report if communication get stuck
            log.NBXreport(NBX_cnt,req,NBX_prc_reached_bar_req[NBX_prc_qcnt],bar_stat);

        } while (flag == false);

        // Remove the executed request

        req.clear();
        stat.clear();
        log.clear();

        // Circular counter
        NBX_cnt = (NBX_cnt + 1) % nbx_cycle;
        NBX_prc_qcnt = -1;

        #ifdef VCLUSTER_PERF_REPORT
        nbx_timer.stop();
        time_spent += nbx_timer.getwct();
        #endif
    }

    void sendrecvMultipleMessagesNBXAsync(size_t n_send , size_t sz[],
                                     size_t prc[] , void * ptr[],
                                     void * (* msg_alloc)(size_t,size_t,size_t,size_t,size_t,size_t,void *),
                                     void * ptr_arg, long int opt = NONE)
    {
        NBX_prc_qcnt++;
        queue_all_sends(n_send,sz,prc,ptr);

        this->NBX_prc_ptr_arg[NBX_prc_qcnt] = ptr_arg;
        this->NBX_prc_msg_alloc[NBX_prc_qcnt] = msg_alloc;

        rid[NBX_prc_qcnt] = 0;

        NBX_prc_reached_bar_req[NBX_prc_qcnt] = false;
        NBX_active[NBX_prc_qcnt] = NBX_Type::NBX_UNKNOWN;

        log.start(10);
        if (NBX_prc_qcnt == 0)
        {NBX_prc_cnt_base = NBX_cnt;}

        return;
    }

    void sendrecvMultipleMessagesNBXWait()
    {
        for (unsigned int j = 0 ; j < NQUEUE ; j++)
        {
            if (NBX_active[j] == NBX_Type::NBX_UNACTIVE)
            {continue;}

            if (NBX_active[j] == NBX_Type::NBX_KNOWN_PRC)
            {
                execute();

                // Circular counter
                NBX_cnt = (NBX_cnt + 1) % nbx_cycle;

                // Allocate the buffers

                for (size_t i = 0 ; i < NBX_prc_n_send[j] ; i++)
                {send(NBX_prc_prc[j][i],SEND_SPARSE + NBX_cnt*131072,NBX_prc_ptr[j][i],NBX_prc_sz[j][i]);}

                for (size_t i = 0 ; i < NBX_prc_n_recv[j] ; i++)
                {
                    void * ptr_recv = NBX_prc_msg_alloc[j](sz_recv_tmp.get(i),0,0,NBX_prc_prc_recv[j][i],i,0,this->NBX_prc_ptr_arg[j]);

                    recv(NBX_prc_prc_recv[j][i],SEND_SPARSE + NBX_cnt*131072,ptr_recv,sz_recv_tmp.get(i));
                }

                NBX_active[j] = NBX_Type::NBX_KNOWN;
            }

            if (NBX_active[j] == NBX_Type::NBX_KNOWN)
            {
                execute();

                // Circular counter
                NBX_cnt = (NBX_cnt + 1) % nbx_cycle;
                NBX_active[j] = NBX_Type::NBX_UNACTIVE;

                continue;
            }

            int flag = false;

            // Wait that all the send are acknowledge
            do
            {
                progressCommunication();

                // Check if all processor reached the async barrier
                if (NBX_prc_reached_bar_req[j])
                {MPI_SAFE_CALL(MPI_Test(&bar_req,&flag,&bar_stat))};

                // produce a report if communication get stuck
                log.NBXreport(NBX_cnt,req,NBX_prc_reached_bar_req[j],bar_stat);

            } while (flag == false);

            // Remove the executed request

            req.clear();
            stat.clear();
            log.clear();

            // Circular counter
            NBX_cnt = (NBX_cnt + 1) % nbx_cycle;
            NBX_active[j] = NBX_Type::NBX_UNACTIVE;

        }

        NBX_prc_qcnt = -1;
        return;
    }

    bool send(size_t proc, size_t tag, const void * mem, size_t sz)
    {
        // send over MPI

        // Create one request
        req.add();

        // send
        MPI_IsendWB::send(proc,SEND_RECV_BASE + tag,mem,sz,req.last());

        return true;
    }


    template<typename T, typename Mem, template<typename> class gr> bool send(size_t proc, size_t tag, openfpm::vector<T,Mem,gr> & v)
    {
#ifdef SE_CLASS1
        checkType<T>();
#endif

        // send over MPI

        // Create one request
        req.add();

        // send
        MPI_IsendW<T,Mem,gr>::send(proc,SEND_RECV_BASE + tag,v,req.last());

        return true;
    }

    bool recv(size_t proc, size_t tag, void * v, size_t sz)
    {
        // recv over MPI

        // Create one request
        req.add();

        // receive
        MPI_IrecvWB::recv(proc,SEND_RECV_BASE + tag,v,sz,req.last());

        return true;
    }

    template<typename T, typename Mem, template<typename> class gr> bool recv(size_t proc, size_t tag, openfpm::vector<T,Mem,gr> & v)
    {
#ifdef SE_CLASS1
            checkType<T>();
#endif

            // recv over MPI

            // Create one request
            req.add();

            // receive
            MPI_IrecvW<T>::recv(proc,SEND_RECV_BASE + tag,v,req.last());

            return true;
    }

    template<typename T, typename Mem, template<typename> class gr> bool allGather(T & send, openfpm::vector<T,Mem,gr> & v)
    {
#ifdef SE_CLASS1
        checkType<T>();
#endif

        // Create one request
        req.add();

        // Number of processors
        v.resize(getProcessingUnits());

        // gather
        MPI_IAllGatherW<T>::gather(&send,1,v.getPointer(),1,req.last());

        return true;
    }

    template<typename T, typename Mem, template<typename> class layout_base >
    bool Bcast(openfpm::vector<T,Mem,layout_base> & v, size_t root)
    {
#ifdef SE_CLASS1
        checkType<T>();
#endif

        b_cast_helper<openfpm::vect_isel<T>::value == STD_VECTOR || is_layout_mlin<layout_base<T>>::value >::bcast_(req,v,root);

        return true;
    }

    void execute()
    {
        // if req == 0 return
        if (req.size() == 0)
            return;

        // Wait for all the requests
        stat.resize(req.size());

        MPI_SAFE_CALL(MPI_Waitall(req.size(),&req.get(0),&stat.get(0)));

        // Remove executed request and status
        req.clear();
        stat.clear();
    }

    void clear()
    {
        for (size_t i = 0 ; i < NQUEUE ; i++)
        {recv_buf[i].clear();}
    }
};




#endif