Factor out parallel processing of work items that have dependencies

This commit is contained in:
Eelco Dolstra 2016-04-22 20:50:06 +02:00
parent 91539d305f
commit c879a20850
2 changed files with 78 additions and 55 deletions

View file

@ -6,6 +6,7 @@
#include <queue> #include <queue>
#include <functional> #include <functional>
#include <thread> #include <thread>
#include <map>
namespace nix { namespace nix {
@ -54,4 +55,63 @@ private:
void workerEntry(); void workerEntry();
}; };
/* Process in parallel a set of items of type T that have a partial
ordering between them. Thus, any item is only processed after all
its dependencies have been processed. */
template<typename T>
void processGraph(
ThreadPool & pool,
const std::set<T> & nodes,
std::function<std::set<T>(const T &)> getEdges,
std::function<void(const T &)> processNode)
{
struct Graph {
std::set<T> left;
std::map<T, std::set<T>> refs, rrefs;
std::function<void(T)> wrap;
};
ref<Sync<Graph>> graph_ = make_ref<Sync<Graph>>();
auto wrapWork = [&pool, graph_, processNode](const T & node) {
processNode(node);
/* Enqueue work for all nodes that were waiting on this one. */
{
auto graph(graph_->lock());
graph->left.erase(node);
for (auto & rref : graph->rrefs[node]) {
auto & refs(graph->refs[rref]);
auto i = refs.find(node);
assert(i != refs.end());
refs.erase(i);
if (refs.empty())
pool.enqueue(std::bind(graph->wrap, rref));
}
}
};
{
auto graph(graph_->lock());
graph->left = nodes;
graph->wrap = wrapWork;
}
/* Build the dependency graph; enqueue all nodes with no
dependencies. */
for (auto & node : nodes) {
auto refs = getEdges(node);
{
auto graph(graph_->lock());
for (auto & ref : refs)
if (ref != node && graph->left.count(ref)) {
graph->refs[node].insert(ref);
graph->rrefs[ref].insert(node);
}
if (graph->refs[node].empty())
pool.enqueue(std::bind(graph->wrap, node));
}
}
}
} }

View file

@ -58,70 +58,33 @@ struct CmdCopy : StorePathsCommand
progressBar.updateStatus(showProgress()); progressBar.updateStatus(showProgress());
struct Graph
{
std::set<Path> left;
std::map<Path, std::set<Path>> refs, rrefs;
};
Sync<Graph> graph_;
{
auto graph(graph_.lock());
graph->left = PathSet(storePaths.begin(), storePaths.end());
}
ThreadPool pool; ThreadPool pool;
std::function<void(const Path &)> doPath; processGraph<Path>(pool,
PathSet(storePaths.begin(), storePaths.end()),
doPath = [&](const Path & storePath) { [&](const Path & storePath) {
checkInterrupt(); return srcStore->queryPathInfo(storePath)->references;
},
if (!dstStore->isValidPath(storePath)) { [&](const Path & storePath) {
auto activity(progressBar.startActivity(format("copying %s...") % storePath)); checkInterrupt();
StringSink sink; if (!dstStore->isValidPath(storePath)) {
srcStore->exportPaths({storePath}, false, sink); auto activity(progressBar.startActivity(format("copying %s...") % storePath));
StringSource source(*sink.s); StringSink sink;
dstStore->importPaths(false, source, 0); srcStore->exportPaths({storePath}, false, sink);
done++; StringSource source(*sink.s);
} else dstStore->importPaths(false, source, 0);
total--;
progressBar.updateStatus(showProgress()); done++;
} else
total--;
/* Enqueue all paths that were waiting for this one. */ progressBar.updateStatus(showProgress());
{ });
auto graph(graph_.lock());
graph->left.erase(storePath);
for (auto & rref : graph->rrefs[storePath]) {
auto & refs(graph->refs[rref]);
auto i = refs.find(storePath);
assert(i != refs.end());
refs.erase(i);
if (refs.empty())
pool.enqueue(std::bind(doPath, rref));
}
}
};
/* Build the dependency graph; enqueue all paths with no
dependencies. */
for (auto & storePath : storePaths) {
auto info = srcStore->queryPathInfo(storePath);
{
auto graph(graph_.lock());
for (auto & ref : info->references)
if (ref != storePath && graph->left.count(ref)) {
graph->refs[storePath].insert(ref);
graph->rrefs[ref].insert(storePath);
}
if (graph->refs[storePath].empty())
pool.enqueue(std::bind(doPath, storePath));
}
}
pool.process(); pool.process();