Merge branch 'master' into ungrafting

5 files changed, 148 insertions, 233 deletions

diff --git a/nix/libstore/build.cc b/nix/libstore/build.cc
index 8413819114..b5551b87ae 100644
--- a/nix/libstore/build.cc
+++ b/nix/libstore/build.cc
@@ -262,6 +262,7 @@ public:
     LocalStore & store;
 
     std::shared_ptr<Agent> hook;
+    std::shared_ptr<Agent> substituter;
 
     Worker(LocalStore & store);
     ~Worker();
@@ -2773,15 +2774,6 @@ private:
     /* Path info returned by the substituter's query info operation. */
     SubstitutablePathInfo info;
 
-    /* Pipe for the substituter's standard output. */
-    Pipe outPipe;
-
-    /* Pipe for the substituter's standard error. */
-    Pipe logPipe;
-
-    /* The process ID of the builder. */
-    Pid pid;
-
     /* Lock on the store path. */
     std::shared_ptr<PathLocks> outputLock;
 
@@ -2795,6 +2787,17 @@ private:
     typedef void (SubstitutionGoal::*GoalState)();
     GoalState state;
 
+    /* The substituter. */
+    std::shared_ptr<Agent> substituter;
+
+    /* Either the empty string, or the expected hash as returned by the
+       substituter.  */
+    string expectedHashStr;
+
+    /* Either the empty string, or the status phrase returned by the
+       substituter.  */
+    string status;
+
     void tryNext();
 
 public:
@@ -2840,7 +2843,7 @@ SubstitutionGoal::SubstitutionGoal(const Path & storePath, Worker & worker, bool
 
 SubstitutionGoal::~SubstitutionGoal()
 {
-    if (pid != -1) worker.childTerminated(pid);
+    if (substituter) worker.childTerminated(substituter->pid);
 }
 
 
@@ -2848,9 +2851,9 @@ void SubstitutionGoal::timedOut()
 {
     if (settings.printBuildTrace)
         printMsg(lvlError, format("@ substituter-failed %1% timeout") % storePath);
-    if (pid != -1) {
-        pid_t savedPid = pid;
-        pid.kill();
+    if (substituter) {
+        pid_t savedPid = substituter->pid;
+	substituter.reset();
         worker.childTerminated(savedPid);
     }
     amDone(ecFailed);
@@ -2977,44 +2980,29 @@ void SubstitutionGoal::tryToRun()
 
     printMsg(lvlInfo, format("fetching path `%1%'...") % storePath);
 
-    outPipe.create();
-    logPipe.create();
-
     destPath = repair ? storePath + ".tmp" : storePath;
 
     /* Remove the (stale) output path if it exists. */
     if (pathExists(destPath))
         deletePath(destPath);
 
-    worker.store.setSubstituterEnv();
-
-    /* Fill in the arguments. */
-    Strings args;
-    args.push_back("guix");
-    args.push_back("substitute");
-    args.push_back("--substitute");
-    args.push_back(storePath);
-    args.push_back(destPath);
-
-    /* Fork the substitute program. */
-    pid = startProcess([&]() {
-
-        commonChildInit(logPipe);
-
-        if (dup2(outPipe.writeSide, STDOUT_FILENO) == -1)
-            throw SysError("cannot dup output pipe into stdout");
+    if (!worker.substituter) {
+	const Strings args = { "substitute", "--substitute" };
+	const std::map<string, string> env = { { "_NIX_OPTIONS", settings.pack() } };
+	worker.substituter = std::make_shared<Agent>(settings.guixProgram, args, env);
+    }
 
-        execv(settings.guixProgram.c_str(), stringsToCharPtrs(args).data());
+    /* Borrow the worker's substituter.  */
+    if (!substituter) substituter.swap(worker.substituter);
 
-        throw SysError(format("executing `%1% substitute'") % settings.guixProgram);
-    });
+    /* Send the request to the substituter.  */
+    writeLine(substituter->toAgent.writeSide,
+	      (format("substitute %1% %2%") % storePath % destPath).str());
 
-    pid.setSeparatePG(true);
-    pid.setKillSignal(SIGTERM);
-    outPipe.writeSide.close();
-    logPipe.writeSide.close();
-    worker.childStarted(shared_from_this(),
-        pid, singleton<set<int> >(logPipe.readSide), true, true);
+    set<int> fds;
+    fds.insert(substituter->fromAgent.readSide);
+    fds.insert(substituter->builderOut.readSide);
+    worker.childStarted(shared_from_this(), substituter->pid, fds, true, true);
 
     state = &SubstitutionGoal::finished;
 
@@ -3029,54 +3017,51 @@ void SubstitutionGoal::finished()
 {
     trace("substitute finished");
 
-    /* Since we got an EOF on the logger pipe, the substitute is
-       presumed to have terminated.  */
-    pid_t savedPid = pid;
-    int status = pid.wait(true);
-
-    /* So the child is gone now. */
-    worker.childTerminated(savedPid);
+    /* Remove the 'guix substitute' process from the list of children.  */
+    worker.childTerminated(substituter->pid);
 
-    /* Close the read side of the logger pipe. */
-    logPipe.readSide.close();
-
-    /* Get the hash info from stdout. */
-    string dummy = readLine(outPipe.readSide);
-    string expectedHashStr = statusOk(status) ? readLine(outPipe.readSide) : "";
-    outPipe.readSide.close();
+    /* If max-jobs > 1, the worker might have created a new 'substitute'
+       process in the meantime.  If that is the case, terminate ours;
+       otherwise, give it back to the worker.  */
+    if (worker.substituter) {
+	substituter.reset ();
+    } else {
+	worker.substituter.swap(substituter);
+    }
 
     /* Check the exit status and the build result. */
     HashResult hash;
     try {
 
-        if (!statusOk(status))
-            throw SubstError(format("fetching path `%1%' %2%")
-                % storePath % statusToString(status));
+        if (status != "success")
+            throw SubstError(format("fetching path `%1%' (status: '%2%')")
+                % storePath % status);
 
         if (!pathExists(destPath))
             throw SubstError(format("substitute did not produce path `%1%'") % destPath);
 
+	if (expectedHashStr == "")
+	    throw SubstError(format("substituter did not communicate hash for `%1'") % storePath);
+
         hash = hashPath(htSHA256, destPath);
 
         /* Verify the expected hash we got from the substituer. */
-        if (expectedHashStr != "") {
-            size_t n = expectedHashStr.find(':');
-            if (n == string::npos)
-                throw Error(format("bad hash from substituter: %1%") % expectedHashStr);
-            HashType hashType = parseHashType(string(expectedHashStr, 0, n));
-            if (hashType == htUnknown)
-                throw Error(format("unknown hash algorithm in `%1%'") % expectedHashStr);
-            Hash expectedHash = parseHash16or32(hashType, string(expectedHashStr, n + 1));
-            Hash actualHash = hashType == htSHA256 ? hash.first : hashPath(hashType, destPath).first;
-            if (expectedHash != actualHash) {
-		if (settings.printBuildTrace)
-		    printMsg(lvlError, format("@ hash-mismatch %1% %2% %3% %4%")
-			     % storePath % "sha256"
-			     % printHash16or32(expectedHash)
-			     % printHash16or32(actualHash));
-                throw SubstError(format("hash mismatch for substituted item `%1%'") % storePath);
-	    }
-        }
+	size_t n = expectedHashStr.find(':');
+	if (n == string::npos)
+	    throw Error(format("bad hash from substituter: %1%") % expectedHashStr);
+	HashType hashType = parseHashType(string(expectedHashStr, 0, n));
+	if (hashType == htUnknown)
+	    throw Error(format("unknown hash algorithm in `%1%'") % expectedHashStr);
+	Hash expectedHash = parseHash16or32(hashType, string(expectedHashStr, n + 1));
+	Hash actualHash = hashType == htSHA256 ? hash.first : hashPath(hashType, destPath).first;
+	if (expectedHash != actualHash) {
+	    if (settings.printBuildTrace)
+		printMsg(lvlError, format("@ hash-mismatch %1% %2% %3% %4%")
+			 % storePath % "sha256"
+			 % printHash16or32(expectedHash)
+			 % printHash16or32(actualHash));
+	    throw SubstError(format("hash mismatch for substituted item `%1%'") % storePath);
+	}
 
     } catch (SubstError & e) {
 
@@ -3122,16 +3107,40 @@ void SubstitutionGoal::finished()
 
 void SubstitutionGoal::handleChildOutput(int fd, const string & data)
 {
-    assert(fd == logPipe.readSide);
-    if (verbosity >= settings.buildVerbosity) writeToStderr(data);
-    /* Don't write substitution output to a log file for now.  We
-       probably should, though. */
+    if (verbosity >= settings.buildVerbosity
+	&& fd == substituter->builderOut.readSide) {
+	writeToStderr(data);
+	/* Don't write substitution output to a log file for now.  We
+	   probably should, though. */
+    }
+
+    if (fd == substituter->fromAgent.readSide) {
+	/* DATA may consist of several lines.  Process them one by one.  */
+	string input = data;
+	while (!input.empty()) {
+	    /* Process up to the first newline.  */
+	    size_t end = input.find_first_of("\n");
+	    string trimmed = (end != string::npos) ? input.substr(0, end) : input;
+
+	    /* Update the goal's state accordingly.  */
+	    if (expectedHashStr == "") {
+		expectedHashStr = trimmed;
+	    } else if (status == "") {
+		status = trimmed;
+		worker.wakeUp(shared_from_this());
+	    } else {
+		printMsg(lvlError, format("unexpected substituter message '%1%'") % input);
+	    }
+
+	    input = (end != string::npos) ? input.substr(end + 1) : "";
+	}
+    }
 }
 
 
 void SubstitutionGoal::handleEOF(int fd)
 {
-    if (fd == logPipe.readSide) worker.wakeUp(shared_from_this());
+    worker.wakeUp(shared_from_this());
 }
 
 
diff --git a/nix/libstore/local-store.cc b/nix/libstore/local-store.cc
index 8c479002ec..c304e2ddd1 100644
--- a/nix/libstore/local-store.cc
+++ b/nix/libstore/local-store.cc
@@ -57,7 +57,6 @@ void checkStoreNotSymlink()
 
 
 LocalStore::LocalStore(bool reserveSpace)
-    : didSetSubstituterEnv(false)
 {
     schemaPath = settings.nixDBPath + "/schema";
 
@@ -183,21 +182,6 @@ LocalStore::LocalStore(bool reserveSpace)
 LocalStore::~LocalStore()
 {
     try {
-	if (runningSubstituter) {
-	    RunningSubstituter &i = *runningSubstituter;
-            if (!i.disabled) {
-		i.to.close();
-		i.from.close();
-		i.error.close();
-		if (i.pid != -1)
-		    i.pid.wait(true);
-	    }
-        }
-    } catch (...) {
-        ignoreException();
-    }
-
-    try {
         if (fdTempRoots != -1) {
             fdTempRoots.close();
             unlink(fnTempRoots.c_str());
@@ -796,96 +780,31 @@ Path LocalStore::queryPathFromHashPart(const string & hashPart)
     });
 }
 
-
-void LocalStore::setSubstituterEnv()
-{
-    if (didSetSubstituterEnv) return;
-
-    /* Pass configuration options (including those overridden with
-       --option) to substituters. */
-    setenv("_NIX_OPTIONS", settings.pack().c_str(), 1);
-
-    didSetSubstituterEnv = true;
-}
-
-
-void LocalStore::startSubstituter(RunningSubstituter & run)
-{
-    if (run.disabled || run.pid != -1) return;
-
-    debug(format("starting substituter program `%1% substitute'")
-	  % settings.guixProgram);
-
-    Pipe toPipe, fromPipe, errorPipe;
-
-    toPipe.create();
-    fromPipe.create();
-    errorPipe.create();
-
-    setSubstituterEnv();
-
-    run.pid = startProcess([&]() {
-        if (dup2(toPipe.readSide, STDIN_FILENO) == -1)
-            throw SysError("dupping stdin");
-        if (dup2(fromPipe.writeSide, STDOUT_FILENO) == -1)
-            throw SysError("dupping stdout");
-        if (dup2(errorPipe.writeSide, STDERR_FILENO) == -1)
-            throw SysError("dupping stderr");
-        execl(settings.guixProgram.c_str(), "guix", "substitute", "--query", NULL);
-        throw SysError(format("executing `%1%'") % settings.guixProgram);
-    });
-
-    run.to = toPipe.writeSide.borrow();
-    run.from = run.fromBuf.fd = fromPipe.readSide.borrow();
-    run.error = errorPipe.readSide.borrow();
-
-    toPipe.readSide.close();
-    fromPipe.writeSide.close();
-    errorPipe.writeSide.close();
-
-    /* The substituter may exit right away if it's disabled in any way
-       (e.g. copy-from-other-stores.pl will exit if no other stores
-       are configured). */
-    try {
-        getLineFromSubstituter(run);
-    } catch (EndOfFile & e) {
-        run.to.close();
-        run.from.close();
-        run.error.close();
-        run.disabled = true;
-        if (run.pid.wait(true) != 0) throw;
-    }
-}
-
-
 /* Read a line from the substituter's stdout, while also processing
    its stderr. */
-string LocalStore::getLineFromSubstituter(RunningSubstituter & run)
+string LocalStore::getLineFromSubstituter(Agent & run)
 {
     string res, err;
 
-    /* We might have stdout data left over from the last time. */
-    if (run.fromBuf.hasData()) goto haveData;
-
     while (1) {
         checkInterrupt();
 
         fd_set fds;
         FD_ZERO(&fds);
-        FD_SET(run.from, &fds);
-        FD_SET(run.error, &fds);
+        FD_SET(run.fromAgent.readSide, &fds);
+        FD_SET(run.builderOut.readSide, &fds);
 
         /* Wait for data to appear on the substituter's stdout or
            stderr. */
-        if (select(run.from > run.error ? run.from + 1 : run.error + 1, &fds, 0, 0, 0) == -1) {
+        if (select(std::max(run.fromAgent.readSide, run.builderOut.readSide) + 1, &fds, 0, 0, 0) == -1) {
             if (errno == EINTR) continue;
             throw SysError("waiting for input from the substituter");
         }
 
         /* Completely drain stderr before dealing with stdout. */
-        if (FD_ISSET(run.error, &fds)) {
+        if (FD_ISSET(run.builderOut.readSide, &fds)) {
             char buf[4096];
-            ssize_t n = read(run.error, (unsigned char *) buf, sizeof(buf));
+            ssize_t n = read(run.builderOut.readSide, (unsigned char *) buf, sizeof(buf));
             if (n == -1) {
                 if (errno == EINTR) continue;
                 throw SysError("reading from substituter's stderr");
@@ -903,23 +822,20 @@ string LocalStore::getLineFromSubstituter(RunningSubstituter & run)
         }
 
         /* Read from stdout until we get a newline or the buffer is empty. */
-        else if (run.fromBuf.hasData() || FD_ISSET(run.from, &fds)) {
-        haveData:
-            do {
-                unsigned char c;
-                run.fromBuf(&c, 1);
-                if (c == '\n') {
-                    if (!err.empty()) printMsg(lvlError, "substitute: " + err);
-                    return res;
-                }
-                res += c;
-            } while (run.fromBuf.hasData());
+        else if (FD_ISSET(run.fromAgent.readSide, &fds)) {
+	    unsigned char c;
+	    readFull(run.fromAgent.readSide, (unsigned char *) &c, 1);
+	    if (c == '\n') {
+		if (!err.empty()) printMsg(lvlError, "substitute: " + err);
+		return res;
+	    }
+	    res += c;
         }
     }
 }
 
 
-template<class T> T LocalStore::getIntLineFromSubstituter(RunningSubstituter & run)
+template<class T> T LocalStore::getIntLineFromSubstituter(Agent & run)
 {
     string s = getLineFromSubstituter(run);
     T res;
@@ -934,51 +850,47 @@ PathSet LocalStore::querySubstitutablePaths(const PathSet & paths)
 
     if (!settings.useSubstitutes || paths.empty()) return res;
 
-    if (!runningSubstituter) {
-	std::unique_ptr<RunningSubstituter>fresh(new RunningSubstituter);
-	runningSubstituter.swap(fresh);
-    }
+    Agent & run = *substituter();
 
-    RunningSubstituter & run = *runningSubstituter;
-    startSubstituter(run);
-
-    if (!run.disabled) {
-	string s = "have ";
-	foreach (PathSet::const_iterator, j, paths)
-	    if (res.find(*j) == res.end()) { s += *j; s += " "; }
-	writeLine(run.to, s);
-	while (true) {
-	    /* FIXME: we only read stderr when an error occurs, so
-	       substituters should only write (short) messages to
-	       stderr when they fail.  I.e. they shouldn't write debug
-	       output. */
-	    Path path = getLineFromSubstituter(run);
-	    if (path == "") break;
-	    res.insert(path);
-	}
+    string s = "have ";
+    foreach (PathSet::const_iterator, j, paths)
+	if (res.find(*j) == res.end()) { s += *j; s += " "; }
+    writeLine(run.toAgent.writeSide, s);
+    while (true) {
+	/* FIXME: we only read stderr when an error occurs, so
+	   substituters should only write (short) messages to
+	   stderr when they fail.  I.e. they shouldn't write debug
+	   output. */
+	Path path = getLineFromSubstituter(run);
+	if (path == "") break;
+	res.insert(path);
     }
 
     return res;
 }
 
 
-void LocalStore::querySubstitutablePathInfos(PathSet & paths, SubstitutablePathInfos & infos)
+std::shared_ptr<Agent> LocalStore::substituter()
 {
-    if (!settings.useSubstitutes) return;
-
     if (!runningSubstituter) {
-	std::unique_ptr<RunningSubstituter>fresh(new RunningSubstituter);
-	runningSubstituter.swap(fresh);
+	const Strings args = { "substitute", "--query" };
+	const std::map<string, string> env = { { "_NIX_OPTIONS", settings.pack() } };
+	runningSubstituter = std::make_shared<Agent>(settings.guixProgram, args, env);
     }
 
-    RunningSubstituter & run = *runningSubstituter;
-    startSubstituter(run);
-    if (run.disabled) return;
+    return runningSubstituter;
+}
+
+void LocalStore::querySubstitutablePathInfos(PathSet & paths, SubstitutablePathInfos & infos)
+{
+    if (!settings.useSubstitutes) return;
+
+    Agent & run = *substituter();
 
     string s = "info ";
     foreach (PathSet::const_iterator, i, paths)
         if (infos.find(*i) == infos.end()) { s += *i; s += " "; }
-    writeLine(run.to, s);
+    writeLine(run.toAgent.writeSide, s);
 
     while (true) {
         Path path = getLineFromSubstituter(run);
diff --git a/nix/libstore/local-store.hh b/nix/libstore/local-store.hh
index 2e48cf03e6..9ba37219da 100644
--- a/nix/libstore/local-store.hh
+++ b/nix/libstore/local-store.hh
@@ -38,21 +38,14 @@ struct OptimiseStats
 };
 
 
-struct RunningSubstituter
-{
-    Pid pid;
-    AutoCloseFD to, from, error;
-    FdSource fromBuf;
-    bool disabled;
-    RunningSubstituter() : disabled(false) { };
-};
-
-
 class LocalStore : public StoreAPI
 {
 private:
     /* The currently running substituter or empty.  */
-    std::unique_ptr<RunningSubstituter> runningSubstituter;
+    std::shared_ptr<Agent> runningSubstituter;
+
+    /* Ensure the substituter is running and return it.  */
+    std::shared_ptr<Agent> substituter();
 
     Path linksDir;
 
@@ -178,8 +171,6 @@ public:
 
     void markContentsGood(const Path & path);
 
-    void setSubstituterEnv();
-
     void createUser(const std::string & userName, uid_t userId);
 
 private:
@@ -213,8 +204,6 @@ private:
     /* Cache for pathContentsGood(). */
     std::map<Path, bool> pathContentsGoodCache;
 
-    bool didSetSubstituterEnv;
-
     /* The file to which we write our temporary roots. */
     Path fnTempRoots;
     AutoCloseFD fdTempRoots;
@@ -262,11 +251,9 @@ private:
 
     void removeUnusedLinks(const GCState & state);
 
-    void startSubstituter(RunningSubstituter & runningSubstituter);
-
-    string getLineFromSubstituter(RunningSubstituter & run);
+    string getLineFromSubstituter(Agent & run);
 
-    template<class T> T getIntLineFromSubstituter(RunningSubstituter & run);
+    template<class T> T getIntLineFromSubstituter(Agent & run);
 
     Path createTempDirInStore();
 
diff --git a/nix/libutil/util.cc b/nix/libutil/util.cc
index 59a2981359..69f1c634a9 100644
--- a/nix/libutil/util.cc
+++ b/nix/libutil/util.cc
@@ -1173,7 +1173,7 @@ void commonChildInit(Pipe & logPipe)
 
 //////////////////////////////////////////////////////////////////////
 
-Agent::Agent(const string &command, const Strings &args)
+Agent::Agent(const string &command, const Strings &args, const std::map<string, string> &env)
 {
     debug(format("starting agent '%1%'") % command);
 
@@ -1191,6 +1191,10 @@ Agent::Agent(const string &command, const Strings &args)
 
         commonChildInit(fromAgent);
 
+	for (auto pair: env) {
+	    setenv(pair.first.c_str(), pair.second.c_str(), 1);
+	}
+
         if (chdir("/") == -1) throw SysError("changing into `/");
 
         /* Dup the communication pipes. */
diff --git a/nix/libutil/util.hh b/nix/libutil/util.hh
index 13cff44316..880b0e93b2 100644
--- a/nix/libutil/util.hh
+++ b/nix/libutil/util.hh
@@ -7,6 +7,7 @@
 #include <dirent.h>
 #include <unistd.h>
 #include <signal.h>
+#include <map>
 #include <functional>
 
 #include <cstdio>
@@ -281,8 +282,10 @@ struct Agent
     /* The process ID of the agent. */
     Pid pid;
 
-    /* The command and arguments passed to the agent.  */
-    Agent(const string &command, const Strings &args);
+    /* The command and arguments passed to the agent along with a list of
+       environment variable name/value pairs.  */
+    Agent(const string &command, const Strings &args,
+	  const std::map<string, string> &env = std::map<string, string>());
 
     ~Agent();
 };