DistributedGraph.h

Go to the documentation of this file.

/*
 * This file belongs to the Galois project, a C++ library for exploiting
 * parallelism. The code is being released under the terms of the 3-Clause BSD
 * License (a copy is located in LICENSE.txt at the top-level directory).
 *
 * Copyright (C) 2018, The University of Texas at Austin. All rights reserved.
 * UNIVERSITY EXPRESSLY DISCLAIMS ANY AND ALL WARRANTIES CONCERNING THIS
 * SOFTWARE AND DOCUMENTATION, INCLUDING ANY WARRANTIES OF MERCHANTABILITY,
 * FITNESS FOR ANY PARTICULAR PURPOSE, NON-INFRINGEMENT AND WARRANTIES OF
 * PERFORMANCE, AND ANY WARRANTY THAT MIGHT OTHERWISE ARISE FROM COURSE OF
 * DEALING OR USAGE OF TRADE.  NO WARRANTY IS EITHER EXPRESS OR IMPLIED WITH
 * RESPECT TO THE USE OF THE SOFTWARE OR DOCUMENTATION. Under no circumstances
 * shall University be liable for incidental, special, indirect, direct or
 * consequential damages or loss of profits, interruption of business, or
 * related expenses which may arise from use of Software or Documentation,
 * including but not limited to those resulting from defects in Software and/or
 * Documentation, or loss or inaccuracy of data of any kind.
 */

#ifndef _GALOIS_DIST_HGRAPH_H_
#define _GALOIS_DIST_HGRAPH_H_

#include <unordered_map>
#include <fstream>

#include "galois/graphs/LC_CSR_Graph.h"
#include "galois/graphs/BufferedGraph.h"
#include "galois/runtime/DistStats.h"
#include "galois/graphs/OfflineGraph.h"
#include "galois/DynamicBitset.h"

/*
 * Headers for boost serialization
 */

namespace galois {
namespace graphs {
enum MASTERS_DISTRIBUTION {
  BALANCED_MASTERS,
  BALANCED_EDGES_OF_MASTERS,
  BALANCED_MASTERS_AND_EDGES
};

template <typename NodeTy, typename EdgeTy>
class DistGraph {
private:
  constexpr static const char* const GRNAME = "dGraph";

  using GraphTy = galois::graphs::LC_CSR_Graph<NodeTy, EdgeTy, true>;

protected:
  GraphTy graph;

  bool transposed;

  // global graph variables
  uint64_t numGlobalNodes; 
  uint64_t numGlobalEdges; 
  uint32_t numNodes;       
  uint64_t numEdges;       

  const unsigned id;       
  const uint32_t numHosts; 

  // local graph
  // size() = Number of nodes created on this host (masters + mirrors)
  uint32_t numOwned;    
  uint32_t beginMaster; 
  uint32_t numNodesWithEdges; 

  std::vector<std::pair<uint64_t, uint64_t>> gid2host;
  std::vector<std::vector<size_t>> mirrorNodes;

  std::vector<uint64_t> localToGlobalVector;
  std::unordered_map<uint64_t, uint32_t> globalToLocalMap;

private:
  // vector for determining range objects for master nodes + nodes
  // with edges (which includes masters)
  std::vector<uint32_t> allNodesRanges;
  std::vector<uint32_t> masterRanges;
  std::vector<uint32_t> withEdgeRanges;
  std::vector<uint32_t> allNodesRangesIn;
  std::vector<uint32_t> masterRangesIn;

  using NodeRangeType =
      galois::runtime::SpecificRange<boost::counting_iterator<size_t>>;

  std::vector<NodeRangeType> specificRanges;
  std::vector<NodeRangeType> specificRangesIn;

protected:
  void inline increment_evilPhase() {
    ++galois::runtime::evilPhase;
    if (galois::runtime::evilPhase >=
        static_cast<uint32_t>(
            std::numeric_limits<int16_t>::max())) { // limit defined by MPI or
                                                    // LCI
      galois::runtime::evilPhase = 1;
    }
  }

  unsigned inline evilPhasePlus1() {
    unsigned result = galois::runtime::evilPhase + 1;

    // limit defined by MPI or LCI
    if (result >= uint32_t{std::numeric_limits<int16_t>::max()}) {
      return 1;
    }
    return result;
  }

  template <typename GraphNode, typename ET>
  struct IdLess {
    bool
    operator()(const galois::graphs::EdgeSortValue<GraphNode, ET>& e1,
               const galois::graphs::EdgeSortValue<GraphNode, ET>& e2) const {
      return e1.dst < e2.dst;
    }
  };

private:
  void computeMastersBlockedNodes(galois::graphs::OfflineGraph& g,
                                  const std::vector<unsigned>& scalefactor,
                                  unsigned DecomposeFactor = 1) {
    uint64_t numNodes_to_divide = g.size();
    if (scalefactor.empty() || (numHosts * DecomposeFactor == 1)) {
      for (unsigned i = 0; i < numHosts * DecomposeFactor; ++i)
        gid2host.push_back(galois::block_range(uint64_t{0}, numNodes_to_divide,
                                               i, numHosts * DecomposeFactor));
      return;
    }

    // TODO: not compatible with DecomposeFactor.
    assert(scalefactor.size() == numHosts);

    unsigned numBlocks = 0;

    for (unsigned i = 0; i < numHosts; ++i) {
      numBlocks += scalefactor[i];
    }

    std::vector<std::pair<uint64_t, uint64_t>> blocks;
    for (unsigned i = 0; i < numBlocks; ++i) {
      blocks.push_back(
          galois::block_range(uint64_t{0}, numNodes_to_divide, i, numBlocks));
    }

    std::vector<unsigned> prefixSums;
    prefixSums.push_back(0);

    for (unsigned i = 1; i < numHosts; ++i) {
      prefixSums.push_back(prefixSums[i - 1] + scalefactor[i - 1]);
    }

    for (unsigned i = 0; i < numHosts; ++i) {
      unsigned firstBlock = prefixSums[i];
      unsigned lastBlock  = prefixSums[i] + scalefactor[i] - 1;
      gid2host.push_back(
          std::make_pair(blocks[firstBlock].first, blocks[lastBlock].second));
    }
  }

  void computeMastersBalancedEdges(galois::graphs::OfflineGraph& g,
                                   const std::vector<unsigned>& scalefactor,
                                   uint32_t edgeWeight,
                                   unsigned DecomposeFactor = 1) {
    if (edgeWeight == 0) {
      edgeWeight = 1;
    }

    auto& net = galois::runtime::getSystemNetworkInterface();

    gid2host.resize(numHosts * DecomposeFactor);
    for (unsigned d = 0; d < DecomposeFactor; ++d) {
      auto r = g.divideByNode(0, edgeWeight, (id + d * numHosts),
                              numHosts * DecomposeFactor, scalefactor);
      gid2host[id + d * numHosts].first  = *(r.first.first);
      gid2host[id + d * numHosts].second = *(r.first.second);
    }

    for (unsigned h = 0; h < numHosts; ++h) {
      if (h == id) {
        continue;
      }
      galois::runtime::SendBuffer b;
      for (unsigned d = 0; d < DecomposeFactor; ++d) {
        galois::runtime::gSerialize(b, gid2host[id + d * numHosts]);
      }
      net.sendTagged(h, galois::runtime::evilPhase, b);
    }
    net.flush();
    unsigned received = 1;
    while (received < numHosts) {
      decltype(net.recieveTagged(galois::runtime::evilPhase, nullptr)) p;
      do {
        p = net.recieveTagged(galois::runtime::evilPhase, nullptr);
      } while (!p);
      assert(p->first != id);
      auto& b = p->second;
      for (unsigned d = 0; d < DecomposeFactor; ++d) {
        galois::runtime::gDeserialize(b, gid2host[p->first + d * numHosts]);
      }
      ++received;
    }
    increment_evilPhase();

#ifndef NDEBUG
    for (unsigned h = 0; h < numHosts; h++) {
      if (h == 0) {
        assert(gid2host[h].first == 0);
      } else if (h == numHosts - 1) {
        assert(gid2host[h].first == gid2host[h - 1].second);
        assert(gid2host[h].second == g.size());
      } else {
        assert(gid2host[h].first == gid2host[h - 1].second);
        assert(gid2host[h].second == gid2host[h + 1].first);
      }
    }
#endif
  }

  void computeMastersBalancedNodesAndEdges(
      galois::graphs::OfflineGraph& g, const std::vector<unsigned>& scalefactor,
      uint32_t nodeWeight, uint32_t edgeWeight, unsigned) {
    if (nodeWeight == 0) {
      nodeWeight = g.sizeEdges() / g.size(); // average degree
    }
    if (edgeWeight == 0) {
      edgeWeight = 1;
    }

    auto& net = galois::runtime::getSystemNetworkInterface();
    gid2host.resize(numHosts);
    auto r = g.divideByNode(nodeWeight, edgeWeight, id, numHosts, scalefactor);
    gid2host[id].first  = *r.first.first;
    gid2host[id].second = *r.first.second;
    for (unsigned h = 0; h < numHosts; ++h) {
      if (h == id)
        continue;
      galois::runtime::SendBuffer b;
      galois::runtime::gSerialize(b, gid2host[id]);
      net.sendTagged(h, galois::runtime::evilPhase, b);
    }
    net.flush();
    unsigned received = 1;
    while (received < numHosts) {
      decltype(net.recieveTagged(galois::runtime::evilPhase, nullptr)) p;
      do {
        p = net.recieveTagged(galois::runtime::evilPhase, nullptr);
      } while (!p);
      assert(p->first != id);
      auto& b = p->second;
      galois::runtime::gDeserialize(b, gid2host[p->first]);
      ++received;
    }
    increment_evilPhase();
  }

protected:
  uint64_t computeMasters(MASTERS_DISTRIBUTION masters_distribution,
                          galois::graphs::OfflineGraph& g,
                          const std::vector<unsigned>& scalefactor,
                          uint32_t nodeWeight = 0, uint32_t edgeWeight = 0,
                          unsigned DecomposeFactor = 1) {
    galois::Timer timer;
    timer.start();
    g.reset_seek_counters();

    uint64_t numNodes_to_divide = g.size();

    // compute masters for all nodes
    switch (masters_distribution) {
    case BALANCED_MASTERS:
      computeMastersBlockedNodes(g, scalefactor, DecomposeFactor);
      break;
    case BALANCED_MASTERS_AND_EDGES:
      computeMastersBalancedNodesAndEdges(g, scalefactor, nodeWeight,
                                          edgeWeight, DecomposeFactor);
      break;
    case BALANCED_EDGES_OF_MASTERS:
    default:
      computeMastersBalancedEdges(g, scalefactor, edgeWeight, DecomposeFactor);
      break;
    }

    timer.stop();

    galois::runtime::reportStatCond_Tmax<MORE_DIST_STATS>(
        GRNAME, "MasterDistTime", timer.get());

    galois::gPrint(
        "[", id, "] Master distribution time : ", timer.get_usec() / 1000000.0f,
        " seconds to read ", g.num_bytes_read(), " bytes in ", g.num_seeks(),
        " seeks (", g.num_bytes_read() / (float)timer.get_usec(), " MBPS)\n");
    return numNodes_to_divide;
  }

  void readersFromFile(galois::graphs::OfflineGraph& g, std::string filename) {
    // read file lines
    std::ifstream mappings(filename);
    std::string curLine;

    unsigned timesToRead = id + 1;

    for (unsigned i = 0; i < timesToRead; i++) {
      std::getline(mappings, curLine);
    }

    std::vector<char> modifyLine(curLine.begin(), curLine.end());
    char* tokenizedString = modifyLine.data();
    char* token;
    token = strtok(tokenizedString, " ");

    // loop 6 more times
    for (unsigned i = 0; i < 6; i++) {
      token = strtok(NULL, " ");
    }
    std::string left(token);

    // 3 more times for right
    for (unsigned i = 0; i < 3; i++) {
      token = strtok(NULL, " ");
    }
    std::string right(token);

    gid2host.resize(numHosts);
    gid2host[id].first  = std::stoul(left);
    gid2host[id].second = std::stoul(right) + 1;
    galois::gPrint("[", id, "] Left: ", gid2host[id].first,
                   ", Right: ", gid2host[id].second, "\n");

    // send/recv from other hosts
    auto& net = galois::runtime::getSystemNetworkInterface();

    for (unsigned h = 0; h < numHosts; ++h) {
      if (h == id)
        continue;
      galois::runtime::SendBuffer b;
      galois::runtime::gSerialize(b, gid2host[id]);
      net.sendTagged(h, galois::runtime::evilPhase, b);
    }
    net.flush();
    unsigned received = 1;
    while (received < numHosts) {
      decltype(net.recieveTagged(galois::runtime::evilPhase, nullptr)) p;
      do {
        p = net.recieveTagged(galois::runtime::evilPhase, nullptr);
      } while (!p);
      assert(p->first != id);
      auto& b = p->second;
      galois::runtime::gDeserialize(b, gid2host[p->first]);
      ++received;
    }
    increment_evilPhase();

    // sanity checking assignment
    for (unsigned h = 0; h < numHosts; h++) {
      if (h == 0) {
        GALOIS_ASSERT(gid2host[h].first == 0);
      } else if (h == numHosts - 1) {
        GALOIS_ASSERT(gid2host[h].first == gid2host[h - 1].second,
                      gid2host[h].first, " ", gid2host[h - 1].second);
        GALOIS_ASSERT(gid2host[h].second == g.size(), gid2host[h].second, " ",
                      g.size());
      } else {
        GALOIS_ASSERT(gid2host[h].first == gid2host[h - 1].second,
                      gid2host[h].first, " ", gid2host[h - 1].second);
        GALOIS_ASSERT(gid2host[h].second == gid2host[h + 1].first,
                      gid2host[h].second, " ", gid2host[h + 1].first);
      }
    }
  }

  uint32_t G2L(uint64_t gid) const {
    assert(isLocal(gid));
    return globalToLocalMap.at(gid);
  }

  uint64_t L2G(uint32_t lid) const { return localToGlobalVector[lid]; }

public:
  using GraphNode = typename GraphTy::GraphNode;
  using EdgeType = EdgeTy;
  using iterator = typename GraphTy::iterator;
  using const_iterator = typename GraphTy::const_iterator;
  using edge_iterator = typename GraphTy::edge_iterator;

  DistGraph(unsigned host, unsigned numHosts)
      : transposed(false), id(host), numHosts(numHosts) {
    mirrorNodes.resize(numHosts);
    numGlobalNodes = 0;
    numGlobalEdges = 0;
  }

  std::vector<std::pair<uint32_t, uint32_t>> getMirrorRanges() const {
    std::vector<std::pair<uint32_t, uint32_t>> mirrorRangesVector;
    // order of nodes locally is masters, outgoing mirrors, incoming mirrors,
    // so just get from numOwned to end
    if (numOwned != numNodes) {
      assert(numOwned < numNodes);
      mirrorRangesVector.push_back(std::make_pair(numOwned, numNodes));
    }
    return mirrorRangesVector;
  }

  std::vector<std::vector<size_t>>& getMirrorNodes() { return mirrorNodes; }

  virtual unsigned getHostID(uint64_t) const = 0;
  virtual bool isOwned(uint64_t) const = 0;
  virtual bool isLocal(uint64_t) const = 0;
  virtual bool is_vertex_cut() const = 0;
  virtual std::pair<unsigned, unsigned> cartesianGrid() const {
    return std::make_pair(0u, 0u);
  }

  bool isTransposed() { return transposed; }

  inline uint64_t getGID(const uint32_t nodeID) const { return L2G(nodeID); }

  inline uint32_t getLID(const uint64_t nodeID) const { return G2L(nodeID); }

  inline NodeTy&
  getData(GraphNode N,
          galois::MethodFlag mflag = galois::MethodFlag::UNPROTECTED) {
    auto& r = graph.getData(N, mflag);
    return r;
  }

  inline typename GraphTy::edge_data_reference
  getEdgeData(edge_iterator ni,
              galois::MethodFlag mflag = galois::MethodFlag::UNPROTECTED) {
    auto& r = graph.getEdgeData(ni, mflag);
    return r;
  }

  GraphNode getEdgeDst(edge_iterator ni) { return graph.getEdgeDst(ni); }

  inline edge_iterator edge_begin(GraphNode N) {
    return graph.edge_begin(N, galois::MethodFlag::UNPROTECTED);
  }

  inline edge_iterator edge_end(GraphNode N) {
    return graph.edge_end(N, galois::MethodFlag::UNPROTECTED);
  }

  inline galois::runtime::iterable<galois::NoDerefIterator<edge_iterator>>
  edges(GraphNode N) {
    return galois::graphs::internal::make_no_deref_range(edge_begin(N),
                                                         edge_end(N));
  }

  inline size_t size() const { return graph.size(); }

  inline size_t sizeEdges() const { return graph.sizeEdges(); }

  inline size_t numMasters() const { return numOwned; }

  inline size_t getNumNodesWithEdges() const { return numNodesWithEdges; }

  inline size_t globalSize() const { return numGlobalNodes; }

  inline size_t globalSizeEdges() const { return numGlobalEdges; }

  inline const NodeRangeType& allNodesRange() const {
    assert(specificRanges.size() == 3);
    return specificRanges[0];
  }

  inline const NodeRangeType& masterNodesRange() const {
    assert(specificRanges.size() == 3);
    return specificRanges[1];
  }

  inline const NodeRangeType& allNodesWithEdgesRange() const {
    assert(specificRanges.size() == 3);
    return specificRanges[2];
  }

protected:
  inline void determineThreadRanges() {
    allNodesRanges = galois::graphs::determineUnitRangesFromPrefixSum(
        galois::runtime::activeThreads, graph.getEdgePrefixSum());
  }

  inline void determineThreadRangesMaster() {
    // make sure this hasn't been called before
    assert(masterRanges.size() == 0);

    // first check if we even need to do any work; if already calculated,
    // use already calculated vector
    if (beginMaster == 0 && (beginMaster + numOwned) == size()) {
      masterRanges = allNodesRanges;
    } else if (beginMaster == 0 &&
               (beginMaster + numOwned) == numNodesWithEdges &&
               withEdgeRanges.size() != 0) {
      masterRanges = withEdgeRanges;
    } else {
      galois::gDebug("Manually det. master thread ranges");
      masterRanges = galois::graphs::determineUnitRangesFromGraph(
          graph, galois::runtime::activeThreads, beginMaster,
          beginMaster + numOwned, 0);
    }
  }

  inline void determineThreadRangesWithEdges() {
    // make sure not called before
    assert(withEdgeRanges.size() == 0);

    // first check if we even need to do any work; if already calculated,
    // use already calculated vector
    if (numNodesWithEdges == size()) {
      withEdgeRanges = allNodesRanges;
    } else if (beginMaster == 0 &&
               (beginMaster + numOwned) == numNodesWithEdges &&
               masterRanges.size() != 0) {
      withEdgeRanges = masterRanges;
    } else {
      galois::gDebug("Manually det. with edges thread ranges");
      withEdgeRanges = galois::graphs::determineUnitRangesFromGraph(
          graph, galois::runtime::activeThreads, 0, numNodesWithEdges, 0);
    }
  }

  void initializeSpecificRanges() {
    assert(specificRanges.size() == 0);

    // TODO/FIXME assertion likely not safe if a host gets no nodes
    // make sure the thread ranges have already been calculated
    // for the 3 ranges
    assert(allNodesRanges.size() != 0);
    assert(masterRanges.size() != 0);
    assert(withEdgeRanges.size() != 0);

    // 0 is all nodes
    specificRanges.push_back(galois::runtime::makeSpecificRange(
        boost::counting_iterator<size_t>(0),
        boost::counting_iterator<size_t>(size()), allNodesRanges.data()));

    // 1 is master nodes
    specificRanges.push_back(galois::runtime::makeSpecificRange(
        boost::counting_iterator<size_t>(beginMaster),
        boost::counting_iterator<size_t>(beginMaster + numOwned),
        masterRanges.data()));

    // 2 is with edge nodes
    specificRanges.push_back(galois::runtime::makeSpecificRange(
        boost::counting_iterator<size_t>(0),
        boost::counting_iterator<size_t>(numNodesWithEdges),
        withEdgeRanges.data()));

    assert(specificRanges.size() == 3);
  }

  void edgesEqualMasters() { specificRanges[2] = specificRanges[1]; }

public:
  void save_local_graph_to_file(std::string) { GALOIS_DIE("not implemented"); }

  void read_local_graph_from_file(std::string) {
    GALOIS_DIE("not implemented");
  }

  void deallocate() {
    galois::gDebug("Deallocating CSR in DistGraph");
    graph.deallocate();
  }

  void sortEdgesByDestination() {
    using GN = typename GraphTy::GraphNode;
    galois::do_all(
        galois::iterate(graph),
        [&](GN n) { graph.sortEdges(n, IdLess<GN, EdgeTy>()); },
        galois::no_stats(), galois::loopname("CSREdgeSort"), galois::steal());
  }
};

template <typename NodeTy, typename EdgeTy>
constexpr const char* const galois::graphs::DistGraph<NodeTy, EdgeTy>::GRNAME;
} // end namespace graphs
} // end namespace galois

#endif //_GALOIS_DIST_HGRAPH_H