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
This commit is contained in:
florijan
@@ -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)
|
||||
|
||||
2
customers/CMakeLists.txt
Normal file
2
customers/CMakeLists.txt
Normal file
@@ -0,0 +1,2 @@
|
||||
project(mg_customers)
|
||||
add_subdirectory(otto)
|
||||
3
customers/otto/CMakeLists.txt
Normal file
3
customers/otto/CMakeLists.txt
Normal file
@@ -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)
|
||||
209
customers/otto/parallel_connected_components.cpp
Normal file
209
customers/otto/parallel_connected_components.cpp
Normal file
@@ -0,0 +1,209 @@
|
||||
#include <algorithm>
|
||||
#include <limits>
|
||||
#include <mutex>
|
||||
#include <random>
|
||||
#include <set>
|
||||
#include <stack>
|
||||
#include <thread>
|
||||
|
||||
#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<int64_t> ids(size);
|
||||
for (size_t i = 0; i < size; ++i) ids[i] = i;
|
||||
std::random_shuffle(ids.begin(), ids.end());
|
||||
return ids;
|
||||
};
|
||||
|
||||
std::vector<VertexAccessor> 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<std::thread> 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<int64_t> 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<int64_t> 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<int64_t> connectivity{FLAGS_vertex_count};
|
||||
SpinLock connectivity_lock;
|
||||
|
||||
// Define bounds of vertex IDs for each thread to use.
|
||||
std::vector<PropertyValue> 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<int64_t>::max());
|
||||
|
||||
std::vector<std::thread> 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<SpinLock> 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<int> component_ids(FLAGS_vertex_count, -1);
|
||||
int next_component_id{0};
|
||||
std::stack<VertexAccessor> 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;
|
||||
}
|
||||
@@ -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<uintXX_t> rank;
|
||||
std::vector<uintXX_t> rank_;
|
||||
|
||||
// array of tree indices
|
||||
std::vector<uintXX_t> parent;
|
||||
std::vector<uintXX_t> 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;
|
||||
}
|
||||
@@ -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) {
|
||||
Reference in New Issue
Block a user