从多个线程访问随机数引擎

Question

这是我的第一个问题，因此请原谅我违反了您的政策。我希望每个线程有一个全局随机数引擎，为此我设计了以下方案：我启动的每个线程都从原子全局 int 获取唯一索引。有一个随机引擎的静态向量，其第 i 个成员被认为由索引为 i 的线程使用。如果索引大于向量大小的元素以同步方式添加到它。为了防止性能损失，我检查了两次索引是否大于向量大小：一次以非同步方式，另一次在锁定互斥锁后。到目前为止一切顺利，但以下示例因各种错误（堆损坏、malloc 错误等）而失败。

#include<vector>
#include<thread>
#include<mutex>
#include<atomic>
#include<random>
#include<iostream>

using std::cout;

std::atomic_uint INDEX_GEN{};
std::vector<std::mt19937> RNDS{};
float f = 0.0f;
std::mutex m{};

class TestAThread {
public:
TestAThread() :thread(nullptr){
    cout << "Calling constructor TestAThread\n";
    thread = new std::thread(&TestAThread::run, this);
}

TestAThread(TestAThread&& source) : thread(source.thread){
    source.thread = nullptr;
    cout << "Calling move constructor TestAThread. My ptr is " << thread << ". Source ptr is" << source.thread << "\n";
}

TestAThread(const TestAThread& source) = delete;

~TestAThread() {
    cout << "Calling destructor TestAThread. Pointer is " << thread << "\n";
    if (thread != nullptr){
        cout << "Deleting thread pointer\n";
        thread->join();
        delete thread;
        thread = nullptr;
    }
}

void run(){
    int index = INDEX_GEN.fetch_add(1);
    std::uniform_real_distribution<float> uniformRnd{ 0.0f, 1.0f };

    while (true){
        if (index >= RNDS.size()){
            m.lock();
            // add randoms in a synchronized manner.
            while (index >= RNDS.size()){
                cout << "index is " << index << ", size is " << RNDS.size() << std::endl;
                RNDS.emplace_back();
            }
            m.unlock();
        }

        f += uniformRnd(RNDS[index]);
    }
}

std::thread*    thread;
};

int main(int argc, char* argv[]){
std::vector<TestAThread> threads;
for (int i = 0; i < 10; ++i){
    threads.emplace_back();
}

cout << f;
}

我做错了什么？！

Answer 1

显然f += ... 将是一个竞争条件，无论右侧如何，但我想你已经知道了。

我看到的主要问题是您使用全局std::vector<std::mt19937> RNDS。您的受互斥锁保护的关键部分仅包含添加新元素；不访问现有元素：

... uniformRnd(RNDS[index]);

这不是线程安全的，因为在另一个线程中调整 RNDS 的大小可能会导致 RNDS[index] 被移动到新的内存位置。实际上，这可能发生在计算引用 RNDS[index] 之后但在 uniformRnd 开始使用它之前，在这种情况下，uniformRnd 认为 Generator& 将是一个悬空指针，可能指向一个新创建的目的。无论如何，uniformRnd 的 operator() 不保证数据竞争 [注 1]，RNDS 的 operator[] 也不保证。

您可以通过以下方式解决此问题：

1. 在受保护部分计算对生成器的引用（或指针）（不能取决于容器的大小是否足够），以及

2. 使用std::deque 而不是std::vector，它不会在调整大小时使引用无效（除非引用的对象已通过调整大小从容器中移除）。

类似这样的事情（关注比赛条件；还有其他事情我可能会做不同的事情）：

std::mt19937& get_generator(int index) {
    std::lock_guard<std::mutex> l(m);
    if (index <= RNDS.size()) RNDS.resize(index + 1);
    return RNDS[index];
}
void run(){
    int index = INDEX_GEN.fetch_add(1);
    auto& gen = get_generator(index);
    std::uniform_real_distribution<float> uniformRnd{ 0.0f, 1.0f };
    while (true) {
        /* Do something with uniformRnd(gen); */
    }
} 

---

[1] uniformRnd 的 operator() 的原型是 template< class Generator > result_type operator()( Generator& g );。换句话说，参数必须是一个 mutable 引用，这意味着它不是隐式线程安全的；只有标准库函数的 const& 参数没有数据竞争。