VCluster_base.hpp

#ifndef VCLUSTER_BASE_HPP_
#define VCLUSTER_BASE_HPP_

#include "config.h"
#include <mpi.h>
#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"

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

#define MSG_LENGTH 1024
#define MSG_SEND_RECV 1025
#define SEND_SPARSE 4096
#define NONE 1
#define NEED_ALL_SIZE 2

#define SERIVCE_MESSAGE_TAG 16384
#define SEND_RECV_BASE 8192
#define GATHER_BASE 24576

#define RECEIVE_KNOWN 4
#define KNOWN_ELEMENT_OR_BYTE 8

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


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

class Vcluster_base
{
    Vcluster_log log;

    size_t NBX_cnt;

    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;

    // Object array


    // Single objects

    int m_size;
    int m_rank;

    int numPE = 1;

    std::vector<red> r;

    openfpm::vector<void *> ptr_send;

    openfpm::vector<size_t> sz_send;

    MPI_Request bar_req;

    MPI_Status bar_stat;

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

    Vcluster_base(const Vcluster_base &)
    :NBX_cnt(0)
    {};

protected:

    openfpm::vector<BHeapMemory> recv_buf;

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";
                }
            }
        }
    }

    Vcluster_base(int *argc, char ***argv)
    :NBX_cnt(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);
        }

        // 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 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

    MPI_Comm getMPIComm()
    {
        return MPI_COMM_WORLD;
    }

    size_t getProcessingUnits()
    {
        return m_size*numPE;
    }

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

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

    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,void *),
                                                          void * ptr_arg,
                                                          long int opt=NONE)
    {
        // Allocate the buffers

        for (size_t i = 0 ; i < prc.size() ; i++)
            send(prc.get(i),SEND_SPARSE + NBX_cnt,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,ptr_arg);

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

        execute();

        // Circular counter
        NBX_cnt = (NBX_cnt + 1) % 1024;
    }


    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,void *),
                                     void * ptr_arg, long int opt=NONE)
    {
#ifdef SE_CLASS1
        checkType<typename T::value_type>();
#endif
        // resize the pointer list
        ptr_send.resize(prc.size());
        sz_send.resize(prc.size());

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

        sendrecvMultipleMessagesNBX(prc.size(),(size_t *)sz_send.getPointer(),(size_t *)prc.getPointer(),(void **)ptr_send.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,void *),
                                     void * ptr_arg, long int opt=NONE)
    {
        // Allocate the buffers

        for (size_t i = 0 ; i < n_send ; i++)
            send(prc[i],SEND_SPARSE + NBX_cnt,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,ptr_arg);

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

        execute();

        // Circular counter
        NBX_cnt = (NBX_cnt + 1) % 1024;
    }

    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,void *),
                                     void * ptr_arg, long int opt=NONE)
    {
        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,&sz[i],sizeof(size_t));}

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

        execute();

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

        // Allocate the buffers

        for (size_t i = 0 ; i < n_send ; i++)
        {send(prc[i],SEND_SPARSE + NBX_cnt,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,ptr_arg);

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

        execute();

        // Circular counter
        NBX_cnt = (NBX_cnt + 1) % 1024;
    }

    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,void *),
                                     void * ptr_arg, long int opt = NONE)
    {
        if (stat.size() != 0 || req.size() != 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";


        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];
                MPI_SAFE_CALL(MPI_Issend(ptr[i], sz[i], MPI_BYTE, prc[i], SEND_SPARSE + NBX_cnt, MPI_COMM_WORLD,&req.last()));
                log.logSend(prc[i]);
            }
        }

        size_t rid = 0;
        int flag = false;

        bool reached_bar_req = false;

        log.start(10);

        // Wait that all the send are acknowledge
        do
        {

            // flag that notify that this processor reach the barrier
            // Barrier request

            MPI_Status stat_t;
            int stat = false;
            MPI_SAFE_CALL(MPI_Iprobe(MPI_ANY_SOURCE,SEND_SPARSE + NBX_cnt,MPI_COMM_WORLD,&stat,&stat_t));

            // If I have an incoming message and is related to this NBX communication
            if (stat == true)
            {
                // Get the message size
                int msize;
                MPI_SAFE_CALL(MPI_Get_count(&stat_t,MPI_BYTE,&msize));

                // Get the pointer to receive the message
                void * ptr = msg_alloc(msize,0,0,stat_t.MPI_SOURCE,rid,ptr_arg);

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

                rid++;

                // Check the pointer
#ifdef SE_CLASS2
                check_valid(ptr,msize);
#endif
                tot_recv += msize;
                MPI_SAFE_CALL(MPI_Recv(ptr,msize,MPI_BYTE,stat_t.MPI_SOURCE,SEND_SPARSE+NBX_cnt,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

            if (reached_bar_req == 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));reached_bar_req = true;}
            }

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

            // produce a report if communication get stuck
            log.NBXreport(NBX_cnt,req,reached_bar_req,bar_stat);

        } while (flag == false);

        // Remove the executed request

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

        // Circular counter
        NBX_cnt = (NBX_cnt + 1) % 1024;
    }

    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, typename 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, typename 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, typename 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, typename gr> bool Bcast(openfpm::vector<T,Mem,gr> & v, size_t root)
    {
#ifdef SE_CLASS1
        checkType<T>();
#endif

        // Create one request
        req.add();

        // gather
        MPI_IBcastW<T>::bcast(root,v,req.last());

        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()
    {
        recv_buf.clear();
    }
};




#endif