/* * MIT License * * Copyright (c) 2022 Colion Braley * * Permission is hereby granted, free of charge, to any person obtaining a copy * of this software and associated documentation files (the "Software"), to deal * in the Software without restriction, including without limitation the rights * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell * copies of the Software, and to permit persons to whom the Software is * furnished to do so, subject to the following conditions: * * The above copyright notice and this permission notice shall be included in all * copies or substantial portions of the Software. * * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE * SOFTWARE. */ // From https://raw.githubusercontent.com/cbraley/threadpool/master/src/thread_pool.h #pragma once // A simple thread pool class. // Usage examples: // // { // ThreadPool pool(16); // 16 worker threads. // for (int i = 0; i < 100; ++i) { // pool.Schedule([i]() { // DoSlowExpensiveOperation(i); // }); // } // // // `pool` goes out of scope here - the code will block in the ~ThreadPool // // destructor until all work is complete. // } // // // TODO(cbraley): Add examples with std::future. #include #include #include #include #include #include #include // We want to use std::invoke if C++17 is available, and fallback to "hand // crafted" code if std::invoke isn't available. #if __cplusplus >= 201703L || defined(_MSC_VER) #define INVOKE_MACRO(CALLABLE, ARGS_TYPE, ARGS) std::invoke(CALLABLE, std::forward(ARGS)...) #elif __cplusplus >= 201103L // Update this with http://www.open-std.org/jtc1/sc22/wg21/docs/papers/2014/n4169.html. #define INVOKE_MACRO(CALLABLE, ARGS_TYPE, ARGS) CALLABLE(std::forward(ARGS)...) #else #error ("C++ version is too old! C++98 is not supported.") #endif namespace cb { class ThreadPool { public: // Create a thread pool with `num_workers` dedicated worker threads. explicit ThreadPool(int num_workers); // Default construction is disallowed. ThreadPool() = delete; // Get the number of logical cores on the CPU. This is implemented using // std::thread::hardware_concurrency(). // https://en.cppreference.com/w/cpp/thread/thread/hardware_concurrency static unsigned int GetNumLogicalCores(); // The `ThreadPool` destructor blocks until all outstanding work is complete. ~ThreadPool(); // No copying, assigning, or std::move-ing. ThreadPool& operator=(const ThreadPool&) = delete; ThreadPool(const ThreadPool&) = delete; ThreadPool(ThreadPool&&) = delete; ThreadPool& operator=(ThreadPool&&) = delete; // Add the function `func` to the thread pool. `func` will be executed at some // point in the future on an arbitrary thread. void Schedule(std::function func); // Add `func` to the thread pool, and return a std::future that can be used to // access the function's return value. // // *** Usage example *** // Don't be alarmed by this function's tricky looking signature - this is // very easy to use. Here's an example: // // int ComputeSum(std::vector& values) { // int sum = 0; // for (const int& v : values) { // sum += v; // } // return sum; // } // // ThreadPool pool = ...; // std::vector numbers = ...; // // std::future sum_future = ScheduleAndGetFuture( // []() { // return ComputeSum(numbers); // }); // // // Do other work... // // std::cout << "The sum is " << sum_future.get() << std::endl; // // *** Details *** // Given a callable `func` that returns a value of type `RetT`, this // function returns a std::future that can be used to access // `func`'s results. template auto ScheduleAndGetFuture(FuncT&& func, ArgsT&&... args) -> std::future; // Wait for all outstanding work to be completed. void Wait(); // Return the number of outstanding functions to be executed. int OutstandingWorkSize() const; // Return the number of threads in the pool. int NumWorkers() const; void SetWorkDoneCallback(std::function func); private: void ThreadLoop(); // Number of worker threads - fixed at construction time. int num_workers_; // The destructor sets `exit_` to true and then notifies all workers. `exit_` // causes each thread to break out of their work loop. bool exit_; mutable std::mutex mu_; // Work queue. Guarded by `mu_`. struct WorkItem { std::function func; }; std::queue work_; // Condition variable used to notify worker threads that new work is // available. std::condition_variable condvar_; // Worker threads. std::vector workers_; // Condition variable used to notify that all work is complete - the work // queue has "run dry". std::condition_variable work_done_condvar_; // Whenever a work item is complete, we call this callback. If this is empty, // nothing is done. std::function work_done_callback_; }; namespace impl { // This helper class simply returns a std::function that executes: // ReturnT x = func(); // promise->set_value(x); // However, this is tricky in the case where T == void. The code above won't // compile if ReturnT == void, and neither will // promise->set_value(func()); // To workaround this, we use a template specialization for the case where // ReturnT is void. If the "regular void" proposal is accepted, this could be // simpler: // http://www.open-std.org/jtc1/sc22/wg21/docs/papers/2016/p0146r1.html. // The non-specialized `FuncWrapper` implementation handles callables that // return a non-void value. template struct FuncWrapper { template std::function GetWrapped( FuncT&& func, std::shared_ptr> promise, ArgsT&&... args) { // TODO(cbraley): Capturing by value is inefficient. It would be more // efficient to move-capture everything, but we can't do this until C++14 // generalized lambda capture is available. Can we use std::bind instead to // make this more efficient and still use C++11? return [promise, func, args...]() mutable { promise->set_value(INVOKE_MACRO(func, ArgsT, args)); }; } }; template void InvokeVoidRet(FuncT&& func, std::shared_ptr> promise, ArgsT&&... args) { INVOKE_MACRO(func, ArgsT, args); promise->set_value(); } // This `FuncWrapper` specialization handles callables that return void. template <> struct FuncWrapper { template std::function GetWrapped(FuncT&& func, std::shared_ptr> promise, ArgsT&&... args) { return [promise, func, args...]() mutable { INVOKE_MACRO(func, ArgsT, args); promise->set_value(); }; } }; } // namespace impl template auto ThreadPool::ScheduleAndGetFuture(FuncT&& func, ArgsT&&... args) -> std::future { using ReturnT = decltype(INVOKE_MACRO(func, ArgsT, args)); // We are only allocating this std::promise in a shared_ptr because // std::promise is non-copyable. std::shared_ptr> promise = std::make_shared>(); std::future ret_future = promise->get_future(); impl::FuncWrapper func_wrapper; std::function wrapped_func = func_wrapper.GetWrapped( std::move(func), std::move(promise), std::forward(args)...); // Acquire the lock, and then push the WorkItem onto the queue. { std::lock_guard scoped_lock(mu_); WorkItem work; work.func = std::move(wrapped_func); work_.emplace(std::move(work)); } condvar_.notify_one(); // Tell one worker we are ready. return ret_future; } } // namespace cb #undef INVOKE_MACRO