Merge branch 'master' of https://github.com/NixOS/nix into upstream/master

Author: Anton-Latukha

src/libstore/build.cc

@@ -2202,7 +2202,7 @@ void DerivationGoal::initEnv()
     env["NIX_BUILD_CORES"] = (format("%d") % settings.buildCores).str();
 
     /* In non-structured mode, add all bindings specified in the
-       derivation via the environments, except those listed in the
+       derivation via the environment, except those listed in the
        passAsFile attribute. Those are passed as file names pointing
        to temporary files containing the contents. Note that
        passAsFile is ignored in structure mode because it's not
@@ -2258,6 +2258,11 @@ void DerivationGoal::initEnv()
         Strings varNames = tokenizeString<Strings>(get(drv->env, "impureEnvVars"));
         for (auto & i : varNames) env[i] = getEnv(i);
     }
+
+    /* Currently structured log messages piggyback on stderr, but we
+       may change that in the future. So tell the builder which file
+       descriptor to use for that. */
+    env["NIX_LOG_FD"] = "2";
 }
 
 

src/libstore/legacy-ssh-store.cc

@@ -105,6 +105,13 @@ struct LegacySSHStore : public Store
             readLongLong(conn->from); // download size
             info->narSize = readLongLong(conn->from);
 
+            if (GET_PROTOCOL_MINOR(conn->remoteVersion) >= 4) {
+                auto s = readString(conn->from);
+                info->narHash = s.empty() ? Hash() : Hash(s);
+                conn->from >> info->ca;
+                info->sigs = readStrings<StringSet>(conn->from);
+            }
+
             auto s = readString(conn->from);
             assert(s == "");
 

src/libstore/serve-protocol.hh

@@ -5,7 +5,7 @@ namespace nix {
 #define SERVE_MAGIC_1 0x390c9deb
 #define SERVE_MAGIC_2 0x5452eecb
 
-#define SERVE_PROTOCOL_VERSION 0x203
+#define SERVE_PROTOCOL_VERSION 0x204
 #define GET_PROTOCOL_MAJOR(x) ((x) & 0xff00)
 #define GET_PROTOCOL_MINOR(x) ((x) & 0x00ff)
 

src/libstore/store-api.cc

@@ -593,23 +593,19 @@ void copyStorePath(ref<Store> srcStore, ref<Store> dstStore,
     MyStringSink sink(progress);
     srcStore->narFromPath({storePath}, sink);
 
-    if (!info->narHash && !checkSigs) {
+    if (!info->narHash) {
         auto info2 = make_ref<ValidPathInfo>(*info);
         info2->narHash = hashString(htSHA256, *sink.s);
         if (!info->narSize) info2->narSize = sink.s->size();
         info = info2;
     }
 
-    assert(info->narHash);
-
     if (info->ultimate) {
         auto info2 = make_ref<ValidPathInfo>(*info);
         info2->ultimate = false;
         info = info2;
     }
 
-    assert(info->narHash);
-
     dstStore->addToStore(*info, sink.s, repair, checkSigs);
 }
 

src/libutil/args.hh

@@ -80,6 +80,19 @@ public:
         FlagMaker & arity(size_t arity) { flag->arity = arity; return *this; };
         FlagMaker & handler(std::function<void(Strings)> handler) { flag->handler = handler; return *this; };
         FlagMaker & category(const std::string & s) { flag->category = s; return *this; };
+
+        FlagMaker & dest(std::string * dest) {
+            assert(flag->arity == 1);
+            flag->handler = [=](Strings ss) { *dest = ss.front(); };
+            return *this;
+        };
+
+        template<class T>
+        FlagMaker & set(T * dest, const T & val) {
+            assert(flag->arity == 0);
+            flag->handler = [=](Strings ss) { *dest = val; };
+            return *this;
+        };
     };
 
     FlagMaker mkFlag();

src/libutil/thread-pool.cc

@@ -21,7 +21,7 @@ ThreadPool::~ThreadPool()
     std::vector<std::thread> workers;
     {
         auto state(state_.lock());
-        state->quit = true;
+        quit = true;
         std::swap(workers, state->workers);
     }
 
@@ -36,7 +36,7 @@ ThreadPool::~ThreadPool()
 void ThreadPool::enqueue(const work_t & t)
 {
     auto state(state_.lock());
-    if (state->quit)
+    if (quit)
         throw ThreadPoolShutDown("cannot enqueue a work item while the thread pool is shutting down");
     state->left.push(t);
     if (state->left.size() > state->workers.size() && state->workers.size() < maxThreads)
@@ -46,53 +46,84 @@ void ThreadPool::enqueue(const work_t & t)
 
 void ThreadPool::process()
 {
+    /* Loop until there are no active work items *and* there either
+       are no queued items or there is an exception. The
+       post-condition is that no new items will become active. */
     while (true) {
         auto state(state_.lock());
-        if (state->exception)
-            std::rethrow_exception(state->exception);
-        if (state->left.empty() && !state->pending) break;
+        if (!state->active) {
+            if (state->exception)
+                std::rethrow_exception(state->exception);
+            if (state->left.empty())
+                break;
+        }
         state.wait(done);
     }
 }
 
 void ThreadPool::workerEntry()
 {
+    interruptCheck = [&]() { return (bool) quit; };
+
     bool didWork = false;
+    std::exception_ptr exc;
 
     while (true) {
         work_t w;
         {
             auto state(state_.lock());
+
+            if (didWork) {
+                assert(state->active);
+                state->active--;
+
+                if (exc) {
+
+                    if (!state->exception) {
+                        state->exception = exc;
+                        // Tell the other workers to quit.
+                        quit = true;
+                        work.notify_all();
+                    } else {
+                        /* Print the exception, since we can't
+                           propagate it. */
+                        try {
+                            std::rethrow_exception(exc);
+                        } catch (std::exception & e) {
+                            if (!dynamic_cast<Interrupted*>(&e) &&
+                                !dynamic_cast<ThreadPoolShutDown*>(&e))
+                                ignoreException();
+                        } catch (...) {
+                        }
+                    }
+                }
+            }
+
+            /* Wait until a work item is available or another thread
+               had an exception or we're asked to quit. */
             while (true) {
-                if (state->quit || state->exception) return;
-                if (didWork) {
-                    assert(state->pending);
-                    state->pending--;
-                    didWork = false;
+                if (quit) {
+                    if (!state->active)
+                        done.notify_one();
+                    return;
                 }
                 if (!state->left.empty()) break;
-                if (!state->pending)
-                    done.notify_all();
+                if (!state->active) {
+                    done.notify_one();
+                    return;
+                }
                 state.wait(work);
             }
-            w = state->left.front();
+
+            w = std::move(state->left.front());
             state->left.pop();
-            state->pending++;
+            state->active++;
         }
 
         try {
             w();
-        } catch (std::exception & e) {
-            auto state(state_.lock());
-            if (state->exception) {
-                if (!dynamic_cast<Interrupted*>(&e) &&
-                    !dynamic_cast<ThreadPoolShutDown*>(&e))
-                    printError(format("error: %s") % e.what());
-            } else {
-                state->exception = std::current_exception();
-                work.notify_all();
-                done.notify_all();
-            }
+        } catch (...) {
+            exc = std::current_exception();
         }
 
         didWork = true;

src/libutil/thread-pool.hh

@@ -7,6 +7,7 @@
 #include <functional>
 #include <thread>
 #include <map>
+#include <atomic>
 
 namespace nix {
 
@@ -44,12 +45,13 @@ private:
     struct State
     {
         std::queue<work_t> left;
-        size_t pending = 0;
+        size_t active = 0;
         std::exception_ptr exception;
         std::vector<std::thread> workers;
-        bool quit = false;
     };
 
+    std::atomic_bool quit{false};
+
     Sync<State> state_;
 
     std::condition_variable work, done;

src/libutil/util.cc

@@ -1002,6 +1002,7 @@ void closeOnExec(int fd)
 bool _isInterrupted = false;
 
 static thread_local bool interruptThrown = false;
+thread_local std::function<bool()> interruptCheck;
 
 void setInterruptThrown()
 {
@@ -1020,7 +1021,6 @@ void _interrupted()
 }
 
 
-
 //////////////////////////////////////////////////////////////////////
 
 

src/libutil/util.hh

@@ -273,13 +273,16 @@ void closeOnExec(int fd);
 
 extern bool _isInterrupted;
 
+extern thread_local std::function<bool()> interruptCheck;
+
 void setInterruptThrown();
 
 void _interrupted();
 
 void inline checkInterrupt()
 {
-    if (_isInterrupted) _interrupted();
+    if (_isInterrupted || (interruptCheck && interruptCheck()))
+        _interrupted();
 }
 
 MakeError(Interrupted, BaseError)

src/nix-store/nix-store.cc

@@ -858,6 +858,8 @@ static void opServe(Strings opFlags, Strings opArgs)
                         // !!! Maybe we want compression?
                         out << info->narSize // downloadSize
                             << info->narSize;
+                        if (GET_PROTOCOL_MINOR(clientVersion) >= 4)
+                            out << (info->narHash ? info->narHash.to_string() : "") << info->ca << info->sigs;
                     } catch (InvalidPath &) {
                     }
                 }