mirror of
https://github.com/yhirose/cpp-httplib.git
synced 2026-10-24 12:04:40 +00:00
Spawn a dynamic thread when queued jobs outnumber idle threads
ThreadPool::enqueue only spawned a dynamic thread when no thread was idle. Threads already notified for an earlier job still counted as idle until they took it, so a burst of connections could leave jobs queued with no thread while the pool was below max_threads. With keep-alive, such a connection waited until another connection closed. Fix #2633
This commit is contained in:
1 parent
062666f3a1
commit
e97165b5dc
2 files changed
+35
-2
No files matched your search
@@ -12423,8 +12423,8 @@ inline bool ThreadPool::enqueue(std::function<void()> fn) {
|
|||||||
}
|
}
|
||||||
jobs_.push_back(std::move(fn));
|
jobs_.push_back(std::move(fn));
|
||||||
|
|
||||||
// Spawn a dynamic thread if no idle threads and under max
|
// Spawn a dynamic thread if jobs outnumber idle threads and under max
|
||||||
if (idle_thread_count_ == 0 &&
|
if (jobs_.size() > idle_thread_count_ &&
|
||||||
threads_.size() + dynamic_threads_.size() < max_thread_count_) {
|
threads_.size() + dynamic_threads_.size() < max_thread_count_) {
|
||||||
cleanup_finished_threads();
|
cleanup_finished_threads();
|
||||||
dynamic_threads_.emplace_back(std::thread([this]() { worker(true); }));
|
dynamic_threads_.emplace_back(std::thread([this]() { worker(true); }));
|
||||||
|
|||||||
@@ -19553,6 +19553,39 @@ TEST(TaskQueueTest, IdleTimeoutAtRuntime) {
|
|||||||
EXPECT_EQ(3u, count.load());
|
EXPECT_EQ(3u, count.load());
|
||||||
}
|
}
|
||||||
|
|
||||||
|
TEST(TaskQueueTest, BurstOfTasksGetsDynamicThreads) {
|
||||||
|
static constexpr size_t number_of_tasks{32};
|
||||||
|
std::unique_ptr<TaskQueue> task_queue{
|
||||||
|
new ThreadPool{/*num_threads=*/16, /*max_threads=*/64}};
|
||||||
|
|
||||||
|
// Let the base threads go idle so the burst below finds them waiting.
|
||||||
|
std::this_thread::sleep_for(std::chrono::milliseconds(10));
|
||||||
|
|
||||||
|
std::mutex mtx;
|
||||||
|
std::condition_variable cv;
|
||||||
|
size_t running = 0;
|
||||||
|
bool release = false;
|
||||||
|
|
||||||
|
for (size_t i = 0; i < number_of_tasks; i++) {
|
||||||
|
EXPECT_TRUE(task_queue->enqueue([&] {
|
||||||
|
std::unique_lock<std::mutex> lock(mtx);
|
||||||
|
running++;
|
||||||
|
cv.notify_all();
|
||||||
|
cv.wait(lock, [&] { return release; });
|
||||||
|
}));
|
||||||
|
}
|
||||||
|
|
||||||
|
{
|
||||||
|
std::unique_lock<std::mutex> lock(mtx);
|
||||||
|
EXPECT_TRUE(cv.wait_for(lock, std::chrono::seconds(5),
|
||||||
|
[&] { return running == number_of_tasks; }));
|
||||||
|
release = true;
|
||||||
|
}
|
||||||
|
cv.notify_all();
|
||||||
|
|
||||||
|
task_queue->shutdown();
|
||||||
|
}
|
||||||
|
|
||||||
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}};
|
||||||
|
|||||||
Reference in new issue
Block a user