libfuse: thread_pool/objpool/fuse_loop - safer memory and iteration

- ThreadPool::destroy used to iterate _threads directly while another
  thread could be adjusting the same vector. Snapshot the vector
  under the mutex and iterate the local copy for pthread_cancel /
  pthread_join.

- ObjPool::construct now returns NULL when the pooled allocator
  returns NULL instead of invoking placement-new on a null
  pointer. In destroy, compute to_node(obj) before running the
  destructor - the destructor may scrub bytes the to_node helper
  reads.

- In the async/sync workers in fuse_loop, msgbuf_alloc can return
  NULL under memory pressure. Fail the worker cleanly with a
  diagnostic rather than dereferencing. Also introduce a small
  _nproc() helper that treats hardware_concurrency()==0 as 1, so
  the downstream min/max arithmetic doesn't produce a zero thread
  count.

vendored/libfuse/include/objpool.hpp

@@ -214,6 +214,9 @@ private:
             static_cast<void*>(node) :
             _allocator.allocate(size_, alignof(T)));
 
+    if(not mem)
+      return nullptr;
+
     try
       {
         return new(mem) T(std::forward<Args>(args_)...);
@@ -242,12 +245,13 @@ public:
     if(not obj_)
       return;
 
-    bool should_pool = _should_pool(obj_);
+    bool  should_pool = _should_pool(obj_);
+    Node *node        = to_node(obj_);
 
     obj_->~T();
 
     if(should_pool)
-      _push_node(to_node(obj_),size_);
+      _push_node(node,size_);
     else
       _allocator.deallocate(obj_,size_,alignof(T));
   }

vendored/libfuse/include/thread_pool.hpp

@@ -180,14 +180,21 @@ ThreadPool::ThreadPool(const unsigned     thread_count_,
 inline
 ThreadPool::~ThreadPool()
 {
+  std::vector<pthread_t> threads;
+
+  {
+    mutex_lockguard(_threads_mutex);
+    threads = _threads;
+  }
+
   syslog(LOG_DEBUG,
          "threadpool (%s): destroying %zu threads",
          _name.c_str(),
-         _threads.size());
+         threads.size());
 
-  for(auto t : _threads)
+  for(auto t : threads)
     pthread_cancel(t);
-  for(auto t : _threads)
+  for(auto t : threads)
     pthread_join(t,NULL);
 }
 

vendored/libfuse/lib/fuse_loop.cpp

@@ -57,6 +57,13 @@ _print_error(int rv_)
              -rv_);
 }
 
+static
+int
+_nproc()
+{
+  return std::max(1,(int)std::thread::hardware_concurrency());
+}
+
 struct AsyncWorker
 {
   fuse_session *_se;
@@ -86,6 +93,11 @@ struct AsyncWorker
         fuse_msgbuf_t *msgbuf;
 
         msgbuf = msgbuf_alloc();
+        if(not msgbuf)
+          {
+            fmt::println(stderr,"mergerfs: failure to allocate msgbuf, exiting");
+            return;
+          }
 
         while(true)
           {
@@ -141,6 +153,11 @@ struct SyncWorker
         fuse_msgbuf_t *msgbuf;
 
         msgbuf = msgbuf_alloc();
+        if(not msgbuf)
+          {
+            fmt::println(stderr,"mergerfs: failure to allocate msgbuf, exiting");
+            return;
+          }
 
         while(true)
           {
@@ -166,31 +183,6 @@ struct SyncWorker
   }
 };
 
-int
-fuse_start_thread(pthread_t *thread_id,
-                  void      *(*func)(void *),
-                  void      *arg)
-{
-  int res;
-  sigset_t oldset;
-  sigset_t newset;
-
-  sigfillset(&newset);
-  pthread_sigmask(SIG_BLOCK,&newset,&oldset);
-  res = pthread_create(thread_id,NULL,func,arg);
-  pthread_sigmask(SIG_SETMASK,&oldset,NULL);
-
-  if(res != 0)
-    {
-      fprintf(stderr,
-              "fuse: error creating thread: %s\n",
-              strerror(res));
-      return -1;
-    }
-
-  return 0;
-}
-
 static
 int
 _calculate_thread_count(const int raw_thread_count_)
@@ -199,12 +191,12 @@ _calculate_thread_count(const int raw_thread_count_)
 
   if(raw_thread_count_ == 0)
     {
-      thread_count = std::thread::hardware_concurrency();
+      thread_count = ::_nproc();
       thread_count = std::min(8,thread_count);
     }
   else if(raw_thread_count_ < 0)
     {
-      thread_count = (std::thread::hardware_concurrency() / -raw_thread_count_);
+      thread_count = (::_nproc() / -raw_thread_count_);
       thread_count = std::max(1,thread_count);
     }
   else if(raw_thread_count_ > 0)
@@ -225,14 +217,14 @@ _calculate_thread_counts(int *read_thread_count_,
     {
       int nproc;
 
-      nproc = std::thread::hardware_concurrency();
+      nproc = ::_nproc();
       *read_thread_count_ = std::min(8,nproc);
     }
   else if((*read_thread_count_ == 0) && (*process_thread_count_ == 0))
     {
       int nproc;
 
-      nproc = std::thread::hardware_concurrency();
+      nproc = ::_nproc();
       *read_thread_count_ = 2;
       *process_thread_count_ = std::max(2,(nproc - 2));
       *process_thread_count_ = std::min(8,*process_thread_count_);