mit-6.5840(6.824) lab1

思路

map和reduce都是无状态的纯函数，所以wowker也不应该涉及任何状态的保存

所以map只有需要操作两件事情

申请任务
报告任务

分开看，申请任务不需要区分申请map或者reduce，所以只要在worker中一直申请就好，任务一旦完成，就立即报告任务完成

1
func Worker(mapf func(string, string) []KeyValue,
2
  reducef func(string, []string) string) {
3
out:
4
  for {
5
    res := RequestNewTaskFunc()
6
    switch res.Type {
7
    case Map:
8
      HandleMap(mapf, res.Map.FileName, res.Map.ReduceTaskNum, res.Map.TaskID)
9
      ReportTaskCompleteFunc(Map, res.Map.TaskID)
10
    case Reduce:
11
      HandleReduce(reducef, res.Reduce.TaskID, res.Reduce.MapTaskNum)
12
      ReportTaskCompleteFunc(Reduce, res.Reduce.TaskID)
13
    case Continue:
14
      continue out
15
    case Stop:
16
      break out
17
    }
18
  }
19
}

对于coordinator，利用channel的机制来避免外部的锁 channel 和其他语言的管道应该是一样的东西，理解起来还是挺简单的

1
type Coordinator struct {
2
  // Your definitions here.
3
  mapTask                []MapTask
4
  reduceTask             []ReduceTask
5
  mapChan                chan int //需要执行的map任务taskID
6
  reduceChan             chan int//需要执行的reduce任务taskID
7
  mapTaskNum             int
8
  reduceTaskNum          int
9
  completedMapTaskNum    int
10
  completedReduceTaskNum int
11
  reportChan             chan ReportTaskCompleteArgs
12
}

使用了两个channel分别保存了需要执行的map任务和reduce任务使用reportChan来处理时间完成的状态修改，来避免多个worker上报同一个任务时状态不对的问题

1
func MakeCoordinator(files []string, nReduce int) *Coordinator {
2
  c := Coordinator{}
3

4
  // 初始化reduce任务
5
  c.completedReduceTaskNum = 0
6
  c.reduceChan = make(chan int, nReduce)
7
  for i := range nReduce {
8
    c.reduceTask = append(c.reduceTask, ReduceTask{partition: i, completed: false})
9
    c.reduceChan <- i
10
  }
11

12
  // 初始化map任务
13
  c.completedMapTaskNum = 0
14
  c.mapChan = make(chan int, len(files))
15
  for i, file := range files {
16
    c.mapTask = append(c.mapTask, MapTask{filename: file, completed: false})
17
    c.mapChan <- i
18
  }
19

20
  c.reportChan = make(chan ReportTaskCompleteArgs)
21

22
  //处理汇报事件
23
  go func() {
24
    for {
25
      task := <-c.reportChan
26
      // log.Printf("%v", task)
27
      taskID := task.TaskID
28
      switch task.TaskType {
29
      case Map:
30
        if !c.mapTask[taskID].completed {
31
          c.mapTask[taskID].completed = true
32
          c.completedMapTaskNum += 1
33
        }
34

35
      case Reduce:
36
        if !c.reduceTask[taskID].completed {
37
          c.reduceTask[taskID].completed = true
38
          c.completedReduceTaskNum += 1
39
        }
40
      }
41
    }
42
  }()
43
  c.server()
44
  return &c
45
}

根据已经完成的任务数量来判断当前阶段，只有当map全部完成时才进入reduce，如果map没有完成，有worker请求任务，则让其等待

1
func (c *Coordinator) Call(req *RequestArgs, res *RPCResponseArgs) error {
2
  // log.Printf("%v", req)
3
  switch req.Type {
4
  case RequestNewTask:
5
    if c.completedReduceTaskNum == len(c.reduceTask) {
6
      res.Type = Stop
7
      return nil
8
    }
9
    if c.completedMapTaskNum == len(c.mapTask) {
10
      select {
11
      case taskID := <-c.reduceChan:
12
        res.Type = Reduce
13
        res.Reduce = ReduceArgs{TaskID: taskID, MapTaskNum: len(c.mapTask)}
14
        go func() {
15
          time.AfterFunc(10*time.Second, func() {
16
            if !c.reduceTask[taskID].completed {
17
              c.reduceChan <- taskID
18
            }
19
          })
20
        }()
21
      default:
22
        res.Type = Continue
23
      }
24

25
      return nil
26
    }
27

28
    select {
29
    case taskID := <-c.mapChan:
30
      res.Type = Map
31
      res.Map = MapArgs{TaskID: taskID, FileName: c.mapTask[taskID].filename, ReduceTaskNum: len(c.reduceTask)}
32
      go func() {
33
        time.AfterFunc(10*time.Second, func() {
34
          if !c.mapTask[taskID].completed {
35
            c.mapChan <- taskID
36
          }
37
        })
38
      }()
39
    default:
40
      res.Type = Continue
41
    }
42
    return nil
43
  case ReportTaskComplete:
44
    c.reportChan <- req.ReportTaskCompleteArgs
45
    res.Type = OK
46
  }
47

48
  return nil
49
}