官方源码阅读

main/mrsequential.go

先看它的loadPlugin函数：

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
func loadPlugin(filename string) (func(string, string) []mr.KeyValue, func(string, []string) string) {
	p, err := plugin.Open(filename)
	if err != nil {
		log.Fatalf("cannot load plugin %v", filename)
	}
	xmapf, err := p.Lookup("Map")
	if err != nil {
		log.Fatalf("cannot find Map in %v", filename)
	}
	mapf := xmapf.(func(string, string) []mr.KeyValue)
	xreducef, err := p.Lookup("Reduce")
	if err != nil {
		log.Fatalf("cannot find Reduce in %v", filename)
	}
	reducef := xreducef.(func(string, []string) string)

	return mapf, reducef
}

这个函数传入了一个.so常见文件的路径，返回了两个函数

一个是mafp：两个字符串（文件名、文件内容）进一组kv出

一个是reducef：一个key和一组值进入，一个结果字符出

这里的if err != nil是用来记录错误日志的，我们先跳过

那么也就是说这个loadPlugin函数总共做了四步：

p, err := plugin.Open(filename)把文件动态加载到当前进程
xmapf, err := p.Lookup("Map")按照名字"Map"查找符号，拿到interface
mapf := xmapf.(func(string, string) []mr.KeyValue)断言，如果不是这个签名的函数就panic
xreducef, err := p.Lookup("Reduce")同上两步

在这个文件里只有这么一个重要函数，那么我们现在去看main函数：

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
func main() {
    if len(os.Args) < 3 {
       fmt.Fprintf(os.Stderr, "Usage: mrsequential xxx.so inputfiles...\n")
       os.Exit(1)
    }

    mapf, reducef := loadPlugin(os.Args[1])

    //
    // read each input file,
    // pass it to Map,
    // accumulate the intermediate Map output.
    //
    intermediate := []mr.KeyValue{}
    for _, filename := range os.Args[2:] {
       file, err := os.Open(filename)
       if err != nil {
          log.Fatalf("cannot open %v", filename)
       }
       content, err := ioutil.ReadAll(file)
       if err != nil {
          log.Fatalf("cannot read %v", filename)
       }
       file.Close()
       kva := mapf(filename, string(content))
       intermediate = append(intermediate, kva...)
    }

    //
    // a big difference from real MapReduce is that all the
    // intermediate data is in one place, intermediate[],
    // rather than being partitioned into NxM buckets.
    //

    sort.Sort(ByKey(intermediate))

    oname := "mr-out-0"
    ofile, _ := os.Create(oname)

    //
    // call Reduce on each distinct key in intermediate[],
    // and print the result to mr-out-0.
    //
    i := 0
    for i < len(intermediate) {
       j := i + 1
       for j < len(intermediate) && intermediate[j].Key == intermediate[i].Key {
          j++
       }
       values := []string{}
       for k := i; k < j; k++ {
          values = append(values, intermediate[k].Value)
       }
       output := reducef(intermediate[i].Key, values)

       // this is the correct format for each line of Reduce output.
       fmt.Fprintf(ofile, "%v %v\n", intermediate[i].Key, output)

       i = j
    }

    ofile.Close()
}

这个就比较好理解了，首先要校验参数的数量，因为这个程序需要大于等于两个参数查新，然后就是利用之前的loadPlugin函数来加载后面会提到的mc.so文件得到两个函数mapf和reducef 然后是map阶段，我们会创建一组kv叫intermediate，然后用一个for循环把剩下的参数也就是文件用我们之前得到的mapf函数解析成kv对然后丢进kv池里面

然后用一个sort函数根据key的值进行排序

最后就是reduce阶段，用前面的reducef函数和双指针分组调用的方法得到结果

后续要做的呢就是把这两个for循环里的东西都分配给woker然后交给coordinator来管理

main/mrcoordinator.go

这个文件就是到时候写完我们自己的MapReduce和RPC系统以后测试用的脚本，看看他是那些部分组成的

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
func main() {
	if len(os.Args) < 2 {
		fmt.Fprintf(os.Stderr, "Usage: mrcoordinator inputfiles...\n")
		os.Exit(1)
	}

	m := mr.MakeCoordinator(os.Args[1:], 10)
	for m.Done() == false {
		time.Sleep(time.Second)
	}

	time.Sleep(time.Second)
}

只有一个简单的main函数，开头依然是参数校验，后面的m := mr.MakeCoordinator(os.Args[1:], 10)就是调用后面我们会写的coordinator代码来注册我们的coordinator，然后就是轮询代码，最后等待一秒钟退出

main/mrworker.go

这个则更简单，一笔带过了

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
func main() {
    if len(os.Args) != 2 {
       fmt.Fprintf(os.Stderr, "Usage: mrworker xxx.so\n")
       os.Exit(1)
    }

    mapf, reducef := loadPlugin(os.Args[1])

    mr.Worker(mapf, reducef)
}

// load the application Map and Reduce functions
// from a plugin file, e.g. ../mrapps/wc.so
func loadPlugin(filename string) (func(string, string) []mr.KeyValue, func(string, []string) string) {
    p, err := plugin.Open(filename)
    if err != nil {
       log.Fatalf("cannot load plugin %v", filename)
    }
    xmapf, err := p.Lookup("Map")
    if err != nil {
       log.Fatalf("cannot find Map in %v", filename)
    }
    mapf := xmapf.(func(string, string) []mr.KeyValue)
    xreducef, err := p.Lookup("Reduce")
    if err != nil {
       log.Fatalf("cannot find Reduce in %v", filename)
    }
    reducef := xreducef.(func(string, []string) string)

    return mapf, reducef
}

开头一个参数校验，然后就是和main/mrsequential一样的注册mapf和reducef函数，结束

mr/coordinator.go、mr/worker.go、mr/rpc.go空壳

这三个文件就是空壳，需要我们来写入的，我们可以参考论文原文来进行设计

mr/rpc.go

照例先来看一下源码，首先就是两个结构体，里面各自包了一个int变量：

1
2
3
4
5
6
7
type ExampleArgs struct {
    X int
}

type ExampleReply struct {
    Y int
}

这个是官方的样例，不用管他，然后看后面的coordinatorSock函数：

1
2
3
4
5
func coordinatorSock() string {
    s := "/var/tmp/5840-mr-"
    s += strconv.Itoa(os.Getuid())
    return s
}

它是 server()（监听端）和 call()（拨号端）约定socket 路径的“接头暗号” 我们在这里只需要写入四个结构体，来方便worker和coordinator两个独立进程的沟通,具体的流程如下：

sequenceDiagram
    participant W as Worker 进程
    participant C as Coordinator 进程

    W->>W: ① 客户端打包 args<br/>(gob 序列化成字节流)
    W->>C: ② 字节流经 Unix socket 发出<br/>(附带上方法名 "Coordinator.AssignTask")
    C->>C: ③ 服务端按方法名查表<br/>找到 c 对象的 AssignTask 方法
    C->>C: ④ 反序列化字节流 → 填充 *TaskRequest<br/>分配一个零值 TaskReply
    C->>C: ⑤ 执行你的 AssignTask 函数体<br/>(读任务表, 填 reply, 返回 nil)
    C->>W: ⑥ 把 reply 序列化发回
    W->>W: ⑦ 反序列化 → 填进你的 &reply<br/>call() 返回 true

所以我们考虑两组结构体，第一组是worker向coordinator要任务：

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
type TaskRequest struct {
	// 什么都不用带，代表着像coordinator要任务
	// 以后如果做优化(比如 worker 声明"我刚做完map任务X,它的中间文件我本地还有"),
	// 这里才有东西可加
}

type TaskReply struct {
	Type      string // "map" / "reduce" / "wait" / "exit" 四种指令
	TaskID    int    // 任务编号: map任务的X(对应mr-X-*), reduce任务的Y
	InputFile string // 仅map任务用: 要处理的文件名
	NMap      int    // reduce任务需要: 知道要去读 mr-0-Y ~ mr-(NMap-1)-Y
	NReduce   int    // map任务需要: 分桶数, ihash(key)%NReduce
}

第二组是worker向coordinator汇报任务完成/失败：

1
2
3
4
5
6
type ReportArgs struct {
	Type   string // 哪类任务完成了
	TaskID int    // 哪一个
}

type ReportReply struct{} // 惯例留空: 汇报不需要回执内容, 但Reply结构体必须存在

coordinator.go

先写服务端，方便理解，首先依然是需要想清楚作为一个服务端他需要哪些信息：

1
2
3
4
5
6
7
8
9
type Coordinator struct {
	mu        sync.Mutex
	files     []string    // M 个 map 任务的任务表
	nReduce   int
	mapDone   []bool      // 各 map 任务状态(false=可派发)
	nextMap   int         // 下一个待派的 map 任务下标
	reduceDone    []bool  // 各 reduce 任务状态
	nextReduce    int     // 待派的 reduce 任务游标
}

接下来可以看看项目自带的一个server函数

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
func (c *Coordinator) server() {
    rpc.Register(c)
    rpc.HandleHTTP()
    //l, e := net.Listen("tcp", ":1234")
    sockname := coordinatorSock()
    os.Remove(sockname)
    l, e := net.Listen("unix", sockname)
    if e != nil {
       log.Fatal("listen error:", e)
    }
    go http.Serve(l, nil)
}

第一行就是一个注册函数，把Coordinator类里面的所有方法都登记到一张方法表里面，写的所有方法都可以被远程调用

第二行就是注册http处理器，net/tpc的报文是跑在http协议上的，也就是说每个RPC请求都是一个HTTP POST

第三行被注释掉了，他注册了一个跨机器的tcp链接，如果想要模拟多机器的话要用

第四行这是我们之前看到coordinatorSock函数拿到本机的socket路径

第五行调用了一个库函数os.Remove是用来删除可能残留的旧socket文件

第六行则是创建ocket端点并开始监听，从这里开始所有的worker都可以进入了

最后就是做一个后台的服务循环，阻塞式的接受链接，每来一个链接就处理上面的RPC请求，也就是第一行中登记的方法，注意这个东西永不返回，出发socket报错了

接下来就是考虑创建Coordinator服务了，写一个简单的构造函数：

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
func MakeCoordinator(files []string, nReduce int) *Coordinator {
	c := Coordinator{
		files:      files,
		nReduce:    nReduce,
		mapDone:    make([]bool, len(files)),
		reduceDone: make([]bool, nReduce),
	}
	c.server()
	return &c
}

这里有个编译器抓不到的坑：如果把结构体初始化写成空的c := Coordinator{}，len(c.files)永远是0，所有worker只能收到wait；更糟的是第一个ReportDone进来时c.mapDone[0] = true会直接panic——对nil slice赋值。所以四个字段必须在这里全部初始化好，尤其两个bool切片要用make按任务数开好长度。

接下来就是分配任务的函数：

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
func (c *Coordinator) AssignTask(args *TaskRequest, reply *TaskReply) error {
	c.mu.Lock()
	defer c.mu.Unlock()

	if c.nextMap < len(c.files) { // 还有 map 任务没派完
		reply.Type = "map"
		reply.TaskID = c.nextMap
		reply.InputFile = c.files[c.nextMap]
		reply.NReduce = c.nReduce
		c.nextMap++
		return nil
	}

	// 阶段2: map 全部完成才开始派 reduce(桶文件此时才齐)
	if allTrue(c.mapDone) {
		if c.nextReduce < c.nReduce {
			reply.Type = "reduce"
			reply.TaskID = c.nextReduce
			reply.NMap = len(c.files) // reduce 收货时需要知道有几个来源
			c.nextReduce++
			return nil
		}
		if allTrue(c.reduceDone) { // 阶段3: 全部完成, 通知 worker 收工
			reply.Type = "exit"
			return nil
		}
	}

	reply.Type = "wait" // 本阶段已派完但没全跑完, worker 先等等
	return nil
}

这里是先加了一个锁，然后defer了锁，让他完成任务后自动解锁

后面就是判断活是否派完了，这里的reply指的是RPC的返回载体，具体流程如下：

flowchart LR
    A["worker侧: 声明零值的reply"] --> B["worker发起call, 请求飞出"]
    B --> C["coordinator框架收到请求<br/>分配一个零值TaskReply"]
    C --> D["你的handler往reply里填字段<br/>然后return nil"]
    D --> E["框架把reply序列化发回worker"]
    E --> F["worker侧的reply被填充<br/>call返回true"]

正如之前所说，net/rpc规定的handler无法返回业务数据，唯一的办法就是reply这个指针参数，吧task相关新的信息给到worker，其实也就是相当于return

如果派完了任务，但是仍然跑了这个函数，那么就会标记为wait

这里要注意两个细节：第一，判断能不能派reduce用的是allTrue(c.mapDone)而不是nextMap >= len(files)——因为派完不等于跑完，桶文件要等所有map真正写完才齐全，reduce提前开跑会收到os.Open失败。第二，派reduce时必须填reply.NMap = len(c.files)，因为doReduce要靠它决定去收几家的货，不填就一个桶文件都收不到。

另外如果只有map和wait两个分支，map全跑完之后没有人切换阶段，所有worker会无限wait下去，所以阶段2和阶段3是整个状态机真正转起来的关键：map全完成→派reduce→reduce全完成→发exit，Done()才有机会变成true让mrcoordinator退出。

接着就是一个回复的函数：

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
func (c *Coordinator) ReportDone(args *ReportArgs, reply *ReportReply) error {
	c.mu.Lock()
	defer c.mu.Unlock()
	if args.Type == "map" {
		c.mapDone[args.TaskID] = true
	} else if args.Type == "reduce" {
		c.reduceDone[args.TaskID] = true
	}
	return nil
}

这个函数相当于worker完成任务以后寄回给coordinator的完工信

第一行同样的是一个加锁然后第二行defer这是因为两个worker可能同时完工，而这个后面的mapDone是共享账本，并发写入就会有问题

注意else if那个reduce分支不能少：如果只处理map，reduce的完工汇报会被静默丢弃，reduceDone永远全是false，Done()永远不成立，mrcoordinator就永远不会退出

最后是一个表示完成的Done函数

1
2
3
4
5
6
7
func (c *Coordinator) Done() bool {
	c.mu.Lock()
	defer c.mu.Unlock()
	return c.nextMap >= len(c.files) &&      // map 全派完
		allTrue(c.mapDone) &&                // map 全完成
		allTrue(c.reduceDone)                // reduce 全完成 
}

这个是到时候运行main/mrcoordinator.go的时候那个轮询，来知道是否完成所有任务

Done()里用到的allTrue是一个小工具函数，见到一个false就整体false，全true才返回true：

1
2
3
4
5
6
7
8
func allTrue(flags []bool) bool {
	for _, b := range flags {
		if !b {
			return false
		}
	}
	return true
}

worker.go

我们先来看看worer里面带了哪些预设的可用的东西

首先就是一个kv结构体和一个哈希函数，就不多赘述了：

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
type KeyValue struct {
    Key   string
    Value string
}

//
// use ihash(key) % NReduce to choose the reduce
// task number for each KeyValue emitted by Map.
//
func ihash(key string) int {
    h := fnv.New32a()
    h.Write([]byte(key))
    return int(h.Sum32() & 0x7fffffff)
}

下面就是一个call函数：

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
func call(rpcname string, args interface{}, reply interface{}) bool {
    // c, err := rpc.DialHTTP("tcp", "127.0.0.1"+":1234")
    sockname := coordinatorSock()
    c, err := rpc.DialHTTP("unix", sockname)
    if err != nil {
       log.Fatal("dialing:", err)
    }
    defer c.Close()

    err = c.Call(rpcname, args, reply)
    if err == nil {
       return true
    }

    fmt.Println(err)
    return false
}

首先定义一个socket,然后我们用Dial.HTTP来做三件事情：打开 socket 文件 → 和 coordinator 完成 HTTP 握手 → 协商出 RPC 通道。最后返回一条连接*rpc.Client

接着defer一个关闭这连接的语句重点就是后一句，调用连接中的Call函数，他是同步阻塞的，主要作用就是你传的 args 序列化后经 socket 发给 coordinator，等它执行完对应方法、把填好的 reply 序列化传回来填进你的 reply 指针。

接着就是一个call函数的样例，我们后续就根据这个样例来写我们的函数：

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
func CallExample() {

    // declare an argument structure.
    args := ExampleArgs{}

    // fill in the argument(s).
    args.X = 99

    // declare a reply structure.
    reply := ExampleReply{}

    // send the RPC request, wait for the reply.
    // the "Coordinator.Example" tells the
    // receiving server that we'd like to call
    // the Example() method of struct Coordinator.
    ok := call("Coordinator.Example", &args, &reply)
    if ok {
       // reply.Y should be 100.
       fmt.Printf("reply.Y %v\n", reply.Y)
    } else {
       fmt.Printf("call failed!\n")
    }
}

我们需要的是一个一个客户端循环，所以我们可以考虑这么设计：

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
func Worker(mapf func(string, string) []KeyValue,
	reducef func(string, []string) string) {

	for {
		args := TaskRequest{}
		reply := TaskReply{}
		ok := call("Coordinator.AssignTask", &args, &reply)
		if !ok {
			return // coordinator 已退出
		}

		switch reply.Type {
		case "map":
			if err := doMap(mapf, reply); err != nil {      
				log.Printf("map task %d failed: %v", reply.TaskID, err)
				continue                                    // 失败不汇报, 靠超时重派兜底
			}
			call("Coordinator.ReportDone",
				&ReportArgs{Type: "map", TaskID: reply.TaskID}, &ReportReply{})
		case "reduce":                                      
			if err := doReduce(reducef, reply); err != nil {
				log.Printf("reduce task %d failed: %v", reply.TaskID, err)
				continue
			}
			call("Coordinator.ReportDone",
				&ReportArgs{Type: "reduce", TaskID: reply.TaskID}, &ReportReply{})
		case "wait":
			time.Sleep(500 * time.Millisecond)
		case "exit":
			return
		}
	}
}

他是无限循环的，直到最后到了触发条件的时候才退出首先会创建一个空的请求和空的回复，args和reply

然后使用一个call方法，找到coordinator中的AssignTask方法，同时载入args和reply

如果这call返回false那么说明coordinator已经退出，这时候就直接return 退出循环

这时候我们会用switch case语句来判断worker类型，如果是map或者reduce那么就干活，然后同样使用call方法汇报完工情况，如果是wait那就等，如果是exit那就返回

下面写一下具体干活的doMap和doReduce是具体怎么完成这个工作的：首先是doMap

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
func doMap(mapf func(string, string) []KeyValue, reply TaskReply) error {
	content, err := os.ReadFile(reply.InputFile)   // 读文件(原来在main循环里)
	if err != nil {
		return err
	}
	kva := mapf(reply.InputFile, string(content))  // 调插件把文件解析为kv对

	// 分桶写文件: 先全部写临时文件
	encs := make([]*json.Encoder, reply.NReduce)
	tmps := make([]*os.File, reply.NReduce)
	for i := 0; i < reply.NReduce; i++ {
		f, err := os.CreateTemp(".", "tmp-*")      // 临时文件
		if err != nil {
			return err
		}
		tmps[i] = f
		encs[i] = json.NewEncoder(f)
	}
	for _, kv := range kva { //取哈希值
		b := ihash(kv.Key) % reply.NReduce         
		encs[b].Encode(&kv)
	}

	// 原子提交: 全部写完才逐个改名(半成品永远不可见)
	for i, f := range tmps {
		f.Close()
		os.Rename(f.Name(), fmt.Sprintf("mr-%d-%d", reply.TaskID, i))
	}
	return nil
}

这一这里的分桶，其实他就是在磁盘上的文件，然后以mr-x-y的形式存储，x就是哪个map任务产生的，y就是桶号，里面就是一堆json格式的kv对

然后这个函数先for循环创建了n个文件然后存储在tmps内，然后通过encs吧编码器绑定在文件上

然后在kv主循环上求出哈希值，再再编码器上把kv对解析成json格式存储

然后就是落盘，形成n个桶文件

然后就是doReduce：

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
func doReduce(reducef func(string, []string) string, reply TaskReply) error {
	// 1. 收集: 把 NMap 个 map 任务产出的第 TaskID 桶全部读进来
	kva := []KeyValue{}
	for x := 0; x < reply.NMap; x++ {
		f, err := os.Open(fmt.Sprintf("mr-%d-%d", x, reply.TaskID))
		if err != nil {
			return err
		}
		dec := json.NewDecoder(f)
		var kv KeyValue
		for dec.More() {
			dec.Decode(&kv)
			kva = append(kva, kv)
		}
		f.Close()
	}

	// 2. 排序 —— 从这里开始和 mrsequential 逐字相同
	sort.Sort(ByKey(kva))

	// 3. 双指针分组 + reducef + 标准格式输出
	tmp, _ := os.CreateTemp(".", "tmp-*")
	i := 0
	for i < len(kva) {
		j := i + 1
		for j < len(kva) && kva[j].Key == kva[i].Key {
			j++
		}
		values := []string{}
		for k := i; k < j; k++ {
			values = append(values, kva[k].Value)
		}
		fmt.Fprintf(tmp, "%v %v\n", kva[i].Key, reducef(kva[i].Key, values))
		i = j
	}
	tmp.Close()
	os.Rename(tmp.Name(), fmt.Sprintf("mr-out-%d", reply.TaskID))  // 原子提交
	return nil
}

首先就是一个for循环来收集我们之前doMap产生的文件，然后就是根据key来排序然后就是经典的双指针分组，然后append到valud里面，最后就是落盘写到mr-our-x文件里，x指的是reduce的任务编号

还有一个容易踩的坑：doReduce里用了sort.Sort(ByKey(kva))，但ByKey定义在mrsequential.go里，那是package main的代码，mr包看不见它——同一个仓库不同包，各自独立编译，所以要自己在worker.go里补一份：

1
2
3
4
5
6
// for sorting by key.
type ByKey []KeyValue

func (a ByKey) Len() int           { return len(a) }
func (a ByKey) Swap(i, j int)      { a[i], a[j] = a[j], a[i] }
func (a ByKey) Less(i, j int) bool { return a[i].Key < a[j].Key }

这是sort.Sort的接口约定：要排序的类型必须实现sort.Interface，也就是Len、Swap、Less这三个方法

mrapps/wc.go

参考文献

https://pdos.csail.mit.edu/6.824/papers/mapreduce.pdf

官方源码阅读#

main/mrsequential.go#

main/mrcoordinator.go#

main/mrworker.go#

mr/coordinator.go、mr/worker.go、mr/rpc.go空壳#

mr/rpc.go#

coordinator.go#

worker.go#

mrapps/wc.go#

参考文献#