1 files changed, 602 insertions, 0 deletions
diff --git a/nix/libstore/remote-store.cc b/nix/libstore/remote-store.cc
new file mode 100644
index 0000000000..4619206932
--- /dev/null
+++ b/nix/libstore/remote-store.cc
@@ -0,0 +1,602 @@
+#include "serialise.hh"
+#include "util.hh"
+#include "remote-store.hh"
+#include "worker-protocol.hh"
+#include "archive.hh"
+#include "affinity.hh"
+#include "globals.hh"
+
+#include <sys/types.h>
+#include <sys/stat.h>
+#include <sys/socket.h>
+#include <sys/un.h>
+#include <fcntl.h>
+
+#include <iostream>
+#include <unistd.h>
+#include <cstring>
+
+namespace nix {
+
+
+Path readStorePath(Source & from)
+{
+    Path path = readString(from);
+    assertStorePath(path);
+    return path;
+}
+
+
+template<class T> T readStorePaths(Source & from)
+{
+    T paths = readStrings<T>(from);
+    foreach (typename T::iterator, i, paths) assertStorePath(*i);
+    return paths;
+}
+
+template PathSet readStorePaths(Source & from);
+
+
+RemoteStore::RemoteStore()
+{
+    initialised = false;
+}
+
+
+void RemoteStore::openConnection(bool reserveSpace)
+{
+    if (initialised) return;
+    initialised = true;
+
+    string remoteMode = getEnv("NIX_REMOTE");
+
+    if (remoteMode == "daemon")
+        /* Connect to a daemon that does the privileged work for
+           us. */
+        connectToDaemon();
+    else
+        throw Error(format("invalid setting for NIX_REMOTE, `%1%'") % remoteMode);
+
+    from.fd = fdSocket;
+    to.fd = fdSocket;
+
+    /* Send the magic greeting, check for the reply. */
+    try {
+        writeInt(WORKER_MAGIC_1, to);
+        to.flush();
+        unsigned int magic = readInt(from);
+        if (magic != WORKER_MAGIC_2) throw Error("protocol mismatch");
+
+        daemonVersion = readInt(from);
+        if (GET_PROTOCOL_MAJOR(daemonVersion) != GET_PROTOCOL_MAJOR(PROTOCOL_VERSION))
+            throw Error("Nix daemon protocol version not supported");
+        writeInt(PROTOCOL_VERSION, to);
+
+        if (GET_PROTOCOL_MINOR(daemonVersion) >= 14) {
+            int cpu = settings.lockCPU ? lockToCurrentCPU() : -1;
+            if (cpu != -1) {
+                writeInt(1, to);
+                writeInt(cpu, to);
+            } else
+                writeInt(0, to);
+        }
+
+        if (GET_PROTOCOL_MINOR(daemonVersion) >= 11)
+            writeInt(reserveSpace, to);
+
+        processStderr();
+    }
+    catch (Error & e) {
+        throw Error(format("cannot start worker (%1%)")
+            % e.msg());
+    }
+
+    setOptions();
+}
+
+
+void RemoteStore::connectToDaemon()
+{
+    fdSocket = socket(PF_UNIX, SOCK_STREAM, 0);
+    if (fdSocket == -1)
+        throw SysError("cannot create Unix domain socket");
+    closeOnExec(fdSocket);
+
+    string socketPath = settings.nixDaemonSocketFile;
+
+    /* Urgh, sockaddr_un allows path names of only 108 characters.  So
+       chdir to the socket directory so that we can pass a relative
+       path name.  !!! this is probably a bad idea in multi-threaded
+       applications... */
+    AutoCloseFD fdPrevDir = open(".", O_RDONLY);
+    if (fdPrevDir == -1) throw SysError("couldn't open current directory");
+    chdir(dirOf(socketPath).c_str());
+    Path socketPathRel = "./" + baseNameOf(socketPath);
+
+    struct sockaddr_un addr;
+    addr.sun_family = AF_UNIX;
+    if (socketPathRel.size() >= sizeof(addr.sun_path))
+        throw Error(format("socket path `%1%' is too long") % socketPathRel);
+    using namespace std;
+    strcpy(addr.sun_path, socketPathRel.c_str());
+
+    if (connect(fdSocket, (struct sockaddr *) &addr, sizeof(addr)) == -1)
+        throw SysError(format("cannot connect to daemon at `%1%'") % socketPath);
+
+    if (fchdir(fdPrevDir) == -1)
+        throw SysError("couldn't change back to previous directory");
+}
+
+
+RemoteStore::~RemoteStore()
+{
+    try {
+        to.flush();
+        fdSocket.close();
+        if (child != -1)
+            child.wait(true);
+    } catch (...) {
+        ignoreException();
+    }
+}
+
+
+void RemoteStore::setOptions()
+{
+    writeInt(wopSetOptions, to);
+
+    writeInt(settings.keepFailed, to);
+    writeInt(settings.keepGoing, to);
+    writeInt(settings.tryFallback, to);
+    writeInt(verbosity, to);
+    writeInt(settings.maxBuildJobs, to);
+    writeInt(settings.maxSilentTime, to);
+    if (GET_PROTOCOL_MINOR(daemonVersion) >= 2)
+        writeInt(settings.useBuildHook, to);
+    if (GET_PROTOCOL_MINOR(daemonVersion) >= 4) {
+        writeInt(settings.buildVerbosity, to);
+        writeInt(logType, to);
+        writeInt(settings.printBuildTrace, to);
+    }
+    if (GET_PROTOCOL_MINOR(daemonVersion) >= 6)
+        writeInt(settings.buildCores, to);
+    if (GET_PROTOCOL_MINOR(daemonVersion) >= 10)
+        writeInt(settings.useSubstitutes, to);
+
+    if (GET_PROTOCOL_MINOR(daemonVersion) >= 12) {
+        Settings::SettingsMap overrides = settings.getOverrides();
+        writeInt(overrides.size(), to);
+        foreach (Settings::SettingsMap::iterator, i, overrides) {
+            writeString(i->first, to);
+            writeString(i->second, to);
+        }
+    }
+
+    processStderr();
+}
+
+
+bool RemoteStore::isValidPath(const Path & path)
+{
+    openConnection();
+    writeInt(wopIsValidPath, to);
+    writeString(path, to);
+    processStderr();
+    unsigned int reply = readInt(from);
+    return reply != 0;
+}
+
+
+PathSet RemoteStore::queryValidPaths(const PathSet & paths)
+{
+    openConnection();
+    if (GET_PROTOCOL_MINOR(daemonVersion) < 12) {
+        PathSet res;
+        foreach (PathSet::const_iterator, i, paths)
+            if (isValidPath(*i)) res.insert(*i);
+        return res;
+    } else {
+        writeInt(wopQueryValidPaths, to);
+        writeStrings(paths, to);
+        processStderr();
+        return readStorePaths<PathSet>(from);
+    }
+}
+
+
+PathSet RemoteStore::queryAllValidPaths()
+{
+    openConnection();
+    writeInt(wopQueryAllValidPaths, to);
+    processStderr();
+    return readStorePaths<PathSet>(from);
+}
+
+
+PathSet RemoteStore::querySubstitutablePaths(const PathSet & paths)
+{
+    openConnection();
+    if (GET_PROTOCOL_MINOR(daemonVersion) < 12) {
+        PathSet res;
+        foreach (PathSet::const_iterator, i, paths) {
+            writeInt(wopHasSubstitutes, to);
+            writeString(*i, to);
+            processStderr();
+            if (readInt(from)) res.insert(*i);
+        }
+        return res;
+    } else {
+        writeInt(wopQuerySubstitutablePaths, to);
+        writeStrings(paths, to);
+        processStderr();
+        return readStorePaths<PathSet>(from);
+    }
+}
+
+
+void RemoteStore::querySubstitutablePathInfos(const PathSet & paths,
+    SubstitutablePathInfos & infos)
+{
+    if (paths.empty()) return;
+
+    openConnection();
+
+    if (GET_PROTOCOL_MINOR(daemonVersion) < 3) return;
+
+    if (GET_PROTOCOL_MINOR(daemonVersion) < 12) {
+
+        foreach (PathSet::const_iterator, i, paths) {
+            SubstitutablePathInfo info;
+            writeInt(wopQuerySubstitutablePathInfo, to);
+            writeString(*i, to);
+            processStderr();
+            unsigned int reply = readInt(from);
+            if (reply == 0) continue;
+            info.deriver = readString(from);
+            if (info.deriver != "") assertStorePath(info.deriver);
+            info.references = readStorePaths<PathSet>(from);
+            info.downloadSize = readLongLong(from);
+            info.narSize = GET_PROTOCOL_MINOR(daemonVersion) >= 7 ? readLongLong(from) : 0;
+            infos[*i] = info;
+        }
+
+    } else {
+
+        writeInt(wopQuerySubstitutablePathInfos, to);
+        writeStrings(paths, to);
+        processStderr();
+        unsigned int count = readInt(from);
+        for (unsigned int n = 0; n < count; n++) {
+            Path path = readStorePath(from);
+            SubstitutablePathInfo & info(infos[path]);
+            info.deriver = readString(from);
+            if (info.deriver != "") assertStorePath(info.deriver);
+            info.references = readStorePaths<PathSet>(from);
+            info.downloadSize = readLongLong(from);
+            info.narSize = readLongLong(from);
+        }
+
+    }
+}
+
+
+ValidPathInfo RemoteStore::queryPathInfo(const Path & path)
+{
+    openConnection();
+    writeInt(wopQueryPathInfo, to);
+    writeString(path, to);
+    processStderr();
+    ValidPathInfo info;
+    info.path = path;
+    info.deriver = readString(from);
+    if (info.deriver != "") assertStorePath(info.deriver);
+    info.hash = parseHash(htSHA256, readString(from));
+    info.references = readStorePaths<PathSet>(from);
+    info.registrationTime = readInt(from);
+    info.narSize = readLongLong(from);
+    return info;
+}
+
+
+Hash RemoteStore::queryPathHash(const Path & path)
+{
+    openConnection();
+    writeInt(wopQueryPathHash, to);
+    writeString(path, to);
+    processStderr();
+    string hash = readString(from);
+    return parseHash(htSHA256, hash);
+}
+
+
+void RemoteStore::queryReferences(const Path & path,
+    PathSet & references)
+{
+    openConnection();
+    writeInt(wopQueryReferences, to);
+    writeString(path, to);
+    processStderr();
+    PathSet references2 = readStorePaths<PathSet>(from);
+    references.insert(references2.begin(), references2.end());
+}
+
+
+void RemoteStore::queryReferrers(const Path & path,
+    PathSet & referrers)
+{
+    openConnection();
+    writeInt(wopQueryReferrers, to);
+    writeString(path, to);
+    processStderr();
+    PathSet referrers2 = readStorePaths<PathSet>(from);
+    referrers.insert(referrers2.begin(), referrers2.end());
+}
+
+
+Path RemoteStore::queryDeriver(const Path & path)
+{
+    openConnection();
+    writeInt(wopQueryDeriver, to);
+    writeString(path, to);
+    processStderr();
+    Path drvPath = readString(from);
+    if (drvPath != "") assertStorePath(drvPath);
+    return drvPath;
+}
+
+
+PathSet RemoteStore::queryValidDerivers(const Path & path)
+{
+    openConnection();
+    writeInt(wopQueryValidDerivers, to);
+    writeString(path, to);
+    processStderr();
+    return readStorePaths<PathSet>(from);
+}
+
+
+PathSet RemoteStore::queryDerivationOutputs(const Path & path)
+{
+    openConnection();
+    writeInt(wopQueryDerivationOutputs, to);
+    writeString(path, to);
+    processStderr();
+    return readStorePaths<PathSet>(from);
+}
+
+
+PathSet RemoteStore::queryDerivationOutputNames(const Path & path)
+{
+    openConnection();
+    writeInt(wopQueryDerivationOutputNames, to);
+    writeString(path, to);
+    processStderr();
+    return readStrings<PathSet>(from);
+}
+
+
+Path RemoteStore::queryPathFromHashPart(const string & hashPart)
+{
+    openConnection();
+    writeInt(wopQueryPathFromHashPart, to);
+    writeString(hashPart, to);
+    processStderr();
+    Path path = readString(from);
+    if (!path.empty()) assertStorePath(path);
+    return path;
+}
+
+
+Path RemoteStore::addToStore(const Path & _srcPath,
+    bool recursive, HashType hashAlgo, PathFilter & filter, bool repair)
+{
+    if (repair) throw Error("repairing is not supported when building through the Nix daemon");
+
+    openConnection();
+
+    Path srcPath(absPath(_srcPath));
+
+    writeInt(wopAddToStore, to);
+    writeString(baseNameOf(srcPath), to);
+    /* backwards compatibility hack */
+    writeInt((hashAlgo == htSHA256 && recursive) ? 0 : 1, to);
+    writeInt(recursive ? 1 : 0, to);
+    writeString(printHashType(hashAlgo), to);
+    dumpPath(srcPath, to, filter);
+    processStderr();
+    return readStorePath(from);
+}
+
+
+Path RemoteStore::addTextToStore(const string & name, const string & s,
+    const PathSet & references, bool repair)
+{
+    if (repair) throw Error("repairing is not supported when building through the Nix daemon");
+
+    openConnection();
+    writeInt(wopAddTextToStore, to);
+    writeString(name, to);
+    writeString(s, to);
+    writeStrings(references, to);
+
+    processStderr();
+    return readStorePath(from);
+}
+
+
+void RemoteStore::exportPath(const Path & path, bool sign,
+    Sink & sink)
+{
+    openConnection();
+    writeInt(wopExportPath, to);
+    writeString(path, to);
+    writeInt(sign ? 1 : 0, to);
+    processStderr(&sink); /* sink receives the actual data */
+    readInt(from);
+}
+
+
+Paths RemoteStore::importPaths(bool requireSignature, Source & source)
+{
+    openConnection();
+    writeInt(wopImportPaths, to);
+    /* We ignore requireSignature, since the worker forces it to true
+       anyway. */
+    processStderr(0, &source);
+    return readStorePaths<Paths>(from);
+}
+
+
+void RemoteStore::buildPaths(const PathSet & drvPaths, BuildMode buildMode)
+{
+    if (buildMode != bmNormal) throw Error("repairing or checking is not supported when building through the Nix daemon");
+    openConnection();
+    writeInt(wopBuildPaths, to);
+    if (GET_PROTOCOL_MINOR(daemonVersion) >= 13)
+        writeStrings(drvPaths, to);
+    else {
+        /* For backwards compatibility with old daemons, strip output
+           identifiers. */
+        PathSet drvPaths2;
+        foreach (PathSet::const_iterator, i, drvPaths)
+            drvPaths2.insert(string(*i, 0, i->find('!')));
+        writeStrings(drvPaths2, to);
+    }
+    processStderr();
+    readInt(from);
+}
+
+
+void RemoteStore::ensurePath(const Path & path)
+{
+    openConnection();
+    writeInt(wopEnsurePath, to);
+    writeString(path, to);
+    processStderr();
+    readInt(from);
+}
+
+
+void RemoteStore::addTempRoot(const Path & path)
+{
+    openConnection();
+    writeInt(wopAddTempRoot, to);
+    writeString(path, to);
+    processStderr();
+    readInt(from);
+}
+
+
+void RemoteStore::addIndirectRoot(const Path & path)
+{
+    openConnection();
+    writeInt(wopAddIndirectRoot, to);
+    writeString(path, to);
+    processStderr();
+    readInt(from);
+}
+
+
+void RemoteStore::syncWithGC()
+{
+    openConnection();
+    writeInt(wopSyncWithGC, to);
+    processStderr();
+    readInt(from);
+}
+
+
+Roots RemoteStore::findRoots()
+{
+    openConnection();
+    writeInt(wopFindRoots, to);
+    processStderr();
+    unsigned int count = readInt(from);
+    Roots result;
+    while (count--) {
+        Path link = readString(from);
+        Path target = readStorePath(from);
+        result[link] = target;
+    }
+    return result;
+}
+
+
+void RemoteStore::collectGarbage(const GCOptions & options, GCResults & results)
+{
+    openConnection(false);
+
+    writeInt(wopCollectGarbage, to);
+    writeInt(options.action, to);
+    writeStrings(options.pathsToDelete, to);
+    writeInt(options.ignoreLiveness, to);
+    writeLongLong(options.maxFreed, to);
+    writeInt(0, to);
+    if (GET_PROTOCOL_MINOR(daemonVersion) >= 5) {
+        /* removed options */
+        writeInt(0, to);
+        writeInt(0, to);
+    }
+
+    processStderr();
+
+    results.paths = readStrings<PathSet>(from);
+    results.bytesFreed = readLongLong(from);
+    readLongLong(from); // obsolete
+}
+
+
+PathSet RemoteStore::queryFailedPaths()
+{
+    openConnection();
+    writeInt(wopQueryFailedPaths, to);
+    processStderr();
+    return readStorePaths<PathSet>(from);
+}
+
+
+void RemoteStore::clearFailedPaths(const PathSet & paths)
+{
+    openConnection();
+    writeInt(wopClearFailedPaths, to);
+    writeStrings(paths, to);
+    processStderr();
+    readInt(from);
+}
+
+
+void RemoteStore::processStderr(Sink * sink, Source * source)
+{
+    to.flush();
+    unsigned int msg;
+    while ((msg = readInt(from)) == STDERR_NEXT
+        || msg == STDERR_READ || msg == STDERR_WRITE) {
+        if (msg == STDERR_WRITE) {
+            string s = readString(from);
+            if (!sink) throw Error("no sink");
+            (*sink)((const unsigned char *) s.data(), s.size());
+        }
+        else if (msg == STDERR_READ) {
+            if (!source) throw Error("no source");
+            size_t len = readInt(from);
+            unsigned char * buf = new unsigned char[len];
+            AutoDeleteArray<unsigned char> d(buf);
+            writeString(buf, source->read(buf, len), to);
+            to.flush();
+        }
+        else {
+            string s = readString(from);
+            writeToStderr(s);
+        }
+    }
+    if (msg == STDERR_ERROR) {
+        string error = readString(from);
+        unsigned int status = GET_PROTOCOL_MINOR(daemonVersion) >= 8 ? readInt(from) : 1;
+        throw Error(format("%1%") % error, status);
+    }
+    else if (msg != STDERR_LAST)
+        throw Error("protocol error processing standard error");
+}
+
+
+}