Browse Source

Make ThreadPool idle timeout configurable at runtime (Fix #2481)

yhirose 1 month ago
parent
commit
32ff75e355
3 changed files with 58 additions and 6 deletions
  1. 9 0
      README.md
  2. 11 6
      httplib.h
  3. 38 0
      test/test.cc

+ 9 - 0
README.md

@@ -824,6 +824,15 @@ svr.new_task_queue = [] { return new ThreadPool(/*base_threads=*/12, /*max_threa
 Default limit is 0 (unlimited). Once the limit is reached, the listener
 Default limit is 0 (unlimited). Once the limit is reached, the listener
 will shutdown the client connection.
 will shutdown the client connection.
 
 
+#### Idle timeout for dynamic threads
+
+The idle timeout for dynamic threads can also be set at runtime via the
+fourth parameter (in seconds):
+
+```cpp
+svr.new_task_queue = [] { return new ThreadPool(/*base_threads=*/8, /*max_threads=*/64, /*max_queued_requests=*/0, /*idle_timeout_sec=*/10); };
+```
+
 ### Override the default thread pool with yours
 ### Override the default thread pool with yours
 
 
 You can supply your own thread pool implementation according to your need.
 You can supply your own thread pool implementation according to your need.

+ 11 - 6
httplib.h

@@ -1541,7 +1541,9 @@ public:
 
 
 class ThreadPool final : public TaskQueue {
 class ThreadPool final : public TaskQueue {
 public:
 public:
-  explicit ThreadPool(size_t n, size_t max_n = 0, size_t mqr = 0);
+  explicit ThreadPool(
+      size_t n, size_t max_n = 0, size_t mqr = 0,
+      time_t idle_timeout_sec = CPPHTTPLIB_THREAD_POOL_IDLE_TIMEOUT);
   ThreadPool(const ThreadPool &) = delete;
   ThreadPool(const ThreadPool &) = delete;
   ~ThreadPool() override = default;
   ~ThreadPool() override = default;
 
 
@@ -1556,6 +1558,7 @@ private:
   size_t base_thread_count_;
   size_t base_thread_count_;
   size_t max_thread_count_;
   size_t max_thread_count_;
   size_t max_queued_requests_;
   size_t max_queued_requests_;
+  time_t idle_timeout_sec_;
   size_t idle_thread_count_;
   size_t idle_thread_count_;
 
 
   bool shutdown_;
   bool shutdown_;
@@ -10304,8 +10307,10 @@ inline ssize_t detail::BodyReader::read(char *buf, size_t len) {
 }
 }
 
 
 // ThreadPool implementation
 // ThreadPool implementation
-inline ThreadPool::ThreadPool(size_t n, size_t max_n, size_t mqr)
-    : base_thread_count_(n), max_queued_requests_(mqr), idle_thread_count_(0),
+inline ThreadPool::ThreadPool(size_t n, size_t max_n, size_t mqr,
+                              time_t idle_timeout_sec)
+    : base_thread_count_(n), max_queued_requests_(mqr),
+      idle_timeout_sec_(idle_timeout_sec), idle_thread_count_(0),
       shutdown_(false) {
       shutdown_(false) {
 #ifndef CPPHTTPLIB_NO_EXCEPTIONS
 #ifndef CPPHTTPLIB_NO_EXCEPTIONS
   if (max_n != 0 && max_n < n) {
   if (max_n != 0 && max_n < n) {
@@ -10415,9 +10420,9 @@ inline void ThreadPool::worker(bool is_dynamic) {
       idle_thread_count_++;
       idle_thread_count_++;
 
 
       if (is_dynamic) {
       if (is_dynamic) {
-        auto has_work = cond_.wait_for(
-            lock, std::chrono::seconds(CPPHTTPLIB_THREAD_POOL_IDLE_TIMEOUT),
-            [&] { return !jobs_.empty() || shutdown_; });
+        auto has_work =
+            cond_.wait_for(lock, std::chrono::seconds(idle_timeout_sec_),
+                           [&] { return !jobs_.empty() || shutdown_; });
         if (!has_work) {
         if (!has_work) {
           // Timed out with no work - exit this dynamic thread
           // Timed out with no work - exit this dynamic thread
           idle_thread_count_--;
           idle_thread_count_--;

+ 38 - 0
test/test.cc

@@ -13476,6 +13476,44 @@ TEST(TaskQueueTest, IncreaseAtomicIntegerWithQueueLimit) {
   EXPECT_TRUE(queued_count >= qlimit);
   EXPECT_TRUE(queued_count >= qlimit);
 }
 }
 
 
+TEST(TaskQueueTest, IdleTimeoutAtRuntime) {
+  // Use a short idle timeout so a dynamic thread spawns, times out and
+  // exits during the test, and the pool keeps working afterwards.
+  std::unique_ptr<TaskQueue> task_queue{new ThreadPool{
+      /*num_threads=*/1, /*max_threads=*/2, /*max_queued_requests=*/0,
+      /*idle_timeout_sec=*/1}};
+
+  std::atomic_uint count{0};
+  std::condition_variable cv;
+  std::mutex mtx;
+  bool release = false;
+
+  // Block the base thread so the second task spawns a dynamic thread.
+  EXPECT_TRUE(task_queue->enqueue([&] {
+    std::unique_lock<std::mutex> lock(mtx);
+    while (!release) {
+      cv.wait(lock);
+    }
+    count++;
+  }));
+  EXPECT_TRUE(task_queue->enqueue([&] { count++; }));
+
+  // Let the dynamic thread finish its task and exceed the idle timeout.
+  std::this_thread::sleep_for(std::chrono::milliseconds(1500));
+
+  {
+    std::unique_lock<std::mutex> lock(mtx);
+    release = true;
+  }
+  cv.notify_all();
+
+  // The pool must still accept and run tasks after the dynamic thread exited.
+  EXPECT_TRUE(task_queue->enqueue([&] { count++; }));
+
+  task_queue->shutdown();
+  EXPECT_EQ(3u, count.load());
+}
+
 TEST(TaskQueueTest, MaxQueuedRequests) {
 TEST(TaskQueueTest, MaxQueuedRequests) {
   static constexpr unsigned int qlimit{3};
   static constexpr unsigned int qlimit{3};
   std::unique_ptr<TaskQueue> task_queue{new ThreadPool{1, 1, qlimit}};
   std::unique_ptr<TaskQueue> task_queue{new ThreadPool{1, 1, qlimit}};