merge sparse and min-cut

Author: etwest

CMakeLists.txt

@@ -91,13 +91,16 @@ endif()
 FetchContent_MakeAvailable(GutterTree StreamingUtilities VieCut tlx)
 
 # AVAILABLE COMPILATION DEFINITIONS:
-# VERIFY_SAMPLES_F   Use a deterministic connected-components
-#                    algorithm to verify post-processing.
 # NO_EAGER_DSU       Do not use the eager DSU query optimization
 #                    if this flag is present.
 # L0_SAMPLING        Run the CubeSketch l0 sampling algorithm
 #                    to ensure that we sample uniformly.
 #                    Otherwise, run a support finding algorithm.
+# L0_FULLY_DENSE     Fully allocate the sketch matrix at the beginning
+#                    of the program. If this flag is not used, sketches
+#                    are allocated dynamically.
+# VERIFY_SAMPLES_F   Use a deterministic connected-components
+#                    algorithm to verify post-processing.
 #
 # Example:
 # cmake -DCMAKE_CXX_FLAGS="-DL0_SAMPLING" ..
@@ -109,7 +112,8 @@ add_library(GraphZeppelin
   src/return_types.cpp
   src/driver_configuration.cpp
   src/cc_alg_configuration.cpp
-  src/sketch.cpp
+  src/sparse_sketch.cpp
+  src/dense_sketch.cpp
   src/util.cpp)
 add_dependencies(GraphZeppelin GutterTree StreamingUtilities VieCut tlx)
 target_link_libraries(GraphZeppelin PUBLIC xxhash GutterTree StreamingUtilities VieCut tlx)
@@ -125,7 +129,8 @@ add_library(GraphZeppelinVerifyCC
   src/return_types.cpp
   src/driver_configuration.cpp
   src/cc_alg_configuration.cpp
-  src/sketch.cpp
+  src/sparse_sketch.cpp
+  src/dense_sketch.cpp
   src/util.cpp
   test/util/graph_verifier.cpp)
 add_dependencies(GraphZeppelinVerifyCC GutterTree StreamingUtilities VieCut tlx)

include/bucket.h

@@ -1,73 +1,94 @@
 #pragma once
-#include <vector>
 #include <xxhash.h>
+
 #include <iostream>
+#include <vector>
+
 #include "types.h"
 
-#pragma pack(push,1)
+#pragma pack(push, 1)
 struct Bucket {
   vec_t alpha;
   vec_hash_t gamma;
 };
 #pragma pack(pop)
 
-namespace Bucket_Boruvka {
-  static constexpr size_t col_hash_bits = sizeof(col_hash_t) * 8;
-  /**
-   * Hashes the column index and the update index together to determine the depth of an update
-   * This is used as a parameter to Bucket::contains.
-   * @param update_idx    Vector index to update
-   * @param seed_and_col  Combination of seed and column
-   * @param max_depth  The maximum depth to return
-   * @return              The hash of update_idx using seed_and_col as a seed.
-   */
-  inline static col_hash_t get_index_depth(const vec_t update_idx, const long seed_and_col,
-   const vec_hash_t max_depth);
+namespace SketchBucket {
 
-  /**
-   * Hashes the index for checksumming
-   * This is used to as a parameter to Bucket::update
-   * @param index      Vector index to update
-   * @param seed       The seed of the Sketch this Bucket belongs to
-   * @return           The depth of the bucket to update
-   */
-  inline static vec_hash_t get_index_hash(const vec_t index, const long sketch_seed);
-
-  /**
-   * Checks whether a Bucket is good, assuming the Bucket contains all elements.
-   * @param bucket       The bucket to check
-   * @param sketch_seed  The seed of the Sketch this Bucket belongs to.
-   * @return             true if this Bucket is good, else false.
-   */
-  inline static bool is_good(const Bucket &bucket, const long sketch_seed);
-
-  /**
-   * Updates a Bucket with the given update index
-   * @param bucket      The bucket to update
-   * @param update_idx  The update index
-   * @param update_hash The hash of the update index, generated with Bucket::index_hash.
-   */
-  inline static void update(Bucket& bucket, const vec_t update_idx,
-                            const vec_hash_t update_hash);
-} // namespace Bucket_Boruvka
+struct Depths {
+ private:
+  col_hash_t depths[2];
+ public:
+  col_hash_t& operator[](size_t i) { return depths[i]; }
+};
 
-inline col_hash_t Bucket_Boruvka::get_index_depth(const vec_t update_idx, const long seed_and_col,
-                                                  const vec_hash_t max_depth) {
+static constexpr size_t col_hash_bits = sizeof(col_hash_t) * 8;
+/**
+ * Hashes the column index and the update index together to determine the depth of an update
+ * This is used as a parameter to Bucket::contains.
+ * @param update_idx    Vector index to update
+ * @param seed_and_col  Combination of seed and column
+ * @param max_depth  The maximum depth to return
+ * @return              The hash of update_idx using seed_and_col as a seed.
+ */
+inline static col_hash_t get_index_depth(const vec_t update_idx, const long seed_and_col,
+                                         const vec_hash_t max_depth) {
   col_hash_t depth_hash = col_hash(&update_idx, sizeof(vec_t), seed_and_col);
-  depth_hash |= (1ull << max_depth); // assert not > max_depth by ORing
+  depth_hash |= (1ull << max_depth);  // assert not > max_depth by ORing
   return __builtin_ctzll(depth_hash);
 }
 
-inline vec_hash_t Bucket_Boruvka::get_index_hash(const vec_t update_idx, const long sketch_seed) {
-  return vec_hash(&update_idx, sizeof(vec_t), sketch_seed);
+inline static Depths get_index_depths(vec_t update_idx, size_t seed, col_hash_t max_depth) {
+  uint64_t depth_hash = col_hash(&update_idx, sizeof(vec_t), seed);
+  Depths ret;
+
+  depth_hash |= (1ull << max_depth);  // assert not > max_depth by ORing
+  ret[0] = __builtin_ctzll(depth_hash);
+
+  // shift hash over and reassert max_depth
+  depth_hash >>= 32;
+  depth_hash |= (1ull << max_depth);
+  ret[1] = __builtin_ctzll(depth_hash);
+
+  return ret;
 }
 
-inline bool Bucket_Boruvka::is_good(const Bucket &bucket, const long sketch_seed) {
+/**
+ * Hashes the index for checksumming
+ * This is used to as a parameter to Bucket::update
+ * @param index      Vector index to update
+ * @param seed       The seed of the Sketch this Bucket belongs to
+ * @return           The depth of the bucket to update
+ */
+inline static vec_hash_t get_index_hash(const vec_t index, const long sketch_seed) {
+  return vec_hash(&index, sizeof(vec_t), sketch_seed);
+}
+
+/**
+ * Checks whether a Bucket is good.
+ * @param bucket       The bucket to check
+ * @param sketch_seed  The seed of the Sketch this Bucket belongs to.
+ * @return             true if this Bucket is good, else false.
+ */
+inline static bool is_good(const Bucket &bucket, const long sketch_seed) {
   return bucket.gamma == get_index_hash(bucket.alpha, sketch_seed);
 }
 
-inline void Bucket_Boruvka::update(Bucket& bucket, const vec_t update_idx,
-                                   const vec_hash_t update_hash) {
+/**
+ * Checks whether a Bucket is empty.
+ * @return  true if this Bucket is empty (alpha and gamma == 0), else false.
+ */
+inline static bool is_empty(const Bucket &bucket) { return bucket.alpha == 0 && bucket.gamma == 0; }
+
+/**
+ * Updates a Bucket with the given update index
+ * @param bucket      The bucket to update
+ * @param update_idx  The update index
+ * @param update_hash The hash of the update index, generated with Bucket::index_hash.
+ */
+inline static void update(Bucket &bucket, const vec_t update_idx, const vec_hash_t update_hash) {
   bucket.alpha ^= update_idx;
   bucket.gamma ^= update_hash;
 }
+
+}  // namespace SketchBucket

include/cc_sketch_alg.h

@@ -209,10 +209,11 @@ class CCSketchAlg {
   /**
    * Apply a batch of updates that have already been processed into a sketch delta.
    * 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 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

@@ -0,0 +1,188 @@
+#pragma once
+#include <graph_zeppelin_common.h>
+#include <gtest/gtest_prod.h>
+#include <sys/mman.h>
+
+#include <fstream>
+#include <unordered_set>
+#include <cmath>
+#include <cassert>
+#include <mutex>
+
+#include "util.h"
+#include "bucket.h"
+#include "sketch_types.h"
+
+/**
+ * Sketch for graph processing, either CubeSketch or CameoSketch.
+ * Sub-linear representation of a vector.
+ */
+class DenseSketch {
+ private:
+  const uint64_t seed;     // seed for hash functions
+  size_t num_samples;      // number of samples we can perform
+  size_t cols_per_sample;  // number of columns to use on each sample
+  size_t num_columns;      // Total number of columns. (product of above 2)
+  size_t bkt_per_col;      // maximum number of buckets per column (max number of rows)
+  size_t num_buckets;      // number of total buckets product of above two
+  size_t sample_idx = 0;   // number of samples performed so far
+
+  // Allocated buckets
+  Bucket* buckets;
+
+  inline Bucket& deterministic_bucket() {
+    return buckets[0];
+  }
+  inline const Bucket& deterministic_bucket() const {
+    return buckets[0];
+  }
+
+  // return the bucket at a particular index in bucket array
+  inline Bucket& bucket(size_t col, size_t row) {
+    return buckets[col * bkt_per_col + row + 1];
+  }
+  inline const Bucket& bucket(size_t col, size_t row) const {
+    return buckets[col * bkt_per_col + row + 1];
+  }
+
+ public:
+  /**
+   * The below constructors use vector length as their input. However, in graph sketching our input
+   * is the number of vertices. This function converts from number of graph vertices to vector
+   * length.
+   * @param num_vertices  Number of graph vertices
+   * @return              The length of the vector to sketch
+   */
+  static vec_t calc_vector_length(node_id_t num_vertices) {
+    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 std::max(size_t(18), (size_t) ceil(f * log2(num_vertices) / num_samples_div));
+  }
+
+  /**
+   * Construct a sketch object
+   * @param vector_len       Length of the vector we are sketching
+   * @param seed             Random seed of the 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, size_t num_samples = 1,
+              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_buckets,
+              size_t num_samples = 1, size_t cols_per_sample = default_cols_per_sample);
+
+  /**
+   * Sketch copy constructor
+   * @param s  The sketch to copy.
+   */
+  DenseSketch(const DenseSketch& s);
+
+  ~DenseSketch();
+
+  /**
+   * Update a sketch based on information about one of its indices.
+   * @param update   the point update.
+   */
+  void update(const vec_t update);
+
+  /**
+   * Function to sample from the sketch.
+   * cols_per_sample determines the number of columns we allocate to this query
+   * @return   A pair with the result index and a code indicating the type of result.
+   */
+  SketchSample sample();
+
+  /**
+   * Function to sample from the appropriate columns to return 1 or more non-zero indices
+   * @return   A pair with the result indices and a code indicating the type of result.
+   */
+  ExhaustiveSketchSample exhaustive_sample();
+
+  std::mutex mutex; // lock the sketch for applying updates in multithreaded processing
+
+  /**
+   * In-place merge function.
+   * @param other  Sketch to merge into caller
+   */
+  void merge(const DenseSketch &other);
+
+  /**
+   * In-place range merge function. Updates the caller Sketch.
+   * The range merge only merges some of the Sketches
+   * This function should only be used if you know what you're doing
+   * @param other         Sketch to merge into caller
+   * @param start_sample  Index of first sample to merge
+   * @param n_samples     Number of samples to merge
+   */
+  void range_merge(const DenseSketch &other, size_t start_sample, size_t n_samples);
+
+  /**
+   * 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 n_raw_buckets  Size of raw_buckets in number of Bucket data-structures
+   */
+  void merge_raw_bucket_buffer(const Bucket *raw_buckets, size_t n_raw_buckets);
+
+  /**
+   * Zero out all the buckets of a sketch.
+   */
+  void zero_contents();
+
+  friend bool operator==(const DenseSketch& sketch1, const DenseSketch& sketch2);
+  friend std::ostream& operator<<(std::ostream& os, const DenseSketch& sketch);
+
+  /**
+   * Serialize the sketch to a binary output stream.
+   * @param binary_out   the stream to write to.
+   */
+  void serialize(std::ostream& binary_out) const;
+
+  inline void reset_sample_state() {
+    sample_idx = 0;
+  }
+
+  // return the size of the sketching datastructure in bytes (just the buckets, not the metadata)
+  inline size_t bucket_array_bytes() const { 
+    return num_buckets * sizeof(Bucket); 
+  }
+
+  // return the size of a sketch given vector size n and number of samples s
+  static size_t estimate_bytes(size_t n, size_t s) { 
+    return (1 + calc_bkt_per_col(n) * s * default_cols_per_sample) * sizeof(Bucket);
+  }
+
+  inline const Bucket* get_readonly_bucket_ptr() const { return (const Bucket*) buckets; }
+  inline uint64_t get_seed() const { return seed; }
+  inline size_t column_seed(size_t column_idx) const { return seed + column_idx * 5; }
+  inline size_t checksum_seed() const { return seed; }
+  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; }
+
+  static size_t calc_bkt_per_col(size_t n) { return ceil(log2(n)) + 1; }
+
+  static constexpr size_t default_cols_per_sample = 1;
+  static constexpr double num_samples_div = 1 - log2(2 - 0.8);
+};