Lab1具体实现

要求简介

实现 mrsequential.go 中的单词计数的分布式版本（其实只是单机跑）

要求跟论文2.1的单词统计挺像的，就是论文里是master分配任务，我们的程序要求worker去向coordinator要任务

整体框架图

image-20220103013740271

再看一下生成上述文件名的四个函数:

// 生成临时 map file name
func genTmpMapName(wid, ti, r int) string {
	return fmt.Sprintf("tmp-worker-%d-%d-%d", wid, ti, r)
}

// 生成最终map 输出文件
func genFinalMapName(taskIndex, n int) string {
	return fmt.Sprintf("mr-%d-%d",taskIndex,n)
}

// 生成临时 reduce file name
func genTmpReduceName(wid, ti int) string {
	return fmt.Sprintf("tmp-worker-%d-out-%d", wid, ti)
}

// s
func genFinaReduceName(index int) string {
	return fmt.Sprintf("mr-out-%d",index)
}

开始

按lab1题目给我们的提示是：一个开始的方法是修改mr/worker.go的Worker()，向协调器发送一个要求任务的RPC。然后修改协调器，使其回应一个尚未启动的map任务的文件名。然后修改Worker以读取该文件并调用应用程序的Map函数，如mrsequential.go。

从提示和对整个mr包的阅读我们容易看出mr.Worker()是本次lab的着手点，因为tesh-mr.sh里面是启动了2-3个worker来测试，所有我们需要区分不同的终端产生的请求。并且由于worker需要不断向coordinator发送rpc请求任务，所有我们需要套一个死循环来请求rpc。

我们在请求rpc的时候需要拿取任务，也需要告知coordinator我们完成了哪些任务，方便coordinator的状态转换，这里有多种方法可以实现，像：

• 在这次请求时带上上一次请求执行的信息，告知coordinator我上次请求已经完成了

• 在完成时额外发送一个rpc，告知coordinator我这次请求已经完成了

因此可以大概写出worker函数的框架：

func Worker(mapf func(string, string) []KeyValue,
	reducef func(string, []string) string) {
	wid := os.Getpid()
	log.Printf("worker %v started", wid)
    
	var lastTaskIndex int
	var lastTaskType string
	// Your worker implementation here.
	// worker 需要不断向coordinator要任务
	for {
		// rpc请求
		args := AskTaskArgs{
			WorkerPID:     wid,
			LastTaskIndex: lastTaskIndex,
			LastTaskType:  lastTaskType,
		}
		reply := AskTaskReply{}
		call("Coordinator.AskForTask", &args, &reply)

		// mr任务均完成，可以退出
		if reply.Stage == "" {
			log.Printf("所有任务均已完成")
			break
		}
		
		if reply.TaskType == MapTask {
			// 如果coordinator返回的是Map任务
		} else if reply.TaskType == ReduceTask {
			// 如果coordinator返回的是reduce任务
		}
        // 让rpc携带上次请求的任务信息
		lastTaskType = reply.TaskType
		lastTaskIndex = reply.TaskIndex
	}
}

由上述的整体框架图，我们可知整个mr过程分为四个step

在step1开始前，coordinator需要有分配任务的阶段,这一阶段可在rpc请求中直接完成：

func (c *Coordinator) AskForTask(args *AskTaskArgs, reply *AskTaskReply) error {
	// 说明上一个是有任务的
	if args.LastTaskType != "" {
		c.mu.Lock()
		// 处理map和reduce任务
		c.mu.Unlock()
	}

	// 分配任务
	task,ok := <- c.cn
	if !ok {
		return nil
	}

	c.mu.Lock()
	defer c.mu.Unlock()
	// 可能会有多个 worker同时来要任务
	log.Printf("分配%v任务%v给worker %v",task.TaskType,task.TaskIndex,args.WorkerPID)
    // 设置超时时间
	task.Deadline = time.Now().Add(10*time.Second)
	task.WorkerPID = args.WorkerPID
	// 上面更改了，所以要重新写上去
	c.tasks[genTaskKey(task.TaskIndex,task.TaskType)] = task
    // 封装reply信息
	reply.TaskType = task.TaskType
	reply.TaskIndex = task.TaskIndex
	reply.NReduce = c.NReduce
	reply.NMap = c.NMap
	reply.File = task.FileName
	reply.Stage = c.stage
	return nil
}

Step 1

step 1 是worker生成 map临时文件，步骤是：

• worker向coordinator发送rpc请求，获取到任务，回复内容如下：

type AskTaskReply struct {
	Stage string // 状态
	File string // 文件名
	NReduce int // NReduce，reduce数
	NMap int // 任务数
	TaskIndex int // 任务索引
	TaskType string // 任务类型
}

• 根据 reply.File 打开我们需要读取的文件

• 根据 ihash(val.Key) % reply.NReduce 写入临时文件

实现如下：

if reply.TaskType == MapTask {
    // 生成临时的文件名 tmp-worker-workerID-TaskIndex-_NReduce
    fileName := reply.File
    file, err := os.Open(fileName)
    if err != nil {
        log.Fatalf("open file %v err:%v", fileName, err)
    }
    content, err := ioutil.ReadAll(file)
    if err != nil {
        log.Fatalf("cannot read %v err:%v", fileName, err)
    }
    file.Close()
    kva := mapf(fileName, string(content))
    // 写进临时文件，按key的ihash进行分桶
    hashed := make(map[int][]KeyValue)
    for _, val := range kva {
        key := ihash(val.Key) % reply.NReduce
        hashed[key] = append(hashed[key], val)
    }
    // 写入中间文件
    for j := 0; j < reply.NReduce; j++ {
        name := genTmpMapName(wid, reply.TaskIndex, j)
        mapfile, err := os.Create(name)
        if err != nil {
            log.Fatalf("create file %v err:%v", name, err)
        }
        for _, val := range hashed[j] {
            fmt.Fprintf(mapfile, "%v\t%v\n", val.Key, val.Value)
        }
        mapfile.Close()
    }
} 

Step 2

step 2 是若worker带过来的rpc请求里有上一个TaskInfo，则说明上一个map任务已经完成，这时候就可以原子的重命名这些临时文件

coordinator的操作比较简单:

// 说明上一个是有任务的
	if args.LastTaskType != "" {
		c.mu.Lock()
		taskKey := genTaskKey(args.LastTaskIndex, args.LastTaskType)
		if t,exist := c.tasks[taskKey];exist && t.WorkerPID == args.WorkerPID {
			log.Printf(
				"%s任务%d 在 %d 上完成\n",
				t.TaskType, t.TaskIndex, args.WorkerPID)
			if args.LastTaskType == MapTask {
				// 如果是map任务,且属于这个主机
				for i := 0; i < c.NReduce; i++ {
					// 忽略掉workerID合并成 NReduce*TaskNum 个文件
					err := os.Rename(genTmpMapName(args.WorkerPID, args.LastTaskIndex, i),
						genFinalMapName(args.LastTaskIndex, i))
					if err != nil {
						log.Fatalf("os rename %v error:%v",
							genTmpMapName(args.WorkerPID, args.LastTaskIndex, i), err)
					}
				}
			} else if args.LastTaskType == ReduceTask {
				
			}
			// 删除现在的task
			delete(c.tasks,taskKey)
			// 如果没任务了就需要更改状态了
			if len(c.tasks) == 0 {
				c.transition(c.NMap,c.NReduce)
			}
		}
		c.mu.Unlock()
	}

step 3

step 3 是在map任务已经全部完成的前提下，worker开始处理reduce任务，这里涉及到状态转换:

func (c *Coordinator) transition(files,nReduce int) {
	// debug
	if c.stage == MapTask {
		log.Println("开始分配reduce任务")
		c.stage = ReduceTask
		// 将Reduce任务加进去
		for i := 0;i < c.NReduce;i++ {
			task := Task{
				TaskType:  ReduceTask,
				// 0~NReudce-1
				TaskIndex: i,
			}
			c.tasks[genTaskKey(task.TaskIndex,task.TaskType)] = task
			c.cn <- task	
		}
	}else if c.stage == ReduceTask {
		log.Printf("mr结束，准备退出")
		close(c.cn)
		c.stage = ""
	}
}

接着便是worker对reduce任务的处理：

else if reply.TaskType == ReduceTask {
    var contents []string
    for i := 0; i < reply.NMap; i++ {
        // 这个循环意在将NReduce*len(files)个文件重新分成NReduce个文件
        // 即分为len(files)个tmpReduceFile
        // i的范围是0~7，代表输入文件的数量，或者说Map任务的数量
        // reply.TaskInfo.TaskIndex是0~9，代表Reduce任务的数量
        // 这里把前面多个主机完成的Map任务给汇总起来
        // Map任务的实现是通过ihash来选reduce，所以这里合并所有map产生的文件
        // 变成NReduce个文件，每个reduce一份文件
        fname := genFinalMapName(i, reply.TaskIndex)
        file, err := os.Open(fname)
        if err != nil {
            log.Fatalf("open %v err:%v", fname, err)
        }
        // 读文件
        content, err := ioutil.ReadAll(file)
        if err != nil {
            log.Fatalf("read file %s err:%v", fname, err)
        }
        contents = append(contents, strings.Split(string(content), "\n")...)
    }
    
    var kva []KeyValue
    for _, content := range contents {
        // 如果一行为空，就不算单词
        if strings.TrimSpace(content) == "" {
            continue
        }
        s := strings.Split(content, "\t")
        kva = append(kva, KeyValue{
            Key:   s[0],
            Value: s[1],
        })
    }
    // 排序
    sort.Sort(ByKey(kva))
    // 写入真正的reduce 临时文件
    fname := genTmpReduceName(wid, reply.TaskIndex)
    file, err := os.Create(fname)
    if err != nil {
        log.Fatalf("create file %v err:%v", fname, err)
    }

    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)
        }
        output := reducef(kva[i].Key, values)
        fmt.Fprintf(file, "%v %v\n", kva[i].Key, output)
        i = j
    }
    file.Close()
}

step 4

step 4 是处理worker完成的reduce任务，比较简单，不多说：

func (c *Coordinator) AskForTask(args *AskTaskArgs, reply *AskTaskReply) error {
	// 说明上一个是有任务的
	if args.LastTaskType != "" {
		c.mu.Lock()
		taskKey := genTaskKey(args.LastTaskIndex, args.LastTaskType)
		if t,exist := c.tasks[taskKey];exist && t.WorkerPID == args.WorkerPID {
			...
		} else if args.LastTaskType == ReduceTask {
				// 如果是reduce任务,且属于这个主机
				err := os.Rename(genTmpReduceName(args.WorkerPID, args.LastTaskIndex),
					genFinaReduceName(args.LastTaskIndex))

				if err != nil {
					log.Fatalf("os rename %v error:%v",
						genTmpReduceName(args.WorkerPID, args.LastTaskIndex), err)
				}
			}
			// 删除现在的task
			delete(c.tasks,taskKey)
			// 如果没任务了就需要更改状态了
			if len(c.tasks) == 0 {
				c.transition(c.NMap,c.NReduce)
			}
		}
		c.mu.Unlock()
	}

	// 分配任务
	...
}

至此，lab 1完成，debug真的痛苦

记录一下三天的debug之旅，第一天写了又写，多种方案都不行，后面上网查了一些思路，看到了呆呆大神的思路讲解，也才终于清晰了很多。

image-20220102155301075

可能出现的错误

我在测试的时候出现了一个错误

reduce parallelism test fail --- too few parallel reduces.

--- reduce parallelism test: FAIL

2022/01/02 15:16:36 分配reduce任务0给worker 5338
2022/01/02 15:16:36 Received reduce task 0 from coordinator
2022/01/02 15:16:37 Mark reduce task 0 as finished on worker 5338
2022/01/02 15:16:37 分配reduce任务1给worker 5338
2022/01/02 15:16:37 Received reduce task 1 from coordinator
2022/01/02 15:16:38 Mark reduce task 1 as finished on worker 5338
2022/01/02 15:16:38 分配reduce任务2给worker 5338
2022/01/02 15:16:38 Received reduce task 2 from coordinator
2022/01/02 15:16:39 Mark reduce task 2 as finished on worker 5338
2022/01/02 15:16:39 分配reduce任务3给worker 5338
2022/01/02 15:16:39 Received reduce task 3 from coordinator
2022/01/02 15:16:41 Mark reduce task 3 as finished on worker 5338
2022/01/02 15:16:41 分配reduce任务4给worker 5338
2022/01/02 15:16:41 Received reduce task 4 from coordinator
2022/01/02 15:16:43 Mark reduce task 4 as finished on worker 5338
2022/01/02 15:16:43 分配reduce任务5给worker 5338
2022/01/02 15:16:43 Received reduce task 5 from coordinator
2022/01/02 15:16:44 Mark reduce task 5 as finished on worker 5338
2022/01/02 15:16:44 分配reduce任务6给worker 5338
2022/01/02 15:16:44 Received reduce task 6 from coordinator
2022/01/02 15:16:45 Mark reduce task 6 as finished on worker 5338
2022/01/02 15:16:45 分配reduce任务7给worker 5338
2022/01/02 15:16:45 Received reduce task 7 from coordinator
2022/01/02 15:16:45 Mark reduce task 7 as finished on worker 5338
2022/01/02 15:16:45 分配reduce任务8给worker 5338
2022/01/02 15:16:45 Received reduce task 8 from coordinator
2022/01/02 15:16:45 Mark reduce task 8 as finished on worker 5338
2022/01/02 15:16:45 分配reduce任务9给worker 5338
2022/01/02 15:16:45 Received reduce task 9 from coordinator
2022/01/02 15:16:46 Mark reduce task 9 as finished on worker 5338
2022/01/02 15:16:46 All REDUCE tasks finished. Prepare to exit

从脚本输出可以看到：

开了两个窗口来接任务，map的时候还有两个，结果到了reduce的时候却没了，只剩下一个了，这样很显然就是错的，查看代码发现，代码中写了一行：

​ 在transition函数中：

// 清空管道
c.cn = make(chan Task,int(math.Max(float64(files),float64(nReduce))))

当时是想着，如果不清空管道，要是里面还有任务堆积，那下面的 c.cn <- task 不就会阻塞了吗（因为存在定时清除且放回channel的操作）

for i := 0;i < c.NReduce;i++ {
			task := Task{
				TaskType:  ReduceTask,
				// 0~9
				TaskIndex: i,
			}
			c.tasks[genTaskKey(task.TaskIndex,task.TaskType)] = task
			c.cn <- task
}

后面发现，只要管道不为0，len(c.tasks)是不可能为0的，也就是说，只有当管道和任务同时清空的时候才代表map or reduce任务结束，因为要是任务被放回channel，那只能是代表失败了，那肯定是没有执行过delete操作的

注：early exit test 会报错，换一台Linux机或把 test-mr.sh 中 wait -n 的 -n 去掉