mr

Documentation

Index

• Constants
• func CallExample()
• func PingPong()
• func Register() uint64
• func Report(res *ResultReq)
• func Worker(mapf func(string, string) []KeyValue, reducef func(string, []string) string)
• type ByKey
  • func (a ByKey) Len() int
  • func (a ByKey) Less(i, j int) bool
  • func (a ByKey) Swap(i, j int)
• type Dispatcher
• type ExampleArgs
• type ExampleReply
• type GetTaskReq
• type GetTaskRes
• type JobState
• type KeyValue
• type Master
  • func MakeMaster(files []string, nReduce int) *Master
  • func (m *Master) Done() bool
  • func (m *Master) Example(args *ExampleArgs, reply *ExampleReply) error
  • func (m *Master) GetTaskWorker(args *GetTaskReq, reply *GetTaskRes) error
  • func (m *Master) PingPong(args *Ping, reply *Pong) error
  • func (m *Master) RegisterWorker(args *RegisterReq, reply *RegisterRes) error
  • func (m *Master) ReportResult(args *ResultReq, reply *ResultRes) error
• type Ping
• type Pong
• type ReduceSource
• type RegisterReq
• type RegisterRes
• type ResultReq
  • func ExecTask(mapf func(string, string) []KeyValue, reducef func(string, []string) string, ...) (*ResultReq, error)
• type ResultRes
• type Task
  • func GetTask() *Task
• type TaskConf
• type TaskPool
• type WorkerSession
  • func (w *WorkerSession) PingPong(ts time.Duration)

Constants

const (
	CallRegister = "Master.RegisterWorker"
	CallPingPong = "Master.PingPong"
	CallGetTask  = "Master.GetTaskWorker"
	CallReport   = "Master.ReportResult"
)

Functions

func CallExample

func CallExample()

func PingPong

func PingPong()

func Register

func Register() uint64

TODO: 是否可以包装成对象

func Report

func Report(res *ResultReq)

func Worker

func Worker(mapf func(string, string) []KeyValue,
	reducef func(string, []string) string)

Types

type ByKey

type ByKey []KeyValue

for sorting by key.

func (ByKey) Len

func (a ByKey) Len() int

for sorting by key.

func (ByKey) Less

func (a ByKey) Less(i, j int) bool

func (ByKey) Swap

func (a ByKey) Swap(i, j int)

type Dispatcher

type Dispatcher struct {
	TimeOut          time.Duration      //默认10秒
	M                *Master            //主节点全局结构
	ReduceSourceChan chan *ReduceSource // 发送 reduce 的任务 执行内容
	CleanWorkerChan  chan uint64        // 清理失效的worker
}

定时清理器

type ExampleArgs

type ExampleArgs struct {
	X int
}

type ExampleReply

type ExampleReply struct {
	Y int
}

type GetTaskReq

type GetTaskReq struct {
	WorkerID uint64
}

获取任务

type GetTaskRes

type GetTaskRes struct {
	Code int
	Msg  string
	T    *Task
}

type JobState

type JobState struct {
	MatrixSource [][]string // MC * RC
	MC           int
	RC           int
	MCDone       int32
	// contains filtered or unexported fields
}

Job 状态

type KeyValue

type KeyValue struct {
	Key   string
	Value string
}

Map functions return a slice of KeyValue.

type Master

type Master struct {
	// Your definitions here.
	S  *JobState
	TP *TaskPool
	W  *sync.Map
}

主节点

func MakeMaster

func MakeMaster(files []string, nReduce int) *Master

create a Master.

func (*Master) Done

func (m *Master) Done() bool

main/mrmaster.go calls Done() periodically to find out if the entire job has finished.

func (*Master) Example

func (m *Master) Example(args *ExampleArgs, reply *ExampleReply) error

an example RPC handler.

func (*Master) GetTaskWorker

func (m *Master) GetTaskWorker(args *GetTaskReq, reply *GetTaskRes) error

func (*Master) PingPong

func (m *Master) PingPong(args *Ping, reply *Pong) error

func (*Master) RegisterWorker

func (m *Master) RegisterWorker(args *RegisterReq, reply *RegisterRes) error

func (*Master) ReportResult

func (m *Master) ReportResult(args *ResultReq, reply *ResultRes) error

type Ping

type Ping struct {
	WorkerID uint64
}

健康检查

type Pong

type Pong struct {
	Code int
}

type ReduceSource

type ReduceSource struct {
	MIdx      int
	MapSource []string // map 任务返回的 source 列表
}

type RegisterReq

type RegisterReq struct {
}

Add your RPC definitions here. 注册

type RegisterRes

type RegisterRes struct {
	WorkerID uint64
}

type ResultReq

type ResultReq struct {
	WorkerID uint64
	Code     int // 0 代表 map 1 代表 reduce 2代表 失败
	Msg      string
	M        []string
}

返回结果

func ExecTask

func ExecTask(mapf func(string, string) []KeyValue,
	reducef func(string, []string) string, task *Task) (*ResultReq, error)

type ResultRes

type ResultRes struct {
	Code int
	Msg  string
}

type Task

type Task struct {
	Status int // 0 未完成 1工作中 2已完成
	Type   int // 0 map 任务 1 reduce 任务 2 shut down 3 retry
	Conf   *TaskConf
}

任务

func GetTask

func GetTask() *Task

type TaskConf

type TaskConf struct {
	Source []string // 兼容两种任务
	RNum   int      // 当前 map 任务的 任务编号 如果是reduce任务 则为-1
	MNum   int      // 当前 reduce 任务的 任务编号 如果是map任务 则为-1
	RC     int      // reduce 的任务数
}

任务配置

type TaskPool

type TaskPool struct {
	Pool chan *Task
}

任务池

type WorkerSession

type WorkerSession struct {
	WorkerID     uint64
	Status       int // 0 空闲状态 1 工作状态 2 无法正常工作
	T            *Task
	Mux          *sync.RWMutex
	LastPingTs   int64
	PingPongChan chan struct{}
}

工作者会话管理器

func (*WorkerSession) PingPong

func (w *WorkerSession) PingPong(ts time.Duration)