hadoop-hdfs-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From ji...@apache.org
Subject svn commit: r1616890 - in /hadoop/common/branches/branch-2/hadoop-hdfs-project/hadoop-hdfs: ./ src/main/java/org/apache/hadoop/hdfs/server/balancer/ src/test/java/org/apache/hadoop/hdfs/server/balancer/
Date Fri, 08 Aug 2014 21:35:45 GMT
Author: jing9
Date: Fri Aug  8 21:35:44 2014
New Revision: 1616890

URL: http://svn.apache.org/r1616890
Log:
HDFS-6828. Merge r1616889 from trunk.

Added:
    hadoop/common/branches/branch-2/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/balancer/Dispatcher.java
      - copied unchanged from r1616889, hadoop/common/trunk/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/balancer/Dispatcher.java
Modified:
    hadoop/common/branches/branch-2/hadoop-hdfs-project/hadoop-hdfs/CHANGES.txt
    hadoop/common/branches/branch-2/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/balancer/Balancer.java
    hadoop/common/branches/branch-2/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/balancer/NameNodeConnector.java
    hadoop/common/branches/branch-2/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/balancer/TestBalancer.java
    hadoop/common/branches/branch-2/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/balancer/TestBalancerWithHANameNodes.java
    hadoop/common/branches/branch-2/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/balancer/TestBalancerWithMultipleNameNodes.java
    hadoop/common/branches/branch-2/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/balancer/TestBalancerWithNodeGroup.java

Modified: hadoop/common/branches/branch-2/hadoop-hdfs-project/hadoop-hdfs/CHANGES.txt
URL: http://svn.apache.org/viewvc/hadoop/common/branches/branch-2/hadoop-hdfs-project/hadoop-hdfs/CHANGES.txt?rev=1616890&r1=1616889&r2=1616890&view=diff
==============================================================================
--- hadoop/common/branches/branch-2/hadoop-hdfs-project/hadoop-hdfs/CHANGES.txt (original)
+++ hadoop/common/branches/branch-2/hadoop-hdfs-project/hadoop-hdfs/CHANGES.txt Fri Aug  8 21:35:44 2014
@@ -126,6 +126,9 @@ Release 2.6.0 - UNRELEASED
 
     HDFS-573. Porting libhdfs to Windows. (cnauroth)
 
+    HDFS-6828. Separate block replica dispatching from Balancer. (szetszwo via
+    jing9)
+
   OPTIMIZATIONS
 
     HDFS-6690. Deduplicate xattr names in memory. (wang)

Modified: hadoop/common/branches/branch-2/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/balancer/Balancer.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/branch-2/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/balancer/Balancer.java?rev=1616890&r1=1616889&r2=1616890&view=diff
==============================================================================
--- hadoop/common/branches/branch-2/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/balancer/Balancer.java (original)
+++ hadoop/common/branches/branch-2/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/balancer/Balancer.java Fri Aug  8 21:35:44 2014
@@ -18,19 +18,9 @@
 package org.apache.hadoop.hdfs.server.balancer;
 
 import static com.google.common.base.Preconditions.checkArgument;
-import static org.apache.hadoop.fs.CommonConfigurationKeys.IPC_CLIENT_FALLBACK_TO_SIMPLE_AUTH_ALLOWED_DEFAULT;
-import static org.apache.hadoop.fs.CommonConfigurationKeys.IPC_CLIENT_FALLBACK_TO_SIMPLE_AUTH_ALLOWED_KEY;
-import static org.apache.hadoop.hdfs.protocolPB.PBHelper.vintPrefixed;
-
-import java.io.BufferedInputStream;
-import java.io.BufferedOutputStream;
-import java.io.DataInputStream;
-import java.io.DataOutputStream;
+
 import java.io.IOException;
-import java.io.InputStream;
-import java.io.OutputStream;
 import java.io.PrintStream;
-import java.net.Socket;
 import java.net.URI;
 import java.text.DateFormat;
 import java.util.ArrayList;
@@ -38,20 +28,11 @@ import java.util.Arrays;
 import java.util.Collection;
 import java.util.Collections;
 import java.util.Date;
-import java.util.EnumMap;
 import java.util.Formatter;
-import java.util.HashMap;
-import java.util.HashSet;
 import java.util.Iterator;
 import java.util.LinkedList;
 import java.util.List;
-import java.util.Map;
 import java.util.Set;
-import java.util.concurrent.ExecutionException;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
-import java.util.concurrent.Future;
-import java.util.concurrent.atomic.AtomicLong;
 
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
@@ -63,31 +44,15 @@ import org.apache.hadoop.hdfs.DFSConfigK
 import org.apache.hadoop.hdfs.DFSUtil;
 import org.apache.hadoop.hdfs.HdfsConfiguration;
 import org.apache.hadoop.hdfs.StorageType;
-import org.apache.hadoop.hdfs.protocol.Block;
-import org.apache.hadoop.hdfs.protocol.DatanodeInfo;
-import org.apache.hadoop.hdfs.protocol.ExtendedBlock;
-import org.apache.hadoop.hdfs.protocol.HdfsConstants;
-import org.apache.hadoop.hdfs.protocol.HdfsConstants.DatanodeReportType;
-import org.apache.hadoop.hdfs.protocol.datatransfer.IOStreamPair;
-import org.apache.hadoop.hdfs.protocol.datatransfer.Sender;
-import org.apache.hadoop.hdfs.protocol.datatransfer.TrustedChannelResolver;
-import org.apache.hadoop.hdfs.protocol.datatransfer.sasl.DataTransferSaslUtil;
-import org.apache.hadoop.hdfs.protocol.datatransfer.sasl.SaslDataTransferClient;
-import org.apache.hadoop.hdfs.protocol.proto.DataTransferProtos.BlockOpResponseProto;
-import org.apache.hadoop.hdfs.protocol.proto.DataTransferProtos.Status;
-import org.apache.hadoop.hdfs.security.token.block.BlockTokenIdentifier;
+import org.apache.hadoop.hdfs.server.balancer.Dispatcher.BalancerDatanode;
+import org.apache.hadoop.hdfs.server.balancer.Dispatcher.Source;
+import org.apache.hadoop.hdfs.server.balancer.Dispatcher.Task;
+import org.apache.hadoop.hdfs.server.balancer.Dispatcher.Util;
 import org.apache.hadoop.hdfs.server.blockmanagement.BlockPlacementPolicy;
 import org.apache.hadoop.hdfs.server.blockmanagement.BlockPlacementPolicyDefault;
-import org.apache.hadoop.hdfs.server.common.HdfsServerConstants;
 import org.apache.hadoop.hdfs.server.namenode.UnsupportedActionException;
-import org.apache.hadoop.hdfs.server.protocol.BlocksWithLocations.BlockWithLocations;
 import org.apache.hadoop.hdfs.server.protocol.DatanodeStorageReport;
 import org.apache.hadoop.hdfs.server.protocol.StorageReport;
-import org.apache.hadoop.io.IOUtils;
-import org.apache.hadoop.net.NetUtils;
-import org.apache.hadoop.net.NetworkTopology;
-import org.apache.hadoop.security.token.Token;
-import org.apache.hadoop.util.HostsFileReader;
 import org.apache.hadoop.util.StringUtils;
 import org.apache.hadoop.util.Time;
 import org.apache.hadoop.util.Tool;
@@ -201,15 +166,7 @@ public class Balancer {
 
   private static final long GB = 1L << 30; //1GB
   private static final long MAX_SIZE_TO_MOVE = 10*GB;
-  private static final long MAX_BLOCKS_SIZE_TO_FETCH = 2*GB;
 
-  /** The maximum number of concurrent blocks moves for 
-   * balancing purpose at a datanode
-   */
-  private static final int MAX_NO_PENDING_BLOCK_ITERATIONS = 5;
-  public static final long DELAY_AFTER_ERROR = 10 * 1000L; //10 seconds
-  public static final int BLOCK_MOVE_READ_TIMEOUT=20*60*1000; // 20 minutes
-  
   private static final String USAGE = "Usage: java "
       + Balancer.class.getSimpleName()
       + "\n\t[-policy <policy>]\tthe balancing policy: "
@@ -221,16 +178,9 @@ public class Balancer {
       + "\n\t[-include [-f <hosts-file> | comma-sperated list of hosts]]"
       + "\tIncludes only the specified datanodes.";
   
-  private final NameNodeConnector nnc;
-  private final KeyManager keyManager;
-
+  private final Dispatcher dispatcher;
   private final BalancingPolicy policy;
-  private final SaslDataTransferClient saslClient;
   private final double threshold;
-  // set of data nodes to be excluded from balancing operations.
-  Set<String> nodesToBeExcluded;
-  //Restrict balancing to the following nodes.
-  Set<String> nodesToBeIncluded;
   
   // all data node lists
   private final Collection<Source> overUtilized = new LinkedList<Source>();
@@ -239,634 +189,6 @@ public class Balancer {
       = new LinkedList<BalancerDatanode.StorageGroup>();
   private final Collection<BalancerDatanode.StorageGroup> underUtilized
       = new LinkedList<BalancerDatanode.StorageGroup>();
-  
-  private final Collection<Source> sources = new HashSet<Source>();
-  private final Collection<BalancerDatanode.StorageGroup> targets
-      = new HashSet<BalancerDatanode.StorageGroup>();
-  
-  private final Map<Block, BalancerBlock> globalBlockList
-                 = new HashMap<Block, BalancerBlock>();
-  private final MovedBlocks<BalancerDatanode.StorageGroup> movedBlocks;
-
-  /** Map (datanodeUuid,storageType -> StorageGroup) */
-  private final StorageGroupMap storageGroupMap = new StorageGroupMap();
-  
-  private NetworkTopology cluster;
-
-  private final ExecutorService moverExecutor;
-  private final ExecutorService dispatcherExecutor;
-  private final int maxConcurrentMovesPerNode;
-
-
-  private static class StorageGroupMap {
-    private static String toKey(String datanodeUuid, StorageType storageType) {
-      return datanodeUuid + ":" + storageType;
-    }
-
-    private final Map<String, BalancerDatanode.StorageGroup> map
-        = new HashMap<String, BalancerDatanode.StorageGroup>();
-    
-    BalancerDatanode.StorageGroup get(String datanodeUuid, StorageType storageType) {
-      return map.get(toKey(datanodeUuid, storageType));
-    }
-    
-    void put(BalancerDatanode.StorageGroup g) {
-      final String key = toKey(g.getDatanode().getDatanodeUuid(), g.storageType);
-      final BalancerDatanode.StorageGroup existing = map.put(key, g);
-      Preconditions.checkState(existing == null);
-    }
-    
-    int size() {
-      return map.size();
-    }
-    
-    void clear() {
-      map.clear();
-    }
-  }
-  /* This class keeps track of a scheduled block move */
-  private class PendingBlockMove {
-    private BalancerBlock block;
-    private Source source;
-    private BalancerDatanode proxySource;
-    private BalancerDatanode.StorageGroup target;
-    
-    /** constructor */
-    private PendingBlockMove() {
-    }
-    
-    @Override
-    public String toString() {
-      final Block b = block.getBlock();
-      return b + " with size=" + b.getNumBytes() + " from "
-          + source.getDisplayName() + " to " + target.getDisplayName()
-          + " through " + proxySource.datanode;
-    }
-
-    /* choose a block & a proxy source for this pendingMove 
-     * whose source & target have already been chosen.
-     * 
-     * Return true if a block and its proxy are chosen; false otherwise
-     */
-    private boolean chooseBlockAndProxy() {
-      // iterate all source's blocks until find a good one
-      for (Iterator<BalancerBlock> blocks=
-        source.getBlockIterator(); blocks.hasNext();) {
-        if (markMovedIfGoodBlock(blocks.next())) {
-          blocks.remove();
-          return true;
-        }
-      }
-      return false;
-    }
-    
-    /* Return true if the given block is good for the tentative move;
-     * If it is good, add it to the moved list to marked as "Moved".
-     * A block is good if
-     * 1. it is a good candidate; see isGoodBlockCandidate
-     * 2. can find a proxy source that's not busy for this move
-     */
-    private boolean markMovedIfGoodBlock(BalancerBlock block) {
-      synchronized(block) {
-        synchronized(movedBlocks) {
-          if (isGoodBlockCandidate(source, target, block)) {
-            this.block = block;
-            if ( chooseProxySource() ) {
-              movedBlocks.put(block);
-              if (LOG.isDebugEnabled()) {
-                LOG.debug("Decided to move " + this);
-              }
-              return true;
-            }
-          }
-        }
-      }
-      return false;
-    }
-    
-    /* Now we find out source, target, and block, we need to find a proxy
-     * 
-     * @return true if a proxy is found; otherwise false
-     */
-    private boolean chooseProxySource() {
-      final DatanodeInfo targetDN = target.getDatanode();
-      // if node group is supported, first try add nodes in the same node group
-      if (cluster.isNodeGroupAware()) {
-        for (BalancerDatanode.StorageGroup loc : block.getLocations()) {
-          if (cluster.isOnSameNodeGroup(loc.getDatanode(), targetDN) && addTo(loc)) {
-            return true;
-          }
-        }
-      }
-      // check if there is replica which is on the same rack with the target
-      for (BalancerDatanode.StorageGroup loc : block.getLocations()) {
-        if (cluster.isOnSameRack(loc.getDatanode(), targetDN) && addTo(loc)) {
-          return true;
-        }
-      }
-      // find out a non-busy replica
-      for (BalancerDatanode.StorageGroup loc : block.getLocations()) {
-        if (addTo(loc)) {
-          return true;
-        }
-      }
-      return false;
-    }
-    
-    /** add to a proxy source for specific block movement */
-    private boolean addTo(BalancerDatanode.StorageGroup g) {
-      final BalancerDatanode bdn = g.getBalancerDatanode();
-      if (bdn.addPendingBlock(this)) {
-        proxySource = bdn;
-        return true;
-      }
-      return false;
-    }
-    
-    /* Dispatch the block move task to the proxy source & wait for the response
-     */
-    private void dispatch() {
-      Socket sock = new Socket();
-      DataOutputStream out = null;
-      DataInputStream in = null;
-      try {
-        sock.connect(
-            NetUtils.createSocketAddr(target.getDatanode().getXferAddr()),
-            HdfsServerConstants.READ_TIMEOUT);
-        /* Unfortunately we don't have a good way to know if the Datanode is
-         * taking a really long time to move a block, OR something has
-         * gone wrong and it's never going to finish. To deal with this 
-         * scenario, we set a long timeout (20 minutes) to avoid hanging
-         * the balancer indefinitely.
-         */
-        sock.setSoTimeout(BLOCK_MOVE_READ_TIMEOUT);
-
-        sock.setKeepAlive(true);
-        
-        OutputStream unbufOut = sock.getOutputStream();
-        InputStream unbufIn = sock.getInputStream();
-        ExtendedBlock eb = new ExtendedBlock(nnc.getBlockpoolID(), block.getBlock());
-        Token<BlockTokenIdentifier> accessToken = keyManager.getAccessToken(eb);
-        IOStreamPair saslStreams = saslClient.socketSend(sock, unbufOut,
-          unbufIn, keyManager, accessToken, target.getDatanode());
-        unbufOut = saslStreams.out;
-        unbufIn = saslStreams.in;
-        out = new DataOutputStream(new BufferedOutputStream(unbufOut,
-            HdfsConstants.IO_FILE_BUFFER_SIZE));
-        in = new DataInputStream(new BufferedInputStream(unbufIn,
-            HdfsConstants.IO_FILE_BUFFER_SIZE));
-        
-        sendRequest(out, eb, StorageType.DEFAULT, accessToken);
-        receiveResponse(in);
-        bytesMoved.addAndGet(block.getNumBytes());
-        LOG.info("Successfully moved " + this);
-      } catch (IOException e) {
-        LOG.warn("Failed to move " + this + ": " + e.getMessage());
-        /* proxy or target may have an issue, insert a small delay
-         * before using these nodes further. This avoids a potential storm
-         * of "threads quota exceeded" Warnings when the balancer
-         * gets out of sync with work going on in datanode.
-         */
-        proxySource.activateDelay(DELAY_AFTER_ERROR);
-        target.getBalancerDatanode().activateDelay(DELAY_AFTER_ERROR);
-      } finally {
-        IOUtils.closeStream(out);
-        IOUtils.closeStream(in);
-        IOUtils.closeSocket(sock);
-        
-        proxySource.removePendingBlock(this);
-        target.getBalancerDatanode().removePendingBlock(this);
-
-        synchronized (this ) {
-          reset();
-        }
-        synchronized (Balancer.this) {
-          Balancer.this.notifyAll();
-        }
-      }
-    }
-    
-    /* Send a block replace request to the output stream*/
-    private void sendRequest(DataOutputStream out, ExtendedBlock eb,
-        StorageType storageType, 
-        Token<BlockTokenIdentifier> accessToken) throws IOException {
-      new Sender(out).replaceBlock(eb, storageType, accessToken,
-          source.getDatanode().getDatanodeUuid(), proxySource.datanode);
-    }
-    
-    /* Receive a block copy response from the input stream */ 
-    private void receiveResponse(DataInputStream in) throws IOException {
-      BlockOpResponseProto response = BlockOpResponseProto.parseFrom(
-          vintPrefixed(in));
-      if (response.getStatus() != Status.SUCCESS) {
-        if (response.getStatus() == Status.ERROR_ACCESS_TOKEN)
-          throw new IOException("block move failed due to access token error");
-        throw new IOException("block move is failed: " +
-            response.getMessage());
-      }
-    }
-
-    /* reset the object */
-    private void reset() {
-      block = null;
-      source = null;
-      proxySource = null;
-      target = null;
-    }
-    
-    /* start a thread to dispatch the block move */
-    private void scheduleBlockMove() {
-      moverExecutor.execute(new Runnable() {
-        @Override
-        public void run() {
-          if (LOG.isDebugEnabled()) {
-            LOG.debug("Start moving " + PendingBlockMove.this);
-          }
-          dispatch();
-        }
-      });
-    }
-  }
-  
-  /* A class for keeping track of blocks in the Balancer */
-  static class BalancerBlock extends MovedBlocks.Locations<BalancerDatanode.StorageGroup> {
-    BalancerBlock(Block block) {
-      super(block);
-    }
-  }
-  
-  /* The class represents a desired move of bytes between two nodes 
-   * and the target.
-   * An object of this class is stored in a source. 
-   */
-  static private class Task {
-    private final BalancerDatanode.StorageGroup target;
-    private long size;  //bytes scheduled to move
-    
-    /* constructor */
-    private Task(BalancerDatanode.StorageGroup target, long size) {
-      this.target = target;
-      this.size = size;
-    }
-  }
-  
-  
-  /* A class that keeps track of a datanode in Balancer */
-  private static class BalancerDatanode {
-    
-    /** A group of storages in a datanode with the same storage type. */
-    private class StorageGroup {
-      final StorageType storageType;
-      final double utilization;
-      final long maxSize2Move;
-      private long scheduledSize = 0L;
-
-      private StorageGroup(StorageType storageType, double utilization,
-          long maxSize2Move) {
-        this.storageType = storageType;
-        this.utilization = utilization;
-        this.maxSize2Move = maxSize2Move;
-      }
-      
-      BalancerDatanode getBalancerDatanode() {
-        return BalancerDatanode.this;
-      }
-
-      DatanodeInfo getDatanode() {
-        return BalancerDatanode.this.datanode;
-      }
-
-      /** Decide if still need to move more bytes */
-      protected synchronized boolean hasSpaceForScheduling() {
-        return availableSizeToMove() > 0L;
-      }
-
-      /** @return the total number of bytes that need to be moved */
-      synchronized long availableSizeToMove() {
-        return maxSize2Move - scheduledSize;
-      }
-      
-      /** increment scheduled size */
-      synchronized void incScheduledSize(long size) {
-        scheduledSize += size;
-      }
-      
-      /** @return scheduled size */
-      synchronized long getScheduledSize() {
-        return scheduledSize;
-      }
-      
-      /** Reset scheduled size to zero. */
-      synchronized void resetScheduledSize() {
-        scheduledSize = 0L;
-      }
-
-      /** @return the name for display */
-      String getDisplayName() {
-        return datanode + ":" + storageType;
-      }
-      
-      @Override
-      public String toString() {
-        return "" + utilization;
-      }
-    }
-
-    final DatanodeInfo datanode;
-    final EnumMap<StorageType, StorageGroup> storageMap
-        = new EnumMap<StorageType, StorageGroup>(StorageType.class);
-    protected long delayUntil = 0L;
-    //  blocks being moved but not confirmed yet
-    private final List<PendingBlockMove> pendingBlocks;
-    private final int maxConcurrentMoves;
-    
-    @Override
-    public String toString() {
-      return getClass().getSimpleName() + ":" + datanode + ":" + storageMap;
-    }
-
-    /* Constructor 
-     * Depending on avgutil & threshold, calculate maximum bytes to move 
-     */
-    private BalancerDatanode(DatanodeStorageReport report,
-        double threshold, int maxConcurrentMoves) {
-      this.datanode = report.getDatanodeInfo();
-      this.maxConcurrentMoves = maxConcurrentMoves;
-      this.pendingBlocks = new ArrayList<PendingBlockMove>(maxConcurrentMoves);
-    }
-    
-    private void put(StorageType storageType, StorageGroup g) {
-      final StorageGroup existing = storageMap.put(storageType, g);
-      Preconditions.checkState(existing == null);
-    }
-
-    StorageGroup addStorageGroup(StorageType storageType, double utilization,
-        long maxSize2Move) {
-      final StorageGroup g = new StorageGroup(storageType, utilization,
-          maxSize2Move);
-      put(storageType, g);
-      return g;
-    }
-
-    Source addSource(StorageType storageType, double utilization,
-        long maxSize2Move, Balancer balancer) {
-      final Source s = balancer.new Source(storageType, utilization,
-          maxSize2Move, this);
-      put(storageType, s);
-      return s;
-    }
-
-    synchronized private void activateDelay(long delta) {
-      delayUntil = Time.now() + delta;
-    }
-
-    synchronized private boolean isDelayActive() {
-      if (delayUntil == 0 || Time.now() > delayUntil){
-        delayUntil = 0;
-        return false;
-      }
-        return true;
-    }
-    
-    /* Check if the node can schedule more blocks to move */
-    synchronized private boolean isPendingQNotFull() {
-      if ( pendingBlocks.size() < this.maxConcurrentMoves ) {
-        return true;
-      }
-      return false;
-    }
-    
-    /* Check if all the dispatched moves are done */
-    synchronized private boolean isPendingQEmpty() {
-      return pendingBlocks.isEmpty();
-    }
-    
-    /* Add a scheduled block move to the node */
-    private synchronized boolean addPendingBlock(
-        PendingBlockMove pendingBlock) {
-      if (!isDelayActive() && isPendingQNotFull()) {
-        return pendingBlocks.add(pendingBlock);
-      }
-      return false;
-    }
-    
-    /* Remove a scheduled block move from the node */
-    private synchronized boolean  removePendingBlock(
-        PendingBlockMove pendingBlock) {
-      return pendingBlocks.remove(pendingBlock);
-    }
-  }
-
-  /** A node that can be the sources of a block move */
-  private class Source extends BalancerDatanode.StorageGroup {
-    
-    /* A thread that initiates a block move 
-     * and waits for block move to complete */
-    private class BlockMoveDispatcher implements Runnable {
-      @Override
-      public void run() {
-        dispatchBlocks();
-      }
-    }
-    
-    private final List<Task> tasks = new ArrayList<Task>(2);
-    private long blocksToReceive = 0L;
-    /* source blocks point to balancerBlocks in the global list because
-     * we want to keep one copy of a block in balancer and be aware that
-     * the locations are changing over time.
-     */
-    private final List<BalancerBlock> srcBlockList
-            = new ArrayList<BalancerBlock>();
-    
-    /* constructor */
-    private Source(StorageType storageType, double utilization,
-        long maxSize2Move, BalancerDatanode dn) {
-      dn.super(storageType, utilization, maxSize2Move);
-    }
-    
-    /** Add a task */
-    private void addTask(Task task) {
-      Preconditions.checkState(task.target != this,
-          "Source and target are the same storage group " + getDisplayName());
-      incScheduledSize(task.size);
-      tasks.add(task);
-    }
-    
-    /* Return an iterator to this source's blocks */
-    private Iterator<BalancerBlock> getBlockIterator() {
-      return srcBlockList.iterator();
-    }
-    
-    /* fetch new blocks of this source from namenode and
-     * update this source's block list & the global block list
-     * Return the total size of the received blocks in the number of bytes.
-     */
-    private long getBlockList() throws IOException {
-      final long size = Math.min(MAX_BLOCKS_SIZE_TO_FETCH, blocksToReceive);
-      final BlockWithLocations[] newBlocks = nnc.getNamenode().getBlocks(
-          getDatanode(), size).getBlocks();
-
-      long bytesReceived = 0;
-      for (BlockWithLocations blk : newBlocks) {
-        bytesReceived += blk.getBlock().getNumBytes();
-        BalancerBlock block;
-        synchronized(globalBlockList) {
-          block = globalBlockList.get(blk.getBlock());
-          if (block==null) {
-            block = new BalancerBlock(blk.getBlock());
-            globalBlockList.put(blk.getBlock(), block);
-          } else {
-            block.clearLocations();
-          }
-        
-          synchronized (block) {
-            // update locations
-            final String[] datanodeUuids = blk.getDatanodeUuids();
-            final StorageType[] storageTypes = blk.getStorageTypes();
-            for (int i = 0; i < datanodeUuids.length; i++) {
-              final BalancerDatanode.StorageGroup g = storageGroupMap.get(
-                  datanodeUuids[i], storageTypes[i]);
-              if (g != null) { // not unknown
-                block.addLocation(g);
-              }
-            }
-          }
-          if (!srcBlockList.contains(block) && isGoodBlockCandidate(block)) {
-            // filter bad candidates
-            srcBlockList.add(block);
-          }
-        }
-      }
-      return bytesReceived;
-    }
-
-    /* Decide if the given block is a good candidate to move or not */
-    private boolean isGoodBlockCandidate(BalancerBlock block) {
-      for (Task t : tasks) {
-        if (Balancer.this.isGoodBlockCandidate(this, t.target, block)) {
-          return true;
-        }
-      }
-      return false;
-    }
-
-    /* Return a block that's good for the source thread to dispatch immediately
-     * The block's source, target, and proxy source are determined too.
-     * When choosing proxy and target, source & target throttling
-     * has been considered. They are chosen only when they have the capacity
-     * to support this block move.
-     * The block should be dispatched immediately after this method is returned.
-     */
-    private PendingBlockMove chooseNextBlockToMove() {
-      for (Iterator<Task> i = tasks.iterator(); i.hasNext();) {
-        final Task task = i.next();
-        final BalancerDatanode target = task.target.getBalancerDatanode();
-        PendingBlockMove pendingBlock = new PendingBlockMove();
-        if (target.addPendingBlock(pendingBlock)) { 
-          // target is not busy, so do a tentative block allocation
-          pendingBlock.source = this;
-          pendingBlock.target = task.target;
-          if ( pendingBlock.chooseBlockAndProxy() ) {
-            long blockSize = pendingBlock.block.getNumBytes();
-            incScheduledSize(-blockSize);
-            task.size -= blockSize;
-            if (task.size == 0) {
-              i.remove();
-            }
-            return pendingBlock;
-          } else {
-            // cancel the tentative move
-            target.removePendingBlock(pendingBlock);
-          }
-        }
-      }
-      return null;
-    }
-
-    /* iterate all source's blocks to remove moved ones */    
-    private void filterMovedBlocks() {
-      for (Iterator<BalancerBlock> blocks=getBlockIterator();
-            blocks.hasNext();) {
-        if (movedBlocks.contains(blocks.next().getBlock())) {
-          blocks.remove();
-        }
-      }
-    }
-    
-    private static final int SOURCE_BLOCK_LIST_MIN_SIZE=5;
-    /* Return if should fetch more blocks from namenode */
-    private boolean shouldFetchMoreBlocks() {
-      return srcBlockList.size()<SOURCE_BLOCK_LIST_MIN_SIZE &&
-                 blocksToReceive>0;
-    }
-    
-    /* This method iteratively does the following:
-     * it first selects a block to move,
-     * then sends a request to the proxy source to start the block move
-     * when the source's block list falls below a threshold, it asks
-     * the namenode for more blocks.
-     * It terminates when it has dispatch enough block move tasks or
-     * it has received enough blocks from the namenode, or 
-     * the elapsed time of the iteration has exceeded the max time limit.
-     */ 
-    private static final long MAX_ITERATION_TIME = 20*60*1000L; //20 mins
-    private void dispatchBlocks() {
-      long startTime = Time.now();
-      long scheduledSize = getScheduledSize();
-      this.blocksToReceive = 2*scheduledSize;
-      boolean isTimeUp = false;
-      int noPendingBlockIteration = 0;
-      while(!isTimeUp && getScheduledSize()>0 &&
-          (!srcBlockList.isEmpty() || blocksToReceive>0)) {
-        PendingBlockMove pendingBlock = chooseNextBlockToMove();
-        if (pendingBlock != null) {
-          // move the block
-          pendingBlock.scheduleBlockMove();
-          continue;
-        }
-        
-        /* Since we can not schedule any block to move,
-         * filter any moved blocks from the source block list and
-         * check if we should fetch more blocks from the namenode
-         */
-        filterMovedBlocks(); // filter already moved blocks
-        if (shouldFetchMoreBlocks()) {
-          // fetch new blocks
-          try {
-            blocksToReceive -= getBlockList();
-            continue;
-          } catch (IOException e) {
-            LOG.warn("Exception while getting block list", e);
-            return;
-          }
-        } else {
-          // source node cannot find a pendingBlockToMove, iteration +1
-          noPendingBlockIteration++;
-          // in case no blocks can be moved for source node's task,
-          // jump out of while-loop after 5 iterations.
-          if (noPendingBlockIteration >= MAX_NO_PENDING_BLOCK_ITERATIONS) {
-            resetScheduledSize();
-          }
-        }
-        
-        // check if time is up or not
-        if (Time.now()-startTime > MAX_ITERATION_TIME) {
-          isTimeUp = true;
-          continue;
-        }
-        
-        /* Now we can not schedule any block to move and there are
-         * no new blocks added to the source block list, so we wait. 
-         */
-        try {
-          synchronized(Balancer.this) {
-            Balancer.this.wait(1000);  // wait for targets/sources to be idle
-          }
-        } catch (InterruptedException ignored) {
-        }
-      }
-    }
-  }
 
   /* Check that this Balancer is compatible with the Block Placement Policy
    * used by the Namenode.
@@ -888,38 +210,12 @@ public class Balancer {
    * when connection fails.
    */
   Balancer(NameNodeConnector theblockpool, Parameters p, Configuration conf) {
+    this.dispatcher = new Dispatcher(theblockpool, p.nodesToBeIncluded,
+        p.nodesToBeExcluded, conf);
     this.threshold = p.threshold;
     this.policy = p.policy;
-    this.nodesToBeExcluded = p.nodesToBeExcluded;
-    this.nodesToBeIncluded = p.nodesToBeIncluded;
-    this.nnc = theblockpool;
-    this.keyManager = nnc.getKeyManager();
-    
-    final long movedWinWidth = conf.getLong(
-        DFSConfigKeys.DFS_BALANCER_MOVEDWINWIDTH_KEY, 
-        DFSConfigKeys.DFS_BALANCER_MOVEDWINWIDTH_DEFAULT);
-    movedBlocks = new MovedBlocks<BalancerDatanode.StorageGroup>(movedWinWidth);
-
-    cluster = NetworkTopology.getInstance(conf);
-
-    this.moverExecutor = Executors.newFixedThreadPool(
-            conf.getInt(DFSConfigKeys.DFS_BALANCER_MOVERTHREADS_KEY,
-                        DFSConfigKeys.DFS_BALANCER_MOVERTHREADS_DEFAULT));
-    this.dispatcherExecutor = Executors.newFixedThreadPool(
-            conf.getInt(DFSConfigKeys.DFS_BALANCER_DISPATCHERTHREADS_KEY,
-                        DFSConfigKeys.DFS_BALANCER_DISPATCHERTHREADS_DEFAULT));
-    this.maxConcurrentMovesPerNode =
-        conf.getInt(DFSConfigKeys.DFS_DATANODE_BALANCE_MAX_NUM_CONCURRENT_MOVES_KEY,
-        DFSConfigKeys.DFS_DATANODE_BALANCE_MAX_NUM_CONCURRENT_MOVES_DEFAULT);
-    this.saslClient = new SaslDataTransferClient(
-      DataTransferSaslUtil.getSaslPropertiesResolver(conf),
-      TrustedChannelResolver.getInstance(conf),
-      conf.getBoolean(
-        IPC_CLIENT_FALLBACK_TO_SIMPLE_AUTH_ALLOWED_KEY,
-        IPC_CLIENT_FALLBACK_TO_SIMPLE_AUTH_ALLOWED_DEFAULT));
   }
   
-  
   private static long getCapacity(DatanodeStorageReport report, StorageType t) {
     long capacity = 0L;
     for(StorageReport r : report.getStorageReports()) {
@@ -940,26 +236,6 @@ public class Balancer {
     return remaining;
   }
 
-  private boolean shouldIgnore(DatanodeInfo dn) {
-    //ignore decommissioned nodes
-    final boolean decommissioned = dn.isDecommissioned();
-    //ignore decommissioning nodes
-    final boolean decommissioning = dn.isDecommissionInProgress();
-    // ignore nodes in exclude list
-    final boolean excluded = Util.shouldBeExcluded(nodesToBeExcluded, dn);
-    // ignore nodes not in the include list (if include list is not empty)
-    final boolean notIncluded = !Util.shouldBeIncluded(nodesToBeIncluded, dn);
-    
-    if (decommissioned || decommissioning || excluded || notIncluded) {
-      if (LOG.isTraceEnabled()) {
-        LOG.trace("Excluding datanode " + dn + ": " + decommissioned + ", "
-            + decommissioning + ", " + excluded + ", " + notIncluded);
-      }
-      return true;
-    }
-    return false;
-  }
-
   /**
    * Given a datanode storage set, build a network topology and decide
    * over-utilized storages, above average utilized storages, 
@@ -967,16 +243,11 @@ public class Balancer {
    * The input datanode storage set is shuffled in order to randomize
    * to the storage matching later on.
    *
-   * @return the total number of bytes that are 
-   *                needed to move to make the cluster balanced.
-   * @param reports a set of datanode storage reports
+   * @return the number of bytes needed to move in order to balance the cluster.
    */
-  private long init(DatanodeStorageReport[] reports) {
+  private long init(List<DatanodeStorageReport> reports) {
     // compute average utilization
     for (DatanodeStorageReport r : reports) {
-      if (shouldIgnore(r.getDatanodeInfo())) {
-        continue; 
-      }
       policy.accumulateSpaces(r);
     }
     policy.initAvgUtilization();
@@ -984,15 +255,8 @@ public class Balancer {
     // create network topology and classify utilization collections: 
     //   over-utilized, above-average, below-average and under-utilized.
     long overLoadedBytes = 0L, underLoadedBytes = 0L;
-    for(DatanodeStorageReport r : DFSUtil.shuffle(reports)) {
-      final DatanodeInfo datanode = r.getDatanodeInfo();
-      if (shouldIgnore(datanode)) {
-        continue; // ignore decommissioning or decommissioned nodes
-      }
-      cluster.add(datanode);
-
-      final BalancerDatanode dn = new BalancerDatanode(r, underLoadedBytes,
-          maxConcurrentMovesPerNode);
+    for(DatanodeStorageReport r : reports) {
+      final BalancerDatanode dn = dispatcher.newDatanode(r);
       for(StorageType t : StorageType.asList()) {
         final Double utilization = policy.getUtilization(r, t);
         if (utilization == null) { // datanode does not have such storage type 
@@ -1007,7 +271,7 @@ public class Balancer {
 
         final BalancerDatanode.StorageGroup g;
         if (utilizationDiff > 0) {
-          final Source s = dn.addSource(t, utilization, maxSize2Move, this);
+          final Source s = dn.addSource(t, utilization, maxSize2Move, dispatcher);
           if (thresholdDiff <= 0) { // within threshold
             aboveAvgUtilized.add(s);
           } else {
@@ -1024,14 +288,15 @@ public class Balancer {
             underUtilized.add(g);
           }
         }
-        storageGroupMap.put(g);
+        dispatcher.getStorageGroupMap().put(g);
       }
     }
 
     logUtilizationCollections();
     
-    Preconditions.checkState(storageGroupMap.size() == overUtilized.size()
-        + underUtilized.size() + aboveAvgUtilized.size() + belowAvgUtilized.size(),
+    Preconditions.checkState(dispatcher.getStorageGroupMap().size()
+        == overUtilized.size() + underUtilized.size() + aboveAvgUtilized.size()
+           + belowAvgUtilized.size(),
         "Mismatched number of storage groups");
     
     // return number of bytes to be moved in order to make the cluster balanced
@@ -1078,7 +343,7 @@ public class Balancer {
    */
   private long chooseStorageGroups() {
     // First, match nodes on the same node group if cluster is node group aware
-    if (cluster.isNodeGroupAware()) {
+    if (dispatcher.getCluster().isNodeGroupAware()) {
       chooseStorageGroups(Matcher.SAME_NODE_GROUP);
     }
     
@@ -1087,15 +352,7 @@ public class Balancer {
     // At last, match all remaining nodes
     chooseStorageGroups(Matcher.ANY_OTHER);
     
-    Preconditions.checkState(storageGroupMap.size() >= sources.size() + targets.size(),
-        "Mismatched number of datanodes (" + storageGroupMap.size() + " < "
-            + sources.size() + " sources, " + targets.size() + " targets)");
-
-    long bytesToMove = 0L;
-    for (Source src : sources) {
-      bytesToMove += src.getScheduledSize();
-    }
-    return bytesToMove;
+    return dispatcher.bytesToMove();
   }
 
   /** Decide all <source, target> pairs according to the matcher. */
@@ -1167,9 +424,8 @@ public class Balancer {
     long size = Math.min(source.availableSizeToMove(), target.availableSizeToMove());
     final Task task = new Task(target, size);
     source.addTask(task);
-    target.incScheduledSize(task.size);
-    sources.add(source);
-    targets.add(target);
+    target.incScheduledSize(task.getSize());
+    dispatcher.add(source, target);
     LOG.info("Decided to move "+StringUtils.byteDesc(size)+" bytes from "
         + source.getDisplayName() + " to " + target.getDisplayName());
   }
@@ -1183,7 +439,8 @@ public class Balancer {
         final C c = candidates.next();
         if (!c.hasSpaceForScheduling()) {
           candidates.remove();
-        } else if (matcher.match(cluster, g.getDatanode(), c.getDatanode())) {
+        } else if (matcher.match(dispatcher.getCluster(),
+            g.getDatanode(), c.getDatanode())) {
           return c;
         }
       }
@@ -1191,172 +448,16 @@ public class Balancer {
     return null;
   }
 
-  private final AtomicLong bytesMoved = new AtomicLong();
-  
-  /* 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.
-   */
-  private long dispatchBlockMoves() throws InterruptedException {
-    long bytesLastMoved = bytesMoved.get();
-    Future<?>[] futures = new Future<?>[sources.size()];
-    int i=0;
-    for (Source source : sources) {
-      futures[i++] = dispatcherExecutor.submit(source.new BlockMoveDispatcher());
-    }
-    
-    // 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;
-  }
-  
-  // The sleeping period before checking if block move is completed again
-  static private long blockMoveWaitTime = 30000L;
-  
-  /** set the sleeping period for block move completion check */
-  static void setBlockMoveWaitTime(long time) {
-    blockMoveWaitTime = time;
-  }
-  
-  /* wait for all block move confirmations 
-   * by checking each target's pendingMove queue 
-   */
-  private void waitForMoveCompletion() {
-    boolean shouldWait;
-    do {
-      shouldWait = false;
-      for (BalancerDatanode.StorageGroup target : targets) {
-        if (!target.getBalancerDatanode().isPendingQEmpty()) {
-          shouldWait = true;
-          break;
-        }
-      }
-      if (shouldWait) {
-        try {
-          Thread.sleep(blockMoveWaitTime);
-        } catch (InterruptedException ignored) {
-        }
-      }
-    } while (shouldWait);
-  }
-
-  /* Decide if it is OK to move the given block from source to target
-   * A block is a good candidate if
-   * 1. the block is not in the process of being moved/has not been moved;
-   * 2. the block does not have a replica on the target;
-   * 3. doing the move does not reduce the number of racks that the block has
-   */
-  private boolean isGoodBlockCandidate(Source source, 
-      BalancerDatanode.StorageGroup target, BalancerBlock block) {
-    if (source.storageType != target.storageType) {
-      return false;
-    }
-    // check if the block is moved or not
-    if (movedBlocks.contains(block.getBlock())) {
-      return false;
-    }
-    if (block.isLocatedOn(target)) {
-      return false;
-    }
-    if (cluster.isNodeGroupAware() && 
-        isOnSameNodeGroupWithReplicas(target, block, source)) {
-      return false;
-    }
-
-    boolean goodBlock = false;
-    if (cluster.isOnSameRack(source.getDatanode(), target.getDatanode())) {
-      // good if source and target are on the same rack
-      goodBlock = true;
-    } else {
-      boolean notOnSameRack = true;
-      synchronized (block) {
-        for (BalancerDatanode.StorageGroup loc : block.getLocations()) {
-          if (cluster.isOnSameRack(loc.getDatanode(), target.getDatanode())) {
-            notOnSameRack = false;
-            break;
-          }
-        }
-      }
-      if (notOnSameRack) {
-        // good if target is target is not on the same rack as any replica
-        goodBlock = true;
-      } else {
-        // good if source is on the same rack as on of the replicas
-        for (BalancerDatanode.StorageGroup loc : block.getLocations()) {
-          if (loc != source && 
-              cluster.isOnSameRack(loc.getDatanode(), source.getDatanode())) {
-            goodBlock = true;
-            break;
-          }
-        }
-      }
-    }
-    return goodBlock;
-  }
-
-  /**
-   * Check if there are any replica (other than source) on the same node group
-   * with target. If true, then target is not a good candidate for placing 
-   * specific block replica as we don't want 2 replicas under the same nodegroup 
-   * after balance.
-   * @param target targetDataNode
-   * @param block dataBlock
-   * @param source sourceDataNode
-   * @return true if there are any replica (other than source) on the same node
-   * group with target
-   */
-  private boolean isOnSameNodeGroupWithReplicas(BalancerDatanode.StorageGroup target,
-      BalancerBlock block, Source source) {
-    final DatanodeInfo targetDn = target.getDatanode();
-    for (BalancerDatanode.StorageGroup loc : block.getLocations()) {
-      if (loc != source && 
-          cluster.isOnSameNodeGroup(loc.getDatanode(), targetDn)) {
-        return true;
-      }
-    }
-    return false;
-  }
-
   /* reset all fields in a balancer preparing for the next iteration */
   private void resetData(Configuration conf) {
-    this.cluster = NetworkTopology.getInstance(conf);
     this.overUtilized.clear();
     this.aboveAvgUtilized.clear();
     this.belowAvgUtilized.clear();
     this.underUtilized.clear();
-    this.storageGroupMap.clear();
-    this.sources.clear();
-    this.targets.clear();  
     this.policy.reset();
-    cleanGlobalBlockList();
-    this.movedBlocks.cleanup();
+    dispatcher.reset(conf);;
   }
   
-  /* Remove all blocks from the global block list except for the ones in the
-   * moved list.
-   */
-  private void cleanGlobalBlockList() {
-    for (Iterator<Block> globalBlockListIterator=globalBlockList.keySet().iterator();
-    globalBlockListIterator.hasNext();) {
-      Block block = globalBlockListIterator.next();
-      if(!movedBlocks.contains(block)) {
-        globalBlockListIterator.remove();
-      }
-    }
-  }
-
   // Exit status
   enum ReturnStatus {
     // These int values will map directly to the balancer process's exit code.
@@ -1380,11 +481,8 @@ public class Balancer {
   private ReturnStatus run(int iteration, Formatter formatter,
       Configuration conf) {
     try {
-      /* get all live datanodes of a cluster and their disk usage
-       * decide the number of bytes need to be moved
-       */
-      final long bytesLeftToMove = init(
-          nnc.getClient().getDatanodeStorageReport(DatanodeReportType.LIVE));
+      final List<DatanodeStorageReport> reports = dispatcher.init();
+      final long bytesLeftToMove = init(reports);
       if (bytesLeftToMove == 0) {
         System.out.println("The cluster is balanced. Exiting...");
         return ReturnStatus.SUCCESS;
@@ -1410,7 +508,7 @@ public class Balancer {
       formatter.format("%-24s %10d  %19s  %18s  %17s%n",
           DateFormat.getDateTimeInstance().format(new Date()),
           iteration,
-          StringUtils.byteDesc(bytesMoved.get()),
+          StringUtils.byteDesc(dispatcher.getBytesMoved()),
           StringUtils.byteDesc(bytesLeftToMove),
           StringUtils.byteDesc(bytesToMove)
           );
@@ -1421,7 +519,7 @@ public class Balancer {
        * available to move.
        * Exit no byte has been moved for 5 consecutive iterations.
        */
-      if (!this.nnc.shouldContinue(dispatchBlockMoves())) {
+      if (!dispatcher.dispatchAndCheckContinue()) {
         return ReturnStatus.NO_MOVE_PROGRESS;
       }
 
@@ -1436,9 +534,7 @@ public class Balancer {
       System.out.println(e + ".  Exiting ...");
       return ReturnStatus.INTERRUPTED;
     } finally {
-      // shutdown thread pools
-      dispatcherExecutor.shutdownNow();
-      moverExecutor.shutdownNow();
+      dispatcher.shutdownNow();
     }
   }
 
@@ -1547,76 +643,6 @@ public class Balancer {
     }
   }
 
-  static class Util {
-
-    /**
-     * @param datanode
-     * @return returns true if data node is part of the excludedNodes.
-     */
-    static boolean shouldBeExcluded(Set<String> excludedNodes, DatanodeInfo datanode) {
-      return isIn(excludedNodes, datanode);
-    }
-
-    /**
-     * @param datanode
-     * @return returns true if includedNodes is empty or data node is part of the includedNodes.
-     */
-    static boolean shouldBeIncluded(Set<String> includedNodes, DatanodeInfo datanode) {
-      return (includedNodes.isEmpty() ||
-          isIn(includedNodes, datanode));
-    }
-    /**
-     * Match is checked using host name , ip address with and without port number.
-     * @param datanodeSet
-     * @param datanode
-     * @return true if the datanode's transfer address matches the set of nodes.
-     */
-    private static boolean isIn(Set<String> datanodeSet, DatanodeInfo datanode) {
-      return isIn(datanodeSet, datanode.getPeerHostName(), datanode.getXferPort()) ||
-          isIn(datanodeSet, datanode.getIpAddr(), datanode.getXferPort()) ||
-          isIn(datanodeSet, datanode.getHostName(), datanode.getXferPort());
-    }
-
-    /**
-     * returns true if nodes contains host or host:port
-     * @param nodes
-     * @param host
-     * @param port
-     * @return
-     */
-    private static boolean isIn(Set<String> nodes, String host, int port) {
-      if (host == null) {
-        return false;
-      }
-      return (nodes.contains(host) || nodes.contains(host +":"+ port));
-    }
-
-    /**
-     * parse a comma separated string to obtain set of host names
-     * @param string
-     * @return
-     */
-    static Set<String> parseHostList(String string) {
-      String[] addrs = StringUtils.getTrimmedStrings(string);
-      return new HashSet<String>(Arrays.asList(addrs));
-    }
-
-    /**
-     * read set of host names from a file
-     * @param fileName
-     * @return
-     */
-    static Set<String> getHostListFromFile(String fileName) {
-      Set<String> nodes = new HashSet <String> ();
-      try {
-        HostsFileReader.readFileToSet("nodes", fileName, nodes);
-        return StringUtils.getTrimmedStrings(nodes);
-      } catch (IOException e) {
-        throw new IllegalArgumentException("Unable to open file: " + fileName);
-      }
-    }
-  }
-
   static class Cli extends Configured implements Tool {
     /**
      * Parse arguments and then run Balancer.
@@ -1689,7 +715,7 @@ public class Balancer {
                 checkArgument(++i < args.length,
                     "File containing nodes to exclude is not specified: args = "
                     + Arrays.toString(args));
-                nodesTobeExcluded = Util.getHostListFromFile(args[i]);
+                nodesTobeExcluded = Util.getHostListFromFile(args[i], "exclude");
               } else {
                 nodesTobeExcluded = Util.parseHostList(args[i]);
               }
@@ -1701,7 +727,7 @@ public class Balancer {
                 checkArgument(++i < args.length,
                     "File containing nodes to include is not specified: args = "
                     + Arrays.toString(args));
-                nodesTobeIncluded = Util.getHostListFromFile(args[i]);
+                nodesTobeIncluded = Util.getHostListFromFile(args[i], "include");
                } else {
                 nodesTobeIncluded = Util.parseHostList(args[i]);
               }

Modified: hadoop/common/branches/branch-2/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/balancer/NameNodeConnector.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/branch-2/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/balancer/NameNodeConnector.java?rev=1616890&r1=1616889&r2=1616890&view=diff
==============================================================================
--- hadoop/common/branches/branch-2/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/balancer/NameNodeConnector.java (original)
+++ hadoop/common/branches/branch-2/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/balancer/NameNodeConnector.java Fri Aug  8 21:35:44 2014
@@ -34,6 +34,10 @@ import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.hdfs.NameNodeProxies;
 import org.apache.hadoop.hdfs.protocol.AlreadyBeingCreatedException;
 import org.apache.hadoop.hdfs.protocol.ClientProtocol;
+import org.apache.hadoop.hdfs.protocol.DatanodeInfo;
+import org.apache.hadoop.hdfs.protocol.HdfsConstants.DatanodeReportType;
+import org.apache.hadoop.hdfs.server.protocol.BlocksWithLocations;
+import org.apache.hadoop.hdfs.server.protocol.DatanodeStorageReport;
 import org.apache.hadoop.hdfs.server.protocol.NamenodeProtocol;
 import org.apache.hadoop.hdfs.server.protocol.NamespaceInfo;
 import org.apache.hadoop.io.IOUtils;
@@ -90,14 +94,16 @@ public class NameNodeConnector implement
     return blockpoolID;
   }
 
-  /** @return the namenode proxy. */
-  public NamenodeProtocol getNamenode() {
-    return namenode;
+  /** @return blocks with locations. */
+  public BlocksWithLocations getBlocks(DatanodeInfo datanode, long size)
+      throws IOException {
+    return namenode.getBlocks(datanode, size);
   }
 
-  /** @return the client proxy. */
-  public ClientProtocol getClient() {
-    return client;
+  /** @return live datanode storage reports. */
+  public DatanodeStorageReport[] getLiveDatanodeStorageReport()
+      throws IOException {
+    return client.getDatanodeStorageReport(DatanodeReportType.LIVE);
   }
 
   /** @return the key manager */

Modified: hadoop/common/branches/branch-2/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/balancer/TestBalancer.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/branch-2/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/balancer/TestBalancer.java?rev=1616890&r1=1616889&r2=1616890&view=diff
==============================================================================
--- hadoop/common/branches/branch-2/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/balancer/TestBalancer.java (original)
+++ hadoop/common/branches/branch-2/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/balancer/TestBalancer.java Fri Aug  8 21:35:44 2014
@@ -89,7 +89,7 @@ public class TestBalancer {
   private static final Random r = new Random();
 
   static {
-    Balancer.setBlockMoveWaitTime(1000L) ;
+    Dispatcher.setBlockMoveWaitTime(1000L) ;
   }
 
   static void initConf(Configuration conf) {
@@ -305,12 +305,12 @@ public class TestBalancer {
       for (DatanodeInfo datanode : datanodeReport) {
         double nodeUtilization = ((double)datanode.getDfsUsed())
             / datanode.getCapacity();
-        if (Balancer.Util.shouldBeExcluded(p.nodesToBeExcluded, datanode)) {
+        if (Dispatcher.Util.isExcluded(p.nodesToBeExcluded, datanode)) {
           assertTrue(nodeUtilization == 0);
           actualExcludedNodeCount++;
           continue;
         }
-        if (!Balancer.Util.shouldBeIncluded(p.nodesToBeIncluded, datanode)) {
+        if (!Dispatcher.Util.isIncluded(p.nodesToBeIncluded, datanode)) {
           assertTrue(nodeUtilization == 0);
           actualExcludedNodeCount++;
           continue;

Modified: hadoop/common/branches/branch-2/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/balancer/TestBalancerWithHANameNodes.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/branch-2/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/balancer/TestBalancerWithHANameNodes.java?rev=1616890&r1=1616889&r2=1616890&view=diff
==============================================================================
--- hadoop/common/branches/branch-2/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/balancer/TestBalancerWithHANameNodes.java (original)
+++ hadoop/common/branches/branch-2/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/balancer/TestBalancerWithHANameNodes.java Fri Aug  8 21:35:44 2014
@@ -44,7 +44,7 @@ public class TestBalancerWithHANameNodes
   ClientProtocol client;
 
   static {
-    Balancer.setBlockMoveWaitTime(1000L);
+    Dispatcher.setBlockMoveWaitTime(1000L);
   }
 
   /**

Modified: hadoop/common/branches/branch-2/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/balancer/TestBalancerWithMultipleNameNodes.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/branch-2/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/balancer/TestBalancerWithMultipleNameNodes.java?rev=1616890&r1=1616889&r2=1616890&view=diff
==============================================================================
--- hadoop/common/branches/branch-2/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/balancer/TestBalancerWithMultipleNameNodes.java (original)
+++ hadoop/common/branches/branch-2/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/balancer/TestBalancerWithMultipleNameNodes.java Fri Aug  8 21:35:44 2014
@@ -73,7 +73,7 @@ public class TestBalancerWithMultipleNam
   private static final Random RANDOM = new Random();
 
   static {
-    Balancer.setBlockMoveWaitTime(1000L) ;
+    Dispatcher.setBlockMoveWaitTime(1000L) ;
   }
 
   /** Common objects used in various methods. */

Modified: hadoop/common/branches/branch-2/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/balancer/TestBalancerWithNodeGroup.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/branch-2/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/balancer/TestBalancerWithNodeGroup.java?rev=1616890&r1=1616889&r2=1616890&view=diff
==============================================================================
--- hadoop/common/branches/branch-2/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/balancer/TestBalancerWithNodeGroup.java (original)
+++ hadoop/common/branches/branch-2/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/balancer/TestBalancerWithNodeGroup.java Fri Aug  8 21:35:44 2014
@@ -75,7 +75,7 @@ public class TestBalancerWithNodeGroup {
   static final int DEFAULT_BLOCK_SIZE = 100;
 
   static {
-    Balancer.setBlockMoveWaitTime(1000L) ;
+    Dispatcher.setBlockMoveWaitTime(1000L) ;
   }
 
   static Configuration createConf() {



Mime
View raw message