diff --git a/CMakeLists.txt b/CMakeLists.txt index 95e90896..b70d3a3d 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -3,7 +3,7 @@ project(GraphZeppelin) include (FetchContent) -set(CMAKE_CXX_STANDARD 14) +set(CMAKE_CXX_STANDARD 17) set(CMAKE_CXX_STANDARD_REQUIRED ON) set(CMAKE_CXX_EXTENSIONS ON) set(CMAKE_ARCHIVE_OUTPUT_DIRECTORY ${PROJECT_BINARY_DIR}/lib) @@ -21,7 +21,7 @@ message(STATUS "GraphZeppelin Build Type: ${CMAKE_BUILD_TYPE}") if("${CMAKE_CXX_COMPILER_ID}" STREQUAL "GNU") message(STATUS "Adding GNU compiler flags") set(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -W -Wall") -elseif(STATUS "${CMAKE_CXX_COMPILER_ID}" STREQUAL "MSVC") +elseif("${CMAKE_CXX_COMPILER_ID}" STREQUAL "MSVC") message("Adding MSVC compiler flags") set(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} /Wall") else() @@ -59,6 +59,22 @@ FetchContent_Declare( GIT_TAG main ) +# Get VieCut +FetchContent_Declare( + VieCut + + GIT_REPOSITORY https://github.com/etwest/VieCut.git + GIT_TAG master +) + +# Get tlx +FetchContent_Declare( + tlx + + GIT_REPOSITORY https://github.com/tlx/tlx.git + GIT_TAG master +) + if (BUILD_BENCH) # Get Google Benchmark FetchContent_Declare( @@ -72,7 +88,7 @@ if (BUILD_BENCH) FetchContent_MakeAvailable(benchmark) endif() -FetchContent_MakeAvailable(GutterTree StreamingUtilities) +FetchContent_MakeAvailable(GutterTree StreamingUtilities VieCut tlx) # AVAILABLE COMPILATION DEFINITIONS: # VERIFY_SAMPLES_F Use a deterministic connected-components @@ -88,13 +104,15 @@ FetchContent_MakeAvailable(GutterTree StreamingUtilities) add_library(GraphZeppelin src/cc_sketch_alg.cpp + src/edge_store.cpp + src/min_cut_sketch_alg.cpp src/return_types.cpp src/driver_configuration.cpp src/cc_alg_configuration.cpp src/sketch.cpp src/util.cpp) -add_dependencies(GraphZeppelin GutterTree StreamingUtilities) -target_link_libraries(GraphZeppelin PUBLIC xxhash GutterTree StreamingUtilities) +add_dependencies(GraphZeppelin GutterTree StreamingUtilities VieCut tlx) +target_link_libraries(GraphZeppelin PUBLIC xxhash GutterTree StreamingUtilities VieCut tlx) target_include_directories(GraphZeppelin PUBLIC include/) target_compile_options(GraphZeppelin PUBLIC -fopenmp) target_link_options(GraphZeppelin PUBLIC -fopenmp) @@ -102,14 +120,16 @@ target_compile_definitions(GraphZeppelin PUBLIC XXH_INLINE_ALL) add_library(GraphZeppelinVerifyCC src/cc_sketch_alg.cpp + src/edge_store.cpp + src/min_cut_sketch_alg.cpp src/return_types.cpp src/driver_configuration.cpp src/cc_alg_configuration.cpp src/sketch.cpp src/util.cpp test/util/graph_verifier.cpp) -add_dependencies(GraphZeppelinVerifyCC GutterTree StreamingUtilities) -target_link_libraries(GraphZeppelinVerifyCC PUBLIC xxhash GutterTree StreamingUtilities) +add_dependencies(GraphZeppelinVerifyCC GutterTree StreamingUtilities VieCut tlx) +target_link_libraries(GraphZeppelinVerifyCC PUBLIC xxhash GutterTree StreamingUtilities VieCut tlx) target_include_directories(GraphZeppelinVerifyCC PUBLIC include/ include/test/) target_compile_options(GraphZeppelinVerifyCC PUBLIC -fopenmp) target_link_options(GraphZeppelinVerifyCC PUBLIC -fopenmp) @@ -119,6 +139,8 @@ if (BUILD_EXE) add_executable(tests test/test_runner.cpp test/cc_alg_test.cpp + test/min_cut_test.cpp + test/edge_store_test.cpp test/sketch_test.cpp test/dsu_test.cpp test/util_test.cpp @@ -137,6 +159,11 @@ if (BUILD_EXE) tools/process_stream.cpp) target_link_libraries(process_stream PRIVATE GraphZeppelin) + # executable for processing a binary graph stream + add_executable(min_cut_exe + tools/minimum_cut.cpp) + target_link_libraries(min_cut_exe PRIVATE GraphZeppelin) + # executable for performing in depth correctness testing add_executable(test_correctness tools/test_correctness.cpp) diff --git a/include/cc_sketch_alg.h b/include/cc_sketch_alg.h index 9e9d3f8c..884638c6 100644 --- a/include/cc_sketch_alg.h +++ b/include/cc_sketch_alg.h @@ -60,6 +60,7 @@ struct alignas(64) GlobalMergeData { enum QueryCode { CONNECTIVITY, // connected components and spanning forest of graph KSPANNINGFORESTS, // k disjoint spanning forests + MINIMUMCUT, // minimum cut query }; /** @@ -71,12 +72,13 @@ class CCSketchAlg { node_id_t num_vertices; size_t seed; bool update_locked = false; - // a set containing one "representative" from each supernode - std::set *representatives; Sketch **sketches; // DSU representation of supernode relationship DisjointSetUnion_MT dsu; + // TODO: Remove this variable later + std::atomic num_updates; + // if dsu valid then we have a cached query answer. Additionally, we need to update the DSU in // pre_insert() bool dsu_valid = true; @@ -84,6 +86,7 @@ class CCSketchAlg { // for accessing if the DSU is valid from threads that do not perform updates std::atomic shared_dsu_valid; + // Most recent spanning forest computed by algorithm and associated locks std::unordered_set *spanning_forest; std::mutex *spanning_forest_mtx; @@ -105,8 +108,8 @@ class CCSketchAlg { /** * Sample a single supernode represented by a single sketch containing one or more vertices. * Updates the dsu and spanning forest with query results if edge contains new connectivity info. - * @param skt sketch to sample - * @return [bool] true if the query result indicates we should run an additional round. + * param: skt sketch to sample + * return: [bool] true if the query result indicates we should run an additional round. */ bool sample_supernode(Sketch &skt); @@ -116,18 +119,25 @@ class CCSketchAlg { void create_merge_instructions(std::vector &merge_instr); /** - * @param reps set containing the roots of each supernode - * @param merge_instr a list of lists of supernodes to be merged + * param: reps set containing the roots of each supernode + * param: merge_instr a list of lists of supernodes to be merged */ bool perform_boruvka_round(const size_t cur_round, const std::vector &merge_instr, std::vector &global_merges); /** - * Main parallel algorithm utilizing Boruvka and L_0 sampling. - * Ensures that the DSU represents the Connected Components of the stream when called + * Main parallel algorithm subroutine for query computation */ void boruvka_emulation(); + /** + * Ensures that the DSU represents the Connected Components of the stream when called + * and that spanning_forest is correct. + */ + void compute_dsu(); + + void filter_sf_edges(SpanningForest &sf); + // constructor for use when reading from a serialized file CCSketchAlg(node_id_t num_vertices, size_t seed, std::ifstream &binary_stream, CCAlgConfiguration config); @@ -170,9 +180,9 @@ class CCSketchAlg { /** * Update all the sketches for a node, given a batch of updates. - * @param thr_id The id of the thread performing the update [0, num_threads) - * @param src_vertex The vertex where the edges originate. - * @param dst_vertices A vector of destinations. + * param: thr_id The id of the thread performing the update [0, num_threads) + * param: src_vertex The vertex where the edges originate. + * param: dst_vertices A vector of destinations. */ void apply_update_batch(int thr_id, node_id_t src_vertex, const std::vector &dst_vertices); @@ -199,8 +209,8 @@ 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 */ void apply_raw_buckets_update(node_id_t src_vertex, Bucket *raw_buckets); @@ -214,15 +224,15 @@ class CCSketchAlg { /** * Main parallel query algorithm utilizing Boruvka and L_0 sampling. - * @return the connected components in the graph. + * return: the connected components in the graph. */ ConnectedComponents connected_components(); /** * Point query algorithm utilizing Boruvka and L_0 sampling. * Allows for additional updates when done. - * @param a, b - * @return true if a and b are in the same connected component, false otherwise. + * param: a, b vertices of the graph. Check if these are connected. + * return: true if a and b are in the same connected component, false otherwise. */ bool point_query(node_id_t a, node_id_t b); @@ -230,10 +240,18 @@ class CCSketchAlg { * Return a spanning forest of the graph utilizing Boruvka and L_0 sampling * IMPORTANT: The updates to this algorithm MUST NOT be a function of the output of this query * that is, unless you really know what you're doing. - * @return the spanning forest of the graph + * return: the spanning forest of the graph */ SpanningForest calc_spanning_forest(); + /** + * Return k edge-disjoint spanning forests of the graph. + * IMPORTANT: The updates to this algorithm MUST NOT be a function of the output of this query + * that is, unless you really know what you're doing. + * return: k edge-disjoint spanning forests of the graph. + */ + std::vector calc_disjoint_spanning_forests(size_t k); + #ifdef VERIFY_SAMPLES_F void set_verifier(std::unique_ptr verifier) { this->verifier = std::move(verifier); @@ -242,7 +260,7 @@ class CCSketchAlg { /** * Serialize the graph data to a binary file. - * @param filename the name of the file to (over)write data to. + * param: filename the name of the file to (over)write data to. */ void write_binary(const std::string &filename); @@ -255,4 +273,10 @@ class CCSketchAlg { inline node_id_t get_num_vertices() { return num_vertices; } inline size_t get_seed() { return seed; } inline size_t max_rounds() { return sketches[0]->get_num_samples(); } + inline size_t get_num_updates() { return num_updates.load(); } + + void invalidate_dsu() { + dsu_valid = false; + shared_dsu_valid = false; + } }; diff --git a/include/edge_store.h b/include/edge_store.h new file mode 100644 index 00000000..54164ddc --- /dev/null +++ b/include/edge_store.h @@ -0,0 +1,81 @@ +#pragma once + +#include +#include +#include +#include +#include + +#include "types.h" + +class EdgeStore { + private: + static constexpr size_t store_edge_bytes = sizeof(SubgraphTaggedUpdate); // Bytes of one edge + static constexpr double contract_factor = 2; // switch to sketch when within this factor of max + + size_t seed; + node_id_t num_vertices; + size_t num_subgraphs; + volatile size_t cur_subgraph = 0; // subgraph depth at which edges enter the edge store + volatile size_t true_min_subgraph = 0; // the minimum subgraph of elements in the store + + std::atomic num_edges; + std::atomic needs_contraction; + + std::vector> adjlist; + + // This is a vector of booleans BUT we don't want to use vector because its not + // multithread friendly + std::vector vertex_contracted; + + size_t max_edges; // Bytes of sketch graph + size_t default_buffer_allocation; // size we allocate each buffer in adjlist to + + // locks that protect the adjacency list + // we have a single lock for each vertex and a lock for handling contraction logic + std::mutex* adj_mutex; + std::mutex contract_lock; + + std::vector vertex_contract(node_id_t src); + void check_if_too_big(); + +#ifdef VERIFY_SAMPLES_F + void verify_contract_complete(); + std::atomic num_inserted; + std::atomic num_duplicate; + std::atomic num_returned; +#endif + public: + /** + * EdgeStore Constructor + * param: seed seed to depth hash function. (MUST be the same as the MinCutAlgorithm's) + * param: num_vertices number of vertices in the graph + * param: sketch_bytes number of bytes per vertex in sketch data-structure + * param: num_subgraphs maximum number of subgraphs + * param: start_subgraph Number of subgraphs we start with sketches for (updates don't go here) + */ + EdgeStore(size_t seed, node_id_t num_vertices, size_t sketch_bytes, size_t num_subgraphs, + size_t start_subgraph = 0); + ~EdgeStore(); + + // functions for adding data to the edge store + // may return a vector of edges that need to be applied to + + // this first function is only called when there exist no sketch subgraphs + TaggedUpdateBatch insert_adj_edges(node_id_t src, const std::vector& dst_vertices); + + // this function is called when there are some sketch subgraphs. + TaggedUpdateBatch insert_adj_edges(node_id_t src, node_id_t caller_first_es_subgraph, + std::vector &dst_data); + + // contract vertex data by removing all updates bound for lower subgraphs than the store + // is responsible for + TaggedUpdateBatch vertex_advance_subgraph(node_id_t cur_first_es_subgraph); + + // Get methods + size_t get_num_edges() { return num_edges; } + size_t get_footprint() { return num_edges * store_edge_bytes; } + size_t get_first_store_subgraph() { return cur_subgraph; } + std::vector get_edges(); + bool contract_in_progress() { return true_min_subgraph < cur_subgraph; } +}; diff --git a/include/graph_sketch_driver.h b/include/graph_sketch_driver.h index e627e443..9b15d657 100644 --- a/include/graph_sketch_driver.h +++ b/include/graph_sketch_driver.h @@ -140,8 +140,8 @@ class GraphSketchDriver { */ void process_stream_until(edge_id_t break_edge_idx) { if (!stream->set_break_point(break_edge_idx)) { - DriverException("Could not correctly set breakpoint: " + std::to_string(break_edge_idx)); - exit(EXIT_FAILURE); + throw DriverException("Could not correctly set breakpoint: " + + std::to_string(break_edge_idx)); } worker_threads->resume_workers(); diff --git a/include/mc_configuration.h b/include/mc_configuration.h new file mode 100644 index 00000000..a599c3d5 --- /dev/null +++ b/include/mc_configuration.h @@ -0,0 +1,61 @@ +#pragma once +#include + +// Configuration options for the minimum cut sketch algorithm +class MCAlgConfiguration { + private: + // How large to make update batches as factor of sketch size + double _batch_factor = 1; + + // Returned min-cut guaranteed to be a +/- epsilon multiplicative approx of the true min cut. + double _epsilon = 0.5; + + // Number of sketch subgraphs to create immediately. + // Preallocating these sketches improves performance at the cost of additional memory consumption + // if you already that you will need more (say 5) then setting this to the higher value is good + size_t _initial_subgraphs = 1; + + friend class MinCutSketchAlg; + public: + // setters + MCAlgConfiguration& batch_factor(double batch_factor) { + if (batch_factor <= 0) { + std::cerr << "WARNING: MCAlgConfiguration, batch factor must be > 0." << std::endl; + std::cerr << " Setting to default value: " << _batch_factor << std::endl; + } else { + _batch_factor = batch_factor; + } + return *this; + } + MCAlgConfiguration& epsilon(double epsilon) { + if (epsilon <= 0 || epsilon > 1) { + std::cerr << "WARNING: MCAlgConfiguration epsilon must be in range (0, 1]." << std::endl; + std::cerr << " Setting to default value: " << _epsilon << std::endl; + } else { + _epsilon = epsilon; + } + return *this; + } + MCAlgConfiguration& initial_subgraphs(size_t num_subgraphs) { + if (num_subgraphs == 0) { + std::cerr << "WARNING: MCAlgConfiguration, initial subgraphs must be > 0." << std::endl; + std::cerr << " Setting to default value: " << _initial_subgraphs << std::endl; + } else { + _initial_subgraphs = num_subgraphs; + } + return *this; + } + + // getters + double get_batch_factor() { return _batch_factor; } + double get_epsilon() { return _epsilon; } + size_t get_initial_subgraphs() { return _initial_subgraphs; } + + friend std::ostream& operator<< (std::ostream &out, const MCAlgConfiguration &conf) { + out << "Minimum Cut Algorithm Configuration:" << std::endl; + out << " batch_factor = " << conf._batch_factor << std::endl; + out << " epsilon = " << conf._epsilon << std::endl; + out << " initial_sketch_subgraphs = " << conf._initial_subgraphs << std::endl; + return out; + } +}; diff --git a/include/min_cut_sketch_alg.h b/include/min_cut_sketch_alg.h new file mode 100644 index 00000000..2e165679 --- /dev/null +++ b/include/min_cut_sketch_alg.h @@ -0,0 +1,130 @@ +#pragma once +#include +#include +#include + +#include "cc_sketch_alg.h" +#include "edge_store.h" +#include "mc_configuration.h" + +// Minimum cut sketch algorithm class +class MinCutSketchAlg { + private: + struct ThreadData { + std::vector> cc_buffers; + std::vector edge_store_buffer; + }; + + const node_id_t num_vertices; + const size_t seed; + const size_t subgraph_seed; + MCAlgConfiguration config; + const size_t max_subgraphs; + const size_t k; + std::atomic cur_subgraphs; + std::mutex advance_subgraph_lock; + + const double sketch_factor; + const size_t sketch_samples; + const size_t buffer_elms; + + CCSketchAlg **cc_sketches; + EdgeStore edge_store; + + Sketch *delta_sketches = nullptr; + ThreadData *thread_data = nullptr; + size_t num_delta_sketches = 0; + size_t num_workers; + +#ifdef VERIFY_SAMPLES_F + std::unique_ptr verifier; + std::unique_ptr adj_verifier; +#endif + + CCAlgConfiguration cc_config; + + void advance_cur_subgraph(size_t new_cur_subgraphs); + + void create_subgraph_verifiers(); + public: + /** + * Construct an instance of the Minimum Cut Sketching Algorithm + * param _num_vertices number of graph vertices + * param _seed seed to hash functions + * param _config Configuration options for minimum cut sketch algorithm + */ + MinCutSketchAlg(node_id_t _num_vertices, size_t _seed, + MCAlgConfiguration _config = MCAlgConfiguration()); + + ~MinCutSketchAlg(); + + /** + * Allocate memory for the worker threads to use when updating this algorithm's sketches + */ + void allocate_worker_memory(size_t num_workers); + + /** + * Returns the number of buffered updates we would like to have in the update batches + */ + size_t get_desired_updates_per_batch() { + return cc_sketches[0]->get_desired_updates_per_batch(); + } + + /** + * Action to take on an update before inserting it to the guttering system. + * We use this function to manage the eager dsu. + */ + void pre_insert(GraphUpdate upd, node_id_t thr_id); + + + /** + * Update all the sketches for a vertex, given a batch of updates. + * param thr_id The id of the thread performing the update [0, num_threads) + * param src_vertex The vertex where the edges originate. + * param dst_vertices A vector of destinations. + */ + void apply_update_batch(size_t thr_id, node_id_t src_vertex, + const std::vector &dst_vertices); + + /** + * Set the verifier this algorithm will use to check its correctness + * param: _verifier the verifier to use, should contain all edges processed at this point + */ +#ifdef VERIFY_SAMPLES_F + void set_verifier(std::unique_ptr _verifier); +#endif + + /** + * Main query routine of this algorithm. + * Returns an approximation of the minimum cut of the graph defined by the graph stream + * seen thus far. This approximation is guaranteed to be within 1 +/- epsilon of the true + * minimum cut. + */ + MinCut calc_minimum_cut(); + + /** + * Return if we have cached an answer to query. + * This allows the driver to avoid flushing the gutters before calling query functions. + * TODO: Is there something intelligent we can do here for mincut/k-conn + */ + bool has_cached_query(int query_type) { + if (query_type != MINIMUMCUT) return cc_sketches[0]->has_cached_query(query_type); + return false; + } + + /** + * Print the configuration of minimum cut graph sketching algorithm. + */ + void print_configuration() { + std::cout << config; + std::cout << "MCAlg using the following CCAlg config:" << std::endl; + std::cout << cc_config << std::endl; + } + + node_id_t get_num_vertices() { return num_vertices; } + + // time hooks for experiments + std::chrono::duration total_mc_duration; + std::chrono::duration sf_total_duration; + std::chrono::duration viecut_duration; +}; diff --git a/include/return_types.h b/include/return_types.h index b329bfe6..9bd166a8 100644 --- a/include/return_types.h +++ b/include/return_types.h @@ -28,10 +28,19 @@ class ConnectedComponents { // This class defines a spanning forest of a graph class SpanningForest { private: - std::vector edges; node_id_t num_vertices; + std::vector edges; + std::vector sorted_adjacency; + bool has_adjacency = false; public: SpanningForest(node_id_t num_vertices, const std::unordered_set *spanning_forest); const std::vector& get_edges() const { return edges; } + const std::vector& get_sorted_adjacency(); +}; + +struct MinCut { + std::set left_vertices; + std::set right_vertices; + size_t value; }; diff --git a/include/sketch.h b/include/sketch.h index 80473ecc..a91cf7fd 100644 --- a/include/sketch.h +++ b/include/sketch.h @@ -39,7 +39,7 @@ struct ExhaustiveSketchSample { */ class Sketch { private: - const uint64_t seed; // seed for hash functions + size_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) @@ -49,15 +49,15 @@ class Sketch { size_t sample_idx = 0; // number of samples performed so far // bucket data - Bucket* buckets; + Bucket* buckets = nullptr; 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 + * 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); @@ -67,59 +67,68 @@ class Sketch { * 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 + * 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)); } + // default constructor doesn't do anything, initialize later with operator=(Sketch &&) + Sketch(){}; + /** * 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) + * 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) */ Sketch(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_samples [Optional] Number of samples this sketch supports (default = 1) - * @param cols_per_sample [Optional] Number of sketch columns for each sample (default = 1) + * 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_samples [Optional] Number of samples this sketch supports (default = 1) + * param cols_per_sample [Optional] Number of sketch columns for each sample (default = 1) */ Sketch(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); /** * Sketch copy constructor - * @param s The sketch to copy. + * param s The sketch to copy. */ Sketch(const Sketch& s); + /** + * Sketch move constructor + * param oth The sketch to move into this one + */ + Sketch& operator=(Sketch &&oth); + ~Sketch(); /** * Update a sketch based on information about one of its indices. - * @param update the point update. + * param update the point update. */ - void update(const vec_t update); + void update(const vec_t update_idx); /** * 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. + * 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. + * return A pair with the result indices and a code indicating the type of result. */ ExhaustiveSketchSample exhaustive_sample(); @@ -127,7 +136,7 @@ class Sketch { /** * In-place merge function. - * @param other Sketch to merge into caller + * param other Sketch to merge into caller */ void merge(const Sketch &other); @@ -135,9 +144,9 @@ class Sketch { * 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 + * 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 Sketch &other, size_t start_sample, size_t n_samples); @@ -145,7 +154,7 @@ class Sketch { * 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 */ void merge_raw_bucket_buffer(const Bucket *raw_buckets); @@ -159,7 +168,7 @@ class Sketch { /** * Serialize the sketch to a binary output stream. - * @param binary_out the stream to write to. + * param binary_out the stream to write to. */ void serialize(std::ostream& binary_out) const; @@ -167,6 +176,11 @@ class Sketch { sample_idx = 0; } + // 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); + } + // 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); } diff --git a/include/types.h b/include/types.h index 6fea6b26..040a232a 100644 --- a/include/types.h +++ b/include/types.h @@ -1,8 +1,10 @@ #pragma once -#include +#include #include +#include + #include -#include +#include typedef uint64_t col_hash_t; static const auto& vec_hash = XXH3_64bits_withSeed; @@ -13,3 +15,26 @@ struct GraphUpdate { Edge edge; UpdateType type; }; + +struct SubgraphTaggedUpdate { + node_id_t subgraph; // highest index subgraph the edge maps to (same src, dst -> same subgraph) + node_id_t dst; // destination vertex of edge + + bool operator<(const SubgraphTaggedUpdate& oth) const { + if (subgraph == oth.subgraph) return dst < oth.dst; + + return subgraph < oth.subgraph; + } + + bool operator>(const SubgraphTaggedUpdate& oth) const { + if (subgraph == oth.subgraph) return dst > oth.dst; + + return subgraph > oth.subgraph; + } +}; + +struct TaggedUpdateBatch { + node_id_t src; + node_id_t edge_store_subgraph; + std::vector dsts_data; +}; diff --git a/src/cc_sketch_alg.cpp b/src/cc_sketch_alg.cpp index a1e688db..5888ee31 100644 --- a/src/cc_sketch_alg.cpp +++ b/src/cc_sketch_alg.cpp @@ -10,14 +10,12 @@ CCSketchAlg::CCSketchAlg(node_id_t num_vertices, size_t seed, CCAlgConfiguration config) : num_vertices(num_vertices), seed(seed), dsu(num_vertices), config(config) { - representatives = new std::set(); 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, 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); } @@ -25,6 +23,8 @@ CCSketchAlg::CCSketchAlg(node_id_t num_vertices, size_t seed, CCAlgConfiguration spanning_forest_mtx = new std::mutex[num_vertices]; dsu_valid = true; shared_dsu_valid = true; + + num_updates = 0; } CCSketchAlg *CCSketchAlg::construct_from_serialized_data(const std::string &input_file, @@ -45,14 +45,12 @@ CCSketchAlg *CCSketchAlg::construct_from_serialized_data(const std::string &inpu CCSketchAlg::CCSketchAlg(node_id_t num_vertices, size_t seed, std::ifstream &binary_stream, CCAlgConfiguration config) : num_vertices(num_vertices), seed(seed), dsu(num_vertices), config(config) { - representatives = new std::set(); 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, 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); } binary_stream.close(); @@ -71,7 +69,6 @@ CCSketchAlg::~CCSketchAlg() { delete[] delta_sketches; } - delete representatives; delete[] spanning_forest; delete[] spanning_forest_mtx; } @@ -109,6 +106,9 @@ void CCSketchAlg::apply_update_batch(int thr_id, node_id_t src_vertex, Sketch &delta_sketch = *delta_sketches[thr_id]; delta_sketch.zero_contents(); + // TODO: Remove this later + num_updates += dst_vertices.size(); + for (const auto &dst : dst_vertices) { delta_sketch.update(static_cast(concat_pairing_fn(src_vertex, dst))); } @@ -136,6 +136,7 @@ void CCSketchAlg::update(GraphUpdate upd) { // that is, 1 or more vertices merged together during Boruvka inline bool CCSketchAlg::sample_supernode(Sketch &skt) { bool modified = false; + SketchSample sample = skt.sample(); Edge e = inv_concat_pairing_fn(sample.idx); @@ -221,6 +222,8 @@ inline bool merge_global(const size_t cur_round, const Sketch &local_sketch, // faster query procedure optimized for when we know there is no merging to do (i.e. round 0) inline bool CCSketchAlg::run_round_zero() { + // std::cout << "Running round zero! " << "num_vertices = " << num_vertices << std::endl; + bool modified = false; bool except = false; std::exception_ptr err; @@ -512,9 +515,7 @@ void CCSketchAlg::boruvka_emulation() { update_locked = false; } -ConnectedComponents CCSketchAlg::connected_components() { - cc_alg_start = std::chrono::steady_clock::now(); - +void CCSketchAlg::compute_dsu() { // if the DSU holds the answer, use that if (shared_dsu_valid) { #ifdef VERIFY_SAMPLES_F @@ -524,30 +525,34 @@ ConnectedComponents CCSketchAlg::connected_components() { } } #endif + return; } // The DSU does not hold the answer, make it so - else { - bool except = false; - std::exception_ptr err; - try { - // auto start = std::chrono::steady_clock::now(); - boruvka_emulation(); - // std::cout << " boruvka's algorithm = " - // << std::chrono::duration(std::chrono::steady_clock::now() - start).count() - // << std::endl; - } catch (...) { - except = true; - err = std::current_exception(); - } - - // get ready for ingesting more from the stream by resetting the sketches sample state - for (node_id_t i = 0; i < num_vertices; i++) { - sketches[i]->reset_sample_state(); - } + bool except = false; + std::exception_ptr err; + try { + // auto start = std::chrono::steady_clock::now(); + boruvka_emulation(); + // std::cout << " boruvka's algorithm = " + // << std::chrono::duration(std::chrono::steady_clock::now() - start).count() + // << std::endl; + } catch (...) { + except = true; + err = std::current_exception(); + } - if (except) std::rethrow_exception(err); + // get ready for ingesting more from the stream by resetting the sketches sample state + for (node_id_t i = 0; i < num_vertices; i++) { + sketches[i]->reset_sample_state(); } + if (except) std::rethrow_exception(err); +} + +ConnectedComponents CCSketchAlg::connected_components() { + cc_alg_start = std::chrono::steady_clock::now(); + + compute_dsu(); ConnectedComponents cc(num_vertices, dsu); #ifdef VERIFY_SAMPLES_F verifier->verify_connected_components(cc); @@ -557,8 +562,7 @@ ConnectedComponents CCSketchAlg::connected_components() { } SpanningForest CCSketchAlg::calc_spanning_forest() { - // TODO: Could probably optimize this a bit by writing new code - connected_components(); + compute_dsu(); SpanningForest ret(num_vertices, spanning_forest); #ifdef VERIFY_SAMPLES_F @@ -567,6 +571,93 @@ SpanningForest CCSketchAlg::calc_spanning_forest() { return ret; } +void CCSketchAlg::filter_sf_edges(SpanningForest &sf) { + // auto start = std::chrono::steady_clock::now(); + + dsu_valid = false; + shared_dsu_valid = false; + + const std::vector &edges = sf.get_sorted_adjacency(); + +#pragma omp parallel + { + size_t thr_id = omp_get_thread_num(); + size_t num_threads = omp_get_num_threads(); + + std::pair partition = get_ith_partition(edges.size(), thr_id, num_threads); + size_t start = partition.first; + size_t end = partition.second; + + if (start != end) { + // check if we collide with previous thread. If so lock and apply those updates. + if (start > 0 && edges[start].src == edges[start - 1].src) { + sketches[edges[start].src]->mutex.lock(); + size_t orig_start = start; + while (start < end && edges[start].src == edges[orig_start].src) { + Edge edge = edges[start]; + sketches[edge.src]->update(static_cast(concat_pairing_fn(edge.src, edge.dst))); + ++start; + } + + sketches[edges[orig_start].src]->mutex.unlock(); + } + + // check if we collide with next thread. If so lock and apply those updates. + if (end > start && end < edges.size() && edges[end - 1].src == edges[end].src) { + sketches[edges[end - 1].src]->mutex.lock(); + size_t orig_end = end; + while (end > start && edges[end - 1].src == edges[orig_end - 1].src) { + Edge edge = edges[end - 1]; + sketches[edge.src]->update(static_cast(concat_pairing_fn(edge.src, edge.dst))); + --end; + } + + sketches[edges[orig_end].src]->mutex.unlock(); + } + + for (size_t i = start; i < end; i++) { + Edge edge = edges[i]; + sketches[edge.src]->update(static_cast(concat_pairing_fn(edge.src, edge.dst))); + } + } + } + + // delete_time += std::chrono::steady_clock::now() - start; +} + +std::vector CCSketchAlg::calc_disjoint_spanning_forests(size_t k) { + std::cout << "Spanning forests query. Number of updates = " << num_updates.load() << std::endl; + + // sf_query_start = std::chrono::steady_clock::now(); + std::vector SFs; + size_t max_rounds = 0; + + for (size_t i = 0; i < k; i++) { + compute_dsu(); + + SFs.emplace_back(num_vertices, spanning_forest); + max_rounds = std::max(last_query_rounds, max_rounds); + + filter_sf_edges(SFs[SFs.size() - 1]); + if (SFs[SFs.size() - 1].get_edges().size() == 0) break; + } + + // revert the state of the sketches to remove all deletions + for (auto &sf : SFs) { + filter_sf_edges(sf); + } + +#ifdef VERIFY_SAMPLES_F + verifier->verify_spanning_forests(SFs); +#endif + + // number of rounds per SF may be non-monotonic, so we track the maximum number of rounds + // among all of the spanning forest queries. + last_query_rounds = max_rounds; + + return SFs; +} + bool CCSketchAlg::point_query(node_id_t a, node_id_t b) { cc_alg_start = std::chrono::steady_clock::now(); diff --git a/src/edge_store.cpp b/src/edge_store.cpp new file mode 100644 index 00000000..3a32e5dc --- /dev/null +++ b/src/edge_store.cpp @@ -0,0 +1,312 @@ +#include "edge_store.h" +#include "bucket.h" +#include "util.h" + +#include + +// Constructor +EdgeStore::EdgeStore(size_t seed, node_id_t num_vertices, size_t sketch_bytes, size_t num_subgraphs, size_t start_subgraph) + : seed(seed), + num_vertices(num_vertices), + num_subgraphs(num_subgraphs), + adjlist(num_vertices), + vertex_contracted(num_vertices, true), + max_edges(num_vertices * sketch_bytes / store_edge_bytes), + default_buffer_allocation(max_edges / num_vertices) { + num_edges = 0; + adj_mutex = new std::mutex[num_vertices]; + + cur_subgraph = start_subgraph; + true_min_subgraph = start_subgraph; + + // reserve some space for each + for (node_id_t i = 0; i < num_vertices; i++) { + adjlist[i].reserve(default_buffer_allocation); + } +#ifdef VERIFY_SAMPLES_F + num_inserted = 0; + num_duplicate = 0; + num_returned = 0; +#endif +} + +EdgeStore::~EdgeStore() { + delete[] adj_mutex; +#ifdef VERIFY_SAMPLES_F + std::cerr << "EdgeStore: Deconstructor" << std::endl; + std::cerr << " num_edges = " << num_edges << std::endl; + std::cerr << " max_edges = " << max_edges << std::endl; + std::cerr << " num_inserted = " << num_inserted << std::endl; + std::cerr << " num_duplicate = " << num_duplicate << std::endl; + std::cerr << " num_returned = " << num_returned << std::endl; +#endif +} + +// caller_first_es_subgraph is implied to be 0 when calling this function +TaggedUpdateBatch EdgeStore::insert_adj_edges(node_id_t src, + const std::vector &dst_vertices) { + + std::vector tagged_updates; + tagged_updates.resize(dst_vertices.size()); + for (node_id_t i = 0; i < dst_vertices.size(); i++) { + node_id_t dst = dst_vertices[i]; + auto idx = concat_pairing_fn(src, dst); + tagged_updates[i] = {Bucket_Boruvka::get_index_depth(idx, seed, num_subgraphs), dst}; + } + return insert_adj_edges(src, 0, tagged_updates); +} + +TaggedUpdateBatch EdgeStore::insert_adj_edges(node_id_t src, node_id_t caller_first_es_subgraph, + std::vector &dst_data) { + std::vector ret; + if (dst_data.size() == 0) return {src, cur_subgraph, ret}; + node_id_t cur_first_es_subgraph; + +#ifdef VERIFY_SAMPLES_F + num_inserted += dst_data.size(); +#endif + + + // Sort the input data + std::sort(dst_data.begin(), dst_data.end()); + + // remove pairs of duplicate updates if there are any + size_t ptr = 0; + size_t dst_ptr = 0; + while (ptr < dst_data.size() - 1) { + if (dst_data[ptr] < dst_data[ptr+1]) { + // not a pair, write to output + dst_data[dst_ptr] = dst_data[ptr]; + ++ptr; + ++dst_ptr; + } else { + // found a pair, skip it + ptr += 2; + } + } + if (ptr < dst_data.size()) { + dst_data[dst_ptr] = dst_data[ptr]; + ++dst_ptr; + } + size_t dst_data_size = dst_ptr; + + // merge the input data into the vertex buffer + { + std::lock_guard lk(adj_mutex[src]); + cur_first_es_subgraph = cur_subgraph; + + if (true_min_subgraph < cur_first_es_subgraph && !vertex_contracted[src]) { + ret = vertex_contract(src); + } + + auto &data_buffer = adjlist[src]; + size_t orig_size = data_buffer.size(); + std::vector new_data_buffer(orig_size + dst_data_size); + + size_t out_ptr = 0; + size_t update_ptr = 0; + size_t buffer_ptr = 0; + + // skip any updates that go to a smaller subgraph + while (update_ptr < dst_data_size && dst_data[update_ptr].subgraph < cur_first_es_subgraph) { + ret.push_back(dst_data[update_ptr]); + ++update_ptr; + } + +#ifdef VERIFY_SAMPLES_F + num_returned += update_ptr; + size_t local_ignored = update_ptr; +#endif + + // merge new updates in, until one of the arrays runs out + while (buffer_ptr < orig_size && update_ptr < dst_data_size) { + if (data_buffer[buffer_ptr] > dst_data[update_ptr]) { + // place update_ptr data into output + new_data_buffer[out_ptr++] = dst_data[update_ptr++]; + } else if (data_buffer[buffer_ptr] < dst_data[update_ptr]) { + // place contents of compare_ptr into out_ptr + new_data_buffer[out_ptr++] = data_buffer[buffer_ptr++]; + } else { + // they are equal! Skip both + ++buffer_ptr; + ++update_ptr; + +#ifdef VERIFY_SAMPLES_F + num_duplicate += 2; + local_ignored += 2; +#endif + } + } + + // place all remaining updates into the buffer + while (buffer_ptr < orig_size) { + new_data_buffer[out_ptr++] = data_buffer[buffer_ptr++]; + } + while (update_ptr < dst_data_size) { + new_data_buffer[out_ptr++] = dst_data[update_ptr++]; + } + new_data_buffer.resize(out_ptr); + std::swap(data_buffer, new_data_buffer); + +#ifdef VERIFY_SAMPLES_F + // verify sorted order + SubgraphTaggedUpdate prev = data_buffer[0]; + for (size_t i = 1; i < data_buffer.size(); i++) { + SubgraphTaggedUpdate data = data_buffer[i]; + if (data < prev || !(data < prev || prev < data)) { + std::cerr << "ERROR: Buffer not sort good!" << std::endl; + std::cerr << "cur " << i << " = {" << data.subgraph << "," << data.dst << "} should be >= "; + std::cerr << "prev = {" << prev.subgraph << "," << prev.dst << "}" << std::endl; + exit(EXIT_FAILURE); + } + } + + if (data_buffer.size() != orig_size + dst_data_size - local_ignored) { + std::cerr << "ERROR: Number of updates incorrect!" << std::endl; + std::cerr << "Expected: " << orig_size + dst_data_size - local_ignored << std::endl; + std::cerr << "Got: " << out_ptr << std::endl; + + std::cerr << "orig_size = " << orig_size << std::endl; + std::cerr << "dst_data_size = " << dst_data_size << std::endl; + std::cerr << "local_ignored = " << local_ignored << std::endl; + exit(EXIT_FAILURE); + } +#endif + + num_edges += data_buffer.size() - orig_size; + } + + if (ret.size() == 0 && true_min_subgraph < cur_first_es_subgraph) { + return vertex_advance_subgraph(cur_first_es_subgraph); + } else { + check_if_too_big(); + return {src, cur_first_es_subgraph, ret}; + } +} + +// IMPORTANT: We must have completed any pending contractions before we call this function +std::vector EdgeStore::get_edges() { + std::vector ret; + ret.reserve(num_edges); + + for (node_id_t src = 0; src < num_vertices; src++) { + for (auto data : adjlist[src]) { + ret.push_back({src, data.dst}); + } + } + + return ret; +} + +#ifdef VERIFY_SAMPLES_F +void EdgeStore::verify_contract_complete() { + for (size_t i = 0; i < num_vertices; i++) { + std::lock_guard lk(adj_mutex[i]); + if (adjlist[i].size() == 0) continue; + + auto it = adjlist[i].begin(); + if (it->subgraph < cur_subgraph) { + std::cerr << "ERROR: Found " << it->subgraph << ", " << it->dst << " which should have been deleted by contraction to " << cur_subgraph << std::endl; + exit(EXIT_FAILURE); + } + } + std::cerr << "Contraction verified!" << std::endl; +} +#endif + +// the thread MUST hold the lock on src before calling this function +std::vector EdgeStore::vertex_contract(node_id_t src) { + std::vector ret; + // someone already contacted this vertex + if (vertex_contracted[src]) + return ret; + + vertex_contracted[src] = true; + auto &data_buffer = adjlist[src]; + size_t orig_size = data_buffer.size(); + + if (data_buffer.size() == 0) { + return ret; + } + + size_t less_than = 0; + while (less_than < data_buffer.size() && data_buffer[less_than].subgraph < cur_subgraph) { + ++less_than; + } + + ret.insert(ret.end(), data_buffer.begin(), data_buffer.begin() + less_than); + + size_t keep_idx = 0; + for (size_t i = less_than; i < data_buffer.size(); i++) { + data_buffer[keep_idx++] = data_buffer[i]; + } + + data_buffer.resize(keep_idx); + + num_edges += data_buffer.size() - orig_size; +#ifdef VERIFY_SAMPLES_F + num_returned += ret.size(); +#endif + return ret; +} + +TaggedUpdateBatch EdgeStore::vertex_advance_subgraph(node_id_t cur_first_es_subgraph) { + node_id_t src = 0; + while (true) { + src = needs_contraction.fetch_add(1); + + if (src >= num_vertices) { + if (src == num_vertices) { + std::lock_guard lk(contract_lock); +#ifdef VERIFY_SAMPLES_F + verify_contract_complete(); +#endif + ++true_min_subgraph; + std::cerr << "EdgeStore: Contraction complete" << std::endl; + } + return {0, cur_first_es_subgraph, std::vector()}; + } + + std::lock_guard lk(adj_mutex[src]); + if (adjlist[src].size() > 0 && !vertex_contracted[src]) + break; + + vertex_contracted[src] = true; + } + + std::lock_guard lk(adj_mutex[src]); + return {src, cur_first_es_subgraph, vertex_contract(src)}; +} + +// checks if we should perform a contraction and begins the process if so +void EdgeStore::check_if_too_big() { + if (num_edges < max_edges) { + // no contraction needed + return; + } + + // we may need to perform a contraction + { + std::lock_guard lk(contract_lock); + if (true_min_subgraph < cur_subgraph) { + // another thread already started contraction + return; + } + + for (node_id_t i = 0; i < num_vertices; i++) { + vertex_contracted[i] = false; + } + needs_contraction = 0; + + cur_subgraph++; + } + +#ifdef VERIFY_SAMPLES_F + std::cerr << "EdgeStore: Contracting to subgraphs " << cur_subgraph << " and above" << std::endl; + std::cerr << " num_edges = " << num_edges << std::endl; + std::cerr << " max_edges = " << max_edges << std::endl; + std::cerr << " num_inserted = " << num_inserted << std::endl; + std::cerr << " num_duplicate = " << num_duplicate << std::endl; + std::cerr << " num_returned = " << num_returned << std::endl; +#endif +} diff --git a/src/min_cut_sketch_alg.cpp b/src/min_cut_sketch_alg.cpp new file mode 100644 index 00000000..30beff71 --- /dev/null +++ b/src/min_cut_sketch_alg.cpp @@ -0,0 +1,299 @@ +#include "min_cut_sketch_alg.h" + +#include +#include +#include +#include +#include + +MinCutSketchAlg::MinCutSketchAlg(node_id_t _num_vertices, size_t _seed, MCAlgConfiguration _config) + : num_vertices(_num_vertices), + seed(_seed), + subgraph_seed(col_hash(&seed, sizeof(seed), seed)), + config(_config), + max_subgraphs(2 * log2(num_vertices)), + k(log2(num_vertices) / (config._epsilon * config._epsilon)), + cur_subgraphs(config._initial_subgraphs), + sketch_factor(1.3 / (config._epsilon * config._epsilon)), + sketch_samples(Sketch::calc_cc_samples(num_vertices, sketch_factor)), + buffer_elms(Sketch::estimate_bytes(Sketch::calc_vector_length(num_vertices), sketch_samples) / + sizeof(node_id_t)), + cc_sketches(new CCSketchAlg *[max_subgraphs]), + edge_store(seed, num_vertices, buffer_elms * sizeof(node_id_t), max_subgraphs, 1) { + if (cur_subgraphs > max_subgraphs) { + std::cerr << "WARNING: MinCutSketchAlg, initial_subgraphs > max_subgraphs. Setting to max." + << std::endl; + cur_subgraphs = max_subgraphs; + } + + cc_config.sketches_factor(sketch_factor); + + for (size_t i = 0; i < cur_subgraphs; i++) { + cc_sketches[i] = new CCSketchAlg(num_vertices, seed, cc_config); + if (i > 0) cc_sketches[i]->invalidate_dsu(); + } +} + +MinCutSketchAlg::~MinCutSketchAlg() { + for (size_t i = 0; i < cur_subgraphs; i++) { + delete cc_sketches[i]; + } + delete[] cc_sketches; + + if (delta_sketches != nullptr) delete[] delta_sketches; + if (thread_data != nullptr) delete[] thread_data; +} + +void MinCutSketchAlg::allocate_worker_memory(size_t _num_workers) { + num_workers = _num_workers; + for (size_t i = 0; i < cur_subgraphs; i++) { + cc_sketches[i]->allocate_worker_memory(num_workers); + } + + thread_data = new ThreadData[num_workers]; + for (size_t t = 0; t < num_workers; t++) { + thread_data[t].cc_buffers.resize(max_subgraphs); + for (size_t b = 1; b < cur_subgraphs; b++) { + thread_data[t].cc_buffers[b].resize(buffer_elms); + } + } +} + +// must hold advance_subgraph_lock when calling this function +void MinCutSketchAlg::advance_cur_subgraph(size_t new_cur_subgraphs) { + for (size_t i = cur_subgraphs; i < new_cur_subgraphs; i++) { + cc_sketches[i] = new CCSketchAlg(num_vertices, seed, cc_config); // TODO: DO WE NEED A DIFFERENT SEED? + cc_sketches[i]->allocate_worker_memory(num_workers); + cc_sketches[i]->invalidate_dsu(); + } + + for (size_t t = 0; t < num_workers; t++) { + for (size_t b = cur_subgraphs; b < new_cur_subgraphs; b++) { + thread_data[t].cc_buffers[b].resize(buffer_elms); + } + } + + cur_subgraphs = new_cur_subgraphs; +} + +void MinCutSketchAlg::pre_insert(GraphUpdate upd, node_id_t thr_id) { + // we just pre-insert to the first subgraph + // TODO: unless there's something more intelligent to do here at some point? + cc_sketches[0]->pre_insert(upd, thr_id); +} + +void MinCutSketchAlg::apply_update_batch(size_t thr_id, node_id_t src_vertex, + const std::vector &dst_vertices) { + assert(dst_vertices.size() <= buffer_elms); + + // everything goes in subgraph 0 + cc_sketches[0]->apply_update_batch(thr_id, src_vertex, dst_vertices); + + size_t num_mapped[max_subgraphs]; + std::fill(&num_mapped[0], &num_mapped[max_subgraphs - 1], 0); + + std::vector> &buffers = thread_data[thr_id].cc_buffers; + std::vector &edge_buf = thread_data[thr_id].edge_store_buffer; + edge_buf.resize(buffer_elms); + + size_t our_cur_subgraphs = cur_subgraphs; + + // map the updates to one of the subgraphs + for (size_t i = 0; i < dst_vertices.size(); i++) { + vec_t idx = concat_pairing_fn(src_vertex, dst_vertices[i]); + node_id_t subgraph_idx = Bucket_Boruvka::get_index_depth(idx, subgraph_seed, max_subgraphs - 1) + 1; + + if (subgraph_idx < our_cur_subgraphs) { + // goes in a sketch! + assert(num_mapped[subgraph_idx] < buffer_elms); + assert(buffers[subgraph_idx].size() == buffer_elms); + buffers[subgraph_idx][num_mapped[subgraph_idx]++] = dst_vertices[i]; + } else { + // goes in edge store! + assert(num_mapped[our_cur_subgraphs] < buffer_elms); + edge_buf[num_mapped[our_cur_subgraphs]++] = {subgraph_idx, dst_vertices[i]}; + } + } + + for (size_t i = 1; i < our_cur_subgraphs; i++) { + buffers[i].resize(num_mapped[i]); + cc_sketches[i]->apply_update_batch(thr_id, src_vertex, buffers[i]); + buffers[i].resize(buffer_elms); + } + + edge_buf.resize(num_mapped[our_cur_subgraphs]); + TaggedUpdateBatch batch = edge_store.insert_adj_edges(src_vertex, our_cur_subgraphs, edge_buf); + + while (batch.dsts_data.size() > 0) { + // we don't have a sketch for this subgraph yet! + if (batch.edge_store_subgraph > cur_subgraphs) { + advance_subgraph_lock.lock(); + // double check that we are the thread who will allocate next subgraph + if (batch.edge_store_subgraph > cur_subgraphs) { + advance_cur_subgraph(batch.edge_store_subgraph); + } + advance_subgraph_lock.unlock(); + } + + std::fill(&num_mapped[0], &num_mapped[max_subgraphs - 1], 0); + for (auto tagged_edge : batch.dsts_data) { + assert(tagged_edge.subgraph < cur_subgraphs); + + buffers[tagged_edge.subgraph][num_mapped[tagged_edge.subgraph]++] = tagged_edge.dst; + } + + for (size_t i = 1; i < batch.edge_store_subgraph; i++) { + if (num_mapped[i] > 0) { + buffers[i].resize(num_mapped[i]); + cc_sketches[i]->apply_update_batch(thr_id, batch.src, buffers[i]); + buffers[i].resize(buffer_elms); + } + } + + // check if there are more contractions to perform + if (edge_store.contract_in_progress()) + batch = edge_store.vertex_advance_subgraph(cur_subgraphs); + else + batch.dsts_data.clear(); + } +} + +static MinCut run_viecut(node_id_t num_vertices, std::vector &edges) { + typedef VieCut::mutable_graph Graph; + typedef std::shared_ptr GraphPtr; + + // Create a VieCut graph + GraphPtr G = std::make_shared(); + G->start_construction(num_vertices, edges.size()); + + // Add edges to VieCut graph + for (auto edge : edges) { + G->new_edge(edge.src, edge.dst); + } + + // finish construction and compute degrees + // TODO: Don't know if degrees are necessary. Its in the VieCut code tho + G->finish_construction(); + G->computeDegrees(); + + // Perform the mincut computation + VieCut::EdgeWeight cut; + VieCut::minimum_cut* mc = new VieCut::viecut(); + cut = mc->perform_minimum_cut(G); + + // Return answer + std::set left; + std::set right; + + for (node_id_t i = 0; i < num_vertices; i++) { + if (G->getNodeInCut(i)) + left.insert(i); + else + right.insert(i); + } + + delete mc; + return {left, right, cut}; +} + +MinCut MinCutSketchAlg::calc_minimum_cut() { +#ifdef VERIFY_SAMPLES_F + create_subgraph_verifiers(); +#endif + + auto start = std::chrono::steady_clock::now(); + sf_total_duration = std::chrono::duration(0); + viecut_duration = std::chrono::duration(0); + + std::cout << "Performing minimum cut query" << std::endl; + for (size_t i = 0; i < cur_subgraphs; i++) { + std::cout << "Sketch " << i << " updates = " << cc_sketches[i]->get_num_updates() << std::endl; + } + + + // iterate over our subgraphs to find the correct value + for (size_t i = 0; i < cur_subgraphs; i++) { + auto sf_start = std::chrono::steady_clock::now(); + std::vector sfs = cc_sketches[i]->calc_disjoint_spanning_forests(k); + std::vector edges; + + for (auto& sf : sfs) { + auto& sf_edges = sf.get_edges(); + edges.insert(edges.end(), sf_edges.begin(), sf_edges.end()); + } + sf_total_duration += std::chrono::steady_clock::now() - sf_start; + + auto viecut_start =std::chrono::steady_clock::now(); + MinCut mc = run_viecut(num_vertices, edges); + size_t adjust_value = mc.value << i; + std::cout << "Subgraph: " << i + 1 << ", cut value = " << mc.value << ", k = " << k + << ", adjusted = " << adjust_value << std::endl; + viecut_duration += std::chrono::steady_clock::now() - viecut_start; + + if (mc.value < k) { + mc.value = adjust_value; + total_mc_duration = std::chrono::steady_clock::now() - start; + return mc; + } + } + + // pull from the adjacency list + std::vector adj_edges = edge_store.get_edges(); + MinCut mc = run_viecut(num_vertices, adj_edges); + std::cout << "Edge Store MinCut = " << mc.value << std::endl; + + // multiply the minimum cut by sampling rate + // -1 because edge store contains everything in remaining subgraphs, geometric series + mc.value <<= (cur_subgraphs - 1); + std::cout << "Adjusted = " << mc.value << std::endl; + total_mc_duration = std::chrono::steady_clock::now() - start; + return mc; +} + +#ifdef VERIFY_SAMPLES_F +void MinCutSketchAlg::set_verifier(std::unique_ptr _verifier) { + verifier = std::make_unique(*_verifier); + cc_sketches[0]->set_verifier(std::make_unique(*verifier)); +} + +void MinCutSketchAlg::create_subgraph_verifiers() { + cc_sketches[0]->set_verifier(std::make_unique(*verifier)); + std::vector> subgraph_verifiers; + + std::cout << "Creating: " << cur_subgraphs + 1 << " verifiers" << std::endl; + + for (size_t i = 0; i <= cur_subgraphs; i++) { + subgraph_verifiers.emplace_back(new GraphVerifier(num_vertices)); + } + + size_t subgraph_sizes[max_subgraphs]; + std::fill(&subgraph_sizes[0], &subgraph_sizes[max_subgraphs - 1], 0); + size_t non_zero = 0; + + std::vector> adj_mat = verifier->extract_adj_matrix(); + for (node_id_t i = 0; i < num_vertices; i++) { + for (node_id_t j = 0; j < num_vertices - i; j++) { + if (adj_mat[i][j]) { + node_id_t dst = i + j; + + non_zero++; + vec_t idx = concat_pairing_fn(i, dst); // edge is + node_id_t subgraph_idx = Bucket_Boruvka::get_index_depth(idx, subgraph_seed, max_subgraphs - 1) + 1; + if (subgraph_idx < cur_subgraphs) { + subgraph_verifiers[subgraph_idx]->edge_update({i, dst}); + subgraph_sizes[subgraph_idx]++; + } else { + subgraph_verifiers[cur_subgraphs]->edge_update({i, dst}); + } + } + } + } + + std::cout << "verifier subgraph 0 size = " << non_zero << std::endl; + for (size_t i = 1; i < cur_subgraphs; i++) { + cc_sketches[i]->set_verifier(std::make_unique(*subgraph_verifiers[i])); + std::cout << "verifier subgraph " << i << " size = " << subgraph_sizes[i] << std::endl; + } + adj_verifier = std::move(subgraph_verifiers[cur_subgraphs]); +} +#endif diff --git a/src/return_types.cpp b/src/return_types.cpp index f4c4998d..9689deb1 100644 --- a/src/return_types.cpp +++ b/src/return_types.cpp @@ -1,5 +1,6 @@ #include "return_types.h" +#include #include ConnectedComponents::ConnectedComponents(node_id_t num_vertices, @@ -39,3 +40,23 @@ SpanningForest::SpanningForest(node_id_t num_vertices, } } } + +const std::vector &SpanningForest::get_sorted_adjacency() { + if (has_adjacency) return sorted_adjacency; + + size_t num = edges.size(); + sorted_adjacency.resize(edges.size() * 2); + +#pragma omp parallel for + for (size_t i = 0; i < num; i++) { + sorted_adjacency[i] = edges[i]; + sorted_adjacency[i + num] = sorted_adjacency[i]; + std::swap(sorted_adjacency[i + num].src, sorted_adjacency[i + num].dst); + } + + // sort the edges + std::sort(sorted_adjacency.begin(), sorted_adjacency.end()); + + has_adjacency = true; + return sorted_adjacency; +} diff --git a/src/sketch.cpp b/src/sketch.cpp index ac674c5e..724f6a02 100644 --- a/src/sketch.cpp +++ b/src/sketch.cpp @@ -45,7 +45,20 @@ Sketch::Sketch(const Sketch &s) : seed(s.seed) { std::memcpy(buckets, s.buckets, bucket_array_bytes()); } -Sketch::~Sketch() { delete[] buckets; } +Sketch& Sketch::operator=(Sketch &&oth) { + seed = oth.seed; + cols_per_sample = oth.cols_per_sample; + num_columns = oth.num_columns; + bkt_per_col = oth.bkt_per_col; + num_buckets = oth.num_buckets; + buckets = oth.buckets; + + oth.buckets = nullptr; + + return *this; +} + +Sketch::~Sketch() { if (buckets != nullptr) delete[] buckets; } #ifdef L0_SAMPLING void Sketch::update(const vec_t update_idx) { diff --git a/src/util.cpp b/src/util.cpp index 9854dbaa..b295c785 100644 --- a/src/util.cpp +++ b/src/util.cpp @@ -8,8 +8,9 @@ typedef uint32_t ul; typedef uint64_t ull; -constexpr ull ULLMAX = std::numeric_limits::max(); -constexpr uint8_t num_bits = sizeof(node_id_t) * 8; +static constexpr ull ULLMAX = std::numeric_limits::max(); +static constexpr uint8_t num_bits = sizeof(node_id_t) * 8; +static constexpr size_t mask = (size_t(1) << num_bits) - 1; unsigned long long int double_to_ull(double d, double epsilon) { return (unsigned long long) (d + epsilon); @@ -42,16 +43,13 @@ Edge inv_nondir_non_self_edge_pairing_fn(uint64_t idx) { return {(node_id_t)i, (node_id_t)j}; } +// smaller node_id concatenated with larger node_id edge_id_t concat_pairing_fn(node_id_t i, node_id_t j) { - // swap i,j if necessary - if (i > j) { - std::swap(i,j); - } - return ((edge_id_t)i << num_bits) | j; + return ((edge_id_t)std::min(i, j) << num_bits) | std::max(i, j); } Edge inv_concat_pairing_fn(ull idx) { - node_id_t j = idx & 0xFFFFFFFF; + node_id_t j = idx & mask; node_id_t i = idx >> num_bits; return {i, j}; } diff --git a/test/edge_store_test.cpp b/test/edge_store_test.cpp new file mode 100644 index 00000000..f4cca809 --- /dev/null +++ b/test/edge_store_test.cpp @@ -0,0 +1,240 @@ +#include "edge_store.h" +#include "bucket.h" +#include "util.h" + +#include +#include +#include + +static size_t get_seed() { + auto now = std::chrono::high_resolution_clock::now(); + return std::chrono::duration_cast(now.time_since_epoch()).count(); +} + +TEST(EdgeStoreTest, no_contract) { + size_t nodes = 1024; + size_t skt_bytes = 1 << 15; // artificially large + EdgeStore edge_store(get_seed(), nodes, skt_bytes, 20); + + std::set edges_added; + + for (size_t i = 0; i < nodes; i++) { + std::vector dsts; + for (size_t j = 0; j < i; j++) { + dsts.push_back(j); + ASSERT_TRUE(edges_added.insert({std::min(i, j), std::max(i, j)}).second); + } + auto more_upds = edge_store.insert_adj_edges(i, dsts); + ASSERT_EQ(more_upds.dsts_data.size(), 0); + } + + ASSERT_EQ(edge_store.get_num_edges(), edges_added.size()); + ASSERT_EQ(edge_store.get_first_store_subgraph(), 0); + + std::vector edges = edge_store.get_edges(); + ASSERT_EQ(edges.size(), edges_added.size()); + + for (auto edge : edges) { + node_id_t src = std::min(edge.src, edge.dst); + node_id_t dst = std::max(edge.src, edge.dst); + ASSERT_NE(edges_added.find({src, dst}), edges_added.end()); + } +} + +TEST(EdgeStoreTest, test_sort) { + size_t nodes = 1024; + size_t skt_bytes = 1 << 15; // artificially large + EdgeStore edge_store(get_seed(), nodes, skt_bytes, 20); + + std::set edges_added; + + std::vector dsts1{1,5,5,10,15,15,15,20,25,30,30,30,40,40}; + std::vector dsts2{3,15,20,100,101,102,103}; + std::set expect{1,3,10,25,30,100,101,102,103}; + + edge_store.insert_adj_edges(0, dsts1); + edge_store.insert_adj_edges(0, dsts2); + + std::vector edges = edge_store.get_edges(); + ASSERT_EQ(edge_store.get_num_edges(), 9); + ASSERT_EQ(edges.size(), 9); + + for (size_t i = 0; i < 9; i++) { + node_id_t src = edges[i].src; + node_id_t dst = edges[i].dst; + ASSERT_EQ(src, 0); + ASSERT_NE(expect.find(dst), expect.end()); + } +} + +TEST(EdgeStoreTest, no_contract_dupl) { + size_t nodes = 1024; + size_t skt_bytes = 1 << 20; // artificially large + EdgeStore edge_store(get_seed(), nodes, skt_bytes, 20); + + std::set edges_added; + + for (size_t i = 0; i < nodes; i++) { + std::vector dsts; + for (size_t j = 0; j < i; j++) { + dsts.push_back(j); + ASSERT_TRUE(edges_added.insert({i, j}).second); + } + auto more_upds = edge_store.insert_adj_edges(i, dsts); + ASSERT_EQ(more_upds.dsts_data.size(), 0); + } + + // remove num_nodes edges from graph + for (size_t i = 0; i < nodes; i++) { + auto it = edges_added.begin(); + edge_store.insert_adj_edges(it->src, std::vector{it->dst}); + edges_added.erase(it); + } + + ASSERT_EQ(edge_store.get_num_edges(), edges_added.size()); + ASSERT_EQ(edge_store.get_first_store_subgraph(), 0); + + std::vector edges = edge_store.get_edges(); + ASSERT_EQ(edges.size(), edges_added.size()); + + for (auto edge : edges) { + ASSERT_NE(edges_added.find(edge), edges_added.end()); + } +} + +TEST(EdgeStoreTest, contract) { + size_t nodes = 1024; + size_t skt_bytes = 1 << 9; // small enough to likely contract thrice + size_t num_subgraphs = 20; + size_t seed = get_seed(); + EdgeStore edge_store(seed, nodes, skt_bytes, num_subgraphs, 1); + + std::set edges_added; + + size_t num_returned[6] = {0, 0, 0, 0, 0, 0}; + size_t num_in_subgraphs[6] = {0, 0, 0, 0, 0, 0}; + + for (size_t i = 0; i < nodes; i++) { + std::vector dsts; + for (size_t j = i + 1; j < nodes; j++) { + node_id_t src = std::min(i, j); + node_id_t dst = std::max(i, j); + auto idx = concat_pairing_fn(src, dst); + size_t depth = Bucket_Boruvka::get_index_depth(idx, seed, num_subgraphs - 1) + 1; + + ++num_in_subgraphs[std::min(size_t(5), depth)]; + + if (depth >= edge_store.get_first_store_subgraph()) { + dsts.push_back({depth, j}); + } else { + ++num_returned[depth]; + } + ASSERT_TRUE(edges_added.insert({src, dst}).second); + } + auto more_upds = edge_store.insert_adj_edges(i, edge_store.get_first_store_subgraph(), dsts); + node_id_t src = more_upds.src; + for (auto dst_data : more_upds.dsts_data) { + ++num_returned[dst_data.subgraph]; + node_id_t s = std::min(src, dst_data.dst); + node_id_t d = std::max(src, dst_data.dst); + ASSERT_NE(edges_added.find({s, d}), edges_added.end()); + } + } + + while (edge_store.contract_in_progress()) { + auto more_upds = edge_store.vertex_advance_subgraph(edge_store.get_first_store_subgraph()); + node_id_t src = more_upds.src; + for (auto dst_data : more_upds.dsts_data) { + ++num_returned[dst_data.subgraph]; + node_id_t s = std::min(src, dst_data.dst); + node_id_t d = std::max(src, dst_data.dst); + ASSERT_NE(edges_added.find({s, d}), edges_added.end()); + } + } + + for (size_t i = 0; i < 6; i++) { + std::cerr << num_returned[i] << " vs " << num_in_subgraphs[i] << std::endl; + } + + for (size_t i = 0; i < edge_store.get_first_store_subgraph(); i++) { + ASSERT_EQ(num_returned[i], num_in_subgraphs[i]); + } + + size_t expected_es_edges = 0; + for (size_t i = edge_store.get_first_store_subgraph(); i <= 5; i++) { + expected_es_edges += num_in_subgraphs[i]; + } + + std::vector edges = edge_store.get_edges(); + ASSERT_EQ(edges.size(), expected_es_edges); + + for (auto edge : edges) { + node_id_t src = std::min(edge.src, edge.dst); + node_id_t dst = std::max(edge.src, edge.dst); + ASSERT_NE(edges_added.find({src, dst}), edges_added.end()); + } +} + +TEST(EdgeStoreTest, contract_parallel) { + size_t nodes = 1024; + size_t skt_bytes = 1 << 9; // small enough to likely contract twice + size_t num_subgraphs = 20; + size_t seed = get_seed(); + EdgeStore edge_store(seed, nodes, skt_bytes, num_subgraphs); + + std::atomic num_returned[6] = {0, 0, 0, 0, 0, 0}; + std::atomic num_in_subgraphs[6] = {0, 0, 0, 0, 0, 0}; + +#pragma omp parallel for + for (size_t i = 0; i < nodes; i++) { + std::vector dsts; + size_t edge_store_subgraph = edge_store.get_first_store_subgraph(); + for (size_t j = i + 1; j < nodes; j++) { + node_id_t src = std::min(i, j); + node_id_t dst = std::max(i, j); + auto idx = concat_pairing_fn(src, dst); + size_t depth = Bucket_Boruvka::get_index_depth(idx, seed, num_subgraphs - 1) + 1; + + ++num_in_subgraphs[std::min(size_t(5), depth)]; + + if (depth >= edge_store_subgraph) { + dsts.push_back({depth, j}); + } else { + ++num_returned[depth]; + } + } + auto more_upds = edge_store.insert_adj_edges(i, edge_store_subgraph, dsts); + for (auto dst_data : more_upds.dsts_data) { + ++num_returned[dst_data.subgraph]; + } + } + +#pragma omp parallel + { + while (edge_store.contract_in_progress()) { + auto more_upds = edge_store.vertex_advance_subgraph(edge_store.get_first_store_subgraph()); + for (auto dst_data : more_upds.dsts_data) { + ++num_returned[dst_data.subgraph]; + } + } + } + + for (size_t i = 0; i < 6; i++) { + std::cerr << num_returned[i] << " vs " << num_in_subgraphs[i] << std::endl; + } + + std::vector edges = edge_store.get_edges(); + std::cerr << "edge store size = " << edges.size() << std::endl; + + for (size_t i = 0; i < edge_store.get_first_store_subgraph(); i++) { + ASSERT_EQ(num_returned[i], num_in_subgraphs[i]); + } + + size_t expected_es_edges = 0; + for (size_t i = edge_store.get_first_store_subgraph(); i <= 5; i++) { + expected_es_edges += num_in_subgraphs[i]; + } + + + ASSERT_EQ(edges.size(), expected_es_edges); +} diff --git a/test/min_cut_test.cpp b/test/min_cut_test.cpp new file mode 100644 index 00000000..06aebd8e --- /dev/null +++ b/test/min_cut_test.cpp @@ -0,0 +1,104 @@ +#include +#include +#include +#include +#include + +#include +#include + +#include "min_cut_sketch_alg.h" +#include "graph_sketch_driver.h" +#include "graph_verifier.h" + +// helper function to generate a dynamic binary stream and its cumulative insert only stream +void generate_stream(size_t seed, node_id_t num_vertices, double density, std::string stream_name) { + // generate new stream files + StaticErdosGenerator stream(seed, num_vertices, density); + stream.to_binary_file(stream_name); +} + +static size_t get_seed() { + auto now = std::chrono::high_resolution_clock::now(); + size_t s = std::chrono::duration_cast(now.time_since_epoch()).count(); + std::cout << "Seed = " << s << std::endl; + return s; +} + +// Test for constructing mincut and driver. Just to ensure everything work +TEST(MinCutTest, Construction) { + const std::string fname = __FILE__; + size_t pos = fname.find_last_of("\\/"); + const std::string curr_dir = (std::string::npos == pos) ? "" : fname.substr(0, pos); + AsciiFileStream stream{curr_dir + "/res/multiples_graph_1024.txt", false}; + + MinCutSketchAlg mc_alg(1024, get_seed()); + GraphSketchDriver driver(&mc_alg, &stream, DriverConfiguration()); +} + +TEST(MinCutTest, DisconnectedMinCut) { + auto driver_config = DriverConfiguration(); + const std::string fname = __FILE__; + size_t pos = fname.find_last_of("\\/"); + const std::string curr_dir = (std::string::npos == pos) ? "" : fname.substr(0, pos); + AsciiFileStream stream{curr_dir + "/res/multiples_graph_1024.txt", false}; + node_id_t num_vertices = stream.vertices(); + + MinCutSketchAlg mc_alg(num_vertices, get_seed()); + + GraphSketchDriver driver(&mc_alg, &stream, driver_config); + driver.process_stream_until(END_OF_STREAM); + driver.prep_query(MINIMUMCUT); + driver.check_verifier(GraphVerifier(num_vertices, curr_dir + "/res/multiples_graph_1024.txt")); + + ASSERT_EQ(0, mc_alg.calc_minimum_cut().value); +} + +TEST(MinCutTest, CompleteGraphMinCut) { + auto driver_config = DriverConfiguration().worker_threads(8); + double epsilon = 0.8; + generate_stream(get_seed(), 1 << 13, 1, "./complete_stream.bin"); + size_t true_mc = (1 << 13) - 1; + + BinaryFileStream stream{"./complete_stream.bin"}; + node_id_t num_vertices = stream.vertices(); + + MinCutSketchAlg mc_alg(num_vertices, get_seed(), MCAlgConfiguration().epsilon(0.8)); + + GraphSketchDriver driver(&mc_alg, &stream, driver_config); + driver.process_stream_until(END_OF_STREAM); + driver.prep_query(MINIMUMCUT); + + MinCut ret = mc_alg.calc_minimum_cut(); + + ASSERT_GT(true_mc, ret.value); + ASSERT_LT(true_mc * (1 - epsilon), ret.value); + std::remove("./complete_stream.bin"); +} + +TEST(MinCutTest, DynamicMinCut) { + auto driver_config = DriverConfiguration().worker_threads(8); + double epsilon = 0.8; + node_id_t num_vertices = 1 << 13; + { + DynamicErdosGenerator gen(get_seed(), num_vertices, 0.5, 0.25, 0.25, 2); + gen.to_binary_file("./dynamic_stream.bin"); + } + + size_t true_mc = num_vertices / 2; // probably about half of complete graph min-cut + + BinaryFileStream stream{"./dynamic_stream.bin"}; + + MinCutSketchAlg mc_alg(num_vertices, get_seed(), MCAlgConfiguration().epsilon(0.8)); + + GraphSketchDriver driver(&mc_alg, &stream, driver_config); + driver.process_stream_until(END_OF_STREAM); + driver.prep_query(MINIMUMCUT); + + MinCut ret = mc_alg.calc_minimum_cut(); + + ASSERT_GT(true_mc, ret.value); + ASSERT_LT(true_mc * (1 - epsilon), ret.value); + std::remove("./dynamic_stream.bin"); +} + diff --git a/tools/minimum_cut.cpp b/tools/minimum_cut.cpp new file mode 100644 index 00000000..df15ed79 --- /dev/null +++ b/tools/minimum_cut.cpp @@ -0,0 +1,143 @@ +#include "min_cut_sketch_alg.h" +#include "graph_sketch_driver.h" +#include +#include +#include // for rusage + +static bool shutdown = false; + +static double get_max_mem_used() { + struct rusage data; + getrusage(RUSAGE_SELF, &data); + return (double) data.ru_maxrss / 1024.0; +} + +static size_t get_seed() { + auto now = std::chrono::high_resolution_clock::now(); + return std::chrono::duration_cast(now.time_since_epoch()).count(); +} + +/* + * Function which is run in a seperate thread and will query + * the graph for the number of updates it has processed + * @param total the total number of edge updates + * @param g the graph object to query + * @param start_time the time that we started stream ingestion + */ +static void track_insertions(uint64_t total, GraphSketchDriver *driver, + std::chrono::steady_clock::time_point start_time) { + total = total * 2; // we insert 2 edge updates per edge + + printf("Insertions\n"); + printf("Progress: | 0%%\r"); fflush(stdout); + std::chrono::steady_clock::time_point start = start_time; + std::chrono::steady_clock::time_point prev = start_time; + uint64_t prev_updates = 0; + + while(true) { + sleep(1); + uint64_t updates = driver->get_total_updates(); + std::chrono::steady_clock::time_point now = std::chrono::steady_clock::now(); + std::chrono::duration total_diff = now - start; + std::chrono::duration cur_diff = now - prev; + + // calculate the insertion rate + uint64_t upd_delta = updates - prev_updates; + // divide insertions per second by 2 because each edge is split into two updates + // we care about edges per second not about stream updates + size_t ins_per_sec = (((double)(upd_delta)) / cur_diff.count()) / 2; + + if (updates >= total || shutdown) + break; + + // display the progress + int progress = updates / (total * .05); + printf("Progress:%s%s", std::string(progress, '=').c_str(), std::string(20 - progress, ' ').c_str()); + printf("| %i%% -- %lu per second\r", progress * 5, ins_per_sec); fflush(stdout); + } + + printf("Progress:====================| Done \n"); + return; +} + +int main(int argc, char **argv) { + if (argc != 5) { + std::cerr << "ERROR: Incorrect number of arguments!" << std::endl; + std::cerr << "Arguments: stream_file, num_queries, graph_workers, reader_threads" << std::endl; + exit(EXIT_FAILURE); + } + + shutdown = false; + std::string stream_file = argv[1]; + int num_queries = std::atoi(argv[2]); + if (num_queries < 1 || num_queries > 1000) { + std::cerr << "ERROR: Invalid number of queries! Must be > 0 and <= 1000" << std::endl; + exit(EXIT_FAILURE); + } + int num_threads = std::atoi(argv[3]); + if (num_threads < 1) { + std::cerr << "ERROR: Invalid number of graph workers! Must be > 0." << std::endl; + exit(EXIT_FAILURE); + } + size_t reader_threads = std::atol(argv[4]); + + double query_percent = 1.0 / num_queries; + size_t queries_in_stream = num_queries - 1; + + BinaryFileStream stream(stream_file); + node_id_t num_nodes = stream.vertices(); + size_t num_updates = stream.edges(); + std::cout << "Processing stream: " << stream_file << std::endl; + std::cout << "nodes = " << num_nodes << std::endl; + std::cout << "num_updates = " << num_updates << std::endl; + std::cout << std::endl; + + auto driver_config = DriverConfiguration().gutter_sys(CACHETREE).worker_threads(num_threads); + auto mc_config = MCAlgConfiguration().batch_factor(1.0); + MinCutSketchAlg mc_alg{num_nodes, get_seed(), mc_config}; + GraphSketchDriver driver{&mc_alg, &stream, driver_config, reader_threads}; + + auto ins_start = std::chrono::steady_clock::now(); + std::thread querier(track_insertions, num_updates, &driver, ins_start); + + for (size_t q = 0; q < queries_in_stream; q++) { + driver.process_stream_until((q+1) * query_percent * num_updates); + auto query_start = std::chrono::steady_clock::now(); + driver.prep_query(CONNECTIVITY); + auto mc = mc_alg.calc_minimum_cut(); + std::chrono::duration mc_time = std::chrono::steady_clock::now() - query_start; + std::chrono::duration flush_time = driver.flush_end - driver.flush_start; + + std::cout << "Query " << q + 1 << std::endl; + std::cout << "Total Minimum cut latency: " << mc_time.count() << std::endl; + std::cout << " Flush Gutters(sec): " << flush_time.count() << std::endl; + std::cout << " Creating SFs(sec): " << mc_alg.sf_total_duration.count() << std::endl; + std::cout << " Viecut algorithm(sec): " << mc_alg.viecut_duration.count() << std::endl; + std::cout << "Minimum Cut: " << mc.value << std::endl; + + } + + // finish the stream + driver.process_stream_until(END_OF_STREAM); + + auto query_start = std::chrono::steady_clock::now(); + driver.prep_query(CONNECTIVITY); + auto mc = mc_alg.calc_minimum_cut(); + std::chrono::duration mc_time = std::chrono::steady_clock::now() - query_start; + std::chrono::duration flush_time = driver.flush_end - driver.flush_start; + + std::chrono::duration insert_time = driver.flush_end - ins_start; + shutdown = true; + querier.join(); + + double num_seconds = insert_time.count(); + + std::cout << "Total insertion time(sec): " << num_seconds << std::endl; + std::cout << "Updates per second: " << stream.edges() / num_seconds << std::endl; + std::cout << "Total Minimum cut latency: " << mc_time.count() << std::endl; + std::cout << " Flush Gutters(sec): " << flush_time.count() << std::endl; + std::cout << " Creating SFs(sec): " << mc_alg.sf_total_duration.count() << std::endl; + std::cout << " Viecut algorithm(sec): " << mc_alg.viecut_duration.count() << std::endl; + std::cout << "Minimum Cut: " << mc.value << std::endl; + std::cout << "Maximum Memory Usage(MiB): " << get_max_mem_used() << std::endl; +}