《C++-Concurrency-in-Action》第四章

碎碎念：接下来的一两周就是考试面试考试面试了，前段时间好不容易说服自己专心搞量化，wlb才是该追求的东西，但是京东的一个面试邀约直接打乱了计划。还是想去互联网刷一段实习，去去魅，毕竟如果秋招去量化的话，社招大概率也去不了互联网了。

努努力准备一下吧，算法，八股，项目，今天下午再看一下dp是怎么回事，重点刷一下，我还是太摆烂了，自己定下的目标很少能完成。

————————————————————————————————————

本书的第四章是同步条件，重点讲的是condition_varible条件变量和future，这里重点讲一下future，并以线程池为例子，第四章后半部分的东西过于小众，我们就略过不讲。

cv这个我们常用的条件变量，会存在一个伪唤醒的情况，即cv.wait()的条件明明没有真正准备好，但就被唤醒了（这让我想到了cas_weak)

伪唤醒 指的是：一个正在条件变量上等待的线程，在没有其他线程显式地通知（signal 或 broadcast）它的情况下，自己从等待状态（如 pthread_cond_wait 或 std::condition_variable::wait）中恢复执行了。

换句话说，线程被“莫名其妙”地唤醒了，但线程等待的那个“条件”实际上并未变为真。

为什么会出现伪唤醒？

伪唤醒听起来像是一个 Bug，但实际上，在某些操作系统和线程库的实现中，这是被允许甚至是有意为之的行为。主要原因有：

性能和实现的简化：在支持多处理器（SMP）的系统上，在条件变量的实现中，要完全避免竞争条件和保证绝对的“不唤醒”，可能需要非常复杂的、低效的代码。为了性能和代码简洁性，标准允许在某些边缘情况下发生伪唤醒。
信号中断：在某些系统上，如果一个线程在等待时收到了一个信号（例如 UNIX 信号），等待函数可能会提前返回。虽然这通常会被归类为错误返回，但在某些实现中也可能被视为一种唤醒。

关键点：POSIX 线程标准和 C++ 标准都明确允许伪唤醒的发生。因此，编写代码时绝不能假设“线程被唤醒”就等于“条件已满足”。

带来的问题

如果代码没有正确处理伪唤醒，会导致严重的并发 Bug：

数据竞争：线程在条件不满足的情况下访问了受保护的数据，可能导致数据损坏或读取到无效状态。
逻辑错误：程序基于一个错误的前提（认为条件已满足）继续执行，产生不可预知的结果。
程序崩溃：例如，线程认为队列非空，结果去取元素时发现是空的，导致未定义行为。

如何避免伪唤醒？—— 正确的使用模式

解决方案非常简单和固定：永远不要在条件变量上使用简单的 if 语句来检查条件，而必须使用 while 循环。

这是因为 while 循环可以在线程被唤醒后，再次检查条件是否真正满足。如果发生了伪唤醒，条件依然不成立，线程会自动再次进入等待状态。

错误示例（使用 `if`，易受伪唤醒影响）

// 错误的写法！
std::unique_lock<std::mutex> lock(mutex);
// 检查条件
if (queue.empty()) { // 使用 if 是危险的
    cond_var.wait(lock);
}
// 被唤醒后，直接使用资源（此时队列可能仍然是空的！）
auto item = queue.front();
queue.pop();

正确示例（使用 `while`，防止伪唤醒）

这是多线程编程中的标准范式。

cpp

std::unique_lock<std::mutex> lock(mutex);
// 使用 while 循环检查条件
while (queue.empty()) { // 正确：使用 while
    cond_var.wait(lock); // 1. 释放锁，进入等待
                         // 2. 被唤醒时，会重新获取锁
                         // 3. 跳出 wait 后，再次检查 while 条件
}
// 能执行到这里，说明队列一定非空
auto item = queue.front();
queue.pop();

流程解释：

线程获取互斥锁，检查 queue.empty() 是否为 true。
如果是，则调用 cond_var.wait(lock)。这个调用会原子地释放互斥锁，并将线程挂起。
当有其他线程向条件变量发出通知，或者发生伪唤醒时，线程会从 wait 中返回。在返回之前，它会重新获取互斥锁。
由于我们使用 while 循环，线程会立刻再次检查 queue.empty()。
- 如果是真通知：另一个线程已经将数据放入队列，queue.empty() 为 false，循环结束，线程安全地处理数据。
- 如果是伪唤醒：queue.empty() 仍然为 true，循环继续，线程再次调用 wait 进入等待状态。

C++ 的进一步简化：带谓词的 `wait`

为了简化代码，C++ 标准库提供了可以接受一个“谓词”（一个可调用对象，返回布尔值）的 wait 重载版本。这个版本在内部自动为我们实现了 while 循环。

上面的正确示例可以简写为：

std::unique_lock<std::mutex> lock(mutex);
// wait 的第一个参数是 lock，第二个参数是一个 lambda 函数作为谓词
cond_var.wait(lock, []{ return !queue.empty(); });
// 当执行到这里时，队列一定非空
auto item = queue.front();
queue.pop();

这个 wait(lock, predicate) 的内部逻辑等价于：

1
2
3

while (!predicate()) {
    wait(lock);
}

总结

伪唤醒是什么：没有收到通知就自行唤醒的线程。
为什么存在：出于性能和实现复杂度的考虑，标准允许其发生。
核心对策：永远使用 while 循环来检查条件，而不是 if 语句。
最佳实践：在 C++ 中，优先使用 condition_variable::wait 带谓词的版本，它更简洁、更安全。

记住这个黄金法则：条件变量的等待必须在一个循环中。 这是编写正确、健壮的多线程代码的基石之一

接下来是future和async

std::future<>这里的模板函数签名表示异步进行的任务的返回值类型

所以其实我们可以用auto自动推导（太好用了家人们）
如果async使用默认的参数，即只传入一个函数，会出现下面的情况
// 默认策略：可能是 async | deferred
auto future = std::async(heavy_computation);

// 这允许实现选择策略：
// - 可能立即在新线程执行
// - 也可能延迟到get()时执行
// 因此不要依赖默认策略的特定行为！
如果使用std::lanuch::async，就会在当场创建一个并行线程，然后在后台执行函数
如果使用std::lanuch::deferred,那么不会创建一个并行线程，而是在原线程阻塞执行，适合小任务

std::future、std::shared_future 和 std::async ——
是 C++11 引入的异步任务与结果通信机制的核心。

🧩 一、`std::future` 是什么？

🌱 定义

std::future<T> 是一个未来结果的占位符，
用来在异步任务和主线程之间传递结果。

可以理解为：我现在没有结果，但我将来能拿到。

🧠 使用场景

在一个线程中启动任务；
返回一个 future；
主线程稍后通过 future.get() 获取任务结果。

✅ 基本用法示例

#include <future>
#include <iostream>

int compute() {
    std::cout << "Computing...\n";
    return 42;
}

int main() {
    std::future<int> fut = std::async(std::launch::async, compute);

    // 可以做其他事...
    std::cout << "Main thread working...\n";

    int result = fut.get(); // 阻塞等待结果
    std::cout << "Result: " << result << "\n";
}

解释：

std::async 启动了一个异步任务（可能是新线程）；
返回一个 std::future<int>；
调用 fut.get() 会阻塞，直到 compute() 执行完毕；
get() 只能调用一次！

⚠️ 注意事项

future.get() 调用一次后，结果被取走，不能再用；
如果不调用 get() 就销毁了 future，程序在某些实现下可能阻塞（等待任务完成）。

🌿 二、`std::shared_future` —— 可共享的 future

std::future 的结果只能取一次，这在某些情况下不方便。
例如多个线程都想读取同一个异步结果。

为了解决这个问题，C++11 提供了 std::shared_future。

✅ 基本用法

#include <future>
#include <iostream>

int compute() {
    std::cout << "Computing...\n";
    return 42;
}

int main() {
    std::future<int> fut = std::async(std::launch::async, compute);
    std::shared_future<int> sfut = fut.share(); // 共享 future

    auto worker = [sfut]() {
        std::cout << "Result = " << sfut.get() << "\n";
    };

    std::thread t1(worker);
    std::thread t2(worker);

    t1.join();
    t2.join();
}

解释：

std::future<int>::share() → 转换为 std::shared_future<int>；
多个线程都可以调用 sfut.get()；
shared_future 内部有引用计数；
结果缓存下来，不会被“取走”。

⚙️ 区别总结

特性	`std::future`	`std::shared_future`
获取结果方式	`get()` 一次性取走	`get()` 可多次读取
拷贝	不可拷贝	可拷贝（引用计数）
线程安全	只能单线程取结果	多线程可同时取结果
典型用途	单消费者	多消费者（广播结果）

🚀 三、`std::async` —— 异步任务启动器

std::async 是最常用的异步执行函数。
它返回一个 std::future<T>，代表任务的最终结果。

✅ 基本语法

1	std::future<T> std::async(std::launch policy, Callable&& f, Args&&... args);

f：要执行的函数；
args...：传递给函数的参数；
policy：执行策略。

🔧 启动策略（policy）

策略	含义
`std::launch::async`	立即启动一个新线程执行任务
`std::launch::deferred`	延迟执行，直到调用 `.get()` 或 `.wait()` 时才执行
`std::launch::async	std::launch::deferred`

🌊 示例 1：结合 `shared_future`

std::shared_future<int> sfut = std::async(std::launch::async, add, 10, 20).share();

std::thread t1([sfut]() { std::cout << "T1: " << sfut.get() << "\n"; });
std::thread t2([sfut]() { std::cout << "T2: " << sfut.get() << "\n"; });

t1.join();
t2.join();

多个线程都可以安全地获取结果。

🧠 四、其它实用操作

`future.wait()`

只等待结果可用，不取结果。

1 2	fut.wait(); std::cout << "Task done\n";

⚠️ 五、常见坑点

future.get() 只能调用一次
- 调用后结果被“取走”，future 失效；
- 如果多线程都需要结果，请使用 shared_future。
async 默认策略可能不一定开线程
- 如果没有明确指定 std::launch::async，可能会延迟执行；
- 要强制开线程，请指定策略。
异常会自动传播
- 如果异步任务抛出异常，调用 .get() 时会重新抛出；
- 要在 .get() 外层捕获。

auto fut = std::async([]() { throw std::runtime_error("Error"); });
try {
    fut.get();
} catch (const std::exception& e) {
    std::cout << "Caught: " << e.what() << "\n";
}

🧭 六、总结对比表

名称	功能	可共享	多线程安全	常用场景
`std::future<T>`	异步结果占位符	❌	❌（单线程取结果）	单一消费者异步任务
`std::shared_future<T>`	可共享结果	✅	✅	多线程共享结果
`std::async()`	启动异步任务并返回 `future`	-	-	快速异步执行任务

✅ 一句话记忆：

std::async 启动异步任务，
std::future 等待并取一次结果，
std::shared_future 可以让多个线程共享这个结果。

#include <type_traits>
#include <utility>

// C++17 风格的改进 submit
template<typename F, typename... Args>
auto submit(F&& f, Args&&... args)
{
    using return_type = std::invoke_result_t<std::decay_t<F>, std::decay_t<Args>...>;

    // 用 shared_ptr 管理 packaged_task 的原因如上
    auto task = std::make_shared<std::packaged_task<return_type()>>(
        // lambda 把 f 和 args 全部 move 进来，避免 bind 的拷贝问题
        [func = std::forward<F>(f), ... captured = std::forward<Args>(args)]() mutable -> return_type {
            return std::invoke(func, std::move(captured)...);
        }
    );

    std::future<return_type> fut = task->get_future();

    {
        std::lock_guard<std::mutex> lock(queueMutex);
        tasks.emplace([task](){ (*task)(); });
    }

    condition.notify_one();
    return fut;
}
一个异步的线程池例子

#include <iostream>

int work(int x) {
    std::this_thread::sleep_for(std::chrono::seconds(1));
    return x * 2;
}

int main() {
    ThreadPool pool(4);

    std::future<int> fut = pool.submit(work, 10);

    std::cout << "Main thread working...\n";

    std::cout << "Result = " << fut.get() << std::endl;
}
使用实例

在这里我们认识了invoke,function，packaged_task这三个相似但完全不同的东西

三者的关系和区别

层次关系

std::invoke (调用工具)
    ↓
std::function (同步包装器)  
    ↓
std::packaged_task (异步包装器 + future)

详细对比

特性	`std::function`	`std::packaged_task`	`std::invoke`
用途	通用函数包装	异步任务包装	统一调用机制
存储	✅ 可以存储可调用对象	✅ 可以存储可调用对象	❌ 只是调用工具
异步	❌ 同步调用	✅ 支持异步执行	❌ 同步调用
Future	❌ 无future支持	✅ 关联future获取结果	❌ 无future支持
成员函数	❌ 不能直接包装	❌ 不能直接包装	✅ 支持成员函数
开销	中等	较大	零运行时开销

invoke的本质是将函数指针，成员函数等特殊类型的函数调用统一，提供一个统一的调用方法，但是是一次性的（可以在编译期去推断调用函数的返回值类型，invoke_result_t，就是上面线程池做的)

而function是一种可以多次调用的invoke，所以要起一个名字

packaged_task就是支持异步返回值的function。

packaged_task的内部本质是一个promise

std::promise：你手动提供结果

#include <future>
#include <thread>
#include <iostream>

int main() {
    std::promise<int> prom;          // 创建 promise
    std::future<int> fut = prom.get_future();  // 获取关联的 future

    std::thread t([&prom]() {
        std::this_thread::sleep_for(std::chrono::seconds(1));
        prom.set_value(42);          // 手动设置结果
    });

    std::cout << fut.get() << std::endl; // 阻塞等待结果
    t.join();
}

🔹特点：

promise 不执行任何函数；
你必须在某个地方**手动调用 set_value()**；
适合线程之间传递结果或信号。

`std::packaged_task`：任务自动产生结果

#include <future>
#include <thread>
#include <iostream>

int add(int a, int b) { return a + b; }

int main() {
    std::packaged_task<int(int,int)> task(add); // 打包函数
    std::future<int> fut = task.get_future();

    std::thread t(std::move(task), 2, 3); // 任务执行自动填充结果
    std::cout << fut.get() << std::endl;  // 输出 5
    t.join();
}

显然promise更灵活，但packaged_task更方便。

template<class F>
class packaged_task {
    std::promise<return_type> promise_; // 内部的 promise
    F func_;

public:
    void operator()() {
        try {
            promise_.set_value(func_()); // 自动调用并设置结果
        } catch (...) {
            promise_.set_exception(std::current_exception());
        }
    }

    std::future<return_type> get_future() {
        return promise_.get_future();
    }
};
伪代码

对于packaged_task：

构造时 ❌ 没执行
get_future() ❌ 没执行
调用 task() ✅ 开始执行
get() 只是等待结果已经存在

这里我们引入一点线程池的原理，众所周知，线程池需要一个东西：类型擦除，即将无论什么参数，什么返回值的函数，都用统一的包装和统一的调用。

对于无返回值，有参数的函数，我们最先见到的往往是bind(函数名，（类的实例），参数1，参数2，Args…)，如果参数暂时不可用，就用占位符。这样，函数被包装成为了function<void()>,无函数签名。

当然，lambda这个语法糖也可以做到，其值捕获可以代替函数参数

🧩 问题 1：

“正常情况下 packaged_task 的函数签名需要参数？
但是用 lambda 去初始化，捕获可以代替函数签名参数，是这个意思吗？”

✅ 正确

🧠 1️⃣ `std::packaged_task` 的签名

std::packaged_task<R(Args...)> 的模板参数 Args... 就是它要“接受的参数列表”。

例子：

std::packaged_task<int(int,int)> task([](int a, int b){
    return a + b;
});
task(2,3);  // 传参数执行

这里：

task 的类型是 std::packaged_task<int(int,int)>
它的 operator()(2,3) 会把参数传递给 lambda。

🧩 2️⃣ 为什么在你的代码中写成了 `std::packaged_task<return_type()>`

你的代码里写的是：

std::packaged_task<return_type()> task(
    [func = ..., ...captured = ...]() mutable -> return_type {
        return std::invoke(func, std::move(captured)...);
    }
);

👉 也就是说：

packaged_task 的签名是无参的；
但是这个 lambda 内部捕获了 f 和所有的 args...；
当调用 (*task)() 时，这个 lambda 内部执行：
1
std::invoke(func, std::move(captured)...);
—— 实际上已经使用了原本的 args...。

✅ 结论：

捕获确实“替代”了函数签名的参数。
因为参数值在创建任务时就被拷贝（或移动）进了 lambda 的闭包体。

所以：

函数签名可以是 ()，因为调用时不再需要额外的参数；
所有参数都在创建时绑定（类似于 std::bind 的效果，但效率更高）。

📦 总结一句话：

在这个写法里，lambda 捕获 相当于“提前绑定参数”，
因此 packaged_task 的签名可以是 ()。

🧩 问题 2：

[task](){ (*task)(); } 在按值捕获的时候拷贝了 shared_ptr，那他是被再次包装成了 function？

完全正确 ✅✅✅
这个过程其实就是“用一个小 lambda 包成 std::function<void()>”。

🧠 1️⃣ 捕获行为

[task] 是 按值捕获，
而 task 是一个 std::shared_ptr<std::packaged_task<return_type()>>。

按值捕获 shared_ptr，会发生：

拷贝 shared_ptr（引用计数 +1）

因此：

这个 lambda 内部有自己的 shared_ptr 副本；
即使外层的 task 离开作用域销毁，队列里的 lambda 仍然持有一份 shared_ptr 引用，保证对象仍然存在。

🧱 2️⃣ lambda 被存进 `std::function<void()>`

tasks.emplace([task](){ (*task)(); });

这里假设 tasks 是：

1	std::queue<std::function<void()>> tasks;

那么：

[task](){ (*task)(); } 是一个无参、无返回值的 lambda；
它可以隐式转换成 std::function<void()>；
队列里存放的是 std::function<void()> 类型；
所以这个 lambda 被复制一份存入 std::function 对象中。

⚙️ 3️⃣ 调用链

当 worker 线程执行任务时：

1 2	std::function<void()> job = std::move(tasks.front()); job(); // 执行

发生的调用链：

job()                     --> 调用 std::function<void()>::operator()
  ↓
lambda()                  --> 调用 [task](){ (*task)(); }
  ↓
(*task)()                 --> 调用 packaged_task 的 operator()
  ↓
执行捕获的 lambda         --> 执行原始 f(args...)
  ↓
packaged_task 设置结果到 future

🎯 一整条调用链串起来，未来结果最终出现在 future.get()。

🧠 小结：两段的关系图

submit(f, args...)

  创建 lambda1: [func, captured...]() -> return_type { return invoke(func, args...); }
       ↓
  用 lambda1 初始化 packaged_task<return_type()>（无参，因为参数都捕获了）
       ↓
  得到 shared_ptr<packaged_task<return_type()>>
       ↓
  创建 lambda2: [task](){ (*task)(); }  ← 捕获 shared_ptr（引用计数 +1）
       ↓
  lambda2 转成 std::function<void()> 放入任务队列
       ↓
  worker 执行 job() → 调用 (*task)() → 调用 lambda1 → 执行 func(args...)

✅ 一句话总结

概念	含义
`packaged_task<return_type()>`	无参数任务，但内部 lambda 捕获了所有实参（提前绑定）
`lambda` 捕获	替代了函数签名参数，把参数“记住”在闭包里
`[task](){ (*task)(); }`	用 shared_ptr 拷贝持有任务对象，保证生命周期
存入队列	lambda 被再包装成 `std::function<void()>`，方便统一调度

而对于有返回值，有参数的函数，bind和lambda之外，我们就要再用packaged_task去包装，提前得到future(相当于返回值)，这样，我们就将有返回值有参数的函数，类型抹除成为了function<void()>这个无参无返回值的包装器（实际是转移了参数和返回值）

同时这里用shared_ptr执行异步任务的操作，以及用lambda拷贝（值捕获）shared_ptr使其引用计数+1，防止提前销毁的策略在协程库的fire-and-forget里面也有体现。（有时间把协程库的东西也写在博客）