hydra-queue-runner: Limit memory usage

[?]

Mar 9, 2016, 1:30 PM

SL3WSRACCX2IMJHHLTRAUQT7QDLCOKYLVO2FEHWIHXM5GPKSRJTQC

Dependencies

[2] B2L4T3X6 Sync with Nix
[3] UVNTWTWG Prevent download of NARs we just uploaded
[4] BRAESISH Warn if PostgreSQL appears stalled
[5] MB3TISH2 Rate-limit the number of threads copying closures at the same time
[6] DIEY5USN Keep better bytesReceived/bytesSent stats
[7] LE4VZIY5 More stats
[8] GJV2J5HX Pool local store connections
[9] 5AIYUMTB Basic remote building
[10] N4IROACV Move buildRemote() into State
[11] FITVNQ2S Keep track of the time we spend copying to/from build machines
[12] A2GL5FOZ Moar stats
[13] CNLNT3T4 Allow only 1 thread to send a closure to a given machine at the same time
[14] OG3Z3QGC Namespace cleanup
[15] 73YR46NJ hydra-queue-runner: Write directly to a binary cache
[16] HJOEIMLR Refactor
[17] 6EO3HVNA Merge remote-tracking branch 'origin/master' into binary-cache
[*] 24BMQDZA Start of single-process hydra-queue-runner
[*] PLOZBRTR Add command ‘hydra-queue-runner --status’ to show current status
[*] IK2UBDAU Revive jobset scheduling
[*] EYR3EW6J Keep stats for the Hydra auto scaler
[*] EOO4EFWD Use a single BinaryCacheStore for all threads

Change contents

replacement in src/hydra-queue-runner/build-remote.cc at line 286

[5.1035]→[5.1286:1404](∅→∅),[5.1286]→[5.1286:1404](∅→∅)

printMsg(lvlDebug, format("copying outputs of ‘%1%’ from ‘%2%’") % step->drvPath % machine->sshName);

[5.1035]

[5.816]

MaintainCount mc(nrStepsCopyingFrom);

auto now1 = std::chrono::steady_clock::now();
replacement in src/hydra-queue-runner/build-remote.cc at line 293

[5.931]→[5.290:336](∅→∅),[5.290]→[5.290:336](∅→∅)

MaintainCount mc(nrStepsCopyingFrom);

[5.931]

[5.217]

/* Query the size of the output paths. */
size_t totalNarSize = 0;
to << cmdQueryPathInfos << outputs;
to.flush();
while (true) {
if (readString(from) == "") break;
readString(from); // deriver
readStrings<PathSet>(from); // references
readLongLong(from); // download size
totalNarSize += readLongLong(from);
}

printMsg(lvlDebug, format("copying outputs of ‘%s’ from ‘%s’ (%d bytes)")
% step->drvPath % machine->sshName % totalNarSize);

/* Block until we have the required amount of memory
available. FIXME: only need this for binary cache
destination stores. */
auto resStart = std::chrono::steady_clock::now();
auto memoryReservation(memoryTokens.get(totalNarSize));
auto resStop = std::chrono::steady_clock::now();
replacement in src/hydra-queue-runner/build-remote.cc at line 316

[5.218]→[3.177:231](∅→∅)

result.accessor = destStore->getFSAccessor();

[5.218]

[3.231]

auto resMs = std::chrono::duration_cast<std::chrono::milliseconds>(resStop - resStart).count();
if (resMs >= 1000)
printMsg(lvlError, format("warning: had to wait %d ms for %d memory tokens for %s")
% resMs % totalNarSize % step->drvPath);
replacement in src/hydra-queue-runner/build-remote.cc at line 321

[3.232]→[5.218:272](∅→∅),[5.218]→[5.218:272](∅→∅)

auto now1 = std::chrono::steady_clock::now();

[3.232]

[5.272]

result.accessor = destStore->getFSAccessor();
edit in src/hydra-queue-runner/hydra-queue-runner.cc at line 20

[19.7501]

[19.7501]

: memoryTokens(4ULL << 30) // FIXME: make this configurable
edit in src/hydra-queue-runner/hydra-queue-runner.cc at line 571

[4.626]

[20.1628]

root.attr("memoryTokensInUse", memoryTokens.currentUse());
edit in src/hydra-queue-runner/hydra-queue-runner.cc at line 595

[20.2016]

[21.1730]
edit in src/hydra-queue-runner/hydra-queue-runner.cc at line 607

[21.2189]

[22.1936]
edit in src/hydra-queue-runner/state.hh at line 12

[5.1299]

[5.1299]

#include "token-server.hh"

#include "derivations.hh"
edit in src/hydra-queue-runner/state.hh at line 17

[5.1342]→[5.1342:1361](∅→∅),[5.1388]→[5.1388:1389](∅→∅)

#include "sync.hh"
replacement in src/hydra-queue-runner/state.hh at line 18

[5.1413]→[5.1413:1439](∅→∅),[5.1439]→[5.1218:1261](∅→∅)

#include "derivations.hh"

#include "binary-cache-store.hh" // FIXME

[5.1413]

[5.1461]

#include "sync.hh"
edit in src/hydra-queue-runner/state.hh at line 355

[23.1502]

[23.1502]

/* Token server to prevent threads from allocating too many big
strings concurrently while importing NARs from the build
machines. When a thread imports a NAR of size N, it will first
acquire N memory tokens, causing it to block until that many
tokens are available. */
nix::TokenServer memoryTokens;
edit in src/hydra-queue-runner/token-server.hh at line 6

[5.1693]

[5.1693]

#include "types.hh"

namespace nix {
edit in src/hydra-queue-runner/token-server.hh at line 10

[5.1694]

[5.1694]

MakeError(NoTokens, Error)
replacement in src/hydra-queue-runner/token-server.hh at line 13

[5.1764]→[5.1764:1945](∅→∅)

available. Calling get() will return a Token object, representing
ownership of a token. If no token is available, get() will sleep
until another thread returns a token. */

[5.1764]

[5.1945]

available. Calling get(N) will return a Token object, representing
ownership of N tokens. If the requested number of tokens is
unavailable, get() will sleep until another thread returns a
token. */
replacement in src/hydra-queue-runner/token-server.hh at line 20

[5.1966]→[5.1966:1994](∅→∅)

unsigned int maxTokens;

[5.1966]

[5.1994]

const size_t maxTokens;
replacement in src/hydra-queue-runner/token-server.hh at line 22

[5.1995]→[5.1995:2032](∅→∅)

Sync<unsigned int> curTokens{0};

[5.1995]

[2.2375]

Sync<size_t> inUse{0};
replacement in src/hydra-queue-runner/token-server.hh at line 26

[5.2081]→[5.2081:2148](∅→∅)

TokenServer(unsigned int maxTokens) : maxTokens(maxTokens) { }

[5.2081]

[5.2148]

TokenServer(size_t maxTokens) : maxTokens(maxTokens) { }
edit in src/hydra-queue-runner/token-server.hh at line 34

[5.2227]

[5.2227]

size_t tokens;
replacement in src/hydra-queue-runner/token-server.hh at line 38

[5.2259]→[5.2259:2322](∅→∅)

Token(TokenServer * ts, unsigned int timeout) : ts(ts)

[5.2259]

[5.2322]

Token(TokenServer * ts, size_t tokens, unsigned int timeout)
: ts(ts), tokens(tokens)
replacement in src/hydra-queue-runner/token-server.hh at line 41

[5.2332]→[5.2332:2430](∅→∅)

auto curTokens(ts->curTokens.lock());
while (*curTokens >= ts->maxTokens)

[5.2332]

[5.2430]

if (tokens >= ts->maxTokens)
throw NoTokens(format("requesting more tokens (%d) than exist (%d)") % tokens);
auto inUse(ts->inUse.lock());
while (*inUse + tokens > ts->maxTokens)
replacement in src/hydra-queue-runner/token-server.hh at line 46

[5.2461]→[5.2461:2623](∅→∅)

if (!curTokens.wait_for(ts->wakeup, std::chrono::seconds(timeout),
[&]() { return *curTokens < ts->maxTokens; }))

[5.2461]

[5.2623]

if (!inUse.wait_for(ts->wakeup, std::chrono::seconds(timeout),
[&]() { return *inUse + tokens <= ts->maxTokens; }))
replacement in src/hydra-queue-runner/token-server.hh at line 50

[5.2678]→[5.2678:2754](∅→∅)

curTokens.wait(ts->wakeup);
(*curTokens)++;

[5.2678]

[5.2754]

inUse.wait(ts->wakeup);
*inUse += tokens;
replacement in src/hydra-queue-runner/token-server.hh at line 64

[5.2983]→[5.2983:3105](∅→∅)

auto curTokens(ts->curTokens.lock());
assert(*curTokens);
(*curTokens)--;

[5.2983]

[5.3105]

auto inUse(ts->inUse.lock());
assert(*inUse >= tokens);
*inUse -= tokens;
replacement in src/hydra-queue-runner/token-server.hh at line 74

[5.3223]→[5.3223:3263](∅→∅)

Token get(unsigned int timeout = 0)

[5.3223]

[5.3263]

Token get(size_t tokens = 1, unsigned int timeout = 0)
{
return Token(this, tokens, timeout);
}

size_t currentUse()
replacement in src/hydra-queue-runner/token-server.hh at line 81

[5.3269]→[5.3269:3306](∅→∅)

return Token(this, timeout);

[5.3269]

[5.3306]

auto inUse_(inUse.lock());
return *inUse_;
edit in src/hydra-queue-runner/token-server.hh at line 85

[5.3315]

}