Update the graph data structure

Use Compressed Sparse Row (CSR) to store graphs instead of N vectors to
save memory.

Author: lqhl

.clang-format

@@ -4,4 +4,4 @@ UseTab: Never
 BreakBeforeBraces: Linux
 AllowShortIfStatementsOnASingleLine: false
 IndentCaseLabels: false
-ForEachMacros: ['RANGES_FOR', 'FOREACH', 'rep']
+ForEachMacros: ['RANGES_FOR', 'FOREACH', 'rep', 'repv']

src/neighbor_partitioner.cpp

@@ -35,15 +35,10 @@ NeighborPartitioner::NeighborPartitioner(std::string basefilename)
     assigned_edges = 0;
     max_sample_size = num_vertices * 2;
     local_average_degree = 2 * (double)max_sample_size / num_vertices;
-    sample_size = 0;
     capacity = (double)num_edges * 1.05 / p + 1;
     occupied.assign(p, 0);
     adj_out.resize(num_vertices);
     adj_in.resize(num_vertices);
-    rep (i, num_vertices) {
-        adj_out[i].reserve(local_average_degree);
-        adj_in[i].reserve(local_average_degree);
-    }
     is_cores.assign(p, boost::dynamic_bitset<>(num_vertices));
     is_boundarys.assign(p, boost::dynamic_bitset<>(num_vertices));
     master.assign(num_vertices, -1);
@@ -57,27 +52,30 @@ NeighborPartitioner::NeighborPartitioner(std::string basefilename)
 
 void NeighborPartitioner::read_more()
 {
-    while (sample_size < max_sample_size && fin_ptr < fin_end) {
+    while (sample_edges.size() < max_sample_size && fin_ptr < fin_end) {
         edge_t *e = (edge_t *)fin_ptr;
         fin_ptr += sizeof(edge_t);
         if (check_edge(e)) {
-            sample_size++;
-            adj_out[e->first].push_back(e->second);
-            adj_in[e->second].push_back(e->first);
+            sample_edges.push_back(*e);
         }
     }
+
+    adj_out.build(sample_edges);
+
+    adj_in.build_reverse(sample_edges);
+
+    sample_edges.clear();
 }
 
 void NeighborPartitioner::read_remaining()
 {
     auto &is_boundary = is_boundarys[p - 1], &is_core = is_cores[p - 1];
 
-    rep (u, num_vertices)
-        for (auto &v : adj_out[u]) {
-            is_boundary[u] = true;
-            is_boundary[v] = true;
-            assign_edge(p - 1, u, v);
-        }
+    for (auto &e : sample_edges) {
+        is_boundary[e.first] = true;
+        is_boundary[e.second] = true;
+        assign_edge(p - 1, e.first, e.second);
+    }
 
     while (fin_ptr < fin_end) {
         edge_t *e = (edge_t *)fin_ptr;
@@ -89,7 +87,7 @@ void NeighborPartitioner::read_remaining()
         }
     }
 
-    rep (i, num_vertices) {
+    repv (i, num_vertices) {
         if (is_boundary[i]) {
             is_core[i] = true;
             rep (j, p - 1)
@@ -103,18 +101,12 @@ void NeighborPartitioner::read_remaining()
 
 void NeighborPartitioner::clean_samples()
 {
-    rep (u, num_vertices) {
+    repv (u, num_vertices) {
         adjlist_t &neighbors = adj_out[u];
-        for (size_t i = 0; i < neighbors.size();) {
-            vid_t &v = neighbors[i];
-            edge_t e(u, v);
-            if (!check_edge(&e)) {
-                sample_size--;
-                erase_one(adj_in[v], u);
-                std::swap(v, neighbors.back());
-                neighbors.pop_back();
-            } else
-                i++;
+        for (size_t i = 0; i < neighbors.size(); i++) {
+            edge_t e(u, neighbors[i]);
+            if (check_edge(&e))
+                sample_edges.push_back(e);
         }
     }
 }
@@ -170,14 +162,16 @@ void NeighborPartitioner::split()
     Timer read_timer, compute_timer;
 
     min_heap.reserve(num_vertices);
+    sample_edges.reserve(max_sample_size);
+    LOG(INFO) << "partitioning...";
     for (bucket = 0; bucket < p - 1; bucket++) {
-        DLOG(INFO) << "start partition " << bucket;
+        std::cerr << bucket << ", ";
         read_timer.start();
         read_more();
         read_timer.stop();
-        DLOG(INFO) << "sample size: " << sample_size;
+        DLOG(INFO) << "sample size: " << adj_out.num_edges();
         compute_timer.start();
-        local_capacity = sample_size / (p - bucket);
+        local_capacity = adj_out.num_edges() / (p - bucket);
         while (occupied[bucket] < local_capacity) {
             vid_t d, vid;
             if (!min_heap.get_min(d, vid)) {
@@ -199,7 +193,7 @@ void NeighborPartitioner::split()
         compute_timer.stop();
     }
     bucket = p - 1;
-    DLOG(INFO) << "start partition " << bucket;
+    std::cerr << bucket << std::endl;
     read_timer.start();
     read_remaining();
     read_timer.stop();

src/neighbor_partitioner.hpp

@@ -15,26 +15,109 @@
 #include "util.hpp"
 #include "min_heap.hpp"
 
-class NeighborPartitioner
+class adjlist_t
+{
+  private:
+    vid_t *start;
+    vid_t len;
+
+  public:
+    adjlist_t() : start(NULL), len(0) {}
+    adjlist_t(vid_t *start, vid_t len) : start(start), len(len) {}
+    vid_t *begin() { return start; }
+    vid_t *end() { return start + len; }
+    void increment() { len++; }
+    size_t size() const { return len; }
+    vid_t &operator[](size_t idx) { return start[idx]; };
+    const vid_t &operator[](size_t idx) const { return start[idx]; };
+    vid_t &back() { return start[len - 1]; };
+    const vid_t &back() const { return start[len - 1]; };
+    void pop_back() { len--; }
+    void clear() { len = 0; }
+};
+
+class graph_t
 {
   private:
-    typedef std::vector<vid_t> adjlist_t;
-    typedef std::vector<adjlist_t> graph_t;
+    vid_t num_vertices;
+    std::vector<vid_t> neighbors;
+    std::vector<adjlist_t> vdata;
+
+  public:
+    void resize(vid_t _num_vertices)
+    {
+        num_vertices = _num_vertices;
+        vdata.resize(num_vertices);
+    }
+
+    size_t num_edges() const { return neighbors.size(); }
+
+    void build(std::vector<edge_t> &edges)
+    {
+        repv (v, num_vertices)
+            vdata[v].clear();
+        std::sort(edges.begin(), edges.end(),
+                  [](const edge_t &a, const edge_t &b) {
+                      return a.first < b.first;
+                  });
+        neighbors.resize(edges.size());
+        vid_t last_v = -1;
+        vid_t *start_ptr = &neighbors[0];
+        for (size_t i = 0; i < edges.size(); i++) {
+            neighbors[i] = edges[i].second;
+            if (edges[i].first == last_v) {
+                vdata[last_v].increment();
+            } else {
+                vdata[edges[i].first] = adjlist_t(start_ptr + i, 1);
+                last_v = edges[i].first;
+            }
+        }
+    }
+
+    void build_reverse(std::vector<edge_t> &edges)
+    {
+        repv (v, num_vertices)
+            vdata[v].clear();
+        std::sort(edges.begin(), edges.end(),
+                  [](const edge_t &a, const edge_t &b) {
+                      return a.second < b.second;
+                  });
+        neighbors.resize(edges.size());
+        vid_t last_v = -1;
+        vid_t *start_ptr = &neighbors[0];
+        for (size_t i = 0; i < edges.size(); i++) {
+            neighbors[i] = edges[i].first;
+            if (edges[i].second == last_v) {
+                vdata[last_v].increment();
+            } else {
+                vdata[edges[i].second] = adjlist_t(start_ptr + i, 1);
+                last_v = edges[i].second;
+            }
+        }
+    }
+
+    adjlist_t &operator[](size_t idx) { return vdata[idx]; };
+    const adjlist_t &operator[](size_t idx) const { return vdata[idx]; };
+};
 
+class NeighborPartitioner
+{
+  private:
     std::string basefilename;
 
     vid_t num_vertices;
     size_t num_edges, assigned_edges;
     int p, bucket;
     double average_degree, local_average_degree;
-    size_t sample_size, max_sample_size;
+    size_t max_sample_size;
     size_t capacity, local_capacity;
 
     // use mmap for file input
     int fin;
     off_t filesize;
     char *fin_map, *fin_ptr, *fin_end;
 
+    std::vector<edge_t> sample_edges;
     graph_t adj_out, adj_in;
     MinHeap<vid_t, vid_t> min_heap;
     std::vector<size_t> occupied;
@@ -126,13 +209,11 @@ class NeighborPartitioner
             for (size_t i = 0; i < neighbors.size();) {
                 vid_t &u = neighbors[i];
                 if (is_core[u]) {
-                    sample_size--;
                     assign_edge(bucket, direction ? vid : u, direction ? u : vid);
                     min_heap.decrease_key(vid);
                     std::swap(u, neighbors.back());
                     neighbors.pop_back();
                 } else if (is_boundary[u] && occupied[bucket] < local_capacity) {
-                    sample_size--;
                     assign_edge(bucket, direction ? vid : u, direction ? u : vid);
                     min_heap.decrease_key(vid);
                     erase_one(adj_r[u], vid);
@@ -155,11 +236,11 @@ class NeighborPartitioner
 
         add_boundary(vid);
 
-        for (auto& w : adj_out[vid])
+        for (auto &w : adj_out[vid])
             add_boundary(w);
         adj_out[vid].clear();
 
-        for (auto& w : adj_in[vid])
+        for (auto &w : adj_in[vid])
             add_boundary(w);
         adj_in[vid].clear();
     }

src/util.hpp

@@ -11,6 +11,7 @@
 #include "threadpool11/threadpool11.hpp"
 
 #define rep(i, n) for (int i = 0; i < (int)(n); ++i)
+#define repv(i, n) for (vid_t i = 0; i < n; ++i)
 
 DECLARE_int32(p);
 DECLARE_uint64(memsize);