Pthreads

Pthreads API

1: Thread Management

Create Threads

pthread_create(thread, attr, startRoutine, arg)
// thread - unique identifier for the new thread (pthread_t)
// attr - attr object used to set thread attributes (pthread_attr) - you can specify a thread attributes object, or NULL for the default values
// startRoutine - C routine that the thread will execute 
// arg - single arg that may be passed to startRoutine - it must be passed by reference (pointer to struct) and NULL may be used if no arg is to be passed

/*
    If successful, the pthread_create() function shall return zero; otherwise, an error number shall be returned to indicate the error
 */
copy

Thread Attributes

By default, a thread is created with certain attributes.

pthread_attr_init(attr) and pthread_attr_destroy(attr) are used to initialize/destroy the thread attribute object.

Other routines are then used to query/set specific attributes in the thread attribute object.

Terminating Thread

Routine: pthread_exit(status)

Cleanup: pthread_exit() does not close files; any files opened inside the thread will remain open after the thread is terminated.

Example

#include <pthread.h>
#include <stdio.h>
#include <stdlib.h>
#define NUM_THREADS 5

void *PrintHello(void *threadid) {
    int *tid;
    tid = (int *)threadid;
    printf("Hello World! It's me, thread #%d!\n", *tid);
    pthread_exit(NULL);
}

int main(int argc, char *argv[]) {
    pthread_t threads[NUM_THREADS];
    int rc, t, tids[NUM_THREADS];
    for (t=0; t< NUM_THREADS; t++) {
        printf("In main: creating thread %d\n", t);
        tids[t] = t;
        rc = pthread_create(&threads[t], NULL, PrintHello, (void *)&tids[t]);

        if (rc) {
            printf("ERROR; return code from pthread_create() is %d\n", rc);
            exit(-1);
        }
    }
    pthread_exit(NULL); 
}
copy

Passing Arguments to Threads

pthread_create() routine permits the programmer to pass one argument to the thread start routine.

For cases where multiple args must be passed, we can create a struct and use the reference pointer as an arg.

All args passed by reference must be cast to (void *)

struct two_args {
    int arg1;
    int arg2;   
};

void *needs_2_args(void *ap) {
    struct two_args *argp;
    int a1, a2;

    argp = (struct two_args *) ap;

    // do stuff here
    
    a1 = argp->arg1;
    a2 = argp->arg2;
    
    // do stuff here 

    free(argp);
    pthread_exit(NULL);
}

int main(int argc, char *argv[]) {
    pthread_t t;
    struct two_args *ap;
    int rc;

    // do stuff here 

    ap = (struct two_args *)malloc(sizeof(struct two_args));
    ap->arg1 = 1;
    ap->arg2 = 2;
    rc = pthread_create(&t, NULL, needs_2_args, (void *) ap);

    // do stuff here 

    pthread_exit(NULL);
}
copy

Joining and Detatching Threads

Routines

To explicitly create a thread as joinable or detached, the attr argument in the pthread_create() routine is used:

Example

void *BusyWork(void *null) {
    // do stuff 
    pthread_exit((void *) 0);
}

int main(int argc, char *argv[]) {
    pthread_attr_t attr;
    int rc, t;
    void *status;

    /* init and set thread detached attribute */
    pthread_attr_init(&attr);
    pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_JOINABLE);

    /* free attribute and wait for the other threads */
    pthread_attr_destory(&attr);
    for (t=0; t< NUM_THREADS; t++) {
        rc = pthread_join(thread[t], &status);
        // do stuff 
        printf("Completed join with thred %d status = %ld\n", t, (long)status);
    }
    pthread_exit(NULL);
}
copy

Syncronisation Issues

When multiple threads attempt to manipulate the same data item, the results can often be incoherent if proper care is not take ie. race conditions.

2: Mutexes

The second class of functions deal with synchronization - called a "mutex", which is an abbreviation for mutual exclusion.

Creating and Destroying Mutexes

Routines

pthread_mutex_init(mutex, attr) pthread_mutex_destroy(mutex) pthread_mutexattr_init(attr) pthread_mutexattr_destroy(attr)

A mutex must be declared with type pthread_mutex_t, and must be initialized before they can be used.

There are two ways to init a mutex variable:

The mutex is initially unlocked.

Locking & Unlocking Mutexes

Routines

pthread_mutex_lock(mutex) pthread_mutex_unlock(mutex) pthread_mutex_trylock(mutex)

Example 1

We can now write our previously incorrect code segment as...

pthread_mutex_t min_value_lock;

main() {
    ...
    pthread_mutex_init(&min_value_lock, NULL);
    ...
}

void *find_min(void *list_ptr) {
    ...
    pthread_mutex_lock(&min_value_lock);
    if (my_cost < best_cost) {
        best_cost = my_cost;
    }   
    pthread_mutex_unlock(&min_value_lock);
}
copy

Example 2

The producer-consumer scenario imposes the following constraints:

pthread_mutex_t task_queue_lock;
int task_available;

main() {
    task_available = 0;
    pthread_mutex_init(&task_queue_lock, NULL);
}

void *producer(void *producer_thread_data) {
    ...
    while (!done()) {
        inserted = 0;
        create_task(&my_task);
        while (inserted == 0) {
            pthread_mutex_lock(&task_queue_lock);
            if (task_available == 0) {
                insert_into_queue(my_task);
                task_available = 1;
                inserted = 1;
            }
            pthread_mutex_unlock(&task_queue_lock);
        }
    }
}

void *consumer(void *consumer_thread_data) {
    ...
    while (!done()) {
        extracted = 0;
        while (extracted == 0) {
            pthread_mutex_lock(&task_queue_lock);
            if (task_available == 1) {
                extract_from_queue(&my_task);
                task_available = 0;
                extracted = 1;
            }
            pthread_mutex_unlock(&task_queue_lock);
        }
        process_task(my_task);
    }
}
copy

Overheads of Locking

Alleviating Locking Overhead

pthread_mutex_t tryLock_lock = PTHREAD_MUTEX_INITIALIZER;

lock_status = pthread_mutex_trylock(&tryLock_lock)
if (lock_status == EBUSY) {
    /* do something else */
    ...
} else {
    /* do one thing */
    ...
    pthread_mutex_unlock(&tryLock_lock);
}
copy

Monitors

Mutexes provide powerful sync tools, but...

A monitor is a high-level abstraction that may provide a convenient and effective mechanism for thread synchronization.

Monitor and Condition Variables

Condition Variables

Create & Destroying Cond Variables

Routines

pthread_cond_init(condition, attr) pthread_cond_destroy(condition) pthread_condattr_init(attr) pthread_condattr_destroy(attr)

Condition variables must be declared with type pthread_cont_t, and must be initialized before they can be used.

2 Ways to declare:

Waiting and Signaling on Condition Variable

pthread_cond_signal() is used to signal (or wake up) another thread which is waiting on the condition variable and should be called after the mutex is locked.

It must unlock mutex in order for pthread_cond_wait() routine to complete.

pthread_cond_broadcast() routine unlocks all of the threads blocked on the condition variable.

Producer-Consumer Using Condition Variables

pthread_cond_t cond_queue_empty, cond_queue_full;
pthread_mutex_t task_queue_cond_lock;
int task_available;
// other data structures here 

main() {
    // declarations and initializations 
    task_available = 0;
    pthread_cond_init(&cond_queue_empty, NULL);
    pthread_cond_init(&cond_queue_full, NULL);
    pthread_mutex_init(&task_queue_cond_lock, NULL);
    // create and join producer and consumer threads
}

void *producer(void *producer_thread_data) {
    while(!done()) {
        create_task();
        pthread_mutex_lock(&task_queue_cond_lock);
        
        while (task_available == 1) {
            pthread_cond_wait(&cond_queue_empty,
                &task_queue_cond_lock);
        }

        insert_into_queue();
        task_available = 1;
        pthread_cond_signal(&cond_queue_full);
        pthread_mutex_unlock(&task_queue_cond_lock);
    }
}

void *consumer(void *consumer_thread_data) {
    while(!done()) {
        pthread_mutex_lock(&task_queue_cond_lock);
        while (task_available == 0) {
            pthread_cond_wait(&cond_queue_full,
                &task_queue_cond_lock);
        }

        my_task = extract_from_queue();
        task_available = 0;
        pthread_cond_signal(&cond_queue_empty);
        pthread_mutex_unlock(&task_queue_cond_lock);
        process_task(my_task);
    }
}
copy