c++
c++26
concurrency
Introduction
In this article, we will examine a concurrent programming example in C++ in detail, using it as a case
study to explore dynamic task generation, work-completion detection, and the safe aggregation of partial
results.
Given the path of a directory, our goal is to produce a statistical summary of the contents of its
directory tree. For each file extension encountered (.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 accumulating partial statistics that will ultimately 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.
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.
Such a structure greatly simplifies the parallelization of graph traversals, including the directory
hierarchies considered in this article.
This class keeps pending tasks of type T in a private standard queue named tasks_, of type std::queue<T>.
In addition, it maintains a counter called active_ that tracks the number of tasks currently being
processed. This counter is incremented whenever a task is acquired and decremented when processing of that task completes.
As a result, the queue can automatically detect global completion, 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::mutex, while a std::condition_variable 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.
It is important to note 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 purpose 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)
that signals that no further work can be generated and that the worker may terminate.
-
complete(): Atomically performs two actions: (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 and 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::async, while the main thread acts as an additional
worker processing tasks from the shared queue.
Each pending directory dir is represented as a task stored in the dynamic_task_queue<filesystem::path>,
with std::filesystem::is_directory(dir) == true. Workers execute the process_directories() function,
which acquires directories from the queue through acquire(), inspects their contents, and records
statistics for 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.
The implementation of process_directories() is based on std::filesystem::directory_iterator, 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, it is worth emphasizing that, 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(), 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,
such as NVMe SSDs and mechanical hard drives, may scale quite differently.
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