C++26 concurrency task-generation

Introduction

This article presents a complete concurrent programming example in C++, using it as a case study to explore dynamic task generation, work-completion detection, and the safe aggregation of partial results.

Given a directory path, our goal is to produce a statistical summary of the contents of its directory tree. For each file extension found (.txt, .zip, and so on), we will determine both the number of files and their cumulative size. The analysis will also report the total number of subdirectories discovered during the traversal.

In this implementation, we will parallelize the directory tree traversal itself. Each worker thread will process directories retrieved from a shared concurrent queue, discover any subdirectories they contain, and dynamically push them into the queue so that they can later be processed by any available worker. As a result, multiple threads will be able to explore different branches of the directory tree simultaneously while collecting partial statistics that will finally be merged into a single global result.

This approach will allow us to generate summaries such as the following for a given root directory:

.docx: 16 files (147.938.289 bytes) .pdf: 28 files (72.191.677 bytes) .png: 1 file (90.575 bytes) .xlsx: 1 file (88.162 bytes) _____________________________________ contains: 46 files, 5 folders size: 210.1 MiB (220.308.703 bytes)
We will use the C++26 standard, particularly contracts and std::optional<T&>, together with the module system introduced in C++20. The latter allows us to organize the code efficiently into well-defined components. If necessary, adapting the solution to the traditional structure based on header (.hpp) and implementation (.cpp) files would be straightforward.

The project can already be compiled with GCC 16.2, which provides support for C++ contracts and modules. However, some module features are still unavailable, most notably private module fragments (module :private;).

Module 1: dynamic_task_queue

We begin by implementing a blocking concurrent work queue, dynamic_task_queue<T>, specifically designed for scenarios in which tasks can dynamically generate additional tasks during their execution. This data structure simplifies the parallelization of graph traversals, including the directory hierarchies considered in this article.

The class stores pending tasks of type T in a private std::queue<T> named tasks_. It also maintains a counter, active_, that tracks the number of tasks currently being processed. This counter is incremented whenever a task is acquired and decremented when the task completes. As a result, the queue can detect global completion automatically, which occurs when there are neither pending tasks nor tasks in progress. The global termination condition is therefore: tasks_.empty() and active_ == 0.

Both the queue and the counter are protected by a std::mutex1, while a std::condition_variable2 is used to block worker threads whenever no work is available and to wake them up when new tasks are added or when global completion is detected.

Notice that an empty tasks_ queue does not necessarily imply that the computation has finished: a worker thread may still be processing a task and could generate additional tasks at a later stage. The role of the active_ counter is precisely to distinguish between these two situations.

The queue interface will be based on an acquire/complete protocol, where every task acquired by a worker through the acquire() member function must eventually be reported as completed through a matching call to the complete() function. In addition, complete() may register any new tasks discovered during the processing of the acquired task. Specifically:

  • acquire(): If work is available, it retrieves a task from the front of the queue, removes it from the pending-work list, and internally increments the active_ task counter. If there are no pending tasks but other workers are still processing work, the call blocks until new work becomes available or the computation finishes. When there are neither pending nor active tasks left, the function returns an empty result (std::nullopt), signalling that the worker may terminate.

  • complete(): Performs two actions while holding the internal mutex: (i) it records the completion of a task previously acquired via acquire() by decrementing the active_ counter, and (ii) it pushes newly discovered tasks into the queue. If no pending or active tasks remain after the operation, it signals global completion so all workers can finish. If there is pending work (tasks_.empty() == false), it wakes up a blocked worker.

To prevent the programmer from having to call complete() manually and to guarantee proper task completion even in the presence of exceptions, acquire() does not directly return a task object of type T. Instead, it returns a std::optional containing an instance of an auxiliary class named acquired_task. This class acts as an RAII handle representing a task currently in progress. In addition to the acquired value, it stores a vector of new tasks discovered during processing. When the object goes out of scope, its destructor automatically invokes complete(), transferring the accumulated new tasks to the queue and signaling the completion of the original task. Thus, the acquire/complete protocol is enforced by design through RAII:

export module concurrency_tools.dynamic_task_queue; import std; export namespace concurrency_tools { template<typename T> concept nothrow_move_constructible = std::is_nothrow_move_constructible_v<T>; // A blocking concurrent work queue for algorithms where // tasks can dynamically generate additional tasks: template<nothrow_move_constructible T> class dynamic_task_queue { public: class acquired_task; template<typename S> requires std::constructible_from<T, S&&> explicit dynamic_task_queue(S&& initial) { tasks_.emplace(std::forward<S>(initial)); } dynamic_task_queue(dynamic_task_queue const&) = delete; auto operator=(dynamic_task_queue const&) -> dynamic_task_queue& = delete; [[nodiscard]] auto acquire() -> std::optional<acquired_task> { auto lock = std::unique_lock{mtx_}; cv_.wait(lock, [this]{ return active_ == 0 or not tasks_.empty(); }); if (tasks_.empty()) { // ⇔ finished() == true return std::nullopt; } auto res = std::optional<acquired_task>{ std::in_place, *this, std::move(tasks_.front()) }; tasks_.pop(); ++active_; return res; } [[nodiscard]] auto empty() const -> bool { auto lock = std::lock_guard{mtx_}; return tasks_.empty(); } [[nodiscard]] auto active() const -> std::size_t { auto lock = std::lock_guard{mtx_}; return active_; } [[nodiscard]] auto finished() const -> bool { auto lock = std::lock_guard{mtx_}; return tasks_.empty() and active_ == 0; } private: std::queue<T> tasks_; std::size_t active_ = 0; mutable std::mutex mtx_; std::condition_variable cv_; // note on noexcept: a failure while pushing new tasks // is considered unrecoverable and triggers std::terminate() void complete(std::vector<T>&& new_tasks) noexcept { auto globally_finished = false; auto has_pending_work = false; { auto lock = std::lock_guard{mtx_}; contract_assert(active_ != 0); --active_; for (auto& t : new_tasks) { tasks_.push(std::move(t)); } globally_finished = tasks_.empty() and active_ == 0; has_pending_work = not tasks_.empty(); } if (globally_finished) { cv_.notify_all(); } else if (has_pending_work) { cv_.notify_one(); } } }; // ----------------------------------------- template<nothrow_move_constructible T> class dynamic_task_queue<T>::acquired_task { public: acquired_task( dynamic_task_queue<T>& queue, T&& value ) noexcept : queue_opt_{queue}, value_{std::move(value)}, new_tasks_{} { } acquired_task(acquired_task const&) = delete; auto operator=(acquired_task const&) -> acquired_task& = delete; acquired_task(acquired_task&& other) noexcept : queue_opt_{other.queue_opt_}, value_{std::move(other.value_)}, new_tasks_{std::move(other.new_tasks_)} { other.queue_opt_ = std::nullopt; } auto operator=(acquired_task&&) -> acquired_task& = delete; [[nodiscard]] auto get() const noexcept -> T const& { return value_; } [[nodiscard]] auto get() noexcept -> T& { return value_; } template<typename S> requires std::constructible_from<T, S&&> void add_task(S&& task) { new_tasks_.emplace_back(std::forward<S>(task)); } ~acquired_task() { if (queue_opt_) { queue_opt_->complete(std::move(new_tasks_)); } } private: std::optional<dynamic_task_queue<T>&> queue_opt_; T value_; std::vector<T> new_tasks_; }; } // namespace concurrency_tools

Module 2: statistics.directory

This module implements the parallel analysis of a directory tree. Given a root directory (root), it produces an object of type directory_statistics containing:

  • A std::map associative container that maps each file extension encountered (.txt, .zip, and so on) to an extension_statistics object, which stores both the number of files with that extension present in the directory tree and their cumulative size. The use of std::map ensures that file extensions appear alphabetically in generated reports by default, improving readability.
  • The total number of subdirectories discovered.
  • The total number of errors encountered during the analysis, whether due to (i) failure to open a directory, (ii) inability to retrieve information about a specific entry, or (iii) inability to continue iterating through a directory. These errors are accumulated in a single counter for simplicity of implementation, but they could easily be broken down into separate counters.

The run_directory_statistics() function is responsible for coordinating the parallel execution. It initializes a dynamic_task_queue<filesystem::path> with the root directory, where each task represents a directory pending analysis within the directory tree. The function then launches a configurable number of worker threads, num_workers, each of which produces a directory_statistics object containing the partial statistics gathered during its execution. Specifically, num_workers - 1 workers are launched asynchronously using std::async3, while the main thread acts as an additional worker processing tasks from the shared queue.

Each worker executes the process_directories() function, which repeatedly acquires directories from the queue through acquire(), inspects their contents, and updates the worker’s local directory_statistics object with the statistics of the files found, grouped by extension and cumulative size. Subdirectories discovered during the traversal are not processed immediately; instead, they are registered as new tasks within the acquired_task object associated with the current directory. Upon destruction, the object automatically transfers the accumulated tasks to the queue and signals completion of the original task. As we previously mentioned, this behavior is implemented using RAII.

Each pending directory dir is represented as a task stored in the dynamic_task_queue<filesystem::path>, with std::filesystem::is_directory(dir) == true. As workers discover new subdirectories, they are dynamically added to the queue as additional tasks, whereas regular files are processed directly and never become tasks themselves.

The implementation of process_directories() is based on std::filesystem::directory_iterator4, which iterates over the entries contained within a directory but does not visit its subdirectories. The iteration order is unspecified by the standard, except that each directory entry is visited exactly once.

As noted earlier, under this design each worker maintains its own directory_statistics object, accumulating statistics only for the directories it processes.

Once the traversal has completed, the partial results are merged by merge_statistics(), which aggregates the file counts and cumulative sizes for each extension, together with the total number of subdirectories visited.

export module statistics.directory; import std; import concurrency_tools.dynamic_task_queue; namespace statistics { export struct extension_statistics {    std::uintmax_t num_files = 0;    std::uintmax_t total_size = 0; }; // total result or partial result produced by a worker: export struct directory_statistics {    std::map<std::string, extension_statistics> files;    std::uintmax_t num_directories = 0;    std::uintmax_t num_errors = 0; }; using directory_queue = concurrency_tools::dynamic_task_queue<std::filesystem::path>; [[nodiscard]] auto process_directories(directory_queue& directories) -> directory_statistics {    auto res = directory_statistics{};    while (auto directory = directories.acquire()) {       // we only process the 'dir' directory; any subdirectories       // discovered are placed in the queue and will be processed // later by some worker:       auto const& dir = directory->get();       auto ec = std::error_code{};       auto it = std::filesystem::directory_iterator{dir, ec};       if (ec) {          // we cannot access the directory; the current task          // will complete automatically after 'continue':          ++res.num_errors;          continue;       }       auto end = std::filesystem::directory_iterator{};       for (; it != end; it.increment(ec)) {          // at this point, ec is clear in the first iteration; thereafter,          // it contains the result of the previous increment():          if (ec) { // error iterating through the directory ++res.num_errors;          break;          }          auto const entry = *it;          auto const status = entry.symlink_status(ec);        if (ec) { ++res.num_errors;           continue;          }        if (std::filesystem::is_symlink(status)) {             continue;          }          else if (std::filesystem::is_directory(status)) {             ++res.num_directories;           directory->add_task(entry.path());         }          else if (std::filesystem::is_regular_file(status)) {             auto const size = entry.file_size(ec);           if (ec) { ++res.num_errors;             continue;           } auto const extension = entry.path().extension().string();           auto& [num_files, total_size] = res.files[extension];           ++num_files;             total_size += size;         } }    } // destruction of the acquired_task 'directory' and automatic call // to complete() before the next loop iteration, adding // the discovered subdirectories to the task queue    return res; } void merge_statistics(    directory_statistics& destination,    directory_statistics const& source ){    destination.num_directories += source.num_directories; destination.num_errors += source.num_errors;    for (auto const& [extension, info] : source.files) {       auto& destination_info = destination.files[extension];       destination_info.num_files += info.num_files;       destination_info.total_size += info.total_size;    } } export [[nodiscard]] auto run_directory_statistics(    std::filesystem::path const& root,    std::size_t num_workers )    -> directory_statistics    pre(num_workers >= 1)    // precondition: std::filesystem::is_directory(root) must be true {    auto directories = directory_queue{root};    auto futures = std::views::indices(num_workers - 1)       | std::views::transform([&]([[maybe_unused]] auto worker_id)            -> std::future<directory_statistics> {            return std::async(               std::launch::async,               process_directories,               std::ref(directories)            );         })       | std::ranges::to<std::vector>();    auto total = process_directories(directories);    for (auto& f : futures) {       merge_statistics(total, f.get());    }    return total; } } // namespace statistics

Auxiliary modules

Before turning to the main program, we will define three simple auxiliary modules. Two of them provide thin wrappers around selected C library facilities, while the third offers formatting utilities for program output. We begin by examining the compatibility modules:

  • c_tools.exit_codes: Exposes the standard termination values EXIT_SUCCESS and EXIT_FAILURE as constexpr variables within the c_tools namespace. Its purpose is to facilitate the use of these values from modular code without relying directly on the macros defined in <cstdlib>.

  • c_tools.standard_streams: Provides noexcept functions that return std::FILE* pointers to the standard stdin, stdout, and stderr streams defined in <cstdio>.

module; #include <cstdlib> export module c_tools.exit_codes; export namespace c_tools { constexpr int exit_success = EXIT_SUCCESS; constexpr int exit_failure = EXIT_FAILURE; } // namespace c_tools
module; #include <cstdio> export module c_tools.standard_streams; export namespace c_tools { [[nodiscard]] auto stdin_stream() noexcept -> std::FILE* { return stdin; } [[nodiscard]] auto stdout_stream() noexcept -> std::FILE* { return stdout; } [[nodiscard]] auto stderr_stream() noexcept -> std::FILE* { return stderr; } } // namespace c_tools

The remaining auxiliary module is:

  • format_tools: Extends std::format through custom std::formatter specializations, allowing binary sizes to be displayed using more readable units (KiB, MiB, and GiB) and integral values to be formatted with thousands separators:
export module format_tools; import std; export namespace format_tools { struct binary_size { std::uint64_t value; }; struct thousands_separated { std::uint64_t value; }; } // namespace format_tools template<> struct std::formatter<format_tools::binary_size> : std::formatter<std::string_view> { template<typename format_context> auto format( format_tools::binary_size const& bsz, format_context& ctx ) const { constexpr auto KiB = 1024.0; constexpr auto MiB = 1024.0*KiB; constexpr auto GiB = 1024.0*MiB; auto format_size = [](std::uint64_t value) -> std::pair<double, std::string_view> { if (value >= GiB) { return {value/GiB, "GiB"}; } if (value >= MiB) { return {value/MiB, "MiB"}; } if (value >= KiB) { return {value/KiB, "KiB"}; } return {static_cast<double>(value), "bytes"}; }; auto const [size, unit] = format_size(bsz.value); auto const str = std::format("{:.1f} {}", size, unit); return std::formatter<std::string_view>::format(str, ctx); } }; template<> struct std::formatter<format_tools::thousands_separated> : std::formatter<std::string_view> { template<typename format_context> auto format( format_tools::thousands_separated const& n, format_context& ctx ) const { auto str = std::to_string(n.value); for (auto i = str.size(); i > 3uz; i -= 3) { str.insert(i - 3, 1, '.'); } return std::formatter<std::string_view>::format(str, ctx); } };

Benchmarks

The following main() function benchmarks the various modules developed throughout this article. It performs the statistical analysis of a directory tree specified as a command-line argument, using an increasing number of worker threads and measuring the execution time in each case.

Based on these measurements, several performance metrics are computed and displayed in the terminal, including the speedup relative to the sequential execution (i.e., a single worker thread) and the percentage improvement achieved when increasing the number of workers:

import std; import c_tools.exit_codes; import c_tools.standard_streams; import format_tools; import statistics.directory; namespace stdc = std::chrono; namespace stdf = std::filesystem; namespace stdv = std::views; struct benchmark_result { statistics::directory_statistics statistics; stdc::milliseconds duration; }; [[nodiscard]] auto run_benchmark( stdf::path const& root, std::size_t num_workers ) -> benchmark_result { using clock = stdc::steady_clock; auto const start = clock::now(); auto const total = statistics::run_directory_statistics(root, num_workers); using ms = stdc::milliseconds; return { .statistics = total, .duration = stdc::duration_cast<ms>(clock::now() - start) }; } auto main(int argc, char* argv[]) -> int { if (argc != 2) { std::println( c_tools::stderr_stream(), "usage: {} <directory>", argv[0] ); return c_tools::exit_failure; } auto const root = stdf::path{argv[1]}; auto ec = std::error_code{}; if (not stdf::is_directory(root, ec)) { std::println( c_tools::stderr_stream(), "error: '{}': {}", root.string(), ec? ec.message() : "not a directory" ); return c_tools::exit_failure; } auto const concurrency = std::thread::hardware_concurrency(); auto const max_workers = std::max(1u, concurrency); std::println( "{:>7} {:>10} {:>10} {:>15} {:>14}", "Workers", "Time (ms)", "Speedup", "vs. 1 worker", "vs. previous" ); auto baseline = stdc::milliseconds{}; auto previous = 0.0; for (auto const num_workers : stdv::iota(1u, max_workers + 1)) { auto const [stats, duration] = run_benchmark(root, num_workers); if (num_workers == 1) { baseline = duration; } auto const current = static_cast<double>(duration.count()); auto const speedup = static_cast<double>(baseline.count())/current; std::println( "{:>7} {:>10} {:>9.1f}× {:>14.1f}% {:>14}", num_workers, current, speedup, 100.0*(1.0 - 1.0/speedup), previous? std::format("{:.1f}%", 100*(1.0 - current/previous)) : "-" ); previous = current; if (num_workers == max_workers) { auto total_files = std::uintmax_t{}; auto total_size = std::uintmax_t{}; for (auto const& [extension, info] : stats.files) { total_files += info.num_files; total_size += info.total_size; } std::println( "{:_^60}\ncontains: {} files, {} folders\nsize: {} ({} bytes)\nerrors: {}", "", format_tools::thousands_separated{total_files}, format_tools::thousands_separated{stats.num_directories}, format_tools::binary_size{total_size}, format_tools::thousands_separated{total_size}, stats.num_errors ); } } return c_tools::exit_success; }

The number of worker threads num_workers defaults to hardware_concurrency()5, which provides an estimate of the concurrency level available on the system, typically matching the number of hardware threads. However, this value should be interpreted as an indicative upper bound and not necessarily as the optimal number of workers. In practice, performance may saturate at a lower thread count due to factors such as contention on the shared queue, the underlying storage device, or limitations imposed by the file system itself. In particular, different storage technologies may scale quite differently: modern NVMe SSDs are designed to handle significantly higher degrees of concurrent access than mechanical drives.

As an example, on a system equipped with an 11th Gen Intel Core i5-1135G7 processor running at 2.40 GHz, 8 hardware threads, and an SK hynix HFM256GD3HX015N SSD, the following results were obtained when analyzing a test directory tree:

Workers Time (ms) Speedup vs. 1 worker vs. previous 1 16331 1.0× 0.0% - 2 10194 1.6× 37.6% 37.6% 3 7777 2.1× 52.4% 23.7% 4 6242 2.6× 61.8% 19.7% 5 4994 3.3× 69.4% 20.0% 6 4545 3.6× 72.2% 9.0% 7 4281 3.8× 73.8% 5.8% 8 4125 4.0× 74.7% 3.6% ____________________________________________________________ contains: 136.613 files, 7.695 folders size: 7.2 GiB (7.679.411.327 bytes) errors: 0

Here, the “vs. 1 worker” column expresses the percentage reduction in execution time relative to the sequential version.

The associative container within the final directory_statistics result can be used, among other things, to generate breakdowns like the one shown in the introduction or to retrieve statistics for a specific file extension. For instance, to determine the total number of .txt files and their cumulative size in a root directory, we would write:

auto const [files, num_directories, num_errors] = statistics::run_directory_statistics(root, 8/*workers*/); auto const extension = ".txt"; if ( auto const it = files.find(extension); it != files.end() ){ auto const& [num_files, total_size] = it->second; std::println( "{} files, {} ({} bytes)", format_tools::thousands_separated{num_files}, format_tools::binary_size{total_size}, format_tools::thousands_separated{total_size} ); } else { std::println("no {} files found", extension); }

Note: This post is an English translation of my earlier post originally published in Spanish on Blogger as Programación Concurrente IX.

Bibliography

  1. cppreference.com – std::mutex ↩

  2. cppreference.com – std::condition_variable ↩

  3. cppreference.com – std::async ↩

  4. cppreference.com – std::filesystem::directory_iterator ↩

  5. cppreference.com – std::thread::hardware_concurrency ↩