hard_sync_linker.h

Go to the documentation of this file.

#ifndef FPMAS_HARD_SYNC_LINKER_H
#define FPMAS_HARD_SYNC_LINKER_H
 
#include <set>
 
#include "fpmas/utils/macros.h"
 
#include "fpmas/graph/distributed_graph.h"
#include "fpmas/synchro/hard/api/hard_sync_mode.h"
#include "fpmas/synchro/ghost/ghost_mode.h"
#include "server_pack.h"
#include "termination.h"
 
namespace fpmas { namespace synchro { namespace hard {
    using api::Tag;
    using api::Epoch;
    using fpmas::api::graph::LocationState;
 
    namespace hard_link {
        template<typename T>
            class LinkServer : public api::LinkServer {
                public:
                    typedef fpmas::api::graph::DistributedEdge<T> EdgeApi;
                    typedef fpmas::api::communication::TypedMpi<graph::EdgePtrWrapper<T>> EdgeMpi;
                    typedef fpmas::api::communication::TypedMpi<DistributedId> IdMpi;
                private:
                    Epoch epoch = Epoch::EVEN;
 
                    fpmas::api::communication::MpiCommunicator& comm;
                    fpmas::api::graph::DistributedGraph<T>& graph;
                    IdMpi& id_mpi;
                    EdgeMpi& edge_mpi;
                    std::set<DistributedId> locked_unlink_edges;
                    std::set<DistributedId> locked_remove_nodes;
 
                public:
                    LinkServer(
                            fpmas::api::communication::MpiCommunicator& comm,
                            fpmas::api::graph::DistributedGraph<T>& graph,
                            IdMpi& id_mpi, EdgeMpi& edge_mpi) :
                        comm(comm), graph(graph),
                        id_mpi(id_mpi), edge_mpi(edge_mpi) {}
 
                    Epoch getEpoch() const override {return epoch;}
                    void setEpoch(api::Epoch epoch) override {this->epoch = epoch;}
 
                    void handleIncomingRequests() override;
 
 
                    void lockUnlink(DistributedId edge_id) override {
                        locked_unlink_edges.insert(edge_id);
                    }
 
                    bool isLockedUnlink(DistributedId edge_id) override {
                        return locked_unlink_edges.count(edge_id) > 0;
                    }
 
                    void unlockUnlink(DistributedId edge_id) override {
                        locked_unlink_edges.erase(edge_id);
                    }
 
                    void lockRemoveNode(DistributedId node_id) override {
                        locked_remove_nodes.insert(node_id);
                    }
 
                    bool isLockedRemoveNode(DistributedId node_id) override {
                        return locked_remove_nodes.count(node_id);
                    }
 
                    void unlockRemoveNode(DistributedId node_id) override {
                        locked_remove_nodes.erase(node_id);
                    }
            };
 
        template<typename T>
            void LinkServer<T>::handleIncomingRequests() {
                fpmas::api::communication::Status status;
                // Check read request
                if(edge_mpi.Iprobe(MPI_ANY_SOURCE, epoch | Tag::LINK, status)) {
                    EdgeApi* edge = edge_mpi.recv(status.source, status.tag);
                    FPMAS_LOGD(this->comm.getRank(), "LINK_SERVER",
                            "receive link request from %i", status.source);
                    graph.importEdge(edge);
                }
                if(id_mpi.Iprobe(MPI_ANY_SOURCE, epoch | Tag::UNLINK, status)) {
                    DistributedId unlink_id = id_mpi.recv(status.source, status.tag);
                    FPMAS_LOGD(this->comm.getRank(), "LINK_SERVER",
                            "receive unlink request %s from %i",
                            FPMAS_C_STR(unlink_id), status.source);
                    if(!isLockedUnlink(unlink_id)) {
                        // The edge is not being unlinked by the local process
                        const auto& edges = graph.getEdges();
                        auto it = edges.find(unlink_id);
                        if(it != edges.end()) {
                            // The edge has not been unlinked by an other UNLINK
                            // operation
                            auto* edge = it->second;
                            // Source or target node is not being removed by the local
                            // process. In this case, the local process is responsible
                            // for all the required unlink operations, so the incoming
                            // request is ignored.
                            if(!(isLockedRemoveNode(edge->getSourceNode()->getId())
                                        || isLockedRemoveNode(edge->getTargetNode()->getId()))) {
                                graph.erase(edge);
                            }
                        }
                    }
                }
                if(id_mpi.Iprobe(MPI_ANY_SOURCE, epoch | Tag::REMOVE_NODE, status)) {
                    DistributedId node_id = id_mpi.recv(status.source, status.tag);
                    FPMAS_LOGD(this->comm.getRank(), "LINK_SERVER",
                            "receive remove node request %s from %i",
                            FPMAS_C_STR(node_id), status.source);
 
                    // Initiates a removeNode operation from the local process,
                    // that will trigger all required UNLINK operations
                    graph.removeNode(graph.getNode(node_id));
                }
            }
 
        template<typename T>
            class LinkClient : public api::LinkClient<T> {
                public:
                    typedef fpmas::api::graph::DistributedEdge<T> EdgeApi;
                    typedef fpmas::api::graph::DistributedNode<T> NodeApi;
                    typedef fpmas::api::communication::TypedMpi<graph::EdgePtrWrapper<T>> EdgeMpi;
                    typedef fpmas::api::communication::TypedMpi<DistributedId> IdMpi;
 
                private:
                    fpmas::api::communication::MpiCommunicator& comm;
                    IdMpi& id_mpi;
                    EdgeMpi& edge_mpi;
                    ServerPackBase& server_pack;
 
                public:
                    LinkClient(
                            fpmas::api::communication::MpiCommunicator& comm,
                            IdMpi& id_mpi, EdgeMpi& edge_mpi,
                            ServerPackBase& server_pack) :
                        comm(comm), id_mpi(id_mpi), edge_mpi(edge_mpi),
                        server_pack(server_pack) {}
 
                    void link(const EdgeApi*) override;
                    void unlink(const EdgeApi*) override;
                    void removeNode(const NodeApi*) override;
            };
 
        template<typename T>
            void LinkClient<T>::link(const EdgeApi* edge) {
                bool distant_src =
                    edge->getSourceNode()->state() == LocationState::DISTANT;
                bool distant_tgt =
                    edge->getTargetNode()->state() == LocationState::DISTANT;
 
                if(edge->getSourceNode()->location() != edge->getTargetNode()->location()) {
                    // The edge is DISTANT, so at least of the two nodes is
                    // DISTANT. In this case, if the two nodes are
                    // DISTANT, they don't have the same location, so two
                    // requests must be performed.
                    fpmas::api::communication::Request req_src;
                    fpmas::api::communication::Request req_tgt;
                    // Simultaneously initiate the two requests, that are potentially
                    // made to different procs
                    if(distant_src) {
                        edge_mpi.Issend(
                                const_cast<EdgeApi*>(edge), edge->getSourceNode()->location(),
                                server_pack.getEpoch() | Tag::LINK, req_src
                                ); 
                    }
                    if(distant_tgt) {
                        edge_mpi.Issend(
                                const_cast<EdgeApi*>(edge), edge->getTargetNode()->location(),
                                server_pack.getEpoch() | Tag::LINK, req_tgt
                                ); 
                    }
                    // Sequentially waits for each request : if req_tgt completes
                    // before req_src, the second waitSendRequest will immediatly return
                    // so there is no performance issue.
                    if(distant_src) {
                        server_pack.waitSendRequest(req_src);
                    }
                    if(distant_tgt) {
                        server_pack.waitSendRequest(req_tgt);
                    }
                } else {
                    // The edge is DISTANT, and its source and target nodes
                    // locations are the same, so the two nodes are necessarily
                    // located on the same DISTANT proc : only need to perform
                    // one request
                    fpmas::api::communication::Request req;
                    edge_mpi.Issend(
                            const_cast<EdgeApi*>(edge), edge->getSourceNode()->location(),
                            server_pack.getEpoch() | Tag::LINK, req
                            ); 
                    server_pack.waitSendRequest(req);
                }
            }
 
        template<typename T>
            void LinkClient<T>::unlink(const EdgeApi* edge) {
 
                bool distant_src
                    = edge->getSourceNode()->state() == LocationState::DISTANT;
                bool distant_tgt
                    = edge->getTargetNode()->state() == LocationState::DISTANT;
 
                fpmas::api::communication::Request req_src;
                fpmas::api::communication::Request req_tgt;
                // Simultaneously initiate the two requests, that are potentially
                // made to different procs
                if(distant_src) {
                    this->id_mpi.Issend(
                            edge->getId(), edge->getSourceNode()->location(),
                            server_pack.getEpoch() | Tag::UNLINK, req_src
                            ); 
                }
                if(distant_tgt) {
                    this->id_mpi.Issend(
                            edge->getId(), edge->getTargetNode()->location(),
                            server_pack.getEpoch() | Tag::UNLINK, req_tgt
                            ); 
                }
                // Sequentially waits for each request : if req_tgt completes
                // before req_src, the second waitSendRequest will immediatly return
                // so there is no performance issue.
                if(distant_src) {
                    server_pack.waitSendRequest(req_src);
                }
                if(distant_tgt) {
                    server_pack.waitSendRequest(req_tgt);
                }
            }
 
        template<typename T>
            void LinkClient<T>::removeNode(const NodeApi* node) {
                fpmas::api::communication::Request req;
                this->id_mpi.Issend(
                        node->getId(), node->location(),
                        server_pack.getEpoch() | Tag::REMOVE_NODE, req
                        );
 
                server_pack.waitSendRequest(req);
            }
 
 
        template<typename T>
            class HardSyncLinker : public api::HardSyncLinker<T> {
                public:
                    typedef fpmas::api::graph::DistributedEdge<T> EdgeApi;
                    typedef fpmas::api::graph::DistributedNode<T> NodeApi;
 
                private:
                    std::vector<EdgeApi*> ghost_edges;
                    fpmas::api::graph::DistributedGraph<T>& graph;
                    api::LinkClient<T>& link_client;
                    ServerPack<api::MutexServer<T>, api::LinkServer>& server_pack;
                    std::vector<fpmas::api::graph::DistributedNode<T>*> nodes_to_remove;
 
                public:
                    void registerNodeToRemove(
                            fpmas::api::graph::DistributedNode<T>* node) override {
                        nodes_to_remove.push_back(node);
                    }
 
                    HardSyncLinker(
                            fpmas::api::graph::DistributedGraph<T>& graph,
                            api::LinkClient<T>& link_client,
                            ServerPack<api::MutexServer<T>, api::LinkServer>& server_pack) :
                        graph(graph), link_client(link_client), server_pack(server_pack) {
                        }
 
                    void link(EdgeApi* edge) override {
                        if(edge->state() == LocationState::DISTANT) {
                            link_client.link(edge);
 
                            if(edge->getSourceNode()->state() == LocationState::DISTANT
                                    && edge->getTargetNode()->state() == LocationState::DISTANT) {
                                ghost_edges.push_back(const_cast<EdgeApi*>(edge));
                            }
                        }
                    };
 
                    void unlink(EdgeApi* edge) override {
                        // Prevents other processes to unlink the edge while the local
                        // process is unlinking it. In this case, incoming unlink requests
                        // have no effect. (see LinkServer::handleIncomingRequests())
                        server_pack.linkServer().lockUnlink(edge->getId());
 
                        if(edge->state() == LocationState::DISTANT) {
                            link_client.unlink(edge);
                        }
 
                        // Unlocks the unlink operation. The edge will actually be erased
                        // from the graph upon return, in the DistributedGraph::unlink()
                        // method.
                        server_pack.linkServer().unlockUnlink(edge->getId());
                    };
 
                    void removeNode(NodeApi* node) override {
                        if(node->state() == LocationState::DISTANT) {
                            link_client.removeNode(node);
                        } else {
                            server_pack.linkServer().lockRemoveNode(node->getId());
                            for(auto edge : node->getOutgoingEdges())
                                graph.unlink(edge);
                            for(auto edge : node->getIncomingEdges())
                                graph.unlink(edge);
                            server_pack.linkServer().unlockRemoveNode(node->getId());
                        }
                        registerNodeToRemove(node);
                    }
 
                    void synchronize() override {
                        FPMAS_LOGI(graph.getMpiCommunicator().getRank(),
                                "HARD_SYNC_LINKER", "Synchronizing sync linker...", "");
                        for(auto edge : ghost_edges)
                            graph.erase(edge);
                        ghost_edges.clear();
 
                        server_pack.terminate();
 
                        for(auto node : nodes_to_remove) {
                            graph.erase(node);
                        }
                        nodes_to_remove.clear();
 
                        FPMAS_LOGI(graph.getMpiCommunicator().getRank(),
                                "HARD_SYNC_LINKER", "Synchronized.", "");
                    };
            };
    }
 
    namespace ghost_link {
        template<typename T>
            class HardSyncLinker : public ghost::GhostSyncLinkerBase<T> {
                private:
                    communication::TypedMpi<Color> color_mpi;
                    TerminationAlgorithm termination;
                    ServerPackBase& server_pack;
 
                public:
                    HardSyncLinker(
                            communication::TypedMpi<graph::EdgePtrWrapper<T>>& edge_mpi,
                            communication::TypedMpi<DistributedId>& id_mpi,
                            fpmas::api::graph::DistributedGraph<T>& graph,
                            ServerPackBase& server_pack) :
                        ghost::GhostSyncLinkerBase<T>(edge_mpi, id_mpi, graph),
                        color_mpi(graph.getMpiCommunicator()),
                        termination(graph.getMpiCommunicator(), color_mpi),
                        server_pack(server_pack) {
                        }
 
                    
                    void synchronize() override {
                        /*
                         * As specified by the
                         * fpmas::api::graph::DistributedGraph::synchronize()
                         * method, when a graph synchronization is performed,
                         * SyncLinker::synchronize() is called before
                         * DataSync::synchronize().
                         *
                         * But, in the context of this component, DataSync
                         * operations are performed on the fly using point-to-point
                         * communications, so it is necessary to finish those
                         * operations before initializing the collective
                         * synchronize_links() communications.
                         */
                        termination.terminate(server_pack);
                        this->synchronize_links();
                    }
            };
 
    }
}}}
#endif