more optimizations and simplifications

Author: etwest

include/driver_configuration.h

@@ -17,9 +17,12 @@ class DriverConfiguration {
   // Where to place on-disk datastructures
   std::string _disk_dir = ".";
 
-  // The number of worker threads
+  // The number of worker threads. These perform the algorithm updates.
   size_t _num_worker_threads = 1;
 
+  // The number of threads that read from the stream
+  size_t _num_stream_threads = 1;
+
   // Configuration for the guttering system
   GutteringConfiguration _gutter_conf;
 
@@ -29,13 +32,15 @@ class DriverConfiguration {
   // setters
   DriverConfiguration& gutter_sys(GutterSystem gutter_sys);
   DriverConfiguration& disk_dir(std::string disk_dir);
-  DriverConfiguration& worker_threads(size_t num_groups);
+  DriverConfiguration& worker_threads(size_t num_threads);
+  DriverConfiguration& stream_threads(size_t num_threads);
   GutteringConfiguration& gutter_conf();
 
   // getters
   GutterSystem get_gutter_sys() { return _gutter_sys; }
   std::string get_disk_dir() { return _disk_dir; }
   size_t get_worker_threads() { return _num_worker_threads; }
+  size_t get_stream_threads() { return _num_stream_threads; }
 
   friend std::ostream& operator<< (std::ostream &out, const DriverConfiguration &conf);
 

include/graph_sketch_driver.h

@@ -85,9 +85,10 @@ class GraphSketchDriver {
 
   std::atomic<size_t> total_updates;
  public:
-  GraphSketchDriver(Alg *sketching_alg, GraphStream *stream, DriverConfiguration config,
-                    size_t num_stream_threads = 1)
-      : sketching_alg(sketching_alg), stream(stream), num_stream_threads(num_stream_threads) {
+  GraphSketchDriver(Alg *sketching_alg, GraphStream *stream, DriverConfiguration config)
+      : sketching_alg(sketching_alg),
+        stream(stream),
+        num_stream_threads(config.get_stream_threads()) {
     sketching_alg->allocate_worker_memory(config.get_worker_threads());
     // set the leaf size of the guttering system appropriately
     if (config.gutter_conf().get_gutter_bytes() == GutteringConfiguration::uninit_param) {

include/sparse_sketch.h

@@ -56,6 +56,10 @@ class SparseSketch {
   // Allocated buckets
   Bucket* buckets;
 
+  // number of dense buckets to sample from when sparse region is populated
+  static constexpr size_t num_dense_to_sample = 2;
+
+  // minimum number of dense rows in a sketch no matter how sparse the vector is.
   static constexpr size_t min_num_dense_rows = 6;
   size_t num_dense_rows = min_num_dense_rows;
 
@@ -131,7 +135,7 @@ class SparseSketch {
   }
 
   inline size_t position_func(size_t col, size_t row, size_t num_rows) const {
-    return col * num_rows + row + 1;
+    return col * num_rows + row + 1; // column-major
   }
 
   // return the bucket at a particular index in bucket array

src/driver_configuration.cpp

@@ -12,12 +12,22 @@ DriverConfiguration& DriverConfiguration::disk_dir(std::string disk_dir) {
   return *this;
 }
 
-DriverConfiguration& DriverConfiguration::worker_threads(size_t num_worker_threads) {
-  _num_worker_threads = num_worker_threads;
-  if (_num_worker_threads < 1) {
-    std::cout << "num_worker_threads="<< _num_worker_threads << " is out of bounds. [1, infty)"
+DriverConfiguration& DriverConfiguration::worker_threads(size_t num_threads) {
+  if (num_threads < 1) {
+    std::cout << "num_worker_threads = "<< num_threads << " is out of bounds. [1, infty)"
               << "Defaulting to 1." << std::endl;
-    _num_worker_threads = 1;
+  } else {
+    _num_worker_threads = num_threads;
+  }
+  return *this;
+}
+
+DriverConfiguration& DriverConfiguration::stream_threads(size_t num_threads) {
+  if (num_threads < 1) {
+    std::cout << "num_stream_threads = "<< num_threads << " is out of bounds. [1, infty)"
+              << "Defaulting to 1." << std::endl;
+  } else {
+    _num_stream_threads = num_threads;
   }
   return *this;
 }
@@ -35,6 +45,7 @@ std::ostream& operator<< (std::ostream &out, const DriverConfiguration &conf) {
       gutter_system = "CacheTree";
     out << " Guttering system      = " << gutter_system << std::endl;
     out << " Worker thread count   = " << conf._num_worker_threads << std::endl;
+    out << " Stream thread count   = " << conf._num_stream_threads << std::endl;
     out << " On disk data location = " << conf._disk_dir;
     return out;
   }

src/sparse_sketch.cpp

@@ -252,17 +252,28 @@ void SparseSketch::update(const vec_t update_idx) {
   SketchBucket::Depths depths;
 
   // Update higher depth buckets
-  for (size_t i = 0; i < num_columns; i++) {
-    size_t bit = i & 0x1;
-    if (bit == 0) {
-      depths = SketchBucket::get_index_depths(update_idx, column_seed(i), bkt_per_col);
+  for (size_t i = 0; i < num_columns - 1; i += 2) {
+    depths = SketchBucket::get_index_depths(update_idx, column_seed(i), bkt_per_col);
+    for (size_t j = 0; j < 2; j++) {
+      col_hash_t depth = depths[j];
+      likely_if(depth < bkt_per_col) {
+        likely_if(depth < num_dense_rows) {
+          SketchBucket::update(bucket(i + j, depth), update_idx, checksum);
+        } else {
+          update_sparse(i + j, {uint8_t(-1), uint8_t(depth), {update_idx, checksum}});
+        }
+      }
     }
-    col_hash_t depth = depths[bit];
+  }
+  if ((num_columns & 0x1) == 1) {
+    size_t col = num_columns - 1;
+
+    size_t depth = SketchBucket::get_index_depth(update_idx, column_seed(col), bkt_per_col);
     likely_if(depth < bkt_per_col) {
       likely_if(depth < num_dense_rows) {
-        SketchBucket::update(bucket(i, depth), update_idx, checksum);
+        SketchBucket::update(bucket(col, depth), update_idx, checksum);
       } else {
-        update_sparse(i, {uint8_t(-1), uint8_t(depth), {update_idx, checksum}});
+        update_sparse(col, {uint8_t(-1), uint8_t(depth), {update_idx, checksum}});
       }
     }
   }
@@ -325,8 +336,14 @@ SketchSample SparseSketch::sample() {
     return sample;
   }
 
+  // if dense region is densely populated then only check the "deepest" few rows 
+  int dense_row_min = 0;
+  if (number_of_sparse_buckets > num_columns || num_dense_rows > min_num_dense_rows) {
+    dense_row_min = num_dense_rows - num_dense_to_sample;
+  }
+
   for (size_t c = 0; c < cols_per_sample; ++c) {
-    for (int r = num_dense_rows - 1; r >= 0; --r) { // TODO: Consider reducing the number of dense rows checked to 1 or 2
+    for (int r = num_dense_rows - 1; r >= dense_row_min; --r) {
       if (SketchBucket::is_good(bucket(c + first_column, r), checksum_seed())) {
         // std::cout << "Found GOOD dense bucket" << std::endl;
         return {bucket(c + first_column, r).alpha, GOOD};

test/cc_alg_test.cpp

@@ -299,7 +299,7 @@ TEST(CCAlgTest, InsertOnlyStream) {
 
 TEST(CCAlgTest, MTStreamWithMultipleQueries) {
   for (int t = 1; t <= 3; t++) {
-    auto driver_config = DriverConfiguration().gutter_sys(STANDALONE);
+    auto driver_config = DriverConfiguration().gutter_sys(STANDALONE).stream_threads(4);
 
     const std::string fname = __FILE__;
     size_t pos = fname.find_last_of("\\/");
@@ -313,7 +313,7 @@ TEST(CCAlgTest, MTStreamWithMultipleQueries) {
     std::cerr << num_nodes << " " << num_edges << std::endl;
 
     CCSketchAlg cc_alg{num_nodes, get_seed()};
-    GraphSketchDriver<CCSketchAlg> driver(&cc_alg, &stream, driver_config, 4);
+    GraphSketchDriver<CCSketchAlg> driver(&cc_alg, &stream, driver_config);
     GraphVerifier verify(num_nodes);
 
     size_t num_queries = 10;

tools/process_stream.cpp

@@ -84,11 +84,14 @@ int main(int argc, char **argv) {
   std::cout << "num_updates = " << num_updates << std::endl;
   std::cout << std::endl;
 
-  auto driver_config = DriverConfiguration().gutter_sys(CACHETREE).worker_threads(num_threads);
-  driver_config.gutter_conf().wq_batch_per_elm(4);
-  auto cc_config = CCAlgConfiguration().batch_factor(1);
+  auto driver_config = DriverConfiguration()
+                           .gutter_sys(CACHETREE)
+                           .worker_threads(num_threads)
+                           .stream_threads(reader_threads);
+  driver_config.gutter_conf().wq_batch_per_elm(2);
+  auto cc_config = CCAlgConfiguration().batch_factor(1.2);
   CCSketchAlg cc_alg{num_nodes, get_seed(), cc_config};
-  GraphSketchDriver<CCSketchAlg> driver{&cc_alg, &stream, driver_config, reader_threads};
+  GraphSketchDriver<CCSketchAlg> driver{&cc_alg, &stream, driver_config};
 
   auto ins_start = std::chrono::steady_clock::now();
   std::thread querier(track_insertions, num_updates, &driver, ins_start);