MIT6.5840 Lab 1：MapReduce

相关：

• 6.5840 Lab 1: MapReduce

实验目标：实现一个 MapReduce 框架。

实验中以 “coordinator” 来替代 paper 中的 “master”

顺序版 MapReduce

试执行

• src/main/mrsequential.go 是一个简单的、已完成的顺序版 MapReduce 实现。它在单个进程中，一个接一个依次执行 Map、Reduce 操作。
  • 输入文件为 pg-xxx.txt
  • 输出文件为 mr-out-0
• 实验提供了几个 MapReduce 应用示例：
  • mrapps/wc.go：词频统计
  • mrapps/indexer.go：文本索引器

可以按照以下步骤来运行词频统计程序

$ cd ~/6.5840
$ cd src/main
$ go build -buildmode=plugin ../mrapps/wc.go  #编译到当前目录下，生成wc.so
$ rm mr-out*    #清理旧输出
$ go run mrsequential.go wc.so pg*.txt #编译并运行单机顺序版的MR框架主程序，传入刚刚编译好的词频统计插件，并传入待处理的输入文本文件。
$ sort mr-out-0

# 1. 编译 MapReduce 应用插件
go build -buildmode=plugin ../mrapps/wc.go

# 2. 编译主程序引擎，生成可执行文件 mrsequential
go build mrsequential.go

# 3. 清理旧输出
rm mr-out*

# 4. 直接运行编译好的二进制文件
./mrsequential wc.so pg*.txt

# 5. 排序查看结果
sort mr-out-0

mrsequential.go 中的代码

实验采用的排序方法是 sort.Sort：定义一个自定义类型ByKey，并实现sort.Interface接口，从而实现ByKey 类型变量的排序。这是 Go 最原始的排序方式，目前在新版本中已被 slices 包取代。：具体见Go语言：从排序功能看接口、反射、闭包、泛型

实验代码中采用 os.Open + ioutil.ReadAll + file.Close进行“打开文件、读取全部内容、自动关闭文件”，这个过程已经被os.ReadFile(filename) 替代：Go语言中的标准库-基础输入输出

它的核心作用是：在单个进程中，按顺序模拟运行标准 MapReduce 的完整生命周期（动态加载插件 $\to$ 执行 Map $\to$ 按 Key 排序归并 $\to$ 执行 Reduce $\to$ 输出到文件）。

main 函数逻辑如下：

1. 初始化与插件加载：
   1. 命令行参数校验。
   2. 用 loadPlugin 动态提取用户自定义的 Map 函数和 Reduce 函数。
2. Map 阶段：
   1. 依次遍历命令行参数中给出的文件，读取文件内容。对于每个文件：
      1. 将内容和文件名传入 mapf 函数，获得一批中间键值对
      2. 将键值对合并到内存切片 intermediate[]
3. Shuffle 阶段（分组与排序）：
   1. 对内存中收集到的所有中间键值对按照 Key 排序。

通过排序，保证所有相同 Key 的数据在内存数组中处于连续紧邻的位置，为接下来的合并分组做准备。

真正的分布式 MapReduce 会将中间数据按 Hash 分发到 $M\times N$ 个 Bucket 中；而这里的串行版本直接将所有数据存在单一内存空间里统一排序。

4. Reduce 阶段
   1. 每次循环处理一个 key（分组）
      1. 将所有 key 相同的键值对提取到 values[] 中
      2. 将 Key 和 values[]传入reducef 函数，进行化简。
      3. 将这个 key 的最终化简结果打印到指定文件中。进行下一个循环

loadPlugin 函数（命令行参数为 wc.so，并以map 的加载为例）

1. plugin.Open(): 将 wc.so 载入当前进程的内存地址空间
2. p.Lookup("Map")：在符号表中查找 "Map" 内存地址，返回一个空接口类型 plugin.Symbol,表示一个通用的内存地址指针，Go 还不知道这个符号具体是什么类型的函数或变量。
3. 类型断言：将 xmapf强转为特定的函数类型（与Map 的函数签名保持一致），返回函数指针，赋值给mapf。

mapf := xmapf.(func(string, string) []mr.KeyValue)

4. 提取出的 mapf 作为普通的 Go 函数变量返回。

func Map(filename string, contents string) []mr.KeyValue

实验任务概览

实验内容

写两个程序：coordinator.go 和 worker.go

• 并行执行一个 coordinator 进程和一个或多个 worker 进程
• worker 进程通过 RPC 和 coordinator 通信

官方给出了两个初始代码，帮忙起步：coordinator（main/mrcoordinator.go）和 worker(main/mrworker.go)。这两个文件不能修改

我们需要把实现写在 mr/coordinator.go、mr/worker.go、mr/rpc.go

运行示例

以word-count 应用为例。

1. 编译 word-count 插件

cd MIT6.5840/6.5840/src/main
go build -buildmode=plugin ../mrapps/wc.go

2. 在其中一个命令行窗口，运行 coordinator
   1. sock123 参数指定了一个套接字（socket），协调器（coordinator）通过它来接收来自 Worker 节点的 RPC 请求。
   2. pg-*.txt 参数则是输入文件；每个文件代表一个“数据分片”（split），并且会作为单个 Map 任务的输入。

cd MIT6.5840/6.5840/src/main
rm mr-out*
go run mrcoordinator.go sock123 pg-*.txt

3. 在另一个命令行窗口，运行一些 worker

cd MIT6.5840/6.5840/src/main
go run mrworker.go wc.so sock123

4. workers 和 coordinator 完成后，输出结果在 mr-out-*。输出结果应该和src/main/mrsequential.go 的实现结果是一致的。

测试

官方提供了测试代码（位于 mr/mr_test.go），可以在src 目录下运行测试：

• 正确性：它会检查 wc和indexer 两个 MapReduce 应用是否能产生正确输出
• 并发：Map 和 Reduce 任务是否是并行运行
• 容错：Worker 节点发生崩溃时，系统能否成功进行故障恢复。

cd src
make mr

预期输出如下：

=== RUN   TestWc
--- PASS: TestWc (8.64s)
=== RUN   TestIndexer
--- PASS: TestIndexer (5.90s)
=== RUN   TestMapParallel         # 测试 Map 是否并行
--- PASS: TestMapParallel (7.05s)
=== RUN   TestReduceParallel      # 测试 Reduce 是否并行
--- PASS: TestReduceParallel (8.05s)
=== RUN   TestJobCount            # 测试 Job 计数
--- PASS: TestJobCount (10.04s)
=== RUN   TestEarlyExit
--- PASS: TestEarlyExit (6.05s)
=== RUN   TestCrashWorker         # 测试 Worker 崩溃容错
2026/01/22 14:58:14 *re*-starting map ... # 看到重新启动任务的日志
--- PASS: TestCrashWorker (40.18s)
PASS
ok      6.5840/mr       86.932s   # 总用时 1 分多钟

如果出现 2026/02/11 16:21:32 dialing:dial unix /var/tmp/5840-mr-501: connect: connection refused 这种报错日志，是正常的。当所有任务做完后，Coordinator 进程会先退出并关闭 RPC 服务。此时可能还有一些“不知情”的 Worker 还在尝试通过 RPC 连 Coordinator，就会报“连接被拒绝”。

规则

• Map 阶段需要将 intermediate key 拆分成 nReduce个桶。每个 Mapper 都需要创建nReduce 个 intermediate 文件供 reduce tasks 使用。
• 应该把 Map 阶段的输出放在当前目录下，方便 Reduce 任务进行读取。
• 第 X 个 Reduce 任务的输出必须写到名为 mr-out-X 的文件中（如mr-out-3）
• mr-out-X 文件中，每一行分别表示一个 Reduce 函数的输出。输出格式为 "%v %v"，分别表示 key 和 value。（可以参考main/mrsequential.go）
• 可以修改 mr/coordinator.go、mr/worker.go、mr/rpc.go。也可以临时修改main/目录下的文件来做本地 Debug，但是官方评测时会重置并使用原始版本的 main/ 文件。
• mr/coordinator需要实现 Done()。mrcoordinator.go 会不断调用 coordinator.Done()。只有当 Map 和 Reduce 所有任务都彻底完成时，Done() 返回 true，mrcoordinator 进程才会安全退出。
• Job 结束后，Coordinator 会退出。Worker 需要感知到这一点并且终止进程。

建议：

• 把 Map 阶段输出的中间文件命名为 mr-X-Y
  • X 表示 Map Task 编号（第几个 Map 任务）
  • Y 表示 Reduce Task 编号（分区 ID）
• 设置 Worker 超时时间为 10s

我的实现流程

我的实现是基于 MapReduce 设计流程的。根据设计，来思考实现方式。
注意：

• 本节中的行文思路并非是结论式的，而是过程式的，它记录的是我在实现过程中如何推进、如何决策，而不是纯粹的结论。
• 最好先看 MapReduce 设计流程 ，再看本文。两篇文章有一定的冗余阐述，但这是为了加深理解，本文的实现不能代替前者的设计思路。
• 折叠的引用块代表扩展的内容，并非实验的重点。但如果你感兴趣，也值得一看。

首先要考虑从哪入手。我考虑过两个方向：

1. 从顶向下：先查看课程给出的所有函数，分析它们的调用关系与参数，再依次填充每个函数的内部逻辑。这样做的问题是，实现会退化成纯粹的工程问题——各个函数之间的调用关系是交叉的，孤立地逐个填充函数，很容易让思路陷入混乱。
2. 按流程推进。根据 MapReduce 设计流程，按顺序推进函数逻辑。简单的说，我遇到什么新问题，我就写什么代码。

• 这一部分的内容是实验中直接给出的，我们无需修改。但我们仍然需要知道实验给出的前提条件是什么，才能基于这些推进流程。

初始数据：src/main/pg-*.txt

设置参数（这些参数都是已经设定的、无需修改的）：

• M：即传给 Coordinator 的输入文件的数量。
• R：即 src/mrcoordinator.go中，调用mr.MakeCoordinator 时传入的参数nReduce: 10
• 哈希分区函数：（由于是在 Map 阶段进行分区的，所以应该在 worker.go 中寻找）ihash（用的是 FNV-1a 算法）

阶段一：输入分片与任务分配

注意：实验中 Coordinator 等价于论文中的 Master

这个阶段中，我们需要做什么？

首先要看我们有什么。实验通过通过命令 go run mrcoordinator.go sock123 pg-*.txt 创建一个 Coordinator 进程。因此，Coordinator 所拥有的数据是：

• sockname（相当于 IP:Port，用来标识 Coordinator 进程的位置）
• 一系列输入文件

我们想要完成的是输入分片与任务分配。对于输入分片：

根据 MapReduce 设计流程，我们需要完成物理存储分块和逻辑分配映射两个任务。

• 物理存储分块是由 Google 的 GFS 实现的，将大文件按大小物理切分为多个数据块（Block），分散存储在不同节点的本地磁盘。而我们的实验环境压根就没有分布式文件系统，因此我们只能略过这一步。
• 逻辑分片映射是 MR 框架实现的，Coordinator 需要向文件系统查询，将 Block 映射成逻辑上的 Input Splits（即 Map Tasks）。我们假设：
  • 每个 pg-*.txt 文本文件恰好是一个数据块（Block）
  • Coordinator 将这些 Block 映射为 Input Split，分片元数据用文件名表示。
  • 不考虑数据本地化调度。

因此，输入分片任务无需我们完成，实验已经提供了假设和前提（既是简化，也是约束）。我们只需考虑任务分配。

从直觉上看，任务分配有两种方向：

1. 推送（Push）：由 coordinator 主动把任务分配给 worker。
   • coordinator 必须实时掌握每一个 worker 的状态，对于分布式系统来说，状态同步、网络变化，都是难以维护的。这可能意味着要实现更复杂的容错。
2. 拉取（Pull）：当有 worker 空闲的时候通知 coordinator，再由 coordinator 将任务分配给该 worker。
   • 它天然实现了负载均衡（任务分配由 Worker 的处理速度驱动，而不需要 Coordinator 进行计算）。降低了 Coordinator 对 Worker 状态的依赖性。

因此，我们采用 Pull 的方式进行任务分配。但问题是：

1. “任务”在代码里是如何表达的？
2. Worker 是如何通知 Coordinator 的？Coordinator 是如何进行任务分配的？

“任务”如何被表达？

任务的定义

由于“任务”是全局性的，Coodinator 需要维护每个任务的状态，而 Worker 只需要知道当前任务的信息就足够了。因此“任务”的定义是在 coordinator.go 中实现的。

于是，我们观察 coordinator.go，会发现它包含：

• Coordinator 结构体
• Coordinator 结构体的方法
• MakeCoordinator 函数：用于创建 Coordinator 对象，并执行相关操作

可以理解为：结构体对象用于存储数据与维护状态、结构体对象的方法用于执行动作、函数用于编排流程。

在结构体对象中，我们要存储和维护所有任务的列表：

//其中mu、MapCount、ReduceCount都是后来定义的，在这个阶段，我只定义了Tasks
type Coordinator struct {
	mu          sync.Mutex //互斥锁，保护所有字段
	Tasks       []Task
	MapCount    int // 未完成的 Map 任务数
	ReduceCount int // 未完成的 Reduce 任务数
}

对于每个任务，无论是 Map 任务还是 Reduce 任务，都有一致的结构和状态。因此我们再定义 Task 类型结构体：

//Task类型的定义同样是我在实现的过程中慢慢修改，不断完善的。比如AssignedAt字段是我在写容错阶段的代码时才加上的。
type Task struct {
	TaskID     int
	TaskType   TaskType
	Status     TaskStatus
	FileName   string
	NReduce    int       //Map任务需要知道NReduce才能将键值对分配到对应的哈希桶
	AssignedAt time.Time //任务分配时间
}

最初我考虑过用 MapTask 和 ReduceTask 两个不同的结构体。但它们有大量的代码冗余，且在 Coordinator 眼里，Map 任务和 Reduce 任务本质上都是一个“需要被分发、跟踪超时、等待完成的有限状态机”。

任务的创建

自然的，我们要接着考虑把数据存入 Coordinator 中。即创建任务。代码逻辑如下：

1. 初始化任务列表（即切片）
   • 这里我用的初始化方法是 s := make([]T, 0, 100)，即声明时指定长度和容量。这是工程上的最佳实践。
2. 给每个输入文件创建一个 map 任务
3. 创建 nReduce 个 reduce 任务

任务分配如何进行？

Worker 是如何通知 Coordinator 的？Coordinator 是如何进行任务分配的？由于 Worker 和 Coordinator 是不同的进程，所以任务分配，实则是进程间通信问题。

Coordinator 和 Worker 通信发生在以下两个情境：

• Worker 向 Coordinator 请求任务，Coordinator 返回任务
• Worker 完成任务后告知结果，Coordinator 无需响应

它的通信特性是点对点、请求 - 响应、同步。我们选择直连调用的技术路线。这种技术路线下，可以选择的技术方案有：RPC 和 REST/HTTP。REST 把系统抽象成一组可操作的资源，RPC 把系统抽象成一组可调用的函数（即一组行为）。MapReduce 的通信语义是行为上的，比如“申请分配任务”、“告知任务已完成”，如果用资源来表达会产生错配，因此我们采用 RPC 的技术方案。

go 语言中，RPC 有多种实现，它们有相同的通信特性，但有不同的工程实现。

• net/rpc：标准库，不需要外部依赖，实现简单，不支持跨语言。
• gRPC Unary：实现复杂，性能更强，适合生产级服务

gRPC 有四种模式，其中 gRPC Unary 是 RPC 模型的直接实现，其她三种模式是 gRPC 借用了 RPC 的接口定义能力，但超出了 RPC 概念所定义的边界。

在实验中，我们的主要目的是实现逻辑而非关注通信细节，net/rpc 让我们把重心放在把设计工程化上。

func (t *T) MethodName(args *ArgsType, reply *ReplyType) error

RPC 能够让 Worker 像调用本地函数一样调用远程方法。在任务分配时，Worker 调用 Coordinator 的 AssignTask 方法，Coordinator 在方法内选择一个任务填入响应结构体，返回给 Worker。

实验中，注册并启动服务端这一步在已经被封装进 server方法，无需实现；客户端连接服务器、发起调用的具体细节也被封装进call 函数，无需实现；

我们只需实现：

• 定义消息结构（rpc.go 中的请求/响应参数结构体：Args/Reply）
• 服务端（Coordinator）处理方法（AssignTask 方法）
• 客户端 (Worker) 调用时传入的方法名、请求和响应

1.定义消息结构

通信实现的是：Coordinator 将任务分配给 Worker。因此传输的消息如下：

• 请求参数为空，Worker 发送消息意味着它处于空闲状态，可以接收任务
• 响应参数为一个具体的任务。Coordinator 选择一个任务交付给 Worker

type AssignTaskArgs struct {
}
type AssignTaskReply struct {
	Task Task
}

2.Coordinator 分配任务

接下来，一旦有 Worker 调用 AssignTask（即向服务端 Coordinator 发起通信），Coordinator 会分配一个未完成的 Map 任务给它（作为响应参数）。代码逻辑如下：

1. 如果还有 Map 任务没有完成，先分配 Map 任务
   • 循环找到一个未分配的 Map 任务。如果所有 Map 任务都分配完，则给 Worker 分配一个空任务
2. 所有 Map 任务完成后，再分配 Reduce 任务
   • 循环找到一个未分配的 Reduce 任务。如果所有 Reduce 任务都分配完，则给 Worker 分配一个空任务
3. 如果全部任务都已完成，则返回关闭任务

任务全都分配完，并不等于任务全都已完成

这里需要注意两点：

• 判断 Map 任务或 Reduce 任务是否已全部完成，需要用计数器。因此我在 Coordinator 类型结构体中定义了 MapCount 和ReduceCount。
• 在选择具体的任务进行分配时，需要修改任务的状态（从空闲状态改为运行状态），涉及写操作。因此在 AssignTask 中，同一块内存会被多个 goroutine 并发访问，且存在写操作，我们需要对共享数据加锁

c.mu.Lock()         //加锁
defer c.mu.Unlock() //解锁

c.mu.Lock()         //加锁
//临界区代码
c.mu.Unlock() //解锁

3.Worker 请求任务

前面提到，实验已经把调用细节封装进 call 函数中，我们只需要传入方法名、请求和响应参数，就可以实现通信。

worker.go 中已经给出了请求调用的实例代码 CallExample，模仿它写一个真实的请求分配任务的代码即可。

func getTask() (Task) {
	var args AssignTaskArgs
	var reply AssignTaskReply
	ok := call("Coordinator.AssignTask", &args, &reply)
	if !ok {
		return Task{}
	}
	return reply.Task

}

在主函数逻辑中，一个 Worker 的生命周期是：

1. 空闲
2. 无限循环请求分配任务（每次循环 getTask 一次）
   1. 分配到 Map 或 Reduce 任务
      1. 调用对应的任务执行函数（mapTask/reduceTask）
      2. 执行完毕，通知 Coordinator（我把这部分的代码放在了任务执行函数内部，因此目前阶段无需实现）
   2. 分配到空任务，等待一段时间后重新请求
   3. 分配到关闭任务，Worker 退出进程。

因此主函数里写一个循环，每次循环 getTask 一次，接着根据不同任务类型来处理任务。

阶段二：Map

这一阶段，我们要写的是 worker.go 中的 Map 任务执行函数，即mapTask。

1. 根据文件名读取文件内容，将它们作为参数传入 Map 函数，获得中间键值对 kva（key-value array）
2. 对中间键值对排序 + 分区
3. 将中间键值对存入 mr-X-Y 文件（一个分区对应一个中间文件）
4. 通知 Coordinator 任务已完成。

第一步的实现十分简单，我们讨论步骤二、三、四的实现。

排序与分区

我们已经拿到了该 Map 任务的全量中间键值对（即切片变量 kva），存储在内存中。

对于中间键值对，可以采取两种方案：

假设中间键值对总数为 N，分区（Reduce）数量为 R，且数据均匀分布。

1. 先全量排序再分区。排序的时间复杂度为 O(N logN)，分区的时间复杂度为O(N)，总时间复杂度为O(N logN)
2. 先分区再局部排序。分区时间复杂度为 O(N)，排序的时间复杂度为R * O((N/R) log(N/R))，总时间复杂度约为O(N log(N/R))

因此，mapTask 中这一部分的实现逻辑如下：

1. 遍历全部中间键值对，对其进行分区，存入对应的 buckets[i]
2. 遍历每个 buckets，对桶内的切片调用 slices.SortFunc 进行排序。（采用这种排序方法的原因见Go语言：从排序功能看接口、反射、闭包、泛型）

写入中间文件

目前为止，我们得到了各个分区的有序数组 buckets[i]。要将数据写入文件，有许多种方法，应该如何决策呢？

type Student struct {
    ID   int
    Name string    // string 本质是一个指针 + 长度，指向堆内存的另一块区域
    Tags []string  // slice 本质是一个指针 + 长度 + 容量，指向堆内存的又一片区域
}

[ 内存地址 A ] Student 结构体主体 (包含 ID，以及指向 Name 和 Tags 的指针)
     │
     ├───► 指向 ───► [ 内存地址 B ] Name 的真实字符串数据 ("Alice")
     │
     └───► 指向 ───► [ 内存地址 C ] Tags 的切片底层数组 ("Math", "CS")

[ 序列化后连续的内存空间：[]byte ]
┌──────────┬─────────────┬─────────────────┐
│ ID: 1001 │ Name: Alice │ Tags: Math, CS  │
└──────────┴─────────────┴─────────────────┘
 (首尾相连，没有指针，不依赖任何特定内存地址)

如果追求极致性能，序列化为二进制格式是最佳的选择。但人类不可读。而实验中，我们需要查看中间文件来进行调试，最终输出文件也要求是人类可读的格式以通过测试。因此，我们只能为序列化为文本字符流。

由于我们编写的是通用的框架，用户输入的文件是不可控的，文件中可能存在 \n、Tab 键等特殊字符。牺牲部分性能，获取代码的健壮性与稳定性，在实验中是合理的选择。因此我们最终采取的序列化方案是 JSON 格式。在 Go 语言中，使用 json.NewEncoder 和 json.NewDecoder 方法（即 Go 标准包 encoding/json）

此外，由于我们处理的是大量数据，应当选择缓冲写入的方式。

通知 Coordinator 任务已完成

Map 任务完成后，需要发起 RPC 通知 Coordinator 任务已完成。

• Worker 新增一个 doneTask 函数，用于发起 RPC
• Coordinator 新增一个 DoneTask 函数，找到对应已完成的任务，标记其已完成，并修改对应任务的计数器（计数器需要添加锁）。

阶段三：Shuffle、Sort 与 Reduce

当所有 Map 任务完成后，才会开始这一阶段，

worker 从 Coordinator 获取任务，如果任务类型是 Reduce，reduceTask 函数要完成的任务是：

1. 从各个中间文件中读取自己分区的数据。
2. 将多个中间文件中的数据进行排序，局部有序变成全局有序。
3. 按 key 分组，并将分组结果依次传入自定义的 Reduce 函数，获取 key 值对应的结果
4. 所有结果写入最终文件

Map 阶段生成的每个文件都是有序的，因此对于第一步、第二步，实则完成的是同一个目标：将 N 个有序的小文件，合并成 1 个全局有序的大文件。这是多路归并问题。

在实现多路归并的过程中，我们发现，每次弹出的记录是有序的，也就是 key 值相同的键值对会接连弹出。因此，我们可以在归并的同时，直接进行分组、reduce

因此整个 reduceTask 函数的流程为：

1. 【初始化】匹配并打开所有中间文件，为每个文件创建一个 json.Decoder，用于逐行读取 JSON 格式的键值对
2. 【初始化】创建一个小根堆（优先队列），进行堆初始化（所有文件的第一个键值对入堆）
3. 【初始化】创建临时输出文件 mr-out-{TaskID}.tmp* 与缓冲写入器。（最终输出文件将通过原子重命名生成，以避免写入过程中因程序崩溃而留下不完整文件。）
4. 【初始化】维护一个 groups，用于暂存当前正在处理的同一 Key 的所有 Value。
5. 循环归并与分组：循环执行以下操作，直到堆为空：
   • 弹出堆顶元素：从堆中弹出 Key 最小的记录（当前所有文件中最小的一条）。
   • 判断 Key 是否变化：
     • 若当前 Key 与 groups.Key 不同，说明上一分组已完整，调用 reducef 处理该分组，将结果写入输出缓冲区，然后重置分组为当前 Key。
     • 若相同，则将当前 Value 追加到 groups.Values 中。
   • 补充下一行数据：从该记录来源的文件中尝试读取下一行键值对。若读取成功，则将其重新推入堆；若遇到 io.EOF，则关闭该文件并不再补充新元素（堆大小会随之减少）。
6. 【收尾】处理最后一个分组
7. 【收尾】所有分组处理完成后，将缓冲区中的剩余数据写入磁盘，然后关闭临时文件。
8. 【原子提交】将临时文件重命名为最终文件 mr-out-{TaskID}。
9. 向 Coordinator 报告任务已完成。

阶段四：收尾

当所有任务结束后，MapCount和 ReduceCount都为 0。mrcoordinator周期性调用Done() 了来判断 Coordinator 任务是否已经完成。

横切关注点

日志

最初，coordinator 和 worker 运行后，控制台没有任何输出，我不知道这些进程正在做什么，卡在哪一步。因此，我需要添加日志，来确认每个进程都在正常运行。

由于课程的要求：

1. 输出结果要干净
2. 不能修改 main 函数所在的文件
   因此，我把 logger 的初始化放在了 coordinator.go 和 worker.go 的入口。并且通过设置 slog.HandlerOptions 来一键显隐不同的日志等级（非常好用！）

由于我需要进行多进程的监控，因此我初始化了一个带有 PID 的 logger。

//初始化logger
func initLogger() {
	handler :=slog.NewTextHandler(os.Stdout, &slog.HandlerOptions{
		Level: slog.LevelDebug})
	logger := slog.New(handler).With("Coordinator.PID", os.Getpid())
	//设置为全局默认Logger
	slog.SetDefault(logger)
	slog.Info("Coordinator 已成功启动")	
}

运行 Coordinator 和 Worker，输出日志如下：

Worker 进程：

Coordinator 进程

并发控制

根据日志输出可知，代码能够实现基本的流程。但在 makr mr 后，出现 DATA RACE 报错。

报错指出：main 协程在调用 Coordinator.Done() 读取数据时，RPC 协程（Goroutine 28）正在调用 Coordinator.DoneTask() 修改数据，且两边没有加锁互斥，访问了同一个内存地址（0x00c000210198）。

因此，我虽然完成了基本的业务逻辑代码，但没有做好临界区的处理。那么有哪些需要处理的临界区变量？

• Coordinator 和不同 Worker 所处的进程空间不同，它们不会访问同一个内存地址。
• 单个 Worker 内部所有的任务处理都是线性的，也无需处理。
• Coordinator 内部，同一时间会启动多个协程
  • main 在循环执行 Done() 时，读 MapCount 和 ReduceCount
  • http.Serve 由 main 函数显式调用，会常驻内存，
  • 每个 RPC 请求都会调用一个 gorotine 来跑对应的函数。如果有 N 个 Worker 同时发起任务分配请求，M 个 Worker 通知 Coordinator 任务已完成，则会有 2+M+N 个协程运行。

为了防止多个协程同时读写计数器、修改任务状态。我们需要给 Coordinator 实例加上互斥锁。对数据进行读写时，都需要加锁

type Coordinator struct {
	mu sync.Mutex   //互斥锁，保护所有字段
	Tasks []Task
	MapCount int
	ReduceCount int
}

例如在分配任务时：

func (c *Coordinator) AssignTask(args *AssignTaskArgs, reply *AssignTaskReply) error {
	c.mu.Lock()         //加锁
	defer c.mu.Unlock() //解锁
	...	
}

容错处理

TestJobCount 测试失败

• 在我最初的设计中，只要有 worker 请求任务，且存在未完成的任务，那么 worker 就能领取到任务，即使这个任务没有超时。这样能够天然实现负载均衡（1.分配一个未完成的、类型为 t 的任务；2.如果全都分配完，但是仍有任务未完成，则分配一个已分配时间最长的任务给 Worker）。但是这无法通过 TestJobCount 测试。实验的要求是：如果任务已分配，未超时，那么它就不能被分配给其她 Worker
  • 从功能性角度上来说，我的逻辑是正确的。但这是一种激进的重试。标准的推测执行是只在拖后腿的任务上启动备份，用少量冗余计算换取总执行时间的缩短，整体资源开销小；我的执行会耗费大量的计算资源和网络带宽，造成 CPU 和内存的浪费。
  • 因此，我需要修改分配的逻辑，Coordinator 只能分配空闲任务；并开一个协程，来定期检测各个任务是否超时未完成，超时的运行中任务的状态设置为空闲，

代码中的错误与调试

• 任务失败后，使用了 log.Fatalf来记录日志。log.Fatalf会指定os.Exit(1)来杀死进程。应该用slog.Error来打印错误日志，并用continue 进入下一次循环来领取任务。
• 我使用 bufio.Write写文件，但是由于关闭文件前没有调用Flush() 刷盘，因此缓冲区中残留的数据没有写入硬盘。导致我每个文件 s~z 的数据丢失。
• 如果使用 range 循环，要修改数组中的数据，则必须用下标来访问数据，而不是直接修改。比如：

for _, task := range c.Tasks{
	if task.TaskType == t && task.Status == Idle{
		task.AssignedAt = time.Now()  //这里的task仅仅是一个副本，并不会修改真正的task的值
		task.Status = Running
		return task
	}
}