parallel.c

   1 /*
   2  * Parallel processing
   3  */
   4
   5 #include "common.h"
   6
   7 #include <pthread.h>
   8
   9 #include "mgraph.h"
  10 #include "parallel.h"
  11
  12 typedef struct {
  13   const struct Vertices *vertices;
  14   Computation *separately;
  15   void *gendata;
  16 } ForAllThreads;
  17
  18 typedef struct {
  19   ForAllThreads *allthreads;
  20   int section;
  21   void *secdata;
  22   pthread_t thread;
  23 } PerThread;
  24
  25 static void *routine(void *thread_v) {
  26   PerThread *t= thread_v;
  27
  28   for (;;) {
  29     inparallel_barrier(); /* wait for work to do */
  30     ForAllThreads *a= t->allthreads;
  31     a->separately(a->vertices, t->section, t->secdata, a->gendata);
  32     inparallel_barrier(); /* synchronise for completion */
  33   }
  34
  35   return 0;
  36 }
  37
  38 static int threads_started;
  39 static pthread_barrier_t threads_barrier;
  40 static PerThread threads[NSECTIONS-1];
  41
  42 void inparallel(const struct Vertices *vertices,
  43                 Computation *separately,
  44                 Computation *combine,
  45                 size_t secdatasz, void *gendata) {
  46   typedef struct { unsigned char secdata[secdatasz]; } SecData;
  47
  48   ForAllThreads allthreads;
  49   SecData secdatas[nsections];
  50
  51   allthreads.vertices= vertices;
  52   allthreads.separately= separately;
  53   allthreads.gendata= gendata;
  54
  55   int s, r;
  56
  57   if (!threads_started) {
  58     r= pthread_barrier_init(&threads_barrier, 0, NSECTIONS);
  59     if (r) { errno=r; diee("pthread_barrier_init"); }
  60
  61     for (s=0; s<NSECTIONS-1; s++) {
  62       r= pthread_create(&threads[s].thread,0,routine,&threads[s]);
  63       if (r) { errno=r; diee("pthread_create"); }
  64     }
  65   }
  66
  67   for (s=0; s<NSECTIONS-1; s++) {
  68     threads[s].allthreads= &allthreads;
  69     threads[s].section= s;
  70     threads[s].secdata= secdatas[s].secdata;
  71   }
  72
  73   inparallel_barrier(); /* announce more work to do */
  74
  75   separately(vertices, NSECTIONS-1, &secdatas[NSECTIONS-1], gendata);
  76
  77   inparallel_barrier(); /* synchronise for completion */
  78
  79   for (s=0; s<nsections; s++)
  80     combine(vertices, s, &secdatas[s].secdata, gendata);
  81 }
  82
  83 void inparallel_barrier(void) {
  84   int r;
  85   r= pthread_barrier_wait(&threads_barrier);
  86   if (r && r!=PTHREAD_BARRIER_SERIAL_THREAD)
  87     { errno=r; diee("pthread_barrier_wait"); }
  88 }