C++ P线程 : Algorithm to run N threads simultaneously when >N threads must be run each iteration

Question

我有一个程序需要每次迭代运行一个函数 M 次，并且这些运行可以并行化。假设我一次只能运行 N 个线程(比如可用内核数)。我需要一种算法来确保我始终运行 N 个线程(只要剩余线程数 >= N)并且该算法需要对这些线程的完成顺序保持不变。此外，线程调度算法不应占用大量 CPU 时间。

我有类似下面的想法，但它显然有缺陷。

#include <iostream>
#include <pthread.h>
#include <cstdlib>

void *find_num(void* arg)
{
    double num = rand();
    for(double q=0; 1; q++)
        if(num == q)
        {
            std::cout << "\n--";
            return 0;
        }
}


int main ()
{
    srand(0);

    const int N = 2;
    pthread_t threads [N];
    for(int q=0; q<N; q++)
        pthread_create(&threads [q], NULL, find_num, NULL);

    int M = 30;
    int launched=N;
    int finnished=0;
    while(1)
    {
        for(int w=0; w<N; w++)
        {
            //inefficient if `threads [1]` done before `threads [2]`
            pthread_join( threads [w], NULL);
            finnished++;
            std::cout << "\n" << finnished;
            if(finnished == M)
                break;
            if(launched < M)
            {
                pthread_create(&threads [w], NULL, find_num, NULL);
                launched++;
            }
        }

        if(finnished == M)
            break;
    }
}

这里明显的问题是，如果 threads[1] 在 threads[0] 之前完成，就会浪费 CPU 时间，我想不出如何获取围绕那个。另外，我假设让主例程等待 pthread_join() 不会显着消耗 CPU 时间？

Answer 1

我建议不要重新生成线程，这是相当严重的开销。相反，创建一个包含 N 个线程的池并通过工作队列向它们提交工作，这是一种相当标准的方法。即使您的剩余工作少于 N，额外的线程也不会造成任何伤害，它们只会在工作队列中阻塞。

如果您坚持当前的方法，您可以这样做:

不要等待带有 pthread_join 的线程，你不需要它，因为你没有将任何东西传回主线程。创建具有属性 PTHREAD_CREATE_DETACHED 的线程让他们退出。

在主线程中，等待信号量，每个退出的线程都会发出信号 - 实际上您会等待任何线程终止。如果你没有<semaphore.h>出于任何原因，使用互斥锁和条件来实现它都是微不足道的。

#include <semaphore.h>
#include <iostream>
#include <pthread.h>
#include <cstdlib>

sem_t exit_sem;

void *find_num(void* arg)
{
    double num = rand();
    for(double q=0; 1; q++)
        if(num == q)
        {
            std::cout << "\n--";
            return 0;
        }

    /* Tell the main thread we have exited.  */
    sem_post (&exit_sem);
    return NULL;
}

int main ()
{
    srand(0);

    /* Initialize pocess private semaphore with 0 initial count.  */
    sem_init (&exit_sem, 0, 0);
    const int N = 2;

    pthread_attr_t attr;
    pthread_attr_init (&attr);
    pthread_attr_setdetachstate (&attr, PTHREAD_CREATE_DETACHED);
    for(int q=0; q<N; q++)
        pthread_create(NULL, &attr, find_num, NULL);

    int M = 30;
    int launched=N;
    int finnished=0;
    while(1)
    {
        for(int w=0; w<N; w++)
        {
            /* Wait for any thread to exit, don't care which.  */
            sem_wait (&exit_sem);

            finnished++;
            std::cout << "\n" << finnished;
            if(finnished == M)
                break;
            if(launched < M)
            {
                pthread_create(NULL, &attr, find_num, NULL);
                launched++;
            }
        }

        if(finnished == M)
            break;
    }
}

无论如何，我会再次推荐线程池/工作队列方法。