5
\$\begingroup\$

This is a follow-up question to Concurrent stack in C

  • The pop function not just removes the top element from the stack, but also returns it.
  • If a pthreads call fails, an error message is displayed and the program exits with EXIT_FAILURE.
  • Program exits also if allocating the memory for the stack fails.

concurrent_stack.h

#ifndef CONCURRENT_STACK_H
#define CONCURRENT_STACK_H

#include <stdlib.h>

typedef struct concurrent_stack
{

    /*******************************************
    The number of elements stored in this stack.
    *******************************************/
    size_t size;

    /**************************************************
    The maximum number of elements this stack can hold.
    **************************************************/
    size_t capacity;

    /**********************************************
    The actual array holding the data of the stack.
    **********************************************/
    void** storage;

    /************************************************
    The mutual exclusion lock for updating the stack.
    ************************************************/
    pthread_mutex_t mutex;

    /*****************************
    Guards against an empty stack.
    *****************************/
    pthread_cond_t  empty_condition_variable;

    /***************************
    Guards against a full stack.
    ***************************/
    pthread_cond_t  full_condition_variable;
}
concurrent_stack;

/*****************************************
Initializes a new, empty concurrent stack.
*****************************************/
void concurrent_stack_init(concurrent_stack* stack, size_t capacity);

/****************************************
Pushes a datum onto the top of the stack.
****************************************/
void concurrent_stack_push(concurrent_stack* stack, void* datum);

/******************************************
Returns, but does not remove the top datum.
******************************************/
void* concurrent_stack_top(concurrent_stack* stack);

/*******************************************************
Removes the topmost datum from the stack and returns it.
*******************************************************/
void* concurrent_stack_pop(concurrent_stack* stack);

/*******************************************
Returns the number of elements in the stack.
*******************************************/
size_t concurrent_stack_size(concurrent_stack* stack);

/*********************************
Returns the capacity of the stack.
*********************************/
size_t concurrent_stack_capacity(concurrent_stack* stack);

/***********************************
Releases all resources of the stack.
***********************************/
void concurrent_stack_free(concurrent_stack* stack);

#endif /* CONCURRENT_STACK_H */

concurrent_stack.c

#include "concurrent_stack.h"
#include <pthread.h>
#include <stdio.h>
#include <stdlib.h>

#define MAX(A,B) (((A) > (B)) ? (A) : (B))
#define CALL_PTHREADS(call) if (call) report_pthreads_error_and_exit();

static const size_t MINIMUM_CAPACITY = 10;

static void report_error_and_exit(const char* msg)
{
    fputs(msg, stderr);
    exit(EXIT_FAILURE);
}

static void report_pthreads_error_and_exit()
{
    report_error_and_exit("pthread library failed.");
}

void concurrent_stack_init(concurrent_stack* stack, size_t capacity)
{
    stack->size = 0;
    stack->capacity = MAX(capacity, MINIMUM_CAPACITY);
    stack->storage = malloc(sizeof(void*) * stack->capacity);

    if (!stack->storage)
    {
        report_error_and_exit("Cannot allocate space for the stack.");
    }

    CALL_PTHREADS(pthread_mutex_init(&stack->mutex, NULL));
    CALL_PTHREADS(pthread_cond_init(&stack->empty_condition_variable, NULL));
    CALL_PTHREADS(pthread_cond_init(&stack->full_condition_variable, NULL));
}

void concurrent_stack_push(concurrent_stack* stack, void* datum)
{
    CALL_PTHREADS(pthread_mutex_lock(&stack->mutex));

    while (stack->size == stack->capacity)
    {
        CALL_PTHREADS(pthread_cond_wait(&stack->full_condition_variable,
                                        &stack->mutex));
    }

    stack->storage[stack->size++] = datum;

    CALL_PTHREADS(pthread_cond_signal(&stack->empty_condition_variable));
    CALL_PTHREADS(pthread_mutex_unlock(&stack->mutex));
}

void* concurrent_stack_pop(concurrent_stack* stack)
{
    void* element;

    CALL_PTHREADS(pthread_mutex_lock(&stack->mutex));

    while (stack->size == 0)
    {
        CALL_PTHREADS(pthread_cond_wait(&stack->empty_condition_variable,
                                        &stack->mutex));
    }

    element = stack->storage[--stack->size];

    CALL_PTHREADS(pthread_cond_signal(&stack->full_condition_variable));
    CALL_PTHREADS(pthread_mutex_unlock(&stack->mutex));

    return element;
}

void* concurrent_stack_top(concurrent_stack* stack)
{
    void* ret;

    CALL_PTHREADS(pthread_mutex_lock(&stack->mutex));

    while (stack->size == 0)
    {
        CALL_PTHREADS(pthread_cond_wait(&stack->empty_condition_variable,
                                        &stack->mutex));
    }

    ret = stack->storage[stack->size - 1];

    CALL_PTHREADS(pthread_cond_signal(&stack->full_condition_variable));
    CALL_PTHREADS(pthread_mutex_unlock(&stack->mutex));

    return ret;
}

size_t concurrent_stack_size(concurrent_stack* stack)
{
    size_t size;

    CALL_PTHREADS(pthread_mutex_lock(&stack->mutex));
    size = stack->size;
    CALL_PTHREADS(pthread_mutex_unlock(&stack->mutex));

    return size;
}

size_t concurrent_stack_capacity(concurrent_stack* stack)
{
    return stack->capacity;
}

void concurrent_stack_free(concurrent_stack* stack)
{
    free(stack->storage);
    CALL_PTHREADS(pthread_mutex_destroy(&stack->mutex));
    CALL_PTHREADS(pthread_cond_destroy(&stack->empty_condition_variable));
    CALL_PTHREADS(pthread_cond_destroy(&stack->full_condition_variable));
}

main.c

#include "concurrent_stack.h"
#include <pthread.h>
#include <stdio.h>

typedef struct thread_config
{
    size_t element_count;
    concurrent_stack* stack;
}
thread_config;

void* producer_code(void* data)
{
    thread_config* cfg = (thread_config*) data;
    size_t limit = cfg->element_count;
    concurrent_stack* stack = cfg->stack;

    for (size_t i = 0; i != limit; ++i)
    {
        concurrent_stack_push(stack, (void*) i);
    }

    return NULL;
}

void* inspector_code(void* data)
{
    thread_config* cfg = (thread_config*) data;
    size_t limit = cfg->element_count;
    concurrent_stack* stack = cfg->stack;

    for (size_t i = 0; i != limit; ++i)
    {
        concurrent_stack_top(stack);
        concurrent_stack_size(stack);
    }

    return NULL;
}

void* consumer_code(void* data)
{
    thread_config* cfg = (thread_config*) data;
    size_t limit = cfg->element_count;
    concurrent_stack* stack = cfg->stack;

    for (size_t i = 0; i != limit; ++i)
    {
        concurrent_stack_pop(stack);
    }

    return NULL;
}

static const size_t PRODUCERS = 3;
static const size_t CONSUMERS = 3;
static const size_t INSPECTORS = 1;

static const size_t PRODUCER_ELEMENTS = 91 * 1000;
static const size_t CONSUMER_ELEMENTS = 90 * 1000;
static const size_t INSPECTOR_ELEMENTS = 50 * 1000;

/**
 In order to make sure that the program exits, you must guarantee that:

 PRODUCER_ELEMENTS * PRODUCERS - CONSUMER_ELEMENTS * CONSUMERS <= STACK_CAPACITY.

 Otherwise, after all consumers have done their job, the producers will fill
 it again and finally block on it.
 */
static const size_t STACK_CAPACITY = 5000;

int main()
{
    concurrent_stack st;
    concurrent_stack_init(&st, STACK_CAPACITY);

    pthread_t threads[PRODUCERS + CONSUMERS + INSPECTORS];
    size_t next_thread_slot = 0;

    thread_config producer_thread_config = { PRODUCER_ELEMENTS, &st };
    thread_config consumer_thread_config = { CONSUMER_ELEMENTS, &st };
    thread_config inspector_thread_config = { INSPECTOR_ELEMENTS, &st };

    for (size_t i = 0; i != CONSUMERS; ++i)
    {
        pthread_create(&threads[next_thread_slot++],
                       NULL,
                       consumer_code,
                       (void*) &consumer_thread_config);
    }

    for (size_t i = 0; i != INSPECTORS; ++i)
    {
        pthread_create(&threads[next_thread_slot++],
                       NULL,
                       inspector_code,
                       (void*) &inspector_thread_config);
    }

    for (size_t i = 0; i != PRODUCERS; ++i)
    {
        pthread_create(&threads[next_thread_slot++],
                       NULL,
                       producer_code,
                       (void*) &producer_thread_config);
    }

    /* All threads created. Now join them. */

    for (size_t i = 0; i != INSPECTORS + CONSUMERS + PRODUCERS; ++i)
    {
        pthread_join(threads[i], NULL);
    }

    size_t expected_stack_size =
    PRODUCERS * PRODUCER_ELEMENTS - CONSUMERS * CONSUMER_ELEMENTS;

    printf("Expected final stack size: %zu\n", expected_stack_size);
    printf("Actual final stack size:   %zu\n", concurrent_stack_size(&st));
    concurrent_stack_free(&st);
}

Critique request

Please tell me anything that comes to mind.

\$\endgroup\$
  • \$\begingroup\$ Out of curiosity, what tools do you use to write C code? Is it just a text editor and command-line GCC, or do you use an IDE? \$\endgroup\$ – Nic Hartley Jan 18 '17 at 22:11
  • \$\begingroup\$ @QPaysTaxes Xcode. Why? \$\endgroup\$ – coderodde Jan 19 '17 at 0:55
  • \$\begingroup\$ Again, curiosity. I'm looking to develop some larger C programs than what I can really manage without a proper IDE, but I can't find a plain C IDE (they're all C++, with the ability to use a C compiler) which I feel like is a requirement. \$\endgroup\$ – Nic Hartley Jan 19 '17 at 23:29
2
\$\begingroup\$

No way to use top() correctly

I'm not sure you should include concurrent_stack_top(), because it is unclear how you could use that function safely/usefully. For example, suppose you have an inspector that does this:

int totalCount = 0;
for (size_t i = 0; i != limit; ++i)
{
    MyType *foo = (MyType *) concurrent_stack_top(stack);

    totalCount += foo->count;
    foo->count = 0;
}

The first problem is, how would the inspector thread be able to inspect all the nodes on the stack? If some other thread pushed two nodes before the inspector was able to call top(), then the inspector would miss a node. Same thing if a consumer consumed two nodes in a row. What's most likely to happen is that the inspector would just keep getting the same nodes over and over again, which isn't that useful.

What's worse is if you have a consumer like this:

void consumer(void)
{
    while (1) {
        MyType *foo = (MyType *) concurrent_stack_pop(stack);

        do_something_with(foo);
        free(foo);
    }
}

Now, if the inspector used top() to retrieve a node, and then a consumer used pop() to retrieve the same node but called free() on it while the inspector was still using it, then the inspector would be dereferencing freed data.

The only safe way to use top() would be to keep the stack locked while using the top node, and then unlock it when you are finished.

Same thing with size()

The same reasoning applies to concurrent_stack_size(). If you call that function and get some size back, then by the time you are doing anything with the size, the stack could be any size including 0. So I'm not sure how you could use the returned size for any useful purpose.

\$\endgroup\$

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.