C++现代并发编程与原子操作

一、内存模型与happens-before

C++11引入的内存模型是现代并发编程的基础。理解 happens-before 关系，是写出正确并发代码的核心。

1.1 线程间同步的六条规则

// C++11定义了6种基本同步关系：
//
// ① program order: 同一线程内，前面的操作happens-before后面的操作
// ② mutex: mutex.unlock() happens-before 后续的mutex.lock()
// ③ thread termination: thread.join() returns happens-before 调用点的后续
// ④ semaphore release/acquire: sem.post() happens-before sem.wait()返回
// ⑤ condition variable: notify_one/all happens_before wait()唤醒后检查条件
// ⑥ memory_order_seq_cst: 顺序一致加载/释放（最强保证）

// 示例：线程间安全传递数据
std::thread t1([&]() {
    data = 42;              // ① data赋值
    flag.store(true);        // ② 标记置1（release）
});

std::thread t2([&]() {
    while (!flag.load()) {}  // ③ 自旋等待
    assert(data == 42);      // ④ 一定能读到42
});
// flag.store()的release和load()的acquire配对
// 保证① happens-before ② happens-before ③ → ④

二、原子类型的五种内存序

2.1 内存序的选择决策树

// std::memory_order的五种语义
//
// memory_order_seq_cst：顺序一致性（默认，最安全但最慢）
//   所有线程看到完全一致的操作顺序
//   性能开销：约10-50倍于普通写操作
//
// memory_order_acq_rel：获取-释放（信号量语义）
//   load() = acquire（获取同步）
//   store() = release（释放同步）
//   fetch_add() = acq_rel
//
// memory_order_consume：消费序（最弱，推荐用于依赖链）
//   避免不必要的指令重排（避免" transitive fence"）
//   std::kill_dependency() 显式切断依赖链
//
// memory_order_relaxed：无同步（仅保证原子性）
//   适用于：计数器、自增ID、统计

// 决策树：
// 是否需要跨线程同步？
//   否 → relaxed（性能最好）
//   是 → 是否有严格的全序要求？
//         是 → seq_cst
//         否 → 是否有acquire/release配对？
//               是 → acq_rel / acquire / release
//               否 → consume（依赖链）

// 生产代码推荐：
// ① 简单标志：release/acquire（flag.test_and_set()）
// ② 计数器统计：relaxed（stats++）
// ③ 跨线程数据传递：seq_cst
// ④ lock-free队列：acq_rel + consume

2.2 生产代码示例

// 生产级lock-free栈
template
class LockFreeStack {
    struct Node {
        T data;
        std::atomic next;
        Node(const T& d) : data(d), next(nullptr) {}
    };

    std::atomic head_{nullptr};
    std::atomic item_count_{0};

public:
    ~LockFreeStack() {
        while (pop()); // 清空
    }

    void push(const T& data) {
        Node* n = new Node(data);
        Node* cur = head_.load(std::memory_order_relaxed);
        do {
            n->next.store(cur, std::memory_order_relaxed);
        } while (!head_.compare_exchange_weak(
            cur, n,
            std::memory_order_release,   // 成功时release
            std::memory_order_relaxed    // 失败时重读
        ));
        item_count_.fetch_add(1, std::memory_order_relaxed);
    }

    bool try_pop(T& result) {
        Node* cur = head_.load(std::memory_order_acquire);
        while (cur) {
            Node* next = cur->next.load(std::memory_order_relaxed);
            if (head_.compare_exchange_weak(
                cur, next,
                std::memory_order_seq_cst,
                std::memory_order_relaxed
            )) {
                result = cur->data;
                delete cur;
                item_count_.fetch_sub(1, std::memory_order_relaxed);
                return true;
            }
        }
        return false;
    }
};

三、线程池与任务调度

// C++20 jthread：自动join的线程
void worker(std::stop_token token) {
    while (!token.stop_requested()) {
        // 处理任务
        std::this_thread::sleep_for(10ms);
    }
}

int main() {
    std::jthread t1(worker);  // 析构时自动join
    std::jthread t2(worker);
    // t1, t2析构时自动等待线程结束
}

// C++17 mutex/condition_variable版本
class ThreadPool {
    std::vector workers_;
    std::queue> tasks_;
    std::mutex mtx_;
    std::condition_variable cv_;
    std::atomic stop_{false};

public:
    explicit ThreadPool(size_t n = std::thread::hardware_concurrency()) {
        workers_.reserve(n);
        for (size_t i = 0; i < n; ++i) {
            workers_.emplace_back([this] {
                while (true) {
                    std::function task;
                    {
                        std::unique_lock lock(mtx_);
                        cv_.wait(lock, [this] { return stop_ || !tasks_.empty(); });
                        if (stop_ && tasks_.empty()) return;
                        task = std::move(tasks_.front());
                        tasks_.pop();
                    }
                    task();
                }
            });
        }
    }

    template
    auto submit(F&& f) -> std::future> {
        auto task = std::make_shared()>>(
            std::forward(f)
        );
        {
            std::lock_guard lock(mtx_);
            tasks_.emplace([task]() { (*task)(); });
        }
        cv_.notify_one();
        return task->get_future();
    }

    ~ThreadPool() {
        stop_ = true;
        cv_.notify_all();
        for (auto& t : workers_) t.join();
    }
};

四、锁的性能对比

// 各锁类型性能对比（10线程，100万次操作）
//
// std::mutex:           12ms  ← 基准
// std::shared_mutex:    18ms  ← 读写锁，读多写少时更优
// std::atomic (relaxed):  0.5ms  ← 仅原子变量
// spinlock (自旋锁):    3ms   ← 短临界区最优
// pthread_rwlock:       15ms
//
// 推荐策略：
// 临界区 < 100ns：自旋锁（避免上下文切换）
// 临界区 100ns ~ 10ms：std::mutex
// 读多写少（>10:1）：shared_mutex
// 纯计数/统计：std::atomic

// 自旋锁实现（短临界区）
class SpinLock {
    std::atomic_flag flag_ = ATOMIC_FLAG_INIT;
public:
    void lock() {
        while (flag_.test_and_set(std::memory_order_acquire)) {
            // pause()提示CPU减少功耗，避免总线风暴
            __builtin_ia32_pause();
        }
    }
    bool try_lock() {
        return !flag_.test_and_set(std::memory_order_acquire);
    }
    void unlock() {
        flag_.clear(std::memory_order_release);
    }
};