VMDThreads.C Source File

00001 /***************************************************************************
00002  *cr
00003  *cr            (C) Copyright 1995-2008 The Board of Trustees of the
00004  *cr                        University of Illinois
00005  *cr                         All Rights Reserved
00006  *cr
00007  ***************************************************************************/
00008 
00009 /***************************************************************************
00010  * RCS INFORMATION:
00011  *
00012  *      $RCSfile: VMDThreads.C,v $
00013  *      $Author: johns $        $Locker:  $             $State: Exp $
00014  *      $Revision: 1.31 $       $Date: 2008/03/27 19:36:49 $
00015  *
00016  ***************************************************************************
00017  * DESCRIPTION:
00018  *
00019  * VMDThreads.C - code for spawning threads on various platforms.
00020  *                Code donated by John Stone, john.stone@gmail.com 
00021  *                This code was originally written for the
00022  *                Tachyon Parallel/Multiprocessor Ray Tracer. 
00023  *                Improvements have been donated by Mr. Stone on an 
00024  *                ongoing basis. 
00025  *
00026  ***************************************************************************/
00027 
00028 #include <stdlib.h>
00029 #include <stdio.h>
00030 
00031 // If compiling on Linux, enable the GNU CPU affinity functions in both
00032 // libc and the libpthreads 
00033 #if defined(__linux)
00034 #define _GNU_SOURCE 1
00035 #endif
00036 
00037 #include "VMDThreads.h"
00038 
00039 #ifdef _MSC_VER
00040 #include <windows.h>
00041 #include <winbase.h>
00042 #endif
00043 
00044 /* needed for call to sysconf() */
00045 #if defined(__sun) || defined(ARCH_IRIX6) || defined(ARCH_IRIX6_64) || defined(ARCH_LINUX) || defined(ARCH_LINUXALPHA) || defined(ARCH_LINUXAMD64) || defined(ARCH_LINUXIA64) || defined(ARCH_LINUXPPC) || defined(ARCH_LINUXPPC64) || defined(_CRAY) || defined(__osf__) || defined(ARCH_AIX4) || defined(ARCH_AIX5) || defined(ARCH_AIX5_64)
00046 #include<unistd.h>
00047 #endif
00048 
00049 #if defined(__APPLE__) && defined(VMDTHREADS)
00050 #include <Carbon/Carbon.h> /* Carbon APIs for Multiprocessing */
00051 #endif
00052 
00053 #if defined(ARCH_HPUX11) ||  defined(ARCH_HPUX10)
00054 #include <sys/mpctl.h>
00055 #endif
00056 
00057 int vmd_thread_numphysprocessors(void) {
00058   int a=1;
00059 
00060 #ifdef VMDTHREADS
00061 #if defined(__APPLE__) 
00062   a = MPProcessorsScheduled(); /* Number of active/running CPUs */
00063 #endif
00064 
00065 #ifdef _MSC_VER
00066   struct _SYSTEM_INFO sysinfo;
00067   GetSystemInfo(&sysinfo);
00068   a = sysinfo.dwNumberOfProcessors; /* total number of CPUs */
00069 #endif /* _MSC_VER */
00070 
00071 #if defined(_CRAY)
00072   a = sysconf(_SC_CRAY_NCPU);
00073 #endif
00074 
00075 #if defined(__sun) || defined(ARCH_LINUX) || defined(ARCH_LINUXALPHA) || defined(ARCH_LINUXAMD64) || defined(ARCH_LINUXIA64) || defined(ARCH_LINUXPPC) || defined(ARCH_LINUXPPC64) || defined(__osf__) || defined(ARCH_AIX4) || defined(ARCH_AIX5) || defined(ARCH_AIX5_64)
00076   a = sysconf(_SC_NPROCESSORS_ONLN); /* number of active/running CPUs */
00077 #endif /* SunOS */
00078 
00079 #if defined(ARCH_IRIX6) || defined(ARCH_IRIX6_64)
00080   a = sysconf(_SC_NPROC_ONLN); /* number of active/running CPUs */
00081 #endif /* IRIX */
00082 
00083 #if defined(ARCH_HPUX11) || defined(ARCH_HPUX10)
00084   a = mpctl(MPC_GETNUMSPUS, 0, 0); /* total number of CPUs */
00085 #endif /* HPUX */
00086 #endif /* VMDTHREADS */
00087 
00088   return a;
00089 }
00090 
00091 int vmd_thread_numprocessors(void) {
00092   int a=1;
00093 
00094 #ifdef VMDTHREADS
00095   // Allow the user to override the number of CPUs for use
00096   // in scalability testing, debugging, etc.
00097   char *forcecount = getenv("VMDFORCECPUCOUNT");
00098   if (forcecount != NULL) {
00099     if (sscanf(forcecount, "%d", &a) == 1) {
00100       return a; // if we got a valid count, return it
00101     } else {
00102       a=1;      // otherwise use the real available hardware CPU count
00103     }
00104   }
00105 
00106   // otherwise the number of physical processors that are currently available
00107   a = vmd_thread_numphysprocessors();
00108 
00109   // XXX we should add checking for the current CPU affinity masks here,
00110   // and return the min of the physical processor count and CPU affinity mask
00111   // enabled CPU count.
00112 #endif /* VMDTHREADS */
00113 
00114   return a;
00115 }
00116 
00117 
00118 int * vmd_cpu_affinitylist(int *cpuaffinitycount) {
00119   int *affinitylist = NULL;
00120   *cpuaffinitycount = -1; // return count -1 if unimplemented or err occurs
00121 
00122 // Win32 process affinity mask query
00123 // XXX untested, but based on the linux code, may work with a few tweaks
00124 #if 0 && defined(_MSC_VER)
00125   HANDLE myproc = GetCurrentProcess(); // returns a psuedo-handle
00126   DWORD affinitymask, sysaffinitymask;
00127 
00128   if (!GetProcessAffinityMask(myproc, &affinitymask, &sysaffinitymask)) {
00129     // count length of affinity list
00130     int affinitycount=0;
00131     int i;
00132     for (i=0; i<31; i++) {
00133       affinitycount += (affinitymask >> i) & 0x1;
00134     }
00135   
00136     // build affinity list
00137     if (affinitycount > 0) {
00138       affinitylist = (int *) malloc(affinitycount * sizeof(int));
00139       if (affinitylist == NULL)
00140         return NULL;
00141 
00142       int curcount = 0;
00143       for (i=0; i<CPU_SETSIZE; i++) {
00144         if (CPU_ISSET(i, &affinitymask)) {
00145           affinitylist[curcount] = i;
00146           curcount++;
00147         }
00148       }
00149     }
00150 
00151     *cpuaffinitycount = affinitycount; // return final affinity list
00152   }
00153 #endif
00154 
00155 // Linux process affinity mask query
00156 #if defined(ARCH_LINUX) || defined(ARCH_LINUXALPHA) || defined(ARCH_LINUXAMD64) || defined(ARCH_LINUXIA64) || defined(ARCH_LINUXPPC) || defined(ARCH_LINUXPPC64)
00157 
00158 // protect ourselves from some older Linux distros
00159 #if defined(CPU_SETSIZE)
00160   int i;
00161   cpu_set_t affinitymask;
00162   int affinitycount=0;
00163 
00164   /* PID 0 refers to the current process */
00165   if (sched_getaffinity(0, sizeof(affinitymask), &affinitymask) < 0) {
00166     perror("vmd_cpu_affinitylist: sched_getaffinity");
00167     return NULL;
00168   }
00169 
00170   // count length of affinity list
00171   for (i=0; i<CPU_SETSIZE; i++) {
00172     affinitycount += CPU_ISSET(i, &affinitymask);
00173   }
00174 
00175   // build affinity list
00176   if (affinitycount > 0) {
00177     affinitylist = (int *) malloc(affinitycount * sizeof(int));
00178     if (affinitylist == NULL)
00179       return NULL;
00180 
00181     int curcount = 0;
00182     for (i=0; i<CPU_SETSIZE; i++) {
00183       if (CPU_ISSET(i, &affinitymask)) {
00184         affinitylist[curcount] = i;
00185         curcount++;
00186       }
00187     }
00188   }
00189 
00190   *cpuaffinitycount = affinitycount; // return final affinity list
00191 #endif
00192 #endif
00193 
00194   // MacOS X doesn't presently (maybe in 10.5?) have a CPU affinity
00195   // query/set capability as their scheduler doesn't know about affinity yet
00196 
00197   // Solaris and HP-UX use pset_bind() and related functions, and they
00198   // don't use the single-level mask-based scheduling mechanism that 
00199   // the others, use.  Instead, they use a hierarchical tree of 
00200   // processor sets and processes float within those, or are tied to one
00201   // processor that's a member of a particular set.
00202 
00203   return affinitylist;
00204 }
00205 
00206 
00207 int vmd_thread_set_self_cpuaffinity(int cpu) {
00208   int status=-1; // unsupported by default
00209 
00210 #ifdef VMDTHREADS
00211 #if defined(__linux)
00212 // XXX this code is too new even for RHEL4, though it will run on Fedora 7
00213 // and other newer revs.
00214 #if 0
00215   // NPTL systems can assign per-thread affinities this way
00216   cpu_set_t affinitymask;
00217   CPU_ZERO(&affinitymask); 
00218   CPU_SET(cpu, &affinitymask);
00219   status = pthread_setaffinity_np(pthread_self(), sizeof(affinitymask), &affinitymask);
00220 #else
00221   // non-NPTL systems based on the clone() interface must use this method
00222   cpu_set_t affinitymask;
00223   CPU_ZERO(&affinitymask); 
00224   CPU_SET(cpu, &affinitymask);
00225 
00226   /* PID 0 refers to the current process */
00227   if ((status=sched_getaffinity(0, sizeof(affinitymask), &affinitymask)) < 0) {
00228     perror("vmd_thread_set_self_cpuaffinitylist: sched_getaffinity");
00229     return status;
00230   }
00231 #endif
00232 #endif /* linux */
00233 #endif
00234 
00235   return status;
00236 }
00237 
00238 
00239 int vmd_thread_setconcurrency(int nthr) {
00240   int status=0;
00241 
00242 #ifdef VMDTHREADS
00243 #if defined(__sun) 
00244 #ifdef USEPOSIXTHREADS 
00245   status = pthread_setconcurrency(nthr);
00246 #else
00247   status = thr_setconcurrency(nthr);
00248 #endif
00249 #endif /* SunOS */
00250 
00251 #if defined(ARCH_IRIX6) || defined(ARCH_IRIX6_64) || defined(ARCH_AIX4) || defined(ARCH_AIX5) || defined(ARCH_AIX5_64)
00252   status = pthread_setconcurrency(nthr);
00253 #endif
00254 #endif /* VMDTHREADS */
00255 
00256   return status;
00257 }
00258 
00259 
00260 
00261 // Typedef to eliminate compiler warning caused by C/C++ linkage conflict.
00262 extern "C" {
00263   typedef void * (*VMDTHREAD_START_ROUTINE)(void *);
00264 }
00265 
00266 int vmd_thread_create(vmd_thread_t * thr, void * routine(void *), void * arg) {
00267   int status=0;
00268 
00269 #ifdef VMDTHREADS 
00270 #ifdef _MSC_VER
00271   int tid; /* thread id, msvc only */
00272  
00273   *thr = CreateThread(NULL, 8192,
00274                     (LPTHREAD_START_ROUTINE) routine, arg, 0, &tid);
00275  
00276   if (*thr == NULL) {
00277     status = -1;
00278   }
00279 #endif /* _MSC_VER */
00280 
00281 #ifdef USEPOSIXTHREADS 
00282 #if defined(ARCH_AIX4) || defined(ARCH_AIX5) || defined(ARCH_AIX5_64)
00283   {
00284     pthread_attr_t attr;
00285     pthread_attr_init(&attr);
00286     pthread_attr_setscope(&attr, PTHREAD_SCOPE_SYSTEM);
00287     status = pthread_create(thr, &attr, routine, arg);
00288     pthread_attr_destroy(&attr);
00289   }
00290 #else   
00291   status = pthread_create(thr, NULL, (VMDTHREAD_START_ROUTINE)routine, arg);
00292 #endif 
00293 #endif /* USEPOSIXTHREADS */
00294 
00295 #ifdef USEUITHREADS 
00296   status = thr_create(NULL, 0, routine, arg, 0, thr); 
00297 #endif /* USEUITHREADS */
00298 #endif /* VMDTHREADS */
00299  
00300   return status;
00301 }
00302 
00303 
00304 int vmd_thread_join(vmd_thread_t thr, void ** stat) {
00305   int status=0;  
00306 
00307 #ifdef VMDTHREADS
00308 #ifdef _MSC_VER
00309   DWORD wstatus = 0;
00310  
00311   wstatus = WAIT_TIMEOUT;
00312  
00313   while (wstatus != WAIT_OBJECT_0) {
00314     wstatus = WaitForSingleObject(thr, INFINITE);
00315   }
00316 #endif /* _MSC_VER */
00317 
00318 #ifdef USEPOSIXTHREADS
00319   status = pthread_join(thr, stat);
00320 #endif /* USEPOSIXTHREADS */
00321 
00322 #ifdef USEUITHREADS
00323   status = thr_join(thr, NULL, stat);
00324 #endif /* USEPOSIXTHREADS */
00325 #endif /* VMDTHREADS */
00326 
00327   return status;
00328 }  
00329 
00330 
00331 int vmd_mutex_init(vmd_mutex_t * mp) {
00332   int status=0;
00333 
00334 #ifdef VMDTHREADS
00335 #ifdef USEPOSIXTHREADS
00336   status = pthread_mutex_init(mp, 0);
00337 #endif /* USEPOSIXTHREADS */
00338 
00339 #ifdef USEUITHREADS 
00340   status = mutex_init(mp, USYNC_THREAD, NULL);
00341 #endif /* USEUITHREADS */
00342 #endif /* VMDTHREADS */
00343 
00344   return status;
00345 }
00346 
00347 
00348 int vmd_mutex_lock(vmd_mutex_t * mp) {
00349   int status=0;
00350 
00351 #ifdef VMDTHREADS
00352 #ifdef USEPOSIXTHREADS
00353   status = pthread_mutex_lock(mp);
00354 #endif /* USEPOSIXTHREADS */
00355 
00356 #ifdef USEUITHREADS
00357   status = mutex_lock(mp);
00358 #endif /* USEUITHREADS */
00359 #endif /* VMDTHREADS */
00360 
00361   return status;
00362 }
00363 
00364 
00365 int vmd_mutex_unlock(vmd_mutex_t * mp) {
00366   int status=0;
00367 
00368 #ifdef VMDTHREADS  
00369 #ifdef USEPOSIXTHREADS
00370   status = pthread_mutex_unlock(mp);
00371 #endif /* USEPOSIXTHREADS */
00372 
00373 #ifdef USEUITHREADS
00374   status = mutex_unlock(mp);
00375 #endif /* USEUITHREADS */
00376 #endif /* VMDTHREADS */
00377 
00378   return status;
00379 }
00380 
00381 
00382 int vmd_mutex_destroy(vmd_mutex_t * mp) {
00383   int status=0;
00384 
00385 #ifdef VMDTHREADS
00386 #ifdef USEPOSIXTHREADS
00387   status = pthread_mutex_destroy(mp);
00388 #endif /* USEPOSIXTHREADS */
00389 
00390 #ifdef USEUITHREADS 
00391   status = mutex_destroy(mp);
00392 #endif /* USEUITHREADS */
00393 #endif /* VMDTHREADS */
00394 
00395   return status;
00396 }
00397 
00398 #if !defined(VMDTHREADS)
00399 
00400 int vmd_thread_barrier_init(vmd_barrier_t *barrier, int n_clients) {
00401   return 0;
00402 }
00403 
00404 void vmd_thread_barrier_destroy(vmd_barrier_t *barrier) {
00405 }
00406 
00407 int vmd_thread_barrier(vmd_barrier_t *barrier, int increment) {
00408   return 0;
00409 }
00410 
00411 #else 
00412 
00413 #ifdef USEPOSIXTHREADS
00414 int vmd_thread_barrier_init(vmd_barrier_t *barrier, int n_clients) {
00415   if (barrier != NULL) {
00416     barrier->n_clients = n_clients;
00417     barrier->n_waiting = 0;
00418     barrier->phase = 0;
00419     barrier->sum = 0;
00420 
00421 #if defined(VMDCAVE)
00422     // XXX the CAVE requires that we use a special synchronization
00423     //     mode so that mutexes and condition variables among multiple 
00424     //     processes in shared memory will work correctly.
00425     pthread_mutexattr_t mattr;
00426     pthread_condattr_t  cattr;
00427 
00428     printf("Setting barriers to have system scope...\n");
00429 
00430     pthread_mutexattr_init(&mattr);
00431     if (pthread_mutexattr_setpshared(&mattr, PTHREAD_PROCESS_SHARED) != 0) {
00432       printf("WARNING: could not set mutex to process shared scope\n");
00433     }
00434 
00435     pthread_condattr_init(&cattr);
00436     if (pthread_condattr_setpshared(&cattr, PTHREAD_PROCESS_SHARED) != 0) {
00437       printf("WARNING: could not set mutex to process shared scope\n");
00438     }
00439 
00440     pthread_mutex_init(&barrier->lock, &mattr);
00441     pthread_cond_init(&barrier->wait_cv, &cattr);
00442 
00443     pthread_condattr_destroy(&cattr);
00444     pthread_mutexattr_destroy(&mattr);
00445 #else
00446     pthread_mutex_init(&barrier->lock, NULL);
00447     pthread_cond_init(&barrier->wait_cv, NULL);
00448 #endif
00449   }
00450 
00451   return 0;
00452 }
00453 
00454 void vmd_thread_barrier_destroy(vmd_barrier_t *barrier) {
00455   pthread_mutex_destroy(&barrier->lock);
00456   pthread_cond_destroy(&barrier->wait_cv);
00457   free(barrier);
00458 }
00459 
00460 int vmd_thread_barrier(vmd_barrier_t *barrier, int increment) {
00461   int my_phase;
00462 
00463   pthread_mutex_lock(&barrier->lock);
00464   my_phase = barrier->phase;
00465   barrier->sum += increment;
00466   barrier->n_waiting++;
00467 
00468   if (barrier->n_waiting == barrier->n_clients) {
00469     barrier->result = barrier->sum;
00470     barrier->sum = 0;
00471     barrier->n_waiting = 0;
00472     barrier->phase = 1 - my_phase;
00473     pthread_cond_broadcast(&barrier->wait_cv);
00474   }
00475 
00476   while (barrier->phase == my_phase) {
00477     pthread_cond_wait(&barrier->wait_cv, &barrier->lock);
00478   }
00479 
00480   pthread_mutex_unlock(&barrier->lock);
00481 
00482   return (barrier->result);
00483 }
00484 #endif
00485 
00486 
00487 #ifdef USEUITHREADS
00488 
00489 int vmd_thread_barrier_init(vmd_barrier_t * barrier, int n_clients) {
00490   if (barrier != NULL) {
00491     barrier->n_clients = n_clients;
00492     barrier->n_waiting = 0;
00493     barrier->phase = 0;
00494     barrier->sum = 0;
00495     mutex_init(&barrier->lock, USYNC_THREAD, NULL);
00496     cond_init(&barrier->wait_cv, USYNC_THREAD, NULL);
00497   }
00498 
00499   return 0;
00500 }
00501 
00502 void vmd_thread_barrier_destroy(vmd_barrier_t *barrier) {
00503   mutex_destroy(&barrier->lock);
00504   cond_destroy(&barrier->wait_cv);
00505   free(barrier);
00506 }
00507 
00508 int vmd_thread_barrier(vmd_barrier_t *barrier, int increment) {
00509   int my_phase;
00510 
00511   mutex_lock(&barrier->lock);
00512   my_phase = barrier->phase;
00513   barrier->sum += increment;
00514   barrier->n_waiting++;
00515 
00516   if (barrier->n_waiting == barrier->n_clients) {
00517     barrier->result = barrier->sum;
00518     barrier->sum = 0;
00519     barrier->n_waiting = 0;
00520     barrier->phase = 1 - my_phase;
00521     cond_broadcast(&barrier->wait_cv);
00522   }
00523 
00524   while (barrier->phase == my_phase) {
00525     cond_wait(&barrier->wait_cv, &barrier->lock);
00526   }
00527 
00528   mutex_unlock(&barrier->lock);
00529 
00530   return (barrier->result);
00531 }
00532 
00533 #endif /* USEUITHREADS */
00534 
00535 
00536 #endif /* VMDTHREADS */
00537 
00538 
00539