Wsrtjtyk

I have a problem that requires me to execute a particular calculation of variable length a large number of times (typically > 10^8) and I have a small number of processors (<=16) to run it on. The simplified code below successfully creates pthreads in batches of NTHREADS at a time, but it has the deficiency that everything else pauses until the slowest thread in each batch completes. Since the slowest thread can occasionally be 10-100 times slower than the fastest thread, this means processors may be idle for a large fraction of the time, on average.

What I would like to do is keep all processors busy by creating a new pthread each time one terminates. I could easily do this if there were a way to retrieve the current number of active pthreads, but I haven't found a way to do that.

Is this possible? If so, how?

#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 %.1fn", 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 %dn", (int) tid[i]) ;

      }



      for (int i = 0; i < N; i++) // wait until all threads in current loop have completed 

        pthread_join(tid[i], NULL);

    }

}

Output is:

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

After posting my question, I worked out what seems to be an acceptable, if possibly naive, solution (before I knew that the best approach is a thread pool; see paddy's comment above). It basically requires some bookkeeping, including the passing of additional variables to the threaded process. Here's what I came up with:

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 = %dn", 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 %dn", (int) tid[ithread]) ;

      }

    }



  for (int i = 0; i < MAXTHREADS; i++) // wait until remaining threads have completed 

    pthread_join(tid[i], NULL) ;



}

The output is then:

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