svn merge -c 1347867 from trunk for HADOOP-8469. Make NetworkTopology class pluggable.

git-svn-id: https://svn.apache.org/repos/asf/hadoop/common/branches/branch-2@1488838 13f79535-47bb-0310-9956-ffa450edef68
This commit is contained in:
Tsz-wo Sze 2013-06-03 03:57:32 +00:00
parent f42fc8f958
commit 1d520009e3
5 changed files with 169 additions and 35 deletions

View File

@ -32,6 +32,9 @@ Release 2.1.0-beta - UNRELEASED
Azure environments. (See breakdown of tasks below for subtasks and
contributors)
HADOOP-8469. Make NetworkTopology class pluggable. (Junping Du via
szetszwo)
IMPROVEMENTS
HADOOP-9253. Capture ulimit info in the logs at service start time.

View File

@ -63,7 +63,9 @@ public class CommonConfigurationKeysPublic {
/** See <a href="{@docRoot}/../core-default.html">core-default.xml</a> */
public static final String NET_TOPOLOGY_NODE_SWITCH_MAPPING_IMPL_KEY =
"net.topology.node.switch.mapping.impl";
/** See <a href="{@docRoot}/../core-default.html">core-default.xml</a> */
public static final String NET_TOPOLOGY_IMPL_KEY =
"net.topology.impl";
/** See <a href="{@docRoot}/../core-default.html">core-default.xml</a> */
public static final String NET_TOPOLOGY_TABLE_MAPPING_FILE_KEY =
"net.topology.table.file.name";

View File

@ -20,6 +20,7 @@ package org.apache.hadoop.net;
import java.util.ArrayList;
import java.util.List;
import java.util.Collection;
import java.util.List;
import java.util.Random;
import java.util.concurrent.locks.ReadWriteLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;
@ -56,8 +57,8 @@ public class NetworkTopology {
/** InnerNode represents a switch/router of a data center or rack.
* Different from a leaf node, it has non-null children.
*/
private class InnerNode extends NodeBase {
private ArrayList<Node> children=new ArrayList<Node>();
static class InnerNode extends NodeBase {
protected List<Node> children=new ArrayList<Node>();
private int numOfLeaves;
/** Construct an InnerNode from a path-like string */
@ -77,7 +78,7 @@ public class NetworkTopology {
}
/** @return its children */
Collection<Node> getChildren() {return children;}
List<Node> getChildren() {return children;}
/** @return the number of children this node has */
int getNumOfChildren() {
@ -183,7 +184,23 @@ public class NetworkTopology {
}
}
}
/**
* Creates a parent node to be added to the list of children.
* Creates a node using the InnerNode four argument constructor specifying
* the name, location, parent, and level of this node.
*
* <p>To be overridden in subclasses for specific InnerNode implementations,
* as alternative to overriding the full {@link #add(Node)} method.
*
* @param parentName The name of the parent node
* @return A new inner node
* @see InnerNode#InnerNode(String, String, InnerNode, int)
*/
protected InnerNode createParentNode(String parentName) {
return new InnerNode(parentName, getPath(this), this, this.getLevel()+1);
}
/** Remove node <i>n</i> from the subtree of this node
* @param n node to be deleted
* @return true if the node is deleted; false otherwise
@ -264,7 +281,7 @@ public class NetworkTopology {
* @param excludedNode an excluded node (can be null)
* @return
*/
private Node getLeaf(int leafIndex, Node excludedNode) {
Node getLeaf(int leafIndex, Node excludedNode) {
int count=0;
// check if the excluded node a leaf
boolean isLeaf =
@ -309,7 +326,21 @@ public class NetworkTopology {
return null;
}
}
/**
* Determine if children a leaves, default implementation calls {@link #isRack()}
* <p>To be overridden in subclasses for specific InnerNode implementations,
* as alternative to overriding the full {@link #getLeaf(int, Node)} method.
*
* @return true if children are leaves, false otherwise
*/
protected boolean areChildrenLeaves() {
return isRack();
}
/**
* Get number of leaves.
*/
int getNumOfLeaves() {
return numOfLeaves;
}
@ -318,18 +349,18 @@ public class NetworkTopology {
/**
* the root cluster map
*/
InnerNode clusterMap = new InnerNode(InnerNode.ROOT);
InnerNode clusterMap;
/** Depth of all leaf nodes */
private int depthOfAllLeaves = -1;
/** rack counter */
private int numOfRacks = 0;
protected int numOfRacks = 0;
/** the lock used to manage access */
private ReadWriteLock netlock;
protected ReadWriteLock netlock = new ReentrantReadWriteLock();
public NetworkTopology() {
netlock = new ReentrantReadWriteLock();
clusterMap = new InnerNode(InnerNode.ROOT);
}
/** Add a leaf node
* Update node counter & rack counter if necessary
* @param node node to be added; can be null
@ -353,7 +384,7 @@ public class NetworkTopology {
"You cannot have a rack and a non-rack node at the same " +
"level of the network topology.");
}
Node rack = getNode(node.getNetworkLocation());
Node rack = getNodeForNetworkLocation(node);
if (rack != null && !(rack instanceof InnerNode)) {
throw new IllegalArgumentException("Unexpected data node "
+ node.toString()
@ -377,7 +408,26 @@ public class NetworkTopology {
netlock.writeLock().unlock();
}
}
/**
* Return a reference to the node given its string representation.
* Default implementation delegates to {@link #getNode(String)}.
*
* <p>To be overridden in subclasses for specific NetworkTopology
* implementations, as alternative to overriding the full {@link #add(Node)}
* method.
*
* @param node The string representation of this node's network location is
* used to retrieve a Node object.
* @return a reference to the node; null if the node is not in the tree
*
* @see #add(Node)
* @see #getNode(String)
*/
protected Node getNodeForNetworkLocation(Node node) {
return getNode(node.getNetworkLocation());
}
/**
* Given a string representation of a rack, return its children
* @param loc a path-like string representation of a rack
@ -426,7 +476,7 @@ public class NetworkTopology {
netlock.writeLock().unlock();
}
}
/** Check if the tree contains node <i>node</i>
*
* @param node a node
@ -466,7 +516,21 @@ public class NetworkTopology {
netlock.readLock().unlock();
}
}
/** Given a string representation of a rack for a specific network
* location
*
* To be overridden in subclasses for specific NetworkTopology
* implementations, as alternative to overriding the full
* {@link #getRack(String)} method.
* @param loc
* a path-like string representation of a network location
* @return a rack string
*/
public String getRack(String loc) {
return loc;
}
/** @return the total number of racks */
public int getNumOfRacks() {
netlock.readLock().lock();
@ -476,7 +540,7 @@ public class NetworkTopology {
netlock.readLock().unlock();
}
}
/** @return the total number of leaf nodes */
public int getNumOfLeaves() {
netlock.readLock().lock();
@ -486,7 +550,7 @@ public class NetworkTopology {
netlock.readLock().unlock();
}
}
/** Return the distance between two nodes
* It is assumed that the distance from one node to its parent is 1
* The distance between two nodes is calculated by summing up their distances
@ -532,8 +596,8 @@ public class NetworkTopology {
return Integer.MAX_VALUE;
}
return dis+2;
}
}
/** Check if two nodes are on the same rack
* @param node1 one node (can be null)
* @param node2 another node (can be null)
@ -548,13 +612,44 @@ public class NetworkTopology {
netlock.readLock().lock();
try {
return node1.getParent()==node2.getParent();
return isSameParents(node1, node2);
} finally {
netlock.readLock().unlock();
}
}
final private static Random r = new Random();
/**
* Check if network topology is aware of NodeGroup
*/
public boolean isNodeGroupAware() {
return false;
}
/**
* Return false directly as not aware of NodeGroup, to be override in sub-class
*/
public boolean isOnSameNodeGroup(Node node1, Node node2) {
return false;
}
/**
* Compare the parents of each node for equality
*
* <p>To be overridden in subclasses for specific NetworkTopology
* implementations, as alternative to overriding the full
* {@link #isOnSameRack(Node, Node)} method.
*
* @param node1 the first node to compare
* @param node2 the second node to compare
* @return true if their parents are equal, false otherwise
*
* @see #isOnSameRack(Node, Node)
*/
protected boolean isSameParents(Node node1, Node node2) {
return node1.getParent()==node2.getParent();
}
final protected static Random r = new Random();
/** randomly choose one node from <i>scope</i>
* if scope starts with ~, choose one from the all nodes except for the
* ones in <i>scope</i>; otherwise, choose one from <i>scope</i>
@ -573,7 +668,7 @@ public class NetworkTopology {
netlock.readLock().unlock();
}
}
private Node chooseRandom(String scope, String excludedScope){
if (excludedScope != null) {
if (scope.startsWith(excludedScope)) {
@ -602,7 +697,25 @@ public class NetworkTopology {
int leaveIndex = r.nextInt(numOfDatanodes);
return innerNode.getLeaf(leaveIndex, node);
}
/** return leaves in <i>scope</i>
* @param scope a path string
* @return leaves nodes under specific scope
*/
public List<Node> getLeaves(String scope) {
Node node = getNode(scope);
List<Node> leafNodes = new ArrayList<Node>();
if (!(node instanceof InnerNode)) {
leafNodes.add(node);
} else {
InnerNode innerNode = (InnerNode) node;
for (int i=0;i<innerNode.getNumOfLeaves();i++) {
leafNodes.add(innerNode.getLeaf(i, null));
}
}
return leafNodes;
}
/** return the number of leaves in <i>scope</i> but not in <i>excludedNodes</i>
* if scope starts with ~, return the number of nodes that are not
* in <i>scope</i> and <i>excludedNodes</i>;
@ -642,7 +755,7 @@ public class NetworkTopology {
netlock.readLock().unlock();
}
}
/** convert a network tree to a string */
@Override
public String toString() {
@ -664,13 +777,12 @@ public class NetworkTopology {
return tree.toString();
}
/* swap two array items */
static private void swap(Node[] nodes, int i, int j) {
/** swap two array items */
static protected void swap(Node[] nodes, int i, int j) {
Node tempNode;
tempNode = nodes[j];
nodes[j] = nodes[i];
nodes[i] = tempNode;
}
/** Sort nodes array by their distances to <i>reader</i>
@ -721,4 +833,5 @@ public class NetworkTopology {
swap(nodes, 0, r.nextInt(nodes.length));
}
}
}

View File

@ -204,10 +204,10 @@ public class DatanodeInfo extends DatanodeID implements Node {
this.xceiverCount = xceiverCount;
}
/** rack name */
/** network location */
public synchronized String getNetworkLocation() {return location;}
/** Sets the rack name */
/** Sets the network location */
public synchronized void setNetworkLocation(String location) {
this.location = NodeBase.normalize(location);
}

View File

@ -39,6 +39,7 @@ import org.apache.hadoop.HadoopIllegalArgumentException;
import org.apache.hadoop.classification.InterfaceAudience;
import org.apache.hadoop.classification.InterfaceStability;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.CommonConfigurationKeysPublic;
import org.apache.hadoop.hdfs.DFSConfigKeys;
import org.apache.hadoop.hdfs.DFSUtil;
import org.apache.hadoop.hdfs.HdfsConfiguration;
@ -113,7 +114,7 @@ public class DatanodeManager {
= new TreeMap<String, DatanodeDescriptor>();
/** Cluster network topology */
private final NetworkTopology networktopology = new NetworkTopology();
private final NetworkTopology networktopology;
/** Host names to datanode descriptors mapping. */
private final Host2NodesMap host2DatanodeMap = new Host2NodesMap();
@ -161,6 +162,12 @@ public class DatanodeManager {
final Configuration conf) throws IOException {
this.namesystem = namesystem;
this.blockManager = blockManager;
Class<? extends NetworkTopology> networkTopologyClass =
conf.getClass(CommonConfigurationKeysPublic.NET_TOPOLOGY_IMPL_KEY,
NetworkTopology.class, NetworkTopology.class);
networktopology = (NetworkTopology) ReflectionUtils.newInstance(
networkTopologyClass, conf);
this.heartbeatManager = new HeartbeatManager(namesystem, blockManager, conf);
@ -291,7 +298,16 @@ public class DatanodeManager {
public void sortLocatedBlocks(final String targethost,
final List<LocatedBlock> locatedblocks) {
//sort the blocks
final DatanodeDescriptor client = getDatanodeByHost(targethost);
// As it is possible for the separation of node manager and datanode,
// here we should get node but not datanode only .
Node client = getDatanodeByHost(targethost);
if (client == null) {
List<String> hosts = new ArrayList<String> (1);
hosts.add(targethost);
String rName = dnsToSwitchMapping.resolve(hosts).get(0);
if (rName != null)
client = new NodeBase(rName + NodeBase.PATH_SEPARATOR_STR + targethost);
}
Comparator<DatanodeInfo> comparator = avoidStaleDataNodesForRead ?
new DFSUtil.DecomStaleComparator(staleInterval) :
@ -301,7 +317,7 @@ public class DatanodeManager {
networktopology.pseudoSortByDistance(client, b.getLocations());
// Move decommissioned/stale datanodes to the bottom
Arrays.sort(b.getLocations(), comparator);
}
}
}
CyclicIteration<String, DatanodeDescriptor> getDatanodeCyclicIteration(