Reduce substitution memory consumption

copyStorePath() now pipes the output of srcStore->narFromPath()
directly into dstStore->addToStore(). The sink used by the former is
converted into a source usable by the latter using
boost::coroutine2. This is based on [1].

This reduces the maximum resident size of

  $ nix build --store ~/my-nix/ /nix/store/b0zlxla7dmy1iwc3g459rjznx59797xy-binutils-2.28.1 --substituters file:///tmp/binary-cache-xz/ --no-require-sigs

from 418592 KiB to 53416 KiB. (The previous commit also reduced the
runtime from ~4.2s to ~3.4s, not sure why.) A further improvement will
be to download files into a Sink.

[1] https://github.com/NixOS/nix/compare/master...Mathnerd314:dump-fix-coroutine#diff-dcbcac55a634031f9cc73707da6e4b18

Issue #1969.

1 files changed, 63 insertions, 0 deletions

diff --git a/src/libutil/serialise.cc b/src/libutil/serialise.cc
index 9e2a502afa..6b79392251 100644
--- a/src/libutil/serialise.cc
+++ b/src/libutil/serialise.cc
@@ -5,6 +5,8 @@
 #include <cerrno>
 #include <memory>
 
+#include <boost/coroutine2/coroutine.hpp>
+
 
 namespace nix {
 
@@ -88,6 +90,23 @@ void Source::operator () (unsigned char * data, size_t len)
 }
 
 
+std::string Source::drain()
+{
+    std::string s;
+    std::vector<unsigned char> buf(8192);
+    while (true) {
+        size_t n;
+        try {
+            n = read(buf.data(), buf.size());
+            s.append((char *) buf.data(), n);
+        } catch (EndOfFile &) {
+            break;
+        }
+    }
+    return s;
+}
+
+
 size_t BufferedSource::read(unsigned char * data, size_t len)
 {
     if (!buffer) buffer = decltype(buffer)(new unsigned char[bufSize]);
@@ -138,6 +157,50 @@ size_t StringSource::read(unsigned char * data, size_t len)
 }
 
 
+std::unique_ptr<Source> sinkToSource(std::function<void(Sink &)> fun)
+{
+    struct SinkToSource : Source
+    {
+        typedef boost::coroutines2::coroutine<std::string> coro_t;
+
+        coro_t::pull_type coro;
+
+        SinkToSource(std::function<void(Sink &)> fun)
+            : coro([&](coro_t::push_type & yield) {
+                LambdaSink sink([&](const unsigned char * data, size_t len) {
+                    if (len) yield(std::string((const char *) data, len));
+                });
+                fun(sink);
+            })
+        {
+        }
+
+        std::string cur;
+        size_t pos = 0;
+
+        size_t read(unsigned char * data, size_t len) override
+        {
+            if (!coro)
+                throw EndOfFile("coroutine has finished");
+
+            if (pos == cur.size()) {
+                if (!cur.empty()) coro();
+                cur = std::move(coro.get());
+                pos = 0;
+            }
+
+            auto n = std::min(cur.size() - pos, len);
+            memcpy(data, (unsigned char *) cur.data() + pos, n);
+            pos += n;
+
+            return n;
+        }
+    };
+
+    return std::make_unique<SinkToSource>(fun);
+}
+
+
 void writePadding(size_t len, Sink & sink)
 {
     if (len % 8) {