宁静的夏天
天空中繁星点点
心里头有些思念
思念着你的脸
——宁夏
完整代码见: https://github.com/SnowLegend-star/6.824
由于这个lab整体难度实在不小,故考虑再三还是决定留下代码仅供参考
6.824的强度早有耳闻,我终于也是到了挑战这座高峰的时候。最开始看完《MapReduce: Simplified Data Processing on Large Clusters》这篇论文的时候,感觉没什么很大的理解难度,比之我平时看的论文有所不如。谁知真正实现它着实让我吃了不少苦头。
首先就是对golang语法不甚熟悉的问题,好几次设计思路有了却不知道怎么具体用golang实现出来,有种一拳打在棉花上的感觉。齐次就是这次是真要直面文件操作了。记得当初c语言最后学到文件操作的时候,我死活理解不了文件的各种操作,感觉抽象无比。在那之后每次遇到文件操作我都有种怯战的感觉。再者平时基本也遇不到文件操作,渐渐文件处理就成了我心里一片挥着不去的阴霾。而MapReduce的本质就是各个worker对任务文件进行处理存储的过程实现,这次我一定要勇敢地克服恐惧,把失去的尊严夺回来。最后就是多文件式编程。虽说以前也遇到过不少多文件编程,6.s081就是如此。但基本都是在各个文件做改进。这次coordinator.go、rpc.go、worker.go三个文件基本全是自己编写,第一次完成这种工作难免让人手足无措。有时候对着当前文件一顿猛看,结果发现是另一个文件出问题了,当然这是后话。
言归正传,下面分析下MapReduce的实现过程。实验说明和Lab原代码要多看几遍,力求理解透彻。
你的任务是实现一个分布式 MapReduce 系统,包括两个程序,coordinator 和 worker。将只有一个 coordinator 进程,和一个或多个并行执行的 worker 进程。在实际系统中,worker 将运行在不同的机器上,但在这个实验中,你将在一台机器上运行它们。worker 将通过 RPC 与 coordinator 通信。每个 worker 进程将在循环中执行以下操作:向 coordinator 请求任务,从一个或多个文件中读取任务的输入,执行任务,将任务的输出写入一个或多个文件,然后再次请求新的任务。coordinator 应该注意到如果一个 worker 在合理的时间内没有完成任务(在本实验中为十秒),并将同一任务交给另一个 worker。
这里就开始给出实验要求了:
- worker要在循环中请求任务,而不是完成了一个任务就直接结束
- worker应在10s完成任务并给master发送完成signal。同时要创建本地文件把任务处理结果存储在其中。
- master如果10s没收到该worker完成任务的signal,就把这个交给其他worker重新实现
我们在 main/test-mr.sh 中提供了一个测试脚本。测试检查 wc 和 indexer MapReduce 应用程序在输入 pg-xxx.txt 文件时是否产生正确的输出。测试还检查你的实现是否并行运行 Map 和 Reduce 任务,以及你的实现是否能从 worker 崩溃中恢复。
由于忽略了上面这句话,我一直对worker要处理什么类型的任务百思不得其解,同时Lab到底用的是什么plugin进行测试抱有疑问。后面还对自己从代码中推断出了plugin应该是和wc.so类似而感到沾沾自喜,直到后面无意中看到了这句话。真是自己一个小时的思考不及多用一分钟阅读这句话,堪称降维打击。
func Worker(mapf func(string, string) []KeyValue,
reducef func(string, []string) string) {}
既然worker要用到map和Reduce处理文件,那应该从哪里读取这些文件呢?答案是通过rpc向master发出任务请求从而获取文件。
论文中的这张图一定要完全理解透彻,不能有些许马虎。
A few rules:
1、Map 阶段应将中间键分成 nReduce 个 Reduce 任务的桶,其中 nReduce 是 Reduce 任务的数量——main/mrcoordinator.go 传递给 MakeCoordinator() 的参数。每个 Mapper 应创建 nReduce 个中间文件供 Reduce 任务使用。
按照MapReduce给的execution overview中,worker在进行map阶段的任务文件处理时,应该生成本地文件并将中间键存储在其中。这里为什么说worker在map阶段呢?难道同一个worker执行完map任务后还可以继续执行Reduce任务吗?Exactly!它甚至在执行完了当前map任务后还可以请求master再给它来点别的map任务。
等到所有map任务结束后,master就开始给worker分配Reduce阶段的任务。
2、worker 实现应将第 X 个 Reduce 任务的输出放在文件 mr-out-X 中。
output文件的名字为mr-out-X,这是Reduce阶段的任务了。
3、一个 mr-out-X 文件应包含每个 Reduce 函数输出的一行。该行应使用 Go 的 "%v %v" 格式生成,使用键和值进行调用。查看 main/mrsequential.go 中注释 "this is the correct format" 的行。如果你的实现与此格式差异过大,测试脚本将失败。
就是用
fmt.Fprintf(ofile, "%v %v\n", key, result)
这种格式把键值对写入output文件中。
4、你可以修改 mr/worker.go、mr/coordinator.go 和 mr/rpc.go。你可以临时修改其他文件进行测试,但确保你的代码与原始版本一起工作;我们将使用原始版本进行测试。
作为新手,我们要有菜鸟的自知之明。能不改的地方尽量不去动,免得给后续实现埋坑。完成Lab后倒是可以自由修改任何地方。
5、worker 应将中间 Map 输出放在当前目录中的文件中,以便你的 worker 在 Reduce 任务中可以读取它们。
这条rule难道不应该放在②吗?Map阶段处理好的intermediate键值对切片要存储起来,供后续Reduce阶段进行读入。输出文件名称有具体要求吗?mr-X-Y。这次的实验说明的内容顺序堪称大坑。混乱的要求次序让人摸不着头脑。
6、main/mrcoordinator.go 期望 mr/coordinator.go 实现一个 Done() 方法,当 MapReduce 作业完全完成时返回 true;此时,mrcoordinator.go 将退出。
如何判断所有任务文件都已经完成了?在master(即coordinator.go)中维护一个Reduce阶段所有任务的状态数组。每当worker完成了一个任务,就向master发送一条signal,master即更新状态数组。
其实也应该为map阶段维护一个数组。毕竟要等所有map任务完成才可以开始分配Reduce阶段的任务。
7、当作业完全完成时,worker 进程应退出。实现此功能的一个简单方法是使用 call() 的返回值:如果 worker 无法联系到 coordinator,可以假定 coordinator 已经退出,因为作业已经完成,因此 worker 也可以终止。根据你的设计,你可能还会发现有一个 "please exit" 伪任务供 coordinator 给 worker 是有帮助的。
这里注意是“当作业完全完成时,worker 进程应退出”而不是完成某项任务worker进程就自顾退出跑路了。
看完了rules之后,我们再来看看hints。其实两者有些内容是重复的,这也是为什么我说实验说明排版混乱的原因。
Hints:
1、Guidance 页面有一些开发和调试的技巧。
2、开始的一个方法是修改 mr/worker.go 的 Worker(),使其发送一个 RPC 请求给 coordinator 要求任务。然后修改 coordinator 以响应一个尚未开始的 Map 任务的文件名。然后修改 worker 以读取该文件并调用应用程序 Map 函数,如 mrsequential.go 中所示。
这里建议多阅读mrcoordinator.go、mrnetworker.go、mrsequential.go等文件,wc.go和indexer.go也要理解透彻。
然后我们可以从运行networker.go中提供的example方法着手。先把它们运行起来,再对example方法进行小小的改动,观察call()方法的执行流程。
3、应用程序 Map 和 Reduce 函数在运行时使用 Go 的 plugin 包加载,文件名以 .so 结尾。
4、如果你更改了 mr/ 目录中的任何内容,可能需要重新构建你使用的任何 MapReduce 插件,如 go build -buildmode=plugin ../mrapps/wc.go。
更改了mr/目录下的任何文件后,应该重构xx.so文件,如果继续使用原来的xx.so文件可能会报如下错误
修改语句如下
go build -race -buildmode=plugin -gcflags="all=-N -l" ../mrapps/wc.go
5、此实验依赖 worker 共享文件系统。当所有 worker 在同一台机器上运行时,这很简单,但如果 worker 在不同的机器上运行,则需要全局文件系统如 GFS。
6、中间文件的合理命名约定是 mr-X-Y,其中 X 是 Map 任务编号,Y 是 Reduce 任务编号。
怎么确定Map的任务编号呢?可以先在master给每个任务文件编号,并用一个切片记录下来。回复worker的任务请求时返回这个文件编号。
7、worker 的 map 任务代码需要一种方式将中间键/值对存储在文件中,以便在 Reduce 任务期间正确读回。一种可能性是使用 Go 的 encoding/json 包。
Q:我直接把中间键值对存在普通文件中进行读取有什么不同呢
A:将中间键值对存储在普通文件中,而不是使用 JSON 格式,有几个关键的不同和潜在的挑战:
1. 数据结构的复杂性
- 普通文件: 如果你使用普通文件存储中间键值对,通常需要自行设计和实现数据的编码和解码方式。你需要确保数据格式的一致性,以及处理可能的特殊字符(如换行符、分隔符等)。
- JSON: 使用 JSON 可以自动处理复杂的数据结构。Go 的 encoding/json 包可以直接将结构体、切片等复杂数据类型序列化和反序列化,无需额外处理。
2. 可读性和调试性
- 普通文件: 如果使用自定义格式存储数据,文件的可读性取决于设计的格式。如果格式复杂或不直观,调试时会比较困难。
- JSON: JSON 是人类可读的文本格式,结构清晰,容易理解和调试。开发者可以直接查看 JSON 文件内容,验证数据是否正确。
3. 跨语言兼容性
- 普通文件: 自定义格式可能需要特定语言或库的支持。如果需要与其他语言编写的系统交互,可能需要额外的工作来解析和生成这种格式。
- JSON: JSON 是一种标准化的数据交换格式,几乎所有编程语言都支持 JSON 的解析和生成,跨语言兼容性好。
4. 编码和解码的实现
- 普通文件: 你需要自己编写编码和解码逻辑,可能涉及字符串处理、文件读写等操作。实现起来可能会比较繁琐,并且容易出错。
- JSON: 使用 Go 的 encoding/json 包,编码和解码变得非常简单和可靠,只需调用标准库中的函数即可。
总结
尽管使用普通文件存储中间键值对也可以实现,但使用 JSON 文件提供了更高的可读性、调试性、跨语言兼容性和实现简单性。JSON 格式的标准化和广泛支持,使得它成为 MapReduce 中间键值对存储的一个理想选择。
8、worker 的 map 任务可以使用 ihash(key) 函数(在 worker.go 中)为给定键选择 Reduce 任务。
这里属于一个小难点,要用合适的方法将中间键写入NBUCKET个文件中,否则多次遍历intermediate会导致处理任务的用时激增。
我采用的方法如下:
- 先将从任务文件读取的内容content传入mapf()中,普通文件读取用ioutil.ReadAll()即可。得到处理过后的切片kva
- 创建一个二维切片intermediate := make([][]KeyValue, reply.NReduce)
- 遍历kva的同时,利用ihash(key)把kva中的每一个元素分配到Nreduce个intermeditae[index]中
- 最后创建Nreduce个“mr-X-Index”中间文件,把对应的intermediate[index]写入到对应的中间文件内部。记得用json的方法写,这是为了Reduce阶段读取中间文件更为方便
9、你可以从 mrsequential.go 中借用一些代码,用于读取 Map 输入文件,在 Map 和Reduce 之间对中间键/值对进行排序,以及将 Reduce 输出存储在文件中。
map阶段对中间键进行排序我个人觉得没什么必要了,Reduce阶段对output文件内容进行排序这就仁者见仁智者见智。为了结果的可读性可以排序,但是排序相当于再次遍历所有内容,时间复杂度就高了。
10、coordinator 作为 RPC 服务器将是并发的;不要忘记锁定共享数据。
coordinator.go中涉及处理单个与worker有关的数据时,应该加锁。包括给worker分配任务,修改任务状态的切片等。
11、使用 Go 的竞态检测器,使用 go run -race。test-mr.sh 在开头有一个注释,告诉你如何使用 -race 运行。当我们评分你的实验时,我们不会使用竞态检测器。然而,如果你的代码存在竞态,尽管没有使用竞态检测器进行测试,你的代码很有可能会失败。
12、worker 有时需要等待,例如 reduce 任务在最后一个 map 完成之前不能开始。一种可能性是 worker 周期性地向 coordinator 请求工作,在每次请求之间使用 time.Sleep()。另一种可能性是让 coordinator 中相关的 RPC 处理器有一个循环,等待,用 time.Sleep() 或 sync.Cond。Go 在每个 RPC 中运行处理器,因此一个处理器在等待并不会阻止 coordinator 处理其他 RPC。
这里我采用的是在每次请求之间使用 time.Sleep(),先把Lab完成了再回头考虑各种优化细节
13、coordinator 不能可靠地区分崩溃的 worker、活着但由于某种原因停滞的 worker 和执行但速度太慢的 worker。你能做的最好的事情是让 coordinator 等待一段时间,然后放弃并重新分配任务给另一个 worker。对于这个实验,让 coordinator 等待十秒;之后 coordinator 应假定 worker 已经死亡(当然,它可能没有)。
超过10s worker还未发送任务完成的signal,master就重新把任务分配给别的worker。由于map阶段和Reduce阶段的各种操作具有幂等性,所以直接不考虑失联的worker就可以。
14、如果你选择实现备用任务(Backup Tasks,第 3.6 节),注意我们测试你的代码在 worker 未崩溃时不会安排多余的任务。只有在相对较长的时间(例如,10秒)之后才应安排备用任务。
15、要测试崩溃恢复,可以使用 mrapps/crash.go 应用程序插件。它在 Map 和 Reduce 函数中随机退出。
16、为了确保在崩溃时没有人观察到部分写入的文件,MapReduce 论文提到使用临时文件并在完全写入后原子重命名的技巧。你可以使用 ioutil.TempFile(或 os.CreateTemp 如果你运行的是 Go 1.17 或更高版本)创建一个临时文件,并使用 os.Rename 原子地重命名它。
17、test-mr.sh 在子目录 mr-tmp 中运行所有进程,因此如果出现问题并且你想查看中间或输出文件,可以在那里查找。可以临时修改 test-mr.sh,使其在失败的测试后退出,因此脚本不会继续测试(并覆盖输出文件)。
这个建议还是有用的。
18、test-mr-many.sh 多次运行 test-mr.sh,你可能希望这样做以发现低概率的错误。它接收一个参数,表示运行测试的次数。你不应该并行运行多个 test-mr.sh 实例,因为 coordinator 将重用相同的套接字,导致冲突。
19、Go RPC 只发送以大写字母开头的字段。子结构也必须具有大写的字段名。
Golang的需要导出的内容都应该以大写字母开头。
20、调用 RPC call() 函数时,回复结构应包含所有默认值。在调用之前不要设置 reply 的任何字段。如果传递包含非默认字段的回复结构,RPC 系统可能会静默返回错误的值。
我在RPC and Threads中看到了上述内容。再次证明Printf神教yyds哈哈哈!
Bugs
最后介绍下完成Lab过程中遇到的各种bug
1、worker通知master任务已完成时,未传入任务类型,这是导致map无限轮回的第一个原因。
2、修改了上述bug后,我在prinrf大法的路上越走越远。我发现worker在完成任务后会及时通知master,master也能成功接收,但是为什么还是永远都运行map阶段的任务呢?
进行了debug我才发现问题出在文件完成状态判断的函数上
我直接把Done()中判断Reduce阶段所有任务是否全部完成的代码给复制过来了,还忘记修改就直接运行。第二天还忘记这事了。这才导致map无限轮回。
3、处理好map阶段的bugs后,终于要成功了。但是Reduce输出的output文件确实空的。看来是Reduce阶段对文件的读取有问题。
多亏我debug时发现读取函数后返回的err有问题——怎么都还没开始读文件,但是文件指针已经移动到了文件末尾呢?
一番排查后我发现又是复制代码的锅。我为了图方便直接把map阶段的处理代码复制到Reduce阶段的处理函数中,结果下面这句话忘记删了。
这个语句直接读取了文件的所有内容,导致用dec := json.NewDecoder(intermediate_File)的时候文件内部的指针已经移到了文件的末尾。
经过我的不懈努力,终于是完成了MapReduce的实现。
美中不足的是被connection refused克制了~
对了,如果想要进行debug,可以在6.5840的 文件夹下创建launch.json文件
{// 使用 IntelliSense 了解相关属性。 // 悬停以查看现有属性的描述。// 欲了解更多信息,请访问: https://go.microsoft.com/fwlink/?linkid=830387"version": "0.2.0","configurations": [{"debugAdapter": "dlv-dap",// "name": "mrcoordinator",// "name": "mrworker","name": "test_test","type": "go","request": "launch","mode": "auto",// "program": "${workspaceFolder}/src/main/mrcoordinator.go","program": "${workspaceFolder}/src/main/mrworker.go","args": [// "wc.so",// "pg-grimm.txt",// "pg-frankenstein.txt"// "wc.so"],"buildFlags": "-race"}]
}
按照上述格式进行调整就行