| SP Parallel Programming II Workshop |
| p o s i x t h r e a d s p r o g r a m m i n g |
| Pthreads Overview |
|
| What Is A Thread? | |
| Pthreads Overview |
|
| What Are Pthreads? | |
| Pthreads Overview |
|
| Why Pthreads? | |
For example, the following table compares timing results for the fork() subroutine and the pthreads_create() subroutine. Timings reflect 50,000 process/thread creations, were performed with the timex utility, and units are in seconds.
| Architecture | fork()
| pthread_create()
| real
| user
| sys
| real
| user
| sys
|
| 92.42
| 2.66
| 105.29
| 8.72
| 4.97
| 3.93
|
| 259.21
| 8.84
| 249.26
| 28.79
| 17.82
| 21.03
|
| 55.54
| 2.31
| 43.49
| 5.05
| 2.54
| 2.41
| | ||||
|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|
| Pthreads Overview |
|
| Designing Threaded Programs | |
| The Pthreads API |
|
| Note: IBM's API includes a Fortran interface which may be used for convenience at the price of portability. |
Example
IBM Fortran Pthreads program
| Routine Prefix | Functional Group |
|---|---|
| pthread_ | Threads themselves and miscellaneous subroutines |
| pthread_attr | Thread attributes objects |
| pthread_mutex | Mutexes |
| pthread_mutexattr | Mutex attributes objects. |
| pthread_cond | Condition variables |
| pthread_condattr | Condition attributes objects |
| pthread_key | Thread-specific data keys |
| Thread Management |
|
| Creating Threads | |
| pthread_create (thread,attr,start_routine,arg) |
|
| Check your local implementation for limits on thread creation. For example, AIX 4.2 specifies a maximum of 512 threads per process with a default thread stack size of 56K bytes. |
| Thread Management |
|
| Terminating Thread Execution | |
| pthread_exit (status) |
| Thread Management |
|
| Example: Pthread Creation and Termination | |
The simple example code creates 5 threads with the pthread_create() routine. Each thread prints a "Hello World!" message, and then terminates with a call to pthread_exit().
Example Code
|
#include <pthread.h>
#include <stdio.h>
#define NUM_THREADS 5
void *PrintHello(void *threadid)
{
printf("\n%d: Hello World!\n", threadid);
pthread_exit(NULL);
}
int main()
{
pthread_t threads[NUM_THREADS];
int rc, t;
for(t=0;t < NUM_THREADS;t++){
printf("Creating thread %d\n", t);
rc = pthread_create(&threads[t], NULL, PrintHello, (void *)t);
if (rc){
printf("ERROR; return code from pthread_create() is %d\n", rc);
exit(-1);
}
}
pthread_exit(NULL);
}
|
| Thread Management |
|
| Passing Arguments To Threads | |
| Important: threads initially access their data structures in the parent thread's memory space. That data structure must not be corrupted/modified until the thread has finished accessing it. |
|
Example Code Fragment
|
int *taskids[NUM_THREADS];
for(t=0;t < NUM_THREADS;t++)
{
taskids[t] = (int *) malloc(sizeof(int));
*taskids[t] = t;
printf("Creating thread %d\n", t);
rc = pthread_create(&threads[t], NULL, PrintHello,
(void *) taskids[t]);
...
}
|
|
Example Code Fragment
|
struct thread_data{
int thread_id;
int sum;
char *message;
};
struct thread_data thread_data_array[NUM_THREADS];
void *PrintHello(void *threadarg)
{
struct thread_data *my_data;
...
my_data = (struct thread_data *) threadarg;
taskid = my_data->thread_id;
sum = my_data->sum;
hello_msg = my_data->message;
...
}
int main()
{
...
thread_data_array[t].thread_id = t;
thread_data_array[t].sum = sum;
thread_data_array[t].message = messages[t];
rc = pthread_create(&threads[t], NULL, PrintHello,
(void *) &thread_data_array[t]);
...
}
|
|
Example Code Fragment
|
int rc, t;
for(t=0;t < NUM_THREADS;t++)
{
printf("Creating thread %d\n", t);
rc = pthread_create(&threads[t], NULL, PrintHello,
(void *) &t);
...
}
|
| Thread Management |
|
| Thread Identifiers | |
|
pthread_self ()
pthread_equal (thread1,thread2) |
| Thread Management |
|
| Joining Threads | |
| pthread_join (threadid,status) |
| Thread Management |
|
| Detaching / Undetaching Threads | |
|
pthread_detach (threadid,status)
pthread_attr_init (attr) pthread_attr_setdetachstate (attr,detachstate) pthread_attr_getdetachstate (attr,detachstate) pthread_attr_destroy (attr) |
|
| Current implementations differ in the default detached status at thread creation time. SGI for example, creates threads as joinable (undetached). IBM's AIX 4.2 implementation creates threads as unjoinable (detached) by default, whereas AIX 4.3 creates threads as joinable (undetached). The final standard specifies undetached as the default. |
| Thread Management |
|
| Example: Pthread Joining | |
|
Example Code
|
#include <pthread.h>
#include <stdio.h>
#define NUM_THREADS 3
void *BusyWork(void *null)
{
int i;
double result=0.0;
for (i=0; i < 1000000; i++)
{
result = result + (double)random();
}
printf("result = %d\n",result);
pthread_exit(NULL);
}
void main()
{
pthread_t thread[NUM_THREADS];
pthread_attr_t attr;
int rc, t;
/* Initialize and set thread detached attribute */
pthread_attr_init(&attr);
pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_UNDETACHED);
for(t=0;t < NUM_THREADS;t++)
{
printf("Creating thread %d\n", t);
rc = pthread_create(&thread[t], &attr, BusyWork, NULL);
if (rc)
{
printf("ERROR; return code from pthread_create()
is %d\n", rc);
exit(-1);
}
}
/* Free attribute and wait for the other threads */
pthread_attr_destroy(&attr);
for(t=0;t < NUM_THREADS;t++)
{
rc = pthread_join(thread[t], NULL);
if (rc)
{
printf("ERROR; return code from pthread_join()
is %d\n", rc);
exit(-1);
}
printf("Completed join with thread %d\n",t);
}
pthread_exit(NULL);
}
|
| Mutex Variables |
|
| Overview | |
| Thread 1 | Thread 2 | Balance |
|---|---|---|
| Read balance: $1000 | $1000 | |
| Read balance: $1000 | $1000 | |
| Deposit $200 | $1000 | |
| Deposit $200 | $1000 | |
| Update balance $1000+$200 | $1200 | |
| Update balance $1000+$200 | $1200 |
| Mutex Variables |
|
| Creating / Destroying Mutexes | |
|
pthread_mutex_init (mutex,attr)
pthread_mutex_destroy (mutex) pthread_mutexattr_init (attr) pthread_mutexattr_destroy (attr) |
|
| Note: Your implementation may/may not provide the three optional mutex attributes. For example, in AIX 4.2 and 4.3 mutex attributes are not defined. |
| Mutex Variables |
|
| Locking / Unlocking Mutexes | |
|
pthread_mutex_lock (mutex)
pthread_mutex_trylock (mutex) pthread_mutex_unlock (mutex) |
| Mutex Variables |
|
| Example: Using Mutexes | |
|
Example Code
|
#include <pthread.h>
#include <stdio.h>
#include <malloc.h>
/*
The following structure contains the necessary information
to allow the function "dotprod" to access its input data and
place its output into the structure. This structure is
unchanged from the sequential version.
*/
typedef struct
{
double *a;
double *b;
double sum;
int veclen;
} DOTDATA;
/* Define globally accessible variables and a mutex */
#define NUMTHRDS 4
#define VECLEN 100
DOTDATA dotstr;
pthread_t callThd[NUMTHRDS];
pthread_mutex_t mutexsum;
/*
The function dotprod is activated when the thread is created.
All input to this routine is obtained from a structure
of type DOTDATA and all output from this function is written into
this structure. The benefit of this approach is apparent for the
multi-threaded program: when a thread is created we pass a single
argument to the activated function - typically this argument
is a thread number. All the other information required by the
function is accessed from the globally accessible structure.
*/
void* dotprod(void *arg)
{
/* Define and use local variables for convenience */
int i, start, end, offset, len ;
double mysum, *x, *y;
offset = (int)arg;
len = dotstr.veclen;
start = offset*len;
end = start + len;
x = dotstr.a;
y = dotstr.b;
/*
Perform the dot product and assign result
to the appropriate variable in the structure.
*/
mysum = 0;
for (i=start; i < end ; i++)
{
mysum += (x[i] * y[i]);
}
/*
Lock a mutex prior to updating the value in the shared
structure, and unlock it upon updating.
*/
pthread_mutex_lock (&mutexsum);
dotstr.sum += mysum;
pthread_mutex_unlock (&mutexsum);
pthread_exit((void*)0);
}
/*
The main program creates threads which do all the work and then
print out result upon completion. Before creating the threads,
the input data is created. Since all threads update a shared structure,
we need a mutex for mutual exclusion. The main thread needs to wait for
all threads to complete, it waits for each one of the threads. We specify
a thread attribute value that allow the main thread to join with the
threads it creates. Note also that we free up handles when they are
no longer needed.
*/
void main (int argc, char* argv[])
{
int i;
double *a, *b;
int status;
pthread_attr_t attr;
/* Assign storage and initialize values */
a = (double*) malloc (NUMTHRDS*VECLEN*sizeof(double));
b = (double*) malloc (NUMTHRDS*VECLEN*sizeof(double));
for (i=0; i < VECLEN*NUMTHRDS; i++)
{
a[i]=1;
b[i]=a[i];
}
dotstr.veclen = VECLEN;
dotstr.a = a;
dotstr.b = b;
dotstr.sum=0;
pthread_mutex_init(&mutexsum, NULL);
/* Create threads to perform the dotproduct */
pthread_attr_init(&attr);
pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_UNDETACHED);
for(i=0;i < NUMTHRDS;i++)
{
/*
Each thread works on a different set of data.
The offset is specified by 'i'. The size of
the data for each thread is indicated by VECLEN.
*/
pthread_create( &callThd[i], &attr, dotprod, (void *)i);
}
pthread_attr_destroy(&attr);
/* Wait on the other threads */
for(i=0;i < NUMTHRDS;i++)
{
pthread_join( callThd[i], (void **)&status);
}
/* After joining, print out the results and cleanup */
printf ("Sum = %f \n", dotstr.sum);
free (a);
free (b);
pthread_mutex_destroy(&mutexsum);
pthread_exit (0);
}
Serial version
Pthreads version
|
| Condition Variables |
|
| Overview | |
| Calling Thread | |
|---|---|
| Declare and initialize global data/variables which require synchronization (such as "count") | |
| Declare and initialize a condition variable object | |
| Declare and initialize an associated mutex | |
| Create threads A and B to do work | |
| Thread A | Thread B |
|
|
| Join / Continue | |
| Condition Variables |
|
| Creating / Destroying Condition Variables | |
|
pthread_cond_init (condition,attr)
pthread_cond_destroy (condition) pthread_condattr_init (attr) pthread_condattr_destroy (attr) |
|
| Check your implementation for the ability to use the condition variable attributes object. For example, in AIX 4.2 and 4.3 this option is not implemented and the value "NULL" should be used. |
| Condition Variables |
|
| Waiting / Signalling On Condition Variables | |
|
pthread_cond_wait (condition,mutex)
pthread_cond_signal (condition) pthread_cond_broadcast (condition) |
| Condition Variables |
|
| Example: Using Condition Variables | |
This simple example code demonstrates the use of several Pthread condition variable routines. The main routine creates three threads. Two of the threads perform work and update a "count" variable. The third thread waits until the count variable reaches a specified value.
|
Example Code
|
#include <pthread.h>
#include <stdio.h>
#define NUM_THREADS 3
#define TCOUNT 10
#define COUNT_LIMIT 12
int count = 0;
int thread_ids[3] = {0,1,2};
pthread_mutex_t count_mutex;
pthread_cond_t count_threshold_cv;
void *inc_count(void *idp)
{
int j,i;
double result=0.0;
int *my_id = idp;
for (i=0; i < TCOUNT; i++) {
pthread_mutex_lock(&count_mutex);
count++;
/*
Check the value of count and signal waiting thread when condition is
reached. Note that this occurs while mutex is locked.
*/
if (count == COUNT_LIMIT) {
pthread_cond_signal(&count_threshold_cv);
printf("inc_count(): thread %d, count = %d Threshold reached.\n",
*my_id, count);
}
printf("inc_count(): thread %d, count = %d, unlocking mutex\n",
*my_id, count);
pthread_mutex_unlock(&count_mutex);
/* Do some work so threads can alternate on mutex lock */
for (j=0; j < 1000; j++)
result = result + (double)random();
}
pthread_exit(NULL);
}
void *watch_count(void *idp)
{
int *my_id = idp;
printf("Starting watch_count(): thread %d\n", *my_id);
/*
Lock mutex and wait for signal. Note that the pthread_cond_wait
routine will automatically and atomically unlock mutex while it waits.
Also, note that if COUNT_LIMIT is reached before this routine is run by
the waiting thread, the loop will be skipped to prevent pthread_cond_wait
from never returning.
*/
pthread_mutex_lock(&count_mutex);
while (count < COUNT_LIMIT) {
pthread_cond_wait(&count_threshold_cv, &count_mutex);
printf("watch_count(): thread %d Condition signal
received.\n", *my_id);
}
pthread_mutex_unlock(&count_mutex);
pthread_exit(NULL);
}
void main()
{
int i, rc;
pthread_t threads[3];
pthread_attr_t attr;
/* Initialize mutex and condition variable objects */
pthread_mutex_init(&count_mutex, NULL);
pthread_cond_init (&count_threshold_cv, NULL);
/*
For portability, explicitly create threads in an undetached state
so that they can be joined later.
*/
pthread_attr_init(&attr);
pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_UNDETACHED);
pthread_create(&threads[0], &attr, inc_count, (void *)&thread_ids[0]);
pthread_create(&threads[1], &attr, inc_count, (void *)&thread_ids[1]);
pthread_create(&threads[2], &attr, watch_count, (void *)&thread_ids[2]);
/* Wait for all threads to complete */
for (i = 0; i < NUM_THREADS; i++) {
pthread_join(threads[i], NULL);
}
printf ("Main(): Waited on %d threads. Done.\n", NUM_THREADS);
/* Clean up and exit */
pthread_attr_destroy(&attr);
pthread_mutex_destroy(&count_mutex);
pthread_cond_destroy(&count_threshold_cv);
pthread_exit (NULL);
}
|
| Pthreads, MPI, SMPs, AIX and IBM's PE |
|
| Compiler Command | Description |
|---|---|
|
xlc_r cc_r | C pthreads compiler with default language level of ANSI |
| xlf_r | Fortran compiler with IBM Pthreads API (non-portable). |
| xlf90_r | Fortran 90 pthreads compiler |
| xlC_r | C++ pthreads compiler |
| mpcc_r | C MPI-pthreads compiler script |
| mpxlf_r | Fortran MPI-pthreads compiler script |
| mpCC_r | C++ MPI-pthreads compiler script |
| Architecture | MPI
| pthreads
|
| 22.54
| 0.11
|
| 92.14
| 0.31
| |
|---|
| Pthread Library Routines Reference |
|
| Pthread Functions | |
| Thread Management | pthread_create |
| pthread_exit | |
| pthread_join | |
| pthread_once | |
| pthread_kill | |
| pthread_self | |
| pthread_equal | |
| pthread_yield | |
| pthread_detach | |
| Thread Specific Data | pthread_key_create |
| pthread_key_delete | |
| pthread_getspecific | |
| pthread_setspecific | |
| Thread Cancellation | pthread_cancel |
| pthread_cleanup_pop | |
| pthread_cleanup_push | |
| pthread_setcancelstate | |
| pthread_getcancelstate | |
| pthread_testcancel | |
| Thread Scheduling | pthread_getschedparam |
| pthread_setschedparam | |
| Signals | pthread_sigmask |
| Pthread Attribute Functions | |
| Basic Management | pthread_attr_init |
| pthread_attr_destroy | |
| Detachable or Joinable | pthread_attr_setdetachstate |
| pthread_attr_getdetachstate | |
| Specifying Stack Information | pthread_attr_getstackaddr |
| pthread_attr_getstacksize | |
| pthread_attr_setstackaddr | |
| pthread_attr_setstacksize | |
| Thread Scheduling Attributes | pthread_attr_getschedparam |
| pthread_attr_setschedparam | |
| pthread_attr_getschedpolicy | |
| pthread_attr_setschedpolicy | |
| pthread_attr_setinheritsched | |
| pthread_attr_getinheritsched | |
| pthread_attr_setscope | |
| pthread_attr_getscope | |
| Mutex Functions | |
| Mutex Management | pthread_mutex_init |
| pthread_mutex_destroy | |
| pthread_mutex_lock | |
| pthread_mutex_unlock | |
| pthread_mutex_trylock | |
| Priority Management | pthread_mutex_setprioceiling |
| pthread_mutex_getprioceiling | |
| Mutex Attribute Functions | |
| Basic Management | pthread_mutexattr_init |
| pthread_mutexattr_destroy | |
| Sharing | pthread_mutexattr_getpshared |
| pthread_mutexattr_setpshared | |
| Protocol Attributes | pthread_mutexattr_getprotocol |
| pthread_mutexattr_setprotocol | |
| Priority Management | pthread_mutexattr_setprioceiling |
| pthread_mutexattr_getprioceiling | |
| Condition Variable Functions | |
| Basic Management | pthread_cond_init |
| pthread_cond_destroy | |
| pthread_cond_signal | |
| pthread_cond_broadcast | |
| pthread_cond_wait | |
| pthread_cond_timedwait | |
| Condition Variable Attribute Functions | |
| Basic Management | pthread_condattr_init |
| pthread_condattr_destroy | |
| Sharing | pthread_condattr_getpshared |
| pthread_condattr_setpshared | |
| References and More Information |
|