0.前言

mit 6.824分布式系统
课程主页

lab1是第一次作业,本菜鸡用了好几天独立完成,经过一次改版优化了数据结构和解决任务元数据并发环境下的data race问题,建议大家做之前有自己独立的思考,有很多可行方案都能完成任务。比如看到有的小伙伴采用master(coordinator)轮询slave(worker)进行交互,我是用slave定时发送请求触发master懒执行大部分任务(后面会聊到原因)。也有的小伙伴用队列增删加锁实现并发安全,本人用的golang自带的channel作为任务队列。不得不感叹人家本科生就有机会学这么有意思的课程,听说lab2更酸爽,后面会接着去冲塔。总之,集中一段时间做这个lab1挺有趣的。

我的代码在这里 :送餐员小李Gitee

1.实验准备

  • MapReduce论文阅读

  • Golang入门:需要掌握 基本语法/IDE配置/数组/面向对象/管道协程,小案例都实现一遍即可,十几个小时够了

    Golang核心-韩顺平

  • 阅读 lab1 note lab1
    看下mrsequential.go的流程,全程一个worker干到底无coordinator,看下map,reduce函数的加载和数据的流向

2 goland配置调试环境

  • windows

windows不支持动态插件编译,硬塞一个linux编译好的so文件也不行,当然可以通过改代码的形式把map reduce 的wordcount函数硬编码进main/worker.go,但不建议这样做后面会很麻烦。

buildmode=plugin not supported on windows/amd64

VMware搞个Ubuntu,然后golang,goland配置好

Ubuntu 16 镜像下载
VMware
VMware如何安装Ubuntu

  • linux

GOPATH选择git clone下来的文件夹6.824

GOROOT是golang的安装位置

wc.so运行前每次都要重新build,因此在写个build-wc.sh脚本放在main路径下面 configuration取名"build wc"

也可以在脚本加一个去掉"mr-"开头文件的句子,把后面频繁调试mr程序生成的结果文件去掉,省的每次运行前都去手删

go build -buildmode=plugin ../mrapps/wc.go
# rm -f mr-*

在这里插入图片描述

mrsequential.go 最下面的Before launch点击加号Run Another Configuration添加"build wc",这样每次运行之前都build一下wc.so

在这里插入图片描述

  • 至此mrsequential.go 应该可以在goland运行起来了

在这里插入图片描述

同理mrworker.go也需要这样配置

在这里插入图片描述

mrcoordinator.go 需要将所有mian下面的txt文本加入进去

pg-being_ernest.txt pg-dorian_gray.txt pg-frankenstein.txt pg-grimm.txt pg-huckleberry_finn.txt pg-metamorphosis.txt pg-sherlock_holmes.txt pg-tom_sawyer.txt

2.任务梳理

lab1 note 里面的内容,内容梳理,一开始比较模糊也没关系,做到哪里回看就行

2.1 描述

实现分布式mr,一个coordinator,一个worker(启动多个),在这次实验都在一个机器上运行。worker通过rpc和coordinator交互。worker请求任务,进行运算,写出结果到文件。coordinator需要关心worker的任务是否完成,在超时情况下将任务重新分配给别的worker。

2.2 规则

  • map阶段需要将中间keys分成nReduce个数, nReduce通过main/mrcoordinator.go传给MakeCoordinator()

  • worker需要将第X个reduce task结果放到mr-out-X中。

  • mr-out-X要一行一行生成,kv形式。main/mrsequential.go中有,拿来就完事了

  • main/mrcoordinator.go从mr/coordinator.go 的 Done()方法得知任务完成并关闭自己。

  • 任务都完成后,worker也得关闭

2.3 提示

  • 一开始可以从mr/worker.go的 Worker()方法做,发送rpc给coordinator请求任务,然后coordinator分配任务,然后worker读文件并且map函数处理。

  • map reduce函数都是通过go插件装载 (.so文件)

  • mr/ 文件变了就需要重新build

  • 都在一个文件系统,worker天然实现文件共享,先凑合着起步

  • 中间文件命名 mr-X-Y X是map任务号,y是reduce任务号

  • worker的map方法用json存储中间kv对,reduce再读回来,因为真正分布式worker都不在一个机器上,涉及网络传输,所以用json编码解码走个过场。

  • worker的map可以用 worker.go里面的ihash(key)得到特定key的reduce任务号

  • mrsequential.go 代码可以借鉴

  • coordinator里面的共享数据需要加锁

  • worker有时候需要等待,比如当map任务都分发出去了,有的worker完成后又来申请任务,此时还有map未完成,reduce不能开始,这个worker需要等待下

  • 如果任务重试机制,记得不要生成重复任务

  • mrapps/crash.go 随机干掉map reduce,看crash.go的代码是有一定几率让worker直接退出或者长时间延迟,可以用来测试恢复功能。这个逻辑是整合在map reduce函数里面的,注意worker被干掉时候任务已经拿到手了。

  • 确保没有人在出现崩溃时观察部分写入的文件,用ioutil.TempFile创建临时文件,用os.Rename重命名

3.开始做任务

每一步提示思路,是我的心路历程,应该符合大多数人的逻辑

3.1 小目标:coordinator分配好任务worker取到后打印出来

首先目标锁定在coordinator和worker的互动,其他的都不需要考虑,worker打印出来自己要做任务的文件名

  • mrsequential.go 里面逻辑看懂

  • 看懂coordinator和worker的rpc交互流程

    call 函数通过1234端口传入args和reply的内存地址,调用rpcname(Coordinator.函数名),通过反射机制"远程"调用Coordinator的该函数,Coordinator通过内存地址读取入参写出结果。worker.go里面的Worker方法调用CallExample,先运行Coordinator,再运行worker,看看worker端打印返回来的经过Coordinator加工过的数字

  • 至此worker和coordinator可以互动了

  • 发放给worker任务的结构体

    Job类型(枚举常量):JobType,以便worker知道这是Map任务还是Reduce任务。

    InputFile 文件名数组 :map情况数组里面就一个分配给worker的文件(Hadoop里面大文件是需要切块的,但是这里面的文件都很小就不切了直接一个文件给一个worker),reduce情况下是worker需要选取的一些需要聚合到一起的中间文件

    JobId/ReducerNum worker需要知道这些以便生成中间结果文件"mr-tmp-x-y"
    x是jobid,y是经过hash后的reduce id, y用来标识哪些文件汇入同一个reduce

type Job struct {
   JobType    JobType
   InputFile  []string
   JobId      int
   ReducerNum int
   //TmpFileList []string
}
  • 任务存放

放channel里面就行,省得自己写队列的各种方法,还能天然并发安全

channel 是golang特有的类型化消息的队列,可以通过它们发送类型化的数据在协程之间通信,可以避开所有内存共享导致的坑;通道的通信方式保证了同步性。数据通过通道:同一时间只有一个协程可以访问数据:所以不会出现数据竞争,设计如此。数据的归属(可以读写数据的能力)被传递。

因此Coordinator定义了两个channel来存放做好的map和reduce,jobMetaHolder(元数据管理相关)和CoordinatorCondition(coordinator状态),暂时不用管可以注释掉

type Coordinator struct {
   // channel which hold uncompleted task
   JobChannelMap        chan *Job
   JobChannelReduce     chan *Job
   ReducerNum           int
   MapNum               int
   CoordinatorCondition Condition
   uniqueJobId          int
   jobMetaHolder        JobMetaHolder
}

  • map任务制作
    Coordinator制作map任务,在一开始程序运行的时候就执行
func (c *Coordinator) makeMapJobs(files []string) {
  for _, v := range files {
  	id := c.generateJobId()
  	//fmt.Println("making map job :", id)
  	job := Job{
  		JobType:    MapJob,
  		InputFile:  []string{v},
  		JobId:      id,
  		ReducerNum: c.ReducerNum,
  	}

  //这下面暂时不需要
  	jobMetaINfo := JobMetaInfo{
  		condition: JobWaiting,
  		JobPtr:    &job,
  	}
  	c.jobMetaHolder.putJob(&jobMetaINfo)
  	fmt.Println("making map job :", &job)
  	c.JobChannelMap <- &job
  }
  //这上面暂时不需要
  fmt.Println("done making map jobs")
  c.jobMetaHolder.checkJobDone()
}

照着把那个样例rpc交互函数写个distribute方法,把的coordinate端和worker端入参数据类型分别改下,加入这句话即可将JobChannelMap里面的一个job给reply

 *reply = *<-c.JobChannelMap

worker端对reply取InputFile第一个元素,打印结果如下,worker线程取到任务了

worker get job which is pg-being_ernest.txt
  • 3.1小目标完成

3.2 worker通过传过来的文件名做map任务,写出结果

这个步骤简单粗暴,照着mrsequential.go里面写一下,记得用ihash处理下key分成Nreduce份用json编码后写出到"mr-tmp-x-y"文件。注意mr论文这步是有排序的,因为真正生产活动数据量是非常巨大的,map端提前排序好后,reduce的排序压力会减小很多。这里排不排序无所谓。

  • 3.2小目标完成

3.3 coordinator感知各个job任务运行完毕和map转reduce时机

  • 任务元数据管理

    从coordinator视角看任务分发

    制作任务 -> 放入队列 -> worker来取

    如果worker维护任务的状态显然不合理,每个任务的运行开始时间,任务状态。这些内容worker没必要知晓,是coordinator用来判断任务超时,或者map转reduce的。因此用JobMetaInfo把Job + Job状态包装下

    制作任务 -> 放入JobMetaInfo(元数据) -> 放入队列 -> worker来取

type JobMetaInfo struct {
condition JobCondition
StartTime time.Time
JobPtr *Job
}


 condition job状态: 包含等待,运行,完成
 StartTime : 开始运行的时间(等待变为运行)
 JobPtr  : job内存地址(golang是值传递,用地址更高效)

所有任务用map存储在JobMetaHolder,key是job的唯一id
```go
type JobMetaHolder struct {
  MetaMap map[int]*JobMetaInfo
}

针对这个JobMetaHolder的操作都可以安排上了,比如放入任务,任务状态更新,检查同一阶段任务是否完成

放入任务函数 putJob

 func (j *JobMetaHolder) putJob(JobInfo *JobMetaInfo) bool {
  jobId := JobInfo.JobPtr.JobId
  meta, _ := j.MetaMap[jobId]
  if meta != nil {
  	fmt.Println("meta contains job which id = ", jobId)
  	return false
  } else {
  	j.MetaMap[jobId] = JobInfo
  }
  return true
}

任务发射方法,当channel给出任务后,元数据管理器对任务元数据进行状态变更和运行开始时间记录(后面超时任务有用)。和Kafka的InflightMessage有点像。

func (j *JobMetaHolder) fireTheJob(jobId int) bool {
	ok, jobInfo := j.getJobMetaInfo(jobId)
	if !ok || jobInfo.condition != JobWaiting {
		return false
	}
	jobInfo.condition = JobWorking
	jobInfo.StartTime = time.Now()
	return true
}

检查当前阶段任务是否完成。因为每次制作jobs后,实在加锁情况下一股脑更新到元数据的,因此这边通过遍历先检查reduce完成未完成数量再检查map就能判断两种情况下的
完成情况。每次也会print 任务的数量信息方便调试

func (j *JobMetaHolder) checkJobDone() bool {
  reduceDoneNum := 0
  reduceUndoneNum := 0
  mapDoneNum := 0
  mapUndoneNum := 0
  for _, v := range j.MetaMap {
  	if v.JobPtr.JobType == MapJob {
  		if v.condition == JobDone {
  			mapDoneNum += 1
  		} else {
  			mapUndoneNum++
  		}
  	} else {
  		if v.condition == JobDone {
  			reduceDoneNum++
  		} else {
  			reduceUndoneNum++
  		}
  	}
  }
  fmt.Printf("%d/%d map jobs are done, %d/%d reduce job are done\n",
  	mapDoneNum, mapDoneNum+mapUndoneNum, reduceDoneNum, reduceDoneNum+reduceUndoneNum)

  return (reduceDoneNum > 0 && reduceUndoneNum == 0) || (mapDoneNum > 0 && mapUndoneNum == 0)
}
  • coordinator的状态转换
    每次当channel的长度为0的时候,去checkJobDone检查一下,当这个阶段所有任务都完成以后,进行状态转换并作出相应操作nextPhase()
func (c *Coordinator) nextPhase() {
	if c.CoordinatorCondition == MapPhase {
		c.makeReduceJobs()
		c.CoordinatorCondition = ReducePhase
	} else if c.CoordinatorCondition == ReducePhase {
		c.CoordinatorCondition = AllDone
	}
}
  • 3.3小目标完成

3.4 coordinator 的任务分配函数

  • 任务分配方法
    任务分配方法是coordinator最核心的函数,worker每次来询问都会调用这个方法
func (c *Coordinator) DistributeJob(args *ExampleArgs, reply *Job) error {
	mu.Lock()
	defer mu.Unlock()
	fmt.Println("coordinator get a request from worker :")
	if c.CoordinatorCondition == MapPhase {
		if len(c.JobChannelMap) > 0 {
			*reply = *<-c.JobChannelMap
			if !c.jobMetaHolder.fireTheJob(reply.JobId) {
				fmt.Printf("[duplicated job id]job %d is running\n", reply.JobId)
			}
		} else {
			reply.JobType = WaittingJob
			if c.jobMetaHolder.checkJobDone() {
				c.nextPhase()
			}
			return nil
		}
	} else if c.CoordinatorCondition == ReducePhase {
		if len(c.JobChannelReduce) > 0 {
			*reply = *<-c.JobChannelReduce
			if !c.jobMetaHolder.fireTheJob(reply.JobId) {
				fmt.Printf("job %d is running\n", reply.JobId)
			}
		} else {
			reply.JobType = WaittingJob
			if c.jobMetaHolder.checkJobDone() {
				c.nextPhase()
			}
			return nil
		}
	} else {
		reply.JobType = KillJob
	}
	return nil
}
  • worker完成任务调用的操作
    这里使用了判断元数据的办法看看任务是否会重复完成,在后面的崩溃测试下,worker会失效,coordinator需要重新发放Jobg给另一个worker,加入刚才那个失效的worker恢复了然后写入回来,不能覆盖已经完成的数据(甚至都下一个阶段了)。

    func (c *Coordinator) JobIsDone(args *Job, reply *ExampleReply) error {
      mu.Lock()
      defer mu.Unlock()
      switch args.JobType {
      case MapJob:
      	ok, meta := c.jobMetaHolder.getJobMetaInfo(args.JobId)
      	//prevent a duplicated work which returned from another worker
      	if ok && meta.condition == JobWorking {
      		meta.condition = JobDone
      		fmt.Printf("Map task on %d complete\n", args.JobId)
      	} else {
      		fmt.Println("[duplicated] job done", args.JobId)
      	}
      	break
      case ReduceJob:
      	fmt.Printf("Reduce task on %d complete\n", args.JobId)
      	ok, meta := c.jobMetaHolder.getJobMetaInfo(args.JobId)
      	//prevent a duplicated work which returned from another worker
      	if ok && meta.condition == JobWorking {
      		meta.condition = JobDone
      	} else {
      		fmt.Println("[duplicated] job done", args.JobId)
      	}
      	break
      default:
      	panic("wrong job done")
      }
      return nil
    

``

3.5 加锁

由于我用了channel所以在任务分配队列实现了天然并发安全,但是在别的地方还是遇到了问题,比如Done函数通过mrcoordinator主线程去时不时读取coordinator的状态来判断是否结束死循环。还有在一个worker调coordinator拉取数据的时候,另一个worker调coordinator的checkJobDone()函数进行检查。因此在响应可能发生冲突的地方加锁。

3.6 crash test

这里我是又起了一个线程去检查JobMetaHolder里面超时的任务,具体可以参考代码
但是无法通过test-mr.sh

去看了crash.go和test-mr.sh里面的逻辑,发现起的worker线程太少了,crash.go里面的maybecrash方法很有可能瘫痪这个worker,只起一个甚至三个都会导致最后没有可用worker,即使将任务重新放到channel里面

因此我修改了test-mr.sh,起了更多的worker,通过了测试。

Logo

为开发者提供学习成长、分享交流、生态实践、资源工具等服务,帮助开发者快速成长。

更多推荐