Concurrency Skill
Production-Grade Development Skill | C++ Concurrent Programming
Master C++ concurrency from C++11 threads to C++20 coroutines and parallel algorithms.
Thread Basics
Creating and Managing Threads
#include <thread>
#include <iostream>
void worker(int id) {
std::cout << "Worker " << id << " running\n";
}
int main() {
std::thread t1(worker, 1);
std::thread t2(worker, 2);
t1.join();
t2.join();
}
void compute(int input, int& result) {
result = input * input;
}
#include <stop_token>
void cancellable_work(std::stop_token st) {
while (!st.stop_requested()) {
}
}
std::jthread worker(cancellable_work);
Thread-Local Storage
thread_local int tls_counter = 0;
void increment() {
++tls_counter;
}
class Logger {
thread_local static Logger* instance_;
public:
static Logger& instance() {
if (!instance_) {
instance_ = new Logger();
}
return *instance_;
}
};
Synchronization Primitives
Mutex and Locks
#include <mutex>
#include <shared_mutex>
std::mutex mtx;
std::shared_mutex shared_mtx;
void exclusive_access() {
std::lock_guard<std::mutex> lock(mtx);
}
void transfer(Account& from, Account& to, int amount) {
std::unique_lock lock1(from.mtx, std::defer_lock);
std::unique_lock lock2(to.mtx, std::defer_lock);
std::lock(lock1, lock2);
from.balance -= amount;
to.balance += amount;
}
void safe_transfer(Account& from, Account& to, int amount) {
std::scoped_lock lock(from.mtx, to.mtx);
from.balance -= amount;
to.balance += amount;
}
void read_data() {
std::shared_lock lock(shared_mtx);
}
void {
;
}
Condition Variables
#include <condition_variable>
#include <queue>
template<typename T>
class ThreadSafeQueue {
std::queue<T> queue_;
mutable std::mutex mtx_;
std::condition_variable cv_;
public:
void push(T value) {
{
std::lock_guard lock(mtx_);
queue_.push(std::move(value));
}
cv_.notify_one();
}
T pop() {
std::unique_lock lock(mtx_);
cv_.wait(lock, [this] { return !queue_.empty(); });
T value = std::move(queue_.front());
queue_.pop();
return value;
}
bool try_pop(T& value) {
std::lock_guard lock(mtx_);
if (queue_.empty()) return false;
value = std::move(queue_.front());
queue_.();
;
}
};
C++20 Synchronization
#include <semaphore>
#include <latch>
#include <barrier>
std::counting_semaphore<10> sem(10);
void limited_access() {
sem.acquire();
sem.release();
}
std::latch start_latch(1);
std::latch done_latch(num_workers);
void worker() {
start_latch.wait();
done_latch.count_down();
}
std::barrier sync_point(num_threads, []() noexcept {
});
void iteration_worker() {
for (int i = 0; i < iterations; ++i) {
sync_point.();
}
}
Atomic Operations
Atomic Types
#include <atomic>
std::atomic<int> counter{0};
std::atomic<bool> flag{false};
std::atomic<std::shared_ptr<Data>> shared_data;
counter.fetch_add(1);
counter.store(0);
int val = counter.load();
int expected = 0;
bool success = counter.compare_exchange_strong(expected, 1);
while (!counter.compare_exchange_weak(expected, expected + 1)) {
}
Memory Ordering
std::atomic<int> data{0};
std::atomic<bool> ready{false};
void produce() {
data.store(42, std::memory_order_relaxed);
ready.store(true, std::memory_order_release);
}
void consume() {
while (!ready.load(std::memory_order_acquire)) {
}
assert(data.load(std::memory_order_relaxed) == 42);
}
Async Programming
std::async and Futures
#include <future>
std::future<int> result = std::async(std::launch::async, []() {
return expensive_computation();
});
int value = result.get();
if (result.wait_for(std::chrono::seconds(0)) == std::future_status::ready) {
int value = result.get();
}
auto f1 = std::async(std::launch::async, func);
auto f2 = std::async(std::launch::deferred, func);
auto f3 = std::async(std::launch::async | std::launch::deferred, func);
Promise and Future
void worker(std::promise<int> promise) {
try {
int result = compute();
promise.set_value(result);
} catch (...) {
promise.set_exception(std::current_exception());
}
}
std::promise<int> promise;
std::future<int> future = promise.get_future();
std::thread t(worker, std::move(promise));
try {
int result = future.get();
} catch (const std::exception& e) {
}
t.join();
Packaged Task
std::packaged_task<int(int, int)> task([](int a, int b) {
return a + b;
});
std::future<int> result = task.get_future();
std::thread t(std::move(task), 2, 3);
t.join();
int sum = result.get();
Parallel Algorithms (C++17)
#include <execution>
#include <algorithm>
#include <numeric>
std::vector<int> v(1'000'000);
std::sort(std::execution::par, v.begin(), v.end());
std::sort(std::execution::par_unseq, v.begin(), v.end());
std::transform(std::execution::par, v.begin(), v.end(), v.begin(),
[](int x) { return x * 2; });
long sum = std::reduce(std::execution::par, v.begin(), v.end(), 0L);
long dot_product = std::transform_reduce(
std::execution::par,
v1.begin(), v1.end(),
v2.begin(),
0L
);
std::for_each(std::execution::par_unseq, v.begin(), v.end(),
[](int& x) { x = process(x); });
Lock-Free Programming
Lock-Free Stack
template<typename T>
class LockFreeStack {
struct Node {
T data;
Node* next;
Node(T val) : data(std::move(val)), next(nullptr) {}
};
std::atomic<Node*> head_{nullptr};
public:
void push(T value) {
Node* new_node = new Node(std::move(value));
new_node->next = head_.load(std::memory_order_relaxed);
while (!head_.compare_exchange_weak(
new_node->next, new_node,
std::memory_order_release,
std::memory_order_relaxed)) {
}
}
std::optional<T> pop() {
Node* old_head = head_.load(std::memory_order_relaxed);
while (old_head && !head_.compare_exchange_weak(
old_head, old_head->next,
std::memory_order_acquire,
std::memory_order_relaxed)) {
}
if (!old_head) return std::nullopt;
T value = std::move(old_head->data);
delete old_head;
return value;
}
};
Common Concurrency Pitfalls
| Pitfall | Description | Solution |
|---|
| Data Race | Unsynchronized access | Use mutex or atomic |
| Deadlock | Circular lock dependency | Lock ordering, std::scoped_lock |
| Livelock | Threads can't progress | Backoff, randomization |
| Priority Inversion | High priority blocked | Priority inheritance |
| False Sharing | Cache line contention | alignas(64) padding |
| ABA Problem | CAS sees same value | Hazard pointers, epoch-based |
Troubleshooting Decision Tree
Concurrency issue?
├── Crash or corruption
│ ├── Data race? → Run with ThreadSanitizer
│ ├── Use after free? → Check thread lifetimes
│ └── Iterator invalidation? → Copy or lock
├── Deadlock (program hangs)
│ ├── Get thread stacks: gdb -p <pid>, thread apply all bt
│ ├── Check lock ordering
│ └── Use std::scoped_lock for multiple locks
├── Performance issues
│ ├── Too much contention? → Reduce critical section
│ ├── False sharing? → Align to cache line
│ └── Lock convoy? → Use reader-writer lock
└── Inconsistent behavior
├── Memory ordering? → Use stronger ordering
├── Visibility? → Use proper synchronization
└── Race condition? → Add mutex protection
Unit Test Template
#include <gtest/gtest.h>
#include <thread>
#include <vector>
#include <atomic>
class ConcurrencyTest : public ::testing::Test {
protected:
static constexpr int NUM_THREADS = 8;
static constexpr int ITERATIONS = 10000;
};
TEST_F(ConcurrencyTest, AtomicCounterIsThreadSafe) {
std::atomic<int> counter{0};
std::vector<std::thread> threads;
for (int i = 0; i < NUM_THREADS; ++i) {
threads.emplace_back([&counter]() {
for (int j = 0; j < ITERATIONS; ++j) {
counter.fetch_add(1);
}
});
}
for (auto& t : threads) t.join();
EXPECT_EQ(counter.load(), NUM_THREADS * ITERATIONS);
}
TEST_F(ConcurrencyTest, MutexProtectsSharedData) {
int counter = 0;
std::mutex mtx;
std::vector<std::thread> threads;
for (int i = 0; i < NUM_THREADS; ++i) {
threads.emplace_back([&]() {
for ( j = ; j < ITERATIONS; ++j) {
std::lock_guard (mtx);
++counter;
}
});
}
(& t : threads) t.();
(counter, NUM_THREADS * ITERATIONS);
}
(ConcurrencyTest, ThreadSafeQueueWorks) {
ThreadSafeQueue<> queue;
std::atomic<> sum{};
;
;
producer.();
consumer.();
(sum.(), );
}
Integration Points
| Component | Interface |
|---|
performance-optimizer | Parallel optimization |
memory-specialist | Thread-safe allocation |
cpp-debugger-agent | Race detection (TSan) |
stl-master | Parallel algorithms |
C++ Plugin v3.0.0 - Production-Grade Development Skill
