nix/libstore/remote-store.cc

   1 #include "serialise.hh"
   2 #include "util.hh"
   3 #include "remote-store.hh"
   4 #include "worker-protocol.hh"
   5 #include "archive.hh"
   6 #include "affinity.hh"
   7 #include "globals.hh"
   8
   9 #include <sys/types.h>
  10 #include <sys/stat.h>
  11 #include <sys/socket.h>
  12 #include <sys/un.h>
  13 #include <errno.h>
  14 #include <fcntl.h>
  15
  16 #include <iostream>
  17 #include <unistd.h>
  18 #include <cstring>
  19
  20 namespace nix {
  21
  22
  23 Path readStorePath(Source & from)
  24 {
  25     Path path = readString(from);
  26     assertStorePath(path);
  27     return path;
  28 }
  29
  30
  31 template<class T> T readStorePaths(Source & from)
  32 {
  33     T paths = readStrings<T>(from);
  34     foreach (typename T::iterator, i, paths) assertStorePath(*i);
  35     return paths;
  36 }
  37
  38 template PathSet readStorePaths(Source & from);
  39
  40
  41 RemoteStore::RemoteStore()
  42 {
  43     initialised = false;
  44 }
  45
  46
  47 void RemoteStore::openConnection(bool reserveSpace)
  48 {
  49     if (initialised) return;
  50     initialised = true;
  51
  52     string remoteMode = getEnv("NIX_REMOTE");
  53
  54     if (remoteMode == "daemon")
  55         /* Connect to a daemon that does the privileged work for
  56            us. */
  57         connectToDaemon();
  58     else
  59         throw Error(format("invalid setting for NIX_REMOTE, `%1%'") % remoteMode);
  60
  61     from.fd = fdSocket;
  62     to.fd = fdSocket;
  63
  64     /* Send the magic greeting, check for the reply. */
  65     try {
  66         writeInt(WORKER_MAGIC_1, to);
  67         to.flush();
  68         unsigned int magic = readInt(from);
  69         if (magic != WORKER_MAGIC_2) throw Error("protocol mismatch");
  70
  71         daemonVersion = readInt(from);
  72         if (GET_PROTOCOL_MAJOR(daemonVersion) != GET_PROTOCOL_MAJOR(PROTOCOL_VERSION))
  73             throw Error("Nix daemon protocol version not supported");
  74         writeInt(PROTOCOL_VERSION, to);
  75
  76         if (GET_PROTOCOL_MINOR(daemonVersion) >= 14) {
  77             int cpu = settings.lockCPU ? lockToCurrentCPU() : -1;
  78             if (cpu != -1) {
  79                 writeInt(1, to);
  80                 writeInt(cpu, to);
  81             } else
  82                 writeInt(0, to);
  83         }
  84
  85         if (GET_PROTOCOL_MINOR(daemonVersion) >= 11)
  86             writeInt(reserveSpace, to);
  87
  88         processStderr();
  89     }
  90     catch (Error & e) {
  91         throw Error(format("cannot start daemon worker: %1%") % e.msg());
  92     }
  93
  94     setOptions();
  95 }
  96
  97
  98 void RemoteStore::connectToDaemon()
  99 {
 100     fdSocket = socket(PF_UNIX, SOCK_STREAM, 0);
 101     if (fdSocket == -1)
 102         throw SysError("cannot create Unix domain socket");
 103     closeOnExec(fdSocket);
 104
 105     string socketPath = settings.nixDaemonSocketFile;
 106
 107     /* Urgh, sockaddr_un allows path names of only 108 characters.  So
 108        chdir to the socket directory so that we can pass a relative
 109        path name.  !!! this is probably a bad idea in multi-threaded
 110        applications... */
 111     AutoCloseFD fdPrevDir = open(".", O_RDONLY);
 112     if (fdPrevDir == -1) throw SysError("couldn't open current directory");
 113     if (chdir(dirOf(socketPath).c_str()) == -1) throw SysError(format("couldn't change to directory of ‘%1%’") % socketPath);
 114     Path socketPathRel = "./" + baseNameOf(socketPath);
 115
 116     struct sockaddr_un addr;
 117     addr.sun_family = AF_UNIX;
 118     if (socketPathRel.size() >= sizeof(addr.sun_path))
 119         throw Error(format("socket path `%1%' is too long") % socketPathRel);
 120     using namespace std;
 121     strcpy(addr.sun_path, socketPathRel.c_str());
 122
 123     if (connect(fdSocket, (struct sockaddr *) &addr, sizeof(addr)) == -1)
 124         throw SysError(format("cannot connect to daemon at `%1%'") % socketPath);
 125
 126     if (fchdir(fdPrevDir) == -1)
 127         throw SysError("couldn't change back to previous directory");
 128 }
 129
 130
 131 RemoteStore::~RemoteStore()
 132 {
 133     try {
 134         to.flush();
 135         fdSocket.close();
 136     } catch (...) {
 137         ignoreException();
 138     }
 139 }
 140
 141
 142 void RemoteStore::setOptions()
 143 {
 144     writeInt(wopSetOptions, to);
 145
 146     writeInt(settings.keepFailed, to);
 147     writeInt(settings.keepGoing, to);
 148     writeInt(settings.tryFallback, to);
 149     writeInt(verbosity, to);
 150     writeInt(settings.maxBuildJobs, to);
 151     writeInt(settings.maxSilentTime, to);
 152     if (GET_PROTOCOL_MINOR(daemonVersion) >= 2)
 153         writeInt(settings.useBuildHook, to);
 154     if (GET_PROTOCOL_MINOR(daemonVersion) >= 4) {
 155         writeInt(settings.buildVerbosity, to);
 156         writeInt(logType, to);
 157         writeInt(settings.printBuildTrace, to);
 158     }
 159     if (GET_PROTOCOL_MINOR(daemonVersion) >= 6)
 160         writeInt(settings.buildCores, to);
 161     if (GET_PROTOCOL_MINOR(daemonVersion) >= 10)
 162         writeInt(settings.useSubstitutes, to);
 163
 164     if (GET_PROTOCOL_MINOR(daemonVersion) >= 12) {
 165         Settings::SettingsMap overrides = settings.getOverrides();
 166         writeInt(overrides.size(), to);
 167         foreach (Settings::SettingsMap::iterator, i, overrides) {
 168             writeString(i->first, to);
 169             writeString(i->second, to);
 170         }
 171     }
 172
 173     processStderr();
 174 }
 175
 176
 177 bool RemoteStore::isValidPath(const Path & path)
 178 {
 179     openConnection();
 180     writeInt(wopIsValidPath, to);
 181     writeString(path, to);
 182     processStderr();
 183     unsigned int reply = readInt(from);
 184     return reply != 0;
 185 }
 186
 187
 188 PathSet RemoteStore::queryValidPaths(const PathSet & paths)
 189 {
 190     openConnection();
 191     if (GET_PROTOCOL_MINOR(daemonVersion) < 12) {
 192         PathSet res;
 193         foreach (PathSet::const_iterator, i, paths)
 194             if (isValidPath(*i)) res.insert(*i);
 195         return res;
 196     } else {
 197         writeInt(wopQueryValidPaths, to);
 198         writeStrings(paths, to);
 199         processStderr();
 200         return readStorePaths<PathSet>(from);
 201     }
 202 }
 203
 204
 205 PathSet RemoteStore::queryAllValidPaths()
 206 {
 207     openConnection();
 208     writeInt(wopQueryAllValidPaths, to);
 209     processStderr();
 210     return readStorePaths<PathSet>(from);
 211 }
 212
 213
 214 PathSet RemoteStore::querySubstitutablePaths(const PathSet & paths)
 215 {
 216     openConnection();
 217     if (GET_PROTOCOL_MINOR(daemonVersion) < 12) {
 218         PathSet res;
 219         foreach (PathSet::const_iterator, i, paths) {
 220             writeInt(wopHasSubstitutes, to);
 221             writeString(*i, to);
 222             processStderr();
 223             if (readInt(from)) res.insert(*i);
 224         }
 225         return res;
 226     } else {
 227         writeInt(wopQuerySubstitutablePaths, to);
 228         writeStrings(paths, to);
 229         processStderr();
 230         return readStorePaths<PathSet>(from);
 231     }
 232 }
 233
 234
 235 void RemoteStore::querySubstitutablePathInfos(const PathSet & paths,
 236     SubstitutablePathInfos & infos)
 237 {
 238     if (paths.empty()) return;
 239
 240     openConnection();
 241
 242     if (GET_PROTOCOL_MINOR(daemonVersion) < 3) return;
 243
 244     if (GET_PROTOCOL_MINOR(daemonVersion) < 12) {
 245
 246         foreach (PathSet::const_iterator, i, paths) {
 247             SubstitutablePathInfo info;
 248             writeInt(wopQuerySubstitutablePathInfo, to);
 249             writeString(*i, to);
 250             processStderr();
 251             unsigned int reply = readInt(from);
 252             if (reply == 0) continue;
 253             info.deriver = readString(from);
 254             if (info.deriver != "") assertStorePath(info.deriver);
 255             info.references = readStorePaths<PathSet>(from);
 256             info.downloadSize = readLongLong(from);
 257             info.narSize = GET_PROTOCOL_MINOR(daemonVersion) >= 7 ? readLongLong(from) : 0;
 258             infos[*i] = info;
 259         }
 260
 261     } else {
 262
 263         writeInt(wopQuerySubstitutablePathInfos, to);
 264         writeStrings(paths, to);
 265         processStderr();
 266         unsigned int count = readInt(from);
 267         for (unsigned int n = 0; n < count; n++) {
 268             Path path = readStorePath(from);
 269             SubstitutablePathInfo & info(infos[path]);
 270             info.deriver = readString(from);
 271             if (info.deriver != "") assertStorePath(info.deriver);
 272             info.references = readStorePaths<PathSet>(from);
 273             info.downloadSize = readLongLong(from);
 274             info.narSize = readLongLong(from);
 275         }
 276
 277     }
 278 }
 279
 280
 281 ValidPathInfo RemoteStore::queryPathInfo(const Path & path)
 282 {
 283     openConnection();
 284     writeInt(wopQueryPathInfo, to);
 285     writeString(path, to);
 286     processStderr();
 287     ValidPathInfo info;
 288     info.path = path;
 289     info.deriver = readString(from);
 290     if (info.deriver != "") assertStorePath(info.deriver);
 291     info.hash = parseHash(htSHA256, readString(from));
 292     info.references = readStorePaths<PathSet>(from);
 293     info.registrationTime = readInt(from);
 294     info.narSize = readLongLong(from);
 295     return info;
 296 }
 297
 298
 299 Hash RemoteStore::queryPathHash(const Path & path)
 300 {
 301     openConnection();
 302     writeInt(wopQueryPathHash, to);
 303     writeString(path, to);
 304     processStderr();
 305     string hash = readString(from);
 306     return parseHash(htSHA256, hash);
 307 }
 308
 309
 310 void RemoteStore::queryReferences(const Path & path,
 311     PathSet & references)
 312 {
 313     openConnection();
 314     writeInt(wopQueryReferences, to);
 315     writeString(path, to);
 316     processStderr();
 317     PathSet references2 = readStorePaths<PathSet>(from);
 318     references.insert(references2.begin(), references2.end());
 319 }
 320
 321
 322 void RemoteStore::queryReferrers(const Path & path,
 323     PathSet & referrers)
 324 {
 325     openConnection();
 326     writeInt(wopQueryReferrers, to);
 327     writeString(path, to);
 328     processStderr();
 329     PathSet referrers2 = readStorePaths<PathSet>(from);
 330     referrers.insert(referrers2.begin(), referrers2.end());
 331 }
 332
 333
 334 Path RemoteStore::queryDeriver(const Path & path)
 335 {
 336     openConnection();
 337     writeInt(wopQueryDeriver, to);
 338     writeString(path, to);
 339     processStderr();
 340     Path drvPath = readString(from);
 341     if (drvPath != "") assertStorePath(drvPath);
 342     return drvPath;
 343 }
 344
 345
 346 PathSet RemoteStore::queryValidDerivers(const Path & path)
 347 {
 348     openConnection();
 349     writeInt(wopQueryValidDerivers, to);
 350     writeString(path, to);
 351     processStderr();
 352     return readStorePaths<PathSet>(from);
 353 }
 354
 355
 356 PathSet RemoteStore::queryDerivationOutputs(const Path & path)
 357 {
 358     openConnection();
 359     writeInt(wopQueryDerivationOutputs, to);
 360     writeString(path, to);
 361     processStderr();
 362     return readStorePaths<PathSet>(from);
 363 }
 364
 365
 366 PathSet RemoteStore::queryDerivationOutputNames(const Path & path)
 367 {
 368     openConnection();
 369     writeInt(wopQueryDerivationOutputNames, to);
 370     writeString(path, to);
 371     processStderr();
 372     return readStrings<PathSet>(from);
 373 }
 374
 375
 376 Path RemoteStore::queryPathFromHashPart(const string & hashPart)
 377 {
 378     openConnection();
 379     writeInt(wopQueryPathFromHashPart, to);
 380     writeString(hashPart, to);
 381     processStderr();
 382     Path path = readString(from);
 383     if (!path.empty()) assertStorePath(path);
 384     return path;
 385 }
 386
 387
 388 Path RemoteStore::addToStore(const string & name, const Path & _srcPath,
 389     bool recursive, HashType hashAlgo, PathFilter & filter, bool repair)
 390 {
 391     if (repair) throw Error("repairing is not supported when building through the Nix daemon");
 392
 393     openConnection();
 394
 395     Path srcPath(absPath(_srcPath));
 396
 397     writeInt(wopAddToStore, to);
 398     writeString(name, to);
 399     /* backwards compatibility hack */
 400     writeInt((hashAlgo == htSHA256 && recursive) ? 0 : 1, to);
 401     writeInt(recursive ? 1 : 0, to);
 402     writeString(printHashType(hashAlgo), to);
 403
 404     try {
 405         to.written = 0;
 406         to.warn = true;
 407         dumpPath(srcPath, to, filter);
 408         to.warn = false;
 409         processStderr();
 410     } catch (SysError & e) {
 411         /* Daemon closed while we were sending the path. Probably OOM
 412            or I/O error. */
 413         if (e.errNo == EPIPE)
 414             try {
 415                 processStderr();
 416             } catch (EndOfFile & e) { }
 417         throw;
 418     }
 419
 420     return readStorePath(from);
 421 }
 422
 423
 424 Path RemoteStore::addTextToStore(const string & name, const string & s,
 425     const PathSet & references, bool repair)
 426 {
 427     if (repair) throw Error("repairing is not supported when building through the Nix daemon");
 428
 429     openConnection();
 430     writeInt(wopAddTextToStore, to);
 431     writeString(name, to);
 432     writeString(s, to);
 433     writeStrings(references, to);
 434
 435     processStderr();
 436     return readStorePath(from);
 437 }
 438
 439
 440 void RemoteStore::exportPath(const Path & path, bool sign,
 441     Sink & sink)
 442 {
 443     openConnection();
 444     writeInt(wopExportPath, to);
 445     writeString(path, to);
 446     writeInt(sign ? 1 : 0, to);
 447     processStderr(&sink); /* sink receives the actual data */
 448     readInt(from);
 449 }
 450
 451
 452 Paths RemoteStore::importPaths(bool requireSignature, Source & source)
 453 {
 454     openConnection();
 455     writeInt(wopImportPaths, to);
 456     /* We ignore requireSignature, since the worker forces it to true
 457        anyway. */
 458     processStderr(0, &source);
 459     return readStorePaths<Paths>(from);
 460 }
 461
 462
 463 void RemoteStore::buildPaths(const PathSet & drvPaths, BuildMode buildMode)
 464 {
 465     if (buildMode != bmNormal) throw Error("repairing or checking is not supported when building through the Nix daemon");
 466     openConnection();
 467     writeInt(wopBuildPaths, to);
 468     if (GET_PROTOCOL_MINOR(daemonVersion) >= 13)
 469         writeStrings(drvPaths, to);
 470     else {
 471         /* For backwards compatibility with old daemons, strip output
 472            identifiers. */
 473         PathSet drvPaths2;
 474         foreach (PathSet::const_iterator, i, drvPaths)
 475             drvPaths2.insert(string(*i, 0, i->find('!')));
 476         writeStrings(drvPaths2, to);
 477     }
 478     processStderr();
 479     readInt(from);
 480 }
 481
 482
 483 void RemoteStore::ensurePath(const Path & path)
 484 {
 485     openConnection();
 486     writeInt(wopEnsurePath, to);
 487     writeString(path, to);
 488     processStderr();
 489     readInt(from);
 490 }
 491
 492
 493 void RemoteStore::addTempRoot(const Path & path)
 494 {
 495     openConnection();
 496     writeInt(wopAddTempRoot, to);
 497     writeString(path, to);
 498     processStderr();
 499     readInt(from);
 500 }
 501
 502
 503 void RemoteStore::addIndirectRoot(const Path & path)
 504 {
 505     openConnection();
 506     writeInt(wopAddIndirectRoot, to);
 507     writeString(path, to);
 508     processStderr();
 509     readInt(from);
 510 }
 511
 512
 513 void RemoteStore::syncWithGC()
 514 {
 515     openConnection();
 516     writeInt(wopSyncWithGC, to);
 517     processStderr();
 518     readInt(from);
 519 }
 520
 521
 522 Roots RemoteStore::findRoots()
 523 {
 524     openConnection();
 525     writeInt(wopFindRoots, to);
 526     processStderr();
 527     unsigned int count = readInt(from);
 528     Roots result;
 529     while (count--) {
 530         Path link = readString(from);
 531         Path target = readStorePath(from);
 532         result[link] = target;
 533     }
 534     return result;
 535 }
 536
 537
 538 void RemoteStore::collectGarbage(const GCOptions & options, GCResults & results)
 539 {
 540     openConnection(false);
 541
 542     writeInt(wopCollectGarbage, to);
 543     writeInt(options.action, to);
 544     writeStrings(options.pathsToDelete, to);
 545     writeInt(options.ignoreLiveness, to);
 546     writeLongLong(options.maxFreed, to);
 547     writeInt(0, to);
 548     if (GET_PROTOCOL_MINOR(daemonVersion) >= 5) {
 549         /* removed options */
 550         writeInt(0, to);
 551         writeInt(0, to);
 552     }
 553
 554     processStderr();
 555
 556     results.paths = readStrings<PathSet>(from);
 557     results.bytesFreed = readLongLong(from);
 558     readLongLong(from); // obsolete
 559 }
 560
 561
 562 PathSet RemoteStore::queryFailedPaths()
 563 {
 564     openConnection();
 565     writeInt(wopQueryFailedPaths, to);
 566     processStderr();
 567     return readStorePaths<PathSet>(from);
 568 }
 569
 570
 571 void RemoteStore::clearFailedPaths(const PathSet & paths)
 572 {
 573     openConnection();
 574     writeInt(wopClearFailedPaths, to);
 575     writeStrings(paths, to);
 576     processStderr();
 577     readInt(from);
 578 }
 579
 580 void RemoteStore::optimiseStore()
 581 {
 582     openConnection();
 583     writeInt(wopOptimiseStore, to);
 584     processStderr();
 585     readInt(from);
 586 }
 587
 588 bool RemoteStore::verifyStore(bool checkContents, bool repair)
 589 {
 590     openConnection();
 591     writeInt(wopVerifyStore, to);
 592     writeInt(checkContents, to);
 593     writeInt(repair, to);
 594     processStderr();
 595     return readInt(from) != 0;
 596 }
 597
 598 void RemoteStore::processStderr(Sink * sink, Source * source)
 599 {
 600     to.flush();
 601     unsigned int msg;
 602     while ((msg = readInt(from)) == STDERR_NEXT
 603         || msg == STDERR_READ || msg == STDERR_WRITE) {
 604         if (msg == STDERR_WRITE) {
 605             string s = readString(from);
 606             if (!sink) throw Error("no sink");
 607             (*sink)((const unsigned char *) s.data(), s.size());
 608         }
 609         else if (msg == STDERR_READ) {
 610             if (!source) throw Error("no source");
 611             size_t len = readInt(from);
 612             unsigned char * buf = new unsigned char[len];
 613             AutoDeleteArray<unsigned char> d(buf);
 614             writeString(buf, source->read(buf, len), to);
 615             to.flush();
 616         }
 617         else {
 618             string s = readString(from);
 619             writeToStderr(s);
 620         }
 621     }
 622     if (msg == STDERR_ERROR) {
 623         string error = readString(from);
 624         unsigned int status = GET_PROTOCOL_MINOR(daemonVersion) >= 8 ? readInt(from) : 1;
 625         throw Error(format("%1%") % error, status);
 626     }
 627     else if (msg != STDERR_LAST)
 628         throw Error("protocol error processing standard error");
 629 }
 630
 631
 632 }