aboutsummaryrefslogtreecommitdiff
path: root/src/nix/copy.cc
diff options
context:
space:
mode:
authorEelco Dolstra <eelco.dolstra@logicblox.com>2016-04-22 18:19:48 +0200
committerEelco Dolstra <eelco.dolstra@logicblox.com>2016-04-22 18:19:48 +0200
commit91539d305ff035d53a6de75f8af1ebbd7df4e622 (patch)
tree35fce6b0a0fdd684802019b3ef96b09f16ec9685 /src/nix/copy.cc
parentb2ce6fde5a46b0ebf32139cbbfe97294120f3a90 (diff)
nix copy: Parallelise
Diffstat (limited to 'src/nix/copy.cc')
-rw-r--r--src/nix/copy.cc71
1 files changed, 58 insertions, 13 deletions
diff --git a/src/nix/copy.cc b/src/nix/copy.cc
index 87e6f604a..16b16910c 100644
--- a/src/nix/copy.cc
+++ b/src/nix/copy.cc
@@ -3,6 +3,7 @@
#include "shared.hh"
#include "store-api.hh"
#include "sync.hh"
+#include "thread-pool.hh"
#include <atomic>
@@ -57,29 +58,73 @@ struct CmdCopy : StorePathsCommand
progressBar.updateStatus(showProgress());
- storePaths.reverse(); // FIXME: assumes reverse topo sort
+ struct Graph
+ {
+ std::set<Path> left;
+ std::map<Path, std::set<Path>> refs, rrefs;
+ };
- for (auto & storePath : storePaths) {
+ Sync<Graph> graph_;
+ {
+ auto graph(graph_.lock());
+ graph->left = PathSet(storePaths.begin(), storePaths.end());
+ }
+
+ ThreadPool pool;
+
+ std::function<void(const Path &)> doPath;
+
+ doPath = [&](const Path & storePath) {
checkInterrupt();
- if (dstStore->isValidPath(storePath)) {
- total--;
- progressBar.updateStatus(showProgress());
- continue;
- }
+ if (!dstStore->isValidPath(storePath)) {
+ auto activity(progressBar.startActivity(format("copying ā€˜%sā€™...") % storePath));
- auto activity(progressBar.startActivity(format("copying ā€˜%sā€™...") % storePath));
+ StringSink sink;
+ srcStore->exportPaths({storePath}, false, sink);
- StringSink sink;
- srcStore->exportPaths({storePath}, false, sink);
+ StringSource source(*sink.s);
+ dstStore->importPaths(false, source, 0);
- StringSource source(*sink.s);
- dstStore->importPaths(false, source, 0);
+ done++;
+ } else
+ total--;
- done++;
progressBar.updateStatus(showProgress());
+
+ /* 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));
+ }
}
+ pool.process();
+
progressBar.done();
}
};