Robust/src/Runtime/workschedule.c

   1 #include <stdlib.h>
   2 #include <stdio.h>
   3 #include <pthread.h>
   4
   5 #include "mem.h"
   6 #include "workschedule.h"
   7 #include "mlp_runtime.h"
   8 #include "coreprof/coreprof.h"
   9
  10 // NOTE: Converting this from a work-stealing strategy
  11 // to a single-queue thread pool protected by a single
  12 // lock.  This will not scale, but it will support
  13 // development of the system for now
  14
  15
  16
  17 // for convenience
  18 typedef struct Queue deq;
  19
  20 typedef struct workerData_t{
  21   pthread_t workerThread;
  22   int id;
  23 } WorkerData;
  24
  25
  26 static pthread_mutex_t systemLockIn;
  27 static pthread_mutex_t systemLockOut;
  28
  29 // implementation internal data
  30 static WorkerData*     workerDataArray;
  31 static pthread_t*      workerArray;
  32
  33 static int systemStarted = 0;
  34
  35 static pthread_cond_t  systemBeginCond  = PTHREAD_COND_INITIALIZER;
  36 static void(*workFunc)(void*);
  37
  38 static pthread_cond_t  workAvailCond  = PTHREAD_COND_INITIALIZER;
  39
  40 int             numWorkers;
  41
  42 int threadcount;
  43 pthread_mutex_t gclock;
  44 pthread_mutex_t gclistlock;
  45 pthread_cond_t gccond;
  46
  47 extern struct listitem * list;
  48 extern __thread struct listitem litem;
  49 extern __thread SESEcommon* seseCommon;
  50
  51 __thread int oid;
  52
  53
  54
  55 void workerExit( void* arg ) {
  56   //printf( "Thread %d canceled.\n", pthread_self() );
  57   CP_EXIT();
  58 }
  59
  60
  61
  62 void* workerMain( void* arg ) {
  63   void*       workUnit;
  64   WorkerData* myData = (WorkerData*) arg;
  65   int         oldState;
  66   int         haveWork;
  67
  68   // once-per-thread stuff
  69   CP_CREATE();
  70
  71   //pthread_cleanup_push( workerExit, NULL );
  72
  73   oid = myData->id;
  74
  75   //pthread_setcanceltype ( PTHREAD_CANCEL_ASYNCHRONOUS, &oldState );
  76   //pthread_setcancelstate( PTHREAD_CANCEL_ENABLE,       &oldState );
  77
  78   // then continue to process work
  79   while( 1 ) {
  80
  81     // wait for work
  82     CP_LOGEVENT( CP_EVENTID_WORKSCHEDGRAB, CP_EVENTTYPE_BEGIN );
  83     haveWork = FALSE;
  84     while( !haveWork ) {
  85       pthread_mutex_lock( &systemLockOut );
  86       if( headqi->next == NULL ) {
  87         pthread_mutex_unlock( &systemLockOut );
  88         sched_yield();
  89         continue;
  90       } else {
  91         haveWork = TRUE;
  92       }
  93     }
  94     struct QI * tmp=headqi;
  95     headqi = headqi->next;
  96     workUnit = headqi->value;
  97     pthread_mutex_unlock( &systemLockOut );
  98     free( tmp );
  99     CP_LOGEVENT( CP_EVENTID_WORKSCHEDGRAB, CP_EVENTTYPE_END );
 100
 101     pthread_mutex_lock(&gclistlock);
 102     threadcount++;
 103     litem.seseCommon=(void*)workUnit;
 104     litem.prev=NULL;
 105     litem.next=list;
 106     if(list!=NULL)
 107       list->prev=&litem;
 108     list=&litem;
 109     seseCommon=(SESEcommon*)workUnit;
 110     pthread_mutex_unlock(&gclistlock);
 111
 112     workFunc( workUnit );
 113
 114     pthread_mutex_lock(&gclistlock);
 115     threadcount--;
 116     if (litem.prev==NULL) {
 117       list=litem.next;
 118     } else {
 119       litem.prev->next=litem.next;
 120     }
 121     if (litem.next!=NULL) {
 122       litem.next->prev=litem.prev;
 123     }
 124     pthread_mutex_unlock(&gclistlock);
 125   }
 126
 127   //pthread_cleanup_pop( 0 );
 128
 129   return NULL;
 130 }
 131
 132 void workScheduleInit( int numProcessors,
 133                        void(*func)(void*) ) {
 134   int i, status;
 135
 136   // the original thread must call this now to
 137   // protect memory allocation events coming, but it
 138   // will also add itself to the worker pool and therefore
 139   // try to call it again, CP_CREATE should just ignore
 140   // duplicate calls
 141   CP_CREATE();
 142
 143   pthread_mutex_init(&gclock, NULL);
 144   pthread_mutex_init(&gclistlock, NULL);
 145   pthread_cond_init(&gccond, NULL);
 146
 147   numWorkers = numProcessors + 1;
 148
 149   workFunc   = func;
 150
 151   headqi=tailqi=RUNMALLOC(sizeof(struct QI));
 152   headqi->next=NULL;
 153
 154   status = pthread_mutex_init( &systemLockIn, NULL );
 155   status = pthread_mutex_init( &systemLockOut, NULL );
 156
 157   // allocate space for one more--the original thread (running
 158   // this code) will become a worker thread after setup
 159   workerDataArray = RUNMALLOC( sizeof( WorkerData ) * (numWorkers+1) );
 160
 161   for( i = 0; i < numWorkers; ++i ) {
 162
 163     // the original thread is ID 1, start counting from there
 164     workerDataArray[i].id = 2 + i;
 165
 166     status = pthread_create( &(workerDataArray[i].workerThread),
 167                              NULL,
 168                              workerMain,
 169                              (void*) &(workerDataArray[i])
 170                            );
 171
 172     if( status != 0 ) { printf( "Error\n" ); exit( -1 ); }
 173
 174     // yield and let all workers get to the begin
 175     // condition variable, waiting--we have to hold them
 176     // so they don't all see empty work queues right away
 177     if( sched_yield() == -1 ) { printf( "Error thread trying to yield.\n" ); exit( -1 ); }
 178   }
 179 }
 180
 181 void workScheduleSubmit( void* workUnit ) {
 182   struct QI* item=RUNMALLOC(sizeof(struct QI));
 183   item->value=workUnit;
 184   item->next=NULL;
 185
 186   pthread_mutex_lock  ( &systemLockIn );
 187   tailqi->next=item;
 188   tailqi=item;
 189   pthread_mutex_unlock( &systemLockIn );
 190 }
 191
 192
 193 // really should be named "add original thread as a worker"
 194 void workScheduleBegin() {
 195   int i;
 196
 197   // space was saved for the original thread to become a
 198   // worker after setup is complete
 199   workerDataArray[numWorkers].id           = 1;
 200   workerDataArray[numWorkers].workerThread = pthread_self();
 201   ++numWorkers;
 202
 203   workerMain( &(workerDataArray[numWorkers-1]) );
 204 }
 205
 206
 207 // the above function does NOT naturally join all the worker
 208 // threads at exit, once the main SESE/Rblock/Task completes
 209 // we know all worker threads are finished executing other
 210 // tasks so we can explicitly kill the workers, and therefore
 211 // trigger any worker-specific cleanup (like coreprof!)
 212 void workScheduleExit() {
 213   int i;
 214
 215   // This is not working well--canceled threads don't run their
 216   // thread-level exit routines?  Anyway, its not critical for
 217   // coreprof but if we ever need a per-worker exit routine to
 218   // run we'll have to look back into this.
 219
 220   //printf( "Thread %d performing schedule exit.\n", pthread_self() );
 221   //
 222   //for( i = 0; i < numWorkers; ++i ) {
 223   //  if( pthread_self() != workerDataArray[i].workerThread ) {
 224   //    pthread_cancel( workerDataArray[i].workerThread );
 225   //  }
 226   //}
 227   //
 228   //// how to let all the threads actually get canceled?
 229   //sleep( 2 );
 230 }