thdsort

/*
  thdsort.c
  S. Edward Dolan
  Wednesday, February 26 2020
*/

#define _GNU_SOURCE             /* for qsort_r() */

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

#define MAX_THREADS 64
#define MIN_ARRAY_SIZE 128

/* Larry Wall says... */
void die(const char* format, ...) {
    va_list ap;
    va_start(ap, format);
    vprintf(format, ap);
    va_end(ap);
    exit(1);
}

/*
  An array slice to be sorted. The members are the same parameters as qsort()
  except the merge index which is used by merge().

  NOTE: size and compar could be hard-coded in sort()
*/
typedef struct slice_t {
    void* base;                 /* pointer to first element of the slice */
    size_t nmemb;               /* slice size */
    size_t size;                /* element size in bytes */
    int (*compar)(const void*, const void*, void*); /* comparison function */
    size_t i;                   /* merge index */
} slice_t;

/* sort comparison function */
int intless(const void* x, const void* y, void* foo) {
    (void)foo;
    return *(int const*)x - *(int const*)y;
}

/* sort the array slice */
void* sortThread(void* slice) {
    slice_t* s = (slice_t*)slice;
    /* man qsort_r to see what this arg is for -----v */
    qsort_r(s->base, s->nmemb, s->size, s->compar, NULL);
    return slice; // don't care about the result, could return NULL
}

/*
  debug: For testing the correctness of a multi-threaded sort. See usage in
         sort() and merge();
*/
void writeArray(int* a, size_t arraySize, const char* fileName) {
    FILE* f = fopen(fileName, "wb");
    fwrite(a, sizeof(int), arraySize, f);
    fclose(f);
}

* debug: For viewing small arrays. */
void printArray(int* a, size_t n) {
    for (size_t i=0; i<n; ++i)
        printf("%d\n", a[i]);
    puts("");
}

/*
  Merge the sorted slices in `a' into a new array.
 */
void merge(int* a, slice_t* slices, int nslices, size_t arraySize) {
    int* out = malloc(sizeof(int) * arraySize);
    if (!out) {
        free(a);
        die("failed to allocate merge array of size: %ld\n", arraySize);
    }
    size_t i = 0;
    slice_t *winner = NULL;     /* slice with the smallest integer */
    while (1) {
        winner = NULL;
        for (int j=0; j<nslices; ++j) {
            if (!winner && slices[j].i < slices[j].nmemb)
                winner = &slices[j];
            else if (slices[j].i < slices[j].nmemb &&
                     *((int*)slices[j].base + slices[j].i) <
                     *((int*)winner->base + winner->i))
                winner = &slices[j]; /* swap */
        }
        if (winner)
            out[i++] = *((int*)winner->base + winner->i++);
        else
            break;              /* all slices merged */
    }
    /* writeArray(out, arraySize, "multi_threaded.bin"); */
    /* printArray(out, arraySize); */
    free(out);
}

/*
  Sort an array of random integers using nThreads.
 */
void sort(int nThreads, size_t arraySize) {
    int* a = malloc(sizeof(int) * arraySize);
    if (!a)
        die("failed to allocate initial array of size: %ld\n", arraySize);
    for (size_t i=0; i<arraySize; ++i)
        a[i] = rand();          /* initialize with random integers */
    if (nThreads == 1) {
        qsort_r(a, arraySize, sizeof(int), intless, NULL);
        /* writeArray(a, arraySize, "single_threaded.bin"); */
        /* printArray(a, arraySize); */
    }
    else {
        int i = 0;
        slice_t slices[MAX_THREADS];
        memset(slices, 0, sizeof(slice_t) * MAX_THREADS);
        size_t sliceSize = arraySize / nThreads; /* see @@@ below */
        for (; i<nThreads; ++i) {
            slices[i].base = a + sliceSize * i;
            slices[i].nmemb = sliceSize;
            slices[i].size = sizeof(int);
            slices[i].compar = intless;
            /* memset above will zero i */
        }
        /* @@@ last slice gets the leftover from the integer division */
        slices[i-1].nmemb += arraySize % nThreads;
        pthread_t threads[MAX_THREADS];
        for (i=0; i<nThreads; ++i)
            if (pthread_create(&threads[i], NULL, sortThread, &slices[i])) {
                free(a);
                die("failed to create thread: %d\n", i);
            }
        for (i=0; i<nThreads; ++i)
            if (pthread_join(threads[i], NULL)) {
                free(a);
                die("failed to join thread: %d\n", i);
            }
        merge(a, slices, nThreads, arraySize);
    }
    free(a);
}

int main(int argc, char** argv) {
    if (argc != 3)
        die("usage: thdsort THREAD_COUNT ARRAY_SIZE\n");
    int nThreads = atoi(argv[1]);
    if (nThreads < 1 || nThreads > MAX_THREADS)
        die("THREAD_COUNT must be 1 to %d\n", MAX_THREADS);
    long arraySize = atol(argv[2]);
    if (arraySize < MIN_ARRAY_SIZE)
        die("ARRAY_SIZE must be >= %d\n", MIN_ARRAY_SIZE);
    sort(nThreads, arraySize);
    return 0;
}