Fix bug in algorithm::traverse where if we fail to create a thread, the exception throw does not cancel and join the existing threads, thus causing std::terminate to be invoked.

2 files changed, 62 insertions, 39 deletions

diff --git a/include/llfio/revision.hpp b/include/llfio/revision.hpp
index 634724d9..ab35fe6a 100644
--- a/include/llfio/revision.hpp
+++ b/include/llfio/revision.hpp
@@ -1,4 +1,4 @@
 // Note the second line of this file must ALWAYS be the git SHA, third line ALWAYS the git SHA update time
-#define LLFIO_PREVIOUS_COMMIT_REF    3354ed31eea1534321d2ea0c05e82e572297609c
-#define LLFIO_PREVIOUS_COMMIT_DATE   "2021-03-17 16:05:09 +00:00"
-#define LLFIO_PREVIOUS_COMMIT_UNIQUE 3354ed31
+#define LLFIO_PREVIOUS_COMMIT_REF    565f27555d13ce2a37346d838f32e03e1972aac1
+#define LLFIO_PREVIOUS_COMMIT_DATE   "2021-04-20 10:44:21 +00:00"
+#define LLFIO_PREVIOUS_COMMIT_UNIQUE 565f2755
diff --git a/include/llfio/v2.0/detail/impl/traverse.ipp b/include/llfio/v2.0/detail/impl/traverse.ipp
index 7f7bf4df..190bcd08 100644
--- a/include/llfio/v2.0/detail/impl/traverse.ipp
+++ b/include/llfio/v2.0/detail/impl/traverse.ipp
@@ -170,7 +170,7 @@ namespace algorithm
 #endif
           std::vector<std::list<workitem>> workqueue;
           size_t workqueue_base{0};
-          size_t dirs_processed{0}, known_dirs_remaining{0}, depth_processed{0}, threads_sleeping{0};
+          size_t dirs_processed{0}, known_dirs_remaining{0}, depth_processed{0}, threads_sleeping{0}, threads_running{0};
 
           explicit state_t(traverse_visitor *_visitor)
               : visitor(_visitor)
@@ -385,49 +385,67 @@ namespace algorithm
           std::condition_variable cond, maincond;
           bool done = false;
           optional<result<void>::error_type> run_error;
-          for(size_t n = 0; n < threads; n++)
           {
-            workerthreads.push_back(std::thread(
-            [&](worker *w) {
+            auto handle_failure = make_scope_fail([&]() noexcept {
               std::unique_lock<std::mutex> g(state.lock);
-              while(!done)
+              done = true;
+              while(state.threads_running > 0)
               {
-                if(state.known_dirs_remaining == 0)
+                g.unlock();
+                cond.notify_all();
+                g.lock();
+              }
+              g.unlock();
+              for(auto &i : workerthreads)
+              {
+                i.join();
+              }
+            });
+            for(size_t n = 0; n < threads; n++)
+            {
+              workerthreads.push_back(std::thread(
+              [&](worker *w) {
+                std::unique_lock<std::mutex> g(state.lock);
+                state.threads_running++;
+                while(!done)
                 {
-                  // sleep
-                  state.threads_sleeping++;
-                  maincond.notify_all();
-                  cond.wait(g);
-                  if(done)
+                  if(state.known_dirs_remaining == 0)
                   {
-                    break;
+                    // sleep
+                    state.threads_sleeping++;
+                    maincond.notify_all();
+                    cond.wait(g);
+                    if(done)
+                    {
+                      break;
+                    }
+                    state.threads_sleeping--;
                   }
-                  state.threads_sleeping--;
-                }
-                else
-                {
-                  // wake everybody
-                  cond.notify_all();
-                }
-                auto r = w->run(g, use_slow_path, topdirh, data);
-                if(!g.owns_lock())
-                {
-                  g.lock();
-                }
-                if(!r)
-                {
-                  done = true;
-                  if(!run_error)
+                  else
+                  {
+                    // wake everybody
+                    cond.notify_all();
+                  }
+                  auto r = w->run(g, use_slow_path, topdirh, data);
+                  if(!g.owns_lock())
                   {
-                    run_error = std::move(r).error();
+                    g.lock();
+                  }
+                  if(!r)
+                  {
+                    done = true;
+                    if(!run_error)
+                    {
+                      run_error = std::move(r).error();
+                    }
+                    break;
                   }
-                  break;
                 }
-              }
-              state.threads_sleeping++;
-              maincond.notify_all();
-            },
-            &workers[n]));
+                state.threads_running--;
+                maincond.notify_all();
+              },
+              &workers[n]));
+            }
           }
           {
             std::unique_lock<std::mutex> g(state.lock);
@@ -436,7 +454,12 @@ namespace algorithm
               maincond.wait(g);
             }
             done = true;
-            cond.notify_all();
+            while(state.threads_running > 0)
+            {
+              g.unlock();
+              cond.notify_all();
+              g.lock();
+            }
           }
           for(auto &i : workerthreads)
           {