Tutorial 3 - Wątki, mutexy i sygnały #

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

#include <math.h>
#include <pthread.h>
#include <stdarg.h>
#include <stddef.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>

#define MAXLINE 4096
#define DEFAULT_THREADCOUNT 10
#define DEFAULT_SAMPLESIZE 100

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

typedef unsigned int UINT;
typedef struct argsEstimation
{
    pthread_t tid;
    UINT seed;
    int samplesCount;
} argsEstimation_t;

void ReadArguments(int argc, char **argv, int *threadCount, int *samplesCount);
void *pi_estimation(void *args);

int main(int argc, char **argv)
{
    int threadCount, samplesCount;
    double *subresult;
    ReadArguments(argc, argv, &threadCount, &samplesCount);
    argsEstimation_t *estimations = (argsEstimation_t *)malloc(sizeof(argsEstimation_t) * threadCount);
    if (estimations == NULL)
        ERR("Malloc error for estimation arguments!");
    srand(time(NULL));
    for (int i = 0; i < threadCount; i++)
    {
        estimations[i].seed = rand();
        estimations[i].samplesCount = samplesCount;
    }
    for (int i = 0; i < threadCount; i++)
    {
        int err = pthread_create(&(estimations[i].tid), NULL, pi_estimation, &estimations[i]);
        if (err != 0)
            ERR("Couldn't create thread");
    }
    double cumulativeResult = 0.0;
    for (int i = 0; i < threadCount; i++)
    {
        int err = pthread_join(estimations[i].tid, (void *)&subresult);
        if (err != 0)
            ERR("Can't join with a thread");
        if (NULL != subresult)
        {
            cumulativeResult += *subresult;
            free(subresult);
        }
    }
    double result = cumulativeResult / threadCount;
    printf("PI ~= %f\n", result);
    free(estimations);
}

void ReadArguments(int argc, char **argv, int *threadCount, int *samplesCount)
{
    *threadCount = DEFAULT_THREADCOUNT;
    *samplesCount = DEFAULT_SAMPLESIZE;

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

void *pi_estimation(void *voidPtr)
{
    argsEstimation_t *args = voidPtr;
    double *result;
    if (NULL == (result = malloc(sizeof(double))))
        ERR("malloc");

    int insideCount = 0;
    for (int i = 0; i < args->samplesCount; i++)
    {
        double x = ((double)rand_r(&args->seed) / (double)RAND_MAX);
        double y = ((double)rand_r(&args->seed) / (double)RAND_MAX);
        if (sqrt(x * x + y * y) <= 1.0)
            insideCount++;
    }
    *result = 4.0 * (double)insideCount / (double)args->samplesCount;
    return result;
}

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

#define MAXLINE 4096
#define DEFAULT_N 1000
#define DEFAULT_K 10
#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 unsigned int UINT;
typedef struct argsThrower
{
    pthread_t tid;
    UINT seed;
    int *pBallsThrown;
    int *pBallsWaiting;
    int *bins;
    pthread_mutex_t *mxBins;
    pthread_mutex_t *pmxBallsThrown;
    pthread_mutex_t *pmxBallsWaiting;
} argsThrower_t;

void ReadArguments(int argc, char **argv, int *ballsCount, int *throwersCount);
void make_throwers(argsThrower_t *argsArray, int throwersCount);
void *throwing_func(void *args);
int throwBall(UINT *seedptr);

int main(int argc, char **argv)
{
    int ballsCount, throwersCount;
    ReadArguments(argc, argv, &ballsCount, &throwersCount);
    int ballsThrown = 0, bt = 0;
    int ballsWaiting = ballsCount;
    pthread_mutex_t mxBallsThrown = PTHREAD_MUTEX_INITIALIZER;
    pthread_mutex_t mxBallsWaiting = PTHREAD_MUTEX_INITIALIZER;
    int bins[BIN_COUNT];
    pthread_mutex_t mxBins[BIN_COUNT];
    for (int i = 0; i < BIN_COUNT; i++)
    {
        bins[i] = 0;
        if (pthread_mutex_init(&mxBins[i], NULL))
            ERR("Couldn't initialize mutex!");
    }
    argsThrower_t *args = (argsThrower_t *)malloc(sizeof(argsThrower_t) * throwersCount);
    if (args == NULL)
        ERR("Malloc error for throwers arguments!");
    srand(time(NULL));
    for (int i = 0; i < throwersCount; i++)
    {
        args[i].seed = (UINT)rand();
        args[i].pBallsThrown = &ballsThrown;
        args[i].pBallsWaiting = &ballsWaiting;
        args[i].bins = bins;
        args[i].pmxBallsThrown = &mxBallsThrown;
        args[i].pmxBallsWaiting = &mxBallsWaiting;
        args[i].mxBins = mxBins;
    }
    make_throwers(args, throwersCount);
    while (bt < ballsCount)
    {
        sleep(1);
        pthread_mutex_lock(&mxBallsThrown);
        bt = ballsThrown;
        pthread_mutex_unlock(&mxBallsThrown);
    }
    int realBallsCount = 0;
    double meanValue = 0.0;
    for (int i = 0; i < BIN_COUNT; i++)
    {
        realBallsCount += bins[i];
        meanValue += bins[i] * i;
    }
    meanValue = meanValue / realBallsCount;
    printf("Bins count:\n");
    for (int i = 0; i < BIN_COUNT; i++)
        printf("%d\t", bins[i]);
    printf("\nTotal balls count : %d\nMean value: %f\n", realBallsCount, meanValue);
    // 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 ReadArguments(int argc, char **argv, int *ballsCount, int *throwersCount)
{
    *ballsCount = DEFAULT_N;
    *throwersCount = DEFAULT_K;
    if (argc >= 2)
    {
        *ballsCount = atoi(argv[1]);
        if (*ballsCount <= 0)
        {
            printf("Invalid value for 'balls count'");
            exit(EXIT_FAILURE);
        }
    }
    if (argc >= 3)
    {
        *throwersCount = atoi(argv[2]);
        if (*throwersCount <= 0)
        {
            printf("Invalid value for 'throwers count'");
            exit(EXIT_FAILURE);
        }
    }
}

void make_throwers(argsThrower_t *argsArray, int throwersCount)
{
    pthread_attr_t threadAttr;
    if (pthread_attr_init(&threadAttr))
        ERR("Couldn't create pthread_attr_t");
    if (pthread_attr_setdetachstate(&threadAttr, PTHREAD_CREATE_DETACHED))
        ERR("Couldn't setdetachsatate on pthread_attr_t");
    for (int i = 0; i < throwersCount; i++)
    {
        if (pthread_create(&argsArray[i].tid, &threadAttr, throwing_func, &argsArray[i]))
            ERR("Couldn't create thread");
    }
    pthread_attr_destroy(&threadAttr);
}

void *throwing_func(void *voidArgs)
{
    argsThrower_t *args = voidArgs;
    while (1)
    {
        pthread_mutex_lock(args->pmxBallsWaiting);
        if (*args->pBallsWaiting > 0)
        {
            (*args->pBallsWaiting) -= 1;
            pthread_mutex_unlock(args->pmxBallsWaiting);
        }
        else
        {
            pthread_mutex_unlock(args->pmxBallsWaiting);
            break;
        }
        int binno = throwBall(&args->seed);
        pthread_mutex_lock(&args->mxBins[binno]);
        args->bins[binno] += 1;
        pthread_mutex_unlock(&args->mxBins[binno]);
        pthread_mutex_lock(args->pmxBallsThrown);
        (*args->pBallsThrown) += 1;
        pthread_mutex_unlock(args->pmxBallsThrown);
    }
    return NULL;
}

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

#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;
}

#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--;
}