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 (NSECTIONS>1 && !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     threads_started= 1;
  66   }
  67
  68   for (s=0; s<NSECTIONS-1; s++) {
  69     threads[s].allthreads= &allthreads;
  70     threads[s].section= s;
  71     threads[s].secdata= secdatas[s].secdata;
  72   }
  73
  74   inparallel_barrier(); /* announce more work to do */
  75
  76   separately(vertices, NSECTIONS-1, &secdatas[NSECTIONS-1], gendata);
  77
  78   inparallel_barrier(); /* synchronise for completion */
  79
  80   for (s=0; s<nsections; s++)
  81     combine(vertices, s, &secdatas[s].secdata, gendata);
  82 }
  83
  84 void inparallel_barrier(void) {
  85   if (NSECTIONS>1) {
  86     int r;
  87     r= pthread_barrier_wait(&threads_barrier);
  88     if (r && r!=PTHREAD_BARRIER_SERIAL_THREAD)
  89       { errno=r; diee("pthread_barrier_wait"); }
  90   }
  91 }