概述
MIT 6.824 Lab1
MapReduce
MapReduce是一种编程模型，用于大规模数据集(大于1TB)的并行运算。"Map(映射)“和"Reduce(归约)”，和它们的主要思想，都是从函数式编程语言里借来的，还有从矢量编程语言里借来的特性。它极大地方便了编程人员在不会分布式并行编程的情况下，将自己的程序运行在分布式系统上。 当前的软件实现是指定一个Map(映射)函数，用来把一组键值对映射成一组新的键值对，指定并发的Reduce(归约)函数，用来保证所有映射的键值对中的每一个共享相同的键组。
完成实验的所需准备
go语言基础知识(goroutine,slice,sync库,struct等)
MIT 6.824课程网站 6.824 Schedule: Spring 2020 (mit.edu)
Lab1 6.824 Lab 1: MapReduce (mit.edu)
阅读论文 MapReduce paper. http://research.google.com/archive/mapreduce-osdi04.pdf
实验环境
deepin系统
Vmware虚拟机
Goland
实验步骤
1. 阅读mrsequential.go源码
我们可以看到源码中所展现的是MapReduce的串行化实现，我们可以以此为基础去完成并行化代码，由于该实验不能在分布式的环境下进行，所以实验是开多个线程来模拟分布式情况下的一系列操作，并通过Rpc的方式进行通信
2. 目标
我们所需要完成的只是完成mr包下中的，master.go,worker.go,rpc.go这三个文件
3. 定义一系列的结构
a. task
type MapTask struct {
TaskStruct
Filename string
}
type ReduceTask struct {
TaskStruct
IntermediateFilenames []string
}
type TaskStruct struct {
State TaskState
StartTime time.Time
Id int
}
type Task struct {
Operation TaskOperation
IsMap bool
NReduce int
Map MapTask
Reduce ReduceTask
}
定义map任务以及reduce任务的结构体，map首先会对某个文件进行映射操作，然后对每个key进行哈希映射从而分配给不同的reduce worker线程进行归约操作，所以每个reduce worker会对多个不同的文件进行规约操作，所以在这里我们要在ReduceTask中定义一个字符串切片数组。
// Task状态枚举变量
type TaskState int
const (
// 等待分配的一个任务
WAITING TaskState = iota
// 进行操作中的一个任务
WORKING
// 已经完成操作的一个任务
FINISHED
)
b. master struct
// 表示master当前需要分配哪些任务,是否已经结束
type JobState int
const (
MAPPING JobState = iota
REDUCING
DONE
)
type Master struct {
State JobState
NReduce int
MapTasks []*MapTask
ReduceTasks []*ReduceTask
MapIdSet []int
MaxTaskId int
Mutex sync.Mutex //信号量，用来进行线程的互斥操作
MapGroup sync.WaitGroup
ReduceGroup sync.WaitGroup
}
其中sync.WaitGroup类似于Java中的CountDownLatch，在这边我用来检测同个操作的所有任务是否已经完成，完成后将线程放行，进行下一个操作。
（init->map->reduce->done）
4. 实现master.go
//
// create a Master.
// main/mrmaster.go calls this function.
// nReduce is the number of reduce tasks to use.
//
func MakeMaster(files []string, nReduce int) *Master {
m := Master{
NReduce: nReduce,
MaxTaskId: 0,
}
// map任务队列
for _, f := range files {
m.MapTasks = append(m.MapTasks, &MapTask{TaskStruct: TaskStruct{State: WAITING}, Filename: f})
m.MapGroup.Add(1)
}
// reduce任务队列
for i := 0; i < nReduce; i++ {
m.ReduceTasks = append(m.ReduceTasks, &ReduceTask{TaskStruct: TaskStruct{State: WAITING, Id: i}})
}
m.ReduceGroup.Add(nReduce)
m.State = MAPPING
m.server()
// 开启一个goroutine检测任务是否已经完成
go func() {
_ = m.checkFinal()
}()
// 开启一个goroutine检测任务是否超时了，如果超时了，将任务重置并重新分配
go func() {
_ = m.checkTime()
}()
return &m
}
其中checkFinal()和checkTime()方法如下
func (m *Master) checkFinal() error {
m.MapGroup.Wait()
m.State = REDUCING
m.ReduceGroup.Wait()
m.State = DONE
return nil
}
同上（其中sync.WaitGroup类似于Java中的CountDownLatch，在这边我用来检测同个操作的所有任务是否已经完成，完成后将线程放行，进行下一个操作。
（init->map->reduce->done））
const TIMEOUT = 10 * time.Second
func (m *Master) checkTime() error {
for {
if m.State == MAPPING {
for _, task := range m.MapTasks {
m.Mutex.Lock()
if task.State == WORKING && task.StartTime.Add(TIMEOUT).Before(time.Now()) {
task.State = WAITING
}
m.Mutex.Unlock()
}
} else if m.State == REDUCING {
for _, task := range m.ReduceTasks {
m.Mutex.Lock()
if task.State == WORKING && task.StartTime.Add(TIMEOUT).Before(time.Now()) {
task.State = WAITING
}
m.Mutex.Unlock()
}
}
if m.State == DONE {
return nil
}
}
}
不断for循环检测任务是否超时(因为在test中，它会模拟现实场景某些操作会莫名其妙的寄掉，将你的worker杀掉几个，所以你必须要对其判断你的worker是否寄了)
func (m *Master) RequestTask(_ *Placeholder, reply *Task) error {
reply.Operation = WAIT
if m.State == MAPPING {
for _, task := range m.MapTasks {
m.Mutex.Lock()
if task.State == WAITING {
task.StartTime = time.Now()
task.State = WORKING
m.MaxTaskId++
task.Id = m.MaxTaskId
m.Mutex.Unlock()
reply.Operation = RUN
reply.IsMap = true
reply.NReduce = m.NReduce
reply.Map = *task
return nil
}
m.Mutex.Unlock()
}
} else if m.State == REDUCING {
for _, task := range m.ReduceTasks {
m.Mutex.Lock()
if task.State == WAITING {
task.StartTime = time.Now()
task.State = WORKING
task.IntermediateFilenames = nil
for _, id := range m.MapIdSet {
task.IntermediateFilenames = append(task.IntermediateFilenames, fmt.Sprintf("mr-%d-%d", id, task.Id))
}
m.Mutex.Unlock()
reply.Operation = RUN
reply.IsMap = false
reply.NReduce = m.NReduce
reply.Reduce = *task
return nil
}
m.Mutex.Unlock()
}
}
return nil
}
分配任务的操作，将任务队列中还在等待的任务分配给worker。
func (m *Master) Finish(args *FinishArgs, _ *Placeholder) error {
if args.IsMap {
for _, task := range m.MapTasks {
if task.Id == args.Id {
task.State = FINISHED
log.Printf("finished task %d, total %d", task.Id, len(m.MapTasks))
m.MapIdSet = append(m.MapIdSet, task.Id)
m.MapGroup.Done()
break
}
}
} else {
for _, task := range m.ReduceTasks {
if task.Id == args.Id {
task.State = FINISHED
m.ReduceGroup.Done()
break
}
}
}
return nil
}
任务完成的操作，worker完成任务后会调用rpc通知master我这个任务已经完成了，这边会对任务标记完成，并且执行m.MapGroup.Done()，就是WaitGroup计数器减一
5. 实现worker.go
//
// main/mrworker.go calls this function.
//
func Worker(mapf MapF, reducef ReduceF) {
for {
time.Sleep(1 * time.Second)
task := Task{}
call("Master.RequestTask", &Placeholder{}, &task)
if task.Operation == WAIT {
continue
}
if task.IsMap {
log.Printf("received map job %s", task.Map.Filename)
err := handleMap(task, mapf)
if err != nil {
log.Fatalf(err.Error())
return
}
} else {
log.Printf("received reduce job %d %v", task.Reduce.Id, task.Reduce.IntermediateFilenames)
err := handleReduce(task, reducef)
if err != nil {
log.Fatalf(err.Error())
return
}
}
}
}
你可以理解为worker的构造函数，通知master给worker分配一个任务，并按照任务的类型去对应的执行handle操作
后续代码就和串行化的操作差不多了，这边不再叙述。
实验结果
PASSED ALL TESTS
实验总结
通过本次实验，让我对分布式系统有了初步的认知，同时让我的go语言编程能力得到了有效的提升
最后
以上就是聪慧帆布鞋为你收集整理的MIT 6.824 Lab1MIT 6.824 Lab1的全部内容，希望文章能够帮你解决MIT 6.824 Lab1MIT 6.824 Lab1所遇到的程序开发问题。
如果觉得靠谱客网站的内容还不错，欢迎将靠谱客网站推荐给程序员好友。
发表评论 取消回复