Improve thread pool (#571)

* WIP

* WIP

* WIP

* WIP

* Update core/utils/async_utils.cpp

core/utils/async_utils.hpp

@@ -23,10 +23,9 @@
 
 #pragma once
 
-#include <atomic>
 #include <condition_variable>
 #include <function2/function2.hpp>
-#include <functional>
+#include <mutex>
 #include <queue>
 #include <thread>
 
@@ -35,8 +34,7 @@
 #include "string.hpp"
 #include "thread_utils.hpp"
 
-namespace irs {
-namespace async_utils {
+namespace irs::async_utils {
 
 //////////////////////////////////////////////////////////////////////////////
 /// @brief spinlock implementation for Win32 since std::mutex cannot be used
@@ -52,86 +50,59 @@ class busywait_mutex final {
  std::atomic<bool> locked_{false};
 };
 
-template<bool UsePriority = true>
-class thread_pool {
+template<bool UseDelay = true>
+class ThreadPool {
  public:
- using native_char_t = std::remove_pointer_t<thread_name_t>;
- using clock_t = std::chrono::steady_clock;
- using func_t = fu2::unique_function<void()>;
-
- explicit thread_pool(size_t max_threads = 0, size_t max_idle = 0,
- std::basic_string_view<native_char_t> worker_name =
- kEmptyStringView<native_char_t>);
- ~thread_pool();
- size_t max_idle() const;
- void max_idle(size_t value);
- void max_idle_delta(int delta); // change value by delta
- size_t max_threads() const;
- void max_threads(size_t value);
- void max_threads_delta(int delta); // change value by delta
-
- // 1st - max_threads(), 2nd - max_idle()
- std::pair<size_t, size_t> limits() const;
- void limits(size_t max_threads, size_t max_idle);
-
- bool run(func_t&& fn, [[maybe_unused]] clock_t::duration delay = {});
- void stop(bool skip_pending = false); // always a blocking call
- size_t tasks_active() const;
- size_t tasks_pending() const;
- size_t threads() const;
+ using Char = std::remove_pointer_t<thread_name_t>;
+ using Clock = std::chrono::steady_clock;
+ using Func = fu2::unique_function<void()>;
+
+ explicit ThreadPool(size_t threads, basic_string_view<Char> name = {});
+ ~ThreadPool() { stop(true); }
+
+ bool run(Func&& fn, Clock::duration delay = {});
+ void stop(bool skip_pending = false) noexcept; // always a blocking call
+ size_t tasks_active() const {
+ std::lock_guard lock{m_};
+ return state_ / 2;
+ }
+ size_t tasks_pending() const {
+ std::lock_guard lock{m_};
+ return tasks_.size();
+ }
+ size_t threads() const {
+ std::lock_guard lock{m_};
+ return threads_.size();
+ }
  // 1st - tasks active(), 2nd - tasks pending(), 3rd - threads()
- std::tuple<size_t, size_t, size_t> stats() const;
+ std::tuple<size_t, size_t, size_t> stats() const {
+ std::lock_guard lock{m_};
+ return {state_ / 2, tasks_.size(), threads_.size()};
+ }
 
  private:
- enum class State { ABORT, FINISH, RUN };
-
- auto& next() {
- if constexpr (UsePriority) {
- return queue_.top();
- } else {
- return queue_.front();
- }
- }
+ struct Task {
+ explicit Task(Func&& fn, Clock::time_point at)
+ : at{at}, fn{std::move(fn)} {}
 
- template<typename T>
- static func_t& func(T& t) {
- if constexpr (UsePriority) {
- return const_cast<func_t&>(t.fn);
- } else {
- return const_cast<func_t&>(t);
- }
- }
+ Clock::time_point at;
+ Func fn;
 
- struct shared_state {
- std::mutex lock;
- std::condition_variable cond;
- std::atomic<State> state{State::RUN};
+ bool operator<(const Task& rhs) const noexcept { return rhs.at < at; }
  };
 
- struct task {
- explicit task(func_t&& fn, clock_t::time_point at)
- : at(at), fn(std::move(fn)) {}
+ void Work();
 
- clock_t::time_point at;
- func_t fn;
-
- bool operator<(const task& rhs) const noexcept { return rhs.at < at; }
- };
+ bool WasStop() const { return state_ % 2 != 0; }
 
- void worker(std::shared_ptr<shared_state> shared_state) noexcept;
- void worker_impl(std::unique_lock<std::mutex>& lock,
- std::shared_ptr<shared_state> shared_state);
- bool maybe_spawn_worker();
-
- std::shared_ptr<shared_state> shared_state_;
- size_t active_{0};
- std::atomic<size_t> threads_{0};
- size_t max_idle_;
- size_t max_threads_;
- std::conditional_t<UsePriority, std::priority_queue<task>, std::queue<func_t>>
- queue_;
- basic_string<native_char_t> worker_name_;
+ basic_string<Char> name_;
+ std::vector<std::thread> threads_;
+ mutable std::mutex m_;
+ std::condition_variable cv_;
+ std::conditional_t<UseDelay, std::priority_queue<Task>, std::queue<Func>>
+ tasks_;
+ // stop flag and active tasks counter
+ uint64_t state_ = 0;
 };
 
-} // namespace async_utils
-} // namespace irs
+} // namespace irs::async_utils

tests/index/index_profile_tests.cpp

@@ -126,7 +126,7 @@ class index_profile_test_case : public tests::index_test_base {
  std::atomic<size_t> writer_import_count(0);
  auto thread_count = (std::max)((size_t)1, num_insert_threads);
  auto total_threads = thread_count + num_import_threads + num_update_threads;
- irs::async_utils::thread_pool thread_pool(total_threads, total_threads);
+ irs::async_utils::ThreadPool<> thread_pool(total_threads);
  std::mutex mutex;
 
  if (!writer) {
@@ -473,7 +473,7 @@ class index_profile_test_case : public tests::index_test_base {
  size_t cleanup_interval) {
  auto* directory = &dir();
  std::atomic<bool> working(true);
- irs::async_utils::thread_pool thread_pool(1, 1);
+ irs::async_utils::ThreadPool<> thread_pool(1);
 
  thread_pool.run([cleanup_interval, directory, &working]() -> void {
  while (working.load()) {
@@ -502,7 +502,7 @@ class index_profile_test_case : public tests::index_test_base {
  options.segment_count_max = 8; // match original implementation or may run
  // out of file handles (e.g. MacOS/Travis)
 
- irs::async_utils::thread_pool thread_pool(commit_threads, commit_threads);
+ irs::async_utils::ThreadPool<> thread_pool(commit_threads);
  auto writer = open_writer(irs::OM_CREATE, options);
 
  for (size_t i = 0; i < commit_threads; ++i) {
@@ -536,7 +536,7 @@ class index_profile_test_case : public tests::index_test_base {
  irs::index_utils::MakePolicy(irs::index_utils::ConsolidateCount());
  irs::IndexWriterOptions options;
  std::atomic<bool> working(true);
- irs::async_utils::thread_pool thread_pool(2, 2);
+ irs::async_utils::ThreadPool<> thread_pool(2);
 
  options.segment_count_max = 8; // match original implementation or may run
  // out of file handles (e.g. MacOS/Travis)

tests/index/index_tests.cpp

@@ -537,7 +537,7 @@ class index_test_case : public tests::index_test_base {
 
  // validate terms async
  {
- irs::async_utils::thread_pool pool(thread_count, thread_count);
+ irs::async_utils::ThreadPool<> pool(thread_count);
 
  {
  std::lock_guard<std::mutex> lock(mutex);
@@ -580,7 +580,7 @@ class index_test_case : public tests::index_test_base {
  ASSERT_EQ(expected_term_itrs[i]->value(), actual_term_itr->value());
  }
 
- irs::async_utils::thread_pool pool(thread_count, thread_count);
+ irs::async_utils::ThreadPool<> pool(thread_count);
 
  {
  std::lock_guard<std::mutex> lock(mutex);

tests/store/directory_test_case.cpp

@@ -374,7 +374,7 @@ TEST_P(directory_test_case, read_multiple_streams) {
  auto in = dir_->open("test_async", irs::IOAdvice::NORMAL);
  std::mutex in_mtx;
  std::mutex mutex;
- irs::async_utils::thread_pool pool(16, 16);
+ irs::async_utils::ThreadPool<> pool(16);
 
  ASSERT_FALSE(!in);
  in = in->reopen();
@@ -383,7 +383,7 @@ TEST_P(directory_test_case, read_multiple_streams) {
  {
  std::lock_guard<std::mutex> lock(mutex);
 
- for (auto i = pool.max_threads(); i; --i) {
+ for (auto i = pool.threads(); i; --i) {
  pool.run([&in, &in_mtx, &mutex]() -> void {
  index_input::ptr input;