mit 6.824 lab1分析

6.824 lab1 笔记

1. 阅读论文

2. 官网rules & hints

2.1 rules

  1. map阶段每个worker应该把中间文件分成nReduce份,nReduce是reduce任务的数量
  2. worker完成reduce任务后生成文件名mr-out-X
  3. mr-out-X文件每行应该是"%v %v"格式,参考main/mrsequential.go
  4. worker处理完map任务,应该把生成的中间文件放到当前目录中,便于worker执行reduce任务时读取中间文件
  5. 当所有任务完成时,Done()函数应该返回true,使得coordinator退出
  6. 所有任务完成时,worker应该退出,方法是:
    1. 当worker调用rpc向coordinator请求任务时,连接不上coordinator,此时可以认为coordinator已经退出因为所有任务已经完成了
    2. 当worker调用rpc向coordinator请求任务时,coordinator可以向其回复所有任务已经完成

2.2 hints

  1. 刚开始可以修改mr/worker.go's ``Worker()向coordinator 发送一个RPC请求一个任务。然后修改coordinator回复一个文件名,代表空闲的map任务。然后worker根据文件名读取文件,调用wc.so-Map函数,调用Map函数可参考mrsequential.go`

  2. 如果修改了mr/目录下任何文件,应该重新build MapReduce plugins,go build -buildmode=plugin ../mrapps/wc.go

  3. worker处理完map任务后产生的中间文件命名格式mr-X-Y,x是map任务的编号,y是reduce任务编号。

    // 初始文件,通过命令行传入的,如
    // pg-being_ernest.txt pg-dorian_gray.txt pg-frankenstein.txt 
    // len(files) = 3 nReduce = 4
    // 中间文件  x:map任务的编号 y:reduce任务编号
    // mr-0-0 mr-1-0 mr-2-0
    // mr-0-1 mr-1-1 mr-2-1
    // mr-0-2 mr-1-2 mr-2-2
    // mr-0-3 mr-1-3 mr-2-3
    
  4. map任务存储数据到文件可以使用json格式,便于reduce任务读取

      // map
      enc := json.NewEncoder(file)
      for _, kv := ... {
        err := enc.Encode(&kv)
          
      // reduce
      dec := json.NewDecoder(file)
      for {
        var kv KeyValue
        if err := dec.Decode(&kv); err != nil {
          break
        }
        kva = append(kva, kv)
      }
    
  5. map阶段使用ihash(key)函数把key映射到哪个reduce任务,如某个worker取得了2号map任务,ihash("apple") = 1,那么就应该把该key放到mr-2-1文件中

  6. 可以参考mrsequential.go代码:读取初始输入文件、排序key、存储reduce输出文件

  7. coordinator是rpc server,将会被并发访问,需要对共享变量加锁

  8. 若当前未有空闲的map任务可以分配,worker应该等待一段时间再请求任务,若worker频繁请求任务,coordinator就会频繁加锁、访问数据、释放锁,浪费资源和时间。如使用time.Sleep(),worker可以每隔一秒发送一次请求任务rpc

  9. coordinator无法辨别某个worker是否crash,有可能某个worker还在运行,但是运行极其慢(由于硬件损坏等原因),最好的办法是:coordinator监控某个任务,若该任务未在规定时间内由worker报告完成,那么coordinator可以把该任务重新分配给其他worker,该lab规定超时时间是10s

  10. 为了确保某个worker在写入文件时,不会有其他worker同时写入;又或者是某个worker写入文件时中途退出了,只写了部分数据,不能让这个没写完的文件让其他worker看到。可以使用临时文件ioutil.TempFile,当写入全部完成时,再使用原子重命名os.Rename

  11. Go RPC只能传struct中大写字母开头的变量

  12. 调用RPC call() 函数时,reply struct应该为空,不然会报错

      reply := SomeType{}
      call(..., &reply)
    

3. 架构设计

3.1 RPC设计

在该lab中,我们需要两个RPC,一个是callTask RPC向coordinator请求一个任务,一个是callTaskDone RPC向coordinator报告某个任务的完成,以下皆在rpc.go中定义

  1. 首先定义一个枚举变量,表示coordinator给worker分配的任务类型,也可用来表示coordinator当前的phase

    type taskType int
    
    const (
        // map任务
    	mapType taskType = iota
        // reduce任务
    	reduceType
        // 当前没有空闲任务,请等待
        waitting
        // 已经完成全部任务,可以退出了
    	done
    )
    
  2. 定义拉取任务RPC的args和reply struct

    CallTaskArgs中不需要包含变量,只需要让coordinator知道该worker正在请求一个任务,coordinator会随机选择空闲任务进行分配填入CallTaskReply

    CallTaskReply包含以下变量:

    • FileName:map阶段,worker需要知道具体的文件名才能解析该文件
    • tp:指示该任务的具体类型
    • TaskID:worker将该变量放入CallTaskDoneArgs中,coordinator可以迅速定位Task[TaskID],并且在reduce阶段中,搭配nFiles变量,worker读取mr-0-TaskIDmr-1-TaskID....mr-nFiles-1-TaskID文件
    • nFiles:初始文件的数量,用于搭配TaskID,在上面已介绍
    • nReduce:用于map阶段,ihash(key) % nReduce
    type CallTaskArgs struct {
    }
    type CallTaskReply struct {
    	FileName string
    	TaskID   int
    	tp       taskType
    	nFiles   int
    	nReduce  int
    }
    
  3. 定义报告任务完成RPC的args和reply struct

    TaskID变量作用在CallTaskReply: TaskID 中提及

    tp的作用是用于coordinator判断该RPC是否是合法的,举例:worker-1成功请求到map-1任务,但是因为worker-1节点硬件问题处理缓慢而导致coordinator检测到该map-1任务超时,于是把map-1任务分配给了worker-2。等到某个时间点,已经完成所有map任务,coordinator进入到了reduce阶段,但此时worker-1节点才刚运行完map-1任务并报告给coordinator,coordinator检测到当前是reduce阶段,但收到报告完成的rpc是map类型,不会对其进行任何操作。

    type CallTaskDoneArgs struct {
    	TaskID int
    	tp     taskType
    }
    type CallTaskDoneReply struct {
    }
    

3.2 Coordinator

3.2.1 结构体设计

type taskState int

const (
	spare taskState = iota
	executing
	finish
)

type task struct {
	fileName string
	id       int
	state    taskState
	start    time.Time
}

首先设计一个task struct,该结构体代表一个任务

  • filename:在map阶段,用于coordinator告知worker要读取的初始文件
  • id: 该任务的id,传给worker,作用在RPC设计中提及
  • state:任务有三个状态:空闲、执行中、已完成。若空闲则可以分配给worker;若执行中,则监视该任务是否超时
  • start:任务刚开始执行的时间
type Coordinator struct {
	// Your definitions here.
	mu         sync.Mutex
	state      taskType
	tasks      []*task
	mapChan    chan *task
	reduceChan chan *task
	nReduce    int
	nFiles     int
	finished   int
}

接着设计主要Coordinator结构体

  • state:当前系统的状态,map阶段(分配map任务)、reduce阶段(分配reduce任务)、全部完成done(可以结束系统运行)
  • tasks: *task的切片,维护了一组任务
  • mapChanreduceChan:用于分发map、reduce任务的channel。map阶段,若有空闲map任务,则放至channel中,当有worker请求任务时,则可取出来。reduce阶段同理
  • finished:当前已完成任务的数量。map阶段,若finished == nFiles,则表示所有map任务完成,可以进入reduce阶段。reduce阶段同理,进入done

3.2.2 初始化

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

	// Your code here.
	c.mapPhase(files, nReduce)
	go c.watch()
	c.server()
	return &c
}

func (c *Coordinator) mapPhase(files []string, nReduce int) {
	c.state = mapType                 //设置系统状态为map阶段
	c.nReduce = nReduce        
	c.nFiles = len(files)
	c.tasks = make([]*task, c.nFiles)
	c.mapChan = make(chan *task, c.nFiles) // c.nFiles长度的map channel
	for i := 0; i < c.nFiles; i++ {
		c.tasks[i] = &task{fileName: files[i], id: i}
		c.mapChan <- c.tasks[i]            // 刚开始所有任务都是空闲状态,放入channel中
	}
}

系统刚开始时即map阶段,mapPhase初始化Coordinator结构体。然后启动c.watch()协程,用于监视任务是否超时,放后面讲

3.2.3 分配任务

func (c *Coordinator) CallTask(args *CallTaskArgs, reply *CallTaskReply) error {
	c.mu.Lock()
	defer c.mu.Unlock()
	if c.state == done {
		reply.Tp = done
	} else if c.state == mapType {
		switch len(c.mapChan) > 0 {
		case true:
			task := <-c.mapChan
			c.setReply(task, reply)
		case false:
			reply.Tp = waitting
		}
	} else {
		switch len(c.reduceChan) > 0 {
		case true:
			task := <-c.reduceChan
			c.setReply(task, reply)
		case false:
			reply.Tp = waitting
		}
	}
	return nil
}

func (c *Coordinator) setReply(t *task, reply *CallTaskReply) {
	if t.state == finish {
		reply.Tp = waitting
		return
	}
	reply.Tp = c.state
	reply.TaskID = t.id
	reply.NReduce = c.nReduce
	reply.NFiles = c.nFiles
	reply.FileName = t.fileName
	t.state = executing
	t.start = time.Now()
}

分配任务的主要函数,worker处会调用call("Coordinator.CallTask", &args, &reply)

  1. 若当前系统状态为done,则返回done,告知worker可以退出了
  2. 若当前系统状态为map阶段:如果当前有任务可以分配len(c.mapChan) > 0,则取出一个task,调用c.setReply(task, reply),将任务的相关信息填入reply中,并把task的当前状态设为执行中,开始时间设为time.Now()。如果没有可分配的任务,则设reply.Tp = waitting,让worker等待一会再请求任务
  3. 若当前系统状态为reduce阶段:同上

3.2.4 任务完成

func (c *Coordinator) CallTaskDone(args *CallTaskDoneArgs, reply *CallTaskDoneReply) error {
	c.mu.Lock()
	defer c.mu.Unlock()
	if c.state != args.Tp || c.state == done {
		return nil
	}
	if c.tasks[args.TaskID].state != finish {
		c.tasks[args.TaskID].state = finish
		c.finished++
		//fmt.Printf("task %v done\n", args.TaskID)
		if c.state == mapType && c.finished == c.nFiles {
			c.reducePhase()
		} else if c.state == reduceType && c.finished == c.nReduce {
			close(c.reduceChan)
			c.state = done
		}
	}
	return nil
}

func (c *Coordinator) reducePhase() {
	//fmt.Printf("reduce phase\n")
	close(c.mapChan)
	c.state = reduceType
	c.tasks = make([]*task, c.nReduce)
	c.finished = 0
	c.reduceChan = make(chan *task, c.nReduce)
	for i := 0; i < c.nReduce; i++ {
		c.tasks[i] = &task{id: i}
		c.reduceChan <- c.tasks[i]
	}
}

worker处会调用call("Coordinator.CallTaskDone", &args, &reply)来报告某任务的完成

首先判断c.state != args.Tp,即报告完成的任务类型和当前系统状态不匹配,可能发生在该情况:work-1请求了map-1任务,但是work-1运行太缓慢导致Coordinator监测到map-1任务超时,于是把map-1任务分配给了work-2。当所有map任务完成时,Coordinator进入了reduce阶段,这时work-1才报告map-1任务完成,与当前系统状态不匹配,故会直接返回

若该任务未完成,则将该任务标记未已完成,c.finished++

  1. 如果当前为map阶段并且所有map任务已完成c.state == mapType && c.finished == c.nFiles,则进入reduce阶段:
    1. 关闭map channel
    2. 将系统状态设为reduce
    3. 重置c.tasks为一系列reduce任务
    4. 创建长度为c.nReduce的reduce channel
    5. 放入任务
  2. 如果当前为reduce阶段并且所有map任务已完成c.state == reduceType && c.finished == c.nReduce,则进入done阶段:
    1. 关闭reduce channel
    2. 将系统状态设为done

3.2.5 监测超时任务goroutine

func (c *Coordinator) watch() {
	for {
		time.Sleep(time.Second)
		c.mu.Lock()
		if c.state == done {
			return
		}
		for _, task := range c.tasks {
			if task.state == executing && time.Since(task.start) > timeout {
				task.state = spare
				switch c.state {
				case mapType:
					c.mapChan <- task
				case reduceType:
					c.reduceChan <- task
				}
			}
		}
		c.mu.Unlock()
	}
}

如果当前系统状态为done了,可以退出协程了

循环任务列表,如果该任务状态是正在执行但是超时了time.Since(task.start) > timeout(time.Since可以计算系统当前时间距离start过去了多少时间),将该任务状态重置为空闲状态,并且根据系统当前状态,把该任务重新放入对应的channel中

3.3 Worker

3.3.1 主流程

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

	for {
		args := CallTaskArgs{}
		reply := CallTaskReply{}
		ok := call("Coordinator.CallTask", &args, &reply)
		//now := time.Now()
		if ok {
			switch reply.Tp {
			case mapType:
				executeMap(reply.FileName, reply.NReduce, reply.TaskID, mapf)
			case reduceType:
				executeReduce(reply.NFiles, reply.TaskID, reducef)
			case waitting:
				time.Sleep(time.Second * 2)
				continue
			case done:
				os.Exit(0)
			}
		} else {
			time.Sleep(time.Second * 2)
			continue
		}
		//fmt.Printf("finish task: %v %v use %v\n", reply.TaskID, rs(reply.Tp), time.Since(now).Seconds())
		a := CallTaskDoneArgs{reply.TaskID, reply.Tp}
		r := CallTaskDoneReply{}
		call("Coordinator.CallTaskDone", &a, &r)
		time.Sleep(time.Second * 2)
	}
}

首先向Coordinator发送请求任务rpc:

  1. map任务:执行
  2. reduce任务:执行
  3. waitting:当前Coordinator没有空闲任务,sleep一段时间再请求
  4. done:所有任务已完成,退出

任务执行完成后,报告任务完成

3.3.2 执行map任务

func executeMap(fileName string, nReduce, taskID int, mapf func(string, string) []KeyValue) {
	file, err := os.Open(fileName)
	if err != nil {
		log.Fatalf("cannot open %v", fileName)
	}
	content, err := ioutil.ReadAll(file)
	if err != nil {
		log.Fatalf("cannot read %v", fileName)
	}
	file.Close()
	kva := mapf(fileName, string(content))
    // 上面的代码参考mrsequential.go
	files := []*os.File{}
	tmpFileNames := []string{}
	encoders := []*json.Encoder{}
	for i := 0; i < nReduce; i++ {
		tempFile, err := ioutil.TempFile("./", "")
		if err != nil {
			log.Fatalf("cannot open temp file")
		}
		files = append(files, tempFile)
		tmpFileNames = append(tmpFileNames, tempFile.Name())
		encoders = append(encoders, json.NewEncoder(tempFile))
	}
	for _, kv := range kva {
		n := ihash(kv.Key) % nReduce
		encoders[n].Encode(kv)
	}
	for i := 0; i < nReduce; i++ {
		files[i].Close()
		os.Rename(tmpFileNames[i], "./"+intermediateFileName(taskID, i))
	}
}

在当前目录创建nReduce个临时文件ioutil.TempFile("./", ""),使用该临时文件创建json.Encoder(在hints第四条),使用ihash函数把每个key映射到哪个文件,写入json格式,然后对每个临时文件重命名为mr-x-y格式

生成中间文件名函数:

func intermediateFileName(x, y int) string {
	return fmt.Sprintf("mr-%v-%v", x, y)
}

3.3.3 执行reduce

func executeReduce(nFiles, taskID int, reducef func(string, []string) string) {
	kvs := []KeyValue{}
	for i := 0; i < nFiles; i++ {
		filename := intermediateFileName(i, taskID)
        // 读取每个中间文件
		file, err := os.Open(filename)
		if err != nil {
			log.Fatalf("cannot open %v", filename)
		}
        // 参考hints第四条,从文件中取出json格式的每条数据
		decoder := json.NewDecoder(file)
		for {
			var kv KeyValue
            // 已读到文件末尾
			if err := decoder.Decode(&kv); err != nil {
				break
			}
			kvs = append(kvs, kv)
		}
		file.Close()
	}
    // 参考mrsequential.go
	oname := fmt.Sprintf("mr-out-%v", taskID)
	tempFile, _ := ioutil.TempFile("./", "")
	tempFileName := tempFile.Name()
	sort.Sort(ByKey(kvs))
	for i := 0; i < len(kvs); {
		j := i + 1
		for j < len(kvs) && kvs[j].Key == kvs[i].Key {
			j++
		}
		values := []string{}
		for k := i; k < j; k++ {
			values = append(values, kvs[k].Value)
		}
		output := reducef(kvs[i].Key, values)
		fmt.Fprintf(tempFile, "%v %v\n", kvs[i].Key, output)
		i = j
	}
	tempFile.Close()
	os.Rename(tempFileName, "./"+oname)
}

热门相关:地球第一剑   日本妈妈   天启预报   刺客之王   寂静王冠