QDQihang Dai
← Writing

MIT 6.824 Lab 1: MapReduce 复盘

·7 min read

总体花了一周，比较摸鱼。不管是视频还是论文看完总是有点模糊，最费时间和理解最深的还是写代码。

mapreduce的结构很简单，就实现三个模块： 分配任务的coordinator， 申请任务和处理任务的worker， 以及他俩之间通信用的rpc的消息结构。大体上是参考这个repo半抄理解。一开始花了很多时间看了好几个repo， 发现用RPC沟通有很多种写法，有的用channel有的只有锁， 有的只定义一个args reply有的定义了 requestArgs 和finishArgs， 有的把rpc传进函数参数进行修改而有的直接传rpc的property然后返回一个新的struct， 迷惑了一段时间，最后发现其实都是一样的，只是写法不同。

Structure

Map reduce: Given a set of files, run map for all of them to map the keys into the corresponding values like word count (generate for each occurence of word for (word, 1)). Then for each key in the kv list, hash the key into a number within the reduce jobs amount, and append that value into the array[key hashed]. Thus we would have a array where keys with same hash are append into a same sublist. Stores this hashed array into intermediate files. After all map is done, run reduce for all intermediate files to generate the final result, like sum the count for the same key. The reduce would sort the subarray first so he can get same key stay together, then do simple loop and count sum.

We spin up multiple workers to do the map or reduce job (in different threads or different machiness to boost performance). But how workers know when to map and when to reduce (reduce can only be started once all map jobs have been done), we need a centralized coordinator for the workers to communicate.

Worker ask for a job for the coordinator. coordinator would check its own stauts to send the worker a job. Map jobs are equal to the input file number, reduce jobs are equal to the number of reduce tasks. Thus coordinator always give worker map job before all map job is marked done. Once all map is done coordinator begin to assign reduce jobs. We can have several workers, but usually much less than the map job or file amount. Thus work would seek a job, and once the job is done and there is still jobs left, the same worker apply for another job in the queue. Thus here are the two job types: MAP and REDUCE.

some map jobs may be finished early like the file is small like one line and other file is still working since the file is big like one billion line. Lets say there is 2 worker and 2 map job in the mentioned situation. Worker 1 finished map early, and worker 2 is still running for a long time. Worker 1 can not start apply for reduce jobs since not all map jobs are marked one yet. Now worker 1 need to wait or sleep use time.Sleep(time.Second) to wait for the other worker to finish. This is the third job type: WAIT.

once all map and reduce jobs are finished, workers can be shut down. This is the fourth job type: EXIT.

Thus there are acutally 4 components: Coordinator + job assign response + Worker + job apply request. The response need to contains the job type, job id, and file name of which the file is going to be processed by the worker. The request usually doesnt need a type, but it do need a job id and file name list when the worker returns the map result to the coordinator so the coordinator can use this information to generate the reduce job.

RPC


//request task from coordinator. also indicate the task status. since args communicate coordinator and worker in both directions, 
//Files is used to send the processed file content to the coordinator but may looks useless when worker request task from coordinator
type TaskArgs struct {
	Type TaskType
	Id  int
	Files []string // when worker report task status, it will send the processed file names to the coordinator
}

type TaskReply struct {
	Type TaskType
	Id  int
	NReduce int
	Files []string //assign single file name to map task in Files[0], assign intermediate file names to reduce tasks 
}

一开始我是从RPC开始写，毕竟没有定义好传递的Args也不知道怎么写worker和coordinator的内容。属于纯抄，很难理解为什么要定义某些参数， 比如args里为啥要传FILE。和File是什么内容.

交流只会从worker主动发起，worker不断询问coordinator来获取任务。 所以一套ARGS REPLY就够了。 但是当worker完成任务后，需要把结果返回给coordinator，这时候就需要另外一套ARGS REPLY。可以新声明另一套struct， 也可以用同一个struct。 用一个struct时可以重复用同一个requestArgs， 也可以新建一个request args 但是type要和response给的type一样从而显示完成了什么类型的任务。

这里把reqeust和response都用一个task args. 其实还是在建struct时候简略（避免让人疑惑为什么相同的东西要用两个不同的名字）， 但在建object的时候区分一下会比较好（如果反复修改通一个object使得同一个object有不同的用途， 也很迷惑）。

Coordinator


type Task struct{
	Type TaskType
	Id int
	Files []string
	Start_time time.Time //default value is long time ago 
	Done bool
}

type Coordinator struct {
	// Your definitions here.
	mu sync.Mutex
	reduceleft int // the number of reduce tasks is defined by the caller
	mapleft int // the number of map tasks is the number of files
	mapTasks []Task
	reduceTasks []Task
}

coordinator中需要有一个新的struct TASK 和 task list来记录任务的状态，包括任务类型，任务id，任务文件名，任务开始时间，任务是否完成。 任务开始时间用来判断是否超时， 任务是否完成用来判断是否可以分配reduce任务。 任务超时的话就让新的worker来做这个任务。 这就是fault tolerance或者backup.

然后是分配任务的逻辑： 首先上锁，switch部分是检查从worker处返回的已完成的任务状态， 更新任务队列和剩余任务数。 switch之后的部分是分配新任务的逻辑。 写在一起而没有抽象成函数显得比较难看，逻辑也不清晰。一堆东西挤在一起，也很容易因为括号和缩进问题而出错，下次写的时候要注意。 一开始把switch后面的部分也成了switch init，导致检查任务结束和分配新任务居然只能有一个进行，分配了一个任务之后就无法分配别的任务了， 又是debug花了半小时。

分配新任务时的逻辑是：有map就分配map，map做完再分配reduce。 任何一种任务如果undone and 超时（任务开始时间的默认值都是很久之前，所以默认超时）则分配此任务给new worker。


func (c *Coordinator) TalktoWorker(request *TaskArgs, response *TaskReply) error{
	c.mu.Lock()
	defer c.mu.Unlock()
	// log.Printf("request: Type: %v , ID: %v", request.Type, request.Id)

	//when the worker finished the task, it will send a request with Map or Reduce statusto the coordinator.
	//we mark the task in the task queue as done if the task is not done yet (in case of duplicate request)

	//check if the request is a done request
	switch request.Type {
		case MAP:
			if !c.mapTasks[request.Id].Done {
				c.mapTasks[request.Id].Done = true
				log.Printf("map task %v done", request.Id)

				//when the worker finished the map task, it will send the processed intermediate file name to the coordinator
				//we will assign the intermediate file name to the reduce tas
				for id, file := range request.Files {
					if len(file) > 0 {
						c.reduceTasks[id].Files = append(c.reduceTasks[id].Files, file)
					}
				}

				c.mapleft--
			}
		case REDUCE:
			if !c.reduceTasks[request.Id].Done {
				log.Printf("reduce task %v done", request.Id)
				c.reduceTasks[request.Id].Done = true
				c.reduceleft--
			}
	}
	
	//assign task to the worker everytime it request a task

	//when the worker request a task, it will send a INIT request to the coordinator
	//we will assign a map task or reduce task to the worker
	now := time.Now()
	timeoutAgo := now.Add(-10 * time.Second)
	if c.mapleft> 0 {
		//check if there is (unfinished map task) and (has been work for more than 10 seconds) 
		//new task will also be assinged since new task its always time out since the start time is long time ago
		// if so, assign the task to a new worker
		for idx := range c.mapTasks {
			t := &c.mapTasks[idx]
			if !t.Done && t.Start_time.Before(timeoutAgo) {
				response.Type = MAP
				response.Id = t.Id
				response.Files = t.Files
				response.NReduce = len(c.reduceTasks)
				t.Start_time = now 
				return nil
			}
		}
		//if all tasks have been started and not timed out || done, ask the worker to wait
		response.Type = WAIT
	} else if c.reduceleft> 0 { //if all map tasks are done, assign reduce task
		for idx := range c.reduceTasks {
			t := &c.reduceTasks[idx]
			if !t.Done && t.Start_time.Before(timeoutAgo) {
				response.Type = REDUCE
				response.Id = t.Id
				response.Files = t.Files
				t.Start_time = now 
				return nil
			}
		}
		response.Type = WAIT //if there is no unfinished reduce task, ask the worker to wait till all reduces are done
	}else{
		response.Type = DONE //if all map and reduce tasks are done, tell the worker to exit
		log.Printf("your job is done")
	}
	//print the assigned task
	log.Printf("response: Type: %v , ID: %v", response.Type, response.Id)
	return nil
}

Worker


func Worker(mapf func(string, string) []KeyValue, reducef func(string, []string) string) {

	// Your worker implementation here.

	// uncomment to send the Example RPC to the coordinator.
	// CallExample()
	
	var response TaskReply = TaskReply{}
	var request TaskArgs = TaskArgs{Type: INIT}

	for {
		response = TalktoMaster(&request) // get task from coordinator +  update req done or not
		switch response.Type {
			case MAP:
				doMap(mapf, &request, &response)
			case REDUCE:
				doReduce(reducef, &request, &response)
			case WAIT:
				log.Printf("wait for 500 ms")
				time.Sleep(500 * time.Millisecond)
				request.Type = INIT
			case DONE:
				return
			default:
				//panic and show the type of task, though it can only be INIT 
				panic(fmt.Sprintf("unknown task type: %v", response.Type))
		}
	}
}

worker的逻辑很简单，就是不断的向coordinator请求任务，接收到response的信息然后根据任务类型做相应的事情。 任务类型有map，reduce，wait，done。 map和reduce的逻辑都是调用mapf和reducef函数，wait的逻辑是等待500ms，done的逻辑是退出。 INIT是worker第一次请求任务时的默认值，不会出现在worker的逻辑中。一个小瑕疵是我把任务昨晚后对request的修改放在了doMAP和doReduce函数里，这样做导致逻辑很不清楚，而且让修改request这个和domap本身无关的逻辑挤在一大堆对文件处理的函数里，果然因为这个瑕疵，我不小心让request的修改进入到了doReduce中的上一个循环而不是在doReduce结束后再修改，导致worker等待的概率大大增加，最后导致了超时。 worker等待是因为任务都被分配完了，但是还有别的worker没有结束。 提前反复修改request可能是降低了worker的速度。

写了另一个bug是在map存文件时想先存到tempfile然后再集体rename。 然而我对于golang的defer file.close()不熟悉， defer是在function return时才启动，会把defer放进一个Stack，后defer的会先运行，我胡乱defer导致rename时temp并没有关闭从而rename失败，到reduce时就出现file cant found的错误。

最后总算是把test-mr都过了。 test-mr-many没有跑，实在de不动了。

➜  main git:(master) ✗ bash 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 job count test.
--- job count test: PASS
*** Starting early exit test.
--- early exit test: PASS
*** Starting crash test.
--- crash test: PASS
*** PASSED ALL TESTS

总结

在已有别人代码参考的情况下，lab1的整体逻辑真的不应该算很难，代码量也不多。 但是我还是花了很多时间在debug上， coordinator的一个bug和worker的两个bug一起大概吃掉了四五个个小时的时间。

教训是能分离的逻辑就分离，不要把不相关的逻辑放在一起， 很容易因为微小的如括号缩进fileclose defer等等的错误导致整个逻辑出错。虽然是ADHD通病，但是这种小错误往往最难发现，因为自己回头分析逻辑发现自己的逻辑是对的， 而实现的时候错了也找不到出错原因，最后只能满世界print log来定位。从去年的555和实习之后我的debug功力有了很大的提升，但是还是很浪费时间。

感叹写代码跟打游戏一样，少失误的收益比经常carry大， 少写bug的收益也比写的快大… 少打点伤害不要紧， 关键团送了就是致命。 copilot也得谨慎使用，补全的时候经常补漏{}, 不该进loop的进了loop，先四处log fmt.print 半小时找到灶点，再line by line读代码找错。 真的宁可自己多手写点也不要让copilot直接带着{}就补全了。