Multithreaded evaluator

configure.ac

@@ -346,7 +346,7 @@ AC_ARG_ENABLE(gc, AS_HELP_STRING([--enable-gc],[enable garbage collection in the
   gc=$enableval, gc=yes)
 if test "$gc" = yes; then
   PKG_CHECK_MODULES([BDW_GC], [bdw-gc])
-  CXXFLAGS="$BDW_GC_CFLAGS $CXXFLAGS"
+  CXXFLAGS="$BDW_GC_CFLAGS -DGC_THREADS $CXXFLAGS"
   AC_DEFINE(HAVE_BOEHMGC, 1, [Whether to use the Boehm garbage collector.])
 
   # See `fixupBoehmStackPointer`, for the integration between Boehm GC

src/libexpr-c/nix_api_value.cc

@@ -181,6 +181,8 @@ ValueType nix_get_type(nix_c_context * context, const nix_value * value)
         switch (v.type()) {
         case nThunk:
             return NIX_TYPE_THUNK;
+        case nFailed:
+            return NIX_TYPE_FAILED;
         case nInt:
             return NIX_TYPE_INT;
         case nFloat:

src/libexpr-c/nix_api_value.h

@@ -31,7 +31,8 @@ typedef enum {
     NIX_TYPE_ATTRS,
     NIX_TYPE_LIST,
     NIX_TYPE_FUNCTION,
-    NIX_TYPE_EXTERNAL
+    NIX_TYPE_EXTERNAL,
+    NIX_TYPE_FAILED,
 } ValueType;
 
 // forward declarations

src/libexpr/eval-gc.cc

@@ -4,6 +4,7 @@
 #include "config-global.hh"
 #include "serialise.hh"
 #include "eval-gc.hh"
+#include "file-system.hh"
 
 #if HAVE_BOEHMGC
 
@@ -32,6 +33,39 @@ static void * oomHandler(size_t requested)
     throw std::bad_alloc();
 }
 
+static size_t getFreeMem()
+{
+    /* On Linux, use the `MemAvailable` or `MemFree` fields from
+       /proc/cpuinfo. */
+#  if __linux__
+    {
+        std::unordered_map<std::string, std::string> fields;
+        for (auto & line : tokenizeString<std::vector<std::string>>(readFile("/proc/meminfo"), "\n")) {
+            auto colon = line.find(':');
+            if (colon == line.npos)
+                continue;
+            fields.emplace(line.substr(0, colon), trim(line.substr(colon + 1)));
+        }
+
+        auto i = fields.find("MemAvailable");
+        if (i == fields.end())
+            i = fields.find("MemFree");
+        if (i != fields.end()) {
+            auto kb = tokenizeString<std::vector<std::string>>(i->second, " ");
+            if (kb.size() == 2 && kb[1] == "kB")
+                return string2Int<size_t>(kb[0]).value_or(0) * 1024;
+        }
+    }
+#  endif
+
+    /* On non-Linux systems, conservatively assume that 25% of memory is free. */
+    long pageSize = sysconf(_SC_PAGESIZE);
+    long pages = sysconf(_SC_PHYS_PAGES);
+    if (pageSize != -1)
+        return (pageSize * pages) / 4;
+    return 0;
+}
+
 static inline void initGCReal()
 {
     /* Initialise the Boehm garbage collector. */
@@ -50,10 +84,12 @@ static inline void initGCReal()
 
     GC_INIT();
 
+    GC_allow_register_threads();
+
     GC_set_oom_fn(oomHandler);
 
-    /* Set the initial heap size to something fairly big (25% of
-       physical RAM, up to a maximum of 384 MiB) so that in most cases
+    /* Set the initial heap size to something fairly big (80% of
+       free RAM, up to a maximum of 8 GiB) so that in most cases
        we don't need to garbage collect at all.  (Collection has a
        fairly significant overhead.)  The heap size can be overridden
        through libgc's GC_INITIAL_HEAP_SIZE environment variable.  We
@@ -64,13 +100,10 @@ static inline void initGCReal()
     if (!getEnv("GC_INITIAL_HEAP_SIZE")) {
         size_t size = 32 * 1024 * 1024;
 #  if HAVE_SYSCONF && defined(_SC_PAGESIZE) && defined(_SC_PHYS_PAGES)
-        size_t maxSize = 384 * 1024 * 1024;
-        long pageSize = sysconf(_SC_PAGESIZE);
-        long pages = sysconf(_SC_PHYS_PAGES);
-        if (pageSize != -1)
-            size = (pageSize * pages) / 4; // 25% of RAM
-        if (size > maxSize)
-            size = maxSize;
+        size_t maxSize = 8ULL * 1024 * 1024 * 1024;
+        auto free = getFreeMem();
+        debug("free memory is %d bytes", free);
+        size = std::min((size_t) (free * 0.8), maxSize);
 #  endif
         debug("setting initial heap size to %1% bytes", size);
         GC_expand_hp(size);

src/libexpr/eval-inline.hh

@@ -32,6 +32,8 @@ Value * EvalState::allocValue()
        GC_malloc_many returns a linked list of objects of the given size, where the first word
        of each object is also the pointer to the next object in the list. This also means that we
        have to explicitly clear the first word of every object we take. */
+    thread_local static std::shared_ptr<void *> valueAllocCache{std::allocate_shared<void *>(traceable_allocator<void *>(), nullptr)};
+
     if (!*valueAllocCache) {
         *valueAllocCache = GC_malloc_many(sizeof(Value));
         if (!*valueAllocCache) throw std::bad_alloc();
@@ -62,6 +64,8 @@ Env & EvalState::allocEnv(size_t size)
 #if HAVE_BOEHMGC
     if (size == 1) {
         /* see allocValue for explanations. */
+        thread_local static std::shared_ptr<void *> env1AllocCache{std::allocate_shared<void *>(traceable_allocator<void *>(), nullptr)};
+
         if (!*env1AllocCache) {
             *env1AllocCache = GC_malloc_many(sizeof(Env) + sizeof(Value *));
             if (!*env1AllocCache) throw std::bad_alloc();
@@ -84,9 +88,33 @@ Env & EvalState::allocEnv(size_t size)
 [[gnu::always_inline]]
 void EvalState::forceValue(Value & v, const PosIdx pos)
 {
+    auto type = v.internalType.load(std::memory_order_acquire);
+
+    if (isFinished(type))
+        goto done;
+
+    if (type == tThunk) {
+        try {
+            if (!v.internalType.compare_exchange_strong(type, tPending, std::memory_order_acquire, std::memory_order_acquire)) {
+                if (type == tPending || type == tAwaited) {
+                    waitOnThunk(v, type == tAwaited);
+                    goto done;
+                }
+                if (isFinished(type))
+                    goto done;
+                printError("NO LONGER THUNK %x %d", this, type);
+                abort();
+            }
+            Env * env = v.payload.thunk.env;
+            Expr * expr = v.payload.thunk.expr;
+            expr->eval(*this, *env, v);
+        } catch (...) {
+            v.mkFailed();
+            throw;
+        }
+    }
+    #if 0
     if (v.isThunk()) {
-        Env * env = v.payload.thunk.env;
-        Expr * expr = v.payload.thunk.expr;
         try {
             v.mkBlackhole();
             //checkInterrupt();
@@ -97,8 +125,33 @@ void EvalState::forceValue(Value & v, const PosIdx pos)
             throw;
         }
     }
-    else if (v.isApp())
-        callFunction(*v.payload.app.left, *v.payload.app.right, v, pos);
+    #endif
+    else if (type == tApp) {
+        try {
+            if (!v.internalType.compare_exchange_strong(type, tPending, std::memory_order_acquire, std::memory_order_acquire)) {
+                if (type == tPending || type == tAwaited) {
+                    waitOnThunk(v, type == tAwaited);
+                    goto done;
+                }
+                if (isFinished(type))
+                    goto done;
+                printError("NO LONGER APP %x %d", this, type);
+                abort();
+            }
+            callFunction(*v.payload.app.left, *v.payload.app.right, v, pos);
+        } catch (...) {
+            v.mkFailed();
+            throw;
+        }
+    }
+    else if (type == tPending || type == tAwaited)
+        type = waitOnThunk(v, type == tAwaited);
+    else
+        abort();
+
+ done:
+    if (type == tFailed)
+        std::rethrow_exception(v.payload.failed->ex);
 }