Pthread_join of one of a number of threads

Question

My question is similar to How do I check if a thread is terminated when using pthread?. but i did not quite get an answer.

My problem is...I create a certain number of threads say n. As soon as main detects the exit of any one thread it creates another thread thus keeping the degree of concurrency as n and so on.

How does the main thread detect the exit of a thread. pthread_join waits for a particular thread to exit but in my case it can be any one of the n threads.

Thanks

Answer 1

Most obvious, without restructuring your code as aix suggests, is to have each thread set something to indicate that it has finished (probably a value in an array shared between all threads, one slot per worker thread), and then signal a condition variable. Main thread waits on the condition variable and each time it wakes up, handle all threads that have indicated themselves finished: there may be more than one.

Of course that means that if the thread is cancelled you never get signalled, so use a cancellation handler or don't cancel the thread.

Answer 2

There are several ways to solve this.

One natural way is to have a thread pool of fixed size n and have a queue into which the main thread would place tasks and from which the workers would pick up tasks and process them. This will maintain a constant degree of concurrency.

An alternative is to have a semaphore with the initial value set to n. Every time a worker thread is created, the value of the semaphore would need to be decremented. Whenever a worker is about to terminate, it would need to increment ("post") the semaphore. Now, waiting on the semaphore in the main thread will block until there's fewer than n workers left; a new worker thread would then be spawned and the wait resumed. Since you won't be using pthread_join on the workers, they should be detached (pthread_detach).

Answer 3

If you want to be informed of a thread exiting (via pthread_exit or cancellation), you can use a handler with pthread_cleanup_push to inform the main thread of the child exiting (via a condition variable, semaphore or similar) so it can either wait on it, or simply start a new one (assuming the child is detached first).

Alternately, I'd suggest having the threads wait for more work (as suggested by @aix), rather than ending.

Answer 4

If your parent thread needs to do other other things, then it can't just constantly be blocking on pthread_join, You will need a way to send a message to the main thread from the child thread to tell it to call pthread_join. There are a number of IPC mechanisms that you could use for this.

When a child thread has done it's work, it would then send some sort of message to the main thread through IPC saying "I completed my job" and also pass its own thread id, then the main thread knows to call pthread_join on that thread id.

Answer 5

One easy way is to use a pipe as a communication channel between the (worker) threads and your main thread. When a thread terminates it writes its result (thread id in the following example) to the pipe. The main thread waits on the pipe and reads the thread result from it as soon as it becomes available.

Unlike mutex or semaphore, a pipe file descriptor can be easily handled by the application main event loop (such as libevent). The writes from different threads to the same pipe are atomic as long as they write PIPE_BUF or less bytes (4096 on my Linux).

Below is a demo that creates ten threads each of which has a different life span. Then the main thread waits for any thread to terminate and prints its thread id. It terminates when all ten threads have completed.

$ cat test.cc
#include <iostream>
#include <pthread.h>
#include <unistd.h>
#include <stdlib.h>
#include <time.h>

void* thread_fun(void* arg) {
    // do something
    unsigned delay = rand() % 10;
    usleep(delay * 1000000);

    // notify termination
    int* thread_completed_fd = static_cast<int*>(arg);
    pthread_t thread_id = pthread_self();
    if(sizeof thread_id != write(*thread_completed_fd, &thread_id, sizeof thread_id))
        abort();

    return 0;
}

int main() {
    int fd[2];
    if(pipe(fd))
        abort();

    enum { THREADS = 10 };

    time_t start = time(NULL);

    // start threads
    for(int n = THREADS; n--;) {
        pthread_t thread_id;
        if(pthread_create(&thread_id, NULL, thread_fun, fd + 1))
            abort();
        std::cout << time(NULL) - start << " sec: started thread " << thread_id << '\n';
    }

    // wait for the threads to finish
    for(int n = THREADS; n--;) {
        pthread_t thread_id;
        if(sizeof thread_id != read(fd[0], &thread_id, sizeof thread_id))
            abort();
        if(pthread_join(thread_id, NULL)) // detached threads don't need this call
            abort();
        std::cout << time(NULL) - start << " sec: thread " << thread_id << " has completed\n";
    }

    close(fd[0]);
    close(fd[1]);
}

$ g++ -o test -pthread -Wall -Wextra -march=native test.cc
$ ./test
0 sec: started thread 140672287479552
0 sec: started thread 140672278759168
0 sec: started thread 140672270038784
0 sec: started thread 140672261318400
0 sec: started thread 140672252598016
0 sec: started thread 140672243877632
0 sec: started thread 140672235157248
0 sec: started thread 140672226436864
0 sec: started thread 140672217716480
0 sec: started thread 140672208996096
1 sec: thread 140672208996096 has completed
2 sec: thread 140672226436864 has completed
3 sec: thread 140672287479552 has completed
3 sec: thread 140672243877632 has completed
5 sec: thread 140672252598016 has completed
5 sec: thread 140672261318400 has completed
6 sec: thread 140672278759168 has completed
6 sec: thread 140672235157248 has completed
7 sec: thread 140672270038784 has completed
9 sec: thread 140672217716480 has completed