diff options
Diffstat (limited to 'src/libstore/remote-store.cc')
| -rw-r--r-- | src/libstore/remote-store.cc | 655 |
1 files changed, 655 insertions, 0 deletions
diff --git a/src/libstore/remote-store.cc b/src/libstore/remote-store.cc new file mode 100644 index 000000000..bc9ef3d47 --- /dev/null +++ b/src/libstore/remote-store.cc @@ -0,0 +1,655 @@ +#include "serialise.hh" +#include "util.hh" +#include "remote-store.hh" +#include "worker-protocol.hh" +#include "archive.hh" +#include "affinity.hh" +#include "globals.hh" +#include "derivations.hh" +#include "pool.hh" + +#include <sys/types.h> +#include <sys/stat.h> +#include <sys/socket.h> +#include <sys/un.h> +#include <errno.h> +#include <fcntl.h> +#include <unistd.h> + +#include <cstring> + +namespace nix { + + +Path readStorePath(Store & store, Source & from) +{ + Path path = readString(from); + store.assertStorePath(path); + return path; +} + + +template<class T> T readStorePaths(Store & store, Source & from) +{ + T paths = readStrings<T>(from); + for (auto & i : paths) store.assertStorePath(i); + return paths; +} + +template PathSet readStorePaths(Store & store, Source & from); +template Paths readStorePaths(Store & store, Source & from); + +/* TODO: Separate these store impls into different files, give them better names */ +RemoteStore::RemoteStore(const Params & params) + : Store(params) + , connections(make_ref<Pool<Connection>>( + std::max(1, (int) maxConnections), + [this]() { return openConnectionWrapper(); }, + [](const ref<Connection> & r) { return r->to.good() && r->from.good(); } + )) +{ +} + + +ref<RemoteStore::Connection> RemoteStore::openConnectionWrapper() +{ + if (failed) + throw Error("opening a connection to remote store ‘%s’ previously failed", getUri()); + try { + return openConnection(); + } catch (...) { + failed = true; + throw; + } +} + + +UDSRemoteStore::UDSRemoteStore(const Params & params) + : Store(params) + , LocalFSStore(params) + , RemoteStore(params) +{ +} + + +std::string UDSRemoteStore::getUri() +{ + return "daemon"; +} + + +ref<RemoteStore::Connection> UDSRemoteStore::openConnection() +{ + auto conn = make_ref<Connection>(); + + /* Connect to a daemon that does the privileged work for us. */ + conn->fd = socket(PF_UNIX, SOCK_STREAM + #ifdef SOCK_CLOEXEC + | SOCK_CLOEXEC + #endif + , 0); + if (!conn->fd) + throw SysError("cannot create Unix domain socket"); + closeOnExec(conn->fd.get()); + + string socketPath = settings.nixDaemonSocketFile; + + struct sockaddr_un addr; + addr.sun_family = AF_UNIX; + if (socketPath.size() + 1 >= sizeof(addr.sun_path)) + throw Error(format("socket path ‘%1%’ is too long") % socketPath); + strcpy(addr.sun_path, socketPath.c_str()); + + if (connect(conn->fd.get(), (struct sockaddr *) &addr, sizeof(addr)) == -1) + throw SysError(format("cannot connect to daemon at ‘%1%’") % socketPath); + + conn->from.fd = conn->fd.get(); + conn->to.fd = conn->fd.get(); + + initConnection(*conn); + + return conn; +} + + +void RemoteStore::initConnection(Connection & conn) +{ + /* Send the magic greeting, check for the reply. */ + try { + conn.to << WORKER_MAGIC_1; + conn.to.flush(); + unsigned int magic = readInt(conn.from); + if (magic != WORKER_MAGIC_2) throw Error("protocol mismatch"); + + conn.from >> conn.daemonVersion; + if (GET_PROTOCOL_MAJOR(conn.daemonVersion) != GET_PROTOCOL_MAJOR(PROTOCOL_VERSION)) + throw Error("Nix daemon protocol version not supported"); + if (GET_PROTOCOL_MINOR(conn.daemonVersion) < 10) + throw Error("the Nix daemon version is too old"); + conn.to << PROTOCOL_VERSION; + + if (GET_PROTOCOL_MINOR(conn.daemonVersion) >= 14) { + int cpu = settings.lockCPU ? lockToCurrentCPU() : -1; + if (cpu != -1) + conn.to << 1 << cpu; + else + conn.to << 0; + } + + if (GET_PROTOCOL_MINOR(conn.daemonVersion) >= 11) + conn.to << false; + + conn.processStderr(); + } + catch (Error & e) { + throw Error("cannot open connection to remote store ‘%s’: %s", getUri(), e.what()); + } + + setOptions(conn); +} + + +void RemoteStore::setOptions(Connection & conn) +{ + conn.to << wopSetOptions + << settings.keepFailed + << settings.keepGoing + << settings.tryFallback + << verbosity + << settings.maxBuildJobs + << settings.maxSilentTime + << settings.useBuildHook + << (settings.verboseBuild ? lvlError : lvlVomit) + << 0 // obsolete log type + << 0 /* obsolete print build trace */ + << settings.buildCores + << settings.useSubstitutes; + + if (GET_PROTOCOL_MINOR(conn.daemonVersion) >= 12) { + auto overrides = settings.getSettings(true); + conn.to << overrides.size(); + for (auto & i : overrides) + conn.to << i.first << i.second; + } + + conn.processStderr(); +} + + +bool RemoteStore::isValidPathUncached(const Path & path) +{ + auto conn(connections->get()); + conn->to << wopIsValidPath << path; + conn->processStderr(); + return readInt(conn->from); +} + + +PathSet RemoteStore::queryValidPaths(const PathSet & paths, bool maybeSubstitute) +{ + auto conn(connections->get()); + if (GET_PROTOCOL_MINOR(conn->daemonVersion) < 12) { + PathSet res; + for (auto & i : paths) + if (isValidPath(i)) res.insert(i); + return res; + } else { + conn->to << wopQueryValidPaths << paths; + conn->processStderr(); + return readStorePaths<PathSet>(*this, conn->from); + } +} + + +PathSet RemoteStore::queryAllValidPaths() +{ + auto conn(connections->get()); + conn->to << wopQueryAllValidPaths; + conn->processStderr(); + return readStorePaths<PathSet>(*this, conn->from); +} + + +PathSet RemoteStore::querySubstitutablePaths(const PathSet & paths) +{ + auto conn(connections->get()); + if (GET_PROTOCOL_MINOR(conn->daemonVersion) < 12) { + PathSet res; + for (auto & i : paths) { + conn->to << wopHasSubstitutes << i; + conn->processStderr(); + if (readInt(conn->from)) res.insert(i); + } + return res; + } else { + conn->to << wopQuerySubstitutablePaths << paths; + conn->processStderr(); + return readStorePaths<PathSet>(*this, conn->from); + } +} + + +void RemoteStore::querySubstitutablePathInfos(const PathSet & paths, + SubstitutablePathInfos & infos) +{ + if (paths.empty()) return; + + auto conn(connections->get()); + + if (GET_PROTOCOL_MINOR(conn->daemonVersion) < 12) { + + for (auto & i : paths) { + SubstitutablePathInfo info; + conn->to << wopQuerySubstitutablePathInfo << i; + conn->processStderr(); + unsigned int reply = readInt(conn->from); + if (reply == 0) continue; + info.deriver = readString(conn->from); + if (info.deriver != "") assertStorePath(info.deriver); + info.references = readStorePaths<PathSet>(*this, conn->from); + info.downloadSize = readLongLong(conn->from); + info.narSize = readLongLong(conn->from); + infos[i] = info; + } + + } else { + + conn->to << wopQuerySubstitutablePathInfos << paths; + conn->processStderr(); + size_t count = readNum<size_t>(conn->from); + for (size_t n = 0; n < count; n++) { + Path path = readStorePath(*this, conn->from); + SubstitutablePathInfo & info(infos[path]); + info.deriver = readString(conn->from); + if (info.deriver != "") assertStorePath(info.deriver); + info.references = readStorePaths<PathSet>(*this, conn->from); + info.downloadSize = readLongLong(conn->from); + info.narSize = readLongLong(conn->from); + } + + } +} + + +void RemoteStore::queryPathInfoUncached(const Path & path, + std::function<void(std::shared_ptr<ValidPathInfo>)> success, + std::function<void(std::exception_ptr exc)> failure) +{ + sync2async<std::shared_ptr<ValidPathInfo>>(success, failure, [&]() { + auto conn(connections->get()); + conn->to << wopQueryPathInfo << path; + try { + conn->processStderr(); + } catch (Error & e) { + // Ugly backwards compatibility hack. + if (e.msg().find("is not valid") != std::string::npos) + throw InvalidPath(e.what()); + throw; + } + if (GET_PROTOCOL_MINOR(conn->daemonVersion) >= 17) { + bool valid; conn->from >> valid; + if (!valid) throw InvalidPath(format("path ‘%s’ is not valid") % path); + } + auto info = std::make_shared<ValidPathInfo>(); + info->path = path; + info->deriver = readString(conn->from); + if (info->deriver != "") assertStorePath(info->deriver); + info->narHash = parseHash(htSHA256, readString(conn->from)); + info->references = readStorePaths<PathSet>(*this, conn->from); + conn->from >> info->registrationTime >> info->narSize; + if (GET_PROTOCOL_MINOR(conn->daemonVersion) >= 16) { + conn->from >> info->ultimate; + info->sigs = readStrings<StringSet>(conn->from); + conn->from >> info->ca; + } + return info; + }); +} + + +void RemoteStore::queryReferrers(const Path & path, + PathSet & referrers) +{ + auto conn(connections->get()); + conn->to << wopQueryReferrers << path; + conn->processStderr(); + PathSet referrers2 = readStorePaths<PathSet>(*this, conn->from); + referrers.insert(referrers2.begin(), referrers2.end()); +} + + +PathSet RemoteStore::queryValidDerivers(const Path & path) +{ + auto conn(connections->get()); + conn->to << wopQueryValidDerivers << path; + conn->processStderr(); + return readStorePaths<PathSet>(*this, conn->from); +} + + +PathSet RemoteStore::queryDerivationOutputs(const Path & path) +{ + auto conn(connections->get()); + conn->to << wopQueryDerivationOutputs << path; + conn->processStderr(); + return readStorePaths<PathSet>(*this, conn->from); +} + + +PathSet RemoteStore::queryDerivationOutputNames(const Path & path) +{ + auto conn(connections->get()); + conn->to << wopQueryDerivationOutputNames << path; + conn->processStderr(); + return readStrings<PathSet>(conn->from); +} + + +Path RemoteStore::queryPathFromHashPart(const string & hashPart) +{ + auto conn(connections->get()); + conn->to << wopQueryPathFromHashPart << hashPart; + conn->processStderr(); + Path path = readString(conn->from); + if (!path.empty()) assertStorePath(path); + return path; +} + + +void RemoteStore::addToStore(const ValidPathInfo & info, const ref<std::string> & nar, + bool repair, bool dontCheckSigs, std::shared_ptr<FSAccessor> accessor) +{ + auto conn(connections->get()); + + if (GET_PROTOCOL_MINOR(conn->daemonVersion) < 18) { + conn->to << wopImportPaths; + + StringSink sink; + sink << 1 // == path follows + ; + assert(nar->size() % 8 == 0); + sink((unsigned char *) nar->data(), nar->size()); + sink + << exportMagic + << info.path + << info.references + << info.deriver + << 0 // == no legacy signature + << 0 // == no path follows + ; + + StringSource source(*sink.s); + conn->processStderr(0, &source); + + auto importedPaths = readStorePaths<PathSet>(*this, conn->from); + assert(importedPaths.size() <= 1); + } + + else { + conn->to << wopAddToStoreNar + << info.path << info.deriver << printHash(info.narHash) + << info.references << info.registrationTime << info.narSize + << info.ultimate << info.sigs << info.ca + << repair << dontCheckSigs; + conn->to(*nar); + conn->processStderr(); + } +} + + +Path RemoteStore::addToStore(const string & name, 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"); + + auto conn(connections->get()); + + Path srcPath(absPath(_srcPath)); + + conn->to << wopAddToStore << name + << ((hashAlgo == htSHA256 && recursive) ? 0 : 1) /* backwards compatibility hack */ + << (recursive ? 1 : 0) + << printHashType(hashAlgo); + + try { + conn->to.written = 0; + conn->to.warn = true; + dumpPath(srcPath, conn->to, filter); + conn->to.warn = false; + conn->processStderr(); + } catch (SysError & e) { + /* Daemon closed while we were sending the path. Probably OOM + or I/O error. */ + if (e.errNo == EPIPE) + try { + conn->processStderr(); + } catch (EndOfFile & e) { } + throw; + } + + return readStorePath(*this, conn->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"); + + auto conn(connections->get()); + conn->to << wopAddTextToStore << name << s << references; + + conn->processStderr(); + return readStorePath(*this, conn->from); +} + + +void RemoteStore::buildPaths(const PathSet & drvPaths, BuildMode buildMode) +{ + auto conn(connections->get()); + conn->to << wopBuildPaths; + if (GET_PROTOCOL_MINOR(conn->daemonVersion) >= 13) { + conn->to << drvPaths; + if (GET_PROTOCOL_MINOR(conn->daemonVersion) >= 15) + conn->to << buildMode; + else + /* Old daemons did not take a 'buildMode' parameter, so we + need to validate it here on the client side. */ + if (buildMode != bmNormal) + throw Error("repairing or checking is not supported when building through the Nix daemon"); + } else { + /* For backwards compatibility with old daemons, strip output + identifiers. */ + PathSet drvPaths2; + for (auto & i : drvPaths) + drvPaths2.insert(string(i, 0, i.find('!'))); + conn->to << drvPaths2; + } + conn->processStderr(); + readInt(conn->from); +} + + +BuildResult RemoteStore::buildDerivation(const Path & drvPath, const BasicDerivation & drv, + BuildMode buildMode) +{ + auto conn(connections->get()); + conn->to << wopBuildDerivation << drvPath << drv << buildMode; + conn->processStderr(); + BuildResult res; + unsigned int status; + conn->from >> status >> res.errorMsg; + res.status = (BuildResult::Status) status; + return res; +} + + +void RemoteStore::ensurePath(const Path & path) +{ + auto conn(connections->get()); + conn->to << wopEnsurePath << path; + conn->processStderr(); + readInt(conn->from); +} + + +void RemoteStore::addTempRoot(const Path & path) +{ + auto conn(connections->get()); + conn->to << wopAddTempRoot << path; + conn->processStderr(); + readInt(conn->from); +} + + +void RemoteStore::addIndirectRoot(const Path & path) +{ + auto conn(connections->get()); + conn->to << wopAddIndirectRoot << path; + conn->processStderr(); + readInt(conn->from); +} + + +void RemoteStore::syncWithGC() +{ + auto conn(connections->get()); + conn->to << wopSyncWithGC; + conn->processStderr(); + readInt(conn->from); +} + + +Roots RemoteStore::findRoots() +{ + auto conn(connections->get()); + conn->to << wopFindRoots; + conn->processStderr(); + size_t count = readNum<size_t>(conn->from); + Roots result; + while (count--) { + Path link = readString(conn->from); + Path target = readStorePath(*this, conn->from); + result[link] = target; + } + return result; +} + + +void RemoteStore::collectGarbage(const GCOptions & options, GCResults & results) +{ + auto conn(connections->get()); + + conn->to + << wopCollectGarbage << options.action << options.pathsToDelete << options.ignoreLiveness + << options.maxFreed + /* removed options */ + << 0 << 0 << 0; + + conn->processStderr(); + + results.paths = readStrings<PathSet>(conn->from); + results.bytesFreed = readLongLong(conn->from); + readLongLong(conn->from); // obsolete + + { + auto state_(Store::state.lock()); + state_->pathInfoCache.clear(); + } +} + + +void RemoteStore::optimiseStore() +{ + auto conn(connections->get()); + conn->to << wopOptimiseStore; + conn->processStderr(); + readInt(conn->from); +} + + +bool RemoteStore::verifyStore(bool checkContents, bool repair) +{ + auto conn(connections->get()); + conn->to << wopVerifyStore << checkContents << repair; + conn->processStderr(); + return readInt(conn->from); +} + + +void RemoteStore::addSignatures(const Path & storePath, const StringSet & sigs) +{ + auto conn(connections->get()); + conn->to << wopAddSignatures << storePath << sigs; + conn->processStderr(); + readInt(conn->from); +} + + +void RemoteStore::queryMissing(const PathSet & targets, + PathSet & willBuild, PathSet & willSubstitute, PathSet & unknown, + unsigned long long & downloadSize, unsigned long long & narSize) +{ + { + auto conn(connections->get()); + if (GET_PROTOCOL_MINOR(conn->daemonVersion) < 19) + // Don't hold the connection handle in the fallback case + // to prevent a deadlock. + goto fallback; + conn->to << wopQueryMissing << targets; + conn->processStderr(); + willBuild = readStorePaths<PathSet>(*this, conn->from); + willSubstitute = readStorePaths<PathSet>(*this, conn->from); + unknown = readStorePaths<PathSet>(*this, conn->from); + conn->from >> downloadSize >> narSize; + return; + } + + fallback: + return Store::queryMissing(targets, willBuild, willSubstitute, + unknown, downloadSize, narSize); +} + + +RemoteStore::Connection::~Connection() +{ + try { + to.flush(); + } catch (...) { + ignoreException(); + } +} + + +void RemoteStore::Connection::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)(s); + } + else if (msg == STDERR_READ) { + if (!source) throw Error("no source"); + size_t len = readNum<size_t>(from); + auto buf = std::make_unique<unsigned char[]>(len); + writeString(buf.get(), source->read(buf.get(), len), to); + to.flush(); + } + else + printError(chomp(readString(from))); + } + if (msg == STDERR_ERROR) { + string error = readString(from); + unsigned int status = readInt(from); + throw Error(status, error); + } + else if (msg != STDERR_LAST) + throw Error("protocol error processing standard error"); +} + + +} |