import org.apache.hadoop.net.NetworkTopologyWithNodeGroup; //导入依赖的package包/类
@Override
DatanodeDescriptor[] getDatanodeDescriptors(Configuration conf) {
  conf.set(CommonConfigurationKeysPublic.NET_TOPOLOGY_IMPL_KEY,
          NetworkTopologyWithNodeGroup.class.getName());
  final String[] racks = {
      "/d1/r1/n1",
      "/d1/r1/n1",
      "/d1/r1/n2",
      "/d1/r2/n3",
      "/d1/r2/n3",
      "/d1/r2/n4",
      "/d2/r3/n5",
      "/d2/r3/n6"
  };
  storages = DFSTestUtil.createDatanodeStorageInfos(racks);
  return DFSTestUtil.toDatanodeDescriptor(storages);
}
 

import org.apache.hadoop.net.NetworkTopologyWithNodeGroup; //导入依赖的package包/类
private DatanodeStorageInfo chooseLocalNodeGroup(
    NetworkTopologyWithNodeGroup clusterMap, Node localMachine,
    Set<Node> excludedNodes, long blocksize, int maxNodesPerRack,
    List<DatanodeStorageInfo> results, boolean avoidStaleNodes,
    EnumMap<StorageType, Integer> storageTypes) throws
    NotEnoughReplicasException {
  // no local machine, so choose a random machine
  if (localMachine == null) {
    return chooseRandom(NodeBase.ROOT, excludedNodes, blocksize,
        maxNodesPerRack, results, avoidStaleNodes, storageTypes);
  }

  // choose one from the local node group
  try {
    return chooseRandom(
        clusterMap.getNodeGroup(localMachine.getNetworkLocation()),
        excludedNodes, blocksize, maxNodesPerRack, results, avoidStaleNodes,
        storageTypes);
  } catch (NotEnoughReplicasException e1) {
    final DatanodeDescriptor newLocal = secondNode(localMachine, results);
    if (newLocal != null) {
      try {
        return chooseRandom(
            clusterMap.getNodeGroup(newLocal.getNetworkLocation()),
            excludedNodes, blocksize, maxNodesPerRack, results,
            avoidStaleNodes, storageTypes);
      } catch(NotEnoughReplicasException e2) {
        //otherwise randomly choose one from the network
        return chooseRandom(NodeBase.ROOT, excludedNodes, blocksize,
            maxNodesPerRack, results, avoidStaleNodes, storageTypes);
      }
    } else {
      //otherwise randomly choose one from the network
      return chooseRandom(NodeBase.ROOT, excludedNodes, blocksize,
          maxNodesPerRack, results, avoidStaleNodes, storageTypes);
    }
  }
}
 

import org.apache.hadoop.net.NetworkTopologyWithNodeGroup; //导入依赖的package包/类
static Configuration createConf() {
  Configuration conf = new HdfsConfiguration();
  TestBalancer.initConf(conf);
  conf.setLong(DFSConfigKeys.DFS_BLOCK_SIZE_KEY, DEFAULT_BLOCK_SIZE);
  conf.set(CommonConfigurationKeysPublic.NET_TOPOLOGY_IMPL_KEY, 
      NetworkTopologyWithNodeGroup.class.getName());
  conf.set(DFSConfigKeys.DFS_BLOCK_REPLICATOR_CLASSNAME_KEY, 
      BlockPlacementPolicyWithNodeGroup.class.getName());
  return conf;
}
 

import org.apache.hadoop.net.NetworkTopologyWithNodeGroup; //导入依赖的package包/类
@Before
public void setUp() throws Exception {
  FileSystem.setDefaultUri(CONF, "hdfs://localhost:0");
  CONF.set(DFSConfigKeys.DFS_NAMENODE_HTTP_ADDRESS_KEY, "0.0.0.0:0");
  // Set properties to make HDFS aware of NodeGroup.
  CONF.set(DFSConfigKeys.DFS_BLOCK_REPLICATOR_CLASSNAME_KEY, 
      BlockPlacementPolicyWithNodeGroup.class.getName());
  CONF.set(CommonConfigurationKeysPublic.NET_TOPOLOGY_IMPL_KEY, 
      NetworkTopologyWithNodeGroup.class.getName());
  
  CONF.setBoolean(DFSConfigKeys.DFS_NAMENODE_AVOID_STALE_DATANODE_FOR_WRITE_KEY, true);
  
  File baseDir = PathUtils.getTestDir(TestReplicationPolicyWithNodeGroup.class);
  
  CONF.set(DFSConfigKeys.DFS_NAMENODE_NAME_DIR_KEY,
      new File(baseDir, "name").getPath());
  
  DFSTestUtil.formatNameNode(CONF);
  namenode = new NameNode(CONF);
  final BlockManager bm = namenode.getNamesystem().getBlockManager();
  replicator = bm.getBlockPlacementPolicy();
  cluster = bm.getDatanodeManager().getNetworkTopology();
  // construct network topology
  for(int i=0; i<NUM_OF_DATANODES; i++) {
    cluster.add(dataNodes[i]);
  }
  setupDataNodeCapacity();
}
 

import org.apache.hadoop.net.NetworkTopologyWithNodeGroup; //导入依赖的package包/类
/** choose local node of <i>localMachine</i> as the target.
 * If localMachine is not available, will fallback to nodegroup/rack
 * when flag <i>fallbackToNodeGroupAndLocalRack</i> is set.
 * @return the chosen node
 */
@Override
protected DatanodeStorageInfo chooseLocalStorage(Node localMachine,
    Set<Node> excludedNodes, long blocksize, int maxNodesPerRack,
    List<DatanodeStorageInfo> results, boolean avoidStaleNodes,
    EnumMap<StorageType, Integer> storageTypes,
    boolean fallbackToNodeGroupAndLocalRack)
    throws NotEnoughReplicasException {
  DatanodeStorageInfo localStorage = chooseLocalStorage(localMachine,
      excludedNodes, blocksize, maxNodesPerRack, results,
      avoidStaleNodes, storageTypes);
  if (localStorage != null) {
    return localStorage;
  }

  if (!fallbackToNodeGroupAndLocalRack) {
    return null;
  }
  // try a node on local node group
  DatanodeStorageInfo chosenStorage = chooseLocalNodeGroup(
      (NetworkTopologyWithNodeGroup)clusterMap, localMachine, excludedNodes, 
      blocksize, maxNodesPerRack, results, avoidStaleNodes, storageTypes);
  if (chosenStorage != null) {
    return chosenStorage;
  }
  // try a node on local rack
  return chooseLocalRack(localMachine, excludedNodes, 
      blocksize, maxNodesPerRack, results, avoidStaleNodes, storageTypes);
}
 

import org.apache.hadoop.net.NetworkTopologyWithNodeGroup; //导入依赖的package包/类
private DatanodeStorageInfo chooseLocalNodeGroup(
    NetworkTopologyWithNodeGroup clusterMap, Node localMachine,
    Set<Node> excludedNodes, long blocksize, int maxNodesPerRack,
    List<DatanodeStorageInfo> results, boolean avoidStaleNodes,
    EnumMap<StorageType, Integer> storageTypes) throws
    NotEnoughReplicasException {
  // no local machine, so choose a random machine
  if (localMachine == null) {
    return chooseRandom(NodeBase.ROOT, excludedNodes, blocksize,
        maxNodesPerRack, results, avoidStaleNodes, storageTypes);
  }

  // choose one from the local node group
  try {
    return chooseRandom(
        clusterMap.getNodeGroup(localMachine.getNetworkLocation()),
        excludedNodes, blocksize, maxNodesPerRack, results, avoidStaleNodes,
        storageTypes);
  } catch (NotEnoughReplicasException e1) {
    final DatanodeDescriptor newLocal = secondNode(localMachine, results);
    if (newLocal != null) {
      try {
        return chooseRandom(
            clusterMap.getNodeGroup(newLocal.getNetworkLocation()),
            excludedNodes, blocksize, maxNodesPerRack, results,
            avoidStaleNodes, storageTypes);
      } catch(NotEnoughReplicasException e2) {
        //otherwise return null
        return null;
      }
    } else {
      //otherwise return null
      return null;
    }
  }
}
 

import org.apache.hadoop.net.NetworkTopologyWithNodeGroup; //导入依赖的package包/类
/** choose local node of localMachine as the target.
 * if localMachine is not available, choose a node on the same nodegroup or 
 * rack instead.
 * @return the chosen node
 */
@Override
protected DatanodeDescriptor chooseLocalNode(
    DatanodeDescriptor localMachine,
    HashMap<Node, Node> excludedNodes,
    long blocksize,
    int maxNodesPerRack,
    List<DatanodeDescriptor> results,
    boolean avoidStaleNodes)
      throws NotEnoughReplicasException {
  // if no local machine, randomly choose one node
  if (localMachine == null)
    return chooseRandom(NodeBase.ROOT, excludedNodes, 
        blocksize, maxNodesPerRack, results, avoidStaleNodes);

  // otherwise try local machine first
  Node oldNode = excludedNodes.put(localMachine, localMachine);
  if (oldNode == null) { // was not in the excluded list
    if (isGoodTarget(localMachine, blocksize,
        maxNodesPerRack, false, results, avoidStaleNodes)) {
      results.add(localMachine);
      // Nodes under same nodegroup should be excluded.
      addNodeGroupToExcludedNodes(excludedNodes,
          localMachine.getNetworkLocation());
      return localMachine;
    }
  } 

  // try a node on local node group
  DatanodeDescriptor chosenNode = chooseLocalNodeGroup(
      (NetworkTopologyWithNodeGroup)clusterMap, localMachine, excludedNodes, 
      blocksize, maxNodesPerRack, results, avoidStaleNodes);
  if (chosenNode != null) {
    return chosenNode;
  }
  // try a node on local rack
  return chooseLocalRack(localMachine, excludedNodes, 
      blocksize, maxNodesPerRack, results, avoidStaleNodes);
}
 

import org.apache.hadoop.net.NetworkTopologyWithNodeGroup; //导入依赖的package包/类
/** choose local node of localMachine as the target.
 * if localMachine is not available, choose a node on the same nodegroup or 
 * rack instead.
 * @return the chosen node
 */
@Override
protected DatanodeStorageInfo chooseLocalStorage(Node localMachine,
    Set<Node> excludedNodes, long blocksize, int maxNodesPerRack,
    List<DatanodeStorageInfo> results, boolean avoidStaleNodes,
    StorageType storageType) throws NotEnoughReplicasException {
  // if no local machine, randomly choose one node
  if (localMachine == null)
    return chooseRandom(NodeBase.ROOT, excludedNodes, 
        blocksize, maxNodesPerRack, results, avoidStaleNodes, storageType);

  // otherwise try local machine first
  if (localMachine instanceof DatanodeDescriptor) {
    DatanodeDescriptor localDataNode = (DatanodeDescriptor)localMachine;
    if (excludedNodes.add(localMachine)) { // was not in the excluded list
      for(DatanodeStorageInfo localStorage : DFSUtil.shuffle(
          localDataNode.getStorageInfos())) {
        if (addIfIsGoodTarget(localStorage, excludedNodes, blocksize,
            maxNodesPerRack, false, results, avoidStaleNodes, storageType) >= 0) {
          return localStorage;
        }
      }
    }
  }

  // try a node on local node group
  DatanodeStorageInfo chosenStorage = chooseLocalNodeGroup(
      (NetworkTopologyWithNodeGroup)clusterMap, localMachine, excludedNodes, 
      blocksize, maxNodesPerRack, results, avoidStaleNodes, storageType);
  if (chosenStorage != null) {
    return chosenStorage;
  }
  // try a node on local rack
  return chooseLocalRack(localMachine, excludedNodes, 
      blocksize, maxNodesPerRack, results, avoidStaleNodes, storageType);
}
 

import org.apache.hadoop.net.NetworkTopologyWithNodeGroup; //导入依赖的package包/类
private DatanodeStorageInfo chooseLocalNodeGroup(
    NetworkTopologyWithNodeGroup clusterMap, Node localMachine,
    Set<Node> excludedNodes, long blocksize, int maxNodesPerRack,
    List<DatanodeStorageInfo> results, boolean avoidStaleNodes,
    StorageType storageType) throws NotEnoughReplicasException {
  // no local machine, so choose a random machine
  if (localMachine == null) {
    return chooseRandom(NodeBase.ROOT, excludedNodes, blocksize,
        maxNodesPerRack, results, avoidStaleNodes, storageType);
  }

  // choose one from the local node group
  try {
    return chooseRandom(
        clusterMap.getNodeGroup(localMachine.getNetworkLocation()),
        excludedNodes, blocksize, maxNodesPerRack, results, avoidStaleNodes,
        storageType);
  } catch (NotEnoughReplicasException e1) {
    final DatanodeDescriptor newLocal = secondNode(localMachine, results);
    if (newLocal != null) {
      try {
        return chooseRandom(
            clusterMap.getNodeGroup(newLocal.getNetworkLocation()),
            excludedNodes, blocksize, maxNodesPerRack, results,
            avoidStaleNodes, storageType);
      } catch(NotEnoughReplicasException e2) {
        //otherwise randomly choose one from the network
        return chooseRandom(NodeBase.ROOT, excludedNodes, blocksize,
            maxNodesPerRack, results, avoidStaleNodes, storageType);
      }
    } else {
      //otherwise randomly choose one from the network
      return chooseRandom(NodeBase.ROOT, excludedNodes, blocksize,
          maxNodesPerRack, results, avoidStaleNodes, storageType);
    }
  }
}
 

import org.apache.hadoop.net.NetworkTopologyWithNodeGroup; //导入依赖的package包/类
static Configuration createConf() {
  Configuration conf = new HdfsConfiguration();
  TestBalancer.initConf(conf);
  conf.set(CommonConfigurationKeysPublic.NET_TOPOLOGY_IMPL_KEY, 
      NetworkTopologyWithNodeGroup.class.getName());
  conf.set(DFSConfigKeys.DFS_BLOCK_REPLICATOR_CLASSNAME_KEY, 
      BlockPlacementPolicyWithNodeGroup.class.getName());
  return conf;
}
 

import org.apache.hadoop.net.NetworkTopologyWithNodeGroup; //导入依赖的package包/类
/** choose local node of localMachine as the target.
 * if localMachine is not available, choose a node on the same nodegroup or 
 * rack instead.
 * @return the chosen node
 */
@Override
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)
    return chooseRandom(NodeBase.ROOT, excludedNodes, 
        blocksize, maxNodesPerRack, results, avoidStaleNodes, storageTypes);

  // otherwise try local machine first
  if (localMachine instanceof DatanodeDescriptor) {
    DatanodeDescriptor localDataNode = (DatanodeDescriptor)localMachine;
    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;
          }
        }
      }
    }
  }

  // try a node on local node group
  DatanodeStorageInfo chosenStorage = chooseLocalNodeGroup(
      (NetworkTopologyWithNodeGroup)clusterMap, localMachine, excludedNodes, 
      blocksize, maxNodesPerRack, results, avoidStaleNodes, storageTypes);
  if (chosenStorage != null) {
    return chosenStorage;
  }

  if (!fallbackToLocalRack) {
    return null;
  }
  // try a node on local rack
  return chooseLocalRack(localMachine, excludedNodes, 
      blocksize, maxNodesPerRack, results, avoidStaleNodes, storageTypes);
}
 

import org.apache.hadoop.net.NetworkTopologyWithNodeGroup; //导入依赖的package包/类
/**
 * choose all good favored nodes as target.
 * If no enough targets, then choose one replica from
 * each bad favored node's node group.
 * @throws NotEnoughReplicasException
 */
@Override
protected void chooseFavouredNodes(String src, int numOfReplicas,
    List<DatanodeDescriptor> favoredNodes,
    Set<Node> favoriteAndExcludedNodes, long blocksize,
    int maxNodesPerRack, List<DatanodeStorageInfo> results,
    boolean avoidStaleNodes, EnumMap<StorageType, Integer> storageTypes)
    throws NotEnoughReplicasException {
  super.chooseFavouredNodes(src, numOfReplicas, favoredNodes,
      favoriteAndExcludedNodes, blocksize, maxNodesPerRack, results,
      avoidStaleNodes, storageTypes);
  if (results.size() < numOfReplicas) {
    // Not enough replicas, choose from unselected Favorednode's Nodegroup
    for (int i = 0;
        i < favoredNodes.size() && results.size() < numOfReplicas; i++) {
      DatanodeDescriptor favoredNode = favoredNodes.get(i);
      boolean chosenNode =
          isNodeChosen(results, favoredNode);
      if (chosenNode) {
        continue;
      }
      NetworkTopologyWithNodeGroup clusterMapNodeGroup =
          (NetworkTopologyWithNodeGroup) clusterMap;
      // try a node on FavouredNode's node group
      DatanodeStorageInfo target = null;
      String scope =
          clusterMapNodeGroup.getNodeGroup(favoredNode.getNetworkLocation());
      try {
        target =
            chooseRandom(scope, favoriteAndExcludedNodes, blocksize,
              maxNodesPerRack, results, avoidStaleNodes, storageTypes);
      } catch (NotEnoughReplicasException e) {
        // catch Exception and continue with other favored nodes
        continue;
      }
      if (target == null) {
        LOG.warn("Could not find a target for file "
            + src + " within nodegroup of favored node " + favoredNode);
        continue;
      }
      favoriteAndExcludedNodes.add(target.getDatanodeDescriptor());
    }
  }
}
 

import org.apache.hadoop.net.NetworkTopologyWithNodeGroup; //导入依赖的package包/类
private DatanodeDescriptor chooseLocalNodeGroup(NetworkTopologyWithNodeGroup clusterMap,
    DatanodeDescriptor localMachine, HashMap<Node, Node> excludedNodes, long blocksize, 
    int maxNodesPerRack, List<DatanodeDescriptor> results, boolean avoidStaleNodes)
        throws NotEnoughReplicasException {
  // no local machine, so choose a random machine
  if (localMachine == null) {
    return chooseRandom(NodeBase.ROOT, excludedNodes, 
    blocksize, maxNodesPerRack, results, avoidStaleNodes);
  }

  // choose one from the local node group
  try {
    return chooseRandom(clusterMap.getNodeGroup(localMachine.getNetworkLocation()),
    excludedNodes, blocksize, maxNodesPerRack, results, avoidStaleNodes);
  } catch (NotEnoughReplicasException e1) {
    // find the second replica
    DatanodeDescriptor newLocal=null;
    for(Iterator<DatanodeDescriptor> iter=results.iterator();
      iter.hasNext();) {
      DatanodeDescriptor nextNode = iter.next();
      if (nextNode != localMachine) {
        newLocal = nextNode;
        break;
      }
    }
    if (newLocal != null) {
      try {
        return chooseRandom(clusterMap.getNodeGroup(newLocal.getNetworkLocation()),
          excludedNodes, blocksize, maxNodesPerRack, results, avoidStaleNodes);
      } catch(NotEnoughReplicasException e2) {
        //otherwise randomly choose one from the network
        return chooseRandom(NodeBase.ROOT, excludedNodes,
            blocksize, maxNodesPerRack, results, avoidStaleNodes);
      }
    } else {
      //otherwise randomly choose one from the network
      return chooseRandom(NodeBase.ROOT, excludedNodes,
          blocksize, maxNodesPerRack, results, avoidStaleNodes);
    }
  }
}