code: private long dispatchBlockMoves() throws InterruptedException {
      long bytesLastMoved = bytesMoved.get();
  
      Future<?>[] futures = new Future<?>[plan.sources.size()];
      int i=0;
      for (Source source : plan.sources) {
        futures[i++] = dispatcherExecutor.submit(
                          source.new BlockMoveDispatcher(Util.now()));
      }
  
      // wait for all dispatcher threads to finish
      for (Future<?> future : futures) {
        try {
          future.get();
        } catch (ExecutionException e) {
          LOG.warn("Dispatcher thread failed", e.getCause());
        }
      }
  
      // wait for all block moving to be done
      waitForMoveCompletion();
  
      return bytesMoved.get()-bytesLastMoved;
    }
docs: /* Start a thread to dispatch block moves for each source.
  The thread selects blocks to move & sends request to proxy source to
  initiate block move. The process is flow controlled. Block selection is
  blocked if there are too many un-confirmed block moves.
  Return the total number of bytes successfully moved in this iteration.