Separate thread pools for http and collection processing.

Fixed a deadlock issue that happened on high concurrency when all threads were exhausted quickly.

include/threadpool.h

@@ -1,110 +1,88 @@
-/*
-Copyright (c) 2012 Jakob Progsch, Václav Zeman
-This software is provided 'as-is', without any express or implied
-warranty. In no event will the authors be held liable for any damages
-arising from the use of this software.
-Permission is granted to anyone to use this software for any purpose,
-including commercial applications, and to alter it and redistribute it
-freely, subject to the following restrictions:
-   1. The origin of this software must not be misrepresented; you must not
-   claim that you wrote the original software. If you use this software
-   in a product, an acknowledgment in the product documentation would be
-   appreciated but is not required.
-   2. Altered source versions must be plainly marked as such, and must not be
-   misrepresented as being the original software.
-   3. This notice may not be removed or altered from any source
-   distribution.
-*/
+// https://github.com/jhasse/ThreadPool
 
 #pragma once
 
-// containers
-#include <vector>
-#include <queue>
-// threading
-#include <thread>
-#include <mutex>
-#include <condition_variable>
-#include <atomic>
-#include <future>
-// utility wrappers
-#include <memory>
 #include <functional>
-// exceptions
-#include <stdexcept>
+#include <future>
+#include <queue>
 
-// std::thread pool for resources recycling
 class ThreadPool {
 public:
-    // the constructor just launches some amount of workers
-    ThreadPool(size_t threads_n = std::thread::hardware_concurrency()) : stop(false)
-    {
-        if(!threads_n)
-            throw std::invalid_argument("more than zero threads expected");
-
-        this->workers.reserve(threads_n);
-        for(; threads_n; --threads_n)
-            this->workers.emplace_back(
-                    [this]
-                    {
-                        while(true)
-                        {
-                            std::function<void()> task;
-
-                            {
-                                std::unique_lock<std::mutex> lock(this->queue_mutex);
-                                this->condition.wait(lock,
-                                                     [this]{ return this->stop || !this->tasks.empty(); });
-                                if(this->stop && this->tasks.empty())
-                                    return;
-                                task = std::move(this->tasks.front());
-                                this->tasks.pop();
-                            }
-
-                            task();
-                        }
-                    }
-            );
-    }
-    // deleted copy&move ctors&assignments
-    ThreadPool(const ThreadPool&) = delete;
-    ThreadPool& operator=(const ThreadPool&) = delete;
-    ThreadPool(ThreadPool&&) = delete;
-    ThreadPool& operator=(ThreadPool&&) = delete;
-    // add new work item to the pool
+    explicit ThreadPool(size_t);
     template<class F, class... Args>
-    std::future<typename std::result_of<F(Args...)>::type> enqueue(F&& f, Args&&... args)
-    {
-        using packaged_task_t = std::packaged_task<typename std::result_of<F(Args...)>::type ()>;
-
-        std::shared_ptr<packaged_task_t> task(new packaged_task_t(
-                std::bind(std::forward<F>(f), std::forward<Args>(args)...)
-        ));
-        auto res = task->get_future();
-        {
-            std::unique_lock<std::mutex> lock(this->queue_mutex);
-            this->tasks.emplace([task](){ (*task)(); });
-        }
-        this->condition.notify_one();
-        return res;
-    }
-    // the destructor joins all threads
-    virtual ~ThreadPool()
-    {
-        this->stop = true;
-        this->condition.notify_all();
-        for(std::thread& worker : this->workers)
-            worker.join();
-    }
+    decltype(auto) enqueue(F&& f, Args&&... args);
+    ~ThreadPool();
 private:
     // need to keep track of threads so we can join them
     std::vector< std::thread > workers;
     // the task queue
-    std::queue< std::function<void()> > tasks;
+    std::queue< std::packaged_task<void()> > tasks;
 
     // synchronization
     std::mutex queue_mutex;
     std::condition_variable condition;
-    // workers finalization flag
-    std::atomic_bool stop;
-};
+    bool stop;
+};
+
+// the constructor just launches some amount of workers
+inline ThreadPool::ThreadPool(size_t threads)
+        :   stop(false)
+{
+    for(size_t i = 0;i<threads;++i)
+        workers.emplace_back(
+                [this]
+                {
+                    for(;;)
+                    {
+                        std::packaged_task<void()> task;
+                        {
+                            std::unique_lock<std::mutex> lock(this->queue_mutex);
+                            this->condition.wait(lock,
+                                                 [this]{ return this->stop || !this->tasks.empty(); });
+                            if(this->stop && this->tasks.empty()) {
+                                return;
+                            }
+                            task = std::move(this->tasks.front());
+                            this->tasks.pop();
+                        }
+
+                        task();
+                    }
+                }
+        );
+}
+
+// add new work item to the pool
+template<class F, class... Args>
+decltype(auto) ThreadPool::enqueue(F&& f, Args&&... args)
+{
+    using return_type = std::invoke_result_t<F, Args...>;
+
+    std::packaged_task<return_type()> task(
+            std::bind(std::forward<F>(f), std::forward<Args>(args)...)
+    );
+
+    std::future<return_type> res = task.get_future();
+    {
+        std::unique_lock<std::mutex> lock(queue_mutex);
+
+        // don't allow enqueueing after stopping the pool
+        if(stop)
+            throw std::runtime_error("enqueue on stopped ThreadPool");
+
+        tasks.emplace(std::move(task));
+    }
+    condition.notify_one();
+    return res;
+}
+
+inline ThreadPool::~ThreadPool() {
+    {
+        std::unique_lock<std::mutex> lock(queue_mutex);
+        stop = true;
+    }
+    condition.notify_all();
+    for (std::thread& worker : workers) {
+        worker.join();
+    }
+}

src/main/main.cpp

@@ -33,7 +33,7 @@ int main(int argc, char* argv[]) {
             field("org", field_types::STRING, true)
     };
 
-    Collection *collection = collectionManager.get_collection("github_top1k");
+    Collection *collection = collectionManager.get_collection("github_top1k").get();
     if(collection == nullptr) {
         collection = collectionManager.create_collection("github_top1k", 4, fields_to_index, "stars").get();
     }

src/typesense_server_utils.cpp

@@ -365,11 +365,12 @@ int run_server(const Config & config, const std::string & version, void (*master
     const size_t num_threads = std::max<size_t>(proc_count * 8, 16);
 
     LOG(INFO) << "Thread pool size: " << num_threads;
-    ThreadPool thread_pool(num_threads);
+    ThreadPool app_thread_pool(num_threads);
+    ThreadPool server_thread_pool(num_threads);
     Store store(db_dir);
 
     CollectionManager & collectionManager = CollectionManager::get_instance();
-    collectionManager.init(&store, &thread_pool, config.get_max_memory_ratio(), config.get_api_key());
+    collectionManager.init(&store, &app_thread_pool, config.get_max_memory_ratio(), config.get_api_key());
 
     curl_global_init(CURL_GLOBAL_SSL);
     HttpClient & httpClient = HttpClient::get_instance();
@@ -382,7 +383,7 @@ int run_server(const Config & config, const std::string & version, void (*master
         config.get_ssl_cert(),
         config.get_ssl_cert_key(),
         config.get_enable_cors(),
-        &thread_pool
+        &server_thread_pool
     );
 
     server->set_auth_handler(handle_authentication);
@@ -396,7 +397,7 @@ int run_server(const Config & config, const std::string & version, void (*master
 
     // first we start the peering service
 
-    ReplicationState replication_state(&store, &thread_pool, server->get_message_dispatcher(),
+    ReplicationState replication_state(&store, &app_thread_pool, server->get_message_dispatcher(),
                                        ssl_enabled, config.get_catch_up_min_sequence_diff(),
                                        config.get_catch_up_threshold_percentage(),
                                        create_init_db_snapshot, quit_raft_service);