连连跨境电商网站开发,网页升级中永久跳转,宁波网站建,wordpress可以接广告吗前置知识
MapReduce#xff1a;Master 将一个 Map 任务或 Reduce 任务分配给一个空闲的 worker。 Map阶段#xff1a;被分配了 map 任务的 worker 程序读取相关的输入数据片段#xff0c;生成并输出中间 k/v 对#xff0c;并缓存在内存中。 Reduce阶段#xff1a;所有 ma…前置知识
MapReduceMaster 将一个 Map 任务或 Reduce 任务分配给一个空闲的 worker。 Map阶段被分配了 map 任务的 worker 程序读取相关的输入数据片段生成并输出中间 k/v 对并缓存在内存中。 Reduce阶段所有 map 任务结束reduce 程序使用 RPC 从 map worker 所在主机的磁盘上读取缓存数据通过对 key 进行排序后使得具有相同 key 值的数据聚合在一起reduce 进行操作后输出为文件。
实验内容
实现一个分布式 MapReduce由两个程序coordinator 和 worker组成。只有一个 coordinator 和一个或多个并行执行的 worker 。在真实系统中 worker 会运行在多台不同的机器上但在本 lab 中将在一台机器上运行所有 worker 。 worker 将通过 RPC 与 coordinator 通话。每个 worker 进程都会向 coordinator 请求 task从一个或多个文件中读取 task 输入执行 task并将 task 输出写入一个或多个文件。 coordinator 应该注意到如果某个 worker 在合理的时间内本 lab 使用 10 秒没有完成任务就会将相同的 task 交给另一个 worker。 rpc举例https://pdos.csail.mit.edu/6.824/notes/kv.go lab内容https://pdos.csail.mit.edu/6.824/labs/lab-mr.html Implmr/coordinator.go、mr/worker.go、mr/rpc.go 总结一下 对于 Coordiantor
map 任务初始化rpc handler回应worker分配任务请求、回应worker任务完成通知自身状态控制处理 map/reduce 阶段还是已经全部完成任务超时重新分配
对于 Worker
给 coordinator 发送 rpc 请求分配任务给 coordinator 发送 rpc 通知任务完成自身状态控制准确来说是 coordinator 不需要 worker 工作时通知 worker 结束运行
具体code见https://github.com/BeGifted/MIT6.5840-2023
实验环境
OSWSL-Ubuntu-18.04 golanggo1.17.6 linux/amd64
概要设计
所设计的Coordinator、Task、rpc消息格式
type WorkerArgs struct {WorkerId intWorkerState int // init\done\failTask *Task
}type WorkerReply struct {WorkerId intWorkerState int // init\done\failTask *Task
}type Coordinator struct {// Your definitions here.MapTaskChan chan *TaskReduceTaskChan chan *TaskNumReduce int // reduce numNumMap int // map numNumDoneReduce int // reduce done numNumDoneMap int // map done numState int // map\reduce\donemu sync.MutexTimeout time.DurationMapTasks map[int]*TaskReduceTasks map[int]*Task
}type Task struct {TaskId intTaskType int // map\reduceTaskState int // int\run\doneNReduce int // nReduceStartTime time.TimeInput []string
}const (StateMap 0StateReduce 1StateDone 2
)const (TaskStateInit 0TaskStateRun 1TaskStateDone 2
)const (TaskTypeMap 0TaskTypeReduce 1
)const (WorkerStateInit 0WorkerStateDone 1WorkerStateFail 2
)TODOWorkerState 出现 fail 的原因主要是在文件无法打开或读取上如果是在处理 map 任务时出现 fail那只有可能是原文件丢失了如果是 reduce 任务时出现 fail表示中间文件丢失需要运行某个特定的 map 任务重新生成然后再重新开始该 reduce 任务。当然不实现这个也不会影响 test。
主要流程
创建 Coordinator
创建 Coordinator 并且初始化将需要处理的数据片段放入 MapTaskChan 信道。这里将单个文件视作一个数据片段进行处理也就是说有 len(files) 个 map 任务。
func MakeCoordinator(files []string, nReduce int) *Coordinator {c : Coordinator{}// Your code here.c.NumMap len(files)c.NumReduce nReducec.MapTaskChan make(chan *Task, len(files))c.ReduceTaskChan make(chan *Task, nReduce)c.MapTasks make(map[int]*Task)c.ReduceTasks make(map[int]*Task)c.NumDoneMap 0c.NumDoneReduce 0c.State StateMapc.Timeout time.Duration(time.Second * 10)for i, file : range files {input : []string{file}task : Task{TaskId: i,TaskType: TaskTypeMap,TaskState: TaskStateInit,Input: input,NReduce: nReduce,StartTime: time.Now(),}c.MapTaskChan - taskc.MapTasks[i] task}c.server()return c
}运行 Worker
Worker 主要处理两类任务map 和 reduce。这两类任务通过 rpc 与 Coordinator 通信获取。 map 任务处理
if task.TaskType TaskTypeMap {filename : task.Input[0]intermediate : []KeyValue{}file, err : os.Open(filename)if err ! nil {log.Fatalf(cannot open %v, filename)continue}content, err : ioutil.ReadAll(file)if err ! nil {log.Fatalf(cannot read %v, filename)continue}file.Close()// log.Println(mapf)// log.Println(task.TaskId)kva : mapf(filename, string(content))intermediate append(intermediate, kva...)// sort.Sort(ByKey(intermediate))ReduceSplit : make(map[int][]KeyValue)for _, kv : range intermediate {ReduceSplit[ihash(kv.Key)%task.NReduce] append(ReduceSplit[ihash(kv.Key)%task.NReduce], kv)}for i : 0; i task.NReduce; i {oname : fmt.Sprintf(mr-%d-%d.tmp, task.TaskId, i)ofile, _ : os.Create(oname)enc : json.NewEncoder(ofile)for _, kv : range ReduceSplit[i] {err : enc.Encode(kv)if err ! nil {log.Fatalf(cannot encode %v, kv)break}}ofile.Close()}// Task Doneargs.Task taskTaskDone(args)
}reduce 任务处理
if task.TaskType TaskTypeReduce {var kva ByKeyfor _, filename : range task.Input {file, err : os.Open(filename)if err ! nil {log.Fatalf(cannot open %v, filename)file.Close()continue}dec : json.NewDecoder(file)for {var kv KeyValueif err : dec.Decode(kv); err ! nil {break}kva append(kva, kv)}file.Close()}sort.Sort(kva)i : 0oname : fmt.Sprintf(mr-out-%d, task.TaskId)ofile, _ : os.Create(oname)for i len(kva) {j : i 1for 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)fmt.Fprintf(ofile, %v %v\n, kva[i].Key, output)i j}// Task Doneargs.Task taskTaskDone(args)
}WorkerHandler
给 worker 分配 map/reduce 任务取决于阶段任务是否全部完成当阶段任务全部完成coordinator 的状态也需要更新。这个过程全局加锁。 处理 map 阶段
if c.State StateMap {select {case reply.Task -c.MapTaskChan:reply.Task.StartTime time.Now()reply.Task.TaskState TaskStateRundefault:for _, mapTask : range c.MapTasks {if mapTask.TaskState TaskStateRun time.Since(mapTask.StartTime) c.Timeout {mapTask.StartTime time.Now()reply.Task mapTaskreturn nil}}}
}处理 reduce 阶段
if c.State StateReduce {select {case reply.Task -c.ReduceTaskChan:reply.Task.StartTime time.Now()reply.Task.TaskState TaskStateRundefault:for _, reduceTask : range c.ReduceTasks {if reduceTask.TaskState TaskStateRun time.Since(reduceTask.StartTime) c.Timeout {reduceTask.StartTime time.Now()reply.Task reduceTaskreturn nil}}}
}需要注意的是除了这两个阶段外还有 StateDone 阶段即 reduce 任务都执行完毕了coordinator 还没完全回收此时 worker 还在请求分配任务这时候就应该通知 worker 停止。
DoneHandler
coordinator 处理任务完成的通知。全程加锁。在这里更新 task/coordinator 状态。
func (c *Coordinator) DoneHandler(args *WorkerArgs, reply *WorkerReply) error {c.mu.Lock()defer c.mu.Unlock()task : args.Taskif task.TaskType TaskTypeMap {if task.TaskState TaskStateRun {task.TaskState TaskStateDonec.MapTasks[task.TaskId].TaskState TaskStateDonec.NumDoneMap}} else if task.TaskType TaskTypeReduce {if task.TaskState TaskStateRun {task.TaskState TaskStateDonec.ReduceTasks[task.TaskId].TaskState TaskStateDonec.NumDoneReduce}}if c.State StateMap {if c.NumDoneMap c.NumMap {c.State StateReducefor i : 0; i c.NumReduce; i {input : []string{}for j : 0; j c.NumMap; j {input append(input, fmt.Sprintf(mr-%d-%d.tmp, j, i))}task : Task{TaskId: i,TaskType: TaskTypeReduce,TaskState: TaskStateInit,NReduce: c.NumReduce,StartTime: time.Now(),Input: input,}c.ReduceTaskChan - taskc.ReduceTasks[i] task}}} else if c.State StateReduce {if c.NumDoneReduce c.NumReduce {c.State StateDone}}return nil
}实验结果
bash test-mr-many.sh 10
