package org.apache.hadoop.hdfs.server.namenode;
import org.apache.commons.logging.*;
import org.apache.hadoop.hdfs.protocol.DatanodeInfo;
import org.apache.hadoop.hdfs.protocol.FSConstants;
import org.apache.hadoop.hdfs.protocol.LocatedBlock;
import org.apache.hadoop.net.NetworkTopology;
import org.apache.hadoop.net.Node;
import org.apache.hadoop.net.NodeBase;
import java.util.*;
class ReplicationTargetChooser {
private final boolean considerLoad;
private NetworkTopology clusterMap;
private FSNamesystem fs;
ReplicationTargetChooser(boolean considerLoad, FSNamesystem fs,
NetworkTopology clusterMap) {
this.considerLoad = considerLoad;
this.fs = fs;
this.clusterMap = clusterMap;
}
private static class NotEnoughReplicasException extends Exception {
NotEnoughReplicasException(String msg) {
super(msg);
}
}
DatanodeDescriptor[] chooseTarget(int numOfReplicas,
DatanodeDescriptor writer,
List<Node> excludedNodes,
long blocksize) {
if (excludedNodes == null) {
excludedNodes = new ArrayList<Node>();
}
return chooseTarget(numOfReplicas, writer,
new ArrayList<DatanodeDescriptor>(), excludedNodes, blocksize);
}
DatanodeDescriptor[] chooseTarget(int numOfReplicas,
DatanodeDescriptor writer,
List<DatanodeDescriptor> choosenNodes,
List<Node> excludedNodes,
long blocksize) {
if (numOfReplicas == 0 || clusterMap.getNumOfLeaves()==0) {
return new DatanodeDescriptor[0];
}
if (excludedNodes == null) {
excludedNodes = new ArrayList<Node>();
}
int clusterSize = clusterMap.getNumOfLeaves();
int totalNumOfReplicas = choosenNodes.size()+numOfReplicas;
if (totalNumOfReplicas > clusterSize) {
numOfReplicas -= (totalNumOfReplicas-clusterSize);
totalNumOfReplicas = clusterSize;
}
int maxNodesPerRack =
(totalNumOfReplicas-1)/clusterMap.getNumOfRacks()+2;
List<DatanodeDescriptor> results =
new ArrayList<DatanodeDescriptor>(choosenNodes);
excludedNodes.addAll(choosenNodes);
if (!clusterMap.contains(writer)) {
writer=null;
}
DatanodeDescriptor localNode = chooseTarget(numOfReplicas, writer,
excludedNodes, blocksize, maxNodesPerRack, results);
results.removeAll(choosenNodes);
return getPipeline((writer==null)?localNode:writer,
results.toArray(new DatanodeDescriptor[results.size()]));
}
private DatanodeDescriptor chooseTarget(int numOfReplicas,
DatanodeDescriptor writer,
List<Node> excludedNodes,
long blocksize,
int maxNodesPerRack,
List<DatanodeDescriptor> results) {
if (numOfReplicas == 0 || clusterMap.getNumOfLeaves()==0) {
return writer;
}
int numOfResults = results.size();
boolean newBlock = (numOfResults==0);
if (writer == null && !newBlock) {
writer = (DatanodeDescriptor)results.get(0);
}
try {
switch(numOfResults) {
case 0:
writer = chooseLocalNode(writer, excludedNodes,
blocksize, maxNodesPerRack, results);
if (--numOfReplicas == 0) {
break;
}
case 1:
chooseRemoteRack(1, results.get(0), excludedNodes,
blocksize, maxNodesPerRack, results);
if (--numOfReplicas == 0) {
break;
}
case 2:
if (clusterMap.isOnSameRack(results.get(0), results.get(1))) {
chooseRemoteRack(1, results.get(0), excludedNodes,
blocksize, maxNodesPerRack, results);
} else if (newBlock){
chooseLocalRack(results.get(1), excludedNodes, blocksize,
maxNodesPerRack, results);
} else {
chooseLocalRack(writer, excludedNodes, blocksize,
maxNodesPerRack, results);
}
if (--numOfReplicas == 0) {
break;
}
default:
chooseRandom(numOfReplicas, NodeBase.ROOT, excludedNodes,
blocksize, maxNodesPerRack, results);
}
} catch (NotEnoughReplicasException e) {
FSNamesystem.LOG.warn("Not able to place enough replicas, still in need of "
+ numOfReplicas);
}
return writer;
}
private DatanodeDescriptor chooseLocalNode(
DatanodeDescriptor localMachine,
List<Node> excludedNodes,
long blocksize,
int maxNodesPerRack,
List<DatanodeDescriptor> results)
throws NotEnoughReplicasException {
if (localMachine == null)
return chooseRandom(NodeBase.ROOT, excludedNodes,
blocksize, maxNodesPerRack, results);
if (!excludedNodes.contains(localMachine)) {
excludedNodes.add(localMachine);
if (isGoodTarget(localMachine, blocksize,
maxNodesPerRack, false, results)) {
results.add(localMachine);
return localMachine;
}
}
return chooseLocalRack(localMachine, excludedNodes,
blocksize, maxNodesPerRack, results);
}
private DatanodeDescriptor chooseLocalRack(
DatanodeDescriptor localMachine,
List<Node> excludedNodes,
long blocksize,
int maxNodesPerRack,
List<DatanodeDescriptor> results)
throws NotEnoughReplicasException {
if (localMachine == null) {
return chooseRandom(NodeBase.ROOT, excludedNodes,
blocksize, maxNodesPerRack, results);
}
try {
return chooseRandom(
localMachine.getNetworkLocation(),
excludedNodes, blocksize, maxNodesPerRack, results);
} catch (NotEnoughReplicasException e1) {
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(
newLocal.getNetworkLocation(),
excludedNodes, blocksize, maxNodesPerRack, results);
} catch(NotEnoughReplicasException e2) {
return chooseRandom(NodeBase.ROOT, excludedNodes,
blocksize, maxNodesPerRack, results);
}
} else {
return chooseRandom(NodeBase.ROOT, excludedNodes,
blocksize, maxNodesPerRack, results);
}
}
}
private void chooseRemoteRack(int numOfReplicas,
DatanodeDescriptor localMachine,
List<Node> excludedNodes,
long blocksize,
int maxReplicasPerRack,
List<DatanodeDescriptor> results)
throws NotEnoughReplicasException {
int oldNumOfReplicas = results.size();
try {
chooseRandom(numOfReplicas, "~"+localMachine.getNetworkLocation(),
excludedNodes, blocksize, maxReplicasPerRack, results);
} catch (NotEnoughReplicasException e) {
chooseRandom(numOfReplicas-(results.size()-oldNumOfReplicas),
localMachine.getNetworkLocation(), excludedNodes, blocksize,
maxReplicasPerRack, results);
}
}
private DatanodeDescriptor chooseRandom(
String nodes,
List<Node> excludedNodes,
long blocksize,
int maxNodesPerRack,
List<DatanodeDescriptor> results)
throws NotEnoughReplicasException {
DatanodeDescriptor result;
do {
DatanodeDescriptor[] selectedNodes =
chooseRandom(1, nodes, excludedNodes);
if (selectedNodes.length == 0) {
throw new NotEnoughReplicasException(
"Not able to place enough replicas");
}
result = (DatanodeDescriptor)(selectedNodes[0]);
} while(!isGoodTarget(result, blocksize, maxNodesPerRack, results));
results.add(result);
return result;
}
private void chooseRandom(int numOfReplicas,
String nodes,
List<Node> excludedNodes,
long blocksize,
int maxNodesPerRack,
List<DatanodeDescriptor> results)
throws NotEnoughReplicasException {
boolean toContinue = true;
do {
DatanodeDescriptor[] selectedNodes =
chooseRandom(numOfReplicas, nodes, excludedNodes);
if (selectedNodes.length < numOfReplicas) {
toContinue = false;
}
for(int i=0; i<selectedNodes.length; i++) {
DatanodeDescriptor result = selectedNodes[i];
if (isGoodTarget(result, blocksize, maxNodesPerRack, results)) {
numOfReplicas--;
results.add(result);
}
} } while (numOfReplicas>0 && toContinue);
if (numOfReplicas>0) {
throw new NotEnoughReplicasException(
"Not able to place enough replicas");
}
}
private DatanodeDescriptor[] chooseRandom(int numOfReplicas,
String nodes,
List<Node> excludedNodes) {
List<DatanodeDescriptor> results =
new ArrayList<DatanodeDescriptor>();
int numOfAvailableNodes =
clusterMap.countNumOfAvailableNodes(nodes, excludedNodes);
numOfReplicas = (numOfAvailableNodes<numOfReplicas)?
numOfAvailableNodes:numOfReplicas;
while(numOfReplicas > 0) {
DatanodeDescriptor choosenNode =
(DatanodeDescriptor)(clusterMap.chooseRandom(nodes));
if (!excludedNodes.contains(choosenNode)) {
results.add(choosenNode);
excludedNodes.add(choosenNode);
numOfReplicas--;
}
}
return (DatanodeDescriptor[])results.toArray(
new DatanodeDescriptor[results.size()]);
}
private boolean isGoodTarget(DatanodeDescriptor node,
long blockSize, int maxTargetPerLoc,
List<DatanodeDescriptor> results) {
return isGoodTarget(node, blockSize, maxTargetPerLoc,
this.considerLoad, results);
}
private boolean isGoodTarget(DatanodeDescriptor node,
long blockSize, int maxTargetPerLoc,
boolean considerLoad,
List<DatanodeDescriptor> results) {
Log logr = FSNamesystem.LOG;
if (node.isDecommissionInProgress() || node.isDecommissioned()) {
logr.debug("Node "+NodeBase.getPath(node)+
" is not chosen because the node is (being) decommissioned");
return false;
}
long remaining = node.getRemaining() -
(node.getBlocksScheduled() * blockSize);
if (blockSize* FSConstants.MIN_BLOCKS_FOR_WRITE>remaining) {
logr.debug("Node "+NodeBase.getPath(node)+
" is not chosen because the node does not have enough space");
return false;
}
if (considerLoad) {
double avgLoad = 0;
int size = clusterMap.getNumOfLeaves();
if (size != 0) {
avgLoad = (double)fs.getTotalLoad()/size;
}
if (node.getXceiverCount() > (2.0 * avgLoad)) {
logr.debug("Node "+NodeBase.getPath(node)+
" is not chosen because the node is too busy");
return false;
}
}
String rackname = node.getNetworkLocation();
int counter=1;
for(Iterator<DatanodeDescriptor> iter = results.iterator();
iter.hasNext();) {
Node result = iter.next();
if (rackname.equals(result.getNetworkLocation())) {
counter++;
}
}
if (counter>maxTargetPerLoc) {
logr.debug("Node "+NodeBase.getPath(node)+
" is not chosen because the rack has too many chosen nodes");
return false;
}
return true;
}
private DatanodeDescriptor[] getPipeline(
DatanodeDescriptor writer,
DatanodeDescriptor[] nodes) {
if (nodes.length==0) return nodes;
synchronized(clusterMap) {
int index=0;
if (writer == null || !clusterMap.contains(writer)) {
writer = nodes[0];
}
for(;index<nodes.length; index++) {
DatanodeDescriptor shortestNode = nodes[index];
int shortestDistance = clusterMap.getDistance(writer, shortestNode);
int shortestIndex = index;
for(int i=index+1; i<nodes.length; i++) {
DatanodeDescriptor currentNode = nodes[i];
int currentDistance = clusterMap.getDistance(writer, currentNode);
if (shortestDistance>currentDistance) {
shortestDistance = currentDistance;
shortestNode = currentNode;
shortestIndex = i;
}
}
if (index != shortestIndex) {
nodes[shortestIndex] = nodes[index];
nodes[index] = shortestNode;
}
writer = shortestNode;
}
}
return nodes;
}
public static int verifyBlockPlacement(LocatedBlock lBlk,
short replication,
NetworkTopology cluster) {
int numRacks = verifyBlockPlacement(lBlk, Math.min(2,replication), cluster);
return numRacks < 0 ? 0 : numRacks;
}
public static int verifyBlockPlacement(LocatedBlock lBlk,
int minRacks,
NetworkTopology cluster) {
DatanodeInfo[] locs = lBlk.getLocations();
if (locs == null)
locs = new DatanodeInfo[0];
int numRacks = cluster.getNumOfRacks();
if(numRacks <= 1) return 0;
minRacks = Math.min(minRacks, numRacks);
Set<String> racks = new TreeSet<String>();
for (DatanodeInfo dn : locs)
racks.add(dn.getNetworkLocation());
return minRacks - racks.size();
}
}