code: func (fs *FileSorter) Input(key []types.Datum, val []types.Datum, handle int64) error {
  	if fs.closed {
  		return errors.New("FileSorter has been closed")
  	}
  	if fs.fetched {
  		return errors.New("call input after output")
  	}
  
  	assigned := false
  	abortTime := time.Duration(1) * time.Minute           // 1 minute
  	cooldownTime := time.Duration(100) * time.Millisecond // 100 milliseconds
  	row := &comparableRow{
  		key:    key,
  		val:    val,
  		handle: handle,
  	}
  
  	origin := time.Now()
  	// assign input row to some worker in a round-robin way
  	for {
  		for i := 0; i < fs.nWorkers; i++ {
  			wid := (fs.cWorker + i) % fs.nWorkers
  			if atomic.LoadInt32(&(fs.workers[wid].busy)) == 0 {
  				fs.workers[wid].input(row)
  				assigned = true
  				fs.cWorker = wid
  				break
  			}
  		}
  		if assigned {
  			break
  		}
  
  		// all workers are busy now, cooldown and retry
  		time.Sleep(cooldownTime)
  
  		if time.Since(origin) >= abortTime {
  			// weird: all workers are busy for at least 1 min
  			// choose to abort for safety
  			return errors.New("can not make progress since all workers are busy")
  		}
  	}
  	return nil
  }
docs: // Input adds one row into FileSorter.
  // Caller should not call Input after calling Output.