如何将 `boost::thread_specific_ptr` 与 `for_each` 一起使用

Question

在下面的代码中，Bar 应该为创建成本适中的线程不安全对象建模。 Foo 包含 Bar 并且是多线程的，因此它使用 thread_specific_ptr<Bar> 来创建每个线程 Bar，可以在多次调用 loop 时重复使用相同的 Foo（因此摊销为每个线程创建Bar 的成本）。 Foo 总是创建一个具有相同 num 的 Bar，因此完整性检查应该总是通过，但它失败了。

原因（我认为）在requirement for the thread_specific_ptr destructor中解释：

与此 thread_specific_ptr 关联的所有线程特定实例（可能与此线程关联的实例除外）必须为空。

所以问题是由三件事共同引起的：

1. Bar 在工作线程中创建的对象在Foos thread_specific_ptr 被清理时不会被清理，因此会在main 中的循环迭代中持续存在（本质上是内存泄漏）
2. C++ 运行时在 main 中循环的迭代之间重用 for_each 中的线程
3. C++ 运行时将main 循环中的每个Foo 重新分配到相同的内存地址

thread_specific_ptrs 的索引方式（通过thread_specific_ptr 的内存地址和线程 ID）导致旧的Bars 被意外重用。我理解这个问题；我不明白该怎么做。请注意文档中的注释：

要求是因为为了删除所有这些实例，应该强制实现维护具有关联特定 ptr 的所有线程的列表，这与线程特定数据的目标背道而驰。

我也想避免这种复杂性。

如何使用for_each进行简单的线程管理，同时避免内存泄漏？解决方案要求：

• 每个Foo 的每个线程应该只创建一个Bar（即，不要在for_each 内创建新的Bar）
• 假设 Bar 不是线程安全的。
• 如果可能，使用for_each 使并行循环尽可能简单
• 循环实际上应该并行运行（即，单个 Bar 周围没有互斥锁）
• 由loop 创建的Bar 对象应该可以使用，直到创建它们的Foo 对象被破坏，此时所有Bar 对象也应该被破坏。

以下代码在具有足够内核的机器上编译并应以高概率退出并返回代码 1。

#include <boost/thread/tss.hpp>
#include <execution>
#include <iostream>
#include <vector>

using namespace std;

class Bar {
    public:
    // models a thread-unsafe object
    explicit Bar(int i) : num(i) { }
    int num;
};

class Foo {
    public:
    explicit Foo(int i) : num(i) { }

    void loop() {
        vector<int> idxs(32);
        iota(begin(idxs), end(idxs), 0);
        for_each(__pstl::execution::par, begin(idxs), end(idxs), [&](int) {
            if (ptr.get() == nullptr) {
                // no `Bar` exists for this thread yet, so create one
                Bar *tmp = new Bar(num);
                ptr.reset(tmp);
            }
            // Get the thread-local Bar
            Bar &b = *ptr;

            // Sanity check: we ALWAYS create a `Bar` with the same num as `Foo`;
            // see the `if` block above.
            // Therefore, this condition shouldn't ever be true (but it is!)
            if (b.num != num) {
               cout << "NOT THREAD SAFE: Foo index is " << num << ", but Bar index is " << b.num << endl;
               exit(1);
            }
        });
    }
    boost::thread_specific_ptr<Bar> ptr;
    int num;
};

int main() {
    for(int i = 0; i < 100; i++) {
        Foo f(i);
        f.loop();
    }
    return 0;
}

Answer 1

According to the documentation

~thread_specific_ptr();

Requires:
  All the thread specific instances associated to this thread_specific_ptr 
  (except maybe the one associated to this thread) must be null.

这意味着在Foo 的所有Bar 都被销毁之前，您不得销毁它。这是一个问题，因为execution_policy::par 不必在新的线程池上操作，也不必在for_each() 完成后终止线程。

这足以让我们回答所提出的问题：您只能在以下情况下使用 thread_specific_ptr 和 execution::par 在同一线程上的各种迭代之间共享数据：

• thread_specific_ptr 永远不会被破坏。这是必需的，因为无法知道for_each 的给定迭代是否将是其分配线程的最后一个迭代，并且该线程可能永远再次被调度。
• 在程序结束之前，您可以轻松地在每个线程中泄漏一个指向对象的实例。

你的代码发生了什么

我们已经处于未定义行为领域，但您所看到的行为仍然可以进一步解释。考虑到：

Boost.Thread 使用 thread_specific_ptr 实例的地址作为线程特定指针的键。这避免了创建/销毁需要锁以防止竞争条件的密钥。这对性能有一点影响，因为必须使用关联容器进行访问。

... 并且 Foo 的所有 100 个实例很可能在内存中的同一位置，当工作线程被回收时，您最终会从之前的 Foo 中看到 Bar 的实例，从而导致你的（不准确，见下文）过牌命中。

解决方案：我认为你应该怎么做

我建议您完全放弃 thread_specific_ptr 并使用关联容器手动管理每个线程/每个Foo Bar 实例的池，这使得管理 Bar 对象的生命周期变得很多更直接：

class per_thread_bar_pool {
  std::map<std::thread::id, Bar> bars_;
  // alternatively: 
  // std::map<std::thread::id, std::unique_ptr<Bar>> bars_;
  std::mutex mtx_;

public:
  Bar& get(int num) {
    auto tid = std::this_thread::get_id();

    std::unique_lock l{mtx_};
    auto found = bars_.find(tid);
    if(found == bars_.end()) {
        l.unlock(); // Let other threads access the map while `Bar` is being built.
        Bar new_bar(num);
        // auto new_bar = std::make_unique<Bar>(num); 
        l.lock();

        assert(bars_.find(tid) == bars_.end());
        found = bars_.emplace(tid, std::move(new_bar)).first;
    }
    return found->second;
    // return *found->second;
  }
};


void loop() {
    per_thread_bar_pool bars;
    vector<int> idxs(32);
    iota(begin(idxs), end(idxs), 0);

    for_each(__pstl::execution::par, begin(idxs), end(idxs), [&](int) {
        Bar& current_bar = bars.get(num);
        // ...
   }
}

thread_specific_ptr 已经在后台使用std::map<>（它为每个线程维护一个）。所以在这里介绍一个没什么大不了的。

我们确实引入了互斥锁，但它只在简单的查找/插入地图时发挥作用，而且由于构造 Bar 应该如此昂贵，它很可能影响很小。它还有一个好处是Foo 的多个实例不再相互交互，因此您可以避免在最终从多个线程调用foo::loop() 时可能发生的令人惊讶的错误。

注意：if (b.num != num) { 不是有效测试，因为来自给定Foo 的所有Bar 实例共享相同的num。不过，这只会导致误报。

解决方案：让你的代码工作（几乎）

话虽如此，如果您对同时使用thread_specific_pointer 和execution::par 绝对有信心，那么您必须执行以下操作：

void loop() {
        static boost::thread_specific_ptr<Bar> ptr; // lives till the end of the program

        vector<int> idxs(32);
        iota(begin(idxs), end(idxs), 0);
        for_each(__pstl::execution::par, begin(idxs), end(idxs), [&](int) {
            if (ptr.get() == nullptr || ptr->num != num) {
                // no `Bar` exists for this thread yet, or it's from a previous run
                Bar *tmp = new Bar(num);
                ptr.reset(tmp);
            }
            // Get the thread-local Bar
            Bar &b = *ptr;

        });

但是，这将每个线程泄漏多达 1 个Bar，因为只有在我们尝试重用之前运行的Bar 时才会进行清理。没有办法解决这个问题。