Build MapReduce in Go

Forget Go for a moment. Your system will look like this:

                       COORDINATOR
                    "Who should do what?"
                    /       |       \\
                   / RPC    | RPC    \\ RPC
                  v         v         v

              Worker 1   Worker 2   Worker 3
                  |          |          |
               Map/Reduce Map/Reduce Map/Reduce
                  \\          |          /
                   \\         |         /
                      shared files

The coordinator does not do MapReduce computation. It decides who should do what and tracks progress. Workers do the actual computation.

Map task 0 -> worker A
Map task 1 -> worker B
Map task 2 -> worker C

worker B finished map 1

Map task 3 -> worker B

Workers repeatedly ask the coordinator for a task, receive it, execute it, write files, report completion, and ask again.

ask coordinator for task
        ↓
receive task
        ↓
execute it
        ↓
write files
        ↓
tell coordinator "done"
        ↓
ask again

There are two communication paths:

• Coordinator ↔ Workers through RPC: mostly the control plane.
• Workers ↔ Shared filesystem: the data plane.

That distinction is important.

Do not learn all of Go first. Start with the concepts you need to understand the lab.

Function declarations

func Map(filename string, contents string) []mr.KeyValue {
    kva := []mr.KeyValue{}

    kv := mr.KeyValue{"hello", "1"}
    kva = append(kva, kv)

    return kva
}

This is roughly equivalent to Python’s def Map(filename: str, contents: str) -> list[KeyValue]. Go puts the type after the variable name: filename string.

:= declares a variable

x := 5

This declares a new variable and infers its type, roughly like x = 5 in Python. You could explicitly write var x int = 5, but Go programmers commonly use :=.

Structs

type KeyValue struct {
    Key   string
    Value string
}

Think of this like a Python dataclass with Key and Value fields. Create one with named fields when learning:

kv := KeyValue{
    Key: "cat",
    Value: "1",
}

Slices

values := []string{}
values = append(values, "hello")

Think of []string as a list of strings. Go uses append(values, "hello") rather than Python’s values.append("hello").

Loops

for _, kv := range kva {
    fmt.Println(kv.Key)
}

This is roughly like for kv in kva: in Python. The underscore means “I don’t care about this value.” range can provide both index and value:

for i, kv := range kva {

Errors

data, err := os.ReadFile(filename)

if err != nil {
    log.Fatal(err)
}

Go normally checks returned errors explicitly, rather than using Python’s try/except for every operation.

Your turn01

Before doing anything distributed, open these files:

• main/mrsequential.go
• mrapps/wc.go
• mr/worker.go

Answer these yourself:

1. What arguments does Map() receive?
2. What does Map() return?
3. What arguments does Reduce() receive?
4. What does Reduce() return?
5. What is a KeyValue?
6. Where does mapf come from inside Worker()?

Then run the sequential implementation exactly as the lab describes. Don’t continue until you can explain:

file → Map() → []KeyValue → sort/group → Reduce() → mr-out-0

Reference: the function types

mapf func(string, string) []KeyValue

reducef func(string, []string) string

The worker doesn’t load these functions itself. main/mrworker.go loads the .so plugin and passes the functions into Worker(sockname, mapf, reducef). For now, you don’t need to worry about how Go plugins work.

The original MapReduce paper had arbitrary input splits and machines distributed across a cluster. This lab simplifies that:

Here, one input file = one Map task.

The starter repo has eight pg-*.txt input files, so normally there are eight Map tasks. The number of workers is separate from the number of tasks.

8 map tasks
3 worker processes
10 reduce tasks

One worker might execute Map 0, then Map 4, then Reduce 2, then Reduce 8 over its lifetime.

TASK ≠ WORKER

A worker is a reusable executor. A task is one piece of work.

Eventually, workers and the coordinator need a simple conversation:

Worker: "Give me something to do."
Coordinator: "Run Map task 3 on pg-frankenstein.txt. There are 10 reducers."
Worker: "Okay."

Worker: "Map task 3 finished."
Coordinator: "Okay."

Worker: "Give me something."
Coordinator: "Nothing currently available. Wait."

Coordinator: "Job is finished. Exit."

Use four task types:

MapTask
ReduceTask
WaitTask
ExitTask

Go’s iota makes it convenient to assign consecutive integer values:

type TaskType int

const (
    MapTask TaskType = iota
    ReduceTask
    WaitTask
    ExitTask
)

These become 0, 1, 2, and 3, respectively.

Your turn02

Open mr/rpc.go and design two RPCs: RequestTask and ReportTaskDone. Decide what information a worker needs from RequestTask.

For a Map task, the worker needs at least:

• task ID
• filename
• number of reducers

For Reduce:

• reduce task ID
• number of map tasks

Reference: a reasonable starting shape

type RequestTaskArgs struct {
}

type RequestTaskReply struct {
    Type     TaskType
    TaskID   int
    Filename string
    NReduce  int
    NMap     int
}

type ReportTaskArgs struct {
    Type   TaskType
    TaskID int
}

type ReportTaskReply struct {
}

Subtle Go / RPC detail: TaskID is visible through RPC; taskID is not. Go’s RPC system only transmits exported fields, which begin with a capital letter.

The coordinator needs to remember whether every task is idle, in progress, or done.

Idle ── assigned ──→ InProgress ── worker finishes ──→ Done
                           │
                           └── timeout ──→ Idle

A useful Go struct might be:

type Task struct {
    ID       int
    Filename string
    State    TaskState
    Started  time.Time
}

And the state type:

type TaskState int

const (
    Idle TaskState = iota
    InProgress
    Completed
)

The coordinator needs collections of tasks:

type Coordinator struct {
    mapTasks    []Task
    reduceTasks []Task

    nReduce int
}

Your turn03

Implement only MakeCoordinator(). Don’t schedule anything yet. Given:

func MakeCoordinator(sockname string, files []string, nReduce int) *Coordinator

initialize one Map task per input file and nReduce Reduce tasks.

Reference: initialize Map tasks

for i, filename := range files {
    task := Task{
        ID:       i,
        Filename: filename,
        State:    Idle,
    }

    c.mapTasks = append(c.mapTasks, task)
}

For reducers:

for i := 0; i < nReduce; i++ {
    c.reduceTasks = append(c.reduceTasks, Task{
        ID:    i,
        State: Idle,
    })
}

Implement the path from worker to coordinator through an RPC. The coordinator finds an idle Map task, marks it in progress, records when it started, and returns it.

Before:
Map 0 Idle
Map 1 Idle
Map 2 Idle

Worker A requests

After:
Map 0 InProgress
Map 1 Idle
Map 2 Idle

Reply: Map task 0

Your turn04

Implement a coordinator RPC handler like this. For now, only assign Map tasks. If no idle Map task exists, return WaitTask. Then modify Worker() so it calls RequestTask and prints what it receives. Don’t execute the Map yet. Start a coordinator and two workers; verify they receive different tasks.

func (c *Coordinator) RequestTask(
    args *RequestTaskArgs,
    reply *RequestTaskReply,
) error

Reference idea

for i := range c.mapTasks {
    if c.mapTasks[i].State == Idle {
        c.mapTasks[i].State = InProgress
        c.mapTasks[i].Started = time.Now()

        reply.Type = MapTask
        reply.TaskID = c.mapTasks[i].ID
        reply.Filename = c.mapTasks[i].Filename
        reply.NReduce = c.nReduce

        return nil
    }
}

reply.Type = WaitTask

Notice for i := range c.mapTasks rather than for _, task := range c.mapTasks. We want to modify the actual element using c.mapTasks[i].State = InProgress.

Multiple worker processes can make RPCs at the same time. Without protection, two handlers could both observe Map 0 as idle and both assign it.

Add this to Coordinator:

mu sync.Mutex

Then use:

c.mu.Lock()
defer c.mu.Unlock()

defer means “execute this when the current function returns.” So the lock is acquired now and released before the function returns, no matter which return path is taken.

Your turn05

Protect RequestTask(), ReportTaskDone(), and Done() with the mutex. Then ask yourself: why should the coordinator not hold this mutex while a worker is processing an entire file?

Answer: If Worker A gets Map 0 and the coordinator holds the lock for five seconds, Worker B’s request may wait five seconds. That destroys parallelism. The coordinator lock protects metadata operations, which should be short. Map and Reduce computation happens outside the coordinator.

Suppose Map task 3 is assigned pg-frankenstein.txt. The worker reads the file, calls mapf(filename, contents), receives a slice of KeyValue pairs, and partitions them into nReduce buckets.

read file
   ↓
call mapf(filename, contents)
   ↓
receive []KeyValue
   ↓
partition into nReduce buckets

The lab’s helper ihash(key) gives you the partitioning rule:

reduceID := ihash(kv.Key) % nReduce

If nReduce = 3, you create buckets 0, 1, and 2. A slice of buckets, where each bucket is itself a slice of KeyValue, has type [][]KeyValue. Think of it as a list of lists.

buckets := make([][]KeyValue, nReduce)

for _, kv := range kva {
    r := ihash(kv.Key) % nReduce
    buckets[r] = append(buckets[r], kv)
}

Your turn06

Write a helper like doMapTask(...). It should use os.ReadFile(), call mapf, create the buckets, and partition every KeyValue. For now, just print how many values ended up in each bucket. Don’t write intermediate files yet.

Checkpoint example

If Map returns cat 1, dog 1, cat 1, bird 1, and hashing gives cat → 0, dog → 1, bird → 0, then bucket 0 contains both cats and bird, while bucket 1 contains dog.

Each bucket needs to survive after the Map worker finishes. The lab’s naming pattern is:

mr-X-Y

Here, X is the Map task number and Y is the Reduce task number. Map task 3 with three reducers creates:

mr-3-0
mr-3-1
mr-3-2

Later, Reduce task 1 knows it must read mr-0-1, mr-1-1, mr-2-1, mr-3-1, and so on.

                 Reduce 0       Reduce 1       Reduce 2

Map 0             mr-0-0        mr-0-1        mr-0-2
Map 1             mr-1-0        mr-1-1        mr-1-2
Map 2             mr-2-0        mr-2-1        mr-2-2

This deterministic naming is why this lab’s coordinator doesn’t need the complicated intermediate-location metadata from Google’s original paper. All workers share the same filesystem.

Use Go’s encoding/json rather than inventing a custom text format.

Writing

enc := json.NewEncoder(file)

for _, kv := range bucket {
    err := enc.Encode(&kv)
    if err != nil {
        // handle error
    }
}

Reading later:

dec := json.NewDecoder(file)

for {
    var kv KeyValue

    err := dec.Decode(&kv)
    if err != nil {
        break
    }

    intermediate = append(intermediate, kv)
}

Each struct becomes JSON like:

{"Key":"cat","Value":"1"}

Your turn07

Extend doMapTask(). For every reduce bucket r, write mr-mapTaskID-r using JSON. Then inspect the files manually with cat mr-0-0. Make sure you understand their contents before moving on.

Imagine a worker starts writing mr-3-2, writes 50% of it, and crashes. Another process might open the file and see half a JSON stream.

Instead, write to a temporary file and only publish it when it is complete:

temporary file
      ↓
write everything
      ↓
close
      ↓
atomic rename
      ↓
mr-3-2

Go provides:

tmp, err := os.CreateTemp(".", "mr-tmp-*")

The "." places the temporary file in the same directory, so os.Rename() can rename it on the same filesystem. Then:

os.Rename(tmp.Name(), finalName)

Your turn08

Change the Map output code from directly creating mr-3-2 to using CreateTemp, writing JSON, closing the file, and renaming it. This is your first taste of writing crash-safe systems code.

Until now, the coordinator still thinks Map 3 is in progress even after the worker has finished. Add a ReportTaskDone RPC.

Worker executes Map 3
        ↓
write all intermediate files successfully
        ↓
ReportTaskDone(Map, 3)

Coordinator:
Map 3 → Completed

Ordering matters. Wrong: tell the coordinator the task is done, then write files. The coordinator might start Reduce even though the files do not exist. Correct: write and rename all files, then report completion.

Your turn09

Wire this up. Your worker loop should request a task, execute and report it if it is a Map task, and then request another task. At this point, you should be able to complete all Map tasks with multiple workers.

A Reduce task cannot safely run if some Map tasks have not produced their intermediate partitions. This lab uses a simple barrier: reducers wait until the last Map has finished.

MAP PHASE

Map 0 Done
Map 1 Done
Map 2 Running
Map 3 Done

             STOP
      no reducers yet

When Map 2 completes, all Maps are complete and the Reduce phase can begin.

Your turn10

Write a helper like func allCompleted(tasks []Task) bool. Modify RequestTask() so it assigns Map tasks if any are idle, returns WaitTask if a Map is still in progress, and starts assigning Reduce tasks only when all Maps are complete.

Easy bug to avoid: no idle Maps is not the same as all Maps done.

No idle maps:
Map 0 Running
Map 1 Running

All maps done:
Map 0 Completed
Map 1 Completed

Suppose the worker receives Reduce task 2. If there are eight Map tasks, it reads these files:

mr-0-2
mr-1-2
mr-2-2
mr-3-2
mr-4-2
mr-5-2
mr-6-2
mr-7-2

The worker needs NMap in the task reply. It reads every corresponding intermediate file, decodes all key/value pairs, combines them into one []KeyValue, sorts by key, groups equal keys, and calls reducef.

read every mr-X-2
        ↓
JSON-decode every KeyValue
        ↓
combine into one []KeyValue
        ↓
sort by Key
        ↓
group equal keys
        ↓
call reducef(key, values)

The sequential implementation in mrsequential.go already demonstrates much of the sorting and grouping logic.

You could define the same ByKey type used in the sequential implementation, or use sort.Slice:

sort.Slice(intermediate, func(i, j int) bool {
    return intermediate[i].Key < intermediate[j].Key
})

If the input is:

dog 1
cat 1
dog 1
bird 1

sorting creates:

bird 1
cat  1
dog  1
dog  1

Then use two indexes: i is the first item of a group, and j is the first item after that group. The sequential code follows this pattern:

i := 0

for i < len(intermediate) {
    j := i + 1

    for j < len(intermediate) &&
        intermediate[j].Key == intermediate[i].Key {
        j++
    }

    // intermediate[i:j] are one key

    i = j
}

Your turn11

Write doReduceTask(...). It should read every mr-X-Y, JSON-decode all key/value pairs, sort them, group identical keys, construct a []string of their values, and call reducef(key, values). Before writing output, print results like cat -> 93 and verify that they make sense.

The final file for reducer X must be named mr-out-X. Each line should look exactly like key value:

fmt.Fprintf(file, "%v %v\n", key, output)

The grader expects this format. As with intermediate files, write to a temporary file, write everything, close it, and rename it to mr-out-X. Then report the Reduce task as finished.

Your turn12

Finish Reduce output. At this point, cat mr-out-* | sort should match the sorted output from the sequential version. If WordCount does not work yet, do not implement crash recovery. Get correctness working first.

The worker now needs a loop that asks for a task and dispatches based on its type.

for {
    task := askCoordinator()

    switch task.Type {

    case MapTask:
        doMapTask(...)
        reportDone(...)

    case ReduceTask:
        doReduceTask(...)
        reportDone(...)

    case WaitTask:
        sleep()

    case ExitTask:
        return
    }
}

Go’s switch does not require a break after every case like C. For waiting, use a short pause:

time.Sleep(500 * time.Millisecond)

Without a sleep, a worker might repeatedly ask for work thousands of times per second. That is busy polling. A small sleep is fine for this lab.

The coordinator’s Done() method should essentially mean that all Reduce tasks have completed.

func (c *Coordinator) Done() bool

Once the job is complete, workers should receive ExitTask. The order matters for the lab’s TestEarlyExit: outputs must already be finalized before processes begin disappearing.

Reduce worker
    ↓
write complete mr-out-7
    ↓
atomic rename
    ↓
ReportTaskDone
    ↓
Coordinator marks Reduce 7 Completed
    ↓
all reduces complete
    ↓
Done() = true

Suppose the coordinator assigns Map 3, and the worker crashes. The coordinator never receives a completion message. If nothing else happens, Map 3 stays in progress forever and the job hangs.

The lab’s solution is intentionally simple: if a task has not completed after 10 seconds, assume its worker is gone or not making progress and assign the task again.

The coordinator does not actually know that the worker crashed. It only knows that it has not seen a result for too long.

The worker could be dead, network-stalled, extremely slow, paused, or still working normally. Lack of response does not tell you which one happened.

You already stored the task’s start time:

Started time.Time

Check the elapsed time:

time.Since(task.Started) > 10*time.Second

If the timeout has passed, transition the task from InProgress back to Idle, allowing another worker to receive it.

Idle → InProgress → Completed
          ↑
          |
          + timeout

Your turn13

Add timeout handling. You can check for timed-out tasks whenever a worker requests a task: look for timeouts, reset expired tasks to idle, then assign available work. You do not need a background timer goroutine for the first implementation.

Then try starting a coordinator and a worker, killing the worker while a Map is running, and starting another worker. The second worker should eventually receive the abandoned task.

Consider this sequence:

t=0   Map 3 → Worker A
t=10  Coordinator decides Worker A is too slow
      Map 3 → Worker B
t=11  Worker A finally finishes
t=12  Worker B finishes

Map 3 executed twice. This is called at-least-once execution. It can happen in this lab.

This is why operations should tolerate retries, files should be committed atomically, and the coordinator must handle duplicate completion messages.

This is not the only possible implementation, but it teaches an important technique. Add an Attempt int field to a task. Every assignment increments it.

Map 3 attempt 1 → Worker A
timeout
Map 3 attempt 2 → Worker B

Worker A reports: Task 3, Attempt 1, Done
Coordinator's current attempt = 2
1 != 2 → ignore stale result

The worker reports the task ID and attempt number. The coordinator checks that the reported attempt is still current before marking it complete.

type RequestTaskReply struct {
    ...
    Attempt int
}

type ReportTaskArgs struct {
    Type    TaskType
    TaskID  int
    Attempt int
}

Your turn14

You can either first pass the lab without attempt IDs and add them later, or add them now. The earlier recommendation was to add them now, because understanding stale messages is valuable distributed-systems practice.

The central completion check becomes something like:

if task.Attempt == args.Attempt &&
   task.State == InProgress {

    task.State = Completed
}

This is one of the first genuinely distributed-systems-looking pieces of code you’ll write.

Don’t confuse them. They protect different kinds of state.

These solve different failure problems. Keep the distinction in mind for future 6.5840 labs.

Your starter call() attempts:

rpc.DialHTTP("unix", coordSockName)

Eventually the coordinator exits. Then a worker asks the coordinator for work, but the socket no longer exists. You can treat that as “the job ended; the worker exits.”

Change the dial-error handling so it returns false rather than terminating through log.Fatal:

c, err := rpc.DialHTTP("unix", coordSockName)

if err != nil {
    return false
}

Then:

if !call(...) {
    return
}

An explicit ExitTask is still cleaner while the coordinator remains alive. The assignment allows either style.

Once everything works, mentally redraw your code like this:

                    Coordinator
             +----------------------+
             | mapTasks []Task      |
             | reduceTasks []Task   |
             | mutex                |
             +----------------------+
                  ^              |
                  | RPC          | RPC
                  |              v
             +---------+     +---------+
             | Worker A|     | Worker B|
             +---------+     +---------+
                  |               |
              Map task         Map task
                  |               |
                  +-------+-------+
                          |
                    intermediate
                       files
                          |
              +-----------+-----------+
              |                       |
          Reduce A                Reduce B
              |                       |
          mr-out-0                mr-out-1

The task state machine is:

Idle → InProgress → Completed
          ↑
          └── timeout

Then the job progresses through the phases:

all maps completed
        ↓
reduce phase
        ↓
all reduces completed
        ↓
job finished

If you can draw this from memory, you understand most of Lab 1.

By the end, you will have used:

// variables
x := 10

// functions
func foo(x int) int

// structs
type Task struct { ... }

// slices
[]Task
append(...)

// loops
for i := range tasks

// switch
switch task.Type

// errors
if err != nil

// pointers
func (c *Coordinator) RequestTask(...)

// mutexes
sync.Mutex

// deferred cleanup
defer c.mu.Unlock()

// time
time.Now()
time.Since(...)

// filesystem
os.ReadFile
os.CreateTemp
os.Rename

// serialization
encoding/json

// sorting
sort.Slice

// RPC
net/rpc

One syntax item may initially look strange:

func (c *Coordinator) Done() bool

The (c *Coordinator) means Done is a method on Coordinator. This is roughly like:

class Coordinator:
    def done(self) -> bool:

c is Go’s explicit version of self. The * means c is a pointer to the actual coordinator rather than a copy. You want that because you are modifying its state.

Don’t repeatedly run the entire 90-second test suite while developing. Use this progression.

Format your code

gofmt -w mr/worker.go mr/coordinator.go mr/rpc.go

First, WordCount

make RUN="-run Wc" mr

Once that passes:

make RUN="-run Indexer" mr

Then parallelism

make RUN="-run MapParallel" mr
make RUN="-run ReduceParallel" mr

Then:

make RUN="-run JobCount" mr

If this fails, you may be executing tasks more than once unnecessarily.

make RUN="-run EarlyExit" mr

If this fails, think carefully about when a task is considered done, when its file is renamed, and when Done() becomes true.

Finally:

make RUN="-run CrashWorker" mr

If this fails, investigate timeouts, reassignment, duplicate execution, partial files, and stale completions.

Then run the full suite:

make mr

The supplied tests exercise correctness, Map parallelism, Reduce parallelism, avoiding unnecessary duplicate work, safe termination, and crash recovery.

Don’t try to solve everything at once. Build it in this progression:

The important dividing line: First make basic MapReduce work. Only after that should you add timeout-based failure handling. Don’t debug RPC, JSON, sorting, concurrency, timeouts, and crashes all at the same time.

Don’t consider the assignment complete just because the tests pass. You should be able to explain these without looking at your code.

1. Why does every Map task produce nReduce files?

Because every mapper partitions its intermediate keys according to which reducer owns them.

2. Why must reducer Y read mr-X-Y from every Map task X?

Because any mapper may have produced keys belonging to reducer Y.

3. Why does the coordinator need a mutex?

Multiple RPC handlers can concurrently inspect and modify shared task metadata.

4. Why does a 10-second timeout not prove that a worker died?

In an asynchronous distributed system, lack of response cannot distinguish a crash from slowness or delay.

5. Why use temporary files plus atomic rename?

Because workers can crash at arbitrary points, and consumers should never observe partially written output.

Your first milestone: Start with Chapters 1–5 and get to the point where two workers successfully request two different Map tasks. That is the first satisfying milestone. Then move forward one chapter at a time rather than reading all the reference snippets immediately.