Pthreads，尝试使用信号量从多个线程中打印出数字部分答案

【问题标题】：Pthreads, trying to use semaphores to print out sections of numbers from multiple threadsPthreads，尝试使用信号量从多个线程中打印出数字部分
【发布时间】：2013-02-25 23:33:08
【问题描述】：

我遇到的问题是：我的“父级”经历了一个从 0 到 4 的 for 循环。在每次迭代中，我有 3 个线程要打印出它们的 12 个数字的“段” （0 到 11）。

例如，当父循环为0时，线程0会打印出0，线程1会打印出4，线程2会打印出8。

当父循环为1时，线程0会打印出1，线程1会打印出5，线程2会打印出9。

理想情况下，输出为 0 1 2 3 4 5 6 7 8 9 10 11 。我知道您无法确定线程何时运行，因此不会完全按照该顺序运行，但我至少希望在线程打印 3 4 或 5 之前打印 0 1 和 2。

这就是我遇到的问题。似乎一个线程或另一个线程被留在尘土中并且没有打印出它的片段，或者没有完全打印出它的片段以及其他线程。这是我试图完全理解信号量的问题。

我有两个信号量，一个在线程工作时阻塞父级（生产者？），另一个阻塞每个线程，直到生产者增加到下一个索引。我认为通过这种方式我可以强迫他们在继续之前等待对方完成，但由于某种原因我遇到了问题。

这是我的代码：

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

#define maxNum 12

sem_t parentSem;
sem_t threadSem;
int finishedThreads;
int done = 0;
int stuff[12];
int i;

void* threadFunction(int m){
    int printNum;
    int baseNum;
    //Determine the value the thread should start at
    baseNum = (double)(maxNum/3) * m;
    while(!done){ //ensure thread doesn't exit before parent is done with whole loop
        //wait for parent to increment
        sem_wait(&threadSem);
        printNum = baseNum + i;

            //keep track of how many threads are finished to let parent continue
        finishedThreads++;
        if(finishedThreads == 3){
                    //let parent continue if all threads are finished
            sem_post(&parentSem);
        }
    }

}

int main(int argc, char** argv[]){

    sem_init(&parentSem, 0, 1);
    sem_init(&threadSem, 0, 0);
    int rc;
    pthread_t threads[3];
    int l; 
    for(l = 0; l < 12; l++){
        stuff[l] = l;
    }
    int j;
    for(j = 0; j < 3; j++){
        rc = pthread_create(&threads[i], NULL, threadFunction, (void*) j);

    }
    int k;
    for(i = 0; i < 4; i++){
        sem_wait(&parentSem); //wait for children here (initially sem set to 1)
        finishedThreads = 0; //set finished thread counter to 0
        for(k = 0; k < 3; k++){
            //increment thread semaphore to 3 so each thread can run
            sem_post(&threadSem);
        }
    }
    for(i = 1; i < 3; i++){
        pthread_join(threads[i], NULL);
    }



}

如何确保所有线程在父级递增之前运行？我如何确保所有线程每“一轮”运行而没有任何卡在后面？有时同一个线程运行两次而不是每个线程运行一次.....帮助？

非常感谢大家的帮助。

编辑：新代码状态：（线程函数）

while(!done){

    printf("Thread %d not done yet...\n", m);
    if(m == 0){
        sem_wait(&threadSem0);
    }else if(m == 1){
        sem_wait(&threadSem1);
    }else if(m == 2){
        sem_wait(&threadSem2);
    }
    printNum = baseNum + i;
    printf("Thread %d past waiting, number segment: %d\n", m, printNum);

    finishedThreads++;
    if(finishedThreads == 3){
        sem_post(&parentSem);
    }
}

父部分：

for(i = 0; i < 4; i++){
    printf("In parent for loop, counter: %d\n", i);
    printf("Parent past wait semaphore\n");
    finishedThreads = 0;
    if(i == 3) done = 1;
    sem_post(&threadSem0);
    sem_post(&threadSem1);
    sem_post(&threadSem2);
    sem_wait(&parentSem);
}
for(i = 1; i < 3; i++){
    pthread_join(threads[i], NULL);
}

【问题讨论】：

您可能应该使用条件变量，并且您必须有一个关于访问finishedThreads 的关键部分。没有临界区，代码就有缺陷（有时会丢失计数）。
我查看了条件变量。我不知道如何正确使用它们。你是说我应该在我增加finishedThreads的地方放一个互斥锁？

标签： c multithreading pthreads semaphore

【解决方案1】：

您需要访问finishedThreads atomic 以避免每个线程仅使用缓存副本。您的程序在没有优化的情况下编译是否可以工作（*）？

在 C++ 中执行此操作的正确方法是使用 std::atomic。请参阅this question 的详尽回答。

更新：

如果线程快速循环（如您的情况），则对所有子线程使用一个信号量可能会产生问题。您必须为每个子线程使用一个信号量来解决此问题。

sem_t threadSem[3];

// ...

sem_wait(&threadSem+m);

// ...

for (i = 0; i < 3; i++)
    sem_init(threadSem+i, 0, 0);

// ...

    for(k = 0; k < 3; k++){
        //increment each thread semaphore so each thread can run
        sem_post(threadSem+k);
    }

另外几件我早期没有发现的东西：

您使用变量i 在threadFunction 中进行增量。这行不通。正确的公式应该只取决于m。由于每个线程都有自己的baseNum 和printNum 副本，因此您可以毫无问题地使用它们：

baseNum = (double)(maxNum/3) * m;
while(!done){ //ensure thread doesn't exit before parent is done with whole loop
    //wait for parent to increment
    sem_wait(&threadSem);
    printNum++;
    // ....

您的代码中有很多东西取决于子线程的数量，我建议为此使用一个常量（就像您为 maxNum 所做的那样）。

您没有设置变量done 来指示子线程的工作结束。

volatile int done;

// in threadFunction

 // wait for start
sem_wait(threadSem+m)
while(!done){ 
    // thread work here
    // ...
    //wait for parent to increment at the end of the loop
    sem_wait(threadSem+m);
}

// in main

for(k = 0; k < 3; k++){
    //increment each thread semaphore so each thread can run
    if (i == 4) done = 1;
    sem_post(threadSem+k);
}

当然done 也必须遵循“volatile 或互斥”规则。

你不会在最后加入所有线程：

for(i = 0; i < 3; i++){
    pthread_join(threads[i], NULL);
}

(*) 如果您使用 gcc，我的意思是使用参数 -O0，甚至使用 -g 进行调试。

【讨论】：

呃，我不完全确定您所说的“没有优化”是什么意思。如果您的意思是“没有信号量”，那么它实际上不起作用，因为 FOR 循环只是通过它计数，并且线程无需等待即可完成，只打印出一个数字。
使用上面的代码，我目前得到的是这样的： Parent FOR is 0 -- Thread 1: 4 Thread 2: 8 Thread 0: 没有运行 Parent FOR is 1 -- Thread 0：1 线程 1：5 线程 2：9 父级 FOR 为 2 -- 线程 0：2 线程 2：10 线程 1：6 父级 FOR 为 3 -- 线程 0：3 线程 1：未运行线程 2：未运行'不运行分段错误
我真的不知道 -O0 是做什么的，但是当我将完成线程作为易失性时，它会编译并给出正确的输出（仍然存在 seg 错误）。即使我不使用 -O0 作为参数也是如此。但是，每隔一段时间就会有同一个线程在一个“回合”中运行多次。它比以前好多了，但仍然不能正常工作。
现在，将其指定为 Volatile，输出如下所示：4 8 0 5 9 1 6 10 2 7 11 3 这是正确的。但是再运行一两次后，它会给出： 8 8 0 9 9 5 10 6 2 11 7 3 这是一个线程多次运行。
一秒钟，尝试实施您的建议

【解决方案2】：

您需要一个屏障，特别是一个可重复使用的屏障。看看 Allen Downey 的信号量小书，第 3.6.7 节，转载here (PDF) 可能会有用。代码是用 Python 编写的，但它的要点很清楚。它需要一个互斥体和两个计数信号量，在您的情况下，障碍中有 4 个参与者。

您可以通过设置线程优先级来控制哪个线程首先被唤醒，以便第一个线程是最重要的，依此类推。

【讨论】：