Mit6.824之lab1

1
2
3
4
5
6
7
8

// TaskType 任务类型，worker会承担map任务或者reduce任务
type TaskType uint8

const (
	Map TaskType = iota + 1
	Reduce
	Done
)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18

type Coordinator struct {
	// Your definitions here.

	mu   sync.Mutex
	cond *sync.Cond

	mapFiles     []string // map任务的输入文件
	nMapTasks    int      // map任务数，lab里规定的一个map任务对应一个worker。len(mapFiles) = nMapTasks
	nReduceTasks int

	// 记录任务的完成情况和何时分配，用以跟踪worker的处理进度
	mapTasksFinished        []bool      // 记录map任务完成情况
	mapTasksWhenAssigned    []time.Time // 记录map任务何时分配
	reduceTasksFinished     []bool      // 记录reduce任务完成情况
	reduceTasksWhenAssigned []time.Time // 记录reduce任务何时分配

	done bool // 所有reduce任务是否已经完成，也就是整个job是否完成
}

 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

func (c *Coordinator) GetTask(args *GetTaskArgs, reply *GetTaskReply) error {
	c.mu.Lock()
	defer c.mu.Unlock()

	reply.NMapTasks = c.nMapTasks
	reply.NReduceTasks = c.nReduceTasks

	// 分配worker map任务，直到所有map任务执行完毕
	for {
		mapDone := true
		for i, done := range c.mapTasksFinished {
			if !done /*task没有完成*/ {
				if c.mapTasksWhenAssigned[i].IsZero() || /*任务是否被分配*/
					time.Since(c.mapTasksWhenAssigned[i]).Seconds() > 10 /*分配出去10s还没完成，认为worker寄掉了*/ {
					reply.TaskType = Map
					reply.TaskId = i
					reply.MapFileName = c.mapFiles[i]
					c.mapTasksWhenAssigned[i] = time.Now()
					return nil
				} else {
					mapDone = false
				}
			}
		}

		if !mapDone { /*没有完成，我们需要阻塞*/
			c.cond.Wait()
		} else { /*全部map任务已经完成了*/
			break
		}
	}

	// 此时，所有的map任务已经做完了，可以开始reduce任务了
	for {
		rDone := true
		for i, done := range c.reduceTasksFinished {
			if !done /*task没有完成*/ {
				if c.reduceTasksWhenAssigned[i].IsZero() || /*任务是否被分配*/
					time.Since(c.reduceTasksWhenAssigned[i]).Seconds() > 10 /*分配出去10s还没完成，认为worker寄掉了*/ {
					reply.TaskType = Reduce
					reply.TaskId = i
					c.reduceTasksWhenAssigned[i] = time.Now()
					return nil
				} else {
					rDone = false
				}
			}
		}
		if !rDone { /*没有完成，我们需要等待*/
			c.cond.Wait()
		} else { /*全部map任务已经完成了*/
			break
		}
	}

	// if the job is done
	reply.TaskType = Done
	c.done = true

	return nil
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16

func (c *Coordinator) FinishedTask(args *FinishedTaskArgs, reply *FinishedTaskReply) error {
	c.mu.Lock()
	defer c.mu.Unlock()
	switch args.TaskType {
	case Map:
		c.mapTasksFinished[args.TaskId] = true
	case Reduce:
		c.reduceTasksFinished[args.TaskId] = true
	default:
		return errors.New("coordinator: not support this task type")
	}

	c.cond.Broadcast()

	return nil
}

1
2
3
4
5
6
7
8
9

	// 唤醒
	go func() {
		for {
			c.mu.Lock()
			c.cond.Broadcast()
			c.mu.Unlock()
			time.Sleep(time.Second)
		}
	}()

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

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

	// Your worker implementation here.
	for {
		args := GetTaskArgs{}
		reply := GetTaskReply{}
		call("Coordinator.GetTask", &args, &reply)
		switch reply.TaskType {
		case Map:
			performMapTask(reply.MapFileName, reply.TaskId, reply.NReduceTasks, mapf)
		case Reduce:
			performReduce(reply.TaskId, reply.NMapTasks, reducef)
		case Done:
			return
		default:
			log.Fatal("worker: not support this task type")
		}

		call("Coordinator.FinishedTask",
			&FinishedTaskArgs{TaskId: reply.TaskId, TaskType: reply.TaskType}, &FinishedTaskReply{})
	}
}

 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

func performMapTask(fn string, taskN int, nReduceTasks int, mapf func(string, string) []KeyValue) {
	f, err := os.Open(fn)
	if err != nil {
		log.Fatalf("open error:%v", err)
	}

	content, err := io.ReadAll(f)
	if err != nil {
		log.Fatalf("read error:%v", err)
	}
	f.Close()

	kvs := mapf(fn, quickB2S(content))

	// 处理中间文件
	tmpFs := make([]*os.File, 0, nReduceTasks)
	tmpFNs := make([]string, 0, nReduceTasks)
	encoders := make([]*json.Encoder, 0, nReduceTasks)

	for r := 0; r < nReduceTasks; r++ {
		tmpF, err := os.CreateTemp("", "")
		if err != nil {
			log.Fatalf("create temporary file error: %v", err)
		}

		tmpFs = append(tmpFs, tmpF)
		tmpFNs = append(tmpFNs, tmpF.Name())
		encoders = append(encoders, json.NewEncoder(tmpF))
	}

	// 写入临时的中间文件
	// 先json编码
	for _, kv := range kvs {
		err := encoders[ihash(kv.Key)%nReduceTasks].Encode(&kv)
		if err != nil {
			log.Fatalf("json encode error: %v", err)
		}
	}

	// 关闭临时文件
	for _, f := range tmpFs {
		f.Close()
	}

	// 原子重命名，其实是为了防止重名和原子性破坏
	for r := 0; r < nReduceTasks; r++ {
		err := renameIntermediateFIle(tmpFNs[r], taskN, r)
		if err != nil {
			log.Fatalf("rename error: %v", err)
		}
	}
}

 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

func performReduce(taskN, nMapTasks int, reducef func(string, []string) string) {
	kvs := make([]KeyValue, 0)
	for i := 0; i < nMapTasks; i++ {
		intermediatef := getIntermediateFIle(i, taskN)
		f, err := os.Open(intermediatef)
		if err != nil {
			log.Fatalf("open error: %v", err)
		}
		dec := json.NewDecoder(f)
		for {
			var kv KeyValue
			if err := dec.Decode(&kv); err != nil {
				break
			}
			kvs = append(kvs, kv)
		}
		f.Close()
	}

	// 排序
	sort.Sort(ByKey(kvs))

	// 再创建临时文件来写reduce的结果，后面通过原子重命名只有一个最终结果文件
	tmpf, err := os.CreateTemp("", "")
	if err != nil {
		log.Fatalf("create temporary file error: %v", err)
	}

	// 把相同的key的所有value聚合在一起
	i := 0
	j := 0
	for i < len(kvs) {
		j = i + 1
		for j < len(kvs) && kvs[j].Key == kvs[i].Key {
			j++
		}
		vs := make([]string, 0, len(kvs))
		for k := i; k < j; k++ {
			vs = append(vs, kvs[k].Value)
		}
		output := reducef(kvs[i].Key, vs)
		_, err := fmt.Fprintf(tmpf, "%v %v\n", kvs[i].Key, output)
		if err != nil {
			log.Fatalf("open or write error: %v", err)
		}
		i = j
	}

	err = renameReduceOutFile(tmpf.Name(), taskN)
	if err != nil {
		log.Fatalf("rename error: %v", err)
	}
}