package mapreduce

import (

   "fmt"

   "sync"

   "strings"

   "time"

)

//

// schedule() starts and waits for all tasks in the given phase (mapPhase

// or reducePhase). the mapFiles argument holds the names of the files that

// are the inputs to the map phase, one per map task. nReduce is the

// number of reduce tasks. the registerChan argument yields a stream

// of registered workers; each item is the worker's RPC address,

// suitable for passing to call(). registerChan will yield all

// existing registered workers (if any) and new ones as they register.

//

func schedule(jobName string, mapFiles []string, nReduce int, phase jobPhase, registerChan chan string) {

   var ntasks int

   var n_other int // number of inputs (for reduce) or outputs (for map)

   switch phase {

   case mapPhase:

      ntasks = len(mapFiles)

      n_other = nReduce

   case reducePhase:

      ntasks = nReduce

      n_other = len(mapFiles)

   }

   fmt.Printf("Schedule: %v %v tasks (%d I/Os)\n", ntasks, phase, n_other)

   // All ntasks tasks have to be scheduled on workers. Once all tasks

   // have completed successfully, schedule() should return.

   //

   // Your code here (Part III, Part IV).

   //

   usableWorkers := make(chan string)

   waitGroup := new(sync.WaitGroup)

   i :=

   for  {

      var work string;

      select {

         case work  =<- usableWorkers:

            //fmt.Printf("usable worker: %v \n", work)

         case work =<- registerChan:

            //fmt.Printf("usable worker: %v \n", work)

      //default:

         //fmt.Printf("not usable worker wait for next loop ...\n")

      }

      if  strings.TrimSpace(work) != "" {

         waitGroup.Add()

         go func(index int) {

            rqArs := new(DoTaskArgs)

            rqArs.JobName=jobName

            if  phase == mapPhase {

               rqArs.File  = mapFiles[index]    // only for map, the input file

            }

            rqArs.Phase=phase   // are we in mapPhase or reducePhase?

            rqArs.TaskNumber=index      // this task's index in the current phase

            // NumOtherPhase is the total number of tasks in other phase; mappers

            // need this to compute the number of output bins, and reducers needs

            // this to know how many input files to collect.

            rqArs.NumOtherPhase=n_other

                defer func(){

                   waitGroup.Done()

                   usableWorkers <- work

                   }()

            call(work, "Worker.DoTask",

               rqArs, nil)

         }(i)

         i++

      } else {

         time.After(time.Second)

      }

      if i == ntasks {

         break

      }

   }

   waitGroup.Wait()

   fmt.Printf("Schedule: %v done\n", phase)

}