thread-pool: add thread pool

core/src/lib/CMakeLists.txt

@@ -99,6 +99,7 @@ set(BAREOS_SRCS
     tree.cc
     try_tls_handshake_as_a_server.cc
     compression.cc
+    thread_pool.cc
     util.cc
     var.cc
     watchdog.cc

core/src/lib/thread_pool.cc

@@ -0,0 +1,157 @@
+/*
+   BAREOS® - Backup Archiving REcovery Open Sourced
+
+   Copyright (C) 2023-2023 Bareos GmbH & Co. KG
+
+   This program is Free Software; you can redistribute it and/or
+   modify it under the terms of version three of the GNU Affero General Public
+   License as published by the Free Software Foundation and included
+   in the file LICENSE.
+
+   This program is distributed in the hope that it will be useful, but
+   WITHOUT ANY WARRANTY; without even the implied warranty of
+   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
+   Affero General Public License for more details.
+
+   You should have received a copy of the GNU Affero General Public License
+   along with this program; if not, write to the Free Software
+   Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA
+   02110-1301, USA.
+*/
+#include "lib/thread_pool.h"
+
+thread_pool::thread_pool(std::size_t num_threads)
+{
+  if (num_threads) ensure_num_workers(num_threads);
+}
+
+void thread_pool::enqueue(task&& t)
+{
+  bool task_submitted = false;
+
+  if (auto wlocked = workers.lock(); wlocked->num_actual_workers() > 0) {
+    auto locked = queue.lock();
+    if (locked->has_value()) {
+      locked->value().emplace_back(std::move(t));
+      tasks_submitted += 1;
+      queue_or_death.notify_one();
+      task_submitted = true;
+    }
+  }
+
+  if (!task_submitted) { t(); }
+}
+
+void thread_pool::ensure_num_workers(std::size_t num_threads)
+{
+  auto locked = workers.lock();
+  locked->min_workers = std::max(locked->min_workers, num_threads);
+  for (std::size_t i = locked->threads.size(); i < num_threads; ++i) {
+    locked->threads.emplace_back(&pool_work, i, this);
+  }
+}
+
+void thread_pool::finish()
+{
+  auto locked = tasks_completed.lock();
+
+  locked.wait(on_task_completion,
+              [this](auto completed) { return completed == tasks_submitted; });
+}
+
+thread_pool::~thread_pool()
+{
+  queue.lock()->reset();
+  queue_or_death.notify_all();
+
+  auto locked = workers.lock();
+  locked.wait(worker_death, [](const auto& pool) {
+    return pool.dead_workers == pool.threads.size();
+  });
+
+  for (auto& thread : locked->threads) { thread.join(); }
+}
+
+void thread_pool::borrow_then_pool_work(task&& t, size_t id, thread_pool* pool)
+{
+  {
+    auto fun = std::move(t);
+    fun();
+  }
+
+  {
+    auto locked = pool->workers.lock();
+    locked->num_borrowed -= 1;
+  }
+
+  pool_work(id, pool);
+}
+
+void thread_pool::borrow_thread(task&& t)
+{
+  enqueue([this, t = std::move(t)]() {
+    {
+      auto locked = workers.lock();
+      auto min_left_over_workers
+          = std::max(std::size_t{1}, locked->min_workers);
+      if (locked->num_actual_workers() <= min_left_over_workers) {
+        locked->threads.emplace_back(borrow_then_pool_work, std::move(t),
+                                     locked->threads.size(), this);
+        locked->num_borrowed += 1;
+        return;
+      }
+      locked->num_borrowed += 1;
+    }
+
+    {
+      auto fun = std::move(t);
+      fun();
+    }
+
+    {
+      auto locked = workers.lock();
+      locked->num_borrowed -= 1;
+    }
+  });
+}
+
+void thread_pool::pool_work(std::size_t id, thread_pool* pool)
+{
+  // id is kept here for debugging purposes.
+  (void)id;
+  try {
+    for (std::optional my_task = pool->dequeue(); my_task.has_value();
+         my_task = pool->finish_and_dequeue()) {
+      std::optional t = std::move(my_task);
+      my_task.reset();
+      (*t)();
+    }
+
+    pool->workers.lock()->dead_workers += 1;
+    pool->worker_death.notify_one();
+  } catch (...) {
+    pool->workers.lock()->dead_workers += 1;
+    pool->worker_death.notify_one();
+    throw;
+  }
+}
+
+auto thread_pool::dequeue() -> std::optional<task>
+{
+  auto locked = queue.lock();
+  locked.wait(queue_or_death, [](auto& queue) {
+    return !queue.has_value() || queue->size() > 0;
+  });
+  if (!locked->has_value()) { return std::nullopt; }
+  task t = std::move(locked->value().front());
+  locked->value().pop_front();
+  return t;
+}
+
+auto thread_pool::finish_and_dequeue() -> std::optional<task>
+{
+  *tasks_completed.lock() += 1;
+  on_task_completion.notify_all();
+
+  return dequeue();
+}

core/src/lib/thread_pool.h

@@ -0,0 +1,143 @@
+/*
+   BAREOS® - Backup Archiving REcovery Open Sourced
+
+   Copyright (C) 2023-2023 Bareos GmbH & Co. KG
+
+   This program is Free Software; you can redistribute it and/or
+   modify it under the terms of version three of the GNU Affero General Public
+   License as published by the Free Software Foundation and included
+   in the file LICENSE.
+
+   This program is distributed in the hope that it will be useful, but
+   WITHOUT ANY WARRANTY; without even the implied warranty of
+   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
+   Affero General Public License for more details.
+
+   You should have received a copy of the GNU Affero General Public License
+   along with this program; if not, write to the Free Software
+   Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA
+   02110-1301, USA.
+*/
+/**
+ * @file
+ * simple thread pool
+ */
+
+#ifndef BAREOS_LIB_THREAD_POOL_H_
+#define BAREOS_LIB_THREAD_POOL_H_
+
+#include <vector>
+#include <thread>
+#include <optional>
+#include <deque>
+#include <mutex>
+#include <shared_mutex>
+#include <condition_variable>
+#include <future>
+#include <type_traits>
+#include <exception>
+
+#include "lib/thread_util.h"
+
+struct worker_pool {
+  std::vector<std::thread> threads{};
+  std::size_t num_borrowed{0};
+  std::size_t min_workers{0};
+  std::size_t dead_workers{0};
+
+  std::size_t num_actual_workers() const
+  {
+    return threads.size() - num_borrowed - dead_workers;
+  }
+};
+
+class thread_pool {
+ public:
+  // fixme(ssura): change this to std::move_only_function (c++23)
+  using task = std::function<void()>;
+
+  thread_pool(std::size_t num_threads = 0);
+
+  void ensure_num_workers(std::size_t num_threads);
+  void finish();   // wait until all submitted tasks are accomplished
+  ~thread_pool();  // stops as fast as possible; dropping outstanding tasks
+
+  void enqueue(task&& t);  // queue task to be worked on by worker threads;
+                           // should not block
+  void borrow_thread(task&& t);  // steal thread to work on task t; can block
+
+ private:
+  std::condition_variable worker_death;
+  synchronized<worker_pool> workers{};
+
+  std::condition_variable queue_or_death;
+  synchronized<std::optional<std::deque<task>>> queue{std::in_place};
+
+  std::size_t tasks_submitted{0};
+  std::condition_variable on_task_completion;
+  synchronized<std::size_t> tasks_completed{0u};
+
+  static void pool_work(std::size_t id, thread_pool* pool);
+  static void borrow_then_pool_work(task&& t,
+                                    std::size_t id,
+                                    thread_pool* pool);
+
+  std::optional<task> dequeue();
+  std::optional<task> finish_and_dequeue();
+};
+
+template <typename F>
+auto enqueue(thread_pool& pool, F&& f) -> std::future<std::invoke_result_t<F>>
+{
+  using result_type = std::invoke_result_t<F>;
+  // todo(ssura): in c++23 we can use std::move_only_function
+  //              and pass the promise directly into the function.
+  //              currently this approach does not work because
+  //              std::function requires the function to be copyable,
+  //              but std::promise obviously is not.
+  std::shared_ptr p = std::make_shared<std::promise<result_type>>();
+  std::future fut = p->get_future();
+  pool.enqueue([f = std::move(f), mp = std::move(p)]() mutable {
+    try {
+      if constexpr (std::is_same_v<result_type, void>) {
+        f();
+        mp->set_value();
+      } else {
+        mp->set_value(f());
+      }
+    } catch (...) {
+      mp->set_exception(std::current_exception());
+    }
+  });
+  return fut;
+}
+
+template <typename F>
+auto borrow_thread(thread_pool& pool, F&& f)
+    -> std::future<std::invoke_result_t<F>>
+{
+  using result_type = std::invoke_result_t<F>;
+  // todo(ssura): in c++23 we can use std::move_only_function
+  //              and pass the promise directly into the function.
+  //              currently this approach does not work because
+  //              std::function requires the function to be copyable,
+  //              but std::promise obviously is not.
+  std::shared_ptr p = std::make_shared<std::promise<result_type>>();
+  std::future fut = p->get_future();
+  pool.borrow_thread([f = std::move(f), mp = std::move(p)]() mutable {
+    try {
+      if constexpr (std::is_same_v<result_type, void>) {
+        f();
+        mp->set_value();
+      } else {
+        mp->set_value(f());
+      }
+    } catch (...) {
+      mp->set_exception(std::current_exception());
+    }
+  });
+  return fut;
+}
+
+
+#endif  // BAREOS_LIB_THREAD_POOL_H_