不阻塞 main() 的 pthread_join() 版本：POSIX

Question

我正在尝试编写一个在调用pthread_join() 时不会阻塞main() 的代码： 即基本上试图实现下面提到的我之前的问题：

https://stackoverflow.com/questions/24509500/pthread-join-and-main-blocking-multithreading

以及对应的解释在：

pthreads - Join on group of threads, wait for one to exit

根据建议的答案：

You'd need to create your own version of it - e.g. an array of flags (one flag per thread) protected by a mutex and a condition variable; where just before "pthread_exit()" each thread acquires the mutex, sets its flag, then does "pthread_cond_signal()". The main thread waits for the signal, then checks the array of flags to determine which thread/s to join (there may be more than one thread to join by then).

我试过如下：

我的状态数组，用于跟踪哪些线程已完成：

typedef struct {
    int Finish_Status[THREAD_NUM];
    int signalled;

    pthread_mutex_t mutex;
    pthread_cond_t FINISHED;
    }THREAD_FINISH_STATE;

线程例程，它在线程完成时设置相应的数组元素，并发出条件变量的信号：

void* THREAD_ROUTINE(void* arg)
{
    THREAD_ARGUMENT* temp=(THREAD_ARGUMENT*) arg;
    printf("Thread created with id %d\n",temp->id);
    waitFor(5);
    pthread_mutex_lock(&(ThreadFinishStatus.mutex));
    ThreadFinishStatus.Finish_Status[temp->id]=TRUE;
    ThreadFinishStatus.signalled=TRUE;
    if(ThreadFinishStatus.signalled==TRUE)
    {
      pthread_cond_signal(&(ThreadFinishStatus.FINISHED));
      printf("Signal that thread %d finished\n",temp->id);
     }
    pthread_mutex_unlock(&(ThreadFinishStatus.mutex));

    pthread_exit((void*)(temp->id));
    }

我无法编写 pthread_join() 和 pthread_cond_wait() 函数的相应部分。有些事情我无法实现。

1) 我的main()中对应的部分pthread_cond_wait()怎么写？

2) 我想把它写成：

   pthread_mutex_lock(&(ThreadFinishStatus.mutex));
    while((ThreadFinishStatus.signalled != TRUE){
     pthread_cond_wait(&(ThreadFinishStatus.FINISHED), &(ThreadFinishStatus.mutex));
     printf("Main Thread signalled\n");
     ThreadFinishStatus.signalled==FALSE; //Reset signalled
     //check which thread to join
    }
    pthread_mutex_unlock(&(ThreadFinishStatus.mutex)); 

但是没有进入while循环。

3) 如何使用pthread_join() 以便我可以获取存储在我的arg[i].returnStatus 中的返回值 即在我的 main 中放置以下语句的位置：

`pthread_join(T[i],&(arg[i].returnStatus));`

完整代码

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

#define THREAD_NUM 5
#define FALSE 0
#define TRUE 1


void waitFor (unsigned int secs) {
    time_t retTime;
    retTime = time(0) + secs;     // Get finishing time.
    while (time(0) < retTime);    // Loop until it arrives.
}

typedef struct {
    int Finish_Status[THREAD_NUM];
    int signalled;

    pthread_mutex_t mutex;
    pthread_cond_t FINISHED;
    }THREAD_FINISH_STATE;

typedef struct {
    int id;
    void* returnStatus;
    }THREAD_ARGUMENT;

THREAD_FINISH_STATE ThreadFinishStatus;

void initializeState(THREAD_FINISH_STATE* state)
{
 int i=0;
 state->signalled=FALSE;
 for(i=0;i<THREAD_NUM;i++)
 {
     state->Finish_Status[i]=FALSE;
     }
 pthread_mutex_init(&(state->mutex),NULL);
 pthread_cond_init(&(state->FINISHED),NULL);
    }

void destroyState(THREAD_FINISH_STATE* state)
{
 int i=0;
 for(i=0;i<THREAD_NUM;i++)
 {
     state->Finish_Status[i]=FALSE;
     }
 pthread_mutex_destroy(&(state->mutex));
 pthread_cond_destroy(&(state->FINISHED));
    }


void* THREAD_ROUTINE(void* arg)
{
    THREAD_ARGUMENT* temp=(THREAD_ARGUMENT*) arg;
    printf("Thread created with id %d\n",temp->id);
    waitFor(5);
    pthread_mutex_lock(&(ThreadFinishStatus.mutex));
    ThreadFinishStatus.Finish_Status[temp->id]=TRUE;
    ThreadFinishStatus.signalled=TRUE;
    if(ThreadFinishStatus.signalled==TRUE)
    {
      pthread_cond_signal(&(ThreadFinishStatus.FINISHED));
      printf("Signal that thread %d finished\n",temp->id);
     }
    pthread_mutex_unlock(&(ThreadFinishStatus.mutex));

    pthread_exit((void*)(temp->id));
    }

int main()
{
    THREAD_ARGUMENT arg[THREAD_NUM];
    pthread_t T[THREAD_NUM];
    int i=0;
    initializeState(&ThreadFinishStatus);

    for(i=0;i<THREAD_NUM;i++)
    {
        arg[i].id=i;
        }

    for(i=0;i<THREAD_NUM;i++)
        {
            pthread_create(&T[i],NULL,THREAD_ROUTINE,(void*)&arg[i]);
        }

    /*
     Join only if signal received
    */

    pthread_mutex_lock(&(ThreadFinishStatus.mutex));
    //Wait
    while((ThreadFinishStatus.signalled != TRUE){
     pthread_cond_wait(&(ThreadFinishStatus.FINISHED), &(ThreadFinishStatus.mutex));
     printf("Main Thread signalled\n");
     ThreadFinishStatus.signalled==FALSE; //Reset signalled
     //check which thread to join
    }
    pthread_mutex_unlock(&(ThreadFinishStatus.mutex));

    destroyState(&ThreadFinishStatus);
    return 0;
    }

Answer 1

这是一个程序示例，该程序使用计数信号量来观察线程是否完成，找出它是哪个线程，并查看来自该线程的一些结果数据。这个程序对锁很有效 - 服务员不会被虚假唤醒（注意线程如何在释放互斥锁保护共享状态之后才发布到信号量）。

这种设计允许主程序在某个线程完成后立即处理来自某个线程的计算结果，并且不需要主程序等待所有线程完成。如果每个线程的运行时间变化很大，这将特别有用。

最重要的是，这个程序不会死锁也不会竞争。

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

void* ThreadEntry(void* args );

typedef struct {
    int threadId;
    pthread_t thread;
    int threadResult;
} ThreadState;

sem_t           completionSema;

pthread_mutex_t resultMutex;

std::queue<int> threadCompletions;

ThreadState* threadInfos;

int main() {

    int numThreads = 10;
    int* threadResults;
    void* threadResult;
    int doneThreadId;


    sem_init( &completionSema, 0, 0 );

    pthread_mutex_init( &resultMutex, 0 );

    threadInfos = new ThreadState[numThreads];

    for ( int i = 0; i < numThreads; i++ ) {

        threadInfos[i].threadId = i;
        pthread_create( &threadInfos[i].thread, NULL, &ThreadEntry, &threadInfos[i].threadId  );
    }

    for ( int i = 0; i < numThreads; i++ ) {
        // Wait for any one thread to complete; ie, wait for someone
        // to queue to the threadCompletions queue.
        sem_wait( &completionSema );


        // Find out what was queued; queue is accessed from multiple threads,
        // so protect with a vanilla mutex.
        pthread_mutex_lock(&resultMutex);
        doneThreadId = threadCompletions.front();
        threadCompletions.pop();
        pthread_mutex_unlock(&resultMutex);

        // Announce which thread ID we saw finish
        printf(
            "Main saw TID %d finish\n\tThe thread's result was %d\n",
            doneThreadId,
            threadInfos[doneThreadId].threadResult
        );

        // pthread_join to clean up the thread.
        pthread_join( threadInfos[doneThreadId].thread, &threadResult );
    }

    delete threadInfos;

    pthread_mutex_destroy( &resultMutex );
    sem_destroy( &completionSema );

}

void* ThreadEntry(void* args ) {
    int threadId = *((int*)args);

    printf("hello from thread %d\n", threadId );

    // This can safely be accessed since each thread has its own space
    // and array derefs are thread safe.
    threadInfos[threadId].threadResult = rand() % 1000;


    pthread_mutex_lock( &resultMutex );
    threadCompletions.push( threadId );
    pthread_mutex_unlock( &resultMutex );

    sem_post( &completionSema );

    return 0;
}

Answer 2

Pthread 条件没有“内存”；如果在 pthread_cond_wait 之前调用 pthread_cond_signal，则 pthread_cond_wait 不会返回，这就是为什么在调用 pthread_cond_wait 之前检查谓词很重要，如果它是真的就不要调用它。但这意味着操作，在这种情况下，“检查要加入的线程”应该只取决于谓词，而不是取决于是否调用了 pthread_cond_wait。

此外，您可能希望让 while 循环真正等待所有线程终止，而您现在没有这样做。

（另外，我认为关于“signalled==FALSE”无害的另一个答案是错误的，它不是无害的，因为有一个 pthread_cond_wait，当它返回时，signaled 会变为 true。）

所以如果我想写一个程序，以这种方式等待所有线程终止，它看起来更像

pthread_mutex_lock(&(ThreadFinishStatus.mutex));
// AllThreadsFinished would check that all of Finish_Status[] is true
// or something, or simpler, count the number of joins completed
while (!AllThreadsFinished()) {
  // Wait, keeping in mind that the condition might already have been
  // signalled, in which case it's too late to call pthread_cond_wait,
  // but also keeping in mind that pthread_cond_wait can return spuriously,
  // thus using a while loop
  while (!ThreadFinishStatus.signalled) {
    pthread_cond_wait(&(ThreadFinishStatus.FINISHED), &(ThreadFinishStatus.mutex));
  }
  printf("Main Thread signalled\n");
  ThreadFinishStatus.signalled=FALSE; //Reset signalled
  //check which thread to join
}
pthread_mutex_unlock(&(ThreadFinishStatus.mutex));

Answer 3

你的代码很活泼。

假设您启动一个线程并在您获取main() 中的互斥锁之前完成。您的while 循环将永远不会运行，因为退出线程已经将signalled 设置为TRUE。

我会回应@antiduh 的建议，即使用一个信号量来计算死但未加入的线程的数量。然后，您循环到在信号量上等待产生的线程数。我要指出，POSIX sem_t 不像 pthread_mutex，因为 sem_wait 可以返回 EINTR。

Answer 4

您的代码看起来不错。你有一个小号：

 ThreadFinishStatus.signalled==FALSE; //Reset signalled

这无济于事。它测试 signaled 是否为 FALSE 并丢弃结果。这是无害的，因为你不需要做任何事情。 （您从不想将signalled 设置为FALSE，因为这会丢失它已发出信号的事实。永远没有任何理由取消它-如果线程完成，那么它就永远完成了。 )

不进入while 循环意味着signalled 为TRUE。这意味着线程已经设置了它，在这种情况下不需要进入循环，因为没有什么可以等待的。所以没关系。

还有：

ThreadFinishStatus.signalled=TRUE;
if(ThreadFinishStatus.signalled==TRUE)

没有必要测试你刚刚设置的东西。不是说套装会失败。

FWIW，我建议重新架构。如果像pthread_join 这样的现有功能不能完全满足您的要求，请不要使用它们。如果您要拥有跟踪已完成工作的结构，则将其与线程终止完全分开。既然您已经知道完成了哪些工作，那么线程终止的时间和方式有什么不同？不要认为这是“我需要一种特殊的方法来知道线程何时终止”，而应该认为这是“我需要知道完成了哪些工作，以便我可以做其他事情”。