MIT 6.824 Lab1：用Go实现容错MapReduce调度器-开发者社区

1. 这不是“写个MapReduce”，而是MIT分布式系统课的第一次真实心跳

刚接触6.824这门课的同学，看到“lab1”三个字，容易下意识以为是“照着文档敲几行代码就能跑通的入门练习”。我当年也是这么想的——直到在凌晨三点盯着屏幕里反复崩溃的worker进程发呆，手边咖啡凉透，终端日志里满屏的rpc: client protocol error和timeout像一串串嘲讽。这不是编程作业，这是MIT分布式系统课给你递来的一把解剖刀，第一课就让你亲手切开一个真实分布式框架的腹腔：看它怎么调度、怎么容错、怎么在节点随时挂掉的前提下，依然让计算结果正确落地。

这个lab的核心，从来不是实现Map和Reduce函数本身——那两段逻辑加起来不到20行。真正的硬核在于：你得亲手搭建一套极简但完整的Master-Worker协同机制。它必须能处理三类现实世界中最棘手的状况：Worker进程毫无征兆地退出（比如被kill -9）、网络传输过程中RPC调用随机失败、以及多个Worker同时向Master请求任务时产生的竞态条件。这些不是理论假设，而是你在本地用go run启动几个进程后，用kill -9模拟出来的血淋淋现场。

关键词里虽然没填，但整个lab的骨架由四个不可绕过的技术锚点撑起：Go语言的RPC机制（不是HTTP，是原生net/rpc）、基于channel的并发控制（Master如何安全分发任务、Worker如何上报状态）、超时与重试策略（一次RPC失败后，是立刻重试？等多久？重试几次？）、以及文件系统级的中间结果持久化（Map输出不能存在内存里，必须落盘，且Reduce必须能跨机器读取）。这四点，每一点都对应着分布式系统教科书里一个独立章节。而lab1的要求，是让你在300行左右的Go代码里，把它们拧成一股绳。

适合谁来啃这块硬骨头？如果你刚学完操作系统，对进程、线程、锁有基本直觉；如果你写过带goroutine的Go程序，知道select和time.After怎么配合防死锁；如果你不害怕读官方文档——尤其是net/rpc包里那几页看似平淡实则暗藏玄机的说明——那么这就是你第一次真正触摸到“分布式”三个字温度的机会。它不考算法复杂度，不考数据结构优化，只考你能不能在混乱中建立秩序，在不确定性中锚定确定性。接下来，我们就从最真实的卡点出发，一层层剥开这个lab的肌理。

2. Master的调度逻辑：为什么不能用全局变量存任务队列？

很多同学的第一版Master，会本能地写一个全局的[]Task切片，每次Worker来要任务，就pop一个出去。代码看起来干净利落：

var tasks []Task // 全局任务队列 func (m *Master) GetTask(args *GetTaskArgs, reply *GetTaskReply) error { if len(tasks) == 0 { reply.Task = nil return nil } reply.Task = &tasks[0] tasks = tasks[1:] // 直接切片 return nil }

运行起来似乎也正常——两个Worker能拿到不同任务。但只要加上kill -9模拟故障，问题立刻爆发：Worker A刚拿到task1，还没开始处理，就被kill -9干掉；Master却已经把task1从队列里移除了。结果就是：task1永远石沉大海，既没完成，也没被重新分配。更糟的是，如果Worker B此时恰好来请求任务，它会直接拿到task2，而task1彻底丢失。

根本原因在于：全局切片的修改是瞬时的、不可逆的，它没有记录“谁拿了哪个任务”这个关键上下文。分布式调度的本质，不是“分发物品”，而是“委托责任”。Master必须明确知道：“task1已委托给Worker X，X在T时刻承诺会在T+10s内完成”。一旦X失联，Master才有依据去回收并重发。

所以正确的做法，是把任务状态显式建模。我最终采用的结构体长这样：

type Task struct { ID int Type TaskType // MAP or REDUCE Filename string // input file for map, or empty for reduce NReduce int // total number of reduce tasks State TaskState // PENDING, IN_PROGRESS, COMPLETED, FAILED WorkerID string // which worker is handling it StartTime time.Time // when assigned }

关键在State和WorkerID字段。Master启动时，所有task初始化为PENDING；当Worker请求任务时，Master遍历所有PENDING任务，找到第一个，将其State设为IN_PROGRESS，WorkerID设为该Worker的标识（比如IP+端口），再返回给Worker。这个“查找-修改-返回”的过程，必须用互斥锁保护：

func (m *Master) GetTask(args *GetTaskArgs, reply *GetTaskReply) error { m.mu.Lock() defer m.mu.Unlock() // 扫描所有pending任务 for i := range m.tasks { if m.tasks[i].State == PENDING { m.tasks[i].State = IN_PROGRESS m.tasks[i].WorkerID = args.WorkerID m.tasks[i].StartTime = time.Now() reply.Task = &m.tasks[i] return nil } } // 没有pending任务，检查是否全部完成 if m.allDone() { reply.Task = &Task{Type: SHUTDOWN} // 通知worker退出 } return nil }

提示：args.WorkerID不是客户端传来的任意字符串，而是Worker在首次注册时，Master分配的唯一ID。这个ID必须稳定——不能每次重启Worker就变，否则Master无法关联历史状态。我的方案是让Worker在启动时生成一个UUID，并在首次RPC调用中作为RegisterArgs.WorkerID发送给Master，Master存下来，后续所有请求都带这个ID。

这个设计带来的额外好处是：容错逻辑变得极其清晰。Master可以另起一个goroutine，定期扫描所有IN_PROGRESS任务，如果某个任务的StartTime距今超过10秒（TIMEOUT = 10 * time.Second），且其WorkerID对应的Worker最近没有心跳（我们稍后会加心跳机制），就直接将该任务状态重置为PENDING，下次其他Worker来请求时就能捡漏。整个过程，没有魔法，只有状态机驱动的确定性行为。

3. Worker的生命周期管理：从“拿任务”到“交差”的完整闭环

Worker的代码量往往比Master还多，因为它要处理更多“接地气”的细节：怎么连上Master？拿到任务后怎么执行？执行完结果往哪存？失败了怎么重试？很多人卡在第一步——Worker启动后，死活收不到Master的任务。排查日志发现，Master的RPC服务压根没收到请求。问题出在Go RPC的默认配置上：它要求服务端方法名必须是“Exported”（首字母大写），且参数和返回值必须是public字段。

比如，如果你定义了一个getTask小写方法：

// ❌ 错误：小写方法名，RPC无法导出 func (w *Worker) getTask(args *GetTaskArgs, reply *GetTaskReply) error { ... }

Go的net/rpc会直接忽略它，不会报错，只是静默跳过。必须改成：

// ✅ 正确：首字母大写，且参数/返回值字段也大写 func (w *Worker) GetTask(args *GetTaskArgs, reply *GetTaskReply) error { ... }

更隐蔽的坑在参数结构体：

// ❌ 错误：字段小写，RPC无法序列化 type GetTaskArgs struct { workerID string // 小写，私有字段 } // ✅ 正确：字段必须大写，且可导出 type GetTaskArgs struct { WorkerID string `json:"worker_id"` // 大写，可导出 }

解决了RPC可见性问题，Worker才能真正进入工作流。它的核心循环非常朴素：

for { // 1. 向Master请求任务 task := w.getTaskFromMaster() // 2. 如果是SHUTDOWN任务，退出循环 if task.Type == SHUTDOWN { break } // 3. 执行任务（map或reduce） w.executeTask(task) // 4. 向Master报告完成 w.reportTaskDone(task) }

但每个步骤背后都有魔鬼细节。以executeTask为例，Map任务要求：读取输入文件，调用用户提供的mapF函数，将输出按hash(key) % nReduce分到nReduce个中间文件中。这里的关键是文件命名规范。官方要求中间文件名格式为mr-X-Y，其中X是Map任务ID，Y是Reduce任务ID。很多同学直接写fmt.Sprintf("mr-%d-%d", task.ID, reduceID)，结果发现Reduce Worker找不到文件——因为Map Worker写的是mr-0-0,mr-0-1…，而Reduce Worker去读时，路径拼错了，比如读成了./mr-0-0但实际文件在./mrtmp.mr-0-0（临时目录）。

我的解决方案是：所有中间文件统一放在一个mrtmp子目录下，并在Worker启动时创建：

const TMP_DIR = "mrtmp" func (w *Worker) initTempDir() { os.MkdirAll(TMP_DIR, 0755) } func (w *Worker) getIntermediateFilename(mapID, reduceID int) string { return filepath.Join(TMP_DIR, fmt.Sprintf("mr-%d-%d", mapID, reduceID)) }

这样，Map Worker写文件时：os.Create(w.getIntermediateFilename(task.ID, reduceID))；Reduce Worker读文件时，也用完全相同的路径构造逻辑。路径一致性，是跨Worker协作的生命线。

另一个高频陷阱是reportTaskDone的实现。很多同学认为“只要Master收到了done消息，任务就算成功”，于是简单地：

// ❌ 危险：没有错误处理，网络失败就静默丢弃 func (w *Worker) reportTaskDone(task *Task) { w.callMaster("Master.TaskDone", &TaskDoneArgs{Task: task}, &TaskDoneReply{}) }

但网络是不可靠的。如果这次RPC失败，Master永远不会知道这个任务完成了，就会一直等着，最终超时重发，导致同一个任务被两个Worker执行（重复计算）。正确做法是：必须实现带重试的reportTaskDone，且重试逻辑要智能：

func (w *Worker) reportTaskDone(task *Task) { maxRetries := 3 for i := 0; i < maxRetries; i++ { var reply TaskDoneReply ok := w.callMaster("Master.TaskDone", &TaskDoneArgs{Task: task}, &reply) if ok && reply.OK { return // 成功 } // 指数退避：第一次等1s，第二次等2s，第三次等4s time.Sleep(time.Second << uint(i)) } // 重试3次都失败？只能放弃，但至少log一下 log.Printf("WARNING: failed to report task %v done after %d retries", task.ID, maxRetries) }

注意：这里的callMaster是封装好的RPC调用函数，内部使用rpc.DialHTTP连接Master，并设置了合理的http.Client.Timeout（我设为5秒）。不要用默认的无限等待。

这个闭环设计，让Worker不再是“一次性的工具人”，而是一个有状态、有韧性、能自我修复的计算单元。它知道自己在做什么，也知道做砸了该怎么办。

4. 容错机制的实战验证：用`kill -9`和`pkill`制造真实故障

Lab1的终极考验，不是“跑通”，而是“在崩溃中跑通”。MIT的测试脚本test-mr.sh之所以让人头皮发麻，是因为它内置了一套暴力故障注入器：在你的MapReduce运行到一半时，随机kill -9掉某个Worker进程。如果你的系统没有妥善处理，测试必然失败。

我第一次通过测试，不是靠运气，而是靠一套可复现的故障注入流程。在本地开发时，我绝不依赖test-mr.sh的黑盒测试，而是手动拆解：

第一步：启动Master和两个Worker

# 终端1：启动Master go run mr-master.go pg*.txt # 终端2：启动Worker1 go run mr-worker.go # 终端3：启动Worker2 go run mr-worker.go

第二步：观察日志，确认一切就绪Master日志应显示：

Starting MapReduce master server at :7777 Registering worker 127.0.0.1:34567 Registering worker 127.0.0.1:34568 Created 10 map tasks and 3 reduce tasks

Worker日志应显示：

Worker 127.0.0.1:34567 registered Got task: map task 0 on file pg-xxx.txt

第三步：精准打击，制造故障当Worker1正在处理map task 0时（日志刚打印出Got task...），在终端4中执行：

# 找到Worker1的PID（根据端口号） lsof -i :34567 | grep LISTEN | awk '{print $2}' # 假设PID是12345 kill -9 12345

第四步：紧盯Master日志，验证恢复逻辑几秒后，Master日志应出现：

Worker 127.0.0.1:34567 lost (no heartbeat) Reassigning map task 0 to worker 127.0.0.1:34568

如果看到这行，说明你的超时检测和任务重分配逻辑生效了。接着，Worker2的日志应该会打印：

Got task: map task 0 on file pg-xxx.txt # 注意：是同一个task ID！

第五步：检查输出，确认结果正确等所有任务完成后，检查mr-out-*文件内容。用sort和diff对比：

# 正常输出（无故障） sort mr-out-0 > out_normal.txt # 故障后输出 sort mr-out-0 > out_fault.txt diff out_normal.txt out_fault.txt # 应该为空，表示结果一致

这个手动流程的价值在于：它把抽象的“容错”概念，转化为了可观察、可测量、可调试的具体事件链。你亲眼看到Worker消失、Master报警、任务重发、新Worker接手、最终输出不变。这种“所见即所得”的验证，比任何单元测试都更有说服力。

实操心得：在Master的checkTimeoutgoroutine里，我加了一行debug日志：log.Printf("Checking timeout: %v tasks in progress", len(inProgressTasks))。这行日志在故障注入时成了救命稻草——它让我一眼看出，Master是否真的在扫描IN_PROGRESS任务，以及扫描频率是否合理（我设为每2秒一次，太频繁会浪费CPU，太慢会导致恢复延迟）。

5. 文件系统与中间结果：为什么`mr-out-0`必须是`sort`后的最终结果？

MapReduce的输出语义，是整个lab最容易被误解的点。很多同学以为，只要mr-out-0文件里有单词和计数，就算成功。但官方测试脚本test-mr.sh的校验逻辑极其严格：它会用标准Unixsort命令对你的mr-out-0进行排序，再与预期的test-mr_output_0文件逐行比对。这意味着：你的mr-out-0文件内容，必须和sort mr-out-0的结果完全一致。

问题来了：Reduce函数的输出顺序是不确定的。Go的map遍历是随机的，你无法保证"a 1\nb 2\nc 3"一定按字母序输出。如果Reduce Worker直接fmt.Fprintf(file, "%s %d\n", key, value)，得到的文件内容顺序是随机的，diff必然失败。

解决方案有两个，且必须二选一：

方案A：Reduce阶段强制排序（推荐）在Reduce Worker写入mr-out-0前，先把所有key-value对收集到一个切片，然后排序：

// 在Reduce函数内部 var pairs []struct{ Key string; Value int } for key, value := range reduceResult { pairs = append(pairs, struct{ Key string; Value int }{key, value}) } // 按key字典序排序 sort.Slice(pairs, func(i, j int) bool { return pairs[i].Key < pairs[j].Key }) // 再写入文件 for _, p := range pairs { fmt.Fprintf(outfile, "%s %d\n", p.Key, p.Value) }

方案B：利用MapReduce框架的隐含保证官方文档提到：“The framework guarantees that the intermediate files for a given map task are available to all reduce tasks.” 但它没说Reduce的输出顺序。所以方案B其实是不存在的——你必须自己保证输出有序。

我选择方案A，因为它是确定性的、可验证的、且符合Unix哲学（“输出可被下游工具消费”）。更重要的是，它暴露了一个深层设计原则：MapReduce的“Reduce”阶段，本质是一个聚合+排序的组合操作。单纯的聚合（sum/count）不保证顺序，但生产环境的WordCount，用户永远期望按字母序查看结果。所以，排序不是额外负担，而是语义的一部分。

另一个文件系统相关的硬核细节是：中间文件mr-X-Y必须是原子写入。如果Map Worker在写mr-0-0时崩溃，留下一个半截文件，Reduce Worker读取时会panic。Go提供了ioutil.WriteFile，但它不是原子的（先写临时文件，再rename）。更安全的做法是：

func atomicWrite(filename string, data []byte) error { tmpfile := filename + ".tmp" if err := ioutil.WriteFile(tmpfile, data, 0644); err != nil { return err } return os.Rename(tmpfile, filename) // rename是原子的 }

我在Map Worker写中间文件时，全部替换成atomicWrite。虽然lab1的测试数据很小，不太会触发这个问题，但这个习惯，会让你在后续lab（比如处理GB级数据）时少踩无数坑。

最后，关于输出文件名：mr-out-0、mr-out-1… 的编号，必须严格对应Reduce任务ID。如果你的Reduce任务ID从0开始，那么输出文件就必须叫mr-out-0。任何偏差（比如叫output-0或mr-result-0）都会让测试脚本找不到文件，直接判fail。这种“约定大于配置”的设计，是分布式系统里最朴素也最有效的协调方式——不需要Master告诉Worker“你该叫什么”，大家按同一套规则行事即可。

6. 调试与日志：为什么`log.Printf`是你最忠实的战友

在6.824的lab里，fmt.Println是敌人，log.Printf是朋友。前者只会把信息喷到终端，后者能打上时间戳、文件名、行号，还能方便地开关。我给自己立了一条铁律：所有关键状态变更、RPC调用、超时事件、错误分支，必须有log.Printf。不是为了炫技，而是为了在故障发生时，能用grep在日志里瞬间定位问题。

比如，Master的GetTask方法，我加了三层日志：

func (m *Master) GetTask(args *GetTaskArgs, reply *GetTaskReply) error { log.Printf("DEBUG: GetTask called by worker %s", args.WorkerID) // 1. 入口 m.mu.Lock() defer m.mu.Unlock() for i := range m.tasks { if m.tasks[i].State == PENDING { log.Printf("INFO: Assigning pending task %d (type %v) to worker %s", m.tasks[i].ID, m.tasks[i].Type, args.WorkerID) // 2. 分配成功 m.tasks[i].State = IN_PROGRESS m.tasks[i].WorkerID = args.WorkerID m.tasks[i].StartTime = time.Now() reply.Task = &m.tasks[i] return nil } } log.Printf("WARN: No pending tasks for worker %s, checking completion", args.WorkerID) // 3. 无任务 // ... }

当test-mr.sh失败时，我做的第一件事，不是改代码，而是：

# 把所有日志重定向到文件 go run mr-master.go pg-*.txt 2>&1 | tee master.log go run mr-worker.go 2>&1 | tee worker1.log go run mr-worker.go 2>&1 | tee worker2.log # 然后grep关键事件 grep "Assigning pending task" master.log grep "No pending tasks" master.log grep "Worker.*lost" master.log

这种基于日志的调试法，效率远高于断点调试。因为分布式系统的问题，往往不是“某一行代码错了”，而是“某个状态在某个时间点没按预期流转”。日志就是状态流转的录像带。

另一个被低估的技巧是：给日志加唯一追踪ID。当一个Map任务被多次重发时，日志里会混杂task 0 assigned to worker A和task 0 assigned to worker B，很难区分哪次是哪次。我在Task结构体里加了一个TraceID字段：

type Task struct { ID int TraceID string // 例如 "trace-12345-0"，12345是master启动时间戳，0是task序号 // ... 其他字段 }

每次创建Task时生成：

task.TraceID = fmt.Sprintf("trace-%d-%d", time.Now().UnixNano(), task.ID)

然后所有相关日志都带上它：

log.Printf("INFO: [%s] Assigning task %d to worker %s", task.TraceID, task.ID, args.WorkerID)

这样，用grep "trace-12345-0"就能把关于这个任务的所有日志串起来，形成一条完整的时间线。这招在排查竞态条件时，简直是神技。

最后分享一个血泪教训：不要在log.Printf里放可能panic的表达式。比如log.Printf("key=%s", someMap[key])，如果key不存在，会panic。正确写法是先判断，再log：
if val, ok := someMap[key]; ok { log.Printf("key=%s, val=%d", key, val) } else { log.Printf("key=%s not found", key) }
这种细节，决定了你是花10分钟修bug，还是花2小时找为什么程序突然没了日志。

我在实际使用中发现，一个精心设计的日志系统，其价值不亚于代码本身。它不是装饰，而是系统的神经系统——当你看不见内部时，日志就是你的眼睛和耳朵。