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"

void bt_sighandler(int sig, siginfo_t * info, void * ctx);

class Vcluster: public Vcluster_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)
        {
            vcl.process_receive_buffer_with_prp<op,T,S,layout_base,prp...>(recv,sz_recv,sz_recv_byte,op_param);
        }
    };

    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)
    {
        openfpm::vector<size_t> sz_recv_byte(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
        openfpm::vector<const void *> send_buf;
        openfpm::vector<size_t> send_sz_byte;

        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<T>::value>::value,op, T, S, layout_base>::packingRequest(send.get(i), req, send_sz_byte);
            tot_size += req;
        }

        HeapMemory pmem;

        ExtPreAlloc<HeapMemory> & mem = *(new ExtPreAlloc<HeapMemory>(tot_size,pmem));
        mem.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<T>::value>::value, op, T, S, layout_base>::packing(mem, send.get(i), sts, send_buf);
        }

        // receive information
        base_info bi(&recv_buf,prc_recv,sz_recv_byte);

        // 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.get(i) = sz_recv.get(i) * sizeof(typename T::value_type);
                }
                else
                {std::cout << __FILE__ << ":" << __LINE__ << " Error " << demangle(typeid(T).name()) << " the type does not work with the option or NO_CHANGE_ELEMENTS" << std::endl;}

                Vcluster_base::sendrecvMultipleMessagesNBX(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.getPointer(),msg_alloc_known,(void *)&bi);
            }
            else
            {
                Vcluster_base::sendrecvMultipleMessagesNBX(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 *)&bi);
                sz_recv_byte = sz_recv_tmp;
            }
        }
        else
        {
            prc_recv.clear();
            sendrecvMultipleMessagesNBX(prc_send.size(),(size_t *)send_sz_byte.getPointer(),(size_t *)prc_send.getPointer(),(void **)send_buf.getPointer(),msg_alloc,(void *)&bi);
        }

        // Reorder the buffer
        reorder_buffer(prc_recv,sz_recv_byte);

        mem.decRef();
        delete &mem;
    }


    void reset_recv_buf()
    {
        for (size_t i = 0 ; i < recv_buf.size() ; i++)
            recv_buf.get(i).resize(0);

        recv_buf.resize(0);
    }

    struct base_info
    {
        openfpm::vector<BHeapMemory> * recv_buf;
        openfpm::vector<size_t> & prc;
        openfpm::vector<size_t> & sz;

        base_info(openfpm::vector<BHeapMemory> * recv_buf, openfpm::vector<size_t> & prc, openfpm::vector<size_t> & sz)
        :recv_buf(recv_buf),prc(prc),sz(sz)
        {}
    };

    static void * msg_alloc(size_t msg_i ,size_t total_msg, size_t total_p, size_t i, size_t ri, void * ptr)
    {
        base_info & rinfo = *(base_info *)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);

        // return the pointer
        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, void * ptr)
    {
        base_info & rinfo = *(base_info *)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)
    {
        if (sz != NULL)
            sz->resize(recv_buf.size());

        pack_unpack_cond_with_prp<has_max_prop<T, has_value_type<T>::value>::value,op, T, S, layout_base, prp... >::unpacking(recv, recv_buf, sz, sz_byte, op_param);
    }

    public:

    Vcluster(int *argc, char ***argv)
    :Vcluster_base(argc,argv)
    {
    }

    template<typename T, typename S> bool SGather(T & send, S & recv,size_t root)
    {
        openfpm::vector<size_t> prc;
        openfpm::vector<size_t> sz;

        return SGather(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 (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;

            // receive information
            base_info bi(&recv_buf,prc,sz);

            // Send and recv multiple messages
            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<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,&sz,NULL,opa);

            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;
            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<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<T>::value>::value,op_ssend_recv_add<void>, T, S, layout_base>::packing(mem, send, sts, send_buf);

            // receive information
            base_info bi(NULL,prc,sz);

            // Send and recv multiple messages
            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;
    }

    template<typename T, typename S> 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 (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);
            }

            // receive information
            base_info bi(&recv_buf,prc,sz);

            // Send and recv multiple messages
            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<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>(*this,recv,NULL,NULL,opa);
        }
        else
        {
            // The non-root receive
            openfpm::vector<size_t> send_req;

            // receive information
            base_info bi(&recv_buf,prc,sz);

            // Send and recv multiple messages
            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<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>(*this,recv,NULL,NULL,opa);
        }

        return true;
    }
    
    void reorder_buffer(openfpm::vector<size_t> & prc, openfpm::vector<size_t> & sz_recv)
    {

        struct recv_buff_reorder
        {
            size_t proc;

            size_t pos;

            recv_buff_reorder()
            :proc(0),pos(0)
            {};

            bool operator<(const recv_buff_reorder & rd) const
            {
                return proc < rd.proc;
            }
        };

        openfpm::vector<recv_buff_reorder> rcv;

        rcv.resize(recv_buf.size());

        for (size_t i = 0 ; i < rcv.size() ; i++)
        {
            rcv.get(i).proc = prc.get(i);
            rcv.get(i).pos = i;
        }

        // we sort based on processor
        rcv.sort();

        openfpm::vector<BHeapMemory> 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(recv_buf.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
        recv_buf.swap(recv_ord);
        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);

        // we generate the list of the properties to pack
        typedef typename ::generate_indexes<int, has_max_prop<T, has_value_type<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);

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

        // 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,opa);

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

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

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

        // process the received information
        process_receive_buffer_with_prp<op,T,S,layout_base,prp...>(recv,NULL,NULL,op_param);

        return true;
    }

};



// Function to initialize the global VCluster //

extern Vcluster * global_v_cluster_private;

static inline void init_global_v_cluster_private(int *argc, char ***argv)
{
    if (global_v_cluster_private == NULL)
        global_v_cluster_private = new Vcluster(argc,argv);
}

static inline void delete_global_v_cluster_private()
{
    delete global_v_cluster_private;
}

static inline Vcluster & create_vcluster()
{
#ifdef SE_CLASS1

    if (global_v_cluster_private == NULL)
        std::cerr << __FILE__ << ":" << __LINE__ << " Error you must call openfpm_init before using any distributed data structures";

#endif

    return *global_v_cluster_private;
}



static inline bool is_openfpm_init()
{
    return ofp_initialized;
}

static inline void openfpm_init(int *argc, char ***argv)
{
#ifdef HAVE_PETSC

    PetscInitialize(argc,argv,NULL,NULL);

#endif

    init_global_v_cluster_private(argc,argv);

#ifdef SE_CLASS1
    std::cout << "OpenFPM is compiled with debug mode LEVEL:1. Remember to remove SE_CLASS1 when you go in production" << std::endl;
#endif

#ifdef SE_CLASS2
    std::cout << "OpenFPM is compiled with debug mode LEVEL:2. Remember to remove SE_CLASS2 when you go in production" << std::endl;
#endif

#ifdef SE_CLASS3
    std::cout << "OpenFPM is compiled with debug mode LEVEL:3. Remember to remove SE_CLASS3 when you go in production" << std::endl;
#endif

    // install segmentation fault signal handler

    struct sigaction sa;

    sa.sa_sigaction = bt_sighandler;
    sigemptyset(&sa.sa_mask);
    sa.sa_flags = SA_RESTART;

    sigaction(SIGSEGV, &sa, NULL);

    if (*argc != 0)
        program_name = std::string(*argv[0]);

    ofp_initialized = true;
}


static inline void openfpm_finalize()
{
#ifdef HAVE_PETSC

    PetscFinalize();

#endif

    delete_global_v_cluster_private();
    ofp_initialized = false;
}


#endif