From 0fdded060618495a24caa55ec66d1538bc65c730 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Adrien=20B=C3=A9raud?= <adrien.beraud@savoirfairelinux.com>
Date: Sat, 19 Aug 2023 13:58:03 -0400
Subject: [PATCH] thread pool: allow pool to reduce in size after some time

---
 include/opendht/thread_pool.h |  9 ++++-
 src/thread_pool.cpp           | 67 ++++++++++++++++++++++++++++++-----
 2 files changed, 66 insertions(+), 10 deletions(-)

diff --git a/include/opendht/thread_pool.h b/include/opendht/thread_pool.h
index cdece0b0..1c4bfffc 100644
--- a/include/opendht/thread_pool.h
+++ b/include/opendht/thread_pool.h
@@ -37,7 +37,7 @@ public:
     static ThreadPool& io();
 
     ThreadPool();
-    ThreadPool(size_t maxThreads);
+    ThreadPool(unsigned minThreads, unsigned maxThreads = 0);
     ~ThreadPool();
 
     void run(std::function<void()>&& cb);
@@ -72,7 +72,14 @@ private:
     unsigned readyThreads_ {0};
     bool running_ {true};
 
+    unsigned minThreads_;
     const unsigned maxThreads_;
+    std::chrono::steady_clock::duration threadExpirationDelay {std::chrono::minutes(1)};
+    double threadDelayRatio_ {2};
+
+    std::vector<std::reference_wrapper<std::thread>> endedThreads_ {};
+    void threadEnded(std::thread&);
+    void cleanupThreads();
 };
 
 class OPENDHT_PUBLIC Executor : public std::enable_shared_from_this<Executor> {
diff --git a/src/thread_pool.cpp b/src/thread_pool.cpp
index 08469ebc..a5f4be57 100644
--- a/src/thread_pool.cpp
+++ b/src/thread_pool.cpp
@@ -38,18 +38,23 @@ ThreadPool::computation()
 ThreadPool&
 ThreadPool::io()
 {
-    static ThreadPool pool(IO_THREADS_MAX);
+    static ThreadPool pool(std::thread::hardware_concurrency(), IO_THREADS_MAX);
     return pool;
 }
 
 
-ThreadPool::ThreadPool(size_t maxThreads) : maxThreads_(maxThreads)
+ThreadPool::ThreadPool(unsigned minThreads, unsigned maxThreads)
+ : minThreads_(std::max(minThreads, 1u))
+ , maxThreads_(maxThreads ? std::max(minThreads_, maxThreads) : minThreads_)
 {
     threads_.reserve(maxThreads_);
+    if (minThreads_ != maxThreads_) {
+        threadDelayRatio_ = std::pow(2, 1.0 / (maxThreads_ - minThreads_));
+    }
 }
 
 ThreadPool::ThreadPool()
- : ThreadPool(std::max<size_t>(std::thread::hardware_concurrency(), 4))
+ : ThreadPool(std::max(std::thread::hardware_concurrency(), 4u))
 {}
 
 ThreadPool::~ThreadPool()
@@ -66,7 +71,9 @@ ThreadPool::run(std::function<void()>&& cb)
     // launch new thread if necessary
     if (not readyThreads_ && threads_.size() < maxThreads_) {
         try {
-            threads_.emplace_back(std::make_unique<std::thread>([this]() {
+            bool permanent_thread = threads_.size() < minThreads_;
+            auto& thread = *threads_.emplace_back(std::make_unique<std::thread>());
+            thread = std::thread([this, permanent_thread, e=threadExpirationDelay, &thread]() {
                 while (true) {
                     std::function<void()> task;
 
@@ -74,11 +81,13 @@ ThreadPool::run(std::function<void()>&& cb)
                     {
                         std::unique_lock<std::mutex> l(lock_);
                         readyThreads_++;
-                        cv_.wait(l, [&](){
-                            return not running_ or not tasks_.empty();
-                        });
+                        auto waitCond = [&](){ return not running_ or not tasks_.empty(); };
+                        if (permanent_thread)
+                            cv_.wait(l, waitCond);
+                        else 
+                            cv_.wait_for(l, e, waitCond);
                         readyThreads_--;
-                        if (not running_)
+                        if (not running_ or tasks_.empty())
                             break;
                         task = std::move(tasks_.front());
                         tasks_.pop();
@@ -92,7 +101,9 @@ ThreadPool::run(std::function<void()>&& cb)
                         std::cerr << "Exception running task: " << e.what() << std::endl;
                     }
                 }
-            }));
+                if (not permanent_thread)
+                    threadEnded(thread);
+            });
         } catch(const std::exception& e) {
             std::cerr << "Exception starting thread: " << e.what() << std::endl;
             if (threads_.empty())
@@ -107,6 +118,43 @@ ThreadPool::run(std::function<void()>&& cb)
     cv_.notify_one();
 }
 
+void
+ThreadPool::threadEnded(std::thread& thread)
+{
+    std::lock_guard<std::mutex> l(lock_);
+    endedThreads_.emplace_back(thread);
+    tasks_.emplace([this]{
+        cleanupThreads();
+    });
+    // A thread expired, maybe after handling a one-time burst of tasks.
+    // If we start new threads later, increase the expiration delay.
+    if (threadExpirationDelay > std::chrono::hours(168)) {
+        // If we reach 7 days, assume the thread pool is often used at max capacity
+        minThreads_ = std::min(minThreads_+1, maxThreads_);
+    } else {
+        threadExpirationDelay *= threadDelayRatio_;
+    }
+
+    cv_.notify_one();
+}
+
+void
+ThreadPool::cleanupThreads()
+{
+    std::lock_guard<std::mutex> l(lock_);
+    if (not endedThreads_.empty()) {
+        for (auto& thread : endedThreads_)
+            thread.get().join();
+        endedThreads_.clear();
+        for (auto it = threads_.begin(); it != threads_.end();) {
+            if (not (*it)->joinable())
+                it = threads_.erase(it);
+            else
+                ++it;
+        }
+    }
+}
+
 void
 ThreadPool::stop()
 {
@@ -123,6 +171,7 @@ ThreadPool::join()
     for (auto& t : threads_)
         t->join();
     threads_.clear();
+    tasks_ = {};
 }
 
 void