C:一次维护N个并发pthread，进行M>>N个独立计算

Question

我遇到一个问题，要求我多次执行特定的可变长度计算(通常> 10^8)，并且我有少量处理器(<=16)来运行它。下面的简化代码一次成功地以 NTHREADS 个批处理创建 pthread，但它的缺陷是其他所有内容都会暂停，直到每个批处理中最慢的线程完成为止。由于最慢的线程有时会比最快的线程慢 10-100 倍，这意味着平均而言，处理器可能会在大部分时间处于空闲状态。

我想做的是通过在每次终止时创建一个新的 pthread 来让所有处理器保持忙碌。如果有一种方法可以检索当前事件 pthread 的数量，我可以轻松地做到这一点，但我还没有找到一种方法来做到这一点。

这可能吗？如果是这样，怎么办？

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

struct arg_struct {
  double x ;
  double y ;
};

int nloops = 0 ;  // initialize loop counter

void process(struct arg_struct  *args)
{
  int thisloop ;

  float x,y ;
  x = args->x ; y = args->y ;
  free(args) ;  // we're done with passed arguments

  nloops++ ;  // increment global counter 
  thisloop = nloops ;  // capture current loop  number

  sleep(11-nloops) ;  // variable delay
  printf("thisloop = %d   threadID = %d args = %.1f %.1f\n", thisloop,  (int) pthread_self(), x, y) ;

  pthread_exit(NULL); // exit thread
}

int main()
{
  const int MINLOOPS = 10 ;  // total number of loops to execute
  const int MAXTHREADS = 4 ;  // maximum number of threads at any one time

  int N, remaining ;
  pthread_t tid[MAXTHREADS];

  while (1)
    {
      remaining = MINLOOPS - nloops ;
      if (remaining == 0) break ;
      if (remaining < MAXTHREADS)
        N = remaining;
      else
        N = MAXTHREADS;

      for (int i = 0; i < N; i++) {  // create a set of simultaneous threads

        struct arg_struct *args = malloc(sizeof(struct arg_struct));  // initialize arguments
        args->x = i;  args->y = -i ;

        pthread_create(&tid[i], NULL, (void *) process,  (void *) args ) ;  
        printf("Created thread %d\n", (int) tid[i]) ;
      }

      for (int i = 0; i < N; i++) // wait until all threads in current loop have completed 
        pthread_join(tid[i], NULL);
    }
}

输出是:

Created thread 216977408
Created thread 217513984
Created thread 218050560
Created thread 218587136
thisloop = 4   threadID = 218587136 args = 3.0 -3.0
thisloop = 3   threadID = 218050560 args = 2.0 -2.0
thisloop = 2   threadID = 217513984 args = 1.0 -1.0
thisloop = 1   threadID = 216977408 args = 0.0 0.0
Created thread 216977408
Created thread 217513984
Created thread 218050560
Created thread 218587136
thisloop = 8   threadID = 218050560 args = 2.0 -2.0
thisloop = 7   threadID = 218587136 args = 3.0 -3.0
thisloop = 6   threadID = 217513984 args = 1.0 -1.0
thisloop = 5   threadID = 216977408 args = 0.0 0.0
Created thread 216977408
Created thread 217513984
thisloop = 10   threadID = 217513984 args = 1.0 -1.0
thisloop = 9   threadID = 216977408 args = 0.0 0.0

Answer 1

在发布我的问题后，我找到了一个似乎可以接受的解决方案，如果可能是天真的解决方案(在我知道最好的方法是线程池之前；请参阅上面帕迪的评论)。它基本上需要一些簿记，包括将附加变量传递给线程进程。这是我想到的:

struct arg_struct {
  double x ;
  double y ;
  int ithread ;
  int loopno ;
};

const int MINLOOPS = 10 ;  // total number of loops to execute
const int MAXTHREADS = 4 ;  // maximum number of threads at any one time
pthread_t tid[MAXTHREADS] ; // table of active threads

int loopno = 0 ;  // initialize loop counter
int nthreads = 0 ; // current number of active threads

void process(struct arg_struct  *args)
{
  int loopno,ithread ;
  float x,y ;
  x = args->x ; y = args->y ; ithread = args->ithread ; loopno = args->loopno ;
  free(args) ;  // we're done with passed arguments

  sleep(MINLOOPS-loopno+1) ;  // variable delay
  printf("thisloop = %d   threadID = %d args = %.1f %.1f ithread = %d\n", loopno,  (int) pthread_self(), x, y, ithread) ;

  nthreads-- ;   // done with current thread
  tid[ithread] = 0 ;
  pthread_exit(NULL); // exit thread
}

int main()
{
  int ithread ;

  for (ithread=0; ithread<MAXTHREADS; ithread++) tid[ithread] = 0 ;  // initialize thread table

  while (loopno < MINLOOPS)
    {
      if (nthreads < MAXTHREADS) {  // check whether new thread needed
        for (int ith=0; ith<MAXTHREADS; ith++)  // find empty table entry
          {
            if (tid[ith] == 0) {
              ithread = ith ;
              break ;
            }
          }

        struct arg_struct *args = malloc(sizeof(struct arg_struct));  // initialize arguments

        loopno++ ;
        args->x = loopno;  args->y = -loopno ; args->ithread = ithread ; args->loopno = loopno ;
        pthread_create(&tid[ithread], NULL, (void *) process,  (void *) args ) ;  
        nthreads++ ;
        printf("Created thread %d\n", (int) tid[ithread]) ;
      }
    }

  for (int i = 0; i < MAXTHREADS; i++) // wait until remaining threads have completed 
    pthread_join(tid[i], NULL) ;

}

输出为:

Created thread 82550784
Created thread 83087360
Created thread 83623936
Created thread 84160512
thisloop = 4   threadID = 84160512 args = 4.0 -4.0 ithread = 3
Created thread 84697088
thisloop = 3   threadID = 83623936 args = 3.0 -3.0 ithread = 2
Created thread 85233664
thisloop = 2   threadID = 83087360 args = 2.0 -2.0 ithread = 1
Created thread 85770240
thisloop = 1   threadID = 82550784 args = 1.0 -1.0 ithread = 0
Created thread 86306816
thisloop = 7   threadID = 85770240 args = 7.0 -7.0 ithread = 1
Created thread 86843392
thisloop = 6   threadID = 85233664 args = 6.0 -6.0 ithread = 2
Created thread 87379968
thisloop = 8   threadID = 86306816 args = 8.0 -8.0 ithread = 0
thisloop = 5   threadID = 84697088 args = 5.0 -5.0 ithread = 3
thisloop = 10   threadID = 87379968 args = 10.0 -10.0 ithread = 2
thisloop = 9   threadID = 86843392 args = 9.0 -9.0 ithread = 1