fix and make somewhat fast

Author: etwest

include/bucket.h

@@ -18,7 +18,7 @@ struct SparseBucket {
     return position >> 8;
   }
   inline uint16_t row() const {
-    return position & 0xFFFF;
+    return position & 0xFF;
   }
   inline void set_col(uint16_t col) {
     position = (col << 8) + row();

include/cc_sketch_alg.h

@@ -201,8 +201,9 @@ class CCSketchAlg {
    * Specifically, the delta is in the form of a pointer to raw bucket data.
    * @param src_vertex   The vertex where the all edges originate.
    * @param raw_buckets  Pointer to the array of buckets from the delta sketch
+   * @param num_buckets  Size of raw_buckets array in number of buckets
    */
-  void apply_raw_buckets_update(node_id_t src_vertex, Bucket *raw_buckets);
+  void apply_raw_buckets_update(node_id_t src_vertex, Bucket *raw_buckets, size_t num_buckets);
 
   /**
    * The function performs a direct update to the associated sketch.

include/dense_sketch.h

@@ -77,18 +77,19 @@ class DenseSketch {
    * @param cols_per_sample  [Optional] Number of sketch columns for each sample (default = 1)
    */
   DenseSketch(vec_t vector_len, uint64_t seed, size_t num_samples = 1,
-         size_t cols_per_sample = default_cols_per_sample);
+              size_t cols_per_sample = default_cols_per_sample);
 
   /**
    * Construct a sketch from a serialized stream
    * @param vector_len       Length of the vector we are sketching
    * @param seed             Random seed of the sketch
    * @param binary_in        Stream holding serialized sketch object
+   * @param num_buckets      Number of buckets in serialized sketch
    * @param num_samples      [Optional] Number of samples this sketch supports (default = 1)
    * @param cols_per_sample  [Optional] Number of sketch columns for each sample (default = 1)
    */
-  DenseSketch(vec_t vector_len, uint64_t seed, std::istream& binary_in, size_t num_samples = 1,
-         size_t cols_per_sample = default_cols_per_sample);
+  DenseSketch(vec_t vector_len, uint64_t seed, std::istream& binary_in, size_t num_buckets,
+              size_t num_samples = 1, size_t cols_per_sample = default_cols_per_sample);
 
   /**
    * Sketch copy constructor
@@ -139,9 +140,10 @@ class DenseSketch {
    * Perform an in-place merge function without another Sketch and instead
    * use a raw bucket memory.
    * We also allow for only a portion of the buckets to be merge at once
-   * @param raw_bucket    Raw bucket data to merge into this sketch
+   * @param raw_bucket     Raw bucket data to merge into this sketch
+   * @param n_raw_buckets  Size of raw_buckets in number of Bucket data-structures
    */
-  void merge_raw_bucket_buffer(const Bucket *raw_buckets);
+  void merge_raw_bucket_buffer(const Bucket *raw_buckets, size_t n_raw_buckets);
 
   /**
    * Zero out all the buckets of a sketch.

include/sparse_sketch.h

@@ -82,13 +82,14 @@ class SparseSketch {
   /**
    * Reallocates the bucket array if necessary to either grow or shrink the dense region
    */
-  void reallocate_if_needed();
+  void reallocate_if_needed(int delta);
+  void dense_realloc(size_t new_num_dense_rows);
 
   // This variable lets us know how many Buckets to allocate to make space for the SparseBuckets
   // that will be using that space
   size_t sparse_data_size = ceil(double(sparse_capacity) * sizeof(SparseBucket) / sizeof(Bucket));
 
-  void update_sparse(uint16_t pos, vec_t update_idx, vec_hash_t checksum);
+  void update_sparse(SparseBucket to_add, bool realloc_if_needed = true);
   SketchSample sample_sparse(size_t first_col, size_t end_col);
 
   inline Bucket& deterministic_bucket() {
@@ -151,11 +152,12 @@ class SparseSketch {
    * @param vector_len       Length of the vector we are sketching
    * @param seed             Random seed of the sketch
    * @param binary_in        Stream holding serialized sketch object
+   * @param num_buckets      Number of buckets in serialized sketch (dense + sparse_capacity)
    * @param num_samples      [Optional] Number of samples this sketch supports (default = 1)
    * @param cols_per_sample  [Optional] Number of sketch columns for each sample (default = 1)
    */
-  SparseSketch(vec_t vector_len, uint64_t seed, std::istream& binary_in, size_t num_samples = 1,
-         size_t cols_per_sample = default_cols_per_sample);
+  SparseSketch(vec_t vector_len, uint64_t seed, std::istream& binary_in, size_t num_buckets,
+               size_t num_samples = 1, size_t cols_per_sample = default_cols_per_sample);
 
   /**
    * SparseSketch copy constructor
@@ -206,9 +208,10 @@ class SparseSketch {
    * Perform an in-place merge function without another Sketch and instead
    * use a raw bucket memory.
    * We also allow for only a portion of the buckets to be merge at once
-   * @param raw_bucket    Raw bucket data to merge into this sketch
+   * @param raw_bucket     Raw bucket data to merge into this sketch
+   * @param n_raw_buckets  Size of raw_buckets in number of Bucket data-structures
    */
-  void merge_raw_bucket_buffer(const Bucket *raw_buckets);
+  void merge_raw_bucket_buffer(const Bucket *raw_buckets, size_t n_raw_buckets);
 
   /**
    * Zero out all the buckets of a sketch.
@@ -240,6 +243,7 @@ class SparseSketch {
   inline size_t get_columns() const { return num_columns; }
   inline size_t get_buckets() const { return num_buckets; }
   inline size_t get_num_samples() const { return num_samples; }
+  inline size_t get_num_dense_rows() const { return num_dense_rows; }
 
   static size_t calc_bkt_per_col(size_t n) { return ceil(log2(n)) + 1; }
 

src/cc_alg_configuration.cpp

@@ -34,6 +34,11 @@ std::ostream& operator<< (std::ostream &out, const CCAlgConfiguration &conf) {
 #else
     out << " Sketching algorithm   = CameoSketch" << std::endl;
 #endif
+#ifdef L0_FULLY_DENSE
+    out << " Sketch storage        = Dense Matrix" << std::endl;
+#else
+    out << " Sketch storage        = Hybrid Matrix" << std::endl;
+#endif
 #ifdef NO_EAGER_DSU
     out << " Using Eager DSU       = False" << std::endl;
 #else

src/cc_sketch_alg.cpp

@@ -53,7 +53,10 @@ CCSketchAlg::CCSketchAlg(node_id_t num_vertices, size_t seed, std::ifstream &bin
 
   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);
+    size_t num_bkts_in_sketch;
+    binary_stream.read((char *) &num_bkts_in_sketch, sizeof(num_bkts_in_sketch));
+    sketches[i] =
+        new Sketch(sketch_vec_len, seed, binary_stream, num_bkts_in_sketch, sketch_num_samples);
   }
   binary_stream.close();
 
@@ -117,9 +120,10 @@ void CCSketchAlg::apply_update_batch(int thr_id, node_id_t src_vertex,
   sketches[src_vertex]->merge(delta_sketch);
 }
 
-void CCSketchAlg::apply_raw_buckets_update(node_id_t src_vertex, Bucket *raw_buckets) {
+void CCSketchAlg::apply_raw_buckets_update(node_id_t src_vertex, Bucket *raw_buckets,
+                                           size_t num_buckets) {
   std::lock_guard<std::mutex> lk(sketches[src_vertex]->mutex);
-  sketches[src_vertex]->merge_raw_bucket_buffer(raw_buckets);
+  sketches[src_vertex]->merge_raw_bucket_buffer(raw_buckets, num_buckets);
 }
 
 // Note: for performance reasons route updates through the driver instead of calling this function
@@ -617,6 +621,8 @@ void CCSketchAlg::write_binary(const std::string &filename) {
   binary_out.write((char *)&num_vertices, sizeof(num_vertices));
   binary_out.write((char *)&config._sketches_factor, sizeof(config._sketches_factor));
   for (node_id_t i = 0; i < num_vertices; ++i) {
+    size_t num_bkts_in_sketch = sketches[i]->get_buckets();
+    binary_out.write((char*) &num_bkts_in_sketch, sizeof(num_bkts_in_sketch));
     sketches[i]->serialize(binary_out);
   }
   binary_out.close();

src/dense_sketch.cpp

@@ -1,9 +1,10 @@
 #include "dense_sketch.h"
 
+#include <cassert>
 #include <cstring>
+#include <exception>
 #include <iostream>
 #include <vector>
-#include <cassert>
 
 DenseSketch::DenseSketch(vec_t vector_len, uint64_t seed, size_t _samples, size_t _cols)
     : seed(seed),
@@ -22,13 +23,17 @@ DenseSketch::DenseSketch(vec_t vector_len, uint64_t seed, size_t _samples, size_
   }
 }
 
-DenseSketch::DenseSketch(vec_t vector_len, uint64_t seed, std::istream &binary_in, size_t _samples,
-                         size_t _cols)
+DenseSketch::DenseSketch(vec_t vector_len, uint64_t seed, std::istream &binary_in,
+                         size_t num_buckets, size_t _samples, size_t _cols)
     : seed(seed),
       num_samples(_samples),
       cols_per_sample(_cols),
       num_columns(cols_per_sample * num_samples),
-      bkt_per_col(calc_bkt_per_col(vector_len)) {
+      bkt_per_col(calc_bkt_per_col(vector_len)),
+      num_buckets(num_buckets) {
+  if (num_buckets != num_columns * bkt_per_col + 1) {
+    throw std::invalid_argument("Serial Constructor: Number of buckets does not match expectation");
+  }
   num_buckets = num_columns * bkt_per_col + 1; // plus 1 for deterministic bucket
   buckets = new Bucket[num_buckets];
 
@@ -188,7 +193,11 @@ void DenseSketch::range_merge(const DenseSketch &other, size_t start_sample, siz
   // std::cout << *this << std::endl;
 }
 
-void DenseSketch::merge_raw_bucket_buffer(const Bucket *raw_buckets) {
+void DenseSketch::merge_raw_bucket_buffer(const Bucket *raw_buckets, size_t n_raw_buckets) {
+  if (n_raw_buckets != num_buckets) {
+    throw std::invalid_argument("Raw bucket buffer is not the same size as DenseSketch");
+  }
+
   for (size_t i = 0; i < num_buckets; i++) {
     buckets[i].alpha ^= raw_buckets[i].alpha;
     buckets[i].gamma ^= raw_buckets[i].gamma;