Mergesort using map-reduce, multithreads, buffers and condition variables

Question

I wrote a map reduce program which uses multi threads, bounded buffers, condition variables. It works perfectly for some types of inputs.

In the program there are N mappers, R reducers, 1 merger. mappers get data from input files, put each string in files to corresponding buffer-X-Y. reducers read from those buffers, sort the strings, and put each sorted sequence to buffer-Y. merger merges given sequences and writes to an output file.

How I run:

make;valgrind --tool=memcheck --leak-check=yes ./program 1 5 file o 10

"File" content:

Take me down to the paradise city

Where the grass is green and the girls are pretty

Take me home (oh won't you please take me home)

Take me down to the paradise city

Where the grass is green and the girls are pretty

Take me home (oh won't you please take me home)

If you make that file content "a a bb bb" it gives different output.

Is there any way I can improve it?

#include <errno.h>
#include <stdio.h>
#include <stdlib.h>
#include <pthread.h>
#include <sys/types.h>
#include <unistd.h>
#include <string.h>
#define FILE_NAME_SIZE 20
#define WORD_LENGTH 286
// a bit extra to carry "\t(occurance)"

struct arg {
    int index;
    char *file_name;
};
struct listNode {
    char *data;
    struct listNode *next;
    int occurrence;
};
int N;
int R;
char output_file[WORD_LENGTH];
int bufsize;

/*  there are n*r buffers between mappers-reducers. 3d array.
    r buffers between reducers-merger. should be 2d array but
    buffer[N] is allocated for second type of buffers. */
char ****buffer;

// buffer[i][j][0]=fill, buffer[i][j][1]=use, 
// buffer[i][j][2]=count, buffer[i][j][3]=to be inserted in total
int ***buffer_info;

pthread_mutex_t **mutex;
pthread_cond_t **c_fill;
pthread_cond_t **c_empty;

void put(int i, int j, char *value) {
    strcpy(buffer[i][j][(buffer_info[i][j][0])], value);
    buffer_info[i][j][0] = (buffer_info[i][j][0] + 1) % bufsize;
    buffer_info[i][j][2]++;
    printf("put into buffer[%d][%d] '%s'\n", i, j, value);
}

char* get(int i, int j) {
    char* tmp = buffer[i][j][(buffer_info[i][j][1])];
    buffer_info[i][j][1] = (buffer_info[i][j][1] + 1) % bufsize;
    buffer_info[i][j][2]--;
    printf("-get '%s' from buffer[%d][%d]\n", tmp, i, j);
    return tmp;
}

void insert (struct listNode **ptr, char *value) {
    struct listNode *newPtr;
    int cmp;

    // find a place to instert node to LL
    while(*ptr){
        // Comparision to detect & remove duplicates nodes

        cmp = strcmp(value, (*ptr)->data);
        // duplicate
        if(cmp == 0){
            (*ptr)->occurrence++;
            return; 
        }
        // the point where i need to add the node
        if(cmp < 0) 
            break;

        ptr = &(*ptr)->next;
    }

    // now here *ptr points to the pointer that i want to change
    // it can be NULL, if we are at the end of the LL

    newPtr = malloc(sizeof *newPtr);
    if(!newPtr)
        return;

    newPtr->data = strdup(value);
    newPtr->occurrence = 1;

    if(newPtr->data == NULL){
        free(newPtr);
        return;     
    }

    // here we are connecting our brand new node to the LL
    newPtr->next = *ptr;
    *ptr = newPtr;
}

static void *merger(){
    FILE *file;
    char temp[WORD_LENGTH];
    char temp2[WORD_LENGTH+30];
    int occurrence=0;
    int i;
    char *tok;

    // once we have the element that comes first we will put write it to file.
    file = fopen(output_file, "w");
    if (file == NULL) {
        printf("Error opening file!\n");
        exit(1);
    }

    // first element (according to asc order) in each file will be here
    // we will use this array to find the next string to write to output
    char child_heads[R][WORD_LENGTH];
    int child_heads2[R];

    // when there is no string left in a temp, we will
    // describe it here. value 1 means we are done.
    int child_done[R];

    for(i = 0; i<R; i++){
        child_done[i] = 0;

        // printf("buffer_info[%d][%d][3]: %d\n", N, i, buffer_info[N][i][3]);
        if(buffer_info[N][i][3] != 0){
            pthread_mutex_lock(&mutex[N][i]); 
            while (buffer_info[N][i][2] == 0) 
                pthread_cond_wait(&c_fill[N][i], &mutex[N][i]); 
            strcpy(temp2, get(N, i));
            buffer_info[N][i][3]--;
            pthread_cond_signal(&c_empty[N][i]); 
            pthread_mutex_unlock(&mutex[N][i]); 

            tok = strtok(temp2, "\t");
            strcpy(temp, tok);
            while((tok = strtok(NULL, "\t")))
                occurrence=atoi(tok);

            strcpy(child_heads[i], temp);
            child_heads2[i] = occurrence;               
        } else {
            child_done[i] = 1;
        }

        // printf("%d.%d\n", i, child_done[i]);
    }

    int done;
    int min_i = -1;

    while(1){

        min_i = -1;

        done = 1;

        // comparisons are not started, we assign the first available
        // item as the minimum for comparisons.
        if(min_i == -1){
            for(i=0; i<R; i++){
                if(child_done[i]==0){
                    min_i = i;
                    break;
                }               
            }
        }

        for(i=0; i<R; i++){
            if(child_done[i]==0 && strcmp(child_heads[i],child_heads[min_i])<0){
                min_i = i;
            }

            // if all data in all files are read stop the outer loop
            if(child_done[i] == 0){
                done = 0;               
            }
        }

        if(done == 1){
            break;
        }

        // write the element into the file
        fprintf(file, "%s\t%d\n", child_heads[min_i], child_heads2[min_i]);

        // so we used the string that comes first in heads, 
        // now we'll update that element's place in heads array.

        if(buffer_info[N][min_i][3] != 0){
            pthread_mutex_lock(&mutex[N][min_i]); 
            while (buffer_info[N][min_i][2] == 0) 
                pthread_cond_wait(&c_fill[N][min_i], &mutex[N][min_i]); 
            strcpy(temp2, get(N, min_i));
            buffer_info[N][min_i][3]--;
            pthread_cond_signal(&c_empty[N][min_i]); 
            pthread_mutex_unlock(&mutex[N][min_i]); 

            tok = strtok(temp2, "\t");
            strcpy(temp, tok);
            while((tok = strtok(NULL, "\t")))
                occurrence=atoi(tok);

            strcpy(child_heads[min_i], temp);
            child_heads2[min_i] = occurrence; 
        } else {
            // there is no element coming from the temp
            child_done[min_i] = 1;          
        }

    }

    // close temp files
    fclose(file);
    pthread_exit(NULL);
}

static void *reducer(void *arg){
    int index=*((int*)arg);

    // printf("--- reducer %d here!\n", index);

    char temp[WORD_LENGTH];
    char temp2[WORD_LENGTH+30];
    strcpy(temp, "");
    strcpy(temp2, "");
    int j, k;
    struct listNode *head = NULL;

    // read from buffer
    for(j=0; j<N; j++){
        for(k=0; k<buffer_info[j][index][3]; k++){
            pthread_mutex_lock(&mutex[j][index]); 
            while (buffer_info[j][index][2] == 0) 
                pthread_cond_wait(&c_fill[j][index], &mutex[j][index]); 
            strcpy(temp, get(j, index));
            pthread_cond_signal(&c_empty[j][index]); 
            pthread_mutex_unlock(&mutex[j][index]); 

            insert(&head, temp);    
        }
    }

    // this buffer must carry a sorted sequence.
    // when there is no items left to send, then merger must be notified
    // how to know when there is no items left to send?
    // her buffer[i][j] için 

    // write to buffer
    struct listNode *ptr = head;
    while (ptr){
        sprintf(temp2, "%s\t%d", ptr->data, ptr->occurrence);

        // write word to a buf.
        pthread_mutex_lock(&mutex[N][index]); 
        while (buffer_info[N][index][2] == bufsize) 
            pthread_cond_wait(&c_empty[N][index], &mutex[N][index]); 
        // see buffer definition to understand why (N) is here
        put(N, index, temp2);
        buffer_info[N][index][3]++;
        pthread_cond_signal(&c_fill[N][index]); 
        pthread_mutex_unlock(&mutex[N][index]); 

        // printf("Reducer %d - %d.%s\t%d\n", index, i, ptr->data, ptr->occurrence);
        ptr = ptr->next;
    }

    // deallocations
    while (head){
        ptr = head;
        head = head->next;
        free(ptr->data);
        free(ptr);
    }

    pthread_exit(NULL);
}

static void *mapper(void *arg_ptr){
    // printf("mapper %s here!\n", ((struct arg *) arg_ptr)->file_name);
    int bytes;
    int j, i;
    i = ((struct arg *) arg_ptr)->index;

    // read input file
    char temp[WORD_LENGTH];
    FILE *file;
    file = fopen(((struct arg *) arg_ptr)->file_name, "r");
    if (file == NULL) {
        printf("Error opening file: %s\n", ((struct arg *) arg_ptr)->file_name);
        return NULL;
    } 

    // scan the next %s from stream and put it to temp
    while(fscanf(file, "%s", temp) > 0){
        bytes = 0; 
        int k;
        for(k=0; k<strlen(temp)+1; k++){
            bytes += temp[k];
        }
        j = bytes % R;

        // write word to a buf.
        pthread_mutex_lock(&mutex[i][j]); 
        while (buffer_info[i][j][2] == bufsize) 
            pthread_cond_wait(&c_empty[i][j], &mutex[i][j]); 
        put(i, j, temp);
        buffer_info[i][j][3]++;
        pthread_cond_signal(&c_fill[i][j]); 
        pthread_mutex_unlock(&mutex[i][j]); 

        // good luck understanding :) 
    }

    fclose(file);


    pthread_exit(NULL);
}

int main(int argc, char *argv[]) {

    printf("_______________________________________\n");

    int i, j, k, ret;

    // program inputs: <N> <R> <infile1> … <infileN> <finalfile> <bufsize>
    N = atoi(argv[1]); // atoi = ascii to int
    R = atoi(argv[2]);
    char input_files[N][WORD_LENGTH];
    for(i=0; i<N; i++){
        strcpy(input_files[i], argv[3+i]);
    }
    strcpy(output_file, argv[3+N]);
    bufsize = atoi(argv[4+N]);
    if(bufsize>10000 || bufsize < 10 || N > 20 || N < 1 || R > 10 || R < 1){
        printf("Input is out of range!\n");
        return 0;
    }

    // create buffer
    buffer = (char* ***)malloc((N+1) * sizeof(char* **));
    for(i = 0; i < (N+1); i++){
        buffer[i] = (char* **)malloc(R * sizeof(char* *));
        for(j = 0; j < R; j++){
            buffer[i][j] = (char* *)malloc(bufsize * sizeof(char*));
            for(k = 0; k < bufsize; k++){
                buffer[i][j][k] = (char*)malloc(WORD_LENGTH * sizeof(char));
            }
        }
    }

    // buffer info. see decleration for explaination
    buffer_info = (int* **)malloc((N+1) * sizeof(int* *));
    for(i = 0; i < (N+1); i++){
        buffer_info[i] = (int* *)malloc(R * sizeof(int*));
        for(j = 0; j < R; j++){
            buffer_info[i][j] = (int* )malloc(4 * sizeof(int));
            for(k = 0; k < 4; k++){
                buffer_info[i][j][k] = 0;
            }
        }
    }

    // create mutex
    mutex = (pthread_mutex_t* *)malloc((N+1) * sizeof(pthread_mutex_t*));
    for(i = 0; i < (N+1); i++){
        mutex[i] = (pthread_mutex_t*)malloc(R * sizeof(pthread_mutex_t));
        for(j = 0; j < R; j++){
            mutex[i][j] = (pthread_mutex_t) PTHREAD_MUTEX_INITIALIZER;
        }
    }

    // create cond vars
    c_empty = (pthread_cond_t* *)malloc((N+1) * sizeof(pthread_cond_t*));
    for(i = 0; i < (N+1); i++){
        c_empty[i] = (pthread_cond_t*)malloc(R * sizeof(pthread_cond_t));
        for(j = 0; j < R; j++){
            c_empty[i][j] = (pthread_cond_t) PTHREAD_COND_INITIALIZER;
        }
    }
    c_fill = (pthread_cond_t* *)malloc((N+1) * sizeof(pthread_cond_t*));
    for(i = 0; i < (N+1); i++){
        c_fill[i] = (pthread_cond_t*)malloc(R * sizeof(pthread_cond_t));
        for(j = 0; j < R; j++){
            c_fill[i][j] = (pthread_cond_t) PTHREAD_COND_INITIALIZER;
        }
    }

    // create mapper threads
    pthread_t tids[N];
    struct arg args[N];

    for(i=0; i<N; i++){
        args[i].index = i;
        args[i].file_name = input_files[i];
        ret = pthread_create(&(tids[i]), NULL, &mapper, (void *) &args[i]);
        if (ret != 0) {
            return 0;
        }
    }
    for(i=0; i<N; i++){
        ret = pthread_join(tids[i], NULL);  
        if (ret != 0) {
            printf("thread join failed \n");
            return 0;
        }
    }

    // create reducer threads
    pthread_t tids_r[R];
    int args_r[R];

    for(i=0; i<R; i++){
        // either pass the addresses of array elements, or allocate new memory 
        // in each iteration and pass the address. otherwise, we have memory problems.
        args_r[i] = i;
        ret = pthread_create(&(tids_r[i]), NULL, &reducer, (void *) &args_r[i]);
        if (ret != 0) {
            printf("thread create failed \n");
            return 0;
        }
    }
    for(i=0; i<R; i++){
        ret = pthread_join(tids_r[i], NULL);    
        if (ret != 0) {
            printf("thread join failed \n");
            return 0;
        }
    }

    // create merger thread
    pthread_t tid;
    ret = pthread_create(&tid, NULL, &merger, NULL);
    if (ret != 0) {
        printf("thread create failed \n");
        return 0;
    }   
    ret = pthread_join(tid, NULL);   
    if (ret != 0) {
        printf("thread join failed \n");
        return 0;
    }


    // freeing
    for(i = 0; i < (N+1); i++){
        for(j = 0; j < R; j++){
            for(k = 0; k < bufsize; k++){
                free(buffer[i][j][k]);
            }
            free(buffer[i][j]);
            free(buffer_info[i][j]);
        }
        free(buffer[i]);
        free(buffer_info[i]);
        free(mutex[i]);
        free(c_empty[i]);
        free(c_fill[i]);
    }
    free(buffer);
    free(buffer_info);
    free(mutex);
    free(c_empty);
    free(c_fill);

    return 0;
}

Answer 1

• I note the code returns silently when a memory allocation fails (if(!newPtr) return;), but printf()s (to stdout) and then exits the whole process when a file operation fails. Neither of these seem like sensible error-handling strategies. In particular, errors are supposed to be put on stderr and not on stdout. What if this is called from a batch process?
• Strongly consider defining a struct that holds both an int * bufferInfo and the char ** buffer and a mutex amd the two pthread_cond_ts, then make a two-dimensional array of that. That way you can guarantee that they match, reduce array lookups, and duplicated code, and alloc/free in one fell swoop.
• What's the point of the other printfs? Are they just for logging/diagnostic purposes?
• merger() is much too long. It would be much more readable if you broke its functionality up into separate functions
• In the body of the while, do you mean to set min_i to -1 and then immediately check if it's -1? That seems pointless.