ToB企服应用市场:ToB评测及商务社交产业平台

标题: mit6.824 lab1 （2022） [打印本页]
作者: 大连全瓷种植牙齿制作中心    时间: 2022-8-24 14:00
标题: mit6.824 lab1 （2022）
  1. <br>　lab1 要求按照论文实现一个mapReduce 框架<br>
复制代码
  1. 　lab1 ：https://pdos.csail.mit.edu/6.824/labs/lab-mr.html
复制代码
　　论文：https://zhuanlan.zhihu.com/p/122571315
　　在mrsequential.go文件中有个单机版mapReduce实现很简单建议阅读。
  1.  
复制代码
  1. [/code] 
  2.  
  3. 整体框架流程：
  4. 　　[img]https://img2022.cnblogs.com/blog/1729513/202207/1729513-20220708145109475-401531304.png[/img]
  5.  
  6.  
  7.  
  8.  
  9.  
  10.  Coordinator 是协调器，负责
  11. 　　　　　　　　　　① 给woker分发任务
  12. 　　　　　　　　　　② 合并由map任务执行产生的中间文件
  13. 　　　　　　　　　　③ 任务超时重新分配任务
  14.  
  15. woker 是工作器，负责
  16. 　　　　　　　　①循环申请map 或reduce任务
  17. 　　　
  18.  
  19. 先看woker：
  20. 　　worker 向 Coordinator 发送任务申请后，判断得到的是什么样类型的任务
  21. [code]//申请任务
  22. for {
  23.    args := Args{}
  24.    args.Signal = REQUEST_WORKER<br><br>   reply := RpcCall(args)
  25.    switch reply.STATUS {
  26.    case COORDINATOR_MAP:    //获得map任务
  27.    　　MapHandle(&reply,mapf)<br>
  28.    case COORDINATOR_REDUCE: //获得reduce任务
  29. 　　　　ReduceHandle(&reply,reducef)
复制代码
  1. 　　case COORDINATOR__MAP_END:  //没申请到任务重新获取<br>   　　continue<br><br>　　case END:                //结束<br>   　　return
复制代码
  1. 　　}　　
复制代码
　　
  1. Recude任务<br>　　处理方式和mrsequential.go中几乎是一样的不多说了。<br><br>map任务<br>　　会从Coordinator 获得文件名、任务id、Nreduce（中间文件个数）<br><br>
复制代码
 
  1. [/code][img]https://img2022.cnblogs.com/blog/1729513/202207/1729513-20220708151531201-347578160.png[/img]
  2.  
  3.  kva是通过mapf 对文件处理得到的数据。
  4. 我开启两个任务分发器，和Nreduce 个文件写入器，进行并发处理数据。将数据写入到Nreduce个中间文件中，分发依据为ihash函数。
  5. [code]kva := MapMachingFile(reply.FileName, mapf)
  6.         midFileName := "mr-out-" + reply.FileName
  7.         chanArray := make([]chan KeyValue, 10)
  8.         for i := 0; i < 10; i++ {
  9.                 chanArray[i] = make(chan KeyValue, 10)
  10.         }<br><br>        //开启reduceNumber个文件写入线程
  11.         var w sync.WaitGroup
  12.         var mapW sync.WaitGroup
  13.         w.Add(reply.Neduce)
  14.         mapW.Add(2)
  15.         for i := 0; i < 10; i++ {
  16.                 go GoMakeMidFile(midFileName+strconv.Itoa(i), chanArray[i], &w)
  17.         }<br><br>        // 开启分发线程，分发数据到文件写入线程
  18.         lenght := len(kva)
  19.         go MapDistributeMidData(chanArray, kva[:lenght/2], &mapW)
  20.         go MapDistributeMidData(chanArray, kva[lenght/2:], &mapW)<br><br>        //所有分发线程结束
  21.         mapW.Wait()
  22.         for cIndex := 0; cIndex < 10; cIndex++ {
  23.                 close(chanArray[cIndex])
  24.         }<br><br>        //所有文件写入线程结束
  25.         w.Wait()
复制代码
　　
 
worker结束剩下看Coordinator 。
 
  1. 1 type Coordinator struct {
  2. 2     filebit                    //数据分发记录
  3. 3     Nreduce       int
  4. 4     midFileMergeC chan int
  5. 5     Mergefiled             //已处理数据记录
  6. 6     monitorC      []chan int //监听每个worker是否按时完成
  7. 7     STATUS
  8. 8     RedeceS
  9. 9     *sync.Mutex
  10. 10     End            bool
  11. 11 }
复制代码
 
  1. Coordinator  结构记录的信息主要为三部分<br>　　　　　　　　2、3、4、5行记录map相关<br>　　　　　　　　6 为监听chan，监听任务是否超时<br>　　　　　　　　7位Coordinator 当前的状态，通过状态判断要分发map任务、reduce任务、结束<br><br><br><br>判断worker的目的，请求任务就分发任务处理，完成map任务就将所有map产生中间数据一一对应合并到Nreduce个文件中。
复制代码
  1. //信号处理
  2. func (c *Coordinator) SignalTask (args *Args, reply *Reply) error {
  3.     switch args.Signal {
  4.     case REQUEST_WORKER:
  5.         c.distributeTask(args,reply)<br><br>    //中间文件处理
  6.     case COMPLETE:
  7.         c.midFileMerge(args,reply)
  8.     }<br><br>    return nil
  9. }
复制代码
  1. [/code] 在初始化Coordinator时，还会打开一些线程。本线程会开启10个中间文件写入线程，当每个worker处理完map任务后，会将自己处理的map文件相关信息传给[i]Coordinator，[/i][i]Coordinator通过chan将数据发给每个文件合并线程[/i]StartMergeFile。
  2. [i]举个例子[/i]
  3. [i]workerMap A产生了  1，2，3 个中间文件[/i]
  4. [i]1号文件 合并到 [i]mr-out-m-1[/i][/i]
  5. [i][i][i]2号文件 合并到 [i]mr-out-m-2[/i][/i][/i][/i]
  6. [i]3号文件 合并到 [i]mr-out-m-3[/i][/i]
  7.  
  8. [i]workerMap  B 又产生1、2、3个中间文件[/i]
  9. [i]1号文件 合并到 [i]mr-out-m-1[/i][/i]
  10. [i][i][i]2号文件 合并到 [i]mr-out-m-2[/i][/i][/i][/i]
  11. [i]3号文件 合并到 [i]mr-out-m-3[/i][/i]
  12. [i][i] [/i][/i][code]<br><br>
复制代码
  1. //开启Nreduce个中间文件写入线程
  2. //返回文件写入chan 切片
  3. func (c *Coordinator)runFileWorker () []chan int {
  4.         cLi := make([]chan int,c.Nreduce)
  5.         for i := 0 ; i < c.Nreduce ; i ++ {
  6.                 cLi[i] = make(chan int,10)
  7.         }
  9.         for fid := 0 ; fid < c.Nreduce ; fid ++ {
  10.                 go c.StartMergeFile(fid,cLi[fid])
  11.         }
  12.         return cLi
  13. }
复制代码
　　
  1. <br>记录信息是否已处理的结构：<br>
复制代码
  1. 　　 filebit 、ReduceS 核心是通过一个简单的bitmap实现的
复制代码
  1. type filebit struct{
  2.         rw                 *sync.Mutex
  3.         bitMap
  4.         file []string
  5. }
  7. type RedeceS struct {
  8.         filebit
  9. }
复制代码
　　
[code][/code][code]type bitMap struct {        bit  int16        size int}//获取一个未使用位置func (b *bitMap) GetOne() int {        for i := int(0) ; i < b.size ; i ++ {                if b.isZero(i) {                        b.seTUsed(i)                        return i                }        }        //这里超过size限制会直接报错        return -1}//第i位是否为0//为0未使用func (b *bitMap) isZero (index int) bool {        return ((1



欢迎光临 ToB企服应用市场:ToB评测及商务社交产业平台 (https://dis.qidao123.com/) Powered by Discuz! X3.4