$ git clone git://g.csail.mit.edu/6.824-golabs-2020 6.824
$ cd 6.824
$ ls
Makefile src

$ cd ~/6.824
$ cd src/main
$ go build -buildmode=plugin ../mrapps/wc.go
$ rm mr-out*
$ go run mrsequential.go wc.so pg*.txt
$ more mr-out-0
A 509
ABOUT 2
ACT 8
...

$ go build -buildmode=plugin ../mrapps/wc.go

$ rm mr-out*
$ go run mrmaster.go pg-*.txt

$ go run mrworker.go wc.so

$ sh ./test-mr.sh
*** Starting wc test.
--- wc test: PASS
*** Starting indexer test.
--- indexer test: PASS
*** Starting map parallelism test.
--- map parallelism test: PASS
*** Starting reduce parallelism test.
--- reduce parallelism test: PASS
*** Starting crash test.
--- crash test: PASS
*** PASSED ALL TESTS
$

// 接口输入
type AskForTaskArgs struct {
}
// 接口输出
type AskForTaskReply struct {
	CurPhase      Phase
	MapFile       string
	JobDone       bool
	MapNum        int
	ReduceNum     int
	MapTaskIdx    int
	ReduceTaskIdx int
}

// worker的rpc调用
func AskForTask() AskForTaskReply {
	args := AskForTaskArgs{}

	reply := AskForTaskReply{}

	call("Master.AskForTask", &args, &reply) // rpc调用

	return reply
}

//
// main/mrworker.go calls this function.

func Worker(mapf func(string, string) []KeyValue,
	reducef func(string, []string) string) {
	for { // 使用轮询的方式不断申请执行任务
	   reply := AskForTask() // 申请任务
	   if reply.JobDone { // 所有任务执行完毕，退出
	   	break
	   }
	   switch reply.CurPhase {
	   	case MapPhase:
			// 执行Map任务
			...
		case ReducePhase:
			// 执行Reduce任务
			...
	   }
	   time.Sleep(time.Second) // 避免太多的忙轮询，这里选择sleep一秒，但是还是会浪费掉cpu以及不能及时的获取到任务进行执行，影响任务执行的总时间
	}
}

	   switch reply.CurPhase {
	   	case MapPhase:
			// 执行Map任务
			...
			fileName := reply.MapFile
			file, _ := os.Open(fileName) 
			content, _ := ioutil.ReadAll(file) // 读取map的输入
			kva := mapf(fileName, string(content)) // 调用应用的map函数生成中间数据
			mapPhaseSucc := true
			for i := 0; i < len(kva); i++ {
				reduceTaskIdx := ihash(kva[i].Key) % askForReply.ReduceNum // 计算key由后续哪个Reduce任务处理
				intermediateEncoder := intermediateEncoderMap[reduceTaskIdx]
				if err := intermediateEncoder.Encode(kva[i]); err != nil { // 序列化到临时文件
					log.Fatalf("encode kv:%v failed", kva[i])
					mapPhaseSucc = false
					NotifyWorkerTaskStatus(MapPhase, askForReply.MapTaskIdx, Fail) // 调用rpc
					，通知master执行过程中存在问题，任务执行失败
					break
				}
			}
			if mapPhaseSucc {
				for reduceIdx, fileName := range intermediateFileNameMap {
					os.Rename(fileName, fmt.Sprintf("mr-%+v-%+v", askForReply.MapTaskIdx, reduceIdx)) // 通过os的原子重命名，保证文件内容的完整性
				}
				NotifyWorkerTaskStatus(MapPhase, askForReply.MapTaskIdx, Success) // 通知master，任务执行成功
			}
}			
			

case ReducePhase:
			intermediate := []KeyValue{} 
			for i := 0; i < reply.MapNum; i++ {
				fileName := fmt.Sprintf("mr-%+v-%+v", i, askForReply.ReduceTaskIdx) // 分别从所有的Map的输出读取当前Reduce任务需要处理的数据
				file, _ := os.Open(fileName)
				dec := json.NewDecoder(file)
				for {
					var kv KeyValue
					if err := dec.Decode(&kv); err != nil {
						break
					}
					intermediate = append(intermediate, kv)
				}
			}
			sort.Sort(ByKey(intermediate)) // 对整合后的数据按key排序

			oname := fmt.Sprintf("mr-out-%+v", reply.ReduceTaskIdx) // Reduce任务的最终输出
			uuidInstance := uuid.New()
			uuidFileName := fmt.Sprintf("mr-tmp-reduce-%v", uuidInstance.String()) // 临时文件
			ofile, _ := os.OpenFile(uuidFileName, os.O_APPEND|os.O_CREATE|os.O_WRONLY, 0644)

			i := 0
			for i < len(intermediate) {
				j := i + 1
				for j < len(intermediate) && intermediate[j].Key == intermediate[i].Key {
					j++
				}
				values := []string{}
				// 对每一个key，合并对应的value
				for k := i; k < j; k++ {
					values = append(values, intermediate[k].Value)
				}
				output := reducef(intermediate[i].Key, values) // 调用应用reduce函数，生成

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

				i = j
			}
			os.Rename(uuidFileName, oname) // 通过os的原子重命名，保证文件内容的完整性
			NotifyWorkerTaskStatus(ReducePhase, askForReply.ReduceTaskIdx, Success) // 通知master，reduce任务执行成功

const (
	NotYetStarted TaskStatus = iota
	Doing
	Done
)

type Master struct {
	// Your definitions here.
	Files               []string // 要处理的文件
	MapNum              int // Map任务总数
	ReduceNum           int // Reduce任务总数
	MapTaskStatusMap    []TaskStatus // Map任务执行状态,初始值都是未开始NotYetStarted
	ReduceTaskStatusMap []TaskStatus // Reduce任务执行状态，初始值也都是未开始NotYetStarted
	CurPhase            Phase // 当前执行阶段，MapPhase还是ReducePhase
	Finished            bool // 整个job是否完成，以便让master退出，还有让worker退出
	DoneMapNum          int  // 记录当前已完成的Map任务数量，决定是否要进入Reduce阶段
	DoneReduceNum       int  // 记录当前已完成的Reduce任务数量，决定是否结束Reduce阶段，并标记整个job完成
}

// create a Master.
// main/mrmaster.go calls this function.
// nReduce is the number of reduce tasks to use.
//
func MakeMaster(files []string, nReduce int) *Master {
	m := Master{}

	// Your code here.

	m.Files = files
	m.MapNum = len(files)
	m.ReduceNum = nReduce
	m.DoneMapNum = 0
	m.DoneReduceNum = 0
	m.Finished = false
	m.MapTaskStatusMap = make([]TaskStatus, len(files))
	for i := 0; i < len(m.Files); i++ {
		m.MapTaskStatusMap[i] = NotYetStarted
	}
	m.ReduceTaskStatusMap = make([]TaskStatus, nReduce)
	for i := 0; i < nReduce; i++ {
		m.ReduceTaskStatusMap[i] = NotYetStarted
	}

	m.server()
	return &m
}

func (m *Master) AskForTask(args *AskForTaskArgs, reply *AskForTaskReply) error {
	mutex.Lock() // 加锁，粒度比较粗
	if m.Finished { // job已经处理完毕，通知worker退出
		mutex.Unlock()
		reply.JobDone = true
		return nil
	}
	reply.JobDone = false
	if m.CurPhase == MapPhase {
		// 遍历Map任务状态，找到一个尚未开始的任务分发给worker
		for i := 0; i < len(m.MapTaskStatusMap); i++ {
			if m.MapTaskStatusMap[i] == NotYetStarted {
				reply.MapFile = m.Files[i]
				reply.CurPhase = MapPhase
				reply.MapNum = m.MapNum
				reply.ReduceNum = m.ReduceNum
				reply.MapTaskIdx = i
				m.MapTaskStatusMap[i] = Doing
				go func(mapTaskIdx int) { // 这里需要为每个Map任务进行倒计时，一旦任务执行超过10秒，便标记未尚未开始，等待下次重新执行
					timer := time.NewTimer(time.Second * 10)
					<-timer.C
					mutex.Lock()
					if m.MapTaskStatusMap[mapTaskIdx] == Doing {
						m.MapTaskStatusMap[mapTaskIdx] = NotYetStarted
					}
					mutex.Unlock()
				}(i)
				break
			}
		}
	} else if m.CurPhase == ReducePhase {
		// 与Map处理方式类似
		...
	}
	mutex.Unlock() // 释放锁
	return nil
}

func (m *Master) NotifyWorkerTaskStatus(args *NotifyWorkerTaskStatusArgs, reply *NotifyWorkerTaskStatusReply) error {
	mutex.Lock()
	if args.WorkerPhase == MapPhase {
		if args.Status == Success {
			m.MapTaskStatusMap[args.TaskIdx] = Done  // 标记任务状态为已完成
			m.DoneMapNum++ // 已完成任务数量加一
			if m.DoneMapNum == m.MapNum { // 所有任务已完成，则开始进入reduce阶段
				m.CurPhase = ReducePhase
			}
		} else if args.Status == Fail {
			m.MapTaskStatusMap[args.TaskIdx] = NotYetStarted // 失败情况下，重新标记为尚未开始，等待下一次调度
		} else {
			mutex.Unlock()
			return fmt.Errorf("unknown task status:%+v phase:%+v taskIdx:%+v", args.Status, args.WorkerPhase, args.TaskIdx)
		}
	} else if args.WorkerPhase == ReducePhase {
		// 与Map处理方式类似
		...
		if m.DoneReduceNum == m.ReduceNum { // 所有reduce任务已完成，标记整个job为完成状态，等待退出
			m.Finished = true
		}
		...
	}
	mutex.Unlock()
	return nil
}

//
// main/mrmaster.go calls Done() periodically to find out
// if the entire job has finished.
//
func (m *Master) Done() bool {
	ret := false

	// Your code here.

	mutex.Lock() // 保证并发安全
	ret = m.Finished
	mutex.Unlock()

	return ret
}

$ sh test-mr.sh
os.Args[2:] [../pg-being_ernest.txt ../pg-dorian_gray.txt ../pg-frankenstein.txt ../pg-grimm.txt ../pg-huckleberry_finn.txt ../pg-metamorphosis.txt ../pg-sherlock_holmes.txt ../pg-tom_sawyer.txt]
filename:../pg-being_ernest.txt
filename:../pg-dorian_gray.txt
filename:../pg-frankenstein.txt
filename:../pg-grimm.txt
filename:../pg-huckleberry_finn.txt
filename:../pg-metamorphosis.txt
filename:../pg-sherlock_holmes.txt
filename:../pg-tom_sawyer.txt
*** Starting wc test.
2020/09/14 15:55:01 rpc.Register: method "Done" has 1 input parameters; needs exactly three
--- wc test: PASS
os.Args[2:] [../pg-being_ernest.txt ../pg-dorian_gray.txt ../pg-frankenstein.txt ../pg-grimm.txt ../pg-huckleberry_finn.txt ../pg-metamorphosis.txt ../pg-sherlock_holmes.txt ../pg-tom_sawyer.txt]
filename:../pg-being_ernest.txt
filename:../pg-dorian_gray.txt
filename:../pg-frankenstein.txt
filename:../pg-grimm.txt
filename:../pg-huckleberry_finn.txt
filename:../pg-metamorphosis.txt
filename:../pg-sherlock_holmes.txt
filename:../pg-tom_sawyer.txt
*** Starting indexer test.
2020/09/14 15:55:14 rpc.Register: method "Done" has 1 input parameters; needs exactly three
--- indexer test: PASS
*** Starting map parallelism test.
2020/09/14 15:55:25 rpc.Register: method "Done" has 1 input parameters; needs exactly three
--- map parallelism test: PASS
*** Starting reduce parallelism test.
2020/09/14 15:55:40 rpc.Register: method "Done" has 1 input parameters; needs exactly three
--- reduce parallelism test: PASS
os.Args[2:] [../pg-being_ernest.txt ../pg-dorian_gray.txt ../pg-frankenstein.txt ../pg-grimm.txt ../pg-huckleberry_finn.txt ../pg-metamorphosis.txt ../pg-sherlock_holmes.txt ../pg-tom_sawyer.txt]
filename:../pg-being_ernest.txt
filename:../pg-dorian_gray.txt
filename:../pg-frankenstein.txt
filename:../pg-grimm.txt
filename:../pg-huckleberry_finn.txt
filename:../pg-metamorphosis.txt
filename:../pg-sherlock_holmes.txt
filename:../pg-tom_sawyer.txt
*** Starting crash test.
2020/09/14 15:55:57 rpc.Register: method "Done" has 1 input parameters; needs exactly three
--- crash test: PASS
*** PASSED ALL TESTS