Method distributeReduceTask

master/master.go:240–300

()

}

func (ms *Master) distributeReduceTask() {
	log.Trace("[Master] Start Reduce task")

	taskStates := sync.Map{}

	finish := make(chan string, 100)
	workerID := 0
	workers, nWorkers := ms.availableWorkers(ms.numReducer)
	for _, reduceTask := range ms.ReduceTasks {
		reduceTask.SetState(TASK_INPROGRESS)
		go func(task ReduceTaskInfo, id int) {
			taskStates.Store(workers[id].UUID, task)
			workers[id].SetState(WORKER_BUSY)
			done := Reduce(workers[id].IP, task.toRPC())
			if !done {
				task.SetState(TASK_IDLE)
				workers[id].SetState(WORKER_UNKNOWN)
				ms.crashChan <- workers[id].UUID
			} else {
				task.SetState(TASK_COMPLETED)
				workers[id].SetState(WORKER_IDLE)
				finish <- workers[id].UUID
			}
		}(reduceTask, workerID)
		workerID = (workerID + 1) % nWorkers
	}

	count := 0
LOOP:
	for {
		select {
		case <-finish:
			count += 1
			if count == len(ms.ReduceTasks) {
				close(finish)
				break LOOP
			}
		case crashUUID := <-ms.crashChan:
			workers, _ = ms.availableWorkers(1)

			log.Info("[Master] Re-execute Map Task from ", crashUUID, "To ", workers[0].UUID)
			reExecuteTask, _ := taskStates.Load(crashUUID)
			go func(task ReduceTaskInfo, id int) {
				taskStates.Delete(crashUUID)
				taskStates.Store(workers[id].UUID, task)
				done := Reduce(workers[id].IP, task.toRPC())
				if !done {
					task.SetState(TASK_IDLE)
					workers[id].SetState(WORKER_UNKNOWN)
					ms.crashChan <- workers[id].UUID
				} else {
					task.SetState(TASK_COMPLETED)
					workers[id].SetState(WORKER_IDLE)
					finish <- workers[id].UUID
				}
			}(reExecuteTask.(ReduceTaskInfo), 0)
		}
	}