Parallel Programming - Commit before I break everything
[matches/honours.git] / course / semester2 / pprog / assignment1 / mthread / nbody.c
index afd1d60..de9b8aa 100644 (file)
@@ -7,20 +7,29 @@
 #include "nbody.h" // Declarations
 #include "../single-thread/nbody.c" // Include all functions from the single threaded version
 
+#include "graphics.h" // For declaration of Graphics_Run only
+
 // --- Variable declarations --- //
 
 pthread_t compute_thread; // The thread responsible for computations; it spawns worker threads
        
 pthread_t * worker_thread = NULL; //Array of worker threads responsible for Force and Position updates
 System * sub_system = NULL; //Array of Systems used to divide up the main "universe" System for worker threads
-pthread_mutex_t mutex_workers; // Mutex used for the barrier between Force and Position updates
-pthread_cond_t workers_done_cv; // Conditional used for the barrier between Force and Position updates
-unsigned workers_busy; // Number of workers currently doing something
+System * nested_sub_system = NULL; //Array of Systems for division of "universe" for the nested worker threads
 
 pthread_mutex_t mutex_runstate; // Mutex around the runstate
 
 
 
+pthread_attr_t attr; //thread attribute for the workers. 
+
+Barrier force_barrier; // I laughed at this variable name. A bit sad really.
+Barrier position_barrier;
+
+
+Barrier graphics_barrier;
+
+
 /**
  * @function Compute_Thread
  * @purpose Thread - Continuously computes steps for a system of bodies. Seperate from graphics functions.
@@ -37,6 +46,9 @@ void * Compute_Thread(void * arg)
        if (options.num_threads <= 0)
                options.num_threads = (DEFAULT_WORKING_THREADS > 1) ? DEFAULT_WORKING_THREADS : 1;
 
+       if (options.nested_threads <= 0)
+               options.nested_threads = 1;
+
        // Do a sanity check; there is no point spawning more threads than bodies.
        if (options.num_threads > s->N)
        {
@@ -46,66 +58,53 @@ void * Compute_Thread(void * arg)
                options.num_threads = s->N;
        }
        
+       pthread_attr_init(&attr);
+       pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_DETACHED); //Needs to be detached, so that memory can be reused.
+       
        if (options.num_threads > 1) // Allocate worker threads and sub systems, as long as there would be more than 1
        {
-               worker_thread = (pthread_t*)(calloc(options.num_threads, sizeof(pthread_t)));
-               if (worker_thread == NULL)
-               {
-                       perror("Couldn't allocate array of worker threads");
-                       QuitProgram(true);
-                       pthread_exit(NULL);
-               }
-               sub_system = (System*)(calloc(options.num_threads, sizeof(System)));
-               if (sub_system == NULL)
+               worker_thread = Allocate_Threads(options.num_threads);
+               sub_system = Split_System(&universe, options.num_threads);
+
+               if (options.nested_threads > 1)
+                       nested_sub_system = Split_System(&universe, options.nested_threads);
+
+               #ifdef PERSISTENT_THREADS
+               for (unsigned i = 0; i < options.num_threads; ++i)
                {
-                       perror("Couldn't allocate array of systems for worker threads to use");
-                       QuitProgram(true);
-                       pthread_exit(NULL);
+                       if (pthread_create(worker_thread+i, & attr, Worker_Thread, (void*)(sub_system+i)) != 0)
+                       {
+                               perror("In compute thread, couldn't create worker thread");
+                               QuitProgram(true);
+                               pthread_exit(NULL);
+                       }
                }
+               #endif //PERSISTENT_THREADS
 
-               // Divide up the Body array owned by s into options.num_threads arrays, one for each worker thread
-               unsigned bodies_per_system = (s->N) / options.num_threads;
-               unsigned remainder = (s->N) % options.num_threads;
-               for (unsigned i = 0; i < options.num_threads; ++i)
+               
+       }
+       #ifdef PERSISTENT_THREADS
+       else
+       {
+               while (!ExitCondition())
                {
-                       sub_system[i].body = (s->body)+(i*bodies_per_system);
-                       sub_system[i].N = bodies_per_system;
-                       sub_system[i].steps = 0;
-                       if (i == options.num_threads - 1)
-                               sub_system[i].N += remainder; // The last thread gets the remainder
+                       if (options.verbosity != 0 && universe.steps % options.verbosity == 1)
+                               DisplayStatistics();
 
+                       // Just do everything in this thread 
+                       System_Forces(s, s);
+                       System_Positions(s);
                }
+               QuitProgram(false);
+               pthread_exit(NULL);
        }
 
-
-       pthread_attr_t attr; //thread attribute for the workers. 
-       pthread_attr_init(&attr);
-       pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_DETACHED); //Needs to be detached, so that memory can be reused.
-
+       #else
        // The main computation loop
-       while (true)
+       while (!ExitCondition())
        {
-               
-               if (runstate != RUN) pthread_exit(NULL); //Check whether the main thread wants to exit
-
-
-       
-               //Check whether the program should quit due to steps being computed, or a timeout
-               if (options.timeout > 0.0)
-               {
-                       if ((unsigned)(time(NULL) - options.start_time.tv_sec) >= options.timeout)
-                               QuitProgram(false);
-               }
-
-               if (options.num_steps > 0 && universe.steps > options.num_steps)
-                       QuitProgram(false);
-
-               if (options.draw_graphics == false && options.verbosity != 0 
-                       && universe.steps % options.verbosity == 1)
-               {
+               if (options.verbosity != 0 && universe.steps % options.verbosity == 1)
                        DisplayStatistics();
-               }
-
 
                if (options.num_threads <= 1)
                {
@@ -117,9 +116,6 @@ void * Compute_Thread(void * arg)
 
                
 
-               
-               workers_busy = options.num_threads; //All threads working
-
                //Compute forces
                for (unsigned i = 0; i < options.num_threads; ++i)
                {
@@ -129,19 +125,18 @@ void * Compute_Thread(void * arg)
                                QuitProgram(true);
                                pthread_exit(NULL);
                        }       
+                       
                }
 
 
-
-               //Barrier - Wait for forces to be computed
-               pthread_mutex_lock(&mutex_workers);
-               while (workers_busy > 0)
-                       pthread_cond_wait(&workers_done_cv, &mutex_workers);
-               pthread_mutex_unlock(&mutex_workers);
+               Barrier_Wait(&force_barrier);
 
                //All the forces are now computed
-               
-               workers_busy = options.num_threads; //All threads working
+
+               if (options.draw_graphics && options.pedantic_graphics)
+               {
+                       Barrier_Wait(&graphics_barrier);
+               }
 
                //Compute positions
                for (unsigned i = 0; i < options.num_threads; ++i)
@@ -155,19 +150,144 @@ void * Compute_Thread(void * arg)
                }
 
                //Wait for positions to be computed
-               pthread_mutex_lock(&mutex_workers);
-               while (workers_busy > 0)
-                       pthread_cond_wait(&workers_done_cv, &mutex_workers);
-               pthread_mutex_unlock(&mutex_workers);
+               Barrier_Wait(&position_barrier);
+
 
                //Update number of steps computed
                universe.steps += 1;
 
+               
+
+               
 
 
        }
+       QuitProgram(false);
+       #endif //PERSISTENT_THREADS
+       return NULL;
 }
 
+/**
+ * @function BeforeDraw
+ * @purpose Called in graphics thread before the draw loop
+ *     When --pedantic-graphics enabled, will wait for position computations to finish before drawing
+ *     Otherwise does nothing
+ */
+void BeforeDraw()
+{
+       if (options.verbosity != 0 && universe.steps % options.verbosity == 0)
+               DisplayStatistics();
+       //printf("BEFORE DRAW\n");
+       if (!options.pedantic_graphics)
+               return;
+       
+       Barrier_Wait(&position_barrier);
+
+
+       Barrier_Enter(&graphics_barrier);       
+}
+
+/**
+ * @function AfterDraw
+ * @purpose Called in graphics thread after the draw loop
+ *     When --pedantic-graphics is supplied, will signal computation thread that drawing is finished
+ *             So that positions can be safely altered
+ *     Otherwise does nothing
+ */
+void AfterDraw()
+{
+       universe.steps += 1;
+       if (!options.pedantic_graphics)
+               return;
+       Barrier_Leave(&graphics_barrier);
+       
+}
+
+#ifdef PERSISTENT_THREADS
+
+/**
+ * @function Worker_Thread
+ * @purpose Thread - Calculate stuff
+ */
+void * Worker_Thread(void * arg)
+{
+       System * s = (System*)(arg);
+
+
+       pthread_t * nested_workers = NULL;
+       System_ForcePair * system_pairs = NULL;
+       System * nested_position = NULL;
+       
+       Barrier nested_barrier; 
+       Barrier_Init(&nested_barrier);
+
+       printf("options.nested_threads == %d\n", (int)(options.nested_threads));
+
+       if (options.nested_threads != 1)        
+       {
+
+               system_pairs = (System_ForcePair*)(calloc(options.nested_threads, sizeof(System_ForcePair)));
+               if (system_pairs == NULL) // Handle tedious error cases
+               {
+                       perror("Couldn't allocate array of system pairs");
+                       QuitProgram(true);
+                       pthread_exit(NULL);
+               }
+               nested_workers = Allocate_Threads(options.nested_threads);
+               nested_position = 
+
+               for (unsigned i = 0; i < options.nested_threads; ++i)
+               {
+                       system_pairs[i].A = s;
+                       system_pairs[i].B = nested_sub_system+i;
+               }
+       }
+
+       while (!ExitCondition())
+       {
+               
+               if (options.nested_threads == 1)
+               {
+                       Barrier_Enter(&force_barrier);
+                       System_Forces(s, &universe);
+                       Barrier_Leave(&force_barrier);
+               }
+               else
+               {
+                       for (unsigned i = 0; i < options.nested_threads; ++i)
+                       {
+                               if (pthread_create(nested_workers+i, &attr, Force_Thread, (void*)(system_pairs+i)) != 0)
+                               {
+                                       perror("In worker thread, couldn't create nested worker thread (force)");
+                                       QuitProgram(true);
+                                       free(nested_workers);
+                                       pthread_exit(NULL);                                     
+                               }       
+                       }
+               }
+               //printf("Computed forces for %p\n", arg);
+               Barrier_Wait(&force_barrier);
+               //printf("Computed ALL forces\n");
+
+               if (options.draw_graphics && options.pedantic_graphics)
+                       Barrier_Wait(&graphics_barrier);
+
+               Barrier_Enter(&position_barrier);
+               
+               System_Positions(s);
+
+               Barrier_Leave(&position_barrier);
+               //printf("Computed positions for %p\n", arg);
+               Barrier_Wait(&position_barrier);
+               //printf("Computed ALL positions\n");
+       }
+       printf("Worker thread exits\n");
+       QuitProgram(false);
+       pthread_exit(NULL);
+}
+
+#endif //PERSISTENT_THREADS
+
 /**
  * @function Force_Thread
  * @purpose Thread - Calculates the forces on objects in a System
@@ -175,19 +295,17 @@ void * Compute_Thread(void * arg)
  */
 void * Force_Thread(void * s)
 {
-       
-       System_Forces((System*)s, &universe); //Simple wrapper
+       System_ForcePair * pair = (System_ForcePair*)s;
+       Barrier_Enter(&force_barrier);
+
+       System_Forces(pair->A, pair->B); //Simple wrapper
+
+       Barrier_Leave(&force_barrier);
 
-       pthread_mutex_lock(&mutex_workers);
-       workers_busy -= 1;      // Count this thread as done
-       if (workers_busy == 0)
-       {
-               pthread_cond_signal(&workers_done_cv); // All threads done; wake up the compute_thread
-       }
-       pthread_mutex_unlock(&mutex_workers);
        return NULL;
 }
 
+
 /**
  * @function Position_Thread
  * @purpose Thread - Calculates the positions of objects in a System 
@@ -195,16 +313,9 @@ void * Force_Thread(void * s)
  */
 void * Position_Thread(void * s)
 {
-       
+       Barrier_Enter(&position_barrier);
        System_Positions((System*)s); // Simple wrapper
-
-       pthread_mutex_lock(&mutex_workers);
-       workers_busy -= 1; // Count this thread as done
-       if (workers_busy == 0)
-       {
-               pthread_cond_signal(&workers_done_cv); //All threads done; wake up the compute_thread
-       }
-       pthread_mutex_unlock(&mutex_workers);
+       Barrier_Leave(&position_barrier);
        return NULL;
 }      
 
@@ -217,7 +328,8 @@ void * Position_Thread(void * s)
  */
 void QuitProgram(bool error)
 {
-       
+       if (runstate == QUIT || runstate == QUIT_ERROR)
+               return; //Don't do anything if already quitting
        pthread_mutex_lock(&mutex_runstate); // aquire mutex
        if (error) // set the runstate
                runstate = QUIT_ERROR;
@@ -247,29 +359,122 @@ void Thread_Cleanup(void)
  * @function Simulation_Run
  * @purpose Initialise and start the simulation. Will be called in the main thread.
  *     Replaces the single-threaded macro that does nothing, and sets up the compute thread
+ * @param argc - Number of arguments - Passed to Graphics_Run if needed
+ * @param argv - Argument strings - Passed to Graphics_Run if needed
  */
-void Simulation_Run()
+void Simulation_Run(int argc, char ** argv)
 {
        atexit(Thread_Cleanup);
 
-       if (options.draw_graphics == false)
+       Barrier_Init(&force_barrier);
+       Barrier_Init(&position_barrier);
+       Barrier_Init(&graphics_barrier);
+
+       
+       if (options.draw_graphics)
        {
+               // The graphics are enabled, so create a thread to do computations
+               // Graphics are done in the main loop
+               //printf("Graphics are enabled\n");
+               #ifdef PERSISTENT_THREADS
                Compute_Thread((void*)(&universe));
+               #else
+               if (pthread_create(&compute_thread, NULL, Compute_Thread, (void*)&universe) != 0)
+               {
+                       perror("Error creating compute thread");
+                       exit(EXIT_FAILURE);
+               }
+               #endif //PERSISTENT_THREADS
+               //printf("Run compute thread\n");
+               Graphics_Run(argc, argv);
+       }
+       else
+       
+               Compute_Thread((void*)(&universe)); // Graphics are disabled, so do computations in the main thread
+}
 
-               // If there are no graphics, run the computation loop in the main thread
-               // The call to pthread_exit(NULL) in the computation loop ends the main thread
 
-               // This means the main function never calls the Graphics_Run function
-               // Without this condition here, Graphics_Display would essentially be running a busy loop in the main thread
-               
-               fprintf(stderr,"Should not see this\n");
-               exit(EXIT_FAILURE);
+
+void Barrier_Init(Barrier * b)
+{
+       b->threads_busy = 0;
+}      
+
+void Barrier_Enter(Barrier * b)
+{
+       pthread_mutex_lock(&(b->mutex));
+       b->threads_busy += 1;
+       pthread_mutex_unlock(&(b->mutex));
+}
+
+void Barrier_Leave(Barrier * b)
+{
+       pthread_mutex_lock(&(b->mutex));
+       b->threads_busy -= 1;
+       if (b->threads_busy == 0)
+       {
+               pthread_cond_signal(&(b->threads_done_cv));
        }
-       
-       // Create a thread to handle computation loop
-       if (pthread_create(&compute_thread, NULL, Compute_Thread, (void*)&universe) != 0)
+       pthread_mutex_unlock(&(b->mutex));
+}
+
+void Barrier_Wait(Barrier * b)
+{
+       pthread_mutex_lock(&(b->mutex));
+       while (b->threads_busy > 0)
+               pthread_cond_wait(&(b->threads_done_cv), &(b->mutex));
+       pthread_mutex_unlock(&(b->mutex));
+}
+
+/**
+ * @function Split_System
+ * @purpose Helper to divide one system into an array of systems
+ *     Each sub system will have N = (s->N / n) bodies in it
+ * @param s - The original system (typically &universe)
+ * @param n - The number of sub systems in the array
+ *
+ * WARNING: It is the caller's responsibility to free() the returned array
+ */
+System * Split_System(System * s, unsigned n)
+{
+       System * result = (System*)(calloc(n, sizeof(System)));
+       if (result == NULL)
+       {
+               perror("Couldn't create array of sub systems");
+               QuitProgram(true);
+               pthread_exit(NULL);
+       }
+
+       unsigned n_per_system = (s->N) / n;
+       unsigned remainder = (s->N) % n;
+
+       for (unsigned i = 0; i < n; ++i)        
+       {
+               result[i].N = n_per_system;
+               if (i == n-1)
+                       result[i].N += remainder;
+               result[i].body = (s->body) + (n_per_system * i);
+               result[i].steps = 0;
+       }
+       return result;
+}
+
+/**
+ * @function Allocate_Threads
+ * @purpose Helper function to allocate an array of pthread_t objects
+ *     Handles all the pointless, er, "important" error checking that should be done
+ * @param n - Number of threads in the array
+ * 
+ * WARNING: Remember to free() the array!!!
+ */
+pthread_t * Allocate_Threads(unsigned n)
+{
+       pthread_t * result = (pthread_t*)(calloc(n, sizeof(pthread_t)));
+       if (result == NULL)
        {
-               perror("Error creating compute thread");
-               exit(EXIT_FAILURE);
+               perror("Unable to allocate memory for threads");
+               QuitProgram(true);
+               pthread_exit(NULL);
        }
+       return result;
 }

UCC git Repository :: git.ucc.asn.au