Базовые операции с параллельными потоками операционной системы Linux 2

#include <stdio.h>
#include <stdarg.h>
#include <time.h>
#include <string.h>
#include <errno.h>
#include <stdlib.h>
 
#define _MULTI_THREADED
#include <pthread.h>
#include <semaphore.h>
 
/*
Variant: 7
Scheme: 5
CR1: vector.
Mutual exclusion: non-blocking semaphore scr1, blocking mutex mcr1.
CR2: sync functions ## 1, 8, 3, 4, 11, 12, 13, 14
    __sync_fetch_and_add
    __sync_sub_and_fetch
    __sync_fetch_and_or
    __sync_fetch_and_and
    __sync_xor_and_fetch
    __sync_nand_and_fetch
    __sync_bool_compare_and_swap
    __sync_val_compare_and_swap
 
Synchronizing: blocking semaphore scr21, multiple signal sig21, single signal sig22.
*/
 
// CR1.
#define VECTOR_TYPE unsigned long
// Vector is able to hold results of atomic operations.
#define VECTOR_LEN 8
VECTOR_TYPE vector[VECTOR_LEN]; //8
 
// Non-blocking semaphore scr1.
sem_t scr1;
// Blocking mutex mcr1.
pthread_mutex_t mcr1 = PTHREAD_MUTEX_INITIALIZER;
 
// CR2.
int common_int_1, common_int_2;
unsigned common_uint_1, common_uint_2;
long common_lint_1, common_lint_2;
unsigned long common_ulint_1, common_ulint_2;
 
// Blocking semaphore scr21.
sem_t scr21, scr22;
// Multiple signal sig21, single signal sig22
pthread_cond_t sig21 = PTHREAD_COND_INITIALIZER, sig22 = PTHREAD_COND_INITIALIZER;
 
void cleanup();
 
void trace(const char *format, ...);
pthread_mutex_t trace_mutex = PTHREAD_MUTEX_INITIALIZER;
 
#define THREADS_COUNT 6
 
void *thread_p1(void *arg);
void *thread_p2(void *arg);
void *thread_p3(void *arg);
void *thread_p4(void *arg);
void *thread_p5(void *arg);
void *thread_p6(void *arg);
 
typedef void *(*thread_type)(void *arg);
 
pthread_t thread_ids[THREADS_COUNT];    //6
thread_type thread_fns[THREADS_COUNT] = { thread_p1, thread_p2, thread_p3, thread_p4, thread_p5, thread_p6 };   //6
 
int main()
{
    trace("main(): start.");
 
    srand(time(NULL));
 
    sem_init(&scr1, 0, 1);
    sem_init(&scr21, 0, 0);
    sem_init(&scr22, 0, 0);
 
    {
        int i;
        trace("main(): creating threads.");
 
        for (i = 0; i < THREADS_COUNT; i++) //6
        {
            int r;
 
            if (0 != (r = pthread_create(&thread_ids[i], NULL, thread_fns[i], NULL)))
            {
                trace("main(): Couldn't pthread_create(). %d (%s).\n", r, strerror(r));
 
                cleanup();
 
                return EXIT_FAILURE;
            }
        }
 
        trace("main(): threads created. Joining.");
 
        for (i = 0; i < THREADS_COUNT; i++) //6
        {
            int r;
 
            if (0 != (r = pthread_join(thread_ids[i], NULL)))
            {
                trace("main(): Couldn't pthread_join(). %d (%s).\n", r, strerror(r));
 
                cleanup();
 
                return EXIT_FAILURE;
            }
        }
    }
 
    cleanup();
 
    trace("main(): threads finished.");
 
    return 0;
}
 
void trace(const char *format, ...)
{
    va_list args;
 
    pthread_mutex_lock(&trace_mutex);
 
    va_start(args, format);
 
    fprintf(stderr, "[%d] ", time(NULL));
    vfprintf(stderr, format, args);
    fprintf(stderr, "\n");
 
    va_end(args);
 
    pthread_mutex_unlock(&trace_mutex);
}
 
void cleanup()
{
    pthread_mutex_destroy(&mcr1);
 
    sem_destroy(&scr1);
    sem_destroy(&scr21);
    sem_destroy(&scr22);
    pthread_cond_destroy(&sig21);
    pthread_cond_destroy(&sig22);
 
    pthread_mutex_destroy(&trace_mutex);
}
 
void *thread_p1(void *arg)
{
    const char name[] = "p1";
 
    trace("Thread %s start.", name);
 
    while (1)
    {
        // Using cr1.
        trace("Thread %s read cr1.", name);
        pthread_mutex_lock(&mcr1);
 
        if (0 == sem_trywait(&scr1))
        {
            unsigned size, i;
            sem_getvalue(&scr1, &size);
 
            size++;
 
            size %= VECTOR_LEN; //8
 
            trace("Thread %s cr1 contents.", name);
            for (i = 0; i < size; i++)
                fprintf(stderr, "%lu ", vector[i]);
 
            fprintf(stderr, "\n");
        }
        else
        {
            trace("Thread %s scr1 is busy.", name);
        }
 
        pthread_mutex_unlock(&mcr1);
 
        // Opening scr21.
        trace("Thread %s opening scr21.", name);
        sem_post(&scr21);
    }
 
    trace("Thread %s end.", name);
}
 
void *thread_p4(void *arg)
{
    const char name[] = "p4";
 
    trace("Thread %s start.", name);
 
    while (1)
    {
        unsigned size;
 
        // Writing cr1.
        trace("Thread %s write cr1.", name);
        pthread_mutex_lock(&mcr1);
 
        sem_getvalue(&scr1, &size);
 
        size = (size + 1) % VECTOR_LEN; //8
 
        vector[size] = rand() % 256;
 
        sem_post(&scr1);
 
        pthread_mutex_unlock(&mcr1);
 
        // Waiting for scr21.
        trace("Thread %s waits scr21.", name);
        sem_wait(&scr21);
        trace("Thread %s has waited scr21.", name);
    }
 
    trace("Thread %s end.", name);
}
 
void *thread_p2(void *arg)
{
    const char name[] = "p2";
 
    trace("Thread %s start.", name);
 
    while (1)
    {
        int t;
        unsigned size;
        // Wait for sig21.
        pthread_mutex_lock(&mcr1);
        pthread_cond_wait(&sig21, &mcr1);
        trace("Thread %s got signal sig21.", name);
        pthread_mutex_unlock(&mcr1);
 
        // Using and modification of cr2.
        t = __sync_fetch_and_add(&common_int_1, rand() % 256);
        trace("Thread %s common_int_1 = %d", name, t);
        t = __sync_sub_and_fetch(&common_int_2, rand() % 256);
        trace("Thread %s common_int_2 = %d", name, t);
 
        // Writing to cr1.
        trace("Thread %s write cr1.", name);
        pthread_mutex_lock(&mcr1);
 
        sem_getvalue(&scr1, &size);
 
        size = (size + 1) % VECTOR_LEN; //8
 
        vector[size] = rand() % 256;
 
        sem_post(&scr1);
 
        pthread_mutex_unlock(&mcr1);
 
        // Sending sig22.
        trace("Thread %s sending sig22.", name);
        pthread_cond_signal(&sig22);
    }
 
    trace("Thread %s end.", name);
}
 
void *thread_p5(void *arg)
{
    const char name[] = "p5";
 
    trace("Thread %s start.", name);
 
    while (1)
    {
        unsigned t;
        // Wait for sig21.
        pthread_mutex_lock(&mcr1);
        pthread_cond_wait(&sig21, &mcr1);
        trace("Thread %s got signal sig21.", name);
        pthread_mutex_unlock(&mcr1);
 
        // Using and modification of cr2.
        t = __sync_fetch_and_or(&common_uint_1, rand() % 256);
        trace("Thread %s common_uint_1 = %d", name, t);
        t = __sync_fetch_and_and(&common_uint_2, rand() % 256);
        trace("Thread %s common_uint_2 = %d", name, t);
 
        // Using cr1.
        trace("Thread %s read cr1.", name);
        pthread_mutex_lock(&mcr1);
 
        if (0 == sem_trywait(&scr1))
        {
            unsigned size, i;
            sem_getvalue(&scr1, &size);
 
            size++;
 
            size %= VECTOR_LEN; //8
 
            trace("Thread %s cr1 contents.", name);
            for (i = 0; i < size; i++)
                fprintf(stderr, "%lu ", vector[i]);
 
            fprintf(stderr, "\n");
        }
        else
        {
            trace("Thread %s scr1 is busy.", name);
        }
 
        pthread_mutex_unlock(&mcr1);
 
        // Closing scr21.
        sem_wait(&scr21);
    }
 
    trace("Thread %s end.", name);
}
 
void *thread_p3(void *arg)
{
    const char name[] = "p3";
 
    trace("Thread %s start.", name);
 
    while (1)
    {
        // Modification cr2.
        trace("Thread %s cr2 modification.", name);
        __sync_xor_and_fetch(&common_lint_1, rand() % 256);
        __sync_nand_and_fetch(&common_lint_2, rand() % 256);
 
        // Sending sig21.
        trace("Thread %s sending sig21.", name);
        pthread_cond_broadcast(&sig21);
 
        // Modification cr2.
        trace("Thread %s cr2 modification.", name);
        __sync_bool_compare_and_swap(&common_ulint_1, rand() % 256, rand() % 256);
        __sync_val_compare_and_swap(&common_ulint_2, rand() % 256, rand() % 256);
    }
 
    trace("Thread %s end.", name);
}
 
void *thread_p6(void *arg)
{
    const char name[] = "p6";
 
    trace("Thread %s start.", name);
 
    while (1)
    {
        // Wait for sig21.
        pthread_mutex_lock(&mcr1);
        pthread_cond_wait(&sig21, &mcr1);
        trace("Thread %s got signal sig21.", name);
        pthread_mutex_unlock(&mcr1);
 
        // Using cr2.
        trace("Thread %s cr2.common_int_1 = %d", name, common_int_1);
        trace("Thread %s cr2.common_int_2 = %d", name, common_int_2);
 
        // Wait for sig22.
        pthread_mutex_lock(&mcr1);
        pthread_cond_wait(&sig22, &mcr1);
        trace("Thread %s got signal sig22.", name);
        pthread_mutex_unlock(&mcr1);
 
        // Using cr2.
        trace("Thread %s cr2.common_uint_1 = %u", name, common_uint_1);
        trace("Thread %s cr2.common_uint_2 = %u", name, common_uint_2);
    }
 
    trace("Thread %s end.", name);
}