Method distributeMapTask

master/master.go:172–238

()

}

func (ms *Master) distributeMapTask() {
	log.Trace("[Master] Start Map task")

	taskStates := sync.Map{}

	finish := make(chan string, 100)
	// crashChan := make(chan MapTaskInfo, 100)
	workerID := 0
	workers, nWorkers := ms.availableWorkers(ms.totalWorkers)
	for _, mapTask := range ms.MapTasks {
		mapTask.setState(TASK_INPROGRESS)
		go func(task MapTaskInfo, id int) {
			taskStates.Store(workers[id].UUID, task)
			done := Map(workers[id].IP, task.toRPC())
			if !done {
				// task.setState(TASK_IDLE)
				workers[id].SetState(WORKER_UNKNOWN)
				ms.crashChan <- workers[id].UUID
				// crashChan <- task
			} else {
				// task.setState(TASK_COMPLETED)
				workers[id].SetState(WORKER_IDLE)
				finish <- workers[id].UUID
			}
		}(mapTask, workerID)
		workerID = (workerID + 1) % nWorkers
	}

	count := 0
LOOP:
	for {
		select {
		case <-finish:
			count += 1
			if count == len(ms.MapTasks) {
				close(finish)
				break LOOP
			}
		case crashUUID := <-ms.crashChan:
			// ms.waitForIDLEWorkers()
			workers, _ := ms.availableWorkers(1)
			log.Info("[Master] Re-execute Map Task from ", crashUUID, " To ", workers[0].UUID)
			if crashUUID == workers[0].UUID {
				log.Panic("Self loop")
			}
			reExecuteTask, ok := taskStates.Load(crashUUID)
			if !ok {
				log.Panic("Load task states from crashUUID fail")
			}
			go func(task MapTaskInfo, id int) {
				// taskStates.Delete(crashUUID)
				taskStates.Store(workers[id].UUID, task)
				done := Map(workers[id].IP, task.toRPC())
				if !done {
					task.setState(TASK_IDLE)
					workers[id].SetState(WORKER_UNKNOWN)
					ms.crashChan <- workers[id].UUID
				} else {