HDFS副本存放策略

 The class is responsible for choosing the desired number of targets
 for placing block replicas.
 The replica placement strategy is that if the writer is on a datanode,
 the 1st replica is placed on the local machine, 
 otherwise a random datanode. The 2nd replica is placed on a datanode
 that is on a different rack. The 3rd replica is placed on a datanode
 which is on a different node of the rack as the second replica.

  /**
   * choose <i>numOfReplicas</i> data nodes for <i>writer</i> 
   * to re-replicate a block with size <i>blocksize</i> 
   * If not, return as many as we can.
   *
   * @param srcPath the file to which this chooseTargets is being invoked.
   * @param numOfReplicas additional number of replicas wanted.
   * @param writer the writer's machine, null if not in the cluster.
   * @param chosen datanodes that have been chosen as targets.
   * @param returnChosenNodes decide if the chosenNodes are returned.
   * @param excludedNodes datanodes that should not be considered as targets.
   * @param blocksize size of the data to be written.
   * @return array of DatanodeDescriptor instances chosen as target
   * and sorted as a pipeline.
   */
  public abstract DatanodeStorageInfo[] chooseTarget(String srcPath,
                                             int numOfReplicas,
                                             Node writer,
                                             List<DatanodeStorageInfo> chosen,
                                             boolean returnChosenNodes,
                                             Set<Node> excludedNodes,
                                             long blocksize,
                                             BlockStoragePolicy storagePolicy);

/**
   * Same as {@link #chooseTarget(String, int, Node, Set, long, List, StorageType)}
   * with added parameter {@code favoredDatanodes}
   * @param favoredNodes datanodes that should be favored as targets. This
   *          is only a hint and due to cluster state, namenode may not be 
   *          able to place the blocks on these datanodes.
   */
  DatanodeStorageInfo[] chooseTarget(String src,
      int numOfReplicas, Node writer,
      Set<Node> excludedNodes,
      long blocksize,
      List<DatanodeDescriptor> favoredNodes,
      BlockStoragePolicy storagePolicy) {
      }          

  /** This is the implementation. */
private DatanodeStorageInfo[] chooseTarget(int numOfReplicas,
                                Node writer,
                                List<DatanodeStorageInfo> chosenStorage,
                                boolean returnChosenNodes,
                                Set<Node> excludedNodes,
                                long blocksize,
                                final BlockStoragePolicy storagePolicy) {
// 如果目标完成副本数为0或机器节点数量为0,返回空
if (numOfReplicas == 0 || clusterMap.getNumOfLeaves()==0) {
  return DatanodeStorageInfo.EMPTY_ARRAY;
}
// 创建黑名单列表集  
if (excludedNodes == null) {
  excludedNodes = new HashSet<Node>();
}
// 计算每个机架所允许最大副本数
int[] result = getMaxNodesPerRack(chosenStorage.size(), numOfReplicas);
numOfReplicas = result[0];
int maxNodesPerRack = result[1];
...

    ...
// 将所选节点加入到结果列表中,同时加入到移除列表中,意为已选择过的节点
final List<DatanodeStorageInfo> results = new ArrayList<DatanodeStorageInfo>(chosenStorage);
for (DatanodeStorageInfo storage : chosenStorage) {
  // add localMachine and related nodes to excludedNodes
  addToExcludedNodes(storage.getDatanodeDescriptor(), excludedNodes);
}
// 计算是否需要避免旧的,未更新的节点
boolean avoidStaleNodes = (stats != null
    && stats.isAvoidingStaleDataNodesForWrite());
// 选择numOfReplicas规定副本数的目标机器,并返回其中第一个节点
final Node localNode = chooseTarget(numOfReplicas, writer, excludedNodes,
    blocksize, maxNodesPerRack, results, avoidStaleNodes, storagePolicy,
// 如果不像返回初始选中的目标节点,则进行移除
if (!returnChosenNodes) {  
  results.removeAll(chosenStorage);
}
...

    ...  
// sorting nodes to form a pipeline
// 根据最短距离排序目标节点列表,形成pipeline
return getPipeline(
    (writer != null && writer instanceof DatanodeDescriptor) ? writer
        : localNode,
    results.toArray(new DatanodeStorageInfo[results.size()]));

final Node localNode = chooseTarget(numOfReplicas, writer, excludedNodes,
        blocksize, maxNodesPerRack, results, avoidStaleNodes, storagePolicy,
        EnumSet.noneOf(StorageType.class), results.isEmpty());

// 如果额外需要请求副本数为0,或者集群中没有可选节点
if (numOfReplicas == 0 || clusterMap.getNumOfLeaves()==0) {
  // 如果writer请求者在其中一个datanode上则返回此节点,否则直接返回null
  return (writer instanceof DatanodeDescriptor) ? writer : null;
}
// 获取已经选择完成的节点数
final int numOfResults = results.size();
// 计算期望希望达到的副本总数
final int totalReplicasExpected = numOfReplicas + numOfResults;
// 如果writer为空或不在datanode上,则取出已选择好列表中的第一个位置所在节点,赋值给writer
if ((writer == null || !(writer instanceof DatanodeDescriptor)) && !newBlock) {
  writer = results.get(0).getDatanodeDescriptor();
}

// Keep a copy of original excludedNodes
// 做一份移除列表名单的拷贝
final Set<Node> oldExcludedNodes = new HashSet<Node>(excludedNodes);

// choose storage types; use fallbacks for unavailable storages
// 根据存储策略获取副本需要满足的存储类型列表,如果有不可用的存储类型,会采用fallback的类型
final List<StorageType> requiredStorageTypes = storagePolicy
    .chooseStorageTypes((short) totalReplicasExpected,
        DatanodeStorageInfo.toStorageTypes(results),
        unavailableStorages, newBlock);
// 将存储类型列表进行计数统计,并存于map中
final EnumMap<StorageType, Integer> storageTypes =
    getRequiredStorageTypes(requiredStorageTypes);
if (LOG.isTraceEnabled()) {
  LOG.trace("storageTypes=" + storageTypes);
}
...

  // 如果已选择的目标节点数量为0,则表示3副本一个都还没开始选,首先从选本地节点开始
  if (numOfResults == 0) {
    writer = chooseLocalStorage(writer, excludedNodes, blocksize,
        maxNodesPerRack, results, avoidStaleNodes, storageTypes, true)
            .getDatanodeDescriptor();
    // 如果此时目标需求完成的副本数为降为0,代表选择目标完成,返回第一个节点writer
    if (--numOfReplicas == 0) {
      return writer;
    }
  }
  // 取出result列表第一个节点
  final DatanodeDescriptor dn0 = results.get(0).getDatanodeDescriptor();
  // 前面的过程已经完成首个本地节点的选择,此时进行不同机房的节点选择
  if (numOfResults <= 1) {
    // 选择1个不同于dn0所在机房的一个目标节点位置
    chooseRemoteRack(1, dn0, excludedNodes, blocksize, maxNodesPerRack,
        results, avoidStaleNodes, storageTypes);
    // 如果此时目标需求完成的副本数为降为0,代表选择目标完成,返回第一个节点writer
    if (--numOfReplicas == 0) {
      return writer;
    }
  }
  // 如果经过前面的处理,节点选择数在2个以内,需要选取第3个副本
  if (numOfResults <= 2) {
    // 取出result列表第二个节点
    final DatanodeDescriptor dn1 = results.get(1).getDatanodeDescriptor();
    // 如果dn0,dn1所在同机房,
    if (clusterMap.isOnSameRack(dn0, dn1)) {
      // 则选择1个不同于dn0,dn1所在机房的副本位置
      chooseRemoteRack(1, dn0, excludedNodes, blocksize, maxNodesPerRack,
          results, avoidStaleNodes, storageTypes);
    } else if (newBlock){
      // 如果是新的block块,则选取1个于dn1所在同机房的节点位置
      chooseLocalRack(dn1, excludedNodes, blocksize, maxNodesPerRack,
          results, avoidStaleNodes, storageTypes);
    } else {
      // 否则选取于writer同机房的位置
      chooseLocalRack(writer, excludedNodes, blocksize, maxNodesPerRack,
          results, avoidStaleNodes, storageTypes);
    }
    // 如果此时目标需求完成的副本数为降为0,代表选择目标完成,返回第一个节点writer
    if (--numOfReplicas == 0) {
      return writer;
    }
  }
  // 如果副本数已经超过2个,说明设置的block的时候,已经设置超过3副本的数量
  // 则剩余位置在集群中随机选择放置节点
  chooseRandom(numOfReplicas, NodeBase.ROOT, excludedNodes, blocksize,
      maxNodesPerRack, results, avoidStaleNodes, storageTypes);

  if (retry) {
    for (DatanodeStorageInfo resultStorage : results) {
      addToExcludedNodes(resultStorage.getDatanodeDescriptor(),
          oldExcludedNodes);
    }
    // 剔除之前完成的选择的目标位置,重新计算当前需要复制的副本数
    numOfReplicas = totalReplicasExpected - results.size();
    // 重新调用自身方法进行复制
    return chooseTarget(numOfReplicas, writer, oldExcludedNodes, blocksize,
        maxNodesPerRack, results, false, storagePolicy, unavailableStorages,
        newBlock);
  }

  protected DatanodeStorageInfo chooseLocalStorage(Node localMachine,
      Set<Node> excludedNodes, long blocksize, int maxNodesPerRack,
      List<DatanodeStorageInfo> results, boolean avoidStaleNodes,
      EnumMap<StorageType, Integer> storageTypes, boolean fallbackToLocalRack)
      throws NotEnoughReplicasException {
    // if no local machine, randomly choose one node
    if (localMachine == null) {
      // 如果本地节点为空,则降级选择1个随机节点
      return chooseRandom(NodeBase.ROOT, excludedNodes, blocksize,
          maxNodesPerRack, results, avoidStaleNodes, storageTypes);
    }
    if (preferLocalNode && localMachine instanceof DatanodeDescriptor) {
      DatanodeDescriptor localDatanode = (DatanodeDescriptor) localMachine;
      // otherwise try local machine first
      if (excludedNodes.add(localMachine)) { // was not in the excluded list
        for (Iterator<Map.Entry<StorageType, Integer>> iter = storageTypes
            .entrySet().iterator(); iter.hasNext(); ) {
          Map.Entry<StorageType, Integer> entry = iter.next();
          // 遍历本地节点可用的存储目录
          for (DatanodeStorageInfo localStorage : DFSUtil.shuffle(
              localDatanode.getStorageInfos())) {
            StorageType type = entry.getKey();
            // 加入满足条件的存储目录位置
            if (addIfIsGoodTarget(localStorage, excludedNodes, blocksize,
                maxNodesPerRack, false, results, avoidStaleNodes, type) >= 0) {
              int num = entry.getValue();
              if (num == 1) {
                iter.remove();
              } else {
                entry.setValue(num - 1);
              }
              return localStorage;
            }
          }
        }
      } 
    }

    if (!fallbackToLocalRack) {
      return null;
    }
    // 本地节点没有满足条件的存储位置,则降级选取同机架的节点
    // try a node on local rack
    return chooseLocalRack(localMachine, excludedNodes, blocksize,
        maxNodesPerRack, results, avoidStaleNodes, storageTypes);
  }

   // no local machine, so choose a random machine
 if (localMachine == null) {
   return chooseRandom(NodeBase.ROOT, excludedNodes, blocksize,
       maxNodesPerRack, results, avoidStaleNodes, storageTypes);
 }
 // 获取本地机架名
 final String localRack = localMachine.getNetworkLocation();
 
 try {
   // choose one from the local rack
   // 将机架名作为scope参数传入
   return chooseRandom(localRack, excludedNodes,
       blocksize, maxNodesPerRack, results, avoidStaleNodes, storageTypes);

    // no local machine, so choose a random machine
   try {
  // choose one from the local rack
  // 将机架名作为scope参数传入
  return chooseRandom(localRack, excludedNodes,
      blocksize, maxNodesPerRack, results, avoidStaleNodes, storageTypes);

  // 获取本地机架名称,带上前缀字符~,作为scope参数传入
 chooseRandom(numOfReplicas, "~" + localMachine.getNetworkLocation(),
     excludedNodes, blocksize, maxReplicasPerRack, results,
     avoidStaleNodes, storageTypes);

  // 获取本地机架名称,带上前缀字符~,作为scope参数传入
 chooseRandom(numOfReplicas, "~" + localMachine.getNetworkLocation(),
     excludedNodes, blocksize, maxReplicasPerRack, results,
     avoidStaleNodes, storageTypes);