欢迎您访问程序员文章站本站旨在为大家提供分享程序员计算机编程知识!
您现在的位置是: 首页

6.824(2020春) Lab1:MapReduce

程序员文章站 2022-03-15 17:13:08
...

前言

本来不打算记录这个实验过程的,一是有点懒;二是没太多好说。后来还是耐着自己那膨胀的小性子,简单记录下这个实验。

一、 概况

本次的实验是6.824分布式课程提供的,其提供了主要的代码框架部分,我们需要做的就是理解MapReduce的原理,然后在框架内实现它。
6.824(2020春) Lab1:MapReduce
  整个实现的MapReduce框架和paper中差不多,一个Master结点用于协调,还有多个Worker结点,每个结点可以向Master请求Map任务或者Reduce任务。

note:
(1)、实验提供的代码中的worker.go文件中的Worker函数,并不是生成所有的worker,而仅仅只是一个worker的执行流程。 多个worker函数是由上层应用进行生成的(实验中是测试代码test-mr.sh生成的)。
(2)、各个worker之间基本上没有通信,信息的传递是利用rpc向Master进行传递,进而被其他的worker了解到的。
(3)、一开始老是通不过最后一个crash测试,后来发现可能是原先用的内部逻辑架构不太好。原先设计的方案是由worker询问master是否所有的任务(CallForReduceFinishReply)都已经完成,然后在这个rpc调用的过程中顺便丢弃还未完成的任务,将其状态重新置为未分配,以便调度其他的worker完成这个任务。但是这样有一个问题,就是可能刚刚被分配的任务,又会被丢弃(因为有其他woker进行询问所有任务是否完成)。
  后来利用了时间片的方式,在Master中每一个时间滴答(TimeTick),将所有正在执行中的任务(RUNNING状态)加一个滴答计时,当超过了10秒之后,就把其状态改为(NOTALLOCATED)。这样操作之后,被动修改就变主动了。优秀的完成了任务。( ●^o^●)

代码

由于官方说明,不推荐贴代码,这里为了记录就粘贴下主要的部分吧。

master部分

// Your code here -- RPC handlers for the worker to call.
//响应Map任务请求
func (m *Master) ReplyForMapTask(args *CallForMapTaskArgs, reply *CallForMapTaskReplyArgs) error {

	reply.FileName = "" //默认返回是空的
	reply.MapId = -1
	reply.Ok = false

	m.MasterLock.Lock()

	//遍历寻找未分配的任务
	for mapId, mapTask := range m.AllMapTask {

		//fmt.Printf("asker:pid:%v--mapId:%v--mapTask:%v\n", args.TestPid, mapId, mapTask)

		if mapTask.status == NOTALLOCATED { //未分配


			reply.MapId = mapId
			reply.FileName = mapTask.TaskFileName
			reply.Ok = true

			m.AllMapTask[mapId].status = RUNNING //已分配,正在运行中
			m.AllMapTask[mapId].UsedTime = 0		//初始化任务的时间


			break
		}
	}

	m.MasterLock.Unlock()

	return nil

}

//响应MapWorker结束的任务
func (m *Master) ReplyForMapFinish(args *CallForMapFinishArgs, reply *CallForMapFinishReplyArgs) error {


	m.MasterLock.Lock()

	MapId := args.MapId //完成任务的MapID号

	m.AllMapTask[MapId].status = FINISHED //已完成

	for _,fileName := range args.FileNames{

		//提取Id号
		idInFileNameIndex := len(fileName) - 1 //文件名中的ReduceId号
		tmpReduceId, _ := strconv.Atoi(string(fileName[idInFileNameIndex]))

		//准备Reduce任务的输入文件
		m.AllReduceTask[tmpReduceId].ReduceTaskFiles = append(m.AllReduceTask[tmpReduceId].ReduceTaskFiles,fileName)

	}


	m.MasterLock.Unlock()

	reply.Ok = true
	return nil
}

//请求是否Map任务都结束了
func (m *Master) ReplyForAllMapFinish(args *CallForAllMapFinishArgs, reply *CallForAllMapFinishReplyArgs) error {

	reply.IsFinished = true
	reply.Ok = true

	m.MasterLock.Lock()

	finishedFlag := true
	for _, mapTask := range m.AllMapTask {

		if mapTask.status != FINISHED {
			reply.IsFinished = false //有任务未结束,需要等待

			//m.AllMapTask[i].status = NOTALLOCATED //需要重新分配
			finishedFlag = false
			break
		}
	}

	m.AllMapTasksFinished = finishedFlag		//更新所有Map任务完成标记

	m.MasterLock.Unlock()



	return nil

}

//响应Reduce申请任务请求
func (m *Master) ReplyForReduceTask(args *CallForReduceTaskArgs, reply *CallForReduceTaskReplyArgs) error {

	reply.Ok = false //默认分配失败
	reply.ReduceId = -1

	m.MasterLock.Lock()

	//分配Reduce编号
	for reduceId, tmpReduceTask := range m.AllReduceTask {

		if tmpReduceTask.status == NOTALLOCATED { //未分配

			m.AllReduceTask[reduceId].status = RUNNING //已分配,运行中
			m.AllReduceTask[reduceId].UsedTime = 0		//初始化使用时间

			//返回的Id和文件名
			reply.ReduceFileNames = m.AllReduceTask[reduceId].ReduceTaskFiles
			reply.ReduceId = reduceId
			reply.Ok = true
			break
		}
	}

	m.MasterLock.Unlock()

	return nil
}

//标记reduceId的任务完成
func (m *Master) ReplyForReduceFinish(args *CallForReduceFinishArgs, reply *CallForReduceFinishReplyArgs) error {

	reply.OK = true
	reduceId := args.ReduceId

	m.MasterLock.Lock()

	m.AllReduceTask[reduceId].status = FINISHED //此编号为ReduceId的任务完成

	m.MasterLock.Unlock()

	return nil

}

//回复是否所有的Reduce任务都已经完成
func (m *Master) ReplyForAllReduceFinish(args *CallForAllReduceFinishArgs, reply *CallForAllReduceFinishReplyArgs) error {

	reply.IsFinished = true
	reply.Ok = true

	m.MasterLock.Lock()
	finishedFlag := true
	for _, tmpReduceTask := range m.AllReduceTask {

		if tmpReduceTask.status != FINISHED {

			finishedFlag = false		//未完成标记
			reply.IsFinished = false

			break
		}
	}

	m.ALlReduceTasksFinished = finishedFlag		//更新所有的reduce任务完成标记

	m.MasterLock.Unlock()

	return nil
}

//不使用参数,无返回值,只是标记所有任务均已完成
func (m *Master) ReplyForAllTaskFinish(args *CallForAllTaskFinishArgs, reply *CallForAllTaskFinishReplyArgs) error {

	m.MasterLock.Lock()
	m.AllTaskFinished = true //标记所有任务均已完成
	m.MasterLock.Unlock()

	return nil
}

//
// start a thread that listens for RPCs from worker.go
//
func (m *Master) server() {
	rpc.Register(m)
	rpc.HandleHTTP()
	//l, e := net.Listen("tcp", ":1234")
	sockname := masterSock()
	os.Remove(sockname)
	l, e := net.Listen("unix", sockname)
	if e != nil {
		log.Fatal("listen error:", e)
	}
	go http.Serve(l, nil)
}

//给运行中的任务计时
func (m *Master) TimeTick()  {

	m.MasterLock.Lock()
	defer m.MasterLock.Unlock()

	if m.AllMapTasksFinished == false{

		//计时Map任务
		for i,tmpMapTask := range m.AllMapTask {
			if tmpMapTask.status == RUNNING {
				m.AllMapTask[i].UsedTime++		//所用时间更新

				if m.AllMapTask[i].UsedTime > 10 {		//如果运行超过10s
					m.AllMapTask[i].status = NOTALLOCATED		//更新状态为未分配
				}
			}

		}
	} else {
		//计时Reduce任务
		for i,tmpReduceTask := range m.AllReduceTask{
			if tmpReduceTask.status == RUNNING {
				m.AllReduceTask[i].UsedTime++

				if m.AllReduceTask[i].UsedTime > 10 {

					m.AllReduceTask[i].status = NOTALLOCATED
				}
			}
		}
	}
}

worker部分

//map任务:或许应该把所有的中间数据暂时放在内存中,完全结束之后再放在文件系统中。
func MapWorker(mapf func(string, string) []KeyValue, fileName string, MapId int) error {

	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()

	kvRet := mapf(fileName, string(content)) //执行用户提供的Map函数,返回键值对

	////将kv写入文件,并通知master///
	ok, outFileNames := WriteToIntFile(kvRet, MapId) //写入到中间文件中去

	if !ok {
		fmt.Printf("mapTask %v failed.\n", MapId)
	} else {
		fmt.Printf("mapTask:%v finished.\n", MapId)
		CallForMapFinishReply(outFileNames, MapId) //返回给master写入的文件名信息
	}
	return nil

}

//写入到中间文件中去
func WriteToIntFile(mapInterData []KeyValue, mapId int) (bool, []string) {

	reduceNum := 10 //默认reducer的数量为10

	//初始化一个二维切片
	tmpInterData := make([][]KeyValue, reduceNum)
	for i := range tmpInterData {
		tmpInterData[i] = make([]KeyValue, reduceNum)
	}

	//将键值对写入到缓冲区中
	for _, keyVal := range mapInterData {
		reduceId := ihash(keyVal.Key) % reduceNum
		tmpInterData[reduceId] = append(tmpInterData[reduceId], keyVal) //添加到缓冲区中
	}

	var outFileNames []string

	for _, tmpKeyVals := range tmpInterData {

		if len(tmpKeyVals) > reduceNum {

			tmpKeyVals = tmpKeyVals[reduceNum:] //从第reduceNum开始为第一个元素
			reduceId := ihash(tmpKeyVals[0].Key) % reduceNum
			outFileName := "mr-" + strconv.Itoa(mapId) + "-" + strconv.Itoa(reduceId) //拼接输出的文件名 mr-X-Y
			outFile, _ := os.OpenFile(outFileName, os.O_WRONLY|os.O_CREATE|os.O_APPEND, 0777)
			enc := json.NewEncoder(outFile)

			//一次写入
			err := enc.Encode(tmpKeyVals) //写入到中间文件中
			if err != nil {
				fmt.Printf("write wrong!\n")

				return false, outFileNames
			}

			//fmt.Printf("reduceId: %v num: %v\n",reduceId,len(tmpKeyVals))
			outFile.Close()
			outFileNames = append(outFileNames, outFileName)
		}
	}

	return true, outFileNames
}

//reducer的任务详情
func ReduceWork(reducef func(string, []string) string, reduceId int, inputFileNames []string) bool {

	var kva []KeyValue //保存从文件中读入的kv值

	//读取所有reduceId匹配的文件
	for _, inputFileName := range inputFileNames {

		inputFile, err := os.OpenFile(inputFileName, os.O_RDONLY, 0777)

		if err != nil {
			fmt.Printf("open mr-*-%v file failed\n", reduceId)
		}

		//成批读入
		dec := json.NewDecoder(inputFile)

		for {
			var tmpKv []KeyValue
			if err := dec.Decode(&tmpKv); err != nil {
				break
			}
			kva = append(kva, tmpKv...)
		}

		inputFile.Close() //关闭输入文件

	}

	//排序,是key值按照从小到大的顺序
	sort.Sort(ByKey(kva))

	//reduce操作后输出到不同的oname文件中去
	oname := "mr-out-" + strconv.Itoa(reduceId)
	ofile, err := ioutil.TempFile("./", "mr-out-tmp*") //先写道tmp文件中

	if err != nil {
		fmt.Printf("open tmp file failed\n")
		return false
	}

	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)
		// this is the correct format for each line of Reduce output.

		fmt.Fprintf(ofile, "%v %v\n", kva[i].Key, output)

		i = j
	}

	ofile.Close()

	os.Rename(ofile.Name(), oname) //Reduce操作,完成之后修改名字

	//删除中间文件
	for _, inputFileName := range inputFileNames {
		os.Remove(inputFileName)
	}

	fmt.Printf("reduce task %v finished \n", reduceId)

	CallForReduceFinishReply(reduceId) //通知master,编号为reduceId的任务已完成

	return true
}


//
// main/mrworker.go calls this function.
//总体的worker调度函数
func Worker(mapf func(string, string) []KeyValue, reducef func(string, []string) string) {


	var sleepTime int = 5		//睡眠时间
	//执行map任务
	for {

		fileName, taskId, ok := CallForMapTask() //向master申请一个任务

		if ok { //返回成功代表,Map任务还未完全处理完毕
			MapWorker(mapf, fileName, taskId) //开启map进程,传入要操作的文件名
			time.Sleep(time.Second * time.Duration(sleepTime))

		} else {		//所有的map任务已分派,但是还未完成

			isAllMapTaskFinished := CallForAllMapFinishReply()
			if isAllMapTaskFinished { //如果全部Map任务都结束了,跳出循环继续执行Reduce任务,否则继续循环
				break
			}else{
				fmt.Printf("map:%v sleep for a while\n",taskId)
				time.Sleep(time.Second * time.Duration(sleepTime)) //任务分配完毕,休息一会儿等待任务执行完毕
			}
		}
	}

	fmt.Printf("all map tasks finished!\n")
	fmt.Printf("reduce tasks begin...\n")
	//此处开始,可以执行reduce操作

	for {
		reduceId, inputFileNames, ok := CallForReduceTask() //向master请求Reduce任务

		if ok {		//返回成功代表reduce还未分配结束

			fmt.Printf("ReduceId:%v running....\n", reduceId)
			ReduceWork(reducef, reduceId, inputFileNames)
			time.Sleep(time.Second * time.Duration(sleepTime))

		} else {		//返回失败,代表所有的任务都已经分配结束,但是还未完成
			isFinished := CallForAllReduceFinishReply() //判断是否全部Reduce任务结束
			if isFinished {
				break
			}else{
				fmt.Printf("reduce:%v sleep for a while\n",reduceId)
				time.Sleep(time.Second * time.Duration(sleepTime)) //任务分配完毕,休息一会儿等待任务执行完毕
			}
		}
	}
	fmt.Printf("all reduce tasks finished!\n")
	CallAllTaskFinished() //通知master,所有任务都已经完成

}

结果

6.824(2020春) Lab1:MapReduce

总结

稍微总结一点无关的内容,就是在做这个实验的时候,一开始一点头绪的没有,虽然基本了解paper的内容,但是,你懂的,感觉好庞大,好多内容,不知从哪开始。
后来,慢慢的,从实现一个小的功能开始,可以跑 word-count。到慢慢通过第一个测试,接着慢慢优化,最后通过所有的测试。
所以,我的感悟是对于一个重大的,庞大的,不知从何下手的任务,可以从小的地方开始,慢慢的,一点一点的,进而完成所有内容。