Tutorial 3 - Threads, mutexes and signals #

Introduction notes:

• Quick look at this material will not suffice, you should compile and run all the programs, check how they work, read additional materials like man pages. As you read the material please do all the exercises and questions. At the end you will find sample task similar to the one you will do during the labs, please do it at home.
• You will find additional information in yellow sections, questions and tasks in blue ones. Under the question you will find the answer, to see it you have to click. Please try to answer on you own before checking.
• Full programs’ codes are placed as attachments at the bottom of this page. On this page only vital parts of the code are displayed
• Codes, information and tasks are organized in logical sequence, in order to fully understand it you should follow this sequence. Sometimes former task makes context for the next one and it is harder to comprehend it without the study of previous parts.
• Most of exercises require command line to practice, I usually assume that all the files are placed in the current working folder and that we do not need to add path parts to file names.
• What you learn and practice in this tutorial will be required for the next ones. If you have a problem with this material after the graded lab you can still ask teachers for help.
• This tutorial is based on student’s examples, it has some minor flaws in parts not related to the threads (I left them on purpose and explain them in comments). Some repeating flaws not mentioned in tasks are:
  • Main functions are too long, it should be easy to split them
  • Bool data type requires extra header and can be easily avoided by using integers (0 and 1)
  • So called “magic numbers” are left in code - those numerical constants should be replaced with “C” preprocessor macro

Thread Management #

Creating Threads #

A thread is created using the pthread_create command. Let’s take a look at the declaration of this function:

int pthread_create(pthread_t *restrict thread,
                  const pthread_attr_t *restrict attr,
                  void *(*start_routine)(void*),
                  void *restrict arg);

Unlike fork, this function returns an integer that is used to determine whether it has succeeded or failed. On success, it writes the newly created thread’s handle to thread.

The second argument is for attributes. It can be left NULL to use the default attributes. (See man 3p pthread_attr_destroy and man -k pthread_attr_set for additional info)

When creating a thread, you have to supply a function pointer to a function of the following signature:

void *function(void *args);

This function is the function that will be executed by the newly created thread. You pass the arguments to the function via the last argument arg.

In C, a void* is a pointer to anything. You can not refer to the underlying value without casting to the actual type, e.g int*.

pthread_t is implementation defined and should be treated as such. (i.e you can’t use it like an integer)

More information:

man 3p pthread_create

Thread Joining #

Much like child processes, a thread should also be joined after it finishes its work. This can be achieved using the pthread_join command.

int pthread_join(pthread_t thread, void **value_ptr);

pthread_join functions similarly to wait, blocking until the thread finishes. POSIX doesn’t define a non-blocking join function.

Like pthread_create, this function returns an integer to indicate success or failure. The thread function’s return value is written to value_ptr if non-NULL storage was provided.

Unlike with processes, you have to precisely specify which thread you’re waiting for by providing a valid thread identifier.

More information:

man 3p pthread_join

Excercise #

Goal: Write a program to approximate PI value with Monte Carlo method, program takes the following parameters:

• k … number of threads used to approximate,
• n … number of random tries of each thread.

Each thread (except for the main) should conduct its own estimation, at the end the main thread gathers the results from co-threads and calculate the average value from all the sub-results.

What you need to know:

• man 7 pthreads
• man 3p pthread_create
• man 3p pthread_join
• man 3p rand (especially rand_r)
• Monte-Carlo method, in paragraph “Monte Carlo methods” on this site.

Solution #

Makefile:

CC=gcc
CFLAGS=-std=gnu99 -Wall -fsanitize=address,undefined
LDFLAGS=-fsanitize=address,undefined
LDLIBS=-lpthread

Flag -lpthread is mandatory for all compilations in this tutorial. Linked library is called libpthread.so (after -l we write the name of the library without first “lib” part)

prog17.c:

#include <pthread.h>
#include <stdio.h>
#include <stdlib.h>
#include <time.h>

#define ERR(source) (perror(source), fprintf(stderr, "%s:%d\n", __FILE__, __LINE__), exit(EXIT_FAILURE))

typedef struct targs
{
    pthread_t tid;
    unsigned int seed;
    int samples_count;
} targs_t;

void* pi_estimation(void* args);

void usage(char* pname)
{
    fprintf(stderr, "Usage: %s [num_threads > 0] [num_samples > 0]\n", pname);
    exit(EXIT_FAILURE);
}

void create_threads(int thread_count, int samples_count, targs_t** estimations)
{
    *estimations = malloc(sizeof(targs_t) * thread_count);
    if (estimations == NULL)
        ERR("malloc");

    targs_t* targs = *estimations;

    srand(time(NULL));
    for (int i = 0; i < thread_count; i++)
    {
        targs[i].seed = rand();
        targs[i].samples_count = samples_count;

        if (pthread_create(&(targs[i].tid), NULL, pi_estimation, &targs[i]) != 0)
            ERR("pthread_create");
    }
}

int main(int argc, char** argv)
{
    if (argc != 3)
        usage(argv[0]);

    int thread_count = atoi(argv[1]);
    int samples_count = atoi(argv[2]);

    if (thread_count <= 0 || samples_count <= 0)
        usage(argv[0]);

    targs_t* estimations;
    create_threads(thread_count, samples_count, &estimations);

    int* subresult = NULL;
    int cumulative_result = 0;
    for (int i = 0; i < thread_count; i++)
    {
        if (pthread_join(estimations[i].tid, (void**)&subresult) != 0)
            ERR("pthread_join");

        if (NULL != subresult)
        {
            cumulative_result += *subresult;
            free(subresult);
        }
    }

    const double sum_avg = (double)cumulative_result / (double)thread_count;
    const double pi_estimate = (2.0 * (double)samples_count) / (sum_avg * sum_avg);
    printf("Estimated value of PI is %f\n", pi_estimate);

    free(estimations);
}

void* pi_estimation(void* void_ptr)
{
    targs_t* args = void_ptr;
    int* result = malloc(sizeof(int));
    if (result == NULL)
        ERR("malloc");

    int inner_count = 0;
    for (int i = 0; i < args->samples_count; i++)
    {
        inner_count += (rand_r(&args->seed) % 2) ? 1 : -1;
    }

    *result = abs(inner_count);

    return result;
}

Notes and questions #

Functions’ declarations at the beginning of the code (not the functions definitions) are quite useful, sometimes mandatory. If you do not know the difference please read this.

In multi threaded processes you can not correctly use rand() function, use rand_r() instead. The later one requires individual seed for every thread.

This program uses the simplest schema for threads lifetime. It creates some threads and then immediately waits for them to finish. More complex scenarios are possible

Please keep in mind that nearly every call to system function (and most calls to library functions) should be followed with the test on errors and if necessary by the proper reaction on the error.

ERR macro does not send “kill” signal as in multi-process program, why?

How is input data passed to the new threads?

Is the thread input data shared between the threads?

How the random seed for rand_r() is prepared for each thread?

In this multi thread program we see srand/rand calls, is this correct? It contradicts what was said a few lines above.

Can we share one input data structure for all the threads instead of having a copy for every thread?

Can we make the array with the thread input data automatic variable (not allocated)?

Why do we need to release the memory returned from the working thread?

Why can’t we return the data as the address of local (to the thread) automatic variable?

Can we avoid memory allocation in the working thread?

Detachable threads & Synchronization #

Overview #

A detached thread is a thread that you need not join. While it’s convenient, it also has downsides, such as not knowing when thread execution ends, as it is not joinable. Because of that, you can’t be sure that a thread has actually terminated, which can lead to problems when exiting from the main thread.

Another issue with detached threads is not being able to obtain anything from them without extra synchronization.

There are two ways to obtain a detached thread, those being either spawning a detached thread or manually detaching an already running thread.

Spawning a detached thread #

In order to spawn a detached thread, you must first create a pthread_attr_t object that will later be passed to the pthread_create function. In order to initialize one to the default state, you must call pthread_attr_init. Then, you have to set its detached state to PTHREAD_CREATE_DETACHED using the pthread_attr_setdetachstate function. Since you have to initialize the attributes structure, you also have to destroy it later using pthread_attr_destroy.

More information:

man 3p pthread_attr_destroy
man 3p pthread_attr_getdetachstate
man 3p pthread_create

Detaching a running thread #

As mentioned above, you may also detach a running thread. You can do it by calling pthread_detach with the thread’s handle. A thread may detach itself in this way by first obtaining its own handle via a pthread_self call, and then calling pthread_detach. Attempting to detach a thread that’s already detached results in undefined behavior and shouldn’t be done.

More information:

man 3p pthread_detach
man 3p pthread_self

Mutual exclusion #

Since a detached thread on its own is effectively useless, we are introducing shared state between the threads. It means that two or more threads have to work on the same structure. This is problematic, as we currently have no way of ensuring that a thread won’t write to a field that is currently being read by another thread. (and vice versa)

This is where mutexes come into play. In short - a mutex is a special object that prevents two or more threads from holding it at the same time. It is recommended to read more about mutexes if you don’t know what a mutex is.

You initialize a mutex by calling the pthread_mutex_init function. You would later destroy it by calling the pthread_mutex_destroy function.

int pthread_mutex_init(pthread_mutex_t *restrict mutex,
   const pthread_mutexattr_t *restrict attr);

Like pthread_create, pthread_mutex_init also accepts an attributes object. Similarly, it can be NULL to use the default attributes. (See man 3p pthread_mutexattr_destroy and man -k pthread_mutexattr_set for additional info)

Just creating a mutex isn’t enough, you also need to acquire it when accessing shared state. That can be done using pthread_mutex_lock. This function blocks until it can acquire the mutex, then tries to acquire it. While it can fail, its failure usually indicates serious memory corruption or very specyfic situation (more on this on OPS2), and thus the return value of this function won’t be checked in the further examples and we don’t require it during labs.

After you have acquired the mutex, you should not attempt to re-acquire it before releasing. This results in a deadlock in the majority of cases.

After you are done modifying the shared state, you should release the mutex by calling pthread_mutex_unlock, so that another thread can lock it. Ideally, you should minimize the time you’re holding the mutex for, as it slows down other threads that use it.

Please note that you should NEVER make copies of a mutex (e.g pthread_mutex_t mtx1_copy = mtx1). POSIX forbids that. A copy of a mutex does not have to be a working mutex! Even if it would work, it should be quite obvious, that a copy would be a different mutex.

More information:

man 3p pthread_mutex_destroy
man 3p pthread_mutex_lock

Excercise #

Goal: Write binomial distribution visualization program, use Bean Machine (Galton board) method with 11 bins for bens. The program takes two parameters:

• k … number of beans trowing threads
• n … total number of beans to throw

Each thread throws beans separately from others, after each throw, number of beans in bins must me updated. Main thread at each second checks if the simulation has completed (not using thread_join). At the end main thread prints the content of the bins and the mean value of beans.

What you need to know:

• man 3p pthread_mutex_destroy (full description)
• man 3p pthread_mutex_lock
• man 3p pthread_mutex_unlock
• man 3p pthread_detach
• man 3p pthread_attr_init
• man 3p pthread_attr_destroy
• man 3p pthread_attr_setdetachstate
• man 3p pthread_attr_getdetachstate
• Bean machine on this site.

Solution #

prog18.c:

#include <pthread.h>
#include <stdio.h>
#include <stdlib.h>
#include <time.h>
#include <unistd.h>

#define BIN_COUNT 11
#define NEXT_DOUBLE(seedptr) ((double)rand_r(seedptr) / (double)RAND_MAX)
#define ERR(source) (perror(source), fprintf(stderr, "%s:%d\n", __FILE__, __LINE__), exit(EXIT_FAILURE))

typedef struct shared_state
{
    int balls_waiting;
    int bins[BIN_COUNT];
    int balls_thrown;
    pthread_mutex_t bins_mtx[BIN_COUNT];
    pthread_mutex_t balls_waiting_mtx;
    pthread_mutex_t balls_thrown_mtx;
} shared_state_t;

typedef struct thrower_args
{
    pthread_t tid;
    unsigned int seed;
    shared_state_t* shared;
} thrower_args_t;

void make_throwers(thrower_args_t** args_ptr, int num_throwers, shared_state_t* shared);
void* throwing_func(void* args);
int throw_ball(unsigned int* seedptr);

void usage(char* pname)
{
    fprintf(stderr, "Usage: %s [num_balls>0] [num_throwers>0]", pname);
    exit(EXIT_FAILURE);
}

int main(int argc, char** argv)
{
    int balls_count, throwers_count;
    if (argc != 3)
        usage(argv[0]);

    balls_count = atoi(argv[1]);
    throwers_count = atoi(argv[2]);

    if (balls_count <= 0 || throwers_count <= 0)
        usage(argv[0]);

    shared_state_t shared = {
        .balls_waiting = balls_count,
        .balls_thrown = 0,
        .balls_waiting_mtx = PTHREAD_MUTEX_INITIALIZER,
        .balls_thrown_mtx = PTHREAD_MUTEX_INITIALIZER,
        .bins = {0},
    };

    int balls_thrown = shared.balls_thrown;

    for (int i = 0; i < BIN_COUNT; i++)
    {
        if (pthread_mutex_init(&shared.bins_mtx[i], NULL))
            ERR("pthread_mutex_init");
    }

    thrower_args_t* args;
    make_throwers(&args, throwers_count, &shared);

    while (balls_thrown < balls_count)
    {
        sleep(1);
        pthread_mutex_lock(&shared.balls_thrown_mtx);
        balls_thrown = shared.balls_thrown;
        pthread_mutex_unlock(&shared.balls_thrown_mtx);
    }

    int final_balls_count = 0;
    double mean_val = 0.0;
    for (int i = 0; i < BIN_COUNT; i++)
    {
        final_balls_count += shared.bins[i];
        mean_val += shared.bins[i] * i;
    }
    mean_val = mean_val / final_balls_count;

    printf("Bins count:\n");
    for (int i = 0; i < BIN_COUNT; i++)
        printf("%d\t", shared.bins[i]);
    printf("\nTotal balls count : %d\nMean value: %f\n", final_balls_count, mean_val);
    // for (int i = 0; i < BIN_COUNT; i++) pthread_mutex_destroy(&mxBins[i]);
    // free(args);
    // The resources used by detached threads cannod be freed as we are not sure
    // if they are running yet.
    exit(EXIT_SUCCESS);
}

void make_throwers(thrower_args_t** args_ptr, int throwers_count, shared_state_t* shared)
{
    *args_ptr = malloc(sizeof(thrower_args_t) * throwers_count);
    thrower_args_t* args = *args_ptr;
    if (args == NULL)
        ERR("malloc");

    pthread_attr_t attr;

    if (pthread_attr_init(&attr))
        ERR("pthread_attr_init");
    if (pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_DETACHED))
        ERR("pthread_attr_setdetachstate");

    srand(time(NULL));
    for (int i = 0; i < throwers_count; i++)
    {
        args[i].seed = rand();
        args[i].shared = shared;
        if (pthread_create(&args[i].tid, &attr, throwing_func, &args[i]))
            ERR("pthread_create");
    }

    pthread_attr_destroy(&attr);
}

void* throwing_func(void* void_args)
{
    thrower_args_t* args = void_args;
    while (1)
    {
        pthread_mutex_lock(&args->shared->balls_waiting_mtx);
        if (args->shared->balls_waiting > 0)
        {
            args->shared->balls_waiting--;
            pthread_mutex_unlock(&args->shared->balls_waiting_mtx);
        }
        else
        {
            pthread_mutex_unlock(&args->shared->balls_waiting_mtx);
            break;
        }

        int binno = throw_ball(&args->seed);
        pthread_mutex_lock(&args->shared->bins_mtx[binno]);
        args->shared->bins[binno]++;
        pthread_mutex_unlock(&args->shared->bins_mtx[binno]);
        pthread_mutex_lock(&args->shared->balls_thrown_mtx);
        args->shared->balls_thrown++;
        pthread_mutex_unlock(&args->shared->balls_thrown_mtx);
    }
    return NULL;
}

/* returns # of bin where ball has landed */
int throw_ball(unsigned int* seedptr)
{
    int result = 0;
    for (int i = 0; i < BIN_COUNT - 1; i++)
        if (NEXT_DOUBLE(seedptr) > 0.5)
            result++;
    return result;
}

Notes and questions #

Once again, all thread input data is passed as pointer to the structure thrower_args_t, treads results modify bins array ( pointer in the same structure), no global variables used.

In this code two mutexes protect two counters and an array of mutexes protects the bins’ array (one mutex for every cell in the array). In total we have BIN_COUNT+2 mutexes.

In this program we use detachable threads. There is no need (nor option) to wait for working threads to finish, thus the lack of pthread_join. As we do not join the threads, a different method must be deployed to test if the main program can exit.

In this example mutexes are created in two ways - as automatic and dynamic variables. The first method is simpler in coding but you need to know exactly how many mutexes you need at coding time. The dynamic creation requires more coding (initiation and removal) but the amount of mutexes also is dynamic.

Is data passed to threads in thrower_args_t structure shared between them?

Why do we use a pointer to the shared_state_t structure?

This program uses a lot of mutexes, can we reduce the number of them?

To check if the working threads terminated, the main threads periodically checks if the numer of thrown beans is equal to the number of beans in total. Is this optimal solution?

Do all the threads created in this program really work?

Threads and Signals #

Excercise #

Goal: The program takes sole ‘k’ parameter and prints the list of numbers form 1 to k at each second. It must handle two signals in dedicated thread, the following action must be taken upon the signal arrival:

• SIGINT (C-c) … removes random number from the list (do nothing if empty),
• SIGQUIT (C-\) … set program ‘STOP’ flag (both threads end).

What you need to know:

• man 3p pthread_sigmask
• man 3p sigwait

prog19.c

#include <errno.h>
#include <pthread.h>
#include <signal.h>
#include <stdarg.h>
#include <stdbool.h>
#include <stddef.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>

#define MAXLINE 4096
#define DEFAULT_ARRAYSIZE 10
#define DELETED_ITEM -1
#define ERR(source) (perror(source), fprintf(stderr, "%s:%d\n", __FILE__, __LINE__), exit(EXIT_FAILURE))

typedef struct argsSignalHandler
{
    pthread_t tid;
    int *pArrayCount;
    int *array;
    pthread_mutex_t *pmxArray;
    sigset_t *pMask;
    bool *pQuitFlag;
    pthread_mutex_t *pmxQuitFlag;
} argsSignalHandler_t;

void ReadArguments(int argc, char **argv, int *arraySize);
void removeItem(int *array, int *arrayCount, int index);
void printArray(int *array, int arraySize);
void *signal_handling(void *);

int main(int argc, char **argv)
{
    int arraySize, *array;
    bool quitFlag = false;
    pthread_mutex_t mxQuitFlag = PTHREAD_MUTEX_INITIALIZER;
    pthread_mutex_t mxArray = PTHREAD_MUTEX_INITIALIZER;
    ReadArguments(argc, argv, &arraySize);
    int arrayCount = arraySize;
    if (NULL == (array = (int *)malloc(sizeof(int) * arraySize)))
        ERR("Malloc error for array!");
    for (int i = 0; i < arraySize; i++)
        array[i] = i + 1;
    sigset_t oldMask, newMask;
    sigemptyset(&newMask);
    sigaddset(&newMask, SIGINT);
    sigaddset(&newMask, SIGQUIT);
    if (pthread_sigmask(SIG_BLOCK, &newMask, &oldMask))
        ERR("SIG_BLOCK error");
    argsSignalHandler_t args;
    args.pArrayCount = &arrayCount;
    args.array = array;
    args.pmxArray = &mxArray;
    args.pMask = &newMask;
    args.pQuitFlag = &quitFlag;
    args.pmxQuitFlag = &mxQuitFlag;
    if (pthread_create(&args.tid, NULL, signal_handling, &args))
        ERR("Couldn't create signal handling thread!");
    while (true)
    {
        pthread_mutex_lock(&mxQuitFlag);
        if (quitFlag == true)
        {
            pthread_mutex_unlock(&mxQuitFlag);
            break;
        }
        else
        {
            pthread_mutex_unlock(&mxQuitFlag);
            pthread_mutex_lock(&mxArray);
            printArray(array, arraySize);
            pthread_mutex_unlock(&mxArray);
            sleep(1);
        }
    }
    if (pthread_join(args.tid, NULL))
        ERR("Can't join with 'signal handling' thread");
    free(array);
    if (pthread_sigmask(SIG_UNBLOCK, &newMask, &oldMask))
        ERR("SIG_BLOCK error");
    exit(EXIT_SUCCESS);
}

void ReadArguments(int argc, char **argv, int *arraySize)
{
    *arraySize = DEFAULT_ARRAYSIZE;

    if (argc >= 2)
    {
        *arraySize = atoi(argv[1]);
        if (*arraySize <= 0)
        {
            printf("Invalid value for 'array size'");
            exit(EXIT_FAILURE);
        }
    }
}

void removeItem(int *array, int *arrayCount, int index)
{
    int curIndex = -1;
    int i = -1;
    while (curIndex != index)
    {
        i++;
        if (array[i] != DELETED_ITEM)
            curIndex++;
    }
    array[i] = DELETED_ITEM;
    *arrayCount -= 1;
}

void printArray(int *array, int arraySize)
{
    printf("[");
    for (int i = 0; i < arraySize; i++)
        if (array[i] != DELETED_ITEM)
            printf(" %d", array[i]);
    printf(" ]\n");
}

void *signal_handling(void *voidArgs)
{
    argsSignalHandler_t *args = voidArgs;
    int signo;
    srand(time(NULL));
    for (;;)
    {
        if (sigwait(args->pMask, &signo))
            ERR("sigwait failed.");
        switch (signo)
        {
            case SIGINT:
                pthread_mutex_lock(args->pmxArray);
                if (*args->pArrayCount > 0)
                    removeItem(args->array, args->pArrayCount, rand() % (*args->pArrayCount));
                pthread_mutex_unlock(args->pmxArray);
                break;
            case SIGQUIT:
                pthread_mutex_lock(args->pmxQuitFlag);
                *args->pQuitFlag = true;
                pthread_mutex_unlock(args->pmxQuitFlag);
                return NULL;
            default:
                printf("unexpected signal %d\n", signo);
                exit(1);
        }
    }
    return NULL;
}

Notes and questions #

Thread input structure argsSignalHandler_t holds the shared threads data (an array and STOP flag) with protective mutexes and not shared (signal mask and tid of thread designated to handle the signals).

In threaded process (one that has more that one thread) you can not use sigprocmask, use pthread_sigmask instead.

Having separated thread to handle the signals (as in this example) is a very common way to deal with signals in multi-threaded code.

How many threads run in this program?

Name differences and similarities between sigwait i sigsuspend:

After successful call to sigwait only one type of pending signal is removed from the pending signals vector thus the problem we experienced with sigsuspend in L2 example can be corrected when using sigwait instead of sigsuspend. Please correct the program in L2 as an exercise.

Does the method of waiting for the end of working thread have the same flow as the method in previous example?

Can we use sigprocmask instead of pthread_sigmask in this program?

Why system calls to functions operating on mutex (acquire, release) are not tested for errors?

Thread cancelation #

Canceling a thread #

A thread can be canceled using the pthread_cancel function. This is useful when the program has to exit before it finishes its work, for example due to a signal.

More information:

man 3p pthread_cancel

Setting cancelability #

A thread can choose how it will respond to cancelation requests using the following functions:

int pthread_setcancelstate(int state, int *oldstate);
int pthread_setcanceltype(int type, int *oldtype);

Cancelation requests can be ignored by setting the cancel state to PTHREAD_CANCEL_DISABLE and then later re-enabled by setting it to PTHREAD_CANCEL_ENABLE. Cancelation is enabled by default.

The cancel type determines when the thread exits upon receiving a cancelation request. It is set to PTHREAD_CANCEL_DEFERRED by default, in which case the thread exits at a cancelation point. If it is set to PTHREAD_CANCEL_ASYNCHRONOUS, the thread will exit as soon as possible.

Generally, it is safer to keep the cancel type as PTHREAD_CANCEL_DEFERRED, as you may have an unpreventable race condition in the case of asynchronous cancelation (e.g a thread gets canceled after you lock a mutex but before you add a cleanup function)

More information:

man 3p pthread_setcancelstate
man 7 pthreads (specifically the "Cancelation Points" section)

Cleanup functions #

In a lot of cases, exiting a thread mid-execution leaks resources. This is problematic, especially if a thread is holding a mutex - as it won’t be released.

To solve this issue, POSIX allows creating cleanup handlers that run when the thread has to exit abruptly. (i.e due to being cancelled or calling pthread_exit)

Those take form of a function with the following signature:

void function(void* arg);

You can add this function as a cleanup handle by issuing the pthread_cleanup_push call. Let’s take a look at the function declaration.

void pthread_cleanup_push(void (*routine)(void*), void *arg);

As you can see, the function accepts the aforementioned function along with the argument that will be passed to it.

When a thread exits abruptly, all of the cleanup functions are executed in the reverse order of their addition.

You also can remove cleanup handlers by issuing pthread_cleanup_pop.

void pthread_cleanup_pop(int execute);

This function removes the handlers in the same order as they would be executed. The execute parameter allows for optional execution of the handler.

For every instance of pthread_cleanup_push in a scope there must be an instance of pthread_cleanup_pop.

More information:

man 3p pthread_cleanup_pop

Joining canceled threads #

Canceled threads remain joinable, since a cancel request isn’t guaranteed to be processed.

If a thread gets canceled, the value pointer passed to pthread_join will be set to PTHREAD_CANCELED

More information:

man 3p pthread_join
man 3p pthread_exit (specifically the last paragraph of the RATIONALE section)

Excercise #

Goal: Program simulates the faith of MiNI students, it takes the following parameter:

• n <= 100 … count of new students

The program stores the counters of students studding on year 1,2,3 and the final BSc year.

Main Thread: Initiate students, then for 4 seconds at random intervals (100-300ms) strike off one of the students (cancel thread). After the 4s period waits for the students threads and prints the counters.

Students threads: Student (thread) adds itself to the counter of the first year, after a second it removes itself from it and adds to the second year counter and so on until it reaches the BSc counter. The thread must be always prepared for cancellation.

What you need to know:

• man 3p pthread_cancel
• man 3 pthread_cleanup_push
• man 3 pthread_cleanup_pop
• man 7 time
• man 3p clock_getres

prog20.c:

#include <errno.h>
#include <pthread.h>
#include <stdarg.h>
#include <stdbool.h>
#include <stddef.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <time.h>
#include <unistd.h>

#define MAXLINE 4096
#define DEFAULT_STUDENT_COUNT 100
#define ELAPSED(start, end) ((end).tv_sec - (start).tv_sec) + (((end).tv_nsec - (start).tv_nsec) * 1.0e-9)
#define ERR(source) (perror(source), fprintf(stderr, "%s:%d\n", __FILE__, __LINE__), exit(EXIT_FAILURE))

typedef unsigned int UINT;
typedef struct timespec timespec_t;
typedef struct studentList
{
    bool *removed;
    pthread_t *thStudents;
    int count;
    int present;
} studentsList_t;
typedef struct yearCounters
{
    int values[4];
    pthread_mutex_t mxCounters[4];
} yearCounters_t;
typedef struct argsModify
{
    yearCounters_t *pYearCounters;
    int year;
} argsModify_t;
void ReadArguments(int argc, char **argv, int *studentsCount);
void *student_life(void *);
void increment_counter(argsModify_t *args);
void decrement_counter(void *_args);
void msleep(UINT milisec);
void kick_student(studentsList_t *studentsList);

int main(int argc, char **argv)
{
    int studentsCount;
    ReadArguments(argc, argv, &studentsCount);
    yearCounters_t counters = {.values = {0, 0, 0, 0},
                               .mxCounters = {PTHREAD_MUTEX_INITIALIZER, PTHREAD_MUTEX_INITIALIZER,
                                              PTHREAD_MUTEX_INITIALIZER, PTHREAD_MUTEX_INITIALIZER}};
    studentsList_t studentsList;
    studentsList.count = studentsCount;
    studentsList.present = studentsCount;
    studentsList.thStudents = (pthread_t *)malloc(sizeof(pthread_t) * studentsCount);
    studentsList.removed = (bool *)malloc(sizeof(bool) * studentsCount);
    if (studentsList.thStudents == NULL || studentsList.removed == NULL)
        ERR("Failed to allocate memory for 'students list'!");
    for (int i = 0; i < studentsCount; i++)
        studentsList.removed[i] = false;
    for (int i = 0; i < studentsCount; i++)
        if (pthread_create(&studentsList.thStudents[i], NULL, student_life, &counters))
            ERR("Failed to create student thread!");
    srand(time(NULL));
    timespec_t start, current;
    if (clock_gettime(CLOCK_REALTIME, &start))
        ERR("Failed to retrieve time!");
    do
    {
        msleep(rand() % 201 + 100);
        if (clock_gettime(CLOCK_REALTIME, &current))
            ERR("Failed to retrieve time!");
        kick_student(&studentsList);
    } while (ELAPSED(start, current) < 4.0);
    for (int i = 0; i < studentsCount; i++)
        if (pthread_join(studentsList.thStudents[i], NULL))
            ERR("Failed to join with a student thread!");
    printf(" First year: %d\n", counters.values[0]);
    printf("Second year: %d\n", counters.values[1]);
    printf(" Third year: %d\n", counters.values[2]);
    printf("  Engineers: %d\n", counters.values[3]);
    free(studentsList.removed);
    free(studentsList.thStudents);
    exit(EXIT_SUCCESS);
}

void ReadArguments(int argc, char **argv, int *studentsCount)
{
    *studentsCount = DEFAULT_STUDENT_COUNT;
    if (argc >= 2)
    {
        *studentsCount = atoi(argv[1]);
        if (*studentsCount <= 0)
        {
            printf("Invalid value for 'studentsCount'");
            exit(EXIT_FAILURE);
        }
    }
}

void *student_life(void *voidArgs)
{
    pthread_setcanceltype(PTHREAD_CANCEL_DEFERRED, NULL);
    argsModify_t args;
    args.pYearCounters = voidArgs;
    for (args.year = 0; args.year < 3; args.year++)
    {
        increment_counter(&args);
        pthread_cleanup_push(decrement_counter, &args);
        msleep(1000);
        pthread_cleanup_pop(1);
    }
    increment_counter(&args);
    return NULL;
}

void increment_counter(argsModify_t *args)
{
    pthread_mutex_lock(&(args->pYearCounters->mxCounters[args->year]));
    args->pYearCounters->values[args->year] += 1;
    pthread_mutex_unlock(&(args->pYearCounters->mxCounters[args->year]));
}

void decrement_counter(void *_args)
{
    argsModify_t *args = _args;
    pthread_mutex_lock(&(args->pYearCounters->mxCounters[args->year]));
    args->pYearCounters->values[args->year] -= 1;
    pthread_mutex_unlock(&(args->pYearCounters->mxCounters[args->year]));
}

void msleep(UINT milisec)
{
    time_t sec = (int)(milisec / 1000);
    milisec = milisec - (sec * 1000);
    timespec_t req = {0};
    req.tv_sec = sec;
    req.tv_nsec = milisec * 1000000L;
    if (nanosleep(&req, &req))
        ERR("nanosleep");
}

void kick_student(studentsList_t *studentsList)
{
    int idx;
    if (0 == studentsList->present)
        return;
    do
    {
        idx = rand() % studentsList->count;
    } while (studentsList->removed[idx] == true);
    pthread_cancel(studentsList->thStudents[idx]);
    studentsList->removed[idx] = true;
    studentsList->present--;
}

Notes and questions #

Threads receive the pointer to the structure with current year and pointer to years counters, structure argsModify_t does not have the same flow as one in task 2 of this tutorial i.e. program is not making too many unnecessary references to the same data.

Structure studentsList_t is only used im main thread, it is not visible for students’ threads.

Cleaver structure yearCounters_t initialization will not work in archaic C standards (c89/c90). It is worth knowing but please use all the improvements of newer standards in your code.

Cleanup handlers in working thread are deployed to safely expel the student while its thread is asleep. Without the handlers the canceled student will occupy the last counter till the end of the program!

Please keep in mind that pthread_cleanup_push must be paired with pthread_cleanup_pop in the same lexical scope (the same braces {}).

How many mutexes this program uses and what they protect?

Must current year of a student be a part of argsModify_t structure?

What does it mean that the thread cancellation state is set to PTHREAD_CANCEL_DEFERRED ?

What functions used in the thread code are cancellation points?

How do we learn what functions are cancellation points?

What one in this call " pthread_cleanup_pop(1);" means ?

When the year counter is decreased?

Algorithm selecting a thread for cancellation has an apparent flow, can you name it and tell what threat it creates?

Improve random selection as an exercise.

Have a look at the method used to measure the 4 seconds life time of the program (clock_gettime, nanosleep). Please change the solution to use alarm function and the SIGALRM handler as an exercise.

Do the example tasks. During the laboratory you will have more time and a starting code. If you do following tasks in the allotted time, it means that you are well-prepared.

• Task 1 ~60 minutes
• Task 2 ~120 minutes
• Task 3 ~100 minutes

Source codes presented in this tutorial #

• prog17.c
• prog18.c
• prog19.c
• prog20.c