Sync with Nix

Dependencies

• [2] RKW3XA4U Fix potential race in dispatcher wakeup
• [3] Y5LMEC35 BinaryCacheStore: Respect build-use-substitutes
• [4] V2UCCYN3 hydra-queue-runner: Get store mode configuration from hydra.conf
• [5] 7LB6QBXY Keep track of the number of build steps that are being built
• [6] MHVIT4JY Split hydra-queue-runner.cc more
• [7] O3NM62IZ Support multiple machines files
• [8] K5G5GZY7 Guard against concurrent invocations of hydra-queue-runner
• [9] W2AOTSS6 Rename class
• [10] ENXUSMSV Make concurrency more robust
• [11] ACBS7C6Q hydra-queue-runner: Detect changes to the scheduling shares
• [12] OG3Z3QGC Namespace cleanup
• [13] N2NKSKHS Refactor local binary cache code into a subclass
• [14] GJV2J5HX Pool local store connections
• [15] ATJ54SPX Use PostgreSQL notifications for queue events
• [16] HJOEIMLR Refactor
• [17] 46ADBTMQ Start steps in order of ascending build ID
• [18] TTBLPQAJ Keep track of wait time per system type
• [19] N4IROACV Move buildRemote() into State
• [20] 73YR46NJ hydra-queue-runner: Write directly to a binary cache
• [21] GTUZLZRH Add an S3-backed binary cache store
• [22] YR2IM6Y5 Temporarily disable machines after a connection failure
• [23] CNLNT3T4 Allow only 1 thread to send a closure to a given machine at the same time
• [24] IK2UBDAU Revive jobset scheduling
• [25] YZAI5GQU Implement a database connection pool
• [26] MB3TISH2 Rate-limit the number of threads copying closures at the same time
• [27] 24BMQDZA Start of single-process hydra-queue-runner
• [28] 5AIYUMTB Basic remote building
• [29] DWFTK56E Keep track of how many threads are waiting
• [30] RND7XFNH getQueuedBuilds(): Periodically stop to handle priority bumps
• [31] EOO4EFWD Use a single BinaryCacheStore for all threads
• [32] QSBS2ISO S3BinaryCacheStore::isValidPath(): Do a GET instead of HEAD
• [33] XLYHZUHT Cache .narinfo lookups
• [34] EYR3EW6J Keep stats for the Hydra auto scaler
• [35] 32HHP5CW hydra-queue-runner: Support generating a signed binary cache
• [36] SOB276BA Keep some statistics for the binary cache stores
• [37] HPJKBFZ4 Handle concurrent finishing of the same build
• [38] GH4S4AWM Rename file

Change contents

• file deletion: pool.hh (----------)
  [5.187]→[5.2674:2705](∅→∅),[5.2705]→[5.889:889](∅→∅)
  − #pragma once

  −

  − #include <memory>

  − #include <list>

  −

  − #include "sync.hh"

  −

  − /* This template class implements a simple pool manager of resources

  − of some type R, such as database connections. It is used as

  − follows:

  −

  − class Connection { ... };

  −

  − Pool<Connection> pool;

  −

  − {

  − auto conn(pool.get());

  − conn->exec("select ...");

  − }

  −

  − Here, the Connection object referenced by ‘conn’ is automatically

  − returned to the pool when ‘conn’ goes out of scope.

  − */

  −

  − template <class R>

  − class Pool

  − {

  − private:

  − struct State

  − {

  − unsigned int count = 0;

  − std::list<std::shared_ptr<R>> idle;

  − };

  −

  − Sync<State> state;

  −

  − public:

  −

  − class Handle

  − {

  − private:

  − Pool & pool;

  − std::shared_ptr<R> r;

  −

  − friend Pool;

  −

  − Handle(Pool & pool, std::shared_ptr<R> r) : pool(pool), r(r) { }

  −

  − public:

  − Handle(Handle && h) : pool(h.pool), r(h.r) { h.r.reset(); }

  −

  − Handle(const Handle & l) = delete;

  −

  − ~Handle()

  − {

  − auto state_(pool.state.lock());

  − if (r) state_->idle.push_back(r);

  − }

  −

  − R * operator -> () { return r.get(); }

  − R & operator * () { return *r; }

  − };

  −

  − Handle get()

  − {

  − {

  − auto state_(state.lock());

  − if (!state_->idle.empty()) {

  − auto p = state_->idle.back();

  − state_->idle.pop_back();

  − return Handle(*this, p);

  − }

  − state_->count++;

  − }

  − /* Note: we don't hold the lock while creating a new instance,

  − because creation might take a long time. */

  − }

  −

  − unsigned int count()

  − {

  − auto state_(state.lock());

  − return state_->count;

  − }

  − };

  − return Handle(*this, factory());

  − Pool(const Factory & factory = []() { return std::make_shared<R>(); })

  − : factory(factory)

  − { }

  −

  −

  − Factory factory;

  −

  − public:

  −

  − typedef std::function<std::shared_ptr<R>()> Factory;

  −

  − #include <functional>
• file deletion: sync.hh (----------)
  [5.187]→[5.10782:10813](∅→∅),[5.10813]→[5.9720:9720](∅→∅)
  − #pragma once

  −

  − #include <mutex>

  − #include <condition_variable>

  −

  − /* This template class ensures synchronized access to a value of type

  − T. It is used as follows:

  −

  − struct Data { int x; ... };

  −

  − Sync<Data> data;

  −

  − {

  − auto data_(data.lock());

  − data_->x = 123;

  − }

  −

  − Here, "data" is automatically unlocked when "data_" goes out of

  − scope.

  − */

  −

  − class Sync

  − {

  − private:

  − std::mutex mutex;

  − T data;

  −

  − public:

  −

  − class Lock

  − {

  − private:

  − Sync * s;

  − friend Sync;

  − Lock(Sync * s) : s(s) { s->mutex.lock(); }

  − public:

  − Lock(Lock && l) : s(l.s) { l.s = 0; }

  − Lock(const Lock & l) = delete;

  − ~Lock() { if (s) s->mutex.unlock(); }

  − T * operator -> () { return &s->data; }

  − T & operator * () { return s->data; }

  −

  − /* FIXME: performance impact of condition_variable_any? */

  − void wait(std::condition_variable_any & cv)

  − {

  − assert(s);

  − cv.wait(s->mutex);

  − }

  − };

  −

  − Lock lock() { return Lock(this); }

  − };

  − }

  −

  − template<class Rep, class Period, class Predicate>

  − bool wait_for(std::condition_variable_any & cv,

  − const std::chrono::duration<Rep, Period> & duration,

  − Predicate pred)

  − {

  − assert(s);

  − return cv.wait_for(s->mutex, duration, pred);

  − }

  −

  − template<class Clock, class Duration>

  − std::cv_status wait_until(std::condition_variable_any & cv,

  − const std::chrono::time_point<Clock, Duration> & duration)

  − {

  − assert(s);

  − return cv.wait_until(s->mutex, duration);

  − Sync() { }

  − Sync(const T & data) : data(data) { }

  −

  − template<class T>

  − #include <cassert>
• replacement in src/hydra-queue-runner/binary-cache-store.cc at line 15
  [5.2032]→[5.0:70](∅→∅)
  − BinaryCacheStore::BinaryCacheStore(const StoreFactory & storeFactory,
  [5.2032]

  [5.307]
  + BinaryCacheStore::BinaryCacheStore(std::shared_ptr<Store> localStore,
• replacement in src/hydra-queue-runner/binary-cache-store.cc at line 17
  [5.367]→[5.71:104](∅→∅)
  − : storeFactory(storeFactory)
  [5.367]

  [5.2180]
  + : localStore(localStore)
• replacement in src/hydra-queue-runner/binary-cache-store.cc at line 240
  [5.106]→[5.106:144](∅→∅)
  − auto localStore = storeFactory();
  [5.106]

  [5.8015]
  + if (!localStore) return;
• replacement in src/hydra-queue-runner/binary-cache-store.cc at line 243
  [5.8053]→[5.0:54](∅→∅)
  − if (!(*localStore)->isValidPath(storePath)) {
  [5.8053]

  [5.8104]
  + if (!localStore->isValidPath(storePath)) {
• replacement in src/hydra-queue-runner/binary-cache-store.cc at line 247
  [5.8172]→[5.55:125](∅→∅)
  − ValidPathInfo info = (*localStore)->queryPathInfo(storePath);
  [5.8172]

  [5.8239]
  + ValidPathInfo info = localStore->queryPathInfo(storePath);
• replacement in src/hydra-queue-runner/binary-cache-store.cc at line 256
  [3.33]→[3.33:98](∅→∅)
  − (*localStore)->querySubstitutablePathInfos(left, infos);
  [3.33]

  [5.8486]
  + localStore->querySubstitutablePathInfos(left, infos);
• edit in src/hydra-queue-runner/binary-cache-store.cc at line 261
  [5.8569]→[5.145:184](∅→∅)
  − auto localStore = storeFactory();

  −
• replacement in src/hydra-queue-runner/binary-cache-store.cc at line 266
  [5.8696]→[5.190:237](∅→∅)
  − (*localStore)->addTempRoot(storePath);
  [5.8696]

  [5.8740]
  + if (!localStore)

  + throw Error(format("don't know how to realise path ‘%1%’ in a binary cache") % storePath);
• replacement in src/hydra-queue-runner/binary-cache-store.cc at line 269
  [5.8741]→[5.238:340](∅→∅)
  − if (!(*localStore)->isValidPath(storePath))

  − (*localStore)->ensurePath(storePath);
  [5.8741]

  [5.8837]
  + localStore->addTempRoot(storePath);
• replacement in src/hydra-queue-runner/binary-cache-store.cc at line 271
  [5.8838]→[5.341:411](∅→∅)
  − ValidPathInfo info = (*localStore)->queryPathInfo(storePath);
  [5.8838]

  [5.8905]
  + if (!localStore->isValidPath(storePath))

  + localStore->ensurePath(storePath);
• edit in src/hydra-queue-runner/binary-cache-store.cc at line 274
  [5.8906]

  [5.8906]
  + ValidPathInfo info = localStore->queryPathInfo(storePath);

  +
• edit in src/hydra-queue-runner/binary-cache-store.hh at line 15
  [5.2138]→[5.185:341](∅→∅),[5.341]→[5.432:535](∅→∅)
  −

  − /* While BinaryCacheStore is thread-safe, LocalStore and RemoteStore

  − aren't. Until they are, use a factory to produce a thread-local

  − local store. */

  − typedef Pool<nix::ref<nix::Store>> StorePool;

  − typedef std::function<StorePool::Handle()> StoreFactory;
• replacement in src/hydra-queue-runner/binary-cache-store.hh at line 23
  [5.9417]→[5.392:423](∅→∅)
  − StoreFactory storeFactory;
  [5.9417]

  [5.620]
  + std::shared_ptr<Store> localStore;
• replacement in src/hydra-queue-runner/binary-cache-store.hh at line 34
  [5.9426]→[5.425:481](∅→∅)
  − BinaryCacheStore(const StoreFactory & storeFactory,
  [5.9426]

  [5.2300]
  + BinaryCacheStore(std::shared_ptr<Store> localStore,
• edit in src/hydra-queue-runner/hydra-queue-runner.cc at line 21
  [5.7501]→[5.536:617](∅→∅)
  − : localStorePool([]() { return std::make_shared<ref<Store>>(openStore()); })
• replacement in src/hydra-queue-runner/hydra-queue-runner.cc at line 48
  [5.1546]→[5.618:659](∅→∅)
  − StorePool::Handle State::getLocalStore()
  [5.1546]

  [5.1580]
  + ref<Store> State::getLocalStore()
• replacement in src/hydra-queue-runner/hydra-queue-runner.cc at line 50
  [5.1582]→[5.660:714](∅→∅)
  − auto conn(localStorePool.get());

  − return conn;
  [5.1582]

  [5.1621]
  + return ref<Store>(_localStore);
• edit in src/hydra-queue-runner/hydra-queue-runner.cc at line 748
  [4.609]

  [5.4619]
  +

  + _localStore = openStore();
• replacement in src/hydra-queue-runner/hydra-queue-runner.cc at line 752
  [4.662]→[4.662:696](∅→∅)
  − _destStore = openStore();
  [4.662]

  [4.696]
  + _destStore = _localStore;
• replacement in src/hydra-queue-runner/hydra-queue-runner.cc at line 760
  [4.960]→[4.960:1134](∅→∅)
  − [this]() { return this->getLocalStore(); },

  − "/home/eelco/Misc/Keys/test.nixos.org/secret",

  − "/home/eelco/Misc/Keys/test.nixos.org/public",
  [4.960]

  [4.1134]
  + _localStore,

  + hydraConfig["binary_cache_secret_key_file"],

  + hydraConfig["binary_cache_public_key_file"],
• replacement in src/hydra-queue-runner/hydra-queue-runner.cc at line 773
  [4.1527]→[4.1527:1583](∅→∅)
  − [this]() { return this->getLocalStore(); },
  [4.1527]

  [4.1583]
  + _localStore,
• replacement in src/hydra-queue-runner/local-binary-cache-store.cc at line 5
  [5.1521]→[5.1016:1096](∅→∅)
  − LocalBinaryCacheStore::LocalBinaryCacheStore(const StoreFactory & storeFactory,
  [5.1521]

  [5.1050]
  + LocalBinaryCacheStore::LocalBinaryCacheStore(std::shared_ptr<Store> localStore,
• replacement in src/hydra-queue-runner/local-binary-cache-store.cc at line 8
  [5.1143]→[5.1097:1164](∅→∅)
  − : BinaryCacheStore(storeFactory, secretKeyFile, publicKeyFile)
  [5.1143]

  [5.1744]
  + : BinaryCacheStore(localStore, secretKeyFile, publicKeyFile)
• replacement in src/hydra-queue-runner/local-binary-cache-store.hh at line 15
  [5.2839]→[5.1165:1226](∅→∅)
  − LocalBinaryCacheStore(const StoreFactory & storeFactory,
  [5.2839]

  [5.1193]
  + LocalBinaryCacheStore(std::shared_ptr<Store> localStore,
• replacement in src/hydra-queue-runner/queue-monitor.cc at line 39
  [5.20820]→[5.1136:1217](∅→∅)
  − bool done = getQueuedBuilds(*conn, *localStore, destStore, lastBuildId);
  [5.20820]

  [5.20872]
  + bool done = getQueuedBuilds(*conn, localStore, destStore, lastBuildId);
• replacement in src/hydra-queue-runner/s3-binary-cache-store.cc at line 34
  [5.2017]→[5.1227:1301](∅→∅)
  − S3BinaryCacheStore::S3BinaryCacheStore(const StoreFactory & storeFactory,
  [5.2017]

  [5.2079]
  + S3BinaryCacheStore::S3BinaryCacheStore(std::shared_ptr<Store> localStore,
• replacement in src/hydra-queue-runner/s3-binary-cache-store.cc at line 37
  [5.2175]→[5.1302:1369](∅→∅)
  − : BinaryCacheStore(storeFactory, secretKeyFile, publicKeyFile)
  [5.2175]

  [5.2240]
  + : BinaryCacheStore(localStore, secretKeyFile, publicKeyFile)
• replacement in src/hydra-queue-runner/s3-binary-cache-store.hh at line 23
  [5.5890]→[5.1370:1428](∅→∅)
  − S3BinaryCacheStore(const StoreFactory & storeFactory,
  [5.5890]

  [5.5936]
  + S3BinaryCacheStore(std::shared_ptr<Store> localStore,
• replacement in src/hydra-queue-runner/state.hh at line 83
  [5.4019]→[5.4019:4061](∅→∅)
  − Sync<std::map<time_t, time_t>> steps;
  [5.4019]

  [5.4061]
  + nix::Sync<std::map<time_t, time_t>> steps;
• replacement in src/hydra-queue-runner/state.hh at line 190
  [5.3142]→[5.3142:3165](∅→∅)
  − Sync<State> state;
  [5.3142]

  [5.3165]
  + nix::Sync<State> state;
• replacement in src/hydra-queue-runner/state.hh at line 230
  [5.1953]→[5.1953:1992](∅→∅)
  − Sync<ConnectInfo> connectInfo;
  [5.1953]

  [5.1992]
  + nix::Sync<ConnectInfo> connectInfo;
• replacement in src/hydra-queue-runner/state.hh at line 269
  [5.4496]→[5.4496:4521](∅→∅)
  − Sync<Builds> builds;
  [5.4496]

  [5.4566]
  + nix::Sync<Builds> builds;
• replacement in src/hydra-queue-runner/state.hh at line 273
  [5.4670]→[5.4670:4697](∅→∅)
  − Sync<Jobsets> jobsets;
  [5.4670]

  [5.4521]
  + nix::Sync<Jobsets> jobsets;
• replacement in src/hydra-queue-runner/state.hh at line 280
  [5.525]→[5.4787:4810](∅→∅),[5.4787]→[5.4787:4810](∅→∅)
  − Sync<Steps> steps;
  [5.525]

  [5.4810]
  + nix::Sync<Steps> steps;
• replacement in src/hydra-queue-runner/state.hh at line 284
  [5.4912]→[5.4912:4941](∅→∅)
  − Sync<Runnable> runnable;
  [5.4912]

  [5.4941]
  + nix::Sync<Runnable> runnable;
• replacement in src/hydra-queue-runner/state.hh at line 287
  [5.4985]→[2.578:663](∅→∅)
  − Sync<bool> dispatcherWakeup;

  − std::condition_variable_any dispatcherWakeupCV;
  [5.4985]

  [5.5063]
  + nix::Sync<bool> dispatcherWakeup;

  + std::condition_variable dispatcherWakeupCV;
• replacement in src/hydra-queue-runner/state.hh at line 291
  [5.5102]→[5.5102:5131](∅→∅)
  − Pool<Connection> dbPool;
  [5.5102]

  [5.5131]
  + nix::Pool<Connection> dbPool;
• replacement in src/hydra-queue-runner/state.hh at line 295
  [5.584]→[5.5215:5276](∅→∅),[5.5215]→[5.5215:5276](∅→∅)
  − Sync<Machines> machines; // FIXME: use atomic_shared_ptr
  [5.584]

  [5.5334]
  + nix::Sync<Machines> machines; // FIXME: use atomic_shared_ptr
• replacement in src/hydra-queue-runner/state.hh at line 317
  [5.6122]→[5.614:666](∅→∅),[5.666]→[5.6169:6222](∅→∅),[5.6169]→[5.6169:6222](∅→∅)
  − Sync<std::queue<nix::Path>> logCompressorQueue;

  − std::condition_variable_any logCompressorWakeup;
  [5.6122]

  [5.6222]
  + nix::Sync<std::queue<nix::Path>> logCompressorQueue;

  + std::condition_variable logCompressorWakeup;
• replacement in src/hydra-queue-runner/state.hh at line 325
  [5.6569]→[5.6569:6691](∅→∅)
  − Sync<std::queue<NotificationItem>> notificationSenderQueue;

  − std::condition_variable_any notificationSenderWakeup;
  [5.6569]

  [5.6691]
  + nix::Sync<std::queue<NotificationItem>> notificationSenderQueue;

  + std::condition_variable notificationSenderWakeup;
• replacement in src/hydra-queue-runner/state.hh at line 339
  [5.2678]→[5.2678:2737](∅→∅)
  − Sync<std::map<std::string, MachineType>> machineTypes;
  [5.2678]

  [5.2737]
  + nix::Sync<std::map<std::string, MachineType>> machineTypes;
• replacement in src/hydra-queue-runner/state.hh at line 353
  [5.6778]→[5.1262:1330](∅→∅),[5.1330]→[5.1429:1458](∅→∅),[5.6778]→[5.1429:1458](∅→∅)
  − /* Pool of local stores. */

  − nix::StorePool localStorePool;

  −

  − /* Destination store. */
  [5.6778]

  [5.1458]
  + std::shared_ptr<nix::Store> _localStore;
• replacement in src/hydra-queue-runner/state.hh at line 363
  [5.13272]→[5.1331:1375](∅→∅)
  − nix::StorePool::Handle getLocalStore();
  [5.13272]

  [5.13314]
  + nix::ref<nix::Store> getLocalStore();
• replacement in src/hydra-queue-runner/token-server.hh at line 17
  [5.2032]→[5.2032:2072](∅→∅)
  − std::condition_variable_any wakeup;
  [5.2032]

  [5.2072]
  + std::condition_variable wakeup;