checkpoint

libfuse/lib/bounded_queue.hpp

@@ -13,25 +13,26 @@ public:
   explicit
   BoundedQueue(std::size_t max_size_,
                bool        block_ = true)
-    : _block(block),
+    : _block(block_),
       _max_size(max_size_)
   {
-    if(_max_size == 0)
-      _max_size = 1;
   }
 
+  BoundedQueue(const BoundedQueue&) = delete;
+  BoundedQueue(BoundedQueue&&) = default;
+
   bool
   push(const T& item_)
   {
     {
-      std::unique_lock guard(_queue_lock);
+      std::unique_lock<std::mutex> guard(_queue_lock);
 
       _condition_push.wait(guard, [&]() { return _queue.size() < _max_size || !_block; });
 
       if(_queue.size() == _max_size)
         return false;
 
-      _queue.push(item);
+      _queue.push(item_);
     }
 
     _condition_pop.notify_one();
@@ -43,7 +44,7 @@ public:
   push(T&& item_)
   {
     {
-      std::unique_lock guard(_queue_lock);
+      std::unique_lock<std::mutex> guard(_queue_lock);
 
       _condition_push.wait(guard, [&]() { return _queue.size() < _max_size || !_block; });
 
@@ -61,7 +62,7 @@ public:
   emplace(Args&&... args_)
   {
     {
-      std::unique_lock guard(_queue_lock);
+      std::unique_lock<std::mutex> guard(_queue_lock);
 
       _condition_push.wait(guard, [&]() { return _queue.size() < _max_size || !_block; });
 
@@ -80,7 +81,7 @@ public:
   pop(T& item_)
   {
     {
-      std::unique_lock guard(_queue_lock);
+      std::unique_lock<std::mutex> guard(_queue_lock);
 
       _condition_pop.wait(guard, [&]() { return !_queue.empty() || !_block; });
       if(_queue.empty())
@@ -99,7 +100,7 @@ public:
   std::size_t
   size() const
   {
-    std::lock_guard guard(_queue_lock);
+    std::lock_guard<std::mutex> guard(_queue_lock);
 
     return _queue.size();
   }
@@ -113,7 +114,7 @@ public:
   bool
   empty() const
   {
-    std::lock_guard guard(_queue_lock);
+    std::lock_guard<std::mutex> guard(_queue_lock);
 
     return _queue.empty();
   }
@@ -121,7 +122,7 @@ public:
   bool
   full() const
   {
-    std::lock_guard lock(_queue_lock);
+    std::lock_guard<std::mutex> lock(_queue_lock);
 
     return (_queue.size() == capacity());
   }
@@ -129,7 +130,7 @@ public:
   void
   block()
   {
-    std::lock_guard guard(_queue_lock);
+    std::lock_guard<std::mutex> guard(_queue_lock);
     _block = true;
   }
 
@@ -137,7 +138,7 @@ public:
   unblock()
   {
     {
-      std::lock_guard guard(_queue_lock);
+      std::lock_guard<std::mutex> guard(_queue_lock);
       _block = false;
     }
 
@@ -148,7 +149,7 @@ public:
   bool
   blocking() const
   {
-    std::lock_guard guard(_queue_lock);
+    std::lock_guard<std::mutex> guard(_queue_lock);
 
     return _block;
   }

libfuse/lib/bounded_thread_pool.hpp

@@ -0,0 +1,130 @@
+#pragma once
+
+#include "bounded_queue.hpp"
+
+#include <tuple>
+#include <atomic>
+#include <vector>
+#include <thread>
+#include <memory>
+#include <future>
+#include <utility>
+#include <functional>
+#include <type_traits>
+
+
+class BoundedThreadPool
+{
+private:
+  using Proc   = std::function<void(void)>;
+  using Queue  = BoundedQueue<Proc>;
+  using Queues = std::vector<std::shared_ptr<Queue>>;
+  
+public:
+  explicit
+  BoundedThreadPool(const std::size_t thread_count_ = std::thread::hardware_concurrency())
+    : _queues(),
+      _count(thread_count_)
+  {
+    printf("threads: %d\n",thread_count_);
+    for(std::size_t i = 0; i < thread_count_; i++)
+      _queues.emplace_back(std::make_shared<Queue>(1));
+    
+    auto worker = [this](std::size_t i)
+    {
+      while(true)
+        {
+          Proc f;
+
+          for(std::size_t n = 0; n < (_count * K); ++n)
+            {
+              if(_queues[(i + n) % _count]->pop(f))
+                break;
+            }
+
+          if(!f && !_queues[i]->pop(f))
+            break;
+
+          f();
+        }
+    };
+
+    _threads.reserve(thread_count_);
+    for(std::size_t i = 0; i < thread_count_; ++i)
+      _threads.emplace_back(worker, i);
+  }
+
+  ~BoundedThreadPool()
+  {
+    for(auto& queue : _queues)
+      queue->unblock();
+    for(auto& thread : _threads)
+      thread.join();
+  }
+
+  template<typename F>
+  void
+  enqueue_work(F&& f_)
+  {
+    auto i = _index++;
+
+    for(std::size_t n = 0; n < (_count * K); ++n)
+      {
+        if(_queues[(i + n) % _count]->push(f_))
+          return;
+      }
+
+    _queues[i % _count]->push(std::move(f_));
+  }
+
+  template<typename F>
+  [[nodiscard]]
+  std::future<typename std::result_of<F()>::type>
+  enqueue_task(F&& f_)
+  {
+    using TaskReturnType = typename std::result_of<F()>::type;
+    using Promise        = std::promise<TaskReturnType>;
+
+    auto i       = _index++;
+    auto promise = std::make_shared<Promise>();
+    auto future  = promise->get_future();
+    auto work    = [=]() {
+      auto rv = f_();
+      promise->set_value(rv);
+    };
+
+    for(std::size_t n = 0; n < (_count * K); ++n)
+      {
+        if(_queues[(i + n) % _count]->push(work))
+          return future;
+      }
+
+    _queues[i % _count]->push(std::move(work));
+
+    return future;
+  }
+
+public:
+  std::vector<pthread_t>
+  threads()
+  {
+    std::vector<pthread_t> rv;
+
+    for(auto &thread : _threads)
+      rv.push_back(thread.native_handle());
+
+    return rv;
+  }
+
+private:
+  Queues _queues;
+
+private:
+  std::vector<std::thread> _threads;
+
+private:
+  const std::size_t _count;
+  std::atomic_uint _index;
+
+  static const unsigned int K = 2;
+};

libfuse/lib/fuse.c

@@ -3730,6 +3730,11 @@ metrics_log_nodes_info(struct fuse *f_,
                        FILE        *file_)
 {
   char buf[1024];
+  uint64_t time_now;
+  uint64_t sizeof_node;
+
+  time_now = time(NULL);
+  sizeof_node = sizeof(struct node);
 
   lfmp_lock(&f_->node_fmp);
   snprintf(buf,sizeof(buf),
@@ -3745,10 +3750,12 @@ metrics_log_nodes_info(struct fuse *f_,
            "node memory pool usage ratio: %f\n"
            "node memory pool avail objs: %"PRIu64"\n"
            "node memory pool total allocated memory: %"PRIu64"\n"
+           "msgbuf allocation count: %"PRIu64"\n"
+           "msgbuf available count: %"PRIu64"\n"
            "\n"
            ,
-           (uint64_t)time(NULL),
-           (uint64_t)sizeof(struct node),
+           (uint64_t)time_now,
+           (uint64_t)sizeof_node,
            (uint64_t)f_->id_table.size,
            (uint64_t)f_->id_table.use,
            (uint64_t)(f_->id_table.size * sizeof(struct node*)),
@@ -3758,7 +3765,9 @@ metrics_log_nodes_info(struct fuse *f_,
            (uint64_t)fmp_slab_count(&f_->node_fmp.fmp),
            fmp_slab_usage_ratio(&f_->node_fmp.fmp),
            (uint64_t)fmp_avail_objs(&f_->node_fmp.fmp),
-           (uint64_t)fmp_total_allocated_memory(&f_->node_fmp.fmp)
+           (uint64_t)fmp_total_allocated_memory(&f_->node_fmp.fmp),
+           (uint64_t)msgbuf_alloc_count(),
+           msgbuf_avail_count()
            );
   lfmp_unlock(&f_->node_fmp);
 
@@ -3818,7 +3827,11 @@ fuse_maintenance_loop(void *fuse_)
 
       // Trigger a followup gc if this gc succeeds
       if(!f->conf.nogc && gc)
-        gc = lfmp_gc(&f->node_fmp);
+        {
+          gc = lfmp_gc(&f->node_fmp);
+        }
+
+      //msgbuf_gc();
 
       if(g_LOG_METRICS)
         metrics_log_nodes_info_to_tmp_dir(f);

libfuse/lib/fuse_loop_mt.cpp

@@ -2,7 +2,7 @@
 #define _GNU_SOURCE
 #endif
 
-#include "thread_pool.hpp"
+#include "bounded_thread_pool.hpp"
 #include "cpu.hpp"
 #include "fmt/core.h"
 
@@ -33,7 +33,7 @@ struct fuse_worker_data_t
   struct fuse_session *se;
   sem_t finished;
   std::function<void(fuse_worker_data_t*,fuse_msgbuf_t*)> msgbuf_processor;
-  std::shared_ptr<ThreadPool> tp;
+  std::shared_ptr<BoundedThreadPool> tp;
 };
 
 class WorkerCleanup
@@ -444,7 +444,7 @@ fuse_session_loop_mt(struct fuse_session *se_,
 
   if(process_thread_count > 0)
     {
-      wd.tp = std::make_shared<ThreadPool>(process_thread_count);
+      wd.tp = std::make_shared<BoundedThreadPool>(process_thread_count);
       wd.msgbuf_processor = process_msgbuf_async;
       process_threads = wd.tp->threads();
     }

libfuse/lib/fuse_msgbuf.cpp

@@ -23,14 +23,18 @@
 #include <cstdint>
 #include <cstdlib>
 #include <mutex>
-#include <stack>
+#include <vector>
+#include <unordered_set>
+#include <atomic>
 
 
-static std::uint32_t g_PAGESIZE = 0;
-static std::uint32_t g_BUFSIZE  = 0;
+static std::uint32_t g_PAGESIZE   = 0;
+static std::uint32_t g_BUFSIZE    = 0;
+static std::uint32_t g_MAX_ALLOCS = 128;
 
 static std::mutex g_MUTEX;
-static std::stack<fuse_msgbuf_t*> g_MSGBUF_STACK;
+static std::vector<fuse_msgbuf_t*> g_MSGBUF_STACK;
+static std::unordered_set<fuse_msgbuf_t*> g_MSGBUF_ALLOCED;
 
 static
 __attribute__((constructor))
@@ -38,6 +42,7 @@ void
 msgbuf_constructor()
 {
   g_PAGESIZE = sysconf(_SC_PAGESIZE);
+  // +2 because to do O_DIRECT we need to offset the buffer to align
   g_BUFSIZE  = (g_PAGESIZE * (FUSE_MAX_MAX_PAGES + 2));
 }
 
@@ -46,7 +51,7 @@ __attribute__((destructor))
 void
 msgbuf_destroy()
 {
-  // TODO: cleanup?
+
 }
 
 uint32_t
@@ -79,12 +84,10 @@ fuse_msgbuf_t*
 msgbuf_alloc()
 {
   fuse_msgbuf_t *msgbuf;
+  std::lock_guard<std::mutex> lck(g_MUTEX);
 
-  g_MUTEX.lock();
   if(g_MSGBUF_STACK.empty())
     {
-      g_MUTEX.unlock();
-
       msgbuf = (fuse_msgbuf_t*)malloc(sizeof(fuse_msgbuf_t));
       if(msgbuf == NULL)
         return NULL;
@@ -97,12 +100,13 @@ msgbuf_alloc()
         }
 
       msgbuf->size = g_BUFSIZE;
+
+      g_MSGBUF_ALLOCED.emplace(msgbuf);
     }
   else
     {
-      msgbuf = g_MSGBUF_STACK.top();
-      g_MSGBUF_STACK.pop();
-      g_MUTEX.unlock();
+      msgbuf = g_MSGBUF_STACK.back();
+      g_MSGBUF_STACK.pop_back();
     }
 
   return msgbuf;
@@ -115,10 +119,42 @@ msgbuf_free(fuse_msgbuf_t *msgbuf_)
 
   if(msgbuf_->size != g_BUFSIZE)
     {
+      g_MSGBUF_ALLOCED.erase(msgbuf_);
       free(msgbuf_->mem);
       free(msgbuf_);
       return;
     }
 
-  g_MSGBUF_STACK.push(msgbuf_);
+  g_MSGBUF_STACK.emplace_back(msgbuf_);
+}
+
+uint64_t
+msgbuf_alloc_count()
+{
+  return g_MSGBUF_ALLOCED.size();
+}
+
+uint64_t
+msgbuf_avail_count()
+{
+  return g_MSGBUF_STACK.size();
+}
+
+void
+msgbuf_gc()
+{
+  std::vector<fuse_msgbuf_t*> oldstack;
+
+  {
+    std::lock_guard<std::mutex> lck(g_MUTEX);
+    oldstack.swap(g_MSGBUF_STACK);
+  }
+
+  fprintf(stderr,"freeing %lu msgbufs\n",oldstack.size());
+  for(auto msgbuf: oldstack)
+    {
+      g_MSGBUF_ALLOCED.erase(msgbuf);
+      free(msgbuf->mem);
+      free(msgbuf);
+    }
 }

libfuse/lib/fuse_msgbuf.hpp

@@ -27,7 +27,11 @@ void     msgbuf_set_bufsize(const uint32_t size);
 uint32_t msgbuf_get_bufsize();
 
 fuse_msgbuf_t* msgbuf_alloc();
-fuse_msgbuf_t* msgbuf_alloc_memonly();
 void           msgbuf_free(fuse_msgbuf_t *msgbuf);
 
+void           msgbuf_gc();
+
+uint64_t       msgbuf_alloc_count();
+uint64_t       msgbuf_avail_count();
+
 EXTERN_C_END

libfuse/lib/thread_pool.hpp

@@ -1,6 +1,7 @@
 #pragma once
 
 #include "unbounded_queue.hpp"
+#include "bounded_queue.hpp"
 
 #include <tuple>
 #include <atomic>