4 files changed, 76 insertions, 17 deletions
diff --git a/src/libutil/compression.cc b/src/libutil/compression.cc
index a3bbb5170d9f..8ffd55efb23c 100644
--- a/src/libutil/compression.cc
+++ b/src/libutil/compression.cc
@@ -89,6 +89,12 @@ static ref<std::string> decompressBzip2(const std::string & in)
     }
 }
 
+static ref<std::string> decompressBrotli(const std::string & in)
+{
+    // FIXME: use libbrotli
+    return make_ref<std::string>(runProgram(BRO, true, {"-d"}, {in}));
+}
+
 ref<std::string> compress(const std::string & method, const std::string & in)
 {
     StringSink ssink;
@@ -106,6 +112,8 @@ ref<std::string> decompress(const std::string & method, const std::string & in)
         return decompressXZ(in);
     else if (method == "bzip2")
         return decompressBzip2(in);
+    else if (method == "br")
+        return decompressBrotli(in);
     else
         throw UnknownCompressionMethod(format("unknown compression method ‘%s’") % method);
 }
@@ -139,7 +147,6 @@ struct XzSink : CompressionSink
 
     ~XzSink()
     {
-        assert(finished);
         lzma_end(&strm);
     }
 
@@ -210,7 +217,6 @@ struct BzipSink : CompressionSink
 
     ~BzipSink()
     {
-        assert(finished);
         BZ2_bzCompressEnd(&strm);
     }
 
@@ -261,6 +267,34 @@ struct BzipSink : CompressionSink
     }
 };
 
+struct BrotliSink : CompressionSink
+{
+    Sink & nextSink;
+    std::string data;
+
+    BrotliSink(Sink & nextSink) : nextSink(nextSink)
+    {
+    }
+
+    ~BrotliSink()
+    {
+    }
+
+    // FIXME: use libbrotli
+
+    void finish() override
+    {
+        flush();
+        nextSink(runProgram(BRO, true, {}, data));
+    }
+
+    void write(const unsigned char * data, size_t len) override
+    {
+        checkInterrupt();
+        this->data.append((const char *) data, len);
+    }
+};
+
 ref<CompressionSink> makeCompressionSink(const std::string & method, Sink & nextSink)
 {
     if (method == "none")
@@ -269,6 +303,8 @@ ref<CompressionSink> makeCompressionSink(const std::string & method, Sink & next
         return make_ref<XzSink>(nextSink);
     else if (method == "bzip2")
         return make_ref<BzipSink>(nextSink);
+    else if (method == "br")
+        return make_ref<BrotliSink>(nextSink);
     else
         throw UnknownCompressionMethod(format("unknown compression method ‘%s’") % method);
 }
diff --git a/src/libutil/local.mk b/src/libutil/local.mk
index cac5c8795db7..0721b21c2089 100644
--- a/src/libutil/local.mk
+++ b/src/libutil/local.mk
@@ -9,3 +9,5 @@ libutil_SOURCES := $(wildcard $(d)/*.cc)
 libutil_LDFLAGS = $(LIBLZMA_LIBS) -lbz2 -pthread $(OPENSSL_LIBS)
 
 libutil_LIBS = libformat
+
+libutil_CXXFLAGS = -DBRO=\"$(bro)\"
diff --git a/src/libutil/util.cc b/src/libutil/util.cc
index 0a5f796e4eaa..99a91c8cc64a 100644
--- a/src/libutil/util.cc
+++ b/src/libutil/util.cc
@@ -1,6 +1,7 @@
 #include "util.hh"
 #include "affinity.hh"
 #include "sync.hh"
+#include "finally.hh"
 
 #include <cctype>
 #include <cerrno>
@@ -10,6 +11,7 @@
 #include <iostream>
 #include <sstream>
 #include <thread>
+#include <future>
 
 #include <sys/wait.h>
 #include <unistd.h>
@@ -676,12 +678,11 @@ Pid::operator pid_t()
 }
 
 
-int Pid::kill(bool quiet)
+int Pid::kill()
 {
     assert(pid != -1);
 
-    if (!quiet)
-        printError(format("killing process %1%") % pid);
+    debug(format("killing process %1%") % pid);
 
     /* Send the requested signal to the child.  If it has its own
        process group, send the signal to every process in the child
@@ -837,23 +838,21 @@ std::vector<char *> stringsToCharPtrs(const Strings & ss)
 
 
 string runProgram(Path program, bool searchPath, const Strings & args,
-    const string & input)
+    const std::experimental::optional<std::string> & input)
 {
     checkInterrupt();
 
     /* Create a pipe. */
     Pipe out, in;
     out.create();
-    if (!input.empty()) in.create();
+    if (input) in.create();
 
     /* Fork. */
     Pid pid = startProcess([&]() {
         if (dup2(out.writeSide.get(), STDOUT_FILENO) == -1)
             throw SysError("dupping stdout");
-        if (!input.empty()) {
-            if (dup2(in.readSide.get(), STDIN_FILENO) == -1)
-                throw SysError("dupping stdin");
-        }
+        if (input && dup2(in.readSide.get(), STDIN_FILENO) == -1)
+            throw SysError("dupping stdin");
 
         Strings args_(args);
         args_.push_front(program);
@@ -870,11 +869,27 @@ string runProgram(Path program, bool searchPath, const Strings & args,
 
     out.writeSide = -1;
 
-    /* FIXME: This can deadlock if the input is too long. */
-    if (!input.empty()) {
+    std::thread writerThread;
+
+    std::promise<void> promise;
+
+    Finally doJoin([&]() {
+        if (writerThread.joinable())
+            writerThread.join();
+    });
+
+
+    if (input) {
         in.readSide = -1;
-        writeFull(in.writeSide.get(), input);
-        in.writeSide = -1;
+        writerThread = std::thread([&]() {
+            try {
+                writeFull(in.writeSide.get(), *input);
+                promise.set_value();
+            } catch (...) {
+                promise.set_exception(std::current_exception());
+            }
+            in.writeSide = -1;
+        });
     }
 
     string result = drainFD(out.readSide.get());
@@ -885,6 +900,9 @@ string runProgram(Path program, bool searchPath, const Strings & args,
         throw ExecError(status, format("program ‘%1%’ %2%")
             % program % statusToString(status));
 
+    /* Wait for the writer thread to finish. */
+    if (input) promise.get_future().get();
+
     return result;
 }
 
diff --git a/src/libutil/util.hh b/src/libutil/util.hh
index 7cb3e68b9ef1..645289f67c9a 100644
--- a/src/libutil/util.hh
+++ b/src/libutil/util.hh
@@ -13,6 +13,8 @@
 #include <limits>
 #include <cstdio>
 #include <map>
+#include <sstream>
+#include <experimental/optional>
 
 #ifndef HAVE_STRUCT_DIRENT_D_TYPE
 #define DT_UNKNOWN 0
@@ -201,7 +203,7 @@ public:
     ~Pid();
     void operator =(pid_t pid);
     operator pid_t();
-    int kill(bool quiet = false);
+    int kill();
     int wait();
 
     void setSeparatePG(bool separatePG);
@@ -231,7 +233,8 @@ pid_t startProcess(std::function<void()> fun, const ProcessOptions & options = P
 /* Run a program and return its stdout in a string (i.e., like the
    shell backtick operator). */
 string runProgram(Path program, bool searchPath = false,
-    const Strings & args = Strings(), const string & input = "");
+    const Strings & args = Strings(),
+    const std::experimental::optional<std::string> & input = {});
 
 class ExecError : public Error
 {