From 968aa4926add0192b745d9b384b291a4afaa80e6 Mon Sep 17 00:00:00 2001 From: florijan Date: Wed, 22 Nov 2017 10:04:12 +0100 Subject: [PATCH] Add parallel customers/Otto test Summary: Looking for connected components in a random graph. This test performs the following: - Generates a random graph that is NOT sequential in memory (otherwise itertion over edges is 2 or more times faster). - Connectivity by iterating over all the edges. - Ditto over vertices. - Ditto over vertices in parallel. Not done: - Edge filtering based on XY. I could/should add that to see how it affects perf. - Getting component info out from union-find. Local results are encouraging. Iterating over the graph is the bottleneck. Still, I get connectivity of 10M vertices/edges in <7sec (parallel over vertices). Will test on 250M remote now. Locally obtained results (20M/20M, 2 threads) ``` I1115 14:57:55.136875 357 otto_parallel.cpp:50] Generating 2000000 vertices... I1115 14:58:19.057734 357 otto_parallel.cpp:74] Generated 2000000 vertices in 23.9208 seconds. I1115 14:58:19.919221 357 otto_parallel.cpp:82] Generating 2000000 edges... I1115 14:58:39.519951 357 otto_parallel.cpp:93] Generated 2000000 edges in 19.3398 seconds. I1115 14:58:39.520349 357 otto_parallel.cpp:196] Running Edge iteration... I1115 14:58:43.857264 357 otto_parallel.cpp:199] Done in 4.33691 seconds, result: 3999860270398 I1115 14:58:43.857316 357 otto_parallel.cpp:196] Running Vertex iteration... I1115 14:58:49.498181 357 otto_parallel.cpp:199] Done in 5.64087 seconds, result: 4000090070787 I1115 14:58:49.498208 357 otto_parallel.cpp:196] Running Connected components - Edges... I1115 14:58:54.232530 357 otto_parallel.cpp:199] Done in 4.73433 seconds, result: 323935 I1115 14:58:54.232570 357 otto_parallel.cpp:196] Running Connected components - Vertices... I1115 14:59:00.412395 357 otto_parallel.cpp:199] Done in 6.17983 seconds, result: 323935 I1115 14:59:00.412422 357 otto_parallel.cpp:196] Running Parallel connected components - Vertices... I1115 14:59:04.662087 357 otto_parallel.cpp:199] Done in 4.24967 seconds, result: 323935 I1115 14:59:04.662116 357 otto_parallel.cpp:196] Running Expansion... I1115 14:59:13.913015 357 otto_parallel.cpp:199] Done in 9.25091 seconds, result: 323935 ``` Reviewers: buda, mislav.bradac, dgleich, teon.banek Reviewed By: buda, teon.banek Subscribers: teon.banek, pullbot Differential Revision: https://phabricator.memgraph.io/D982 --- CMakeLists.txt | 12 +- customers/CMakeLists.txt | 2 + customers/otto/CMakeLists.txt | 3 + .../otto/parallel_connected_components.cpp | 209 ++++++++++++++++++ {poc => src/data_structures}/union_find.hpp | 38 ++-- .../unit/datastructure_union_find.cpp | 42 ++-- 6 files changed, 265 insertions(+), 41 deletions(-) create mode 100644 customers/CMakeLists.txt create mode 100644 customers/otto/CMakeLists.txt create mode 100644 customers/otto/parallel_connected_components.cpp rename {poc => src/data_structures}/union_find.hpp (66%) rename poc/union_find_test.cpp => tests/unit/datastructure_union_find.cpp (60%) diff --git a/CMakeLists.txt b/CMakeLists.txt index 96046fccd..823cd0a2e 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -135,7 +135,10 @@ option(POC "Build proof of concept binaries" ON) message(STATUS "POC binaries: ${POC}") # experimental option(EXPERIMENTAL "Build experimental binaries" OFF) -message(STATUS "POC binaries: ${POC}") +message(STATUS "Experimental binaries: ${EXPERIMENTAL}") +# customers +option(CUSTOMERS "Customer binaries" ON) +message(STATUS "Customers binaries: ${CUSTOMERS}") # tests option(TEST_COVERAGE "Generate coverage reports from unit tests" OFF) message(STATUS "Generate coverage from unit tests: ${TEST_COVERAGE}") @@ -248,6 +251,7 @@ add_dependencies(memgraph_lib generate_opencypher_parser) if (POC) add_subdirectory(poc) endif() +# ----------------------------------------------------------------------------- # experimental if (EXPERIMENTAL) @@ -255,6 +259,12 @@ if (EXPERIMENTAL) endif() # ----------------------------------------------------------------------------- +# customers +if (CUSTOMERS) + add_subdirectory(customers) +endif() +# ----------------------------------------------------------------------------- + # tests enable_testing() add_subdirectory(tests) diff --git a/customers/CMakeLists.txt b/customers/CMakeLists.txt new file mode 100644 index 000000000..f977c36af --- /dev/null +++ b/customers/CMakeLists.txt @@ -0,0 +1,2 @@ +project(mg_customers) +add_subdirectory(otto) diff --git a/customers/otto/CMakeLists.txt b/customers/otto/CMakeLists.txt new file mode 100644 index 000000000..db6f0d531 --- /dev/null +++ b/customers/otto/CMakeLists.txt @@ -0,0 +1,3 @@ +set(exec_name customers_otto_parallel_connected_components) +add_executable(${exec_name} parallel_connected_components.cpp) +target_link_libraries(${exec_name} memgraph_lib) diff --git a/customers/otto/parallel_connected_components.cpp b/customers/otto/parallel_connected_components.cpp new file mode 100644 index 000000000..da2246b66 --- /dev/null +++ b/customers/otto/parallel_connected_components.cpp @@ -0,0 +1,209 @@ +#include +#include +#include +#include +#include +#include +#include + +#include "gflags/gflags.h" +#include "glog/logging.h" + +#include "data_structures/union_find.hpp" +#include "database/graph_db.hpp" +#include "database/graph_db_accessor.hpp" +#include "storage/property_value.hpp" +#include "threading/sync/spinlock.hpp" +#include "utils/bound.hpp" +#include "utils/timer.hpp" + +DEFINE_int32(thread_count, 1, "Number of threads"); +DEFINE_int32(vertex_count, 1000, "Number of vertices"); +DEFINE_int32(edge_count, 1000, "Number of edges"); +DECLARE_int32(gc_cycle_sec); + +static const std::string kLabel{"kLabel"}; +static const std::string kProperty{"kProperty"}; + +void GenerateGraph(GraphDb &db) { + { + GraphDbAccessor dba{db}; + dba.BuildIndex(dba.Label(kLabel), dba.Property(kProperty)); + dba.Commit(); + } + + // Randomize the sequence of IDs of created vertices and edges to simulate + // real-world lack of locality. + auto make_id_vector = [](size_t size) { + std::vector ids(size); + for (size_t i = 0; i < size; ++i) ids[i] = i; + std::random_shuffle(ids.begin(), ids.end()); + return ids; + }; + + std::vector vertices; + vertices.reserve(FLAGS_vertex_count); + { + CHECK(FLAGS_vertex_count % FLAGS_thread_count == 0) + << "Thread count must be a factor of vertex count"; + LOG(INFO) << "Generating " << FLAGS_vertex_count << " vertices..."; + utils::Timer timer; + auto vertex_ids = make_id_vector(FLAGS_vertex_count); + + std::vector threads; + SpinLock vertices_lock; + for (int i = 0; i < FLAGS_thread_count; ++i) { + threads.emplace_back([&db, &vertex_ids, &vertices, &vertices_lock, i]() { + GraphDbAccessor dba{db}; + auto label = dba.Label(kLabel); + auto property = dba.Property(kProperty); + auto batch_size = FLAGS_vertex_count / FLAGS_thread_count; + for (int j = i * batch_size; j < (i + 1) * batch_size; ++j) { + auto vertex = dba.InsertVertex(vertex_ids[j]); + vertex.add_label(label); + vertex.PropsSet(property, vertex_ids[j]); + vertices_lock.lock(); + vertices.emplace_back(vertex); + vertices_lock.unlock(); + } + dba.Commit(); + }); + } + for (auto &t : threads) t.join(); + LOG(INFO) << "Generated " << FLAGS_vertex_count << " vertices in " + << timer.Elapsed().count() << " seconds."; + } + { + GraphDbAccessor dba{db}; + for (int i = 0; i < FLAGS_vertex_count; ++i) + vertices[i] = *dba.Transfer(vertices[i]); + + LOG(INFO) << "Generating " << FLAGS_edge_count << " edges..."; + auto edge_ids = make_id_vector(FLAGS_edge_count); + std::mt19937 pseudo_rand_gen{std::random_device{}()}; + std::uniform_int_distribution<> rand_dist{0, FLAGS_vertex_count - 1}; + auto edge_type = dba.EdgeType("edge"); + utils::Timer timer; + for (int i = 0; i < FLAGS_edge_count; ++i) + dba.InsertEdge(vertices[rand_dist(pseudo_rand_gen)], + vertices[rand_dist(pseudo_rand_gen)], edge_type, + edge_ids[i]); + dba.Commit(); + LOG(INFO) << "Generated " << FLAGS_edge_count << " edges in " + << timer.Elapsed().count() << " seconds."; + } +} + +auto EdgeIteration(GraphDb &db) { + GraphDbAccessor dba{db}; + int64_t sum{0}; + for (auto edge : dba.Edges(false)) sum += edge.from().id() + edge.to().id(); + return sum; +} + +auto VertexIteration(GraphDb &db) { + GraphDbAccessor dba{db}; + int64_t sum{0}; + for (auto v : dba.Vertices(false)) + for (auto e : v.out()) sum += e.id() + e.to().id(); + return sum; +} + +auto ConnectedComponentsEdges(GraphDb &db) { + UnionFind connectivity{FLAGS_vertex_count}; + GraphDbAccessor dba{db}; + for (auto edge : dba.Edges(false)) + connectivity.Connect(edge.from().id(), edge.to().id()); + return connectivity.Size(); +} + +auto ConnectedComponentsVertices(GraphDb &db) { + UnionFind connectivity{FLAGS_vertex_count}; + GraphDbAccessor dba{db}; + for (auto from : dba.Vertices(false)) { + for (auto out_edge : from.out()) + connectivity.Connect(from.id(), out_edge.to().id()); + } + return connectivity.Size(); +} + +auto ConnectedComponentsVerticesParallel(GraphDb &db) { + UnionFind connectivity{FLAGS_vertex_count}; + SpinLock connectivity_lock; + + // Define bounds of vertex IDs for each thread to use. + std::vector bounds; + for (int64_t i = 0; i < FLAGS_thread_count; ++i) + bounds.emplace_back(i * FLAGS_vertex_count / FLAGS_thread_count); + bounds.emplace_back(std::numeric_limits::max()); + + std::vector threads; + for (int i = 0; i < FLAGS_thread_count; ++i) { + threads.emplace_back( + [&connectivity, &connectivity_lock, &bounds, &db, i]() { + GraphDbAccessor dba{db}; + for (auto from : + dba.Vertices(dba.Label(kLabel), dba.Property(kProperty), + utils::MakeBoundInclusive(bounds[i]), + utils::MakeBoundExclusive(bounds[i + 1]), false)) { + for (auto out_edge : from.out()) { + std::lock_guard lock{connectivity_lock}; + connectivity.Connect(from.id(), out_edge.to().id()); + } + } + }); + } + for (auto &t : threads) t.join(); + return connectivity.Size(); +} + +auto Expansion(GraphDb &db) { + std::vector component_ids(FLAGS_vertex_count, -1); + int next_component_id{0}; + std::stack expansion_stack; + GraphDbAccessor dba{db}; + for (auto v : dba.Vertices(false)) { + if (component_ids[v.id()] != -1) + continue; + auto component_id = next_component_id++; + expansion_stack.push(v); + while (!expansion_stack.empty()) { + auto next_v = expansion_stack.top(); + expansion_stack.pop(); + if (component_ids[next_v.id()] != -1) + continue; + component_ids[next_v.id()] = component_id; + for (auto e : next_v.out()) + expansion_stack.push(e.to()); + for (auto e : next_v.in()) + expansion_stack.push(e.from()); + } + } + + return next_component_id; +} + +int main(int argc, char **argv) { + gflags::ParseCommandLineFlags(&argc, &argv, true); + google::InitGoogleLogging(argv[0]); + FLAGS_gc_cycle_sec = -1; + + GraphDb db; + GenerateGraph(db); + auto timed_call = [&db](auto callable, const std::string &descr) { + LOG(INFO) << "Running " << descr << "..."; + utils::Timer timer; + auto result = callable(db); + LOG(INFO) << "\tDone in " << timer.Elapsed().count() + << " seconds, result: " << result; + }; + timed_call(EdgeIteration, "Edge iteration"); + timed_call(VertexIteration, "Vertex iteration"); + timed_call(ConnectedComponentsEdges, "Connected components - Edges"); + timed_call(ConnectedComponentsVertices, "Connected components - Vertices"); + timed_call(ConnectedComponentsVerticesParallel, + "Parallel connected components - Vertices"); + timed_call(Expansion, "Expansion"); + + return 0; +} diff --git a/poc/union_find.hpp b/src/data_structures/union_find.hpp similarity index 66% rename from poc/union_find.hpp rename to src/data_structures/union_find.hpp index 57df4861b..98e3fd92d 100644 --- a/poc/union_find.hpp +++ b/src/data_structures/union_find.hpp @@ -16,8 +16,8 @@ class UnionFind { * * @param n Number of elements in the data structure. */ - UnionFind(uintXX_t n) : set_count(n), rank(n), parent(n) { - for (auto i = 0; i < n; ++i) rank[i] = 0, parent[i] = i; + explicit UnionFind(uintXX_t n) : set_count_(n), rank_(n), parent_(n) { + for (auto i = 0; i < n; ++i) rank_[i] = 0, parent_[i] = i; } /** @@ -29,24 +29,24 @@ class UnionFind { * @param p First element. * @param q Second element. */ - void connect(uintXX_t p, uintXX_t q) { - auto rp = root(p); - auto rq = root(q); + void Connect(uintXX_t p, uintXX_t q) { + auto rp = Root(p); + auto rq = Root(q); // if roots are equal, we don't have to do anything if (rp == rq) return; // merge the subtree with the smaller rank to the root of the subtree with // the larger rank - if (rank[rp] < rank[rq]) - parent[rp] = rq; - else if (rank[rp] > rank[rq]) - parent[rq] = rp; + if (rank_[rp] < rank_[rq]) + parent_[rp] = rq; + else if (rank_[rp] > rank_[rq]) + parent_[rq] = rp; else - parent[rq] = rp, rank[rp] += 1; + parent_[rq] = rp, rank_[rp] += 1; // update the number of groups - set_count--; + set_count_--; } /** @@ -57,33 +57,33 @@ class UnionFind { * @param q Second element. * @return See above. */ - bool find(uintXX_t p, uintXX_t q) { return root(p) == root(q); } + bool Find(uintXX_t p, uintXX_t q) { return Root(p) == Root(q); } /** * Returns the number of disjoint sets in this UnionFind. * * @return See above. */ - uintXX_t size() const { return set_count; } + uintXX_t Size() const { return set_count_; } private: - uintXX_t set_count; + uintXX_t set_count_; // array of subtree ranks - std::vector rank; + std::vector rank_; // array of tree indices - std::vector parent; + std::vector parent_; - uintXX_t root(uintXX_t p) { + uintXX_t Root(uintXX_t p) { auto r = p; auto newp = p; // find the node connected to itself, that's the root - while (parent[r] != r) r = parent[r]; + while (parent_[r] != r) r = parent_[r]; // do some path compression to enable faster searches - while (p != r) newp = parent[p], parent[p] = r, p = newp; + while (p != r) newp = parent_[p], parent_[p] = r, p = newp; return r; } diff --git a/poc/union_find_test.cpp b/tests/unit/datastructure_union_find.cpp similarity index 60% rename from poc/union_find_test.cpp rename to tests/unit/datastructure_union_find.cpp index 42f021037..73c450662 100644 --- a/poc/union_find_test.cpp +++ b/tests/unit/datastructure_union_find.cpp @@ -3,41 +3,41 @@ #include "gtest/gtest.h" -#include "union_find.hpp" +#include "data_structures/union_find.hpp" void ExpectFully(UnionFind<> &uf, bool connected, int from = 0, int to = -1) { - if (to == -1) to = uf.size(); + if (to == -1) to = uf.Size(); for (int i = from; i < to; i++) for (int j = from; j < to; j++) - if (i != j) EXPECT_EQ(uf.find(i, j), connected); + if (i != j) EXPECT_EQ(uf.Find(i, j), connected); } TEST(UnionFindTest, InitialSizeTest) { for (int i = 0; i < 10; i++) { UnionFind<> uf(i); - EXPECT_EQ(i, uf.size()); + EXPECT_EQ(i, uf.Size()); } } TEST(UnionFindTest, ModifiedSizeTest) { UnionFind<> uf(10); - EXPECT_EQ(10, uf.size()); + EXPECT_EQ(10, uf.Size()); - uf.connect(0, 0); - EXPECT_EQ(10, uf.size()); + uf.Connect(0, 0); + EXPECT_EQ(10, uf.Size()); - uf.connect(0, 1); - EXPECT_EQ(9, uf.size()); + uf.Connect(0, 1); + EXPECT_EQ(9, uf.Size()); - uf.connect(2, 3); - EXPECT_EQ(8, uf.size()); + uf.Connect(2, 3); + EXPECT_EQ(8, uf.Size()); - uf.connect(0, 2); - EXPECT_EQ(7, uf.size()); + uf.Connect(0, 2); + EXPECT_EQ(7, uf.Size()); - uf.connect(1, 3); - EXPECT_EQ(7, uf.size()); + uf.Connect(1, 3); + EXPECT_EQ(7, uf.Size()); } TEST(UnionFindTest, Disconectivity) { @@ -47,7 +47,7 @@ TEST(UnionFindTest, Disconectivity) { TEST(UnionFindTest, ConnectivityAlongChain) { UnionFind<> uf(10); - for (unsigned int i = 1; i < uf.size(); i++) uf.connect(i - 1, i); + for (unsigned int i = 1; i < uf.Size(); i++) uf.Connect(i - 1, i); ExpectFully(uf, true); } @@ -55,23 +55,23 @@ TEST(UnionFindTest, ConnectivityOnTree) { UnionFind<> uf(10); ExpectFully(uf, false); - uf.connect(0, 1); - uf.connect(0, 2); + uf.Connect(0, 1); + uf.Connect(0, 2); ExpectFully(uf, true, 0, 3); ExpectFully(uf, false, 2); - uf.connect(2, 3); + uf.Connect(2, 3); ExpectFully(uf, true, 0, 4); ExpectFully(uf, false, 3); } TEST(UnionFindTest, DisjointChains) { UnionFind<> uf(30); - for (int i = 0; i < 30; i++) uf.connect(i, i % 10 == 0 ? i : i - 1); + for (int i = 0; i < 30; i++) uf.Connect(i, i % 10 == 0 ? i : i - 1); for (int i = 0; i < 30; i++) for (int j = 0; j < 30; j++) - EXPECT_EQ(uf.find(i, j), (j - (j % 10)) == (i - (i % 10))); + EXPECT_EQ(uf.Find(i, j), (j - (j % 10)) == (i - (i % 10))); } int main(int argc, char **argv) {