diff --git a/src/libutil/thread-pool.hh b/src/libutil/thread-pool.hh index 939bcf1ef..78b63467d 100644 --- a/src/libutil/thread-pool.hh +++ b/src/libutil/thread-pool.hh @@ -6,6 +6,7 @@ #include #include #include +#include namespace nix { @@ -54,4 +55,63 @@ private: 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 +void processGraph( + ThreadPool & pool, + const std::set & nodes, + std::function(const T &)> getEdges, + std::function processNode) +{ + struct Graph { + std::set left; + std::map> refs, rrefs; + std::function wrap; + }; + + ref> graph_ = make_ref>(); + + 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)); + } + } +} + } diff --git a/src/nix/copy.cc b/src/nix/copy.cc index 16b16910c..b5bd362d6 100644 --- a/src/nix/copy.cc +++ b/src/nix/copy.cc @@ -58,70 +58,33 @@ struct CmdCopy : StorePathsCommand progressBar.updateStatus(showProgress()); - struct Graph - { - std::set left; - std::map> refs, rrefs; - }; - - Sync graph_; - { - auto graph(graph_.lock()); - graph->left = PathSet(storePaths.begin(), storePaths.end()); - } - ThreadPool pool; - std::function doPath; + processGraph(pool, + PathSet(storePaths.begin(), storePaths.end()), - doPath = [&](const Path & storePath) { - checkInterrupt(); + [&](const Path & storePath) { + return srcStore->queryPathInfo(storePath)->references; + }, - if (!dstStore->isValidPath(storePath)) { - auto activity(progressBar.startActivity(format("copying ā€˜%sā€™...") % storePath)); + [&](const Path & storePath) { + checkInterrupt(); - StringSink sink; - srcStore->exportPaths({storePath}, false, sink); + if (!dstStore->isValidPath(storePath)) { + auto activity(progressBar.startActivity(format("copying ā€˜%sā€™...") % storePath)); - StringSource source(*sink.s); - dstStore->importPaths(false, source, 0); + StringSink sink; + srcStore->exportPaths({storePath}, false, sink); - done++; - } else - total--; + StringSource source(*sink.s); + dstStore->importPaths(false, source, 0); - progressBar.updateStatus(showProgress()); + done++; + } else + total--; - /* Enqueue all paths that were waiting for this one. */ - { - 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)); - } - } + progressBar.updateStatus(showProgress()); + }); pool.process();