From ddaac8d0c06cbfdf4b1777603e97cedab0e54ead Mon Sep 17 00:00:00 2001 From: David Robillard Date: Fri, 17 Feb 2017 09:58:10 +0100 Subject: Fix and improve parallel traversal --- src/server/CompiledGraph.cpp | 134 ++++++++++++++++++++++++++++++------------- src/server/CompiledGraph.hpp | 9 ++- 2 files changed, 101 insertions(+), 42 deletions(-) (limited to 'src/server') diff --git a/src/server/CompiledGraph.cpp b/src/server/CompiledGraph.cpp index 5553140c..29b2ab9a 100644 --- a/src/server/CompiledGraph.cpp +++ b/src/server/CompiledGraph.cpp @@ -40,10 +40,22 @@ public: const BlockImpl* root; }; +static bool +has_provider_with_many_dependants(BlockImpl* n) +{ + for (BlockImpl* p : n->providers()) { + if (p->dependants().size() > 1) { + return true; + } + } + + return false; +} + CompiledGraph::CompiledGraph(GraphImpl* graph) : _log(graph->engine().log()) , _path(graph->path()) - , _master(new Task(Task::Mode::PARALLEL)) + , _master(new Task(Task::Mode::SEQUENTIAL)) { compile_graph(graph); } @@ -66,18 +78,31 @@ CompiledGraph::compile(Raul::Maid& maid, GraphImpl& graph) } } -void -CompiledGraph::compile_set(const std::set& blocks, - Task& task, - std::set& k) +static size_t +num_unvisited_dependants(BlockImpl* block) { - // Keep compiling working set until all nodes are visited - for (BlockImpl* block : blocks) { - // Each block is the start of a new sequential task - Task seq(Task::Mode::SEQUENTIAL); - compile_block(block, seq, k); - task.push_front(std::move(seq)); + size_t count = 0; + for (BlockImpl* b : block->dependants()) { + if (b->get_mark() == BlockImpl::Mark::UNVISITED) { + ++count; + } } + return count; +} + +size_t +parallel_depth(BlockImpl* block) +{ + if (has_provider_with_many_dependants(block)) { + return 2; + } + + size_t min_provider_depth = std::numeric_limits::max(); + for (auto p : block->providers()) { + min_provider_depth = std::min(min_provider_depth, parallel_depth(p)); + } + + return 2 + min_provider_depth; } void @@ -97,20 +122,31 @@ CompiledGraph::compile_graph(GraphImpl* graph) } } - // Keep compiling working set until all connected nodes are visited + // Keep compiling working set until all nodes are visited while (!blocks.empty()) { - std::set next; + std::set predecessors; + + // Calculate maximum sequential depth to consume this phase + size_t depth = std::numeric_limits::max(); + for (auto i : blocks) { + depth = std::min(depth, parallel_depth(i)); + } + + // fprintf(stderr, "WORKING SET {\n"); + // for (auto i = blocks.begin(); i != blocks.end(); ++i) { + // fprintf(stderr, "\t%s\n", (*i)->path().c_str()); + // } + // fprintf(stderr, "} DEPTH %zu\n", depth); - // The working set is a parallel task... Task par(Task::Mode::PARALLEL); - for (BlockImpl* block : blocks) { - // ... where each block is the tail of a new sequential task + for (auto b : blocks) { + assert(num_unvisited_dependants(b) == 0); Task seq(Task::Mode::SEQUENTIAL); - compile_block(block, seq, next); + compile_block(b, seq, depth, predecessors); par.push_front(std::move(seq)); } _master->push_front(std::move(par)); - blocks = next; + blocks = predecessors; } _master = Task::simplify(std::move(_master)); @@ -125,25 +161,25 @@ CompiledGraph::compile_graph(GraphImpl* graph) /** Throw a FeedbackException iff `dependant` has `root` as a dependency. */ static void -check_feedback(const BlockImpl* root, BlockImpl* dependant) +check_feedback(const BlockImpl* root, BlockImpl* provider) { - if (dependant == root) { + if (provider == root) { throw FeedbackException(root); } - for (auto& d : dependant->dependants()) { - const BlockImpl::Mark mark = d->get_mark(); + for (auto p : provider->providers()) { + const BlockImpl::Mark mark = p->get_mark(); switch (mark) { case BlockImpl::Mark::UNVISITED: - d->set_mark(BlockImpl::Mark::VISITING); - check_feedback(root, d); + p->set_mark(BlockImpl::Mark::VISITING); + check_feedback(root, p); break; case BlockImpl::Mark::VISITING: - throw FeedbackException(d, root); + throw FeedbackException(p, root); case BlockImpl::Mark::VISITED: break; } - d->set_mark(mark); + p->set_mark(mark); } } @@ -151,29 +187,39 @@ void CompiledGraph::compile_provider(const BlockImpl* root, BlockImpl* block, Task& task, + size_t max_depth, std::set& k) { if (block->dependants().size() > 1) { - /* Provider has other dependants, so this is the start of a sequential task. + /* Provider has other dependants, so this is the tail of a sequential task. Add provider to future working set and stop traversal. */ check_feedback(root, block); - k.insert(block); - } else { - // Provider has only this dependant, add here + if (num_unvisited_dependants(block) == 0) { + k.insert(block); + } + } else if (max_depth > 0) { + // Calling dependant has only this provider, add here if (task.mode() == Task::Mode::PARALLEL) { // Inside a parallel task, compile into a new sequential child Task seq(Task::Mode::SEQUENTIAL); - compile_block(block, seq, k); + compile_block(block, seq, max_depth, k); task.push_front(std::move(seq)); } else { - // Append to given sequential task - compile_block(block, task, k); + // Prepend to given sequential task + compile_block(block, task, max_depth, k); + } + } else { + if (num_unvisited_dependants(block) == 0) { + k.insert(block); } } } void -CompiledGraph::compile_block(BlockImpl* n, Task& task, std::set& k) +CompiledGraph::compile_block(BlockImpl* n, + Task& task, + size_t max_depth, + std::set& k) { switch (n->get_mark()) { case BlockImpl::Mark::UNVISITED: @@ -183,15 +229,23 @@ CompiledGraph::compile_block(BlockImpl* n, Task& task, std::set& k) task.push_front(Task(Task::Mode::SINGLE, n)); if (n->providers().size() < 2) { - // Single provider, append to this sequential task - for (auto& d : n->providers()) { - compile_provider(n, d, task, k); + // Single provider, prepend it to this sequential task + for (auto p : n->providers()) { + compile_provider(n, p, task, max_depth - 1, k); + } + } else if (has_provider_with_many_dependants(n)) { + // Stop recursion and enqueue providers for the next round + for (auto p : n->providers()) { + if (num_unvisited_dependants(p) == 0) { + k.insert(p); + } } } else { - // Multiple providers, create a new parallel task + // Multiple providers with only this node as dependant, + // make a new parallel task to execute them Task par(Task::Mode::PARALLEL); - for (auto& d : n->providers()) { - compile_provider(n, d, par, k); + for (auto p : n->providers()) { + compile_provider(n, p, par, max_depth - 1, k); } task.push_front(std::move(par)); } diff --git a/src/server/CompiledGraph.hpp b/src/server/CompiledGraph.hpp index db9aa288..0578c18a 100644 --- a/src/server/CompiledGraph.hpp +++ b/src/server/CompiledGraph.hpp @@ -61,11 +61,16 @@ private: typedef std::set BlockSet; void compile_graph(GraphImpl* graph); - void compile_set(const BlockSet& blocks, Task& task, BlockSet& k); - void compile_block(BlockImpl* block, Task& task, BlockSet& k); + + void compile_block(BlockImpl* block, + Task& task, + size_t max_depth, + BlockSet& k); + void compile_provider(const BlockImpl* root, BlockImpl* block, Task& task, + size_t max_depth, BlockSet& k); Log& _log; -- cgit v1.2.1