Multithreaded MapReduce

A MapReduce runtime implemented in C using POSIX threads. It runs a configurable number of mapper and reducer threads over an in-memory key-value input, partitioning work deterministically and aggregating intermediate results behind mutex-protected structures.

Features

• Configurable number of mapper and reducer threads
• Deterministic partitioning of input across mappers and of intermediate keys across reducers
• Thread-safe emission functions (mr_emit_i, mr_emit_f) usable from user map/reduce callbacks
• Automatic group-by-key with lexicographic ordering of intermediate and final output
• Supports multiple mr_exec() calls within a single process

Project structure

.
├── CMakeLists.txt
├── include
│   ├── interface.h
│   └── tests.h
├── src
│   ├── interface.c
│   ├── main.c
│   ├── free_output.c
│   ├── map_and_reduce.c
│   ├── number_of_mappers_reducers.c
│   ├── partition.c
│   ├── single_map.c
│   ├── single_reduce.c
│   └── test.c
└── README.md

Programming model

Map stage

Input key-value pairs are partitioned across mapper threads. Each mapper invokes the user-supplied map() callback, which emits intermediate pairs via mr_emit_i().

Group-by-key stage

Intermediate output is sorted lexicographically by key and grouped into (key, [value list]) form.

Reduce stage

Grouped keys are partitioned across reducer threads. Each reducer invokes the user-supplied reduce() callback, which emits final pairs via mr_emit_f().

Final output

Reducer output is sorted lexicographically by key and stored in the caller-supplied output buffer.

Interface

Defined in include/interface.h:

int mr_exec(
    const struct mr_input *input,
    void (*map)(const struct mr_in_kv *),
    size_t mapper_count,
    void (*reduce)(const struct mr_out_kv *),
    size_t reducer_count,
    struct mr_output *output
);

Thread-safe emission functions, callable from map()/reduce():

int mr_emit_i(const char *key, const char *value);
int mr_emit_f(const char *key, const char *value);

Build

cmake -S . -B build
cmake --build build

Produces build/mapreduce. The CMakeLists.txt pins the compiler to clang; edit CMAKE_C_COMPILER if you'd rather use gcc.

Run the test suite

./build/mapreduce

main.c runs a fixed suite covering single-mapper/single-reducer correctness, multi-thread scaling, partition correctness, and an end-to-end word-count job. Each case prints a running score. The program always exits 0, even on failures, so check the printed Success cases: N/25 line instead of the exit code.

Example workload

The test suite's word-count job:

void amr_map(const struct mr_in_kv *in_kv) {
    mr_emit_i(in_kv->value, "1");
}

void amr_reduce(const struct mr_out_kv *inter_kv) {
    size_t cnt = inter_kv->count;
    char cnt_str[MAX_VALUE_SIZE];
    snprintf(cnt_str, MAX_VALUE_SIZE, "%zu", cnt);
    mr_emit_f(inter_kv->key, cnt_str);
}

Concurrency and sanitizers

Shared structures (intermediate/final output buffers, thread counters) are protected with mutexes.

The test suite has been run clean, with no reported errors, under:

# AddressSanitizer + UndefinedBehaviorSanitizer
gcc -Wall -Wextra -Iinclude -pthread -g -O0 \
    -fsanitize=address,undefined -fno-omit-frame-pointer \
    src/*.c -o mapreduce-asan
./mapreduce-asan

# ThreadSanitizer
gcc -Wall -Wextra -Iinclude -pthread -g -O1 \
    -fsanitize=thread \
    src/*.c -o mapreduce-tsan
./mapreduce-tsan

These runs cover the fixed 25-case suite in main.c only; they are not a proof of absence of races or memory errors under other workloads. CI (.github/workflows/ci.yml) builds and runs the suite under both sanitizer combinations on every push and pull request.

Known limitations

• main.c's exit code doesn't reflect test pass/fail. Read the printed score instead.
• No license file is currently included.