Update mr.c

mr.c

@@ -8,5 +8,197 @@
 #include "hash.h"
 #include "kvlist.h"
 
+// If no time, don't need to worry about edge cases
+
+// Structure to pass data to mapper threads
+typedef struct {
+  mapper_t mapper;   // User-supplied map function
+  kvlist_t* input;   // Data assigned to this mapper
+  kvlist_t* output;  // Store mapper's output pairs
+} mapper_arg_t;
+
+// Structure to pass data to reducer threads
+typedef struct {
+  reducer_t reducer;  // User-supplied reduce function
+  kvlist_t* input;    // Data assigned to this reducer
+  kvlist_t* output;   // Store reducer's output pairs
+} reducer_arg_t;
+
+// Shuffle struct maybe not needed? Make just in case why not XD
+typedef struct {
+  kvlist_t** reducer_lists;
+  size_t num_reducer;
+  pthread_mutex_t* mutexes;
+} shuffle_data_t;
+
+// Mapper thread function
+void* mapper_thread(void* arg) {
+  mapper_arg_t* m_arg = (mapper_arg_t*)arg;
+
+  // Process each key-value pair in input list
+  kvlist_iterator_t* itor = kvlist_iterator_new(m_arg->input);
+  for (;;) {
+    kvpair_t* pair = kvlist_iterator_next(itor);
+    if (pair == NULL) {
+      break;
+    }
+    // Use mapper function here
+    m_arg->mapper(pair, m_arg->output);
+  }
+  kvlist_iterator_free(&itor);
+  return NULL;
+}
+
+// Reducer thread function
+void* reducer_thread(void* arg) {
+  reducer_arg_t* r_arg = (reducer_arg_t*)arg;
+
+  // Sort the input list by key to group same keys together
+  kvlist_sort(r_arg->input);
+
+  // Group by key and call reducer for each unique key
+  kvlist_iterator_t* itor = kvlist_iterator_new(r_arg->input);
+  kvlist_t* current_group = kvlist_new();
+  char* current_key = NULL;
+
+  for (;;) {
+    kvpair_t* pair = kvlist_iterator_next(itor);
+
+    // If we reached end or found a new key, process current group
+    if (pair == NULL || (current_key != NULL && strcmp(current_key, pair->key) != 0)) {
+      if (current_key != NULL && current_group->head != NULL) {
+        // Call reducer for current key group
+        r_arg->reducer(current_key, current_group, r_arg->output);
+
+        // Clear current group for next key
+        kvlist_free(&current_group);
+        current_group = kvlist_new();
+      }
+
+      // Need to break?
+      if (pair == NULL) {
+        break;
+      }
+    }
+
+    // Now need to update current key and add pair to current group
+    if (current_key == NULL || strcmp(current_key, pair->key) != 0) {
+      current_key = pair->key;
+    }
+    kvlist_append(current_group, kvpair_clone(pair));
+  }
+
+  kvlist_iterator_free(&itor);
+  kvlist_free(&current_group);
+  return NULL;
+}
+
 void map_reduce(mapper_t mapper, size_t num_mapper, reducer_t reducer,
-                size_t num_reducer, kvlist_t* input, kvlist_t* output) {}
+                size_t num_reducer, kvlist_t* input, kvlist_t* output) {
+  // Check requirements
+  if (num_mapper == 0 || num_reducer == 0 || input == NULL || output == NULL) {
+    return;
+  }
+
+  // 1. Split Phase: Split input into num_mapper lists
+  kvlist_t** mapper_inputs = malloc(num_mapper * sizeof(kvlist_t*));
+  for (size_t i = 0; i < num_mapper; i++) {
+    mapper_inputs[i] = kvlist_new();
+  }
+
+  // Distribute input pairs to mapper lists using round-robin
+  kvlist_iterator_t* input_itor = kvlist_iterator_new(input);
+  size_t mapper_index = 0;
+  for (;;) {
+    kvpair_t* pair = kvlist_iterator_next(input_itor);
+    if (pair == NULL) {
+      break;
+    }
+    kvlist_append(mapper_inputs[mapper_index], kvpair_clone(pair));
+    mapper_index = (mapper_index + 1) % num_mapper;
+  }
+  kvlist_iterator_free(&input_itor);
+
+  // 2. Map Phase: Mapper thread setup
+  pthread_t* mapper_threads = malloc(num_mapper * sizeof(pthread_t));
+  mapper_arg_t* mapper_args = malloc(num_mapper * sizeof(mapper_arg_t));
+  kvlist_t** mapper_outputs = malloc(num_mapper * sizeof(kvlist_t*));
+
+  for (size_t i = 0; i < num_mapper; ++i) {
+    mapper_outputs[i] = kvlist_new();
+
+    mapper_args[i].mapper = mapper;
+    mapper_args[i].input = mapper_inputs[i];
+    mapper_args[i].output = mapper_outputs[i];
+
+    pthread_create(&mapper_threads[i], NULL, mapper_thread, &mapper_args[i]);
+  }
+
+  // Wait for all mapper threads to finish
+  for (size_t i = 0; i < num_mapper; ++i) {
+    pthread_join(mapper_threads[i], NULL);
+  }
+
+  // 3. Shuffle Phase: Distribute mapper outputs to reducer lists
+  kvlist_t** reducer_inputs = malloc(num_reducer * sizeof(kvlist_t*));
+  for (size_t i = 0; i < num_reducer; i++) {
+    reducer_inputs[i] = kvlist_new();
+  }
+
+  // Using hash to route keys
+  for (size_t i = 0; i < num_mapper; ++i) {
+    kvlist_iterator_t* out_itor = kvlist_iterator_new(mapper_outputs[i]);
+    kvpair_t* pair;
+    while ((pair = kvlist_iterator_next(out_itor)) != NULL) {
+      unsigned long hval = hash(pair->key);
+      size_t idx = hval % num_reducer;
+      kvlist_append(reducer_inputs[idx], kvpair_clone(pair));
+    }
+    kvlist_iterator_free(&out_itor);
+  }
+
+  // 4. Reduce Phase: Reducer thread setup
+  pthread_t* reducer_threads = malloc(num_reducer * sizeof(pthread_t));
+  reducer_arg_t* reducer_args = malloc(num_reducer * sizeof(reducer_arg_t));
+  kvlist_t** reducer_outputs = malloc(num_reducer * sizeof(kvlist_t*));
+
+  for (size_t i = 0; i < num_reducer; ++i) {
+    reducer_outputs[i] = kvlist_new();
+
+    reducer_args[i].reducer = reducer;
+    reducer_args[i].input = reducer_inputs[i];
+    reducer_args[i].output = reducer_outputs[i];
+
+    pthread_create(&reducer_threads[i], NULL, reducer_thread, &reducer_args[i]);
+  }
+
+  // Is this needed?? YES - wait for all reducer threads to finish
+  for (size_t i = 0; i < num_reducer; ++i) {
+    pthread_join(reducer_threads[i], NULL);
+  }
+
+  // 5. Combine all reducer outputs into final output
+  for (size_t i = 0; i < num_reducer; ++i) {
+    kvlist_extend(output, reducer_outputs[i]);
+  }
+
+  // Free stuff bye bye
+  for (size_t i = 0; i < num_mapper; ++i) {
+    kvlist_free(&mapper_inputs[i]);
+    kvlist_free(&mapper_outputs[i]);
+  }
+  for (size_t i = 0; i < num_reducer; ++i) {
+    kvlist_free(&reducer_inputs[i]);
+    kvlist_free(&reducer_outputs[i]);
+  }
+
+  // Free all allocated arrays
+  free(mapper_inputs);
+  free(mapper_threads);
+  free(mapper_args);
+  free(mapper_outputs);
+  free(reducer_inputs);
+  free(reducer_threads);
+  free(reducer_args);
+  free(reducer_outputs);
+}