From: Joel Rosdahl Date: Wed, 29 Oct 2025 14:38:03 +0000 (+0100) Subject: enhance: Add util::ThreadPool::enqueue, returning a std::future X-Git-Tag: v4.13~86 X-Git-Url: http://git.ipfire.org/cgi-bin/gitweb.cgi?a=commitdiff_plain;h=6fc9cbe64aa7ed1c4ff3bf17236d4ba0f34dcb62;p=thirdparty%2Fccache.git enhance: Add util::ThreadPool::enqueue, returning a std::future --- diff --git a/src/ccache/core/mainoptions.cpp b/src/ccache/core/mainoptions.cpp index 95796fd0..b20f7d60 100644 --- a/src/ccache/core/mainoptions.cpp +++ b/src/ccache/core/mainoptions.cpp @@ -354,7 +354,7 @@ trim_dir(const std::string& dir, std::atomic incompressible_size = 0; for (auto& file : files) { - thread_pool.enqueue([&] { + thread_pool.enqueue_detach([&] { try { auto new_stat = recompressor.recompress( file, *recompress_level, core::FileRecompressor::KeepAtime::yes); diff --git a/src/ccache/storage/local/localstorage.cpp b/src/ccache/storage/local/localstorage.cpp index c25899f9..99b18c51 100644 --- a/src/ccache/storage/local/localstorage.cpp +++ b/src/ccache/storage/local/localstorage.cpp @@ -906,38 +906,40 @@ LocalStorage::recompress(const std::optional level, const auto& file = files[i]; if (file_type_from_path(file.path()) != FileType::unknown) { - thread_pool.enqueue([=, &recompressor, &incompressible_size] { - try { - DirEntry new_dir_entry = recompressor.recompress( - file, level, core::FileRecompressor::KeepAtime::no); - auto old_size = file.size(); - auto new_size = new_dir_entry.size(); - // LOG_RAW+fmt::format instead of LOG due to GCC 12.3 bug - // #109241 - LOG_RAW(fmt::format("Recompressed {} from {} to {} bytes", - file.path(), - old_size, - new_size)); - auto size_change_kibibyte = - kibibyte_size_diff(file, new_dir_entry); - if (size_change_kibibyte != 0) { - StatsFile(stats_file).update([=](auto& cs) { - cs.increment(Statistic::cache_size_kibibyte, - size_change_kibibyte); - cs.increment_offsetted( - Statistic::subdir_size_kibibyte_base, - l2_index, - size_change_kibibyte); - }); + thread_pool.enqueue_detach( + [=, &recompressor, &incompressible_size] { + try { + DirEntry new_dir_entry = recompressor.recompress( + file, level, core::FileRecompressor::KeepAtime::no); + auto old_size = file.size(); + auto new_size = new_dir_entry.size(); + // LOG_RAW+fmt::format instead of LOG due to GCC 12.3 bug + // #109241 + LOG_RAW(fmt::format("Recompressed {} from {} to {} bytes", + file.path(), + old_size, + new_size)); + auto size_change_kibibyte = + kibibyte_size_diff(file, new_dir_entry); + if (size_change_kibibyte != 0) { + StatsFile(stats_file).update([=](auto& cs) { + cs.increment(Statistic::cache_size_kibibyte, + size_change_kibibyte); + cs.increment_offsetted( + Statistic::subdir_size_kibibyte_base, + l2_index, + size_change_kibibyte); + }); + } + } catch (core::Error& e) { + // LOG_RAW+fmt::format instead of LOG due to GCC 12.3 bug + // #109241 + LOG_RAW(fmt::format("Error when recompressing {}: {}", + file.path(), + e.what())); + incompressible_size += file.size_on_disk(); } - } catch (core::Error& e) { - // LOG_RAW+fmt::format instead of LOG due to GCC 12.3 bug - // #109241 - LOG_RAW(fmt::format( - "Error when recompressing {}: {}", file.path(), e.what())); - incompressible_size += file.size_on_disk(); - } - }); + }); } else if (!util::TemporaryFile::is_tmp_file(file.path())) { incompressible_size += file.size_on_disk(); } diff --git a/src/ccache/util/threadpool.cpp b/src/ccache/util/threadpool.cpp index 78c56ed1..f64d05d0 100644 --- a/src/ccache/util/threadpool.cpp +++ b/src/ccache/util/threadpool.cpp @@ -61,7 +61,7 @@ ThreadPool::~ThreadPool() noexcept } void -ThreadPool::enqueue(std::function function) +ThreadPool::enqueue_detach(std::function function) { // Fast path for worker threads: avoid blocking on a full queue to prevent // deadlocks (all workers waiting inside enqueue() means no one can pop). diff --git a/src/ccache/util/threadpool.hpp b/src/ccache/util/threadpool.hpp index ecc7fe26..aacf1453 100644 --- a/src/ccache/util/threadpool.hpp +++ b/src/ccache/util/threadpool.hpp @@ -23,10 +23,14 @@ #include #include #include +#include #include +#include #include #include #include +#include +#include #include namespace util { @@ -39,7 +43,14 @@ public: size_t task_queue_max_size = std::numeric_limits::max()); ~ThreadPool() noexcept; - void enqueue(std::function function); + void enqueue_detach(std::function function); + + // Enqueue a task that returns a value. Returns a std::future that can be + // used to retrieve the result once the task completes. + template + auto enqueue(F&& f, Args&&... args) + -> std::future::type>; + void shut_down() noexcept; private: @@ -54,4 +65,21 @@ private: void worker_thread_main(); }; +template +auto +ThreadPool::enqueue(F&& f, Args&&... args) + -> std::future::type> +{ + using return_type = typename std::invoke_result::type; + + auto task = std::make_shared>( + std::bind(std::forward(f), std::forward(args)...)); + + std::future result = task->get_future(); + + enqueue_detach([task]() { (*task)(); }); + + return result; +} + } // namespace util diff --git a/unittest/test_util_threadpool.cpp b/unittest/test_util_threadpool.cpp index 08eea374..d329bd19 100644 --- a/unittest/test_util_threadpool.cpp +++ b/unittest/test_util_threadpool.cpp @@ -39,7 +39,7 @@ TEST_CASE("ThreadPool basic functionality") util::ThreadPool pool(2); std::atomic counter{0}; - pool.enqueue([&counter] { ++counter; }); + pool.enqueue_detach([&] { ++counter; }); pool.shut_down(); CHECK(counter == 1); @@ -52,7 +52,7 @@ TEST_CASE("ThreadPool basic functionality") const int num_tasks = 10; for (int i = 0; i < num_tasks; ++i) { - pool.enqueue([&counter] { ++counter; }); + pool.enqueue_detach([&] { ++counter; }); } pool.shut_down(); @@ -66,7 +66,7 @@ TEST_CASE("ThreadPool basic functionality") const int num_tasks = 100; for (int i = 0; i < num_tasks; ++i) { - pool.enqueue([&counter] { ++counter; }); + pool.enqueue_detach([&] { ++counter; }); } pool.shut_down(); @@ -78,7 +78,7 @@ TEST_CASE("ThreadPool basic functionality") util::ThreadPool pool(0); // Should create at least 1 thread std::atomic executed{false}; - pool.enqueue([&executed] { executed = true; }); + pool.enqueue_detach([&] { executed = true; }); pool.shut_down(); CHECK(executed); @@ -98,20 +98,20 @@ TEST_CASE("ThreadPool task queue limits") std::condition_variable cv; bool first_task_can_finish = false; - pool.enqueue([&mutex, &cv, &first_task_can_finish] { + pool.enqueue_detach([&] { std::unique_lock lock(mutex); cv.wait(lock, [&] { return first_task_can_finish; }); }); // Enqueue tasks to fill the queue. for (size_t i = 0; i < max_queue_size; ++i) { - pool.enqueue([&counter] { ++counter; }); + pool.enqueue_detach([&] { ++counter; }); } // Try to enqueue one more task in a separate thread - it should block. std::atomic extra_task_enqueued{false}; - std::thread enqueue_thread([&pool, &counter, &extra_task_enqueued] { - pool.enqueue([&counter] { ++counter; }); + std::thread enqueue_thread([&] { + pool.enqueue_detach([&] { ++counter; }); extra_task_enqueued = true; }); @@ -143,7 +143,7 @@ TEST_CASE("ThreadPool task queue limits") std::condition_variable cv; bool can_finish = false; - pool.enqueue([&mutex, &cv, &can_finish] { + pool.enqueue_detach([&] { std::unique_lock lock(mutex); cv.wait(lock, [&] { return can_finish; }); }); @@ -151,7 +151,7 @@ TEST_CASE("ThreadPool task queue limits") // Enqueue many tasks - should not block. const int num_tasks = 1000; for (int i = 0; i < num_tasks; ++i) { - pool.enqueue([&counter] { ++counter; }); + pool.enqueue_detach([&] { ++counter; }); } { @@ -165,6 +165,36 @@ TEST_CASE("ThreadPool task queue limits") } } +TEST_CASE("ThreadPool inline execution for worker threads") +{ + SUBCASE("worker thread can enqueue without deadlock when queue is full") + { + const size_t max_queue_size = 2; + util::ThreadPool pool(1, max_queue_size); + std::atomic counter{0}; + const int num_enqueues = 7; + + // Use a future to wait for the outer task to complete. + auto outer_task_future = pool.enqueue([&]() { + // Fill the queue from within a worker thread. When the queue is full, + // tasks will execute inline. + for (int i = 0; i < num_enqueues; ++i) { + pool.enqueue_detach([&] { ++counter; }); + } + }); + + // Wait for the outer task to complete. + outer_task_future.get(); + + pool.shut_down(); + + // All tasks should have executed (some inline, some from queue). The exact + // number executed should equal the number we enqueued. + int final_count = counter.load(); + CHECK(final_count == num_enqueues); + } +} + TEST_CASE("ThreadPool shutdown behavior") { SUBCASE("shutdown waits for all tasks to complete") @@ -180,7 +210,7 @@ TEST_CASE("ThreadPool shutdown behavior") std::atomic ready_count{0}; for (int i = 0; i < num_tasks; ++i) { - pool.enqueue([&counter, &mutex, &cv, &ready_count] { + pool.enqueue_detach([&] { // Signal that this task is running. ready_count++; cv.notify_all(); @@ -204,7 +234,7 @@ TEST_CASE("ThreadPool shutdown behavior") pool.shut_down(); - pool.enqueue([&counter] { ++counter; }); + pool.enqueue_detach([&] { ++counter; }); // No need to wait - the enqueue after shutdown should be a no-op. CHECK(counter == 0); @@ -215,7 +245,7 @@ TEST_CASE("ThreadPool shutdown behavior") util::ThreadPool pool(2); std::atomic counter{0}; - pool.enqueue([&counter] { ++counter; }); + pool.enqueue_detach([&] { ++counter; }); pool.shut_down(); pool.shut_down(); // Should be safe to call multiple times @@ -230,7 +260,7 @@ TEST_CASE("ThreadPool shutdown behavior") { util::ThreadPool pool(2); - pool.enqueue([&counter] { ++counter; }); + pool.enqueue_detach([&] { ++counter; }); // Destructor should call shut_down(). } @@ -245,10 +275,10 @@ TEST_CASE("ThreadPool exception handling") util::ThreadPool pool(2); std::atomic counter{0}; - pool.enqueue([] { throw std::runtime_error("Test exception"); }); + pool.enqueue_detach([] { throw std::runtime_error("Test exception"); }); - pool.enqueue([&counter] { ++counter; }); - pool.enqueue([&counter] { ++counter; }); + pool.enqueue_detach([&] { ++counter; }); + pool.enqueue_detach([&] { ++counter; }); pool.shut_down(); @@ -262,8 +292,8 @@ TEST_CASE("ThreadPool exception handling") std::atomic counter{0}; for (int i = 0; i < 5; ++i) { - pool.enqueue([] { throw std::runtime_error("Test exception"); }); - pool.enqueue([&counter] { ++counter; }); + pool.enqueue_detach([] { throw std::runtime_error("Test exception"); }); + pool.enqueue_detach([&] { ++counter; }); } pool.shut_down(); @@ -276,8 +306,8 @@ TEST_CASE("ThreadPool exception handling") util::ThreadPool pool(2); std::atomic counter{0}; - pool.enqueue([] { throw 42; }); - pool.enqueue([&counter] { ++counter; }); + pool.enqueue_detach([] { throw 42; }); + pool.enqueue_detach([&] { ++counter; }); pool.shut_down(); @@ -298,7 +328,7 @@ TEST_CASE("ThreadPool concurrent access") for (int i = 0; i < num_producer_threads; ++i) { producer_threads.emplace_back([&] { for (int j = 0; j < tasks_per_thread; ++j) { - pool.enqueue([&counter] { ++counter; }); + pool.enqueue_detach([&] { ++counter; }); } }); } @@ -326,7 +356,7 @@ TEST_CASE("ThreadPool task ordering") std::condition_variable start_cv; bool can_start = false; - pool.enqueue([&start_mutex, &start_cv, &can_start] { + pool.enqueue_detach([&] { std::unique_lock lock(start_mutex); start_cv.wait(lock, [&] { return can_start; }); }); @@ -334,7 +364,7 @@ TEST_CASE("ThreadPool task ordering") // Enqueue tasks in order. const int num_tasks = 10; for (int i = 0; i < num_tasks; ++i) { - pool.enqueue([&execution_order, &order_mutex, i] { + pool.enqueue_detach([&, i] { std::lock_guard lock(order_mutex); execution_order.push_back(i); }); @@ -354,4 +384,203 @@ TEST_CASE("ThreadPool task ordering") } } +TEST_CASE("ThreadPool enqueue") +{ + SUBCASE("simple return value") + { + util::ThreadPool pool(2); + + auto future = pool.enqueue([]() { return 42; }); + + CHECK(future.get() == 42); + pool.shut_down(); + } + + SUBCASE("function with arguments") + { + util::ThreadPool pool(2); + + auto future = pool.enqueue([](int a, int b) { return a + b; }, 10, 20); + + CHECK(future.get() == 30); + pool.shut_down(); + } + + SUBCASE("string return type") + { + util::ThreadPool pool(2); + + auto future = pool.enqueue([]() { return std::string("Hello, World!"); }); + + CHECK(future.get() == "Hello, World!"); + pool.shut_down(); + } + + SUBCASE("multiple futures") + { + util::ThreadPool pool(4); + + auto future1 = pool.enqueue([]() { return 1; }); + auto future2 = pool.enqueue([]() { return 2; }); + auto future3 = pool.enqueue([]() { return 3; }); + auto future4 = pool.enqueue([]() { return 4; }); + + CHECK(future1.get() == 1); + CHECK(future2.get() == 2); + CHECK(future3.get() == 3); + CHECK(future4.get() == 4); + + pool.shut_down(); + } + + SUBCASE("future with computation") + { + util::ThreadPool pool(2); + + auto future = pool.enqueue([]() { + int sum = 0; + for (int i = 1; i <= 100; ++i) { + sum += i; + } + return sum; + }); + + CHECK(future.get() == 5050); + pool.shut_down(); + } + + SUBCASE("future blocks until result is ready") + { + util::ThreadPool pool(1); + + // Use a condition variable to control when the task completes. + std::mutex mutex; + std::condition_variable cv; + bool can_finish = false; + + auto future = pool.enqueue([&]() { + std::unique_lock lock(mutex); + cv.wait(lock, [&] { return can_finish; }); + return 123; + }); + + // Verify that the future is not immediately ready. + CHECK(future.wait_for(std::chrono::milliseconds(0)) + == std::future_status::timeout); + + // Allow the task to complete. + { + std::lock_guard lock(mutex); + can_finish = true; + } + cv.notify_one(); + + // Now future.get() should succeed. + int result = future.get(); + CHECK(result == 123); + + pool.shut_down(); + } + + SUBCASE("exception in future task") + { + util::ThreadPool pool(2); + + auto future = + pool.enqueue([]() -> int { throw std::runtime_error("Task failed"); }); + + CHECK_THROWS_AS(future.get(), std::runtime_error); + + pool.shut_down(); + } + + SUBCASE("multiple futures with different types") + { + util::ThreadPool pool(3); + + auto int_future = pool.enqueue([]() { return 42; }); + auto str_future = pool.enqueue([]() { return std::string("test"); }); + auto double_future = pool.enqueue([]() { return 3.14; }); + + CHECK(int_future.get() == 42); + CHECK(str_future.get() == "test"); + CHECK(double_future.get() == doctest::Approx(3.14)); + + pool.shut_down(); + } + + SUBCASE("future with captured variables") + { + util::ThreadPool pool(2); + + int x = 10; + int y = 20; + + auto future = pool.enqueue([x, y]() { return x * y; }); + + CHECK(future.get() == 200); + pool.shut_down(); + } + + SUBCASE("void return type") + { + util::ThreadPool pool(2); + std::atomic executed{false}; + + auto future = pool.enqueue([&]() { executed = true; }); + + future.get(); // Should work even with void return + CHECK(executed); + + pool.shut_down(); + } + + SUBCASE("future remains valid after shutdown") + { + util::ThreadPool pool(1); + + auto future = pool.enqueue([]() { return 99; }); + + pool.shut_down(); + + // Should still be able to get the result after shutdown. + CHECK(future.get() == 99); + } + + SUBCASE("parallel computation with futures") + { + util::ThreadPool pool(4); + + std::vector> futures; + const int num_tasks = 10; + + for (int i = 0; i < num_tasks; ++i) { + futures.push_back(pool.enqueue([i]() { return i * i; })); + } + + for (int i = 0; i < num_tasks; ++i) { + CHECK(futures[i].get() == i * i); + } + + pool.shut_down(); + } + + SUBCASE("future with reference capture") + { + util::ThreadPool pool(2); + std::vector data = {1, 2, 3, 4, 5}; + + auto future = pool.enqueue([&]() { + int sum = 0; + for (int val : data) { + sum += val; + } + return sum; + }); + + CHECK(future.get() == 15); + pool.shut_down(); + } +} + TEST_SUITE_END();