为什么将条件写转换为无条件写不是线程安全的优化?

Why is transforming a conditional write to an unconditional write not a thread safe optimization?

本文关键字：安全 线程 优化 无条件 条件 转换 为什么      更新时间：2023-10-16

在一篇关于并发性和c++ 11内存模型的演讲中，Herb Sutter给出了一些非法优化的例子。

http://channel9.msdn.com/Shows/Going +深/cpp -和-在- 2012 -草-萨特-原子武器- 2 - 2

从第17分钟的幻灯片:

void f(vector<widget>& v) {
    if(v.length()>0) xMutex.lock();
    for(int i = 0; i < v.length(); ++i)
        ++x;                                  // write is conditional
    if(v.length()>0) xMutex.unlock();
}

"很可能(如果存在严重缺陷)对中心循环进行转换:"

r1 = x;
for(int i = 0; i < v.length(); ++i)
    ++r1;                                     // oops: write is not conditional
x = r1;

他解释说:"……这个写不是有条件的，它会发生在每次执行，即使doOptionalWork是假的，这将注入一个写不受互斥锁保护，注入一个竞争…"

为什么他说发明的写不受互斥锁的保护?我理解完整的转换描述如下:

// "optimized" version 1
void f(vector<widget>& v) {
    if(v.length() > 0) xMutex.lock()
    r1 = x;
    for(int i = 0; i < v.length(); ++i)
        ++r1;
    x = r1;
    if(v.length() > 0) xMutex.unlock();
}

但也可以是这个

// "optimized" version 2
void f(vector<widget>& v) {
    if(v.length() > 0) xMutex.lock()
    r1 = x;
    for(int i = 0; i < v.length(); ++i)
        ++r1;
    if(v.length() > 0) xMutex.unlock();
    x = r1;
}

显然版本2不是线程安全的，但我不确定版本1。版本1线程安全吗?如果没有其他写x的行呢?

刚才我开始输入"要么v.length()是0，要么不是…"，并意识到即使重言式在多线程世界中也会失败。我不知道该从何说起

互斥锁仅在vector容器内有对象时使用。在两个空向量上并发地运行这个方法会导致数据竞争，因为我们根本没有锁，但是我们写到了x。

假设有另一个线程执行以下代码:

xMutex.lock()
++x;
xMutex.unlock();

如果上面的代码与(转换后的)函数f(带一个空向量)同时执行，那么x的增量可能会丢失，尽管这在源代码级别上是不可能的。

mutex. lock()函数调用锁定互斥锁。互斥锁是一个互斥锁，它只允许一个调用者调用lock()并返回。随后的调用者将阻塞，等待原始调用者调用unlock()。

它是'互斥'的，因为只有一个线程被允许获得锁。

这就是互斥锁的定义。

在你发布的代码中，第二个块没有用lock()和unlock() 调用包装增量循环。

这意味着调用在循环中修改变量的函数的两个线程将彼此踩在一起。可以在另一个循环读取该变量之前(或之后)立即写入该变量。代码期望变量与函数一致，因此函数的行为将是不正确的。

考虑到这一点，我制作了一个演示来演示比赛。

#include <chrono>
#include <iostream>
#include <mutex>
#include <thread>
#include <vector>
// we have very simple widgets
typedef int widget;
// x and its mutex are shared by the threads
static int x = 0;
std::mutex xMutex;
static std::chrono::milliseconds central_loop_dur(10);
static std::chrono::milliseconds before_central_loop(0);
void f(std::vector<widget>& v) {
    // The first thread acquires the lock.
    // The second thread has an empty vector so passes through before the lock is released.
    if(v.size() > 0) xMutex.lock();
    // The first thread will take about 50ms to write to x.
    // The second thread reads x nearly concurrently with the first thread. Both see 0.
    int r = x;
    // The first thread passes through here.
    // The second thread snoozes.
    std::this_thread::sleep_for(before_central_loop);
    before_central_loop = std::chrono::milliseconds(200);
    for(auto i : v) {
        std::this_thread::sleep_for(central_loop_dur);
        ++r;
    }
    // At 50ms, the first thread writes to x.
    // At 200ms, the second thread obliviously writes a previous value of x back to x.
    x = r;
    if(v.size() > 0) xMutex.unlock();
}
void thread_main(size_t vec_size) {
    std::vector<widget> v(vec_size);
    f(v);
}
int main() {
    std::thread t1(thread_main, 5);
    std::thread t2(thread_main, 0);
    t1.join();
    t2.join();
    std::cout << "The value of x = " << x << std::endl;
}

如果没有睡眠指令，如果一切都按顺序发生，结果将是x = 5。相反，程序将(很有可能)打印x = 0，即使线程1增加x 5次!

相关文章：