使用 thread_local 维护并发内存缓冲区时发生错误答案

【问题标题】：Error occurred when using thread_local to maintain a concurrent memory buffer使用 thread_local 维护并发内存缓冲区时发生错误
【发布时间】：2020-10-06 15:04:57
【问题描述】：

在下面的代码中，我想创建一个允许多个线程同时读/写的内存缓冲区。一次，所有线程将并行读取此缓冲区，稍后它们将并行写入缓冲区。但不会同时进行读/写操作。

为此，我使用vector 或shared_ptr<vector<uint64_t>>。当一个新线程到达时，它将被分配一个新的vector<uint64_t>，并且只写入它。两个线程不会写入同一个向量。

我使用 thread_local 来跟踪当前线程将写入的向量索引和偏移量。当我需要向memory_ 变量添加新缓冲区时，我使用互斥锁来保护它。

class TestBuffer {
public:
    thread_local static uint32_t index_;
    thread_local static uint32_t offset_;
    thread_local static bool ready_;

    vector<shared_ptr<vector<uint64_t>>> memory_;
    mutex lock_;

    void init() {
        if (!ready_) {
            new_slab();
            ready_ = true;
        }
    }

    void new_slab() {
        std::lock_guard<mutex> lock(lock_);
        index_ = memory_.size();
        memory_.push_back(make_shared<vector<uint64_t>>(1000));
        offset_ = 0;
    }

    void put(uint64_t value) {
        init();
        if (offset_ == 1000) {
            new_slab();
        }
        if(memory_[index_] == nullptr) {
            cout << "Error" << endl;
        }
        *(memory_[index_]->data() + offset_) = value;
        offset_++;
    }
};

thread_local uint32_t TestBuffer::index_ = 0;
thread_local uint32_t TestBuffer::offset_ = 0;
thread_local bool TestBuffer::ready_ = false;

int main() {
    TestBuffer buffer;
    vector<std::thread> threads;
    for (int i = 0; i < 10; ++i) {
        thread t = thread([&buffer, i]() {
            for (int j = 0; j < 10000; ++j) {
                buffer.put(i * 10000 + j);
            }
        });
        threads.emplace_back(move(t));
    }
    for (auto &t: threads) {
        t.join();
    }
}

代码的行为与预期不符，并在put 函数中报告错误。根本原因是memory_[index_] 有时会返回nullptr。但是，我不明白为什么这是可能的，因为我认为我已经正确设置了这些值。感谢您的帮助！

【问题讨论】：

在您的代码中看到这么多thread_local 是不正常的。不要使用这样的全局变量，如果它是线程本地的，则将其范围限定在线程的函数中。我不确定这种策略是否安全，也不推荐。如果您希望在线程之间共享一个可实现的内存池，但不需要任何thread_local 开销。
这里的目标是拥有某种可以并行写入的容器，然后在所有数据累积后读回吗？如果是这样，为什么所有这些大惊小怪，而不是线程将向量添加到目标queue 当它们已满/准备好时？ C++ 模式将是某种流式对象，您可以在其中执行Buffer b 然后b << value 然后b.close()。
您的 put() 方法只会将向量展开一次。当 offset_ 达到 2000 等时会发生什么？
@JohnSheridan 没关系，new_slab 将 offset_ 重置为零。
@PaulSanders 确实如此

标签： c++ multithreading thread-local

【解决方案1】：

您在put 中存在由new_slab() 引起的竞争条件。当new_slab 调用memory_.push_back() 时，_memory 向量可能需要调整自身大小，如果在调整大小的过程中另一个线程正在执行put，memory_[index_] 可能会访问陈旧数据。

一种解决方案是通过锁定互斥锁来保护_memory 向量：

{
    std::lock_guard<mutex> lock(lock_);

    if(memory_[index_] == nullptr) {
        cout << "Error" << endl;
    }
    *(memory_[index_]->data() + offset_) = value;
}

另一个是提前在memory_向量中预留你需要的空间。

【讨论】：