Chapter 6: Concurrency in C++20_《C++20Get the details》_notes

发布于：2025-04-03 ⋅ 阅读:(21) ⋅ 点赞:(0)

Chapter 6: Concurrency in C++20

6.1 Coroutines

Key Concepts:

  1. Coroutine Framework: Requires implementing promise_type with specific methods (get_return_object, initial_suspend, final_suspend, return_void, unhandled_exception).
  2. Cooperative Suspension: Use co_await to suspend execution and co_yield to produce values.
  3. Generator Pattern: Common use case for producing sequences lazily.

Code Example: Simple Generator

#include <coroutine>
#include <iostream>

template<typename T>
struct Generator {
    struct promise_type {
        T value_;
        auto get_return_object() { return Generator{this}; }
        auto initial_suspend() noexcept { return std::suspend_always{}; }
        auto final_suspend() noexcept { return std::suspend_always{}; }
        void return_void() {}
        void unhandled_exception() { std::terminate(); }
        auto yield_value(T val) {
            value_ = val;
            return std::suspend_always{};
        }
    };

    using Handle = std::coroutine_handle<promise_type>;
    Handle coro_handle;

    explicit Generator(promise_type* p) : coro_handle(Handle::from_promise(*p)) {}
    ~Generator() { if (coro_handle) coro_handle.destroy(); }

    T next() {
        coro_handle.resume();
        return coro_handle.promise().value_;
    }
};

Generator<int> generate_sequence(int start, int step) {
    int current = start;
    while (true) {
        co_yield current;
        current += step;
    }
}

int main() {
    auto gen = generate_sequence(0, 1);
    for (int i = 0; i < 5; ++i) {
        std::cout << gen.next() << " "; // 0 1 2 3 4
    }
}

Compile & Run:

g++ -std=c++20 -fcoroutines -pthread main.cpp && ./a.out

6.2 Atomics

Key Concepts:

  1. Atomic References: std::atomic_ref allows atomic operations on non-atomic objects.
  2. Memory Order: Use memory_order to specify synchronization constraints.

Code Example: Atomic Reference

#include <atomic>
#include <iostream>
#include <thread>

int shared_data = 0;

void increment(std::atomic_ref<int> atomic_ref) {
    for (int i = 0; i < 1000; ++i) {
        atomic_ref.fetch_add(1, std::memory_order_relaxed);
    }
}

int main() {
    std::thread t1(increment, std::atomic_ref<int>(shared_data));
    std::thread t2(increment, std::atomic_ref<int>(shared_data));
    t1.join(); t2.join();
    std::cout << shared_data << "\n"; // 2000 (no data race)
}

Compile & Run:

g++ -std=c++20 -pthread main.cpp && ./a.out

6.3 Semaphores

Key Concepts:

  1. Counting Semaphore: std::counting_semaphore controls access to shared resources.
  2. Acquire/Release: Use acquire() to request a permit and release() to return it.

Code Example: Resource Pool

#include <semaphore>
#include <iostream>
#include <thread>

std::counting_semaphore<3> sem(3); // Allow 3 concurrent accesses

void use_resource(int id) {
    sem.acquire();
    std::cout << "Thread " << id << " acquired resource\n";
    std::this_thread::sleep_for(std::chrono::milliseconds(100));
    sem.release();
}

int main() {
    std::thread workers[5];
    for (int i = 0; i < 5; ++i) {
        workers[i] = std::thread(use_resource, i);
    }
    for (auto& t : workers) t.join();
}

Output:

Thread 0 acquired resource
Thread 1 acquired resource
Thread 2 acquired resource
... (after releases, Threads 3/4 acquire)

6.4 Latches and Barriers

Key Concepts:

  1. Latch (std::latch): Single-use counter for synchronization.
  2. Barrier (std::barrier): Reusable synchronization for phased tasks.

Code Example: Latch

#include <latch>
#include <iostream>
#include <thread>

std::latch sync_point(3); // Wait for 3 threads

void worker() {
    std::cout << "Working...\n";
    sync_point.arrive_and_wait();
    std::cout << "All workers completed\n";
}

int main() {
    std::thread t1(worker), t2(worker), t3(worker);
    t1.join(); t2.join(); t3.join();
}

6.5 Cooperative Interruption

Key Concepts:

  1. Stop Tokens: std::stop_token checks for cancellation requests.
  2. Stop Callbacks: std::stop_callback registers cleanup logic.

Code Example: Interruptible Thread

#include <stop_token>
#include <iostream>
#include <thread>

void task(std::stop_token stoken) {
    while (!stoken.stop_requested()) {
        std::cout << "Working...\n";
        std::this_thread::sleep_for(std::chrono::milliseconds(100));
    }
    std::cout << "Cancelled!\n";
}

int main() {
    std::jthread jt(task);
    std::this_thread::sleep_for(std::chrono::milliseconds(250));
    jt.request_stop(); // Triggers cancellation
}

6.6 std::jthread

Key Concepts:

  1. Automatic Joining: Destructor calls join() automatically.
  2. Interruption Support: Integrates with std::stop_token.

Code Example: (See 6.5 for std::jthread usage)

6.7 Synchronized Output Streams
Key Concepts:

  1. std::osyncstream: Ensures atomicity of output operations.
  2. Thread-Safe Printing: Avoid interleaved characters.

Code Example:

#include <syncstream>
#include <iostream>
#include <thread>

void print(int id) {
    std::osyncstream(std::cout) << "Thread " << id << " says hello\n";
}

int main() {
    std::thread t1(print, 1), t2(print, 2);
    t1.join(); t2.join();
}

Output (No Interleaving):

Thread 1 says hello
Thread 2 says hello

Summary

  • Coroutines: Implement generators via co_yield and custom promise types.
  • Atomics: Use std::atomic_ref for thread-safe operations on non-atomic data.
  • Synchronization: Prefer std::counting_semaphore for resource pools, std::latch for one-time sync.
  • Interruption: Leverage std::jthread and std::stop_token for cooperative cancellation.
  • Output Safety: Use std::osyncstream to prevent garbled console output.

All examples are tested with GCC 11+ (flags: -std=c++20 -pthread -fcoroutines).

Multiple-Choice Questions on C++20 Concurrency (Chapter 6)

Question 1: Coroutine Mechanics
Which statements about C++20 coroutines are TRUE?
A. Coroutines require heap allocation by default.
B. co_await must always suspend execution.
C. The promise_type defines the coroutine’s behavior.
D. Coroutine state is automatically destroyed when co_return is called.

Question 2: std::atomic_ref
Which scenarios are valid uses of std::atomic_ref?
A. Atomic access to a non-atomic variable in a multi-threaded context.
B. Creating an atomic reference to a const object.
C. Enforcing sequential consistency for a shared counter.
D. Replacing std::atomic<T> for better performance.

Question 3: Semaphores
Which statements about std::counting_semaphore are CORRECT?
A. acquire() decrements the semaphore count.
B. try_acquire_for() requires a timeout duration.
C. A binary semaphore is a specialization with max=1.
D. Semaphores guarantee FIFO ordering for blocked threads.

Question 4: Latches & Barriers
Which statements differentiate std::latch from std::barrier?
A. A latch can be reused after reaching zero.
B. A barrier automatically resets its counter.
C. Latches support dynamic thread participation.
D. Barriers execute a completion function when all threads arrive.

Question 5: Cooperative Interruption
Which components interact with std::stop_token?
A. std::stop_source
B. std::stop_callback
C. std::interrupt_flag
D. std::jthread

Question 6: std::jthread Features
What distinguishes std::jthread from std::thread?
A. Automatically joins on destruction.
B. Supports cooperative interruption via request_stop().
C. Uses a shared state for exception propagation.
D. Guarantees lock-free synchronization.

Question 7: Coroutine Handles
Which operations are valid on std::coroutine_handle?
A. Destroying the coroutine state with destroy().
B. Resuming execution with resume().
C. Checking if the coroutine is suspended with done().
D. Accessing the promise object via promise().

Question 8: Atomic Smart Pointers
Which statements about std::atomic<std::shared_ptr<T>> are FALSE?
A. It provides atomic load, store, and exchange operations.
B. It guarantees lock-free implementation.
C. std::atomic_compare_exchange_strong can be used with it.
D. It avoids data races when accessing the managed object.

Question 9: Synchronized Output Streams
What does std::osyncstream ensure?
A. Thread-safe output without interleaving characters.
B. Faster I/O performance compared to std::cout.
C. Atomicity of entire output statements.
D. Compatibility with all stream manipulators.

Question 10: co_await Mechanics
Which types can be used with co_await?
A. Types with a operator co_await() member.
B. Types implementing the Awaitable concept.
C. std::future<T>.
D. std::atomic<int>.

Question 11: Memory Order in Atomics
Which std::memory_order pairs are valid for compare_exchange_strong?
A. std::memory_order_acq_rel (success) and std::memory_order_relaxed (failure).
B. std::memory_order_seq_cst (success) and std::memory_order_acquire (failure).
C. std::memory_order_release (success) and std::memory_order_consume (failure).
D. std::memory_order_relaxed for both success and failure.

Question 12: std::stop_source
Which methods are part of std::stop_source?
A. request_stop()
B. stop_requested()
C. get_token()
D. swap()

Question 13: Coroutine Lifetime
When is a coroutine’s state destroyed?
A. When co_return is executed.
B. When the coroutine handle is explicitly destroyed.
C. When the last reference to the promise object is released.
D. Automatically after the final suspension point.

Question 14: std::atomic_flag
Which operations are valid for std::atomic_flag?
A. test_and_set()
B. clear()
C. wait()
D. is_lock_free()

Question 15: Barriers and Threads
What happens when a thread calls std::barrier::arrive_and_drop()?
A. The thread exits the synchronization group.
B. The barrier’s expected count is reduced by 1.
C. The thread blocks until all threads arrive.
D. The completion function is invoked immediately.

Question 16: co_yield Behavior
What does co_yield value do in a coroutine?
A. Suspends the coroutine and returns value to the caller.
B. Stores value in the promise object.
C. Resumes the coroutine when value is modified.
D. Requires the promise type to have yield_value().

Question 17: std::latch Usage
Which use cases are appropriate for std::latch?
A. Synchronizing a one-time event across multiple threads.
B. Repeating a task until a condition is met.
C. Waiting for a fixed number of threads to reach a point.
D. Implementing a thread-safe queue.

Question 18: Atomic Operations
Which operations on std::atomic<int> are atomic?
A. fetch_add()
B. operator++()
C. load()
D. exchange()

Question 19: Coroutine promise_type
Which functions must be defined in a coroutine’s promise_type?
A. get_return_object()
B. initial_suspend()
C. return_void()
D. unhandled_exception()

Question 20: std::osyncstream Guarantees
Which guarantees does std::osyncstream provide?
A. Output from different threads appears in invocation order.
B. No interleaving of characters from different threads.
C. Synchronization with other std::osyncstream instances.
D. Exception safety equivalent to std::cout.

Answers & Explanations

Question 1
Correct: C, D

  • C: The promise_type controls coroutine behavior (e.g., suspension points).
  • D: The state is destroyed after the final suspension (co_return triggers this).
  • A: Heap allocation depends on the coroutine’s promise type.
  • B: co_await might not suspend if the result is ready (optimization).

Question 2
Correct: A, C

  • A: atomic_ref allows atomic operations on non-atomic variables.
  • C: Sequential consistency (memory_order_seq_cst) is enforced by default.
  • B: atomic_ref requires non-const objects.
  • D: atomic_ref is for existing variables; std::atomic<T> is standalone.

Question 3
Correct: A, B, C

  • A/B/C: Semaphore basics.
  • D: No FIFO guarantee; order is implementation-defined.

Question 4
Correct: B, D

  • B/D: Barriers reset and run completion functions; latches are single-use.
  • A: Latches are single-use.
  • C: Both have fixed thread counts.

Question 5
Correct: A, B, D

  • A/B/D: stop_source produces tokens; jthread uses them.
  • C: interrupt_flag is an implementation detail.

Question 6
Correct: A, B

  • A/B: Key features of jthread.
  • C: Both use shared states.
  • D: No lock-free guarantee.

Question 7
Correct: A, B, C, D

  • All are valid operations on coroutine_handle.

Question 8
Correct: B, C

  • B: atomic<shared_ptr> is rarely lock-free.
  • C: Use std::atomic_compare_exchange_weak instead.

Question 9
Correct: A, C

  • A/C: Ensures atomic output of entire statements.
  • B: Slower due to synchronization.
  • D: Not all manipulators are supported.

Question 10
Correct: A, B

  • A/B: Awaitable types must satisfy the Awaitable concept.
  • C: std::future is not directly awaitable without adaptation.

Question 11
Correct: B, D

  • B/D: Valid combinations.
  • A/C: Failure order must be ≤ success order.

Question 12
Correct: A, B, C, D

  • All are methods of stop_source.

Question 13
Correct: B, D

  • B/D: Explicit destruction or final suspension triggers cleanup.
  • A: co_return leads to final suspension but not immediate destruction.

Question 14
Correct: A, B, C, D

  • All are valid for C++20 atomic_flag.

Question 15
Correct: A, B

  • A/B: arrive_and_drop() reduces the thread count and exits.

Question 16
Correct: A, D

  • A/D: co_yield suspends and requires yield_value() in the promise.

Question 17
Correct: A, C

  • Latches synchronize one-time events with a fixed thread count.

Question 18
Correct: A, B, C, D

  • All are atomic operations.

Question 19
Correct: A, B, D

  • return_void() or return_value() is required (not both).

Question 20
Correct: B

  • B: Prevents interleaving.
  • A: Order is not guaranteed.
  • C: Each osyncstream has its own buffer.
  • D: Exceptions may still occur.

C++20 Concurrency Design Challenges

Problem 1: Thread-Safe Lock-Free Queue with atomic_ref
Task: Implement a lock-free queue using C++20 std::atomic_ref for thread-safe push and pop operations.
Code:

#include <atomic>
#include <memory>

template<typename T>
class LockFreeQueue {
    struct Node {
        T data;
        std::atomic<Node*> next;
        Node(T val) : data(val), next(nullptr) {}
    };
    std::atomic<Node*> head, tail;

public:
    void push(T val) {
        Node* newNode = new Node(val);
        Node* oldTail = tail.load(std::memory_order_relaxed);
        while (!tail.compare_exchange_weak(oldTail, newNode, 
               std::memory_order_release, std::memory_order_relaxed)) {}
        oldTail->next.store(newNode, std::memory_order_release);
    }

    bool pop(T& val) {
        Node* oldHead = head.load(std::memory_order_relaxed);
        if (oldHead == tail.load(std::memory_order_acquire)) return false;
        head.store(oldHead->next, std::memory_order_relaxed);
        val = oldHead->data;
        delete oldHead;
        return true;
    }
};

// Test
#include <thread>
int main() {
    LockFreeQueue<int> q;
    std::jthread t1([&]{ q.push(1); });
    std::jthread t2([&]{ int v; q.pop(v); });
}

Explanation: Uses atomic operations to manage queue nodes. push CAS-loop ensures atomic tail update. pop checks head/tail consistency.

Problem 2: Custom Coroutine Generator with Exception Propagation
Task: Design a coroutine generator that propagates exceptions to the caller.
Code:

#include <iostream>
#include <stdexcept>
#include <coroutine>

template<typename T>
struct Generator {
    struct promise_type {
        T value;
        std::exception_ptr eptr;
        auto yield_value(T v) { value = v; return suspend_always{}; }
        auto initial_suspend() { return suspend_never{}; }
        auto final_suspend() noexcept { return suspend_always{}; }
        void unhandled_exception() { eptr = std::current_exception(); }
        Generator get_return_object() { return Generator{this}; }
    };
    std::coroutine_handle<promise_type> h;
    explicit Generator(promise_type* p) : h(std::coroutine_handle<promise_type>::from_promise(*p)) {}
    T next() { h.resume(); if (h.promise().eptr) std::rethrow_exception(h.promise().eptr); return h.promise().value; }
};

Generator<int> gen() {
    co_yield 1;
    throw std::runtime_error("Error!");
    co_yield 2;
}

// Test
int main() {
    auto g = gen();
    try { 
        std::cout << g.next(); 
        g.next(); 
    } catch(const std::exception& e) {
        std::cout << "Caught: " << e.what();
    }
}

Explanation: The promise_type captures exceptions via unhandled_exception. Caller uses next() to check and rethrow.

Problem 3: Counting Semaphore for Resource Pool
Task: Use std::counting_semaphore to limit concurrent access to a resource pool.
Code:

#include <semaphore>
#include <thread>
#include <vector>

std::counting_semaphore<3> sem(3); // Allow 3 concurrent accesses

void use_resource(int id) {
    sem.acquire();
    std::cout << id << " using resource\n";
    std::this_thread::sleep_for(std::chrono::seconds(1));
    sem.release();
}

// Test
int main() {
    std::vector<std::jthread> threads;
    for (int i = 0; i < 10; ++i)
        threads.emplace_back(use_resource, i);
}

Explanation: Semaphore ensures only 3 threads access the resource simultaneously.

Problem 4: Multi-Phase Processing with std::latch
Task: Coordinate parallel processing phases using std::latch.
Code:

#include <latch>
#include <thread>

void worker(std::latch& latch, int phase) {
    latch.arrive_and_wait();
    std::cout << "Phase " << phase << " done\n";
}

// Test
int main() {
    std::latch latch(3); // 3 workers per phase
    std::jthread t1(worker, std::ref(latch), 1);
    std::jthread t2(worker, std::ref(latch), 1);
    std::jthread t3(worker, std::ref(latch), 1);
}

Explanation: Workers synchronize at the latch, proceeding only when all arrive.

Problem 5: Thread-Safe Object Sharing with atomic_shared_ptr
Task: Implement thread-safe shared object access using std::atomic_shared_ptr.
Code:

#include <atomic>
#include <memory>
#include <thread>

struct Data { int x = 0; };
std::atomic<std::shared_ptr<Data>> ptr(std::make_shared<Data>());

void increment() {
    std::shared_ptr<Data> local = ptr.load();
    while (!ptr.compare_exchange_weak(local, 
           std::make_shared<Data>(*local))) {}
    local->x++;
}

// Test
int main() {
    std::jthread t1(increment);
    std::jthread t2(increment);
    t1.join(); t2.join();
    std::cout << ptr.load()->x; // Should be 2
}

Explanation: Atomic CAS ensures safe updates to the shared pointer.

Problem 6: Barrier for Iterative Parallel Algorithm
Task: Use std::barrier to synchronize iterative parallel tasks.
Code:

#include <barrier>
#include <thread>

std::barrier bar(2, []{ std::cout << "Barrier reached\n"; });

void task() {
    for (int i = 0; i < 3; ++i) {
        std::cout << i << " ";
        bar.arrive_and_wait();
    }
}

// Test
int main() {
    std::jthread t1(task);
    std::jthread t2(task);
}

Explanation: Barrier syncs two threads at each iteration, invoking completion function.

Problem 7: Interruptible jthread with Stop Token
Task: Use std::jthread’s stop token to interrupt a long-running task.
Code:

#include <stop_token>
#include <thread>

void task(std::stop_token stoken) {
    while (!stoken.stop_requested()) {
        std::cout << "Working...\n";
        std::this_thread::sleep_for(std::chrono::milliseconds(100));
    }
    std::cout << "Stopped!\n";
}

// Test
int main() {
    std::jthread t(task);
    std::this_thread::sleep_for(std::chrono::milliseconds(500));
    t.request_stop();
}

Explanation: jthread automatically joins, and stop token checks for interruption.

Problem 8: Interruptible Condition Variable Wait
Task: Implement a wait on a condition variable that can be interrupted via std::stop_token.
Code:

#include <condition_variable>
#include <mutex>
#include <stop_token>

std::mutex mtx;
std::condition_variable_any cv;
bool ready = false;

void wait_for_data(std::stop_token stoken) {
    std::unique_lock lock(mtx);
    cv.wait(lock, stoken, []{ return ready; });
    if (!stoken.stop_requested()) std::cout << "Data ready!\n";
}

// Test
int main() {
    std::jthread t(wait_for_data);
    std::this_thread::sleep_for(std::chrono::seconds(1));
    t.request_stop();
}

Explanation: Overloaded wait accepts stop token, allowing interruption.

Problem 9: Async Task Chain with Coroutines
Task: Implement coroutine-based async tasks chained with co_await.
Code:

#include <coroutine>
#include <future>
#include <iostream>

struct Task {
    struct promise_type {
        std::future<int> fut;
        auto initial_suspend() { return std::suspend_never{}; }
        auto final_suspend() noexcept { return std::suspend_never{}; }
        void return_value(int v) { fut.set_value(v); }
        Task get_return_object() { return {fut.get()}; }
        void unhandled_exception() { std::terminate(); }
    };
    int result;
};

Task async_task() {
    co_return 42;
}

// Test
int main() {
    auto t = async_task();
    std::cout << t.result; // Output: 42
}

Explanation: Coroutine returns a future, resolved when the task completes.

Problem 10: Spinlock with atomic_flag
Task: Implement a spinlock using std::atomic_flag.
Code:

#include <atomic>
#include <thread>

class Spinlock {
    std::atomic_flag flag = ATOMIC_FLAG_INIT;
public:
    void lock() { while(flag.test_and_set(std::memory_order_acquire)); }
    void unlock() { flag.clear(std::memory_order_release); }
};

// Test
Spinlock sl;
void task(int id) {
    sl.lock();
    std::cout << id << " locked\n";
    sl.unlock();
}

int main() {
    std::jthread t1(task, 1);
    std::jthread t2(task, 2);
}

Explanation: test_and_set in loop provides atomic acquire, clear releases.

Problem 11: Synchronized Output with osyncstream
Task: Use std::osyncstream to prevent interleaved console output.
Code:

#include <syncstream>
#include <thread>

void log(int id) {
    std::osyncstream(std::cout) << "Thread " << id << " logging\n";
}

// Test
int main() {
    std::vector<std::jthread> threads;
    for(int i = 0; i < 5; ++i)
        threads.emplace_back(log, i);
}

Explanation: osyncstream ensures atomic output operations across threads.

Problem 12: Custom Awaitable for Thread Pool Execution
Task: Implement an awaitable that schedules coroutine continuation on a thread pool.
Code:

#include <coroutine>
#include <thread>
#include <queue>
#include <mutex>

struct ThreadPool {
    std::queue<std::coroutine_handle<>> tasks;
    std::mutex mtx;
    void enqueue(std::coroutine_handle<> h) {
        std::lock_guard lock(mtx);
        tasks.push(h);
    }
    void run() {
        while (!tasks.empty()) {
            auto h = tasks.front();
            tasks.pop();
            h.resume();
        }
    }
};

struct Awaitable {
    ThreadPool& pool;
    bool await_ready() { return false; }
    void await_suspend(std::coroutine_handle<> h) { pool.enqueue(h); }
    void await_resume() {}
};

// Usage
ThreadPool pool;
Awaitable schedule_on_pool(ThreadPool& p) { return {p}; }

void coro_task() {
    co_await schedule_on_pool(pool);
    std::cout << "Running in pool\n";
}

// Test
int main() {
    coro_task();
    pool.run(); // Output: "Running in pool"
}

Explanation: Awaitable suspends coroutine and enqueues to thread pool for resumption.

Problem 13: Lock-Free Stack with Hazard Pointers
Task: Implement a lock-free stack using hazard pointers for safe memory reclamation.
Code:

#include <atomic>
#include <memory>
#include <thread>

template<typename T>
class LockFreeStack {
    struct Node {
        T data;
        Node* next;
        Node(T d) : data(d), next(nullptr) {}
    };
    std::atomic<Node*> head;
    // Hazard pointer logic omitted for brevity
public:
    void push(T val) {
        Node* newNode = new Node(val);
        newNode->next = head.load();
        while (!head.compare_exchange_weak(newNode->next, newNode));
    }
    bool pop(T& val) {
        Node* oldHead = head.load();
        if (!oldHead) return false;
        while (!head.compare_exchange_weak(oldHead, oldHead->next));
        val = oldHead->data;
        delete oldHead;
        return true;
    }
};

// Test
int main() {
    LockFreeStack<int> stack;
    std::jthread t1([&]{ stack.push(1); });
    std::jthread t2([&]{ int v; stack.pop(v); });
}

Explanation: Basic lock-free stack (hazard pointers omitted for space). CAS manages head updates.

Problem 14: Data Pipeline with Coroutine Generators
Task: Create a data processing pipeline using coroutine generators and range adaptors.
Code:

#include <coroutine>
#include <vector>
#include <ranges>
#include <iostream>

Generator<int> gen() {
    for (int i = 0; ; ++i) co_yield i;
}

// Test
int main() {
    auto numbers = gen() 
        | std::views::take(5) 
        | std::views::filter([](int x){ return x % 2 == 0; });
    for (int i : numbers) std::cout << i << " "; // 0 2 4
}

Explanation: Coroutine generates infinite sequence, filtered by range views.

Problem 15: Multi-Stage Map-Reduce with Futures
Task: Implement parallel map-reduce using std::future and barriers.
Code:

#include <future>
#include <vector>
#include <numeric>
#include <latch>

int main() {
    std::vector<int> data(1000, 1); // Sum should be 1000
    std::latch latch(data.size());
    auto map = [&](int& x) { x *= 2; latch.count_down(); };
    std::vector<std::future<void>> futures;
    for (auto& x : data) 
        futures.push_back(std::async(std::launch::async, map, std::ref(x)));
    latch.wait();
    int sum = std::reduce(data.begin(), data.end());
    std::cout << sum; // 2000
}

Explanation: Parallel map doubles each element, latch syncs before reduction.

Key Takeaways:

  1. Atomic Types: Essential for lock-free structures; use compare_exchange for atomic updates.
  2. Coroutines: Manage state via promise_type; handle exceptions and resumption.
  3. Concurrency Primitives: Semaphores, latches, barriers coordinate complex workflows.
  4. Thread Management: jthread simplifies lifecycle with automatic joining and stop tokens.
  5. Memory Model: Understand acquire/release semantics to avoid data races.

网站公告

今日签到

点亮在社区的每一天
去签到

热门文章

最新发布