Pthread_join одного из нескольких потоков

Question

Мой вопрос похож на Как проверить, завершен ли поток при использовании pthread? . но я не совсем получил ответ.

Моя проблема ... Я создаю определенное количество потоков, скажем n. Как только main обнаруживает выход любого потока, он создает другой поток, сохраняя степень параллелизма как n и т. Д.

Как основной поток обнаруживает выход потока. pthread_join ожидает завершения определенного потока, но в моем случае это может быть любой из n потоков.

Спасибо

Answer 1

Наиболее очевидным, без реструктуризации вашего кода, как предлагает aix, является то, чтобы каждый поток устанавливал что-то, чтобы указать, что он завершил (вероятно, значение в массиве, совместно используемом всеми потоками, один слот на рабочий поток), а затем сигнализирует условная переменная. Основной поток ожидает переменную условия и каждый раз, когда он просыпается, обрабатывает все потоки, которые указали, что они завершены: их может быть больше одного.

Конечно, это означает, что если поток отменен, вы никогда не получите сигнал, поэтому используйте обработчик отмены или не отменяйте поток.

Answer 2

Есть несколько способов решить эту проблему.

Один естественный способ - это иметь пул потоков фиксированного размера n и иметь очередь, в которую основной поток будет помещать задачи и из которых рабочие будутподобрать задачи и обработать их.Это будет поддерживать постоянную степень параллелизма.

Альтернативой является семафор с начальным значением, установленным на n.Каждый раз, когда создается рабочий поток, значение семафора должно уменьшаться.Всякий раз, когда работник собирается завершить работу, он должен увеличивать («публиковать») семафор.Теперь ожидание семафора в основном потоке будет блокироваться до тех пор, пока не останется менее n рабочих;тогда будет создан новый рабочий поток, и ожидание возобновится.Поскольку вы не будете использовать pthread_join на рабочих, они должны быть отключены (pthread_detach).

Answer 3

Одним простым способом является использование канала в качестве канала связи между (рабочими) потоками и вашим основным потоком. Когда поток завершается, он записывает свой результат (идентификатор потока в следующем примере) в канал. Основной поток ожидает в канале и считывает результат потока из него, как только он становится доступным.

В отличие от мьютекса или семафора, дескриптор файла канала может быть легко обработан главным циклом событий приложения (например, libevent). Записи из разных потоков в один и тот же канал являются атомарными, если они записывают PIPE_BUF или менее байтов (4096 в моем Linux).

Ниже приведена демонстрация, которая создает десять потоков, каждый из которых имеет различную продолжительность жизни. Затем основной поток ожидает завершения какого-либо потока и печатает его идентификатор. Завершается, когда завершены все десять потоков.

$ 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

Answer 4

Если ваш родительский поток должен выполнять другие действия, он не может просто постоянно блокироваться на pthread_join, вам понадобится способ отправить сообщение в основной поток из дочернего потока, чтобы сказать ему, чтобы он вызывалpthread_join.Существует несколько механизмов IPC, которые вы могли бы использовать для этого.

Когда дочерний поток выполнил свою работу, он затем отправил бы какое-то сообщение в основной поток через IPC, говорящее «Я выполнил свою работу»а также передать свой собственный идентификатор потока, тогда основной поток знает, чтобы вызвать pthread_join для этого потока идентификатор.

Answer 5

Если вы хотите получать информацию о выходе из потока (через pthread_exit или отмену), вы можете использовать обработчик с pthread_cleanup_push, чтобы сообщить основному потоку о выходе из дочернего процесса (через условие переменная, семафор или аналогичный), поэтому он может либо подождать его, либо просто начать новый (при условии, что дочерний элемент отсоединен первым).

С другой стороны, я бы предложил, чтобы потоки ожидали больше работы (как предложено @aix), а не заканчивали.