如何让N个线程运行特定代码而又不会出现数据竞争错误？

Question

我有以下问题：N进程锁定器是一种同步机制，它允许N个线程，其中N是固定数，等到他们都达到某个程度一旦所有线程都达到此特定点（更衣室），它们都可以继续。可重复使用的储物柜是可以多次使用的储物柜相同的代码，即，它可以用于等待所有N个线程到达某个点，然后释放它们继续，然后在下面的代码行中重用它，以等待所有N个线程到达该点，然后释放它们。我编写了一个代码，可以使用互斥量和信号量解决此问题。我尝试用一​​个简单的代码测试我的代码，但它成功了，但是当我使用valgrind --tool=helgrind ./test运行测试时我知道我有数据竞赛错误！我不明白为什么会出现这种情况，我读了很多遍代码，也无法弄清楚如果例如5个线程到达了wait函数（它将等待所有线程，然后在N个线程释放一次）的情况下会发生什么问题？到达）谁能帮我，只为我指出代码中出现此问题的位置？这是我的代码：（问题出在等待功能中）

#include <iostream>
#include "Locker.h"
using namespace std;




Locker::Locker(unsigned int num_of_threads){
    NumOfThreads=num_of_threads;
    count=0;
    tmp=0;
    flag=0;
    sem_init(&sem, 0, 0);
    sem_init(&wait_sem, 0, 0);
    pthread_mutex_init(&m1,NULL);
    pthread_mutex_init(&m2,NULL);
}






Locker::~Locker(){
    sem_destroy(&sem);
    sem_destroy(&wait_sem);
    pthread_mutex_destroy(&m1);
    pthread_mutex_destroy(&m2);
}


void Locker::wait(){
    pthread_mutex_lock(&m1);
    pthread_mutex_lock(&m2);
    count+=1;
    if(count == NumOfThreads ){
        for (unsigned int i = 0; i < count -1 ; ++i) { //  change it for while loop
            flag=1;
            sem_post(&sem);
        }
    }
    pthread_mutex_unlock(&m2);
    if(!flag){
        pthread_mutex_unlock(&m1);
        sem_wait(&sem);
    }else{
        flag=0;
    }
        //**************************************************** waiting for the N threads
    pthread_mutex_lock(&m2);  // we need to block this section until we finished with count  ** here m2 won't lock untill we finished with N thread
    count-=1;
    if(count == 0){
        pthread_mutex_unlock(&m2);
        sem_post(&wait_sem);
        while(tmp != (NumOfThreads-1)){}
        tmp=0;
        pthread_mutex_unlock(&m1);
        return;
    }
    pthread_mutex_unlock(&m2);
    sem_wait(&wait_sem);
    pthread_mutex_lock(&m2);
    tmp+=1;
    pthread_mutex_unlock(&m2);
    sem_post(&wait_sem);
};

如果需要，这里是h文件：

#ifndef LOCKER_H_
#define LOCKER_H_

#include <semaphore.h>
#include <pthread.h>

class Locker{
public:
    Locker(unsigned int num_of_threads);
    void wait();
    ~Locker();

    unsigned int NumOfThreads=0;
    unsigned int count;
    unsigned int tmp;
    int flag;
    sem_t sem;
    sem_t wait_sem;
    //sem_t sem_lock;
    pthread_mutex_t m1;
    pthread_mutex_t m2;

    // TODO: define the member variables
    // Remember: you can only use semaphores!
};

#endif // LOCKER_H_

Answer 1

TL; DR：看来比赛涉及Locker::tmp。代码“ while (tmp != (NumOfThreads-1)) {}”无需任何同步即可读取它，其他线程对其进行了修改。

---

这里是我在检查多线程代码时要寻找的一些东西：

• 关键部分是否定义明确？

• 每个共享变量与互斥量（或用作互斥量的信号量）之间是否有明确的关联，对每个共享变量的每次访问是否都受到相应互斥量的保护？

• 是否同时保持锁定状态的每对互斥锁总是以相同的相对顺序获取？

• 互斥锁是否总是以相反的顺序释放？

这些都不是硬性要求，但是，如果确保所有问题的答案都是“是”，则编写和维护正确的多线程代码要容易得多。所提供的代码不是这种情况，这是您（和我）无法分析代码的原因之一。

---

考虑：

• 关键部分定义是否正确？

没有您的代码中的某些区域由各种线程执行，并锁定了互斥锁的不同组合。特别是，当此if / else的执行完成时，互斥锁m1可能会锁定或可能不会锁定：

    if (!flag){
        pthread_mutex_unlock(&m1);
        sem_wait(&sem);
    } else {
        flag = 0; // m1 is still locked
    }

这也很难确保始终释放m1，这是关键部分定义良好的另一个方面。

---

考虑：

• 每个共享变量和互斥锁[...]之间是否有明确的关联？

不是。理想情况下，我希望在代码中看到此类关联的文档。最好在标题中，但是如果变量的名称明确传达了适当的关联，这对我来说就足够了。

与Locker::wait()行为一致的唯一关联集是：

• [m1：flag
• [m2：其他所有内容[count，tmp，NumOfThreads）

很难判断这是否是设计的意图。

---

此外，在审查任何类型的代码时，我考虑的标准之一就是它的复杂性和清晰度。这是所提供代码的最大问题。似乎有大约两倍于实际需要的运动部​​件，并且尚不清楚额外的钻头打算用于什么目的。为了您描述的目的，每个Locker需要以下数据：

• 释放前必须达到的线程数-> Locker::NumOfThreads
• 当前正在阻塞的线程数-> Locker::count
• （也许是一个标志，该标志确定释放后是否要重新武装（据我所知，Locker中未实现）

并且（仅需要）两个同步对象：

• 保护访问上述数据的互斥量或二进制信号量->Locker::m1和 Locker::m2
• 一个计数信号量，条件变量或类似的同步对象，可用来阻塞到达的线程，直到屏障打开-> Locker::sem

但是flag？也许某些这样的变量可用作local变量，然后该变量将不会在线程之间共享，但是我什至没有真正的需要。还有tmp？您似乎正在用它来进行某种辅助线程计数，但是这样做的目的对我来说完全消失了。为什么还要两个信号量和两个互斥量？

更糟糕的是，发生数据争夺的原因还包括所有这些额外的东西。当我第一次阅读代码时，这一行引起了我的注意：

        while(tmp != (NumOfThreads-1)){}

这似乎倾向于进入无限循环，但是您声称代码对您有用。一种可能是，当执行第一次到达该循环时，tmp可靠地等于NumOfThreads - 1，但是在这种情况下，它毫无用处。另一种可能性是tmp正在执行循环的线程的控制流之外进行修改，实际上就是这种情况，并且其中存在数据竞争。我们不需要任何上下文即可看到在循环迭代之间没有操作互斥量或信号量，因此只有两种可能性：

1. 执行循环的线程持有一个互斥量，该互斥量将防止其他线程修改tmp。这将产生循环的零次或无界迭代。

2. 执行循环的线程未持有互斥量，该互斥量会阻止其他线程修改tmp，因此，如果任何其他线程do对其进行修改，则数据竞争将由循环线程的未保护访问（任何尽管其他线程可以互斥使用）。

我建议将所有内容全部转储并简化。 Locker::wait()的主体不需要多于10到15的非空白，非注释行（直到您添加检查函数的返回码和响应错误所需的代码）。您只需要我已经描述的数据。进入Locker::wait()后，每个线程在访问Locker的成员之前都需要获取相同（一个）互斥量。然后，

• 如果它没有使线程计数达到完整，则线程释放互斥量并等待信号量。当它成功地从等待中返回时，它将继续前进，而不会与屏障进一步交互。

• 另一方面，如果它是打开屏障所需的最后一个线程，则它将发布到信号量足够的时间以释放其他线程（同时仍保持互斥锁）。如果屏障重新设置，则该线程必须重置等待线程的计数。然后，它释放互斥锁并继续运行，而无需等待信号量或与屏障进一步交互。

• 最后，如果障碍不是自动重新设置的，则线程可能会发现它已经打开。在这种情况下，线程将释放互斥锁并继续执行。