Better, faster, stronger query algorithm for the refactor

CMakeLists.txt

@@ -102,6 +102,7 @@ set(BUILD_SHARED_LIBS "${SAVED_BUILD_SHARED_LIBS}" CACHE BOOL "" FORCE)
 
 add_library(GraphZeppelin
   src/cc_sketch_alg.cpp
+  src/return_types.cpp
   src/driver_configuration.cpp
   src/cc_alg_configuration.cpp
   src/sketch.cpp
@@ -115,6 +116,7 @@ target_compile_definitions(GraphZeppelin PUBLIC XXH_INLINE_ALL)
 
 add_library(GraphZeppelinVerifyCC
   src/cc_sketch_alg.cpp
+  src/return_types.cpp
   src/driver_configuration.cpp
   src/cc_alg_configuration.cpp
   src/sketch.cpp
@@ -184,3 +186,4 @@ if (BUILD_BENCH)
   add_dependencies(bench_cc GraphZeppelin benchmark)
   target_link_libraries(bench_cc GraphZeppelin benchmark::benchmark xxhash)
 endif()
+

include/cc_alg_configuration.h

@@ -31,9 +31,6 @@ class CCAlgConfiguration {
 
   friend std::ostream& operator<< (std::ostream &out, const CCAlgConfiguration &conf);
 
-  // no use of equal operator
-  CCAlgConfiguration& operator=(const CCAlgConfiguration &) = delete;
-
   // moving and copying allowed
   CCAlgConfiguration(const CCAlgConfiguration &oth) = default;
   CCAlgConfiguration (CCAlgConfiguration &&) = default;

include/cc_sketch_alg.h

@@ -1,6 +1,6 @@
 #pragma once
 
-#include <atomic>  // REMOVE LATER
+#include <atomic>
 #include <cstdlib>
 #include <exception>
 #include <fstream>
@@ -13,6 +13,7 @@
 #include <cassert>
 
 #include "cc_alg_configuration.h"
+#include "return_types.h"
 #include "sketch.h"
 #include "dsu.h"
 
@@ -27,6 +28,34 @@ class UpdateLockedException : public std::exception {
   }
 };
 
+struct MergeInstr {
+  node_id_t root;
+  node_id_t child;
+
+  inline bool operator< (const MergeInstr &oth) const {
+    if (root == oth.root)
+      return child < oth.child;
+    return root < oth.root;
+  }
+};
+
+struct alignas(64) GlobalMergeData {
+  Sketch sketch;
+  std::mutex mtx;
+  size_t num_merge_needed = -1;
+  size_t num_merge_done = 0;
+
+  GlobalMergeData(node_id_t num_vertices, size_t seed)
+      : sketch(Sketch::calc_vector_length(num_vertices), seed,
+               Sketch::calc_cc_samples(num_vertices)) {}
+
+  GlobalMergeData(const GlobalMergeData&& other)
+  : sketch(other.sketch) {
+    num_merge_needed = other.num_merge_needed;
+    num_merge_done = other.num_merge_done;
+  }
+};
+
 /**
  * Algorithm for computing connected components on undirected graph streams
  * (no self-edges or multi-edges)
@@ -56,38 +85,37 @@ class CCSketchAlg {
   Sketch **delta_sketches = nullptr;
   size_t num_delta_sketches;
 
+  /**
+   * Run the first round of Boruvka. We can do things faster here because we know there will
+   * be no merging we have to do.
+   */
+  bool run_round_zero();
+
   /**
    * Update the query array with new samples
    * @param query  an array of sketch sample results
    * @param reps   an array containing node indices for the representative of each supernode
    */
-  bool sample_supernodes(std::vector<node_id_t> &merge_instr);
+  bool sample_supernode(Sketch &skt);
+
 
   /**
-   * @param reps         set containing the roots of each supernode
-   * @param merge_instr  a list of lists of supernodes to be merged
+   * Calculate the instructions for what vertices to merge to form each component
    */
-  void merge_supernodes(const size_t next_round,
-                        const std::vector<node_id_t> &merge_instr);
+  void create_merge_instructions(std::vector<MergeInstr> &merge_instr);
 
   /**
    * @param reps         set containing the roots of each supernode
-   * @param merge_instr  an array where each vertex indicates its supernode root
+   * @param merge_instr  a list of lists of supernodes to be merged
    */
-  void undo_merge_supernodes(const size_t cur_round,
-                             const std::vector<node_id_t> &merge_instr);
+  bool perform_boruvka_round(const size_t cur_round, const std::vector<MergeInstr> &merge_instr,
+                             std::vector<GlobalMergeData> &global_merges);
 
   /**
    * Main parallel algorithm utilizing Boruvka and L_0 sampling.
-   * @return a vector of the connected components in the graph.
-   */
-  std::vector<std::set<node_id_t>> boruvka_emulation();
-
-  /**
-   * Generates connected components from this graph's dsu
-   * @return a vector of the connected components in the graph.
+   * Ensures that the DSU represents the Connected Components of the stream when called
    */
-  std::vector<std::set<node_id_t>> cc_from_dsu();
+  void boruvka_emulation();
 
   FRIEND_TEST(GraphTestSuite, TestCorrectnessOfReheating);
 
@@ -174,7 +202,7 @@ class CCSketchAlg {
    * Main parallel query algorithm utilizing Boruvka and L_0 sampling.
    * @return a vector of the connected components in the graph.
    */
-  std::vector<std::set<node_id_t>> connected_components();
+  ConnectedComponents connected_components();
 
   /**
    * Point query algorithm utilizing Boruvka and L_0 sampling.
@@ -190,7 +218,7 @@ class CCSketchAlg {
    * that is, unless you really know what you're doing.
    * @return an adjacency list representation of the spanning forest of the graph
    */
-  std::vector<std::pair<node_id_t, std::vector<node_id_t>>> calc_spanning_forest();
+  SpanningForest calc_spanning_forest();
 
 #ifdef VERIFY_SAMPLES_F
   std::unique_ptr<GraphVerifier> verifier;

include/dsu.h

@@ -5,6 +5,9 @@
 #include <random>
 #include <vector>
 
+#define likely_if(x) if(__builtin_expect((bool)(x), true))
+#define unlikely_if(x) if (__builtin_expect((bool)(x), false))
+
 template <class T>
 struct DSUMergeRet {
   bool merged;  // true if a merge actually occured
@@ -131,7 +134,7 @@ class DisjointSetUnion_MT {
 
   // make a copy of the DSU
   DisjointSetUnion_MT(const DisjointSetUnion_MT& oth)
-      : n(oth.n), parent(new T[n]), priority(new T[n]) {
+      : n(oth.n), parent(new std::atomic<T>[n]), priority(new T[n]) {
     for (T i = 0; i < n; i++) {
       parent[i] = oth.parent[i].load();
       priority[i] = oth.priority[i];

include/return_types.h

@@ -0,0 +1,36 @@
+// This file defines the query return types from the cc algorithm class
+#include <cstddef>
+#include <iterator>
+#include <set>
+#include <unordered_set>
+#include <vector>
+
+#include "dsu.h"
+#include "types.h"
+
+// This class defines the connected components of a graph
+class ConnectedComponents {
+ private:
+  node_id_t *parent_arr;
+  node_id_t num_vertices;
+  node_id_t num_cc;
+
+ public:
+  ConnectedComponents(node_id_t num_vertices, DisjointSetUnion_MT<node_id_t> &dsu);
+  ~ConnectedComponents();
+
+  std::vector<std::set<node_id_t>> get_component_sets();
+  bool is_connected(node_id_t a, node_id_t b) { return parent_arr[a] == parent_arr[b]; }
+  node_id_t size() { return num_cc; }
+};
+
+// This class defines a spanning forest of a graph
+class SpanningForest {
+ private:
+  std::vector<Edge> edges;
+  node_id_t num_vertices;
+ public:
+  SpanningForest(node_id_t num_vertices, const std::unordered_set<node_id_t> *spanning_forest);
+
+  const std::vector<Edge>& get_edges() { return edges; }
+};

include/types.h

@@ -43,6 +43,3 @@ struct GraphUpdate {
   Edge edge;
   UpdateType type;
 };
-
-#define likely_if(x) if(__builtin_expect((bool)(x), true))
-#define unlikely_if(x) if (__builtin_expect((bool)(x), false))

include/util.h

@@ -40,14 +40,5 @@ edge_id_t concat_pairing_fn(node_id_t i, node_id_t j);
  */
 Edge inv_concat_pairing_fn(edge_id_t idx);
 
-/**
- * Configures the system using the configuration file streaming.conf
- * Gets the path prefix where the buffer tree data will be stored and sets
- * with the number of threads used for a variety of tasks.
- * Should be called before creating the buffer tree or starting graph workers.
- * @return a tuple with the following elements
- *   - use_guttertree
- *   - in_memory_backups
- *   - disk_dir
- */
-std::tuple<bool, bool, std::string> configure_system();
+#define likely_if(x) if(__builtin_expect((bool)(x), true))
+#define unlikely_if(x) if (__builtin_expect((bool)(x), false))