MIT6.824-Lab1

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

type RequestTaskReply struct {
	Task       *Task
	NReduce    int
	DoneStatus bool
}

type ReportTaskResponse struct {
	WorkType  int // 0 map 1 reduce
	WorkId    int
	WorkFiles []string
}

type DoneTaskReply struct {
	X int
}

type TaskArgs struct{}

const (
	MapTask    = 0
	ReduceTask = 1

	Maping    = 0
	Reduceing = 1

	Ready    = 0
	Running  = 1
	Finished = 2
)

1
2
3
4
5
6
7
8
9
10
11

type Master struct {
	MapQue      []*Task // Map任务队列 未完成的
	ReduceQue   []*Task // Reduce任务队列 未完成的
	Files       []string // 要处理的文件
	MapCount    int // Map任务 现在看来是没用了，因为本来是我给MapQue写的是channel后来改成了切片
	ReduceCount int // Reduce任务数量 用于work生成temp文件
	Whiching    int // 正在进行什么任务，map or reduce
	Lock        sync.Mutex // 锁
	WorkID      int // id
	IsDone      bool // 是否完成
}

1
2
3
4
5
6
7
8

type Task struct {
	TaskType   int // Task的类型 map or reduce
	MapId      int // map任务的id
	ReduceId   int // reduce任务id
	TaskStatus int // 状态 finished running ready
	InputFile  string // 操作的文件
	BeginTime  time.Time // 开始时间，用于处理crash
}

1
2
3
4
5
6
7
8
9
10

m.Lock.Lock()
for _, file := range files {
	task := Task{
		TaskType:   MapTask,
		InputFile:  file,
		TaskStatus: Ready,
		MapId:      0,
	}
	m.MapQue = append(m.MapQue, &task)
}

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

		if flag := func() bool {
			m.Lock.Lock()
			for i := 0; i < len(m.MapQue); i++ {
				task := m.MapQue[i]
				if task.TaskStatus != Finished {
					m.Lock.Unlock()
					return false
				}
			}
			m.Lock.Unlock()
			return true
		}(); flag {
			m.Whiching = Reduceing
			m.Lock.Lock()
			for i := 0; i < m.ReduceCount; i++ {
				task := Task{
					TaskType:   ReduceTask,
					InputFile:  fmt.Sprintf("/var/tmp/mr-*-%v", i),
					TaskStatus: Ready,
					ReduceId:   0,
				}
				m.ReduceQue = append(m.ReduceQue, &task)
			}
			m.Lock.Unlock()
		}

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

func (m *Master) GetTask(args *TaskArgs, reply *RequestTaskReply) error{
    if m.Done(){
        reply.DoneStatus = true
        return nil
    }
    if m.Whiching == Maping{
        for _, task := range m.MapQue{
            if task.TaskStatus == Ready{
                // 分发
                return nil
            }
        }
        // 根据上面的 判断是否完成map任务并生成reduce任务
    }else{
        for _, task := range m.ReduceQue{
            if task.TaskStatus == Ready{
                // 分发
                return nil
            }
        }
        flag := func(){}() // 判断reduce是否都完成
        if flag{ // 所有任务都已经完成
            reply.DoneStatus = true
			time.Sleep(time.Second * 2)
			m.Lock.Lock()
			m.IsDone = true
			m.Lock.Unlock()
        }
    }
}

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

func (m *Master) ReportWorkDone(args *ReportTaskResponse, reply *DoneTaskReply) error{
    if args.WorkType == MapTask{
        for task := range m.MapQue{
            // 先判断任务是否超时
            if task.MapId == mapWorkId && task.TaskStatus == Running{
                //把temp文件移到工作目录
                task.TaskStatus = Finished
				for _, file := range args.WorkFiles {
					src, err := os.Open(file)
                    defer src.Close()
					if err != nil {
						log.Fatal( err)
					}
					DIR, _ := os.Getwd()
					filename := filepath.Base(file)
					f := DIR + "/" + filename
					dst, err := os.Create(f)
                    defer dst.Close()
					if err != nil {
						log.Fatal(err)
					}
					_, err = io.Copy(dst, src)
					if err != nil {
						log.Fatal(err)
					}
				}
				break
            }
        }
    }else{
        // 同上
    }
}

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

func maybeCrash() {
	max := big.NewInt(1000)
	rr, _ := crand.Int(crand.Reader, max)
	if rr.Int64() < 330 {
		// crash!
		os.Exit(1)
	} else if rr.Int64() < 660 {
		// delay for a while.
		maxms := big.NewInt(10 * 1000)
		ms, _ := crand.Int(crand.Reader, maxms)
		time.Sleep(time.Duration(ms.Int64()) * time.Millisecond)
	}
}

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

func (m *Master) HeartBeatForTimeOut() { // 非心跳机制
	for !m.Done() {
		time.Sleep(2 * time.Second)
		m.Lock.Lock()
		if m.Whiching == Maping {
			for _, task := range m.MapQue {
				if task.TaskStatus == Running && (time.Since(task.BeginTime) > time.Duration(10*time.Second)) {
		task.TaskStatus = Ready
	}
			}
		} else {
			for _, task := range m.ReduceQue {
				if task.TaskStatus == Running && (time.Since(task.BeginTime) > time.Duration(10*time.Second)) {
		task.TaskStatus = Ready
	}
			}
		}
		m.Lock.Unlock()
	}
}

1
2
3
4
5
6
7
8

type OneWorker struct {
	WorkerID int // 工作 id
	Mapf     func(string, string) []KeyValue // 程序提供的Mapf
	Reducef  func(string, []string) string // 程序提供的Reducef
	Task     *Task // 执行的任务
	NReduce  int // educe任务的数量
	IsDone   bool // 是否完成
}

1
2
3
4
5
6
7
8
9
10
11

func Worker(mapf func(string, string) []KeyValue, reducef func(string, []string) string) {
	worker := OneWorker{
		WorkerID: -1,
		Mapf:     mapf,
		Reducef:  reducef,
		IsDone:   false,
	}
	for !worker.IsDone {
		worker.work()
	}
}

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

func (work *OneWorker) work() {
	task := work.RequireTask() // 获取任务
	if task != nil {
		work.Task = task
		if task.TaskType == MapTask {
			work.WorkerID = task.MapId
			intermediate := work.generateMapResult(task) // 调用mapf生成结果
			file := work.writeMapTempFiles(intermediate) // 写到文件里面
			work.TaskDone(file, MapTask) // 上报
		} else {
			work.WorkerID = task.ReduceId
			intermediate := work.generatorReduceResult(task.InputFile) // 调用reducef
			files := work.writeReduceTempFiles(task, intermediate) // 写到文件
			work.TaskDone(files, ReduceTask) // 上报
		}
	}
}

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

func (work *OneWorker) RequireTask() *Task {
	args := TaskArgs{}
	reply := RequestTaskReply{}
	call("Master.GetTask", &args, &reply)
	if &reply != nil {
		if reply.DoneStatus { // 如果任务都完成了
			work.IsDone = true // 当前任务不需要处理
			return nil
		}
		work.NReduce = reply.NReduce
		return reply.Task
	} else {
		return nil
	}
}

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

func (work *OneWorker) generateMapResult(task *Task) []KeyValue {
	NowIntermediate := make([]KeyValue, 0) // 中间文件
	file, err := os.Open(task.InputFile)
	if err != nil {
		log.Fatal(err)
	}
	content, err := ioutil.ReadAll(file)
	if err != nil {
		log.Fatal(err)
	}
	file.Close()
	kva := work.Mapf(task.InputFile, string(content))
	NowIntermediate = append(NowIntermediate, kva...)
	return NowIntermediate
}

func (work *OneWorker) generatorReduceResult(task *Task) []KeyValue {
	intermedia := make([]KeyValue, 0)
	files, err := filepath.Glob(task.InputFile)
	if err != nil {
		log.Fatal(err)
	}
	for _, filepath := range files {
		file, err := os.Open(filepath)
		if err != nil {
			log.Fatal(err)
		}
		dec := json.NewDecoder(file)
		for {
			var kv KeyValue
			if err := dec.Decode(&kv); err != nil {
				break
			}
			intermedia = append(intermedia, kv)
		}
	}
    // mrsequential
	sort.Sort(ByKey(intermedia))
	res := make([]KeyValue, 0)
	i := 0
	for i < len(intermedia) {
		j := i + 1
		for j < len(intermedia) && intermedia[j].Key == intermedia[i].Key {
			j++
		}
		values := make([]string, 0)
		for k := i; k < j; k++ {
			values = append(values, intermedia[k].Value)
		}
		//defer work.CrashRecover()
		output := work.Reducef(intermedia[i].Key, values)
		kv := KeyValue{Key: intermedia[i].Key, Value: output}
		res = append(res, kv)
		i = j
	}
	return res
}

1
2
3
4
5
6
7
8
9
10

func (work *OneWorker) TaskDone(files []string, Type int) {
	args := ReportTaskResponse{
		WorkType:  Type,
		WorkId:    work.WorkerID,
		WorkFiles: files,
	}
	reply := DoneTaskReply{}
	call("Master.ReportWorkDone", &args, &reply)
}