course/semester2/pprog/assignment1/mthread/nbody.c

   1 /**
   2  * @file nbody.c
   3  * @purpose Implentation of multi-threaded N-Body simulator using pthreads
   4  * @author Sam Moore (20503628) - 2012
   5  */
   6
   7 #include "nbody.h" // Declarations
   8 #include "../single-thread/nbody.c" // Include all functions from the single threaded version
   9
  10 // --- Variable declarations --- //
  11
  12 pthread_t compute_thread; // The thread responsible for computations; it spawns worker threads
  13
  14 pthread_t * worker_thread = NULL; //Array of worker threads responsible for Force and Position updates
  15 System * sub_system = NULL; //Array of Systems used to divide up the main "universe" System for worker threads
  16 pthread_mutex_t mutex_workers; // Mutex used for the barrier between Force and Position updates
  17 pthread_cond_t workers_done_cv; // Conditional used for the barrier between Force and Position updates
  18 unsigned workers_busy; // Number of workers currently doing something
  19
  20 pthread_mutex_t mutex_runstate; // Mutex around the runstate
  21
  22
  23
  24 /**
  25  * @function Compute_Thread
  26  * @purpose Thread - Continuously computes steps for a system of bodies. Seperate from graphics functions.
  27  *      Spawns worker threads to divide up computation.
  28  * @param arg - Can be cast to the System for which computations are performed (ie: &universe)
  29  */
  30 void * Compute_Thread(void * arg)
  31 {
  32
  33         System * s = (System*)(arg); //cast argument to a System*
  34
  35
  36         // If no number of threads provided, use the default value, unless someone changed that to a stupid value
  37         if (options.num_threads <= 0)
  38                 options.num_threads = (DEFAULT_WORKING_THREADS > 1) ? DEFAULT_WORKING_THREADS : 1;
  39
  40         // Do a sanity check; there is no point spawning more threads than bodies.
  41         if (options.num_threads > s->N)
  42         {
  43                 fprintf(stderr,
  44                         "Warning: Using %u threads instead of %u specified, because there are only %u bodies to simulate!\n",
  45                         s->N, options.num_threads, s->N);
  46                 options.num_threads = s->N;
  47         }
  48
  49         if (options.num_threads > 1) // Allocate worker threads and sub systems, as long as there would be more than 1
  50         {
  51                 worker_thread = (pthread_t*)(calloc(options.num_threads, sizeof(pthread_t)));
  52                 if (worker_thread == NULL)
  53                 {
  54                         perror("Couldn't allocate array of worker threads");
  55                         QuitProgram(true);
  56                         pthread_exit(NULL);
  57                 }
  58                 sub_system = (System*)(calloc(options.num_threads, sizeof(System)));
  59                 if (sub_system == NULL)
  60                 {
  61                         perror("Couldn't allocate array of systems for worker threads to use");
  62                         QuitProgram(true);
  63                         pthread_exit(NULL);
  64                 }
  65
  66                 // Divide up the Body array owned by s into options.num_threads arrays, one for each worker thread
  67                 unsigned bodies_per_system = (s->N) / options.num_threads;
  68                 unsigned remainder = (s->N) % options.num_threads;
  69                 for (unsigned i = 0; i < options.num_threads; ++i)
  70                 {
  71                         sub_system[i].body = (s->body)+(i*bodies_per_system);
  72                         sub_system[i].N = bodies_per_system;
  73                         sub_system[i].steps = 0;
  74                         if (i == options.num_threads - 1)
  75                                 sub_system[i].N += remainder; // The last thread gets the remainder
  76
  77                 }
  78         }
  79
  80
  81         pthread_attr_t attr; //thread attribute for the workers.
  82         pthread_attr_init(&attr);
  83         pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_DETACHED); //Needs to be detached, so that memory can be reused.
  84
  85         // The main computation loop
  86         while (true)
  87         {
  88
  89                 if (runstate != RUN) pthread_exit(NULL); //Check whether the main thread wants to exit
  90
  91
  92
  93                 //Check whether the program should quit due to steps being computed, or a timeout
  94                 if (options.timeout > 0.0)
  95                 {
  96                         if ((unsigned)(time(NULL) - options.start_time.tv_sec) >= options.timeout)
  97                                 QuitProgram(false);
  98                 }
  99
 100                 if (options.num_steps > 0 && universe.steps > options.num_steps)
 101                         QuitProgram(false);
 102
 103                 if (options.draw_graphics == false && options.verbosity != 0
 104                         && universe.steps % options.verbosity == 1)
 105                 {
 106                         DisplayStatistics();
 107                 }
 108
 109
 110                 if (options.num_threads <= 1)
 111                 {
 112                         // Just do everything in this thread
 113                         System_Forces(s, s);
 114                         System_Positions(s);
 115                         continue;
 116                 }
 117
 118
 119
 120
 121                 workers_busy = options.num_threads; //All threads working
 122
 123                 //Compute forces
 124                 for (unsigned i = 0; i < options.num_threads; ++i)
 125                 {
 126                         if (pthread_create(worker_thread+i, &attr, Force_Thread, (void*)(sub_system+i)) != 0)
 127                         {
 128                                 perror("In compute thread, couldn't create worker thread (force)");
 129                                 QuitProgram(true);
 130                                 pthread_exit(NULL);
 131                         }
 132                 }
 133
 134
 135
 136                 //Barrier - Wait for forces to be computed
 137                 pthread_mutex_lock(&mutex_workers);
 138                 while (workers_busy > 0)
 139                         pthread_cond_wait(&workers_done_cv, &mutex_workers);
 140                 pthread_mutex_unlock(&mutex_workers);
 141
 142                 //All the forces are now computed
 143
 144                 workers_busy = options.num_threads; //All threads working
 145
 146                 //Compute positions
 147                 for (unsigned i = 0; i < options.num_threads; ++i)
 148                 {
 149                         if (pthread_create(worker_thread+i, &attr, Position_Thread, (void*)(sub_system+i)) != 0)
 150                         {
 151                                 perror("In compute thread, couldn't create worker thread (position)");
 152                                 QuitProgram(true);
 153                                 pthread_exit(NULL);
 154                         }
 155                 }
 156
 157                 //Wait for positions to be computed
 158                 pthread_mutex_lock(&mutex_workers);
 159                 while (workers_busy > 0)
 160                         pthread_cond_wait(&workers_done_cv, &mutex_workers);
 161                 pthread_mutex_unlock(&mutex_workers);
 162
 163                 //Update number of steps computed
 164                 universe.steps += 1;
 165
 166
 167
 168         }
 169 }
 170
 171 /**
 172  * @function Force_Thread
 173  * @purpose Thread - Calculates the forces on objects in a System
 174  * @param s - Can be cast to the System*
 175  */
 176 void * Force_Thread(void * s)
 177 {
 178
 179         System_Forces((System*)s, &universe); //Simple wrapper
 180
 181         pthread_mutex_lock(&mutex_workers);
 182         workers_busy -= 1;      // Count this thread as done
 183         if (workers_busy == 0)
 184         {
 185                 pthread_cond_signal(&workers_done_cv); // All threads done; wake up the compute_thread
 186         }
 187         pthread_mutex_unlock(&mutex_workers);
 188         return NULL;
 189 }
 190
 191 /**
 192  * @function Position_Thread
 193  * @purpose Thread - Calculates the positions of objects in a System
 194  * @param s - Can be cast to the System*
 195  */
 196 void * Position_Thread(void * s)
 197 {
 198
 199         System_Positions((System*)s); // Simple wrapper
 200
 201         pthread_mutex_lock(&mutex_workers);
 202         workers_busy -= 1; // Count this thread as done
 203         if (workers_busy == 0)
 204         {
 205                 pthread_cond_signal(&workers_done_cv); //All threads done; wake up the compute_thread
 206         }
 207         pthread_mutex_unlock(&mutex_workers);
 208         return NULL;
 209 }
 210
 211 /**
 212  * @function QuitProgram
 213  * @purpose This function can either be called by the main thread in order to signal other threads
 214  *              that it wants to exit. The main thread then calls pthread_join before exiting.
 215  *      It can also be called by a child thread to request the main thread to exit.
 216  *      It is only used this way if there is an unrecovarable error (ie: Can't allocate memory in a child thread)
 217  */
 218 void QuitProgram(bool error)
 219 {
 220
 221         pthread_mutex_lock(&mutex_runstate); // aquire mutex
 222         if (error) // set the runstate
 223                 runstate = QUIT_ERROR;
 224         else
 225                 runstate = QUIT;
 226         pthread_mutex_unlock(&mutex_runstate); //release mutex
 227 }
 228
 229 /**
 230  * @function Thread_Cleanup
 231  * @purpose Will be called in the main thread when exit() is called
 232  *      Automatically tells all other threads to quit (if they haven't already been told)
 233  *      Then waits for them to finish.
 234  *      Also frees memory associated with the worker threads.
 235  */
 236 void Thread_Cleanup(void)
 237 {
 238         if (runstate == RUN) // If this is true, as far as child threads are concerned, the simulation is still running
 239                 QuitProgram(false); // So call QuitProgram which will set runstate, and cause child threads to exit
 240         pthread_join(compute_thread, NULL);
 241         free(worker_thread);
 242         free(sub_system);
 243 }
 244
 245
 246 /**
 247  * @function Simulation_Run
 248  * @purpose Initialise and start the simulation. Will be called in the main thread.
 249  *      Replaces the single-threaded macro that does nothing, and sets up the compute thread
 250  */
 251 void Simulation_Run()
 252 {
 253         atexit(Thread_Cleanup);
 254
 255         if (options.draw_graphics == false)
 256         {
 257                 Compute_Thread((void*)(&universe));
 258
 259                 // If there are no graphics, run the computation loop in the main thread
 260                 // The call to pthread_exit(NULL) in the computation loop ends the main thread
 261
 262                 // This means the main function never calls the Graphics_Run function
 263                 // Without this condition here, Graphics_Display would essentially be running a busy loop in the main thread
 264
 265                 fprintf(stderr,"Should not see this\n");
 266                 exit(EXIT_FAILURE);
 267         }
 268
 269         // Create a thread to handle computation loop
 270         if (pthread_create(&compute_thread, NULL, Compute_Thread, (void*)&universe) != 0)
 271         {
 272                 perror("Error creating compute thread");
 273                 exit(EXIT_FAILURE);
 274         }
 275 }