diff --git a/CMakeLists.txt b/CMakeLists.txt index 95e90896..1f324a7f 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -130,13 +130,17 @@ if (BUILD_EXE) tools/sketch_testing.cpp) add_dependencies(statistical_sketch_test GraphZeppelinVerifyCC) target_link_libraries(statistical_sketch_test PRIVATE GraphZeppelinVerifyCC) - # executable for processing a binary graph stream add_executable(process_stream tools/process_stream.cpp) target_link_libraries(process_stream PRIVATE GraphZeppelin) + # executable for testing the spanning forest extraction from BinaryFileStream + add_executable(spanning_forest_extract + tools/spanning_forest_extract.cpp) + target_link_libraries(spanning_forest_extract 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..5c820586 100644 --- a/include/cc_sketch_alg.h +++ b/include/cc_sketch_alg.h @@ -128,6 +128,12 @@ class CCSketchAlg { */ void boruvka_emulation(); + /** + * Delete edges found in spanning forest from sketches + * Helper function for calc_disjoint_spanning_forests(k) + */ + 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); @@ -234,6 +240,14 @@ class CCSketchAlg { */ SpanningForest calc_spanning_forest(); + /** + * Returns edges that form 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 vector of edges that together form the 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); @@ -249,6 +263,10 @@ class CCSketchAlg { // time hooks for experiments std::chrono::steady_clock::time_point cc_alg_start; std::chrono::steady_clock::time_point cc_alg_end; + std::chrono::steady_clock::time_point sf_query_start; + std::chrono::steady_clock::time_point sf_query_end; + std::chrono::duration query_time; + std::chrono::duration delete_time; size_t last_query_rounds = 0; // getters diff --git a/include/return_types.h b/include/return_types.h index b329bfe6..e58b8858 100644 --- a/include/return_types.h +++ b/include/return_types.h @@ -28,10 +28,13 @@ 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(); }; diff --git a/src/cc_sketch_alg.cpp b/src/cc_sketch_alg.cpp index a1e688db..2185c925 100644 --- a/src/cc_sketch_alg.cpp +++ b/src/cc_sketch_alg.cpp @@ -80,7 +80,7 @@ void CCSketchAlg::pre_insert(GraphUpdate upd, int /* thr_id */) { #ifdef NO_EAGER_DSU (void)upd; // reason we have an if statement: avoiding cache coherency issues - unlikely_if(dsu_valid) { + unlikely_if (dsu_valid) { dsu_valid = false; shared_dsu_valid = false; } @@ -561,12 +561,89 @@ SpanningForest CCSketchAlg::calc_spanning_forest() { connected_components(); SpanningForest ret(num_vertices, spanning_forest); + #ifdef VERIFY_SAMPLES_F verifier->verify_spanning_forests(std::vector{ret}); #endif 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; + + // 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 (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 < edges.size() && edges[end - 1].src == edges[end].src) { + sketches[edges[end - 1].src]->mutex.lock(); + size_t orig_end = end; + while (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::vector SFs; + std::chrono::steady_clock::time_point start; + + for (size_t i = 0; i < k; i++) { + start = std::chrono::steady_clock::now(); + SFs.push_back(calc_spanning_forest()); + query_time += std::chrono::steady_clock::now() - start; + + filter_sf_edges(SFs[SFs.size() - 1]); + } + + // 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 + + 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/return_types.cpp b/src/return_types.cpp index f4c4998d..af8084a5 100644 --- a/src/return_types.cpp +++ b/src/return_types.cpp @@ -1,6 +1,7 @@ #include "return_types.h" #include +#include ConnectedComponents::ConnectedComponents(node_id_t num_vertices, DisjointSetUnion_MT &dsu) @@ -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/test/cc_alg_test.cpp b/test/cc_alg_test.cpp index 5f395b01..4aa3494a 100644 --- a/test/cc_alg_test.cpp +++ b/test/cc_alg_test.cpp @@ -279,6 +279,28 @@ TEST(CCAlgTest, SpanningForestExtraction) { cc_alg.calc_spanning_forest(); } +TEST(CCAlgTest, MultipleSpanningForests) { + auto driver_config = DriverConfiguration().gutter_sys(STANDALONE); + auto cc_config = CCAlgConfiguration().sketches_factor(1.6); + generate_stream(get_seed(), 1024, 0.1, 0.1, 0.05, 3, "sample.txt", "cumul_sample.txt"); + AsciiFileStream stream{"./sample.txt"}; + node_id_t num_nodes = stream.vertices(); + + CCSketchAlg cc_alg{num_nodes, get_seed()}; + GraphSketchDriver driver(&cc_alg, &stream, driver_config); + + driver.process_stream_until(END_OF_STREAM); + driver.prep_query(CONNECTIVITY); + driver.check_verifier(GraphVerifier(1024, "./cumul_sample.txt")); + + // do a bunch of query calls to ensure sketches are in a good state after queries. + cc_alg.calc_disjoint_spanning_forests(8); + + cc_alg.calc_spanning_forest(); + + cc_alg.calc_disjoint_spanning_forests(8); +} + TEST(CCAlgTest, InsertOnlyStream) { auto driver_config = DriverConfiguration().gutter_sys(STANDALONE); auto cc_config = CCAlgConfiguration(); diff --git a/tools/benchmark/graphcc_bench.cpp b/tools/benchmark/graphcc_bench.cpp index fc5b8995..96a15c1e 100644 --- a/tools/benchmark/graphcc_bench.cpp +++ b/tools/benchmark/graphcc_bench.cpp @@ -534,4 +534,76 @@ static void BM_Parallel_DSU_Root(benchmark::State& state) { } BENCHMARK(BM_Parallel_DSU_Root)->RangeMultiplier(2)->Range(1, 8)->UseRealTime(); +// Test the speed of preforming redundant merge operations + +static void BM_DSU_Redundant_Merge(benchmark::State& state) { + constexpr size_t size_of_dsu = 16 * MB; + + auto rng = std::default_random_engine{}; + + // generate updates + std::vector> updates; + // generate updates + for (size_t iter = 0; ((size_t)2 << iter) <= size_of_dsu; iter++) { + size_t jump = 2 << iter; + std::vector> new_updates; + for (size_t i = 0; i < size_of_dsu; i += jump) { + new_updates.push_back({i, i + jump / 2}); + } + std::shuffle(new_updates.begin(), new_updates.end(), rng); + updates.insert(updates.end(), new_updates.begin(), new_updates.end()); + } + DisjointSetUnion dsu(size_of_dsu); + for (auto upd : updates) { + dsu.merge(upd.first, upd.second); + } + + // Perform merge test + for (auto _ : state) { + for (auto upd : updates) { + dsu.merge(upd.first, upd.second); + } + } + state.counters["Merge_Latency"] = + benchmark::Counter(state.iterations() * updates.size(), + benchmark::Counter::kIsRate | benchmark::Counter::kInvert); +} +BENCHMARK(BM_DSU_Redundant_Merge); + +static void BM_Parallel_DSU_Redundant_Merge(benchmark::State& state) { + constexpr size_t size_of_dsu = 16 * MB; + + auto rng = std::default_random_engine{}; + + // generate updates + std::vector> updates; + // generate updates + for (size_t iter = 0; ((size_t)2 << iter) <= size_of_dsu; iter++) { + size_t jump = 2 << iter; + std::vector> new_updates; + for (size_t i = 0; i < size_of_dsu; i += jump) { + new_updates.push_back({i, i + jump / 2}); + } + std::shuffle(new_updates.begin(), new_updates.end(), rng); + updates.insert(updates.end(), new_updates.begin(), new_updates.end()); + } + + DisjointSetUnion_MT dsu(size_of_dsu); + for (auto upd : updates) { + dsu.merge(upd.first, upd.second); + } + + // Perform merge test + for (auto _ : state) { +#pragma omp parallel for num_threads(state.range(0)) + for (auto upd : updates) { + dsu.merge(upd.first, upd.second); + } + } + state.counters["Merge_Latency"] = + benchmark::Counter(state.iterations() * updates.size(), + benchmark::Counter::kIsRate | benchmark::Counter::kInvert); +} +BENCHMARK(BM_Parallel_DSU_Redundant_Merge)->RangeMultiplier(2)->Range(1, 8)->UseRealTime(); + BENCHMARK_MAIN(); diff --git a/tools/spanning_forest_extract.cpp b/tools/spanning_forest_extract.cpp new file mode 100644 index 00000000..3b54b5c1 --- /dev/null +++ b/tools/spanning_forest_extract.cpp @@ -0,0 +1,127 @@ +#include +#include +#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(); +} + +std::pair calc_stats(const std::vector &data) { + double avg = 0; + size_t total = 0; + for (size_t i = 0; i < data.size(); i++) { + avg += i * data[i]; + total += data[i]; + } + avg /= total; + + double dev = 0; + for (size_t i = 0; i < data.size(); i++) { + dev += data[i] * pow(double(i) - avg, 2); + } + + return {avg, sqrt(dev / total)}; +} + +// Main function that populates a graph and then tests how many SFs can be extracted from it +int main(int argc, char **argv) { + if (argc < 3 || argc > 5) { + std::cout << "ERROR: Incorrect number of arguments!" << std::endl; + std::cout << "Arguments: graph_stream trials [seed]" << std::endl; + std::cout << "\"graph_stream\" must be a BinaryFileStream." << std::endl; + std::cout << "Optionally specify a seed, otherwise one is chosen randomly" << std::endl; + exit(EXIT_FAILURE); + } + + std::string stream_file = argv[1]; + size_t trials = std::stol(argv[2]); + size_t seed = get_seed(); + if (argc == 4) { + seed = std::stol(argv[3]); + } + + size_t num_threads = 40; + size_t reader_threads = 8; + std::vector rounds_required; + std::mt19937_64 seed_gen(seed); + + node_id_t num_vertices; + size_t num_updates; + { + BinaryFileStream stream(stream_file); + num_vertices = stream.vertices(); + num_updates = stream.edges(); + } + size_t vertex_power = ceil(log2(num_vertices)); + size_t errors = 0; + size_t empty = 0; + std::chrono::duration ingest_time(0); + std::chrono::duration query_time(0); + std::chrono::duration sample_time(0); + std::chrono::duration delete_time(0); + + for (size_t trial = 0; trial < trials; trial++) { + BinaryFileStream stream(stream_file); + std::cout << "Extracting " << vertex_power << " Spanning Forests from: " << stream_file << std::endl; + std::cout << "vertices = " << num_vertices << 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 cc_config = CCAlgConfiguration().sketches_factor(1.6); + CCSketchAlg cc_alg{num_vertices, seed_gen(), cc_config}; + GraphSketchDriver driver{&cc_alg, &stream, driver_config, reader_threads}; + + rounds_required.resize(cc_alg.max_rounds()); + + std::cout << "Beginning stream ingestion ... "; fflush(stdout); + auto start = std::chrono::steady_clock::now(); + driver.process_stream_until(END_OF_STREAM); + driver.prep_query(KSPANNINGFORESTS); + std::cout << "Stream processed!" << std::endl; + ingest_time += std::chrono::steady_clock::now() - start; + std::cout << "Ingestion throughput: " << num_updates / std::chrono::duration(std::chrono::steady_clock::now() - start).count() << std::endl; + + // figure out how many rounds are required to extract log V spanning forests + start = std::chrono::steady_clock::now(); + cc_alg.calc_disjoint_spanning_forests(vertex_power); + query_time += std::chrono::steady_clock::now() - start; + + // add number of rounds to get log V spanning forests to vector + rounds_required[cc_alg.last_query_rounds] += 1; + + sample_time += cc_alg.query_time; + delete_time += cc_alg.delete_time; + } + + std::cout << std::endl; + std::cout << "ERRORS = " << errors << std::endl; + std::cout << "EMPTY = " << empty << std::endl; + + // for (size_t i = 0; i < rounds_required.size(); i++) { + // std::cout << i << ", " << rounds_required[i] << std::endl; + // } + + auto stats = calc_stats(rounds_required); + std::cout << "avg = " << stats.first << " std dev = " << stats.second << std::endl; + std::cout << "ingest: " << ingest_time.count() << std::endl; + std::cout << "query: " << query_time.count() << std::endl; + std::cout << " sample: " << sample_time.count() << std::endl; + std::cout << " delete: " << delete_time.count() << std::endl; + + std::ofstream output_file("rounds_required.txt"); + output_file << "ERRORS = " << errors << std::endl; + output_file << "EMPTY = " << empty << std::endl; + for (size_t i = 0; i < rounds_required.size(); i++) { + output_file << i << ", " << rounds_required[i] << std::endl; + } + output_file << "avg, " << stats.first << std::endl; + output_file << "std dev, " << stats.second << std::endl; + output_file << "ingest: " << ingest_time.count() << std::endl; + output_file << "query: " << query_time.count() << std::endl; + output_file << " sample: " << sample_time.count() << std::endl; + output_file << " delete: " << delete_time.count() << std::endl; +}