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;
}
