7
\$\begingroup\$

(There is a next iteration available.)

I was practicing writing some portable C89 code and came up with this parallel for loop construct. Basically, if you have a bunch of tasks that do not depend on each other, it makes sense to create a concurrent queue of those tasks, a pool of worker threads popping the queue and performing a task at a time:

parallel_for.h

#ifndef PARALLEL_FOR_H
#define PARALLEL_FOR_H

#include <stdlib.h>

#define ERROR_FORP_SUCCESS 0
#define ERROR_FORP_NO_ARGS 1
#define ERROR_FORP_UNKNOWN_CORES 2
#define ERROR_FORP_NO_MUTEX_INIT 3
#define ERROR_FORP_NO_MUTEX_DESTROY 4
#define ERROR_FORP_MALLOC_FAIL 5
#define ERROR_FORP_SSCANF_FAIL 6
#define ERROR_FORP_POPEN_FAIL 7
#define ERROR_FORP_CPU_FEOF 8
#define ERROR_FORP_CPU_FERROR 9
#define ERROR_FORP_NO_THREAD 10
#define ERROR_FORP_NO_SETCANCELTYPE 11
#define ERROR_FORP_NO_JOIN 12

/*******************************************************************************
* Runs a multithreaded for loop over the input array producing the results and *
* storing them in the output array.                                            *
*******************************************************************************/
int forp(void** input,
         void** output,
         size_t len,
         void* (*func)(void*));

/*************************************************************************
* Returns a human-readable description of an error code related to forp. *
*************************************************************************/
const char* forp_error(int error_code);

#endif /* PARALLEL_FOR_H */

parallel_for.c

#include "parallel_for.h"

#if defined(__APPLE__)
#include <pthread.h>
#elif defined(__linux__)
#include <pthread.h>
#include <sys/sysinfo.h>
#elif defined(_WIN32)
#include <windows.h>
#else
#error "Platform not supported."
#endif

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

/********************************************
* This struct implements a queue node type. *
********************************************/
typedef struct concurrent_queue_node_t {
    void* queue_node_element;
    struct concurrent_queue_node_t* next_queue_node;
} concurrent_queue_node_t;

/******************************************************************************
* This struct implements a partially concurrent queue implemented as a linked *
* list.                                                                       *
******************************************************************************/
typedef struct concurrent_queue_t {

#if defined(__APPLE__)
    pthread_mutex_t mutex;
#elif defined(__linux__)
    pthread_mutex_t mutex;
#elif defined(_WIN32)
    CRITICAL_SECTION criticalSection;
#else
#error "Platform not supported."
#endif

    size_t queue_size;
    concurrent_queue_node_t* queue_head_node;
    concurrent_queue_node_t* queue_tail_node;
} concurrent_queue_t;

/***********************************************
* Initializes a given queue to an empty state. *
***********************************************/
int concurrent_queue_t_init(concurrent_queue_t* queue)
{
#if defined(__APPLE__)
    int ret = pthread_mutex_init(&queue->mutex, NULL);

    if (ret != 0)
    {
        return ERROR_FORP_NO_MUTEX_INIT;
    }
#elif defined(__linux__)
    int ret = pthread_mutex_init(&queue->mutex, NULL);

    if (ret != 0)
    {
        return ERROR_FORP_NO_MUTEX_INIT;
    }
#elif defined(_WIN32)
    InitializeCriticalSection(&queue->criticalSection);
#else
#error "Platform not supported."
#endif

    queue->queue_size = 0;
    queue->queue_head_node = NULL;
    queue->queue_tail_node = NULL;
    return ERROR_FORP_SUCCESS;
}

/***********************************************************************
* Appends a given element to the tail of the queue. This method is not *
* thread-safe since it is used by a single-threaded process.           *
***********************************************************************/
int concurrent_queue_t_enqueue(concurrent_queue_t* queue, void* element)
{
    concurrent_queue_node_t* queue_node = malloc(sizeof *queue_node);

    if (queue_node == NULL)
    {
        return ERROR_FORP_MALLOC_FAIL;
    }

    queue_node->queue_node_element = element;
    queue_node->next_queue_node = NULL;

    if (queue->queue_size == 0)
    {
        queue->queue_head_node = queue_node;
        queue->queue_tail_node = queue_node;
    }
    else
    {
        queue->queue_tail_node->next_queue_node = queue_node;
        queue->queue_tail_node = queue_node;
    }

    queue->queue_size++;
    return ERROR_FORP_SUCCESS;
}

/*****************************************************************************
* Removes the head element of the queue. This method is thread-safe since it *
* is called from multiple distinct threads.                                  *
*****************************************************************************/
void* concurrent_queue_t_dequeue(concurrent_queue_t* queue)
{
    void* ret;
    concurrent_queue_node_t* head_node;

#if defined(__APPLE__)
    pthread_mutex_lock(&queue->mutex);
#elif defined(__linux__)
    pthread_mutex_lock(&queue->mutex);
#elif defined(_WIN32)
    EnterCriticalSection(&queue->criticalSection);
#else
#error "Platform not supported."
#endif

    if (queue->queue_size == 0)
    {
        ret = NULL;
    }
    else
    {
        head_node = queue->queue_head_node;
        ret = head_node->queue_node_element;
        queue->queue_head_node = queue->queue_head_node->next_queue_node;
        free(head_node);
        queue->queue_size--;
    }

#if defined(__APPLE__)
    pthread_mutex_unlock(&queue->mutex);
#elif defined(__linux__)
    pthread_mutex_unlock(&queue->mutex);
#elif defined(_WIN32)
    LeaveCriticalSection(&queue->criticalSection);
#else
#error "Platform not supported."
#endif
    return ret;
}

/************************************************
* Releases all the resources held by the queue. *
************************************************/
int concurrent_queue_t_destroy(concurrent_queue_t* queue)
{
    while (concurrent_queue_t_dequeue(queue) != NULL) {}

#if defined(__APPLE__)
    int ret = pthread_mutex_destroy(&queue->mutex);
    return ret == 0 ? ERROR_FORP_SUCCESS : ERROR_FORP_NO_MUTEX_DESTROY;
#elif defined(__linux__)
    int ret = pthread_mutex_destroy(&queue->mutex);
    return ret == 0 ? ERROR_FORP_SUCCESS : ERROR_FORP_NO_MUTEX_DESTROY;
#elif defined(_WIN32)
    DeleteCriticalSection(&queue->criticalSection);
    return ERROR_FORP_SUCCESS;
#else
#error "Platform not supported."
#endif
}

/******************************************************************************
* A task descriptor specifying the input element and the address at which the *
* output element should be stored.                                            *
******************************************************************************/
typedef struct task_descriptor {
    void* input_element;
    void** output_element_address;
} task_descriptor;

/**********************************************
* Returns the number of processors on Mac OS. *
**********************************************/
static int get_number_of_processors_apple(size_t* p_number_of_processors)
{
#ifdef __APPLE__
    FILE* fp;
    char output[10];
    int error_code;
    char* fgets_ret;

    fp = popen("sysctl -n hw.ncpu", "r");

    if (fp == NULL)
    {
        return ERROR_FORP_POPEN_FAIL;
    }

    fgets_ret = fgets(output, 9, fp);

    if (fgets_ret != output)
    {
        if (feof(fp))
        {
            return ERROR_FORP_CPU_FEOF;
        }

        if (ferror(fp) != 0)
        {
            return ERROR_FORP_CPU_FERROR;
        }
    }

    error_code = sscanf(output, "%zu", p_number_of_processors);

    if (error_code == EOF)
    {
        return ERROR_FORP_SSCANF_FAIL;
    }
#endif
    return ERROR_FORP_SUCCESS;
}

/*********************************************
* Returns the number of processors on Linux. *
*********************************************/
static int get_number_of_processors_linux(size_t* p_number_of_processors)
{
#ifdef __linux__
    *p_number_of_processors = (size_t)get_nprocs();
#endif
    return ERROR_FORP_SUCCESS;
}

/***********************************************
* Returns the number of processors on Windows. *
***********************************************/
static int get_number_of_processors_windows(size_t* p_number_of_processors)
{
#ifdef _WIN32
    SYSTEM_INFO si;
    GetSystemInfo(&si);
    *p_number_of_processors = (size_t) 2 * si.dwNumberOfProcessors;
#endif
    return ERROR_FORP_SUCCESS;
}

/**************************************************************
* A portable function for returning the number of processors. *
**************************************************************/
static int get_number_of_processors(size_t* p_number_of_processors)
{
#if defined(__APPLE__)
    return get_number_of_processors_apple(p_number_of_processors);
#elif defined(__linux__)
    return get_number_of_processors_linux(p_number_of_processors);
#elif defined(_WIN32)
    return get_number_of_processors_windows(p_number_of_processors);
#else
#error "Platform not supported."
#endif
}

/*****************************************************************************
* Specifies the worker thread arguments. Holds the queue and the function to *
* be applied to each queue element.                                          *
*****************************************************************************/
typedef struct worker_thread_proc_args {
    concurrent_queue_t* concurrent_queue;
    void* (*func)(void*);
    int return_status;
} worker_thread_proc_args;

/*********************************
* Implements the worker threads. *
*********************************/
void* worker_thread_proc(void* args)
{
    worker_thread_proc_args* worker_thread_proc_arguments =
    (worker_thread_proc_args*) args;

    int ret = 0;

#ifdef __APPLE__
    ret = pthread_setcanceltype(PTHREAD_CANCEL_ASYNCHRONOUS, NULL);
#elif defined(__linux__)
    ret = pthread_setcanceltype(PTHREAD_CANCEL_ASYNCHRONOUS, NULL);
#endif

    if (ret != 0)
    {
        worker_thread_proc_arguments->return_status = ret;
        return NULL;
    }
    else
    {
        worker_thread_proc_arguments->return_status = 0;
    }

    concurrent_queue_t* concurrent_queue =
    worker_thread_proc_arguments->concurrent_queue;

    void* (*func)(void*) = worker_thread_proc_arguments->func;
    task_descriptor* task_desc;

    if (worker_thread_proc_arguments->return_status != 0)
    {
        return NULL;
    }

    while ((task_desc = concurrent_queue_t_dequeue(concurrent_queue)) != NULL)
    {
        *task_desc->output_element_address = func(task_desc->input_element);
    }

    return NULL;
}

/***************************************
* Cancels all the first 'len' threads. *
***************************************/
#ifdef __APPLE__
void cancel_threads(pthread_t* pthreads, size_t len)
#elif defined(__linux__)
void cancel_threads(pthread_t* pthreads, size_t len)
#elif defined(_WIN32)
void cancel_threads(HANDLE* threads, size_t len)
#endif
{
    size_t i;

    for (i = 0; i < len; ++i)
    {
#ifdef __APPLE__
        pthread_cancel(pthreads[i]);
#elif defined(__linux__)
        pthread_cancel(pthreads[i]);
#elif defined(_WIN32)
        TerminateThread(threads[i], 0);
#endif
    }
}

/***********************************************************
* The actual implementation of the parallel for construct. *
***********************************************************/
int forp(void** input, void** output, size_t len, void* (*func)(void*))
{
    size_t number_of_cores;
    size_t szi;
    int ret;
    int join_ret = ERROR_FORP_SUCCESS;
    concurrent_queue_t queue;
    task_descriptor* task_desc;
    worker_thread_proc_args* wtpa;

#ifdef __APPLE__
    pthread_t* threads;
#elif defined(__linux__)
    pthread_t* threads;
#elif defined(_WIN32)
    HANDLE* threads;
#endif

    if (input == NULL || output == NULL || func == NULL)
    {
        return ERROR_FORP_NO_ARGS;
    }

    if (len == 0)
    {
        /*****************
        * Nothing to do. *
        *****************/
        return ERROR_FORP_SUCCESS;
    }

    ret = get_number_of_processors(&number_of_cores);

    if (ret != ERROR_FORP_SUCCESS)
    {
        return ret;
    }

    if (number_of_cores == 0)
    {
        return ERROR_FORP_UNKNOWN_CORES;
    }

    if ((ret = concurrent_queue_t_init(&queue)) != ERROR_FORP_SUCCESS)
    {
        return ret;
    }

    /**************************************
    * Create a concurrent queue of tasks. *
    **************************************/
    for (szi = 0; szi < len; szi++)
    {
        task_desc = malloc(sizeof *task_desc);

        if (task_desc == NULL)
        {
            concurrent_queue_t_destroy(&queue);
            return ERROR_FORP_MALLOC_FAIL;
        }

        task_desc->input_element = input[szi];
        task_desc->output_element_address = &output[szi];
        ret = concurrent_queue_t_enqueue(&queue, task_desc);

        if (ret != ERROR_FORP_SUCCESS)
        {
            concurrent_queue_t_destroy(&queue);
            return ret;
        }
    }

    /*****************************
    * Create the worker threads. *
    *****************************/
    threads = malloc(number_of_cores * sizeof(*threads));

    if (threads == NULL)
    {
        concurrent_queue_t_destroy(&queue);
        return ERROR_FORP_MALLOC_FAIL;
    }

    wtpa = malloc(number_of_cores * sizeof(*wtpa));

    if (wtpa == NULL)
    {
        free(threads);
        concurrent_queue_t_destroy(&queue);
        return ERROR_FORP_MALLOC_FAIL;
    }

    for (szi = 0; szi < number_of_cores; szi++)
    {
        wtpa[szi].concurrent_queue = &queue;
        wtpa[szi].func = func;
        wtpa[szi].return_status = 0;

#ifdef __APPLE__
        ret = pthread_create(&threads[szi],
                             NULL,
                             worker_thread_proc,
                             &wtpa[szi]);
#elif defined(__linux__)
        ret = pthread_create(&threads[szi],
                             NULL,
                             worker_thread_proc,
                             &wtpa[szi]);
#elif defined(_WIN32)
        threads[szi] = CreateThread(NULL,
                                    100000,
                                    (LPTHREAD_START_ROUTINE) worker_thread_proc,
                                    (LPVOID) &wtpa[szi],
                                    0,
                                    NULL);
#endif

        if (ret != 0)
        {
            cancel_threads(threads, szi);
            concurrent_queue_t_destroy(&queue);
            return ERROR_FORP_NO_THREAD;
        }

        if (wtpa[szi].return_status != 0)
        {
            cancel_threads(threads, szi + 1);
            concurrent_queue_t_destroy(&queue);
            return ERROR_FORP_NO_SETCANCELTYPE;
        }
    }

    /***********************************************
    * Wait for all the worker threads to complete. *
    ***********************************************/
    for (szi = 0; szi < number_of_cores; szi++)
    {
#ifdef _WIN32
        if (WaitForSingleObject(threads[szi], INFINITE) != 0 && join_ret == 0)
        {
            join_ret = ERROR_FORP_NO_JOIN;
        }
#else
        join_ret = pthread_join(threads[szi], NULL);

        if (ret != 0 && join_ret == ERROR_FORP_SUCCESS)
        {
            join_ret = ERROR_FORP_NO_JOIN;
        }
#endif
    }

    return join_ret;
}

const char* forp_error(int error_code)
{
    switch (error_code)
    {
        case ERROR_FORP_SUCCESS:
            return "forp succeeded.";

        case ERROR_FORP_NO_ARGS:
            return "Some arguments missing.";

        case ERROR_FORP_NO_JOIN:
            return "Could not join a thread.";

        case ERROR_FORP_CPU_FEOF:
            return "Reached EOF while reading the number of processors.";

        case ERROR_FORP_NO_THREAD:
            return "Could create a thread.";

        case ERROR_FORP_CPU_FERROR:
            return "An error occured while reading the number of processors.";

        case ERROR_FORP_POPEN_FAIL:
            return "Could not execute a program in popen.";

        case ERROR_FORP_MALLOC_FAIL:
            return "A call to malloc returned NULL.";

        case ERROR_FORP_SSCANF_FAIL:
            return "sscanf failed.";

        case ERROR_FORP_NO_MUTEX_INIT:
            return "Could not initialize a mutex.";

        case ERROR_FORP_NO_MUTEX_DESTROY:
            return "Could not destroy a mutex.";

        case ERROR_FORP_UNKNOWN_CORES:
            return "Could not determine the number of processors.";

        case ERROR_FORP_NO_SETCANCELTYPE:
            return "setcanceltype failed.";

        default:
            return "Unknown error code.";
    }
}

main.c

#include "parallel_for.h"
#include <stdio.h>
#include <stdlib.h>
#if defined(__APPLE__)
#include <sys/time.h>
#elif defined(__linux__)
#include <sys/time.h>
#elif defined(_WIN32)
#include <windows.h>
#else
#error "Platform not supported."
#endif
#include <time.h>

/*********************************
 * Implements a dummy heavy task. *
 *********************************/
static unsigned long long fibonacci(unsigned long long num)
{
    switch (num)
    {
        case 0:
            return 0;

        case 1:
            return 1;

        default:
            return fibonacci(num - 1) + fibonacci(num - 2);
    }
}

/*******************************
 * The worker thread procedure. *
 *******************************/
static void* fibonacci_func(void* arg)
{
    unsigned long long* pa = (unsigned long long*) arg;
    unsigned long long* result = malloc(sizeof(*result));
    *result = fibonacci(*pa);
    return result;
}

/**************************************
 * Populates randomly the input array. *
 **************************************/
static void populate_input_randomly(void** input_array, size_t len)
{
    unsigned long long* input_datum;
    size_t i;

    for (i = 0; i < len; i++)
    {
        input_datum = malloc(sizeof(unsigned long long));
        *input_datum = 20 + rand() % 21;
        input_array[i] = input_datum;
    }
}

/***************************
 * Prints the output array. *
 ***************************/
static void print_output(void** output, size_t len)
{
    void* raw_datum;
    unsigned long long datum;
    size_t i;
    char* separator = "";
    printf("[");

    for (i = 0; i < len; i++) {
        printf("%s", separator);
        separator = ", ";
        raw_datum = output[i];
        datum = *((unsigned long long*) raw_datum);
        printf("%llu", datum);
    }

    puts("]");
}

/**************************************************************
 * Returns a current millisecond count. Used for benchmarking. *
 **************************************************************/
static unsigned long long get_milliseconds()
{
#ifdef __APPLE__
    struct timeval tv;
    gettimeofday(&tv, NULL);
    return 1000 * tv.tv_sec + tv.tv_usec / 1000;
#elif defined(__linux__)
    struct timeval tv;
    gettimeofday(&tv, NULL);
    return 1000 * tv.tv_sec + tv.tv_usec / 1000;
#elif defined(_WIN32)
    return (unsigned long long) GetTickCount64();
#else
#error "Platform not supported."
#endif
}

#define N 100

int main(int argc, const char * argv[]) {
    void* input[N];
    void* output[N];
    unsigned long long start;
    unsigned long long end;
    size_t i;
    int error_code;

    srand((unsigned int)time(NULL));
    populate_input_randomly(input, N);

    start = get_milliseconds();
    error_code = forp(input, output, N, fibonacci_func);
    end = get_milliseconds();

    print_output(output, N);
    printf("Parallel for took %llu milliseconds. Error message: %s\n\n",
           end - start,
           forp_error(error_code));

    start = get_milliseconds();

    for (i = 0; i < N; i++)
    {
        output[i] = fibonacci_func(input[i]);
    }

    end = get_milliseconds();
    print_output(output, N);
    printf("Sequential for took %llu milliseconds.\n", end - start);

#ifdef _WIN32
    getchar();
#endif

    return 0;
}

Example output, two physical cores, Mac OS X:

[28657, 514229, 24157817, 6765, 514229, 832040, 9227465, 121393, 10946, 63245986, 832040, 63245986, 832040, 102334155, 2178309, 514229,
832040, 317811, 2178309, 317811, 63245986, 46368, 39088169, 5702887,
9227465, 75025, 3524578, 514229, 17711, 196418, 317811, 24157817,
39088169, 24157817, 24157817, 832040, 63245986, 46368, 317811, 196418,
121393, 196418, 10946, 1346269, 9227465, 121393, 3524578, 514229,
46368, 39088169, 9227465, 10946, 832040, 14930352, 3524578, 2178309,
514229, 24157817, 3524578, 75025, 63245986, 28657, 10946, 10946,
17711, 102334155, 317811, 514229, 196418, 14930352, 1346269, 63245986,
75025, 1346269, 24157817, 46368, 196418, 1346269, 9227465, 28657,
39088169, 9227465, 121393, 832040, 9227465, 10946, 514229, 6765,
9227465, 3524578, 10946, 9227465, 9227465, 317811, 10946, 196418,
46368, 6765, 28657, 2178309]
Parallel for took 7385 milliseconds. Error message: forp succeeded.

[28657, 514229, 24157817, 6765, 514229, 832040, 9227465, 121393, 10946, 63245986, 832040, 63245986, 832040, 102334155, 2178309, 514229,
832040, 317811, 2178309, 317811, 63245986, 46368, 39088169, 5702887,
9227465, 75025, 3524578, 514229, 17711, 196418, 317811, 24157817,
39088169, 24157817, 24157817, 832040, 63245986, 46368, 317811, 196418,
121393, 196418, 10946, 1346269, 9227465, 121393, 3524578, 514229,
46368, 39088169, 9227465, 10946, 832040, 14930352, 3524578, 2178309,
514229, 24157817, 3524578, 75025, 63245986, 28657, 10946, 10946,
17711, 102334155, 317811, 514229, 196418, 14930352, 1346269, 63245986,
75025, 1346269, 24157817, 46368, 196418, 1346269, 9227465, 28657,
39088169, 9227465, 121393, 832040, 9227465, 10946, 514229, 6765,
9227465, 3524578, 10946, 9227465, 9227465, 317811, 10946, 196418,
46368, 6765, 28657, 2178309]
Sequential for took 17403 milliseconds.

On portability

I was able to compile/run the program without changes on:

  1. Visual Studio 2017/Windows 10
  2. gcc 7.2.0/Ubuntu Linux
  3. Xcode 8.3.3/Mac OS X
\$\endgroup\$
  • 1
    \$\begingroup\$ #define POSIX isn't valid. You want #define _POSIX_C_SOURCE. (See: pubs.opengroup.org/onlinepubs/9699919799/xrat/…) \$\endgroup\$ – Finn O'leary Jan 21 '18 at 23:06
  • 1
    \$\begingroup\$ @Finn, did you mean to comment on vnp's answer rather than the question? \$\endgroup\$ – Toby Speight Jan 22 '18 at 11:56
  • \$\begingroup\$ WHOOPS. Yes. I misread and assumed that vnp was quoting from coderodde's submission. \$\endgroup\$ – Finn O'leary Jan 22 '18 at 19:10
10
\$\begingroup\$
  • The code for apple and linux cases is identical (except one case). Consider

    #if defined(__APPLE__) || defined(__linux__)
    #define POSIX
    #endif
    

    and consolidate the identical blocks under #ifdef POSIX. Or consistently do

    #ifdef __WIN32__
    ....
    #else
    ....
    #endif
    

    as you did near the end of forp.

  • Determining a number of processors by parsing an output of an external command doesn't look right. See this answer for details.

  • All functions except forp shall be static. Especially concurrent_queue_t_enqueue, which is surely not designed for a public access.

  • I don't see why concurrent_queue_t is implemented as a linked list. An array of task_desc with a mutex-protected cursor seems enough. It is much easier on memory.

  • It is advisable (though not mandatory) to avoid _t type names. See B.2.12 Data Types section of the Posix standard (1003.1, Rationale (Informative)).

\$\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.