Merge pull request #5492 from obsidiansystems/backport/fd-leak-2.3-maintenance

Nix 2.3 Backport: PathSubstitutionGoal: Clean up pipe

src/libstore/build.cc

@@ -179,6 +179,8 @@ public:
 
     virtual string key() = 0;
 
+    virtual void cleanup() { }
+
 protected:
 
     virtual void amDone(ExitCode result);
@@ -424,6 +426,8 @@ void Goal::amDone(ExitCode result)
     }
     waiters.clear();
     worker.removeGoal(shared_from_this());
+
+    cleanup();
 }
 
 
@@ -3895,6 +3899,8 @@ public:
     void handleChildOutput(int fd, const string & data) override;
     void handleEOF(int fd) override;
 
+    void cleanup() override;
+
     Path getStorePath() { return storePath; }
 
     void amDone(ExitCode result) override
@@ -3918,15 +3924,7 @@ SubstitutionGoal::SubstitutionGoal(const Path & storePath, Worker & worker, Repa
 
 SubstitutionGoal::~SubstitutionGoal()
 {
-    try {
+    cleanup();
-        if (thr.joinable()) {
-            // FIXME: signal worker thread to quit.
-            thr.join();
-            worker.childTerminated(this);
-        }
-    } catch (...) {
-        ignoreException();
-    }
 }
 
 
@@ -3961,6 +3959,8 @@ void SubstitutionGoal::tryNext()
 {
     trace("trying next substituter");
 
+    cleanup();
+
     if (subs.size() == 0) {
         /* None left.  Terminate this goal and let someone else deal
            with it. */
@@ -4088,7 +4088,7 @@ void SubstitutionGoal::tryToRun()
     thr = std::thread([this]() {
         try {
             /* Wake up the worker loop when we're done. */
-            Finally updateStats([this]() { outPipe.writeSide = -1; });
+            Finally updateStats([this]() { outPipe.writeSide.close(); });
 
             Activity act(*logger, actSubstitute, Logger::Fields{storePath, sub->getUri()});
             PushActivity pact(act.id);
@@ -4172,6 +4172,20 @@ void SubstitutionGoal::handleEOF(int fd)
     if (fd == outPipe.readSide.get()) worker.wakeUp(shared_from_this());
 }
 
+void SubstitutionGoal::cleanup()
+{
+    try {
+        if (thr.joinable()) {
+            // FIXME: signal worker thread to quit.
+            thr.join();
+            worker.childTerminated(this);
+        }
+
+        outPipe.close();
+    } catch (...) {
+        ignoreException();
+    }
+}
 
 //////////////////////////////////////////////////////////////////////
 

src/libutil/util.cc

@@ -753,6 +753,7 @@ void AutoCloseFD::close()
         if (::close(fd) == -1)
             /* This should never happen. */
             throw SysError(format("closing file descriptor %1%") % fd);
+        fd = -1;
     }
 }
 
@@ -770,6 +771,12 @@ int AutoCloseFD::release()
     return oldFD;
 }
 
+void Pipe::close()
+{
+    readSide.close();
+    writeSide.close();
+}
+
 
 void Pipe::create()
 {
@@ -1080,7 +1087,7 @@ void runProgram2(const RunOptions & options)
         throw SysError("executing '%1%'", options.program);
     }, processOptions);
 
-    out.writeSide = -1;
+    out.writeSide.close();
 
     std::thread writerThread;
 
@@ -1093,7 +1100,7 @@ void runProgram2(const RunOptions & options)
 
 
     if (source) {
-        in.readSide = -1;
+        in.readSide.close();
         writerThread = std::thread([&]() {
             try {
                 std::vector<unsigned char> buf(8 * 1024);
@@ -1110,7 +1117,7 @@ void runProgram2(const RunOptions & options)
             } catch (...) {
                 promise.set_exception(std::current_exception());
             }
-            in.writeSide = -1;
+            in.writeSide.close();
         });
     }
 

src/libutil/util.hh

@@ -190,7 +190,6 @@ public:
 class AutoCloseFD
 {
     int fd;
-    void close();
 public:
     AutoCloseFD();
     AutoCloseFD(int fd);
@@ -202,6 +201,7 @@ public:
     int get() const;
     explicit operator bool() const;
     int release();
+    void close();
 };
 
 
@@ -210,6 +210,7 @@ class Pipe
 public:
     AutoCloseFD readSide, writeSide;
     void create();
+    void close();
 };