dist_map_graph.hpp

/*
 * dist_map_graph.hpp
 *
 *  Created on: Dec 10, 2015
 *      Author: Antonio Leo
 *
 *
 *  The distributed graph, is a graph distributed across processors, each processor store part of the graph
 *
 * ## Dictionary
 *
 * * local vertex id, is the index of the vertex in the local graph
 * * vertex id is the "unique" index in the distribution vector (vtxdist)
 * * global id is the "unique" topological id of the vertex, it never change
 *
 *
 *
 *  Graph structure that store a CSR graph format
 *
 *  This Graph format is suppose to have a list of vertex that store an index x that indicate where
 *  in the adjacency list we have the list of all the neighborhood vertex, plus an adjacency list
 *  that store all the neighborhood for each vertex:
 *
 *  In reality inside DistGraph_CSR
 *
 *  VertexList store the Vertex index that indicate the end of the neighborhood list,
 *  the start is indicated by i * v_slot (id of the vertex, v_slot maximum number of
 *  neighborhood for a vertex)
 *
 *  EdgeList store for each vertex at position i * v_slot (id of the vertex, v_slot maximum
 *  number of neighborhood for a vertex) the list of all the neighborhood of the vertex i
 *
 *  Example
 *
 *  Suppose an undirected graph of 4 vertex
 *
 *  1 -> 2 3 4
 *  2 -> 1
 *  3 -> 4 1
 *  4 -> 1 3
 *
 *  suppose that v_slot is 4 ( each vertex has less tha 4 neighborhood  )
 *
 *  we will have
 *
 *  Vertex list 3 5 10 14
 *  Edge list   2 3 4 0 1 0 0 0 4 1 0 0 1 3 0 0
 *
 *  Vertex properties and edge properties are stored in a separate structure
 *
 * ## Dictionary
 *
 * The distributed
 *
 * * local vertex id is the index of the vertex in
 *
 */
 
#ifndef DIST_MAP_GRAPH_HPP_
#define DIST_MAP_GRAPH_HPP_
 
#include "Vector/map_vector.hpp"
#include "Graph/map_graph.hpp"
#include <unordered_map>
#include "Packer_Unpacker/Packer.hpp"
#include "Packer_Unpacker/Unpacker.hpp"
#include "VCluster/VCluster.hpp"
 
#define NO_EDGE -1
#define DIST_GRAPH_ERROR 7001
 
template<typename V, typename E,
         typename Memory,
         typename layout_v,
         typename layout_e,
         template<typename> class layout_v_base,
         template<typename> class layout_e_base,
         typename grow_p>
class DistGraph_CSR;
 
class v_info
{
public:
    typedef boost::fusion::vector<size_t, size_t> type;
 
    type data;
 
    static const unsigned int id = 0;
    static const unsigned int gid = 1;
    static const unsigned int max_prop = 2;
 
    v_info()
    {
    }
 
    inline void setid(size_t id_)
    {
        boost::fusion::at_c<0>(data) = id_;
    }
 
    inline void setgid(size_t gid_)
    {
        boost::fusion::at_c<1>(data) = gid_;
    }
 
    template<unsigned int id> inline auto get() -> decltype(boost::fusion::at_c < id > (data))
    {
        return boost::fusion::at_c<id>(data);
    }
 
    template<unsigned int id> inline auto get() const -> const decltype(boost::fusion::at_c < id > (data))
    {
        return boost::fusion::at_c<id>(data);
    }
 
    template<unsigned int dim, typename Mem> inline v_info(const encapc<dim, v_info, Mem> & p)
    {
        this->operator=(p);
    }
 
    template<unsigned int dim, typename Mem> inline v_info & operator=(const encapc<dim, v_info, Mem> & p)
    {
        boost::fusion::at_c<0>(data) = p.template get<0>();
        boost::fusion::at_c<1>(data) = p.template get<1>();
 
        return *this;
    }
 
    static bool noPointers()
    {
        return true;
    }
};
 
class e_info
{
public:
    typedef boost::fusion::vector<size_t, size_t> type;
 
    type data;
 
    static const unsigned int sgid = 0;
    static const unsigned int dgid = 1;
    static const unsigned int max_prop = 2;
 
    e_info()
    {
    }
 
    template<unsigned int id> inline auto get() -> decltype(boost::fusion::at_c < id > (data))
    {
        return boost::fusion::at_c<id>(data);
    }
 
    template<unsigned int id> inline auto get() const -> const decltype(boost::fusion::at_c < id > (data))
    {
        return boost::fusion::at_c<id>(data);
    }
 
    template<unsigned int dim, typename Mem> inline e_info(const encapc<dim, e_info, Mem> & p)
    {
        this->operator=(p);
    }
 
    template<unsigned int dim, typename Mem> inline e_info & operator=(const encapc<dim, e_info, Mem> & p)
    {
        boost::fusion::at_c<0>(data) = p.template get<0>();
        boost::fusion::at_c<1>(data) = p.template get<1>();
 
        return *this;
    }
 
    static bool noPointers()
    {
        return true;
    }
};
 
template<typename V, typename E = no_edge,
         typename Memory = HeapMemory,
         typename layout_v = typename memory_traits_lin<V>::type,
         typename layout_e = typename memory_traits_lin<E>::type,
         template <typename> class layout_v_base = memory_traits_lin,
         template <typename> class layout_e_base = memory_traits_lin,
        typename grow_p = openfpm::grow_policy_double>
class DistGraph_CSR
{
    Vcluster<> & vcl;
 
    openfpm::vector<idx_t> vtxdist;
 
    openfpm::vector<idx_t> fvtxdist;
 
    size_t v_slot;
 
    openfpm::vector<V, Memory,layout_v_base,grow_p, openfpm::vect_isel<V>::value> v;
 
    openfpm::vector<v_info, Memory, memory_traits_lin, grow_p, openfpm::vect_isel<v_info>::value> v_m;
 
    openfpm::vector<size_t, Memory, layout_v_base, grow_p, openfpm::vect_isel<size_t>::value> v_l;
 
    openfpm::vector<E, Memory, layout_e_base, grow_p, openfpm::vect_isel<E>::value> e;
 
    openfpm::vector<e_info, Memory, layout_e_base, grow_p, openfpm::vect_isel<e_info>::value> e_m;
 
    openfpm::vector<e_map, Memory, layout_e_base, grow_p, openfpm::vect_isel<e_map>::value> e_l;
 
    openfpm::vector<E, Memory, layout_e_base, grow_p, openfpm::vect_isel<E>::value> e_invalid;
 
    std::unordered_map<size_t, size_t> id2glb;
 
    std::unordered_map<size_t, size_t> glb2id;
 
    std::unordered_map<size_t, size_t> glb2loc;
 
    struct SendGraphPack
    {
        openfpm::vector<V> send_v;
        openfpm::vector<v_info> send_v_m;
        openfpm::vector<E> send_e;
        openfpm::vector<e_info> send_e_m;
        openfpm::vector<size_t> send_el;
        openfpm::vector<size_t> send_es;
        bool isEmpty = true;
    };
 
    openfpm::vector<SendGraphPack> sgp;
 
    openfpm::vector<size_t> v_td;
 
    typedef struct
    {
        size_t id;
        // processor containing the vertex
        size_t pid;
    } GlobalVInfo;
 
    // Map of GlobalVInfo containing informations of vertices of the INITIAL distribution contained in this processor
    // ex. if this will contain the first 4 vertices of the distribution (0,1,2,3) it will maintain informations only about these vertices
    // The key is the vertex global id
    std::unordered_map<size_t, GlobalVInfo> glbi_map;
 
    openfpm::vector<openfpm::vector<size_t>> vr_queue;
 
    std::unordered_map<size_t, bool> ghs_map;
 
    typedef struct
    {
        size_t v1;
        size_t v2;
        size_t v1n;
        size_t v2n;
    } EdgeReq;
 
    openfpm::vector<EdgeReq> e_queue;
 
    template<typename CheckPolicy = NoCheck> inline size_t addEdge_(size_t v1, size_t v2)
    {
        // If v1 and v2 does not satisfy some criteria return
        if (CheckPolicy::valid(v1, v.size()) == false)
        {return (size_t)NO_EDGE;}
        if (CheckPolicy::valid(v2, v.size()) == false)
        {return (size_t)NO_EDGE;}
 
        // get the number of adjacent vertex
        size_t id_x_end = v_l.template get<0>(v1);
 
#ifdef DEBUG
 
        // Check that the edge does not already exist
 
        for (size_t s = 0; s < id_x_end; s++)
        {
            if (e_l.template get<e_map::vid>(v1 * v_slot + s) == v2)
            {
                std::cerr << "Error graph: the edge already exist\n";
            }
        }
#endif
 
        // Check if there is space for another edge
 
        if (id_x_end >= v_slot)
        {
            // Unfortunately there is not space we need to reallocate memory
            // Reallocate with double slot
 
            // Create an new Graph
            DistGraph_CSR<V, E> g_new(2 * v_slot, v.size());
 
            // Copy the graph
            for (size_t i = 0; i < v.size(); i++)
            {
                // copy the property from the old graph
                g_new.v.set(i, v, 2 * i);
            }
 
            // swap the new graph with the old one
            swap(g_new);
        }
 
        // Here we are sure than v and e has enough slots to store a new edge
        // Check that e_l has enough space to store new edge
        // should be always e.size == e_l.size
 
        if (id_x_end >= e_l.size())
        {
            // Resize the basic structure
            e_l.resize(v.size() * v_slot);
        }
 
        // add in e_l the adjacent vertex for v1 and fill the edge id
        e_l.template get<e_map::vid>(v1 * v_slot + id_x_end) = v2;
        e_l.template get<e_map::eid>(v1 * v_slot + id_x_end) = e.size();
 
        // add an empty edge
        e.resize(e.size() + 1);
        e_m.resize(e_m.size() + 1);
 
        // Increment the ending point
        ++v_l.template get<0>(v1);
 
        // return the created edge
        return e.size() - 1;
    }
 
    static void * gr_receive(size_t msg_i, size_t total_msg, size_t total_p, size_t i, size_t ri, size_t tag, void * ptr)
    {
        openfpm::vector<HeapMemory> *v = static_cast<openfpm::vector<HeapMemory> *>(ptr);
 
        v->get(i).allocate(msg_i);
 
        return v->get(i).getPointer();
 
    }
 
    static void * on_receive(size_t msg_i, size_t total_msg, size_t total_p, size_t i, size_t ri, size_t tag, void * ptr)
    {
        openfpm::vector<openfpm::vector<size_t>> *v = static_cast<openfpm::vector<openfpm::vector<size_t>> *>(ptr);
 
        v->get(i).resize(msg_i / sizeof(size_t));
 
        return &(v->get(i).get(0));
    }
 
    void resetExchange()
    {
        if (sgp.size() == vcl.getProcessingUnits())
        {
            for (size_t p = 0; p < vcl.getProcessingUnits(); ++p)
            {
                sgp.get(p).send_v.clear();
                sgp.get(p).send_v_m.clear();
                sgp.get(p).send_e.clear();
                sgp.get(p).send_e_m.clear();
                sgp.get(p).send_es.clear();
                sgp.get(p).send_el.clear();
                sgp.get(p).isEmpty = true;
            }
        }
        else
        {
            sgp.resize(vcl.getProcessingUnits());
 
            for (size_t p = 0; p < vcl.getProcessingUnits(); ++p)
            {
                openfpm::vector<V> s_v;
                openfpm::vector<v_info> s_v_m;
                openfpm::vector<E> s_e;
                openfpm::vector<e_info> s_e_m;
                openfpm::vector<size_t> s_el;
                openfpm::vector<size_t> s_es;
 
                sgp.get(p).send_v = s_v;
                sgp.get(p).send_v_m = s_v_m;
                sgp.get(p).send_e = s_e;
                sgp.get(p).send_e_m = s_e_m;
                sgp.get(p).send_es = s_es;
                sgp.get(p).send_el = s_el;
                sgp.get(p).isEmpty = true;
            }
        }
    }
 
    void deleteMovedVertices()
    {
        // Prepare new graph
        DistGraph_CSR recv_g;
 
        // Reset maps
        glb2loc.clear();
        glb2id.clear();
        id2glb.clear();
 
        size_t local_i = 0;
 
        // Add previous vertices without the sent ones
        for (size_t i = 0; i < this->getNVertex(); ++i)
        {
            if (!isToDelete(i))
            {
                recv_g.add_vertex(this->vertex(i), this->getVertexId(i), this->getVertexGlobalId(i));
 
                for (size_t j = 0; j < this->getNChilds(i); j++)
                {
                    recv_g.addEdge(local_i, this->getChild(i, j), this->getChildEdge(i, j), this->getChildInfo(i, j));
                }
                ++local_i;
            }
        }
 
        recv_g.fvtxdist = fvtxdist;
 
        // Swap temporary graph with the main one
        swap(recv_g);
 
        // Clear vertex to delete array
        v_td.clear();
    }
 
    bool isToDelete(size_t i)
    {
 
        for (size_t j = 0; j < v_td.size(); ++j)
        {
            if (i == v_td.get(j))
                return true;
        }
        return false;
    }
 
    size_t getVProcessor(size_t v)
    {
        for (size_t i = 1; i < vtxdist.size() - 1; ++i)
        {
            if (v < vtxdist.get(i))
            {
                return i - 1;
            }
        }
        return vcl.getProcessingUnits() - 1;
    }
 
    template<bool addAsGhosts>
    void exchangeVertices()
    {
        // If the exchange is not to retrieve ghost vertices delete the vertices this processor is sending
        if (!addAsGhosts)
            deleteMovedVertices();
 
        openfpm::vector<size_t> prc;
        openfpm::vector<size_t> size;
        openfpm::vector<void *> ptr;
        openfpm::vector<HeapMemory> packs(vcl.getProcessingUnits());
        openfpm::vector<HeapMemory *> to_release;
        openfpm::vector<ExtPreAlloc<HeapMemory> *> to_release_ext;
 
        // Total number of vertex to send
        size_t nvts = 0;
        for (size_t i = 0; i < vcl.getProcessingUnits(); i++)
        {
            nvts += sgp.get(i).send_v.size();
        }
 
        // For each processor
        for (size_t i = 0; i < vcl.getProcessingUnits(); i++)
        {
            // if nothing to send continue
            if (sgp.get(i).isEmpty)
                continue;
 
            // process to communicate with TODO remove pc
            size_t pc = i;
 
            size_t vp_size = sgp.get(pc).send_v.size();
 
            size_t req = 0;
 
            // prepare slot for number of vertices
            Packer<size_t, HeapMemory>::packRequest(req);
 
            for (size_t j = 0; j < vp_size; j++)
            {
                // prepare slot for vertex
                Packer<V, HeapMemory>::packRequest(req);
 
                // prepare slot info for vertex
                Packer<v_info, HeapMemory>::packRequest(req);
 
                // prepare slot for the number of children
                Packer<size_t, HeapMemory>::packRequest(req);
 
                // prepare slots for the children
                for (size_t k = 0; k < sgp.get(pc).send_es.get(j); k++)
                {
                    // prepare slot for edge
                    Packer<E, HeapMemory>::packRequest(req);
 
                    // prepare slot for edge info
                    Packer<e_info, HeapMemory>::packRequest(req);
 
                    // prepare slot for edge target id
                    Packer<size_t, HeapMemory>::packRequest(req);
                }
            }
 
            // allocate the memory
            HeapMemory & pmem = *(new HeapMemory());
//          pmem.allocate(req);
            ExtPreAlloc<HeapMemory> & mem = *(new ExtPreAlloc<HeapMemory>(req, pmem));
            mem.incRef();
 
            to_release.add(&pmem);
            to_release_ext.add(&mem);
 
            Pack_stat sts;
            size_t e_it = 0;
 
            // Pack total size
            Packer<size_t, HeapMemory>::pack(mem, vp_size, sts);
 
            for (size_t j = 0; j < vp_size; j++)
            {
                // Pack the vertex
                Packer<decltype(sgp.get(pc).send_v.get(0)), HeapMemory>::pack(mem, sgp.get(pc).send_v.get(j), sts);
 
                // Pack the vertex info
                Packer<decltype(sgp.get(pc).send_v_m.get(0)), HeapMemory>::pack(mem, sgp.get(pc).send_v_m.get(j), sts);
 
                // Pack size of the children
                Packer<size_t, HeapMemory>::pack(mem, sgp.get(pc).send_es.get(j), sts);
 
                // Pack children
                for (size_t k = 0; k < sgp.get(pc).send_es.get(j); k++)
                {
                    // Pack the edge
                    Packer<decltype(sgp.get(pc).send_e.get(0)), HeapMemory>::pack(mem, sgp.get(pc).send_e.get(e_it), sts);
 
                    // Pack the edge info
                    Packer<decltype(sgp.get(pc).send_e_m.get(0)), HeapMemory>::pack(mem, sgp.get(pc).send_e_m.get(e_it), sts);
 
                    // Pack the edge target id
                    Packer<size_t, HeapMemory>::pack(mem, sgp.get(pc).send_el.get(e_it), sts);
 
                    ++e_it;
                }
            }
 
            prc.add(i);
            size.add(pmem.size());
            ptr.add(mem.getPointerBase());
        }
 
        // Exchange informations through processors
        vcl.sendrecvMultipleMessagesNBX(prc.size(), (size_t *)size.getPointer(), (size_t *)prc.getPointer(), (void **)ptr.getPointer(), gr_receive, &packs, NONE);
 
        for (int i = 0 ; i < to_release_ext.size(); i++)
        {
            to_release_ext.get(i)->decRef();
            delete to_release_ext.get(i);
 
            delete to_release.get(i);
        }
 
        for (size_t i = 0; i < vcl.getProcessingUnits(); i++)
        {
 
            if (i != vcl.getProcessUnitID() && packs.get(i).size() > 0)
            {
                Unpack_stat ps;
 
                ExtPreAlloc<HeapMemory> mem(packs.get(i).size(), packs.get(i));
 
                // unpack total number of vertex
                size_t r_size;
                Unpacker<size_t, HeapMemory>::unpack(mem, r_size, ps);
 
                // take previous last item
                size_t prev = getNVertex();
 
                for (size_t j = prev; j < prev + r_size; j++)
                {
                    // unpack the vertex
                    V v_n;
                    Unpacker<V, HeapMemory>::unpack(mem, v_n, ps);
 
                    v_info vm;
                    Unpacker<v_info, HeapMemory>::unpack(mem, vm, ps);
 
                    add_vertex(v_n, vm.template get<v_info::id>(), vm.template get<v_info::gid>());
 
                    if (addAsGhosts)
                        ghs_map.insert( { vm.template get<v_info::gid>(), false });
 
                    // unpack size of children
                    size_t s;
                    Unpacker<size_t, HeapMemory>::unpack(mem, s, ps);
 
                    // prepare slots for the children
                    for (size_t k = 0; k < s; k++)
                    {
                        // unpack edge
                        E e_n;
                        Unpacker<E, HeapMemory>::unpack(mem, e_n, ps);
 
                        // unpack edge
                        e_info e_i;
                        Unpacker<e_info, HeapMemory>::unpack(mem, e_i, ps);
 
                        // unpack vertex id of the edge target
                        size_t el_n;
                        Unpacker<size_t, HeapMemory>::unpack(mem, el_n, ps);
 
                        // add the edge //HERE ERROR modify to add globals
                        addEdge(j, el_n, e_n, e_i);
                    }
                }
            }
        }
 
        // After the exchange reset all the structures needed for it
        resetExchange();
    }
 
    void updateVtxdist()
    {
        // Prepare vtxdist
        vtxdist.resize(vcl.getProcessingUnits() + 1);
 
        // Set first element to 0, always the same
        vtxdist.get(0) = 0;
 
        // Prepare array that will contain the sizes of all the graphs
        openfpm::vector<size_t> recv(vcl.getProcessingUnits());
 
        // Set the local size
        size_t tv = getNVertex();
 
        // Sent and receive the size of each subgraph
        vcl.allGather(tv, recv);
        vcl.execute();
 
        // Set the value for this processor
        recv.get(vcl.getProcessUnitID()) = getNVertex();
 
        // Update vtxdist
        for (size_t i = 1; i <= recv.size(); ++i)
        {
            vtxdist.get(i) = recv.get(i - 1) + vtxdist.get(i - 1);
        }
    }
 
    void remap()
    {
        size_t p_id = vcl.getProcessUnitID();
 
        typedef struct
        {
            // new vertex id
            size_t id;
            // processor rank that contain the vertex
            size_t pid;
        } IdnProc;
 
        // Map that will contain the couples to update the global info map in this processor
        // The key is the (old vertex id)
        std::unordered_map<size_t, IdnProc> on_toup;
 
        // For each processor old, new couples
        openfpm::vector<openfpm::vector<size_t>> on_info(vcl.getProcessingUnits());
 
        std::map<size_t, size_t> old_glob2loc(glb2loc.begin(), glb2loc.end());
        size_t j = vtxdist.get(p_id);
        size_t i = 0, k = 0;
 
        // Reset maps of vertices ids
        id2glb.clear();
        glb2id.clear();
        glb2loc.clear();
 
        // Fix sending couples gid, newid and remove on_toup updating glbi_map here and after receive
        for (auto it : old_glob2loc)
        {
            // The couple to the final map, needed to update the vertices in this sub-graph
            IdnProc nidpid = { j, p_id };
            on_toup.insert( { v_m.get(it.second).template get<v_info::id>(), nidpid });
 
            // fill the re-mapping information for each processors that need it
            on_info.get(getInfoProc(it.first)).add(v_m.get(it.second).template get<v_info::id>());
            on_info.get(getInfoProc(it.first)).add(j);
 
            // Re-map the vertex
            map_v(j, v_m.get(it.second).template get<v_info::gid>(), it.second);
 
            j++;
            i++;
            k += 2;
        }
 
        // Prepare structures to send the old-new couples
        openfpm::vector<size_t> prc;
        openfpm::vector<size_t> size;
        openfpm::vector<void *> ptr;
 
        // Array that will contain the couples, divided per processor
        openfpm::vector<openfpm::vector<size_t>> on_vs(vcl.getProcessingUnits());
 
        fillSendRecvStructs<size_t>(on_info, prc, size, ptr);
 
        // Send on_info
        vcl.sendrecvMultipleMessagesNBX(prc.size(), (size_t *)size.getPointer(), (size_t *)prc.getPointer(), (void **)ptr.getPointer(), on_receive, &on_vs, NONE);
 
        // Insert in the on_toup map the received couples
        for (size_t i = 0; i < vcl.getProcessingUnits(); i++)
        {
            if (i != vcl.getProcessUnitID() && on_vs.get(i).size() > 0) // redundant check 2nd arg in if
            {
                for (size_t j = 0; j < on_vs.get(i).size() - 1; j += 2) // -1 is useless
                {
                    IdnProc nidpid = { on_vs.get(i).get(j + 1), i };
                    on_toup.insert( { on_vs.get(i).get(j), nidpid });
                }
            }
        }
 
        // Update the glbi_map with the new ids and the processor info
        for (auto k : glbi_map)
        {
            auto search = on_toup.find(glbi_map.at(k.first).id); // fix with (k.second).id
            if (search != on_toup.end())
            {
                GlobalVInfo t = { (search->second).id, (search->second).pid };
                glbi_map.at(k.first) = t;
            }
        }
 
        // Vector of vertices global id I need info
        openfpm::vector<openfpm::vector<size_t>> vni(vcl.getProcessingUnits());
 
        // Map of re-mapping info
        std::unordered_map<size_t, size_t> rmi_m(vcl.getProcessingUnits()); // TODO elimin
 
        // Check which vertices I need to ask info about
        for (size_t i = 0; i < getNVertex(); ++i)
        {
            for (size_t j = 0; j < getNChilds(i); ++j)
            {
                // Here we get the global vertex id of all the children
                size_t vid = getChildDstGid(i, j);
                // We check which processor has the information about this vertex
                size_t pid = getInfoProc(vid);
 
                // if the vertex info is not in this processor I have to make a request to the right processor
                if (!vertexIsInThisGraph(vid) && pid != vcl.getProcessUnitID())
                {
                    // add to requests
                    vni.get(pid).add(vid);
                }
                else if (pid == vcl.getProcessUnitID())
                {
                    // if info is in this processor add it in the map
                    rmi_m.insert( { vid, glbi_map.at(vid).id });
                }
                else if (vertexIsInThisGraph(vid)) // check if it is needed, probably not because the glbi_map is update
                {
                    // if the vertex is in this graph, add the new id in the map
                    rmi_m.insert( { vid, glb2id.at(vid) });
                }
            }
        }
 
        // Array that will contain the requests from the other processors
        openfpm::vector<openfpm::vector<size_t>> req_rmi(vcl.getProcessingUnits());
 
        // Fill the structures for sendrecvMultipleMessagesNBX function
        fillSendRecvStructs<size_t>(vni, prc, size, ptr);
 
        // Send and receive requests
        vcl.sendrecvMultipleMessagesNBX(prc.size(), (size_t *)size.getPointer(), (size_t *)prc.getPointer(), (void **)ptr.getPointer(), on_receive, &req_rmi, NONE);
 
        // Re-mapping info map
        openfpm::vector<openfpm::vector<size_t>> rmi(vcl.getProcessingUnits());
 
        // Array that will contain the response to previous requests
        openfpm::vector<openfpm::vector<size_t>> resp_rmi(vcl.getProcessingUnits());
 
        // Prepare re-mapping info response
        for (size_t i = 0; i < req_rmi.size(); ++i)
        {
            for (size_t j = 0; j < req_rmi.get(i).size(); ++j)
            {
                resp_rmi.get(i).add(glbi_map.at(req_rmi.get(i).get(j)).id);
            }
        }
 
        // Fill the structures for sendrecvMultipleMessagesNBX function
        fillSendRecvStructs<size_t>(resp_rmi, prc, size, ptr);
 
        // Send responses
        vcl.sendrecvMultipleMessagesNBX(prc.size(), (size_t *)size.getPointer(), (size_t *)prc.getPointer(), (void **)ptr.getPointer(), on_receive, &rmi, NONE);
 
        // Add received info into re-mapping info map
        for (size_t i = 0; i < rmi.size(); ++i)
        {
            for (size_t j = 0; j < rmi.get(i).size(); ++j)
            {
                rmi_m.insert( { vni.get(i).get(j), rmi.get(i).get(j) });
            }
        }
 
        // Finally re-map the edges
        for (size_t i = 0; i < getNVertex(); ++i)
        {
            for (size_t s = 0; s < getNChilds(i); s++)
            {
                e_l.template get<e_map::vid>(i * v_slot + s) = rmi_m.at(getChildDstGid(i, s));
            }
        }
    }
 
    void initGlbimap()
    {
        size_t pid = vcl.getProcessUnitID();
 
        for (size_t i = 0; i < getNVertex(); ++i)
        {
            GlobalVInfo info;
            info.id = v_m.get(i).template get<v_info::id>();
            info.pid = pid;
            glbi_map.insert( { v_m.get(i).template get<v_info::gid>(), info });
        }
    }
 
    size_t getInfoProc(size_t vid)
    {
        for (size_t i = 0; i < fvtxdist.size() - 1; ++i)
        {
            if (vid >= (size_t)fvtxdist.get(i) && vid < (size_t)fvtxdist.get(i + 1))
            {
                return i;
            }
        }
        return vcl.getProcessingUnits() - 1;
    }
 
    template<typename T>
    void fillSendRecvStructs(openfpm::vector<openfpm::vector<T>> &vec, openfpm::vector<size_t> &prc, openfpm::vector<size_t> &size, openfpm::vector<void *> &ptr)
    {
        // Reset sendrecv structures
        prc.clear();
        size.clear();
        ptr.clear();
 
        for (size_t i = 0; i < vec.size(); ++i)
        {
            if (vec.get(i).size() > 0 && i != vcl.getProcessUnitID())
            {
                prc.add(i);
                size.add(vec.get(i).size() * sizeof(T));
                ptr.add(vec.get(i).getPointer());
            }
        }
    }
 
public:
 
    typedef V V_type;
 
    typedef E E_type;
 
    typedef typename openfpm::vector<V, Memory, layout_v_base, grow_p, openfpm::vect_isel<V>::value>::container V_container;
 
    typedef typename openfpm::vector<E, Memory, layout_e_base, grow_p, openfpm::vect_isel<E>::value>::container E_container;
 
    DistGraph_CSR<V, E, Memory, layout_v, layout_e,layout_v_base,layout_e_base, grow_p> duplicate() const
    {
        DistGraph_CSR<V, E, Memory, layout_v, layout_e,layout_v_base,layout_e_base, grow_p> dup;
 
        dup.v_slot = v_slot;
 
        // duplicate all the structures
 
        dup.v.swap(v.duplicate());
        dup.v_m.swap(v_m.duplicate());
        dup.v_l.swap(v_l.duplicate());
        dup.glb2id = glb2id;
        dup.id2glb = id2glb;
        dup.glb2loc = glb2loc;
        dup.e.swap(e.duplicate());
        dup.e_m.swap(e_m.duplicate());
        dup.e_l.swap(e_l.duplicate());
        dup.e_invalid.swap(e_invalid.duplicate());
        dup.vtxdist.swap(vtxdist.duplicate());
        dup.fvtxdist.swap(fvtxdist.duplicate());
        return dup;
    }
 
    DistGraph_CSR(const DistGraph_CSR & dg) :
            vcl(create_vcluster())
    {
        this->operator=(dg);
    }
 
    DistGraph_CSR(DistGraph_CSR && dg)
    :vcl(create_vcluster())
    {
        this->operator=(dg);
    }
 
    DistGraph_CSR() :
            DistGraph_CSR(0, 16)
    {
    }
 
    DistGraph_CSR(size_t n_vertex) :
            DistGraph_CSR(n_vertex, 16)
    {
    }
 
    DistGraph_CSR(size_t n_vertex, size_t n_slot) :
            vcl(create_vcluster()), v_slot(n_slot)
    {
        // Creating n_vertex into the graph
        v.resize(n_vertex);
        // Creating n_vertex info objects into the graph
        v_m.resize(n_vertex);
        // Creating n_vertex adjacency list counters
        v_l.resize(n_vertex);
        // no edge set the counter to zero
        v_l.fill(0);
        // create one invalid edge
        e_invalid.resize(1);
        // init communication structures
        resetExchange();
    }
 
    void getDecompositionVector(openfpm::vector<idx_t> &v)
    {
        v.resize(vtxdist.size());
 
        for (size_t i = 0; i < vtxdist.size(); ++i)
        {
            v.get(i) = vtxdist.get(i);
        }
    }
 
    openfpm::vector<idx_t>* getVtxdist()
    {
        return &vtxdist;
    }
 
    void initDistributionVector(openfpm::vector<idx_t> &v)
    {
        vtxdist.resize(vcl.getProcessingUnits() + 1);
        fvtxdist.resize(vcl.getProcessingUnits() + 1);
 
        for (size_t i = 0; i < vtxdist.size(); ++i)
        {
            vtxdist.get(i) = v.get(i);
            fvtxdist.get(i) = v.get(i);
        }
    }
 
    void initDistributionVector()
    {
        updateVtxdist();
 
        fvtxdist.resize(vcl.getProcessingUnits() + 1);
 
        for (size_t i = 0; i < vtxdist.size(); ++i)
        {
            fvtxdist.get(i) = vtxdist.get(i);
        }
    }
 
    DistGraph_CSR(Vcluster<> & vcl, DistGraph_CSR<V, E, Memory> && g) :
            vcl(vcl)
    {
        swap(g);
    }
 
    DistGraph_CSR<V, E, Memory> & operator=(DistGraph_CSR<V, E, Memory> && g)
    {
        swap(g);
 
        return *this;
    }
 
    DistGraph_CSR<V, E, Memory> & operator=(const DistGraph_CSR<V, E, Memory> & g)
    {
        swap(g.duplicate());
 
        return *this;
    }
 
    template<unsigned int i> auto vertex_p(size_t id) -> decltype( v.template get<i>(id) )
    {
        return v.template get<i>(id);
    }
 
    template<unsigned int i> auto vertex_p(grid_key_dx<1> id) -> decltype( v.template get<i>(id) )
    {
        return v.template get<i>(id);
    }
 
    auto vertex(size_t id) -> decltype( v.get(id) )
    {
        return v.get(id);
    }
 
    auto vertex(grid_key_dx<1> id) -> decltype( v.get(id.get(0)) )
    {
        return v.get(id.get(0));
    }
 
    auto vertex(openfpm::vector_key_iterator id) -> decltype( v.get(0) )
    {
        return v.get(id.get());
    }
 
    auto vertex(size_t id) const -> const decltype( v.get(id) )
    {
        return v.get(id);
    }
 
    auto vertex(grid_key_dx<1> id) const -> const decltype( v.get(id.get(0)) )
    {
        return v.get(id.get(0));
    }
 
    auto vertex(openfpm::vector_key_iterator id) const -> const decltype( v.get(0) )
    {
        return v.get(id.get());
    }
 
    auto vertex_info(openfpm::vector_key_iterator id) const -> const decltype( v_m.get(0) )
    {
        return v_m.get(id.get());
    }
 
    auto getVertex(size_t id) -> decltype( v.get(id) )
    {
 
#ifdef SE_CLASS1
 
        if (glb2loc.find(id) == glb2loc.end())
        {
            std::cerr << __FILE__ << ":" << __LINE__ << " The vertex with global id " << id << " is not in this sub-graph. Try to call reqVertex(" << id << ") and sync() first.\n";
            ACTION_ON_ERROR(DIST_GRAPH_ERROR);
        }
 
#endif
 
        return v.get(glb2loc.find(id)->second);
    }
 
    auto getVertex(size_t id) const -> const decltype( v.get(0) )
    {
        try
        {
            return v.get(glb2loc.at(id));
        } catch (const std::out_of_range& oor)
        {
            std::cerr << "The vertex with global id " << id << " is not in this sub-graph. Try to call reqVertex(" << id << ") and sync() first.\n";
        }
 
        return v.get(0);
    }
 
    size_t nodeById(size_t id) const
    {
        try
        {
            return glb2loc.at(id2glb.at(id));
        } catch (const std::out_of_range& oor)
        {
            std::cout << "Node not found by glb: " << id << std::endl;
        }
 
        return 0;
    }
 
    size_t firstId() const
    {
        return vtxdist.get(vcl.getProcessUnitID());
    }
 
    size_t lastId() const
    {
        return vtxdist.get(vcl.getProcessUnitID() + 1) - 1;
    }
 
    size_t getVertexId(size_t i) const
    {
        return v_m.get(i).template get<v_info::id>();
    }
 
    size_t getVertexGlobalId(size_t i) const
    {
        return v_m.get(i).template get<v_info::gid>();
    }
 
    bool vertexIsInThisGraph(size_t id)
    {
        auto search = glb2id.find(id);
 
        if (search != glb2id.end())
        {
            return true;
        }
        else
        {
            return false;
        }
    }
 
    void map_v(size_t n, size_t g, size_t l)
    {
        id2glb.insert( { n, g });
        glb2id.insert( { g, n });
        glb2loc.insert( { g, l });
        v_m.get(l).template get<v_info::id>() = n;
    }
 
    void clear()
    {
        v.clear();
        v_m.clear();
        e.clear();
        id2glb.clear();
        glb2id.clear();
        glb2loc.clear();
        v_l.clear();
        e_l.clear();
        e_invalid.clear();
    }
 
    template<unsigned int i> auto edge_p(grid_key_dx<1> id) -> decltype ( e.template get<i>(id) )
    {
        return e.template get<i>(id);
    }
 
    template<unsigned int i> auto edge_p(size_t id) -> decltype ( e.template get<i>(id) )
    {
        return e.template get<i>(id);
    }
 
    auto edge(grid_key_dx<1> id) const -> const decltype ( e.get(id.get(0)) )
    {
        return e.get(id.get(0));
    }
 
    auto edge(edge_key ek) const -> const decltype ( e.get(0) )
    {
        return e.get(e_l.template get<e_map::eid>(ek.pos * v_slot + ek.pos_e));
    }
 
    auto getEdge(edge_key ek) const -> const decltype ( e.get(0) )
    {
        size_t v;
 
        try
        {
            v = glb2loc.at(ek.pos);
        } catch (const std::out_of_range& oor)
        {
            std::cout << "The source vertex of this edge is not in this graph.\n";
        }
 
        return e.get(e_l.template get<e_map::eid>(v * v_slot + ek.pos_e));
    }
 
    auto edge(size_t id) const -> const decltype ( e.get(id) )
    {
        return e.get(id);
    }
 
    inline size_t getNChilds(size_t c) const
    {
        return v_l.template get<0>(c);
    }
 
    inline size_t getNChilds(typename openfpm::vector<V, Memory, layout_v_base, grow_p, openfpm::vect_isel<V>::value>::iterator_key & c)
    {
        return v_l.template get<0>(c.get());
    }
 
    inline size_t getNEdge(size_t v) const
    {
        try
        {
            v = glb2loc.at(v);
        } catch (const std::out_of_range& oor)
        {
            std::cout << "The source vertex of this edge is not in this graph.\n";
        }
 
        return v_l.template get<0>(v);
    }
 
    inline auto getChildEdge(size_t v, size_t v_e) -> decltype(e.get(0))
    {
        return e.get(e_l.template get<e_map::eid>(v * v_slot + v_e));
    }
 
    inline auto getChildInfo(size_t v, size_t v_e) -> decltype(e_m.get(0))
    {
        return e_m.get(e_l.template get<e_map::eid>(v * v_slot + v_e));
    }
 
    inline auto getEdge(size_t v, size_t v_e) -> decltype(e.get(0))
    {
        try
        {
            v = glb2loc.at(v);
        } catch (const std::out_of_range& oor)
        {
            std::cout << "The source vertex of this edge is not in this graph.\n";
        }
 
        return e.get(e_l.template get<e_map::eid>(v * v_slot + v_e));
    }
 
    inline size_t getChild(size_t v, size_t i) const
    {
#ifdef SE_CLASS1
        if (i >= v_l.template get<0>(v))
        {
            std::cerr << "Error " << __FILE__ << " line: " << __LINE__ << "    vertex " << v << " does not have edge " << i << " on processor " << vcl.getProcessUnitID() << "\n";
        }
 
        if (e.size() <= e_l.template get<e_map::eid>(v * v_slot + i))
        {
            std::cerr << "Error " << __FILE__ << " " << __LINE__ << " vertex " << v << " does not have edge " << i << " on processor " << vcl.getProcessUnitID() << " (" << e.size() << "<=" << e_l.template get<e_map::eid>(v * v_slot + i) << ")\n";
        }
#endif
        // Get the edge id
        return e_l.template get<e_map::vid>(v * v_slot + i);
    }
 
    inline size_t getChild(size_t i) const
    {
        // Get the edge id
        return e_l.template get<e_map::vid>(i);
    }
 
    inline size_t getChild(typename openfpm::vector<V, Memory, layout_v_base, grow_p, openfpm::vect_isel<V>::value>::iterator_key & v, size_t i)
    {
#ifdef DEBUG
        if (i >= v_l.template get<0>(v.get()))
        {
            std::cerr << "Error " << __FILE__ << " line: " << __LINE__ << "    vertex " << v.get() << " does not have edge " << i << "\n";
        }
 
        if (e.size() <= e_l.template get<e_map::eid>(v.get() * v_slot + i))
        {
            std::cerr << "Error " << __FILE__ << " " << __LINE__ << " vertex " << v.get() << " does not have edge " << i << "\n";
        }
#endif
 
        // Get the edge id
        return e_l.template get<e_map::vid>(v.get() * v_slot + i);
    }
 
    inline void add_vertex(const V & vrt, size_t id, size_t gid)
    {
        // Create vertex info object
        v_info vm;
        vm.template get<v_info::id>() = id;
        vm.template get<v_info::gid>() = gid;
 
        // Add the vertex info
        v_m.add(vm);
 
        // Add the vertex
        v.add(vrt);
 
        // Update id to global map
        id2glb.insert( { id, gid });
 
        // Update global id to local index
        glb2loc.insert( { gid, v.size() - 1 });
 
        // Update global id to id
        glb2id.insert( { gid, id });
 
        // Set the number of adjacent vertex for this vertex to 0
        v_l.add(0ul);
 
        // Add a slot for the vertex adjacency list
        e_l.resize(e_l.size() + v_slot);
    }
 
    template<unsigned int dim, typename Mem> inline void add_vertex(const encapc<dim, V, Mem> & vrt, size_t id, size_t gid)
    {
        // Create vertex info object
        v_info vm;
        vm.template get<v_info::id>() = id;
        vm.template get<v_info::gid>() = gid;
 
        // Add the vertex info
        v_m.add(vm);
 
        // Add the vertex
        v.add(vrt);
 
        // Update id to global map
        id2glb.insert( { id, gid });
 
        // Update global id to local index
        glb2loc.insert( { gid, v.size() - 1 });
 
        // Update global id to id
        glb2id.insert( { gid, id });
 
        // Set the number of adjacent vertex for this vertex to 0
        v_l.add(0ul);
 
        // Add a slot for the vertex adjacency list
        e_l.resize(e_l.size() + v_slot);
    }
 
    inline void add_vertex(const V & vrt, size_t gid)
    {
        add_vertex(vrt, gid, gid);
    }
 
    void setGlobalMap(size_t g, size_t l, size_t i)
    {
        v_m.template get<v_info::id>(l) = i;
 
        v_m.template get<v_info::gid>(l) = g;
 
        // Set global id to local index
        glb2loc.insert( { g, l });
 
        // Set global id to id
        glb2id.insert( { g, i });
 
        // Set id to global map
        id2glb.insert( { i, g });
    }
 
    inline auto addEdge(size_t v1, size_t v2, size_t srcgid, size_t dstgid) -> decltype(e.get(0))
    {
        // add an edge
        long int id_x_end = addEdge_<NoCheck>(v1, v2);
        // If there is not edge return an invalid edge, is a kind of stub object
        if (id_x_end == NO_EDGE)
            return e_invalid.get(0);
 
        // set source and destination global ids of the edge
        e_m.template get<e_info::sgid>(id_x_end) = srcgid;
        e_m.template get<e_info::dgid>(id_x_end) = dstgid;
 
        // return the edge to change the properties
        return e.get(id_x_end);
    }
 
    inline auto addEdge(size_t v1, size_t v2, size_t srcgid, size_t dstgid, const E & ed) -> decltype(e.get(0))
    {
        // add an edge
        long int id_x_end = addEdge_<NoCheck>(v1, v2);
        // If there is not edge return an invalid edge, is a kind of stub object
        if (id_x_end == NO_EDGE)
            return e_invalid.get(0);
 
        // add in e_l the edge properties
        e.set(id_x_end, ed);
 
        // set source and destination global ids of the edge
        e_m.template get<e_info::sgid>(id_x_end) = srcgid;
        e_m.template get<e_info::dgid>(id_x_end) = dstgid;
 
        // return the edge to change the properties
        return e.get(id_x_end);
    }
 
    template<unsigned int dim, typename Mem, typename Mem1> inline auto addEdge(size_t v1, size_t v2, const encapc<dim, E, Mem> & ed, const encapc<dim, e_info, Mem1> & ei) -> decltype(e.get(0))
    {
        // add an edge
        long int id_x_end = addEdge_<NoCheck>(v1, v2);
        // If there is not edge return an invalid edge, is a kind of stub object
        if (id_x_end == NO_EDGE)
            return e_invalid.get(0);
 
        // set the edge object and the edge info object
        e.set(id_x_end, ed);
        e_m.set(id_x_end, ei);
 
        // return the edge to change the properties
        return e.get(id_x_end);
    }
 
    inline auto addEdge(size_t v1, size_t v2, const E & ed, const e_info & ei) -> decltype(e.get(0))
    {
        // add an edge
        long int id_x_end = addEdge_<NoCheck>(v1, v2);
        // If there is not edge return an invalid edge, is a kind of stub object
        if (id_x_end == NO_EDGE)
            return e_invalid.get(0);
 
        // set the edge object and the edge info object
        e.set(id_x_end, ed);
        e_m.set(id_x_end, ei);
 
        // return the edge to change the properties
        return e.get(id_x_end);
    }
 
    size_t getChildSrcGid(size_t v1, size_t s)
    {
        size_t eid = e_l.template get<e_map::eid>(v1 * v_slot + s);
        return e_m.template get<e_info::sgid>(eid);
    }
 
    size_t getChildDstGid(size_t v1, size_t s)
    {
        size_t eid = e_l.template get<e_map::eid>(v1 * v_slot + s);
        return e_m.template get<e_info::dgid>(eid);
    }
 
    template<typename CheckPolicy = NoCheck> inline void add_edge(size_t v1, size_t v2)
    {
        //if the source vertex is not in this graph, this processor doesn't need to do anything
        if (!vertexIsInThisGraph(v1))
        {
            return;
        }
 
        //if the destination vertex is not in this graph, this processor has to request it and add the edge to a queue
        if (!vertexIsInThisGraph(v2))
        {
            reqVertex(v2);
            EdgeReq er = { v1, v2, 0, 0 };
            e_queue.add(er);
 
            return;
        }
 
        // add an edge
        long int id_x_end = addEdge_<CheckPolicy>(glb2loc.at(v1), glb2id.at(v2));
 
        // If there is not edge return an invalid edge, is a kind of stub object
        if (id_x_end == NO_EDGE)
            return;
 
        // set source and destination ids of the edge
        e_m.get(id_x_end).template get<e_info::sgid>() = v1;
        e_m.get(id_x_end).template get<e_info::dgid>() = v2;
    }
 
    void syncEdge()
    {
        // retrieve ghosts necessary for edge adding
        sync();
 
        // update with real values of edges, we don't add the edge here because it will modify edge's array length
        for (size_t i = 0; i < e_queue.size(); ++i)
        {
            e_queue.get(i).v1n = glb2loc.at(e_queue.get(i).v1);
            e_queue.get(i).v2n = glb2id.at(e_queue.get(i).v2);
        }
 
        deleteGhosts();
 
        for (auto req : e_queue)
        {
            // add an edge
            long int id_x_end = addEdge_<>(req.v1n, req.v2n);
 
            // If there is not edge return an invalid edge, is a kind of stub object
            if (id_x_end == NO_EDGE)
                return;
 
            // set source and destination ids of the edge
            e_m.get(id_x_end).template get<e_info::sgid>() = req.v1;
            e_m.get(id_x_end).template get<e_info::dgid>() = req.v2;
        }
 
        e_queue.clear();
 
    }
 
    inline void swap(DistGraph_CSR<V, E> & g)
    {
        // switch the memory
        v.swap(g.v);
        v_m.swap(g.v_m);
        e.swap(g.e);
        e_m.swap(g.e_m);
        v_l.swap(g.v_l);
        glb2id.swap(g.glb2id);
        id2glb.swap(g.id2glb);
        glb2loc.swap(g.glb2loc);
        e_l.swap(g.e_l);
        e_invalid.swap(g.e_invalid);
        vtxdist.swap(g.vtxdist);
        fvtxdist.swap(g.fvtxdist);
 
        size_t v_slot_tmp = v_slot;
        v_slot = g.v_slot;
        g.v_slot = v_slot_tmp;
    }
 
    inline void swap(DistGraph_CSR<V, E> && g)
    {
        // switch the memory
        v.swap(g.v);
        v_m.swap(g.v_m);
        e.swap(g.e);
        e_m.swap(g.e_m);
        v_l.swap(g.v_l);
        glb2id.swap(g.glb2id);
        id2glb.swap(g.id2glb);
        glb2loc.swap(g.glb2loc);
        e_l.swap(g.e_l);
        e_invalid.swap(g.e_invalid);
        vtxdist.swap(g.vtxdist);
        fvtxdist.swap(g.fvtxdist);
 
        size_t v_slot_tmp = v_slot;
        v_slot = g.v_slot;
        g.v_slot = v_slot_tmp;
    }
 
    inline auto getVertexIterator() const -> decltype(v.getIterator())
    {
        return v.getIterator();
    }
 
    inline edge_iterator<DistGraph_CSR<V, E, Memory>> getEdgeIterator() const
    {
        return edge_iterator<DistGraph_CSR<V, E, Memory>>(*this);
    }
 
    inline size_t getNVertex() const
    {
        return v.size();
    }
 
    inline size_t getTotNVertex() const
    {
        return vtxdist.get(vcl.getProcessingUnits());
    }
 
    inline size_t getNEdge() const
    {
        return e.size();
    }
 
    void init()
    {
        initDistributionVector();
 
        initGlbimap();
 
        remap();
    }
 
    bool isGhost(size_t id)
    {
        auto search = ghs_map.find(id);
 
        if (search != ghs_map.end())
        {
            return true;
        }
        else
        {
            return false;
        }
    }
 
    void deleteGhosts()
    {
        if (ghs_map.size() == 0)
            return;
 
        // Prepare new graph
        DistGraph_CSR recv_g;
 
        size_t fpos = getNVertex() - ghs_map.size();
        size_t epos = 0;
 
        // Reset global to local map
        for (auto gh : ghs_map)
        {
            epos += getNChilds(glb2loc.at(gh.first));
            id2glb.erase(glb2id.at(gh.first));
            glb2loc.erase(gh.first);
            glb2id.erase(gh.first);
 
        }
 
        //resize all structures to delete the ghosts
        v.resize(fpos);
        v_m.resize(fpos);
        v_l.resize(fpos);
        e.resize(e.size() - epos);
        e_m.resize(e_m.size() - epos);
        e_l.resize(fpos * v_slot);
 
        // Clear ghosts map
        ghs_map.clear();
    }
 
    template<bool toRemove = true> //TODO make it private and create wrapper in public
    void q_move(size_t i, size_t t)
    {
        // Check if a 'useless' move has been requested
        if (t == vcl.getProcessUnitID())
        {
            std::cerr << "Warning: " << __FILE__ << ":" << __LINE__ << " target processor is equal the local processor\n";
            return;
        }
 
        // If the array for t processor is empty, set it to not empty
        if (sgp.get(t).isEmpty)
            sgp.get(t).isEmpty = false;
 
        // Add the vertex to the vertex send buffer
        sgp.get(t).send_v.add(vertex(i));
 
        // Add the vertex to the vertex send buffer
        sgp.get(t).send_v_m.add(v_m.get(i));
 
        // Add the info about the number of children
        sgp.get(t).send_es.add(getNChilds(i));
 
        for (size_t s = 0; s < getNChilds(i); s++)
        {
            // Add edge value and object to the respective arrays
            sgp.get(t).send_e.add(getChildEdge(i, s));
            sgp.get(t).send_e_m.add(getChildInfo(i, s));
            sgp.get(t).send_el.add(getChild(i, s));
        }
 
        // If the vertex has to be removed after the send add its index id to the v_td array
        if (toRemove)
            v_td.add(i);
    }
 
    bool moveQueueIsEmpty()
    {
        size_t exs = 0;
        for (size_t i = 0; i < vcl.getProcessingUnits(); ++i)
            if (sgp.get(i).isEmpty)
                exs++;
 
        if (exs == 0)
            return true;
 
        return false;
    }
 
    void redistribute()
    {
        if (glbi_map.size() == 0)
            initGlbimap();
 
        deleteGhosts();
 
        exchangeVertices<false>();
 
        updateVtxdist();
 
        remap();
    }
 
    void reqVertex(size_t gid)
    {
        // if not initialized, prepare vr_queue
        if (vr_queue.size() == 0)
            vr_queue.resize(vcl.getProcessingUnits());
 
        if (!vertexIsInThisGraph(gid))
        {
            vr_queue.get(getInfoProc(gid)).add(gid);
        }
 
    }
 
    void sync()
    {
        // If not initialized, prepare global informations map
        if (glbi_map.size() == 0)
            initGlbimap();
 
        // if not initialized, prepare vr_queue
        if (vr_queue.size() == 0) // TODO remove check on size
            vr_queue.resize(vcl.getProcessingUnits());
 
        // Arrays that will contain temporary requests and responses during communications TODO move to global private and reset method
        openfpm::vector<openfpm::vector<size_t>> resp(vcl.getProcessingUnits());
        openfpm::vector<openfpm::vector<size_t>> reqs(vcl.getProcessingUnits());
 
        // Prepare structures for communication
        openfpm::vector<size_t> prc;
        openfpm::vector<size_t> size;
        openfpm::vector<void *> ptr;
 
        // Fill the structures for sendrecvMultipleMessagesNBX function
        fillSendRecvStructs<size_t>(vr_queue, prc, size, ptr);
 
        // Send/receive requests for info about needed vertices
        vcl.sendrecvMultipleMessagesNBX(prc.size(), (size_t *)size.getPointer(), (size_t *)prc.getPointer(), (void **)ptr.getPointer(), on_receive, &resp, NONE);
 
        // Prepare responses with the containing processors of requested vertices
        for (size_t i = 0; i < resp.size(); ++i)
        {
            reqs.get(i).clear();
 
            for (size_t j = 0; j < resp.get(i).size(); ++j)
            {
                try
                {
                    reqs.get(i).add(glbi_map.at(resp.get(i).get(j)).pid);
                } catch (const std::out_of_range& oor)
                {
                    std::cout << resp.get(i).get(j) << " not found in global info map (proc: " << vcl.getProcessUnitID() << ")\n";
                }
            }
        }
 
        // Fill the structures for sendrecvMultipleMessagesNBX function
        fillSendRecvStructs<size_t>(reqs, prc, size, ptr);
 
        // Reset response array TODO clear and resize not needed
        resp.clear();
        resp.resize(vcl.getProcessingUnits());
 
        // Send/receive responses with the containing processors of requested vertices
        vcl.sendrecvMultipleMessagesNBX(prc.size(), (size_t *)size.getPointer(), (size_t *)prc.getPointer(), (void **)ptr.getPointer(), on_receive, &resp, NONE);
 
        // Clear requests array
        reqs.clear();
        reqs.resize(vcl.getProcessingUnits());
 
        // Prepare vertices requests
        for (size_t i = 0; i < vcl.getProcessingUnits(); ++i)
        {
            // If the info has been retrieved from other processors take it from resp
            if (i != vcl.getProcessUnitID())
            {
                for (size_t j = 0; j < vr_queue.get(i).size(); ++j)
                {
                    reqs.get(resp.get(i).get(j)).add(vr_queue.get(i).get(j));
                }
            }
            // Otherwise take it from the global info map of this processor
            else
            {
                for (size_t j = 0; j < vr_queue.get(i).size(); ++j)
                {
                    reqs.get(glbi_map.at(vr_queue.get(i).get(j)).pid).add(vr_queue.get(i).get(j));
                }
            }
        }
 
        // Fill the structures for sendrecvMultipleMessagesNBX function
        fillSendRecvStructs<size_t>(reqs, prc, size, ptr);
 
        // Reset response array
        resp.clear();
        resp.resize(vcl.getProcessingUnits());
 
        // Send/receive vertices requests
        vcl.sendrecvMultipleMessagesNBX(prc.size(), (size_t *)size.getPointer(), (size_t *)prc.getPointer(), (void **)ptr.getPointer(), on_receive, &resp, NONE);
 
        for (size_t i = 0; i < resp.size(); ++i)
        {
            for (size_t j = 0; j < resp.get(i).size(); ++j)
            {
                try
                {
                    q_move<false>(glb2loc.at(resp.get(i).get(j)), i);
                } catch (const std::out_of_range& oor)
                {
                    std::cout << resp.get(i).get(j) << " not found in global to local map (proc: " << vcl.getProcessUnitID() << ")\n";
                }
            }
        }
 
        // Send and receive vertices, the received ones will be added to the graph as ghosts
        exchangeVertices<true>();
 
        // Empty the requests list
        vr_queue.clear();
    }
};
 
template<typename V, typename E>
class DistGraph_CSR_s: public DistGraph_CSR<V, E>
{
 
};
 
#endif /* DIST_MAP_GRAPH_HPP_ */