Make sketches_factor actually do something and fix exception race-case

include/cc_alg_configuration.h

@@ -26,7 +26,7 @@ class CCAlgConfiguration {
 
   // getters
   std::string get_disk_dir() { return _disk_dir; }
-  double get_sketch_factor() { return _sketches_factor; }
+  double get_sketches_factor() { return _sketches_factor; }
   double get_batch_factor() { return _batch_factor; }
 
   friend std::ostream& operator<< (std::ostream &out, const CCAlgConfiguration &conf);

include/cc_sketch_alg.h

@@ -45,9 +45,9 @@ struct alignas(64) GlobalMergeData {
   size_t num_merge_needed = -1;
   size_t num_merge_done = 0;
 
-  GlobalMergeData(node_id_t num_vertices, size_t seed)
+  GlobalMergeData(node_id_t num_vertices, size_t seed, double sketches_factor)
       : sketch(Sketch::calc_vector_length(num_vertices), seed,
-               Sketch::calc_cc_samples(num_vertices)) {}
+               Sketch::calc_cc_samples(num_vertices, sketches_factor)) {}
 
   GlobalMergeData(const GlobalMergeData&& other)
   : sketch(other.sketch) {
@@ -155,8 +155,9 @@ class CCSketchAlg {
     num_delta_sketches = num_workers;
     delta_sketches = new Sketch *[num_delta_sketches];
     for (size_t i = 0; i < num_delta_sketches; i++) {
-      delta_sketches[i] = new Sketch(Sketch::calc_vector_length(num_vertices), seed,
-                                     Sketch::calc_cc_samples(num_vertices));
+      delta_sketches[i] =
+          new Sketch(Sketch::calc_vector_length(num_vertices), seed,
+                     Sketch::calc_cc_samples(num_vertices, config.get_sketches_factor()));
     }
   }
 

include/sketch.h

@@ -63,6 +63,18 @@ class Sketch {
     return ceil(double(num_vertices) * (num_vertices - 1) / 2);
   }
 
+  /**
+   * This function computes the number of samples a Sketch should support in order to solve
+   * connected components. Optionally, can increase or decrease the number of samples by a
+   * multiplicative factor.
+   * @param num_vertices   Number of graph vertices
+   * @param f              Multiplicative sample factor
+   * @return               The number of samples
+   */
+  static size_t calc_cc_samples(node_id_t num_vertices, double f) {
+    return ceil(f * log2(num_vertices) / num_samples_div);
+  }
+
   /**
    * Construct a sketch object
    * @param vector_len       Length of the vector we are sketching
@@ -167,7 +179,6 @@ class Sketch {
   inline size_t get_num_samples() const { return num_samples; }
 
   static size_t calc_bkt_per_col(size_t n) { return ceil(log2(n)) + 1; }
-  static size_t calc_cc_samples(size_t n) { return ceil(log2(n) / num_samples_div); }
 
 #ifdef L0_SAMPLING
   static constexpr size_t default_cols_per_sample = 7;

src/cc_alg_configuration.cpp

@@ -14,11 +14,6 @@ CCAlgConfiguration& CCAlgConfiguration::sketches_factor(double factor) {
               << "Defaulting to 1." << std::endl;
     _sketches_factor = 1;
   }
-  if (_sketches_factor != 1) {
-    std::cerr << "WARNING: Your graph configuration specifies using a factor " << _sketches_factor 
-              << " of the normal quantity of sketches." << std::endl;
-    std::cerr << "         Is this intentional? If not, set sketches_factor to one!" << std::endl;
-  }
   return *this;
 }
 

src/cc_sketch_alg.cpp

@@ -14,7 +14,8 @@ CCSketchAlg::CCSketchAlg(node_id_t num_vertices, size_t seed, CCAlgConfiguration
   sketches = new Sketch *[num_vertices];
 
   vec_t sketch_vec_len = Sketch::calc_vector_length(num_vertices);
-  size_t sketch_num_samples = Sketch::calc_cc_samples(num_vertices);
+  size_t sketch_num_samples = Sketch::calc_cc_samples(num_vertices, config.get_sketches_factor());
+
   for (node_id_t i = 0; i < num_vertices; ++i) {
     representatives->insert(i);
     sketches[i] = new Sketch(sketch_vec_len, seed, sketch_num_samples);
@@ -48,7 +49,8 @@ CCSketchAlg::CCSketchAlg(node_id_t num_vertices, size_t seed, std::ifstream &bin
   sketches = new Sketch *[num_vertices];
 
   vec_t sketch_vec_len = Sketch::calc_vector_length(num_vertices);
-  size_t sketch_num_samples = Sketch::calc_cc_samples(num_vertices);
+  size_t sketch_num_samples = Sketch::calc_cc_samples(num_vertices, config.get_sketches_factor());
+
   for (node_id_t i = 0; i < num_vertices; ++i) {
     representatives->insert(i);
     sketches[i] = new Sketch(sketch_vec_len, seed, binary_stream, sketch_num_samples);
@@ -227,6 +229,7 @@ inline bool CCSketchAlg::run_round_zero() {
       if (sample_supernode(*sketches[i]) && !modified) modified = true;
     } catch (...) {
       except = true;
+#pragma omp critical
       err = std::current_exception();
     }
   }
@@ -258,14 +261,16 @@ bool CCSketchAlg::perform_boruvka_round(const size_t cur_round,
   {
     // some thread local variables
     Sketch local_sketch(Sketch::calc_vector_length(num_vertices), seed,
-                        Sketch::calc_cc_samples(num_vertices));
+                        Sketch::calc_cc_samples(num_vertices, config.get_sketches_factor()));
 
     size_t thr_id = omp_get_thread_num();
     size_t num_threads = omp_get_num_threads();
     std::pair<node_id_t, node_id_t> partition = get_ith_partition(num_vertices, thr_id, num_threads);
     node_id_t start = partition.first;
     node_id_t end = partition.second;
     assert(start <= end);
+    bool local_except = false;
+    std::exception_ptr local_err;
 
     // node_id_t left_root = merge_instr[start].root;
     // node_id_t right_root = merge_instr[end - 1].root;
@@ -298,8 +303,8 @@ bool CCSketchAlg::perform_boruvka_round(const size_t cur_round,
               // num_query += 1;
               if (sample_supernode(global_merges[thr_id].sketch) && !modified) modified = true;
             } catch (...) {
-              except = true;
-              err = std::current_exception();
+              local_except = true;
+              local_err = std::current_exception();
             }
           }
 
@@ -312,8 +317,8 @@ bool CCSketchAlg::perform_boruvka_round(const size_t cur_round,
             // num_query += 1;
             if (sample_supernode(local_sketch) && !modified) modified = true;
           } catch (...) {
-            except = true;
-            err = std::current_exception();
+            local_except = true;
+            local_err = std::current_exception();
           }
         }
 
@@ -343,8 +348,8 @@ bool CCSketchAlg::perform_boruvka_round(const size_t cur_round,
           // num_query += 1;
           if (sample_supernode(global_merges[global_id].sketch) && !modified) modified = true;
         } catch (...) {
-          except = true;
-          err = std::current_exception();
+          local_except = true;
+          local_err = std::current_exception();
         }
       }
     } else {
@@ -354,10 +359,15 @@ bool CCSketchAlg::perform_boruvka_round(const size_t cur_round,
         // num_query += 1;
         if (sample_supernode(local_sketch) && !modified) modified = true;
       } catch (...) {
-        except = true;
-        err = std::current_exception();
+        local_except = true;
+        local_err = std::current_exception();
       }
     }
+    if (local_except) {
+#pragma omp critical
+      err = local_err;
+      except = true;
+    }
   }
 
   // std::cout << "Number of roots queried = " << num_query << std::endl;
@@ -463,7 +473,7 @@ void CCSketchAlg::boruvka_emulation() {
   std::vector<GlobalMergeData> global_merges;
   global_merges.reserve(num_threads);
   for (size_t i = 0; i < num_threads; i++) {
-    global_merges.emplace_back(num_vertices, seed);
+    global_merges.emplace_back(num_vertices, seed, config.get_sketches_factor());
   }
 
   dsu.reset();

test/sketch_test.cpp

@@ -363,7 +363,7 @@ TEST(SketchTestSuite, TestExhaustiveQuery) {
 
 TEST(SketchTestSuite, TestSampleInsertGrinder) {
   size_t nodes = 4096;
-  Sketch sketch(Sketch::calc_vector_length(nodes), get_seed(), Sketch::calc_cc_samples(nodes));
+  Sketch sketch(Sketch::calc_vector_length(nodes), get_seed(), Sketch::calc_cc_samples(nodes, 1));
 
   for (size_t src = 0; src < nodes - 1; src++) {
     for (size_t dst = src + 7; dst < nodes; dst += 7) {
@@ -372,7 +372,7 @@ TEST(SketchTestSuite, TestSampleInsertGrinder) {
   }
 
   size_t successes = 0;
-  for (size_t i = 0; i < Sketch::calc_cc_samples(nodes); i++) {
+  for (size_t i = 0; i < Sketch::calc_cc_samples(nodes, 1); i++) {
     SketchSample ret = sketch.sample();
     if (ret.result == FAIL) continue;
 
@@ -388,7 +388,7 @@ TEST(SketchTestSuite, TestSampleInsertGrinder) {
 
 TEST(SketchTestSuite, TestSampleDeleteGrinder) {
   size_t nodes = 4096;
-  Sketch sketch(Sketch::calc_vector_length(nodes), get_seed(), Sketch::calc_cc_samples(nodes));
+  Sketch sketch(Sketch::calc_vector_length(nodes), get_seed(), Sketch::calc_cc_samples(nodes, 1));
 
   // insert
   for (size_t src = 0; src < nodes - 1; src++) {
@@ -405,7 +405,7 @@ TEST(SketchTestSuite, TestSampleDeleteGrinder) {
   }
 
   size_t successes = 0;
-  for (size_t i = 0; i < Sketch::calc_cc_samples(nodes); i++) {
+  for (size_t i = 0; i < Sketch::calc_cc_samples(nodes, 1); i++) {
     SketchSample ret = sketch.sample();
     if (ret.result == FAIL) continue;
 

tools/process_stream.cpp

@@ -85,7 +85,7 @@ int main(int argc, char **argv) {
   std::cout << std::endl;
 
   auto driver_config = DriverConfiguration().gutter_sys(CACHETREE).worker_threads(num_threads);
-  auto cc_config = CCAlgConfiguration().batch_factor(1);
+  auto cc_config = CCAlgConfiguration().batch_factor(1).sketches_factor(0.5);
   CCSketchAlg cc_alg{num_nodes, get_seed(), cc_config};
   GraphSketchDriver<CCSketchAlg> driver{&cc_alg, &stream, driver_config, reader_threads};
 

tools/test_correctness.cpp

@@ -22,7 +22,7 @@ CorrectnessResults test_path_correctness(size_t num_vertices, size_t num_graphs,
                                          size_t samples_per_graph) {
   CorrectnessResults results;
 
-  size_t num_rounds = Sketch::calc_cc_samples(num_vertices);
+  size_t num_rounds = Sketch::calc_cc_samples(num_vertices, 1);
   for (size_t r = 0; r < num_rounds; r++)
     results.num_round_hist.push_back(0);