linux - 使用 pthreads 在多个内核上并行处理数据

Question

我的目标是使用多个工作线程在多个核上处理数据，然后在主线程中进一步处理结果。我在 linux 上工作，我想使用 pthreads。我创建了一个简单的示例来学习如何正确执行此操作。我有一个名为“回调”的主线程和 4 个工作线程。这个想法是主线程向工作线程发出信号开始处理，然后 4 个线程在完成时向主线程发出信号，主线程在所有 4 个线程通知它们完成后退出。我希望 4 个工作线程能够并行运行，所以我不希望这些线程中的任何一个等待其他线程。在我的示例中，我试图让每个线程休眠不同的持续时间(1、2、3 和 4 秒)，并认为代码将在 4 秒后退出(即当工作线程 4 完成等待时4 秒)。

由于某种原因，我的代码不正确，它总是立即退出，打印出:

thread 3 start (sleeping 3000 ms)
thread 2 start (sleeping 2000 ms)
thread 1 start (sleeping 1000 ms)
thread 4 start (sleeping 4000 ms)
    thread 1 stop
    thread 2 stop
    thread 3 stop
    thread 4 stop
Main(): Waited on 5  threads. Done.

所以线程似乎确实以正确的顺序退出，但程序运行不需要 4 秒。

这是怎么回事？我已经在下面粘贴了代码

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

// based on https://computing.llnl.gov/tutorials/pthreads/#ConditionVariables

#define DUR 1000
#define NUM_THREADS 5
int state = 0; 

pthread_mutex_t mutex;
pthread_cond_t conddone;
pthread_cond_t condwork;

void* callback(void* t) { 

    // signal worker threads to start work 

        pthread_mutex_lock(&mutex);
        pthread_cond_broadcast(&condwork);
        pthread_mutex_unlock(&mutex);

    // wait for worker threads to finish 

        pthread_mutex_lock(&mutex);
        while (state < 4)
            pthread_cond_wait(&conddone, &mutex);
        pthread_mutex_unlock(&mutex);

   pthread_exit(NULL);

} 

void* worker(void* t) { 

   long id = (long)t;

    // wait for signal from callback to start doing work 

    pthread_mutex_lock(&mutex);
    pthread_cond_wait(&condwork, &mutex);
    pthread_mutex_unlock(&mutex);

    // do work 

    printf("thread %d start (sleeping %d ms)\n", id, id * DUR); 
    usleep(id * DUR); 
    printf("    thread %d stop\n", id); 

    // tell callback we're done 

    pthread_mutex_lock(&mutex);
    state++;
    pthread_cond_signal(&conddone);
    pthread_mutex_unlock(&mutex);

   pthread_exit(NULL);

}


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

   pthread_t threads[5];
   pthread_attr_t attr;

   pthread_mutex_init(&mutex, NULL);
   pthread_cond_init (&condwork, NULL);
   pthread_cond_init (&conddone, NULL);

   pthread_attr_init(&attr);
   pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_JOINABLE);

   pthread_create(&threads[0], &attr, callback, (void *)0);
   pthread_create(&threads[1], &attr, worker, (void *)1);
   pthread_create(&threads[2], &attr, worker, (void *)2);
   pthread_create(&threads[3], &attr, worker, (void *)3);
   pthread_create(&threads[4], &attr, worker, (void *)4);

   for (i=0; i<NUM_THREADS; i++) {
     pthread_join(threads[i], NULL);
   }
   printf ("Main(): Waited on %d  threads. Done.\n", NUM_THREADS);

   pthread_attr_destroy(&attr);
   pthread_mutex_destroy(&mutex);
   pthread_cond_destroy(&condwork);
   pthread_cond_destroy(&conddone);
   pthread_exit(NULL);
} 

Answer 1

你眼前的问题就是usleep()休眠微秒而不是毫秒，因此您的线程休眠的时间是您希望它们休眠的时间的千分之一。

不过，您确实还有另一个问题:您的 condwork条件变量未与共享状态的谓词配对(例如 state < 4 变量的谓词 conddone)。如果您的工作线程之一执行 pthread_cond_wait()在“回调”线程执行完 pthread_cond_broadcast() 之后， worker 将无限期地等待。

您可以通过初始化 state 来解决这个问题变量为 -1 :

int state = -1;

并让您的工作人员等待谓词 state < 0 :

// wait for signal from callback to start doing work 

pthread_mutex_lock(&mutex);
while (state < 0)
    pthread_cond_wait(&condwork, &mutex);
pthread_mutex_unlock(&mutex);

并通过将状态设置为 0 让“回调”向工作人员发出信号:

// signal worker threads to start work 

pthread_mutex_lock(&mutex);
state = 0;
pthread_cond_broadcast(&condwork);
pthread_mutex_unlock(&mutex);