You are viewing a plain text version of this content. The canonical link for it is here.
Posted to hdfs-commits@hadoop.apache.org by sz...@apache.org on 2011/03/25 19:44:47 UTC
svn commit: r1085509 - in /hadoop/hdfs/trunk: CHANGES.txt
src/java/org/apache/hadoop/hdfs/server/datanode/BlockReceiver.java
src/java/org/apache/hadoop/hdfs/server/datanode/DataXceiver.java
Author: szetszwo
Date: Fri Mar 25 18:44:46 2011
New Revision: 1085509
URL: http://svn.apache.org/viewvc?rev=1085509&view=rev
Log:
HDFS-1785. In BlockReceiver and DataXceiver, clientName.length() is used multiple times for determining whether the source is a client or a datanode.
Modified:
hadoop/hdfs/trunk/CHANGES.txt
hadoop/hdfs/trunk/src/java/org/apache/hadoop/hdfs/server/datanode/BlockReceiver.java
hadoop/hdfs/trunk/src/java/org/apache/hadoop/hdfs/server/datanode/DataXceiver.java
Modified: hadoop/hdfs/trunk/CHANGES.txt
URL: http://svn.apache.org/viewvc/hadoop/hdfs/trunk/CHANGES.txt?rev=1085509&r1=1085508&r2=1085509&view=diff
==============================================================================
--- hadoop/hdfs/trunk/CHANGES.txt (original)
+++ hadoop/hdfs/trunk/CHANGES.txt Fri Mar 25 18:44:46 2011
@@ -21,6 +21,10 @@ Trunk (unreleased changes)
HDFS-1675. Support transferring RBW between datanodes. (szetszwo)
+ HDFS-1785. In BlockReceiver and DataXceiver, clientName.length() is used
+ multiple times for determining whether the source is a client or a
+ datanode. (szetszwo)
+
IMPROVEMENTS
HDFS-1510. Added test-patch.properties required by test-patch.sh (nigel)
Modified: hadoop/hdfs/trunk/src/java/org/apache/hadoop/hdfs/server/datanode/BlockReceiver.java
URL: http://svn.apache.org/viewvc/hadoop/hdfs/trunk/src/java/org/apache/hadoop/hdfs/server/datanode/BlockReceiver.java?rev=1085509&r1=1085508&r2=1085509&view=diff
==============================================================================
--- hadoop/hdfs/trunk/src/java/org/apache/hadoop/hdfs/server/datanode/BlockReceiver.java (original)
+++ hadoop/hdfs/trunk/src/java/org/apache/hadoop/hdfs/server/datanode/BlockReceiver.java Fri Mar 25 18:44:46 2011
@@ -36,13 +36,13 @@ import java.util.zip.Checksum;
import org.apache.commons.logging.Log;
import org.apache.hadoop.fs.FSOutputSummer;
import org.apache.hadoop.hdfs.protocol.Block;
-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.hdfs.protocol.DataTransferProtocol.BlockConstructionStage;
-import org.apache.hadoop.hdfs.protocol.DataTransferProtocol.PipelineAck;
import org.apache.hadoop.hdfs.protocol.DataTransferProtocol.PacketHeader;
+import org.apache.hadoop.hdfs.protocol.DataTransferProtocol.PipelineAck;
import org.apache.hadoop.hdfs.protocol.DataTransferProtocol.Status;
+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.hdfs.util.DataTransferThrottler;
import org.apache.hadoop.io.IOUtils;
import org.apache.hadoop.util.Daemon;
@@ -58,7 +58,6 @@ class BlockReceiver implements java.io.C
public static final Log LOG = DataNode.LOG;
static final Log ClientTraceLog = DataNode.ClientTraceLog;
- private Block block; // the block to receive
private DataInputStream in = null; // from where data are read
private DataChecksum checksum; // from where chunks of a block can be read
private OutputStream out = null; // to block file at local disk
@@ -76,28 +75,41 @@ class BlockReceiver implements java.io.C
private Daemon responder = null;
private DataTransferThrottler throttler;
private FSDataset.BlockWriteStreams streams;
- private String clientName;
- DatanodeInfo srcDataNode = null;
+ private DatanodeInfo srcDataNode = null;
private Checksum partialCrc = null;
private final DataNode datanode;
- private final BlockConstructionStage initialStage;
- final private ReplicaInPipelineInterface replicaInfo;
volatile private boolean mirrorError;
- BlockReceiver(Block block, DataInputStream in, String inAddr,
- String myAddr, BlockConstructionStage stage,
- long newGs, long minBytesRcvd, long maxBytesRcvd,
- String clientName, DatanodeInfo srcDataNode, DataNode datanode)
- throws IOException {
+ /** The client name. It is empty if a datanode is the client */
+ private final String clientname;
+ private final boolean isClient;
+ private final boolean isDatanode;
+
+ /** the block to receive */
+ private final Block block;
+ /** the replica to write */
+ private final ReplicaInPipelineInterface replicaInfo;
+ /** pipeline stage */
+ private final BlockConstructionStage initialStage;
+
+ BlockReceiver(final Block block, final DataInputStream in,
+ final String inAddr, final String myAddr,
+ final BlockConstructionStage stage,
+ final long newGs, final long minBytesRcvd, final long maxBytesRcvd,
+ final String clientname, final DatanodeInfo srcDataNode,
+ final DataNode datanode) throws IOException {
try{
this.block = block;
this.in = in;
this.inAddr = inAddr;
this.myAddr = myAddr;
- this.clientName = clientName;
this.srcDataNode = srcDataNode;
this.datanode = datanode;
-
+
+ this.clientname = clientname;
+ this.isDatanode = clientname.length() == 0;
+ this.isClient = !this.isDatanode;
+
//for datanode, we have
//1: clientName.length() == 0, and
//2: stage == null, PIPELINE_SETUP_CREATE or TRANSFER_RBW
@@ -105,7 +117,7 @@ class BlockReceiver implements java.io.C
//
// Open local disk out
//
- if (clientName.length() == 0) { //replication or move
+ if (isDatanode) { //replication or move
replicaInfo = datanode.data.createTemporary(block);
} else {
switch (stage) {
@@ -140,8 +152,8 @@ class BlockReceiver implements java.io.C
this.bytesPerChecksum = checksum.getBytesPerChecksum();
this.checksumSize = checksum.getChecksumSize();
- boolean isCreate = stage == BlockConstructionStage.PIPELINE_SETUP_CREATE
- || clientName.length() == 0;
+ final boolean isCreate = isDatanode
+ || stage == BlockConstructionStage.PIPELINE_SETUP_CREATE;
streams = replicaInfo.createStreams(isCreate,
this.bytesPerChecksum, this.checksumSize);
if (streams != null) {
@@ -526,7 +538,7 @@ class BlockReceiver implements java.io.C
* protocol includes acks and only the last datanode needs to verify
* checksum.
*/
- if (mirrorOut == null || clientName.length() == 0) {
+ if (mirrorOut == null || isDatanode) {
verifyChunks(pktBuf, dataOff, len, pktBuf, checksumOff);
}
@@ -625,7 +637,7 @@ class BlockReceiver implements java.io.C
throttler = throttlerArg;
try {
- if (clientName.length() > 0) {
+ if (isClient) {
responder = new Daemon(datanode.threadGroup,
new PacketResponder(this, block, mirrIn,
replyOut, numTargets,
@@ -649,7 +661,7 @@ class BlockReceiver implements java.io.C
// if this write is for a replication request (and not
// from a client), then finalize block. For client-writes,
// the block is finalized in the PacketResponder.
- if (clientName.length() == 0) {
+ if (isDatanode) {
// close the block/crc files
close();
@@ -688,7 +700,7 @@ class BlockReceiver implements java.io.C
* if this write is for a replication request (and not from a client)
*/
private void cleanupBlock() throws IOException {
- if (clientName.length() == 0
+ if (isDatanode
&& initialStage != BlockConstructionStage.TRANSFER_RBW) {
datanode.data.unfinalizeBlock(block);
}
@@ -925,12 +937,11 @@ class BlockReceiver implements java.io.C
block.setNumBytes(replicaInfo.getNumBytes());
datanode.data.finalizeBlock(block);
datanode.closeBlock(block, DataNode.EMPTY_DEL_HINT);
- if (ClientTraceLog.isInfoEnabled() &&
- receiver.clientName.length() > 0) {
+ if (ClientTraceLog.isInfoEnabled() && isClient) {
long offset = 0;
ClientTraceLog.info(String.format(DN_CLIENTTRACE_FORMAT,
receiver.inAddr, receiver.myAddr, block.getNumBytes(),
- "HDFS_WRITE", receiver.clientName, offset,
+ "HDFS_WRITE", receiver.clientname, offset,
datanode.dnRegistration.getStorageID(), block, endTime-startTime));
} else {
LOG.info("Received block " + block +
Modified: hadoop/hdfs/trunk/src/java/org/apache/hadoop/hdfs/server/datanode/DataXceiver.java
URL: http://svn.apache.org/viewvc/hadoop/hdfs/trunk/src/java/org/apache/hadoop/hdfs/server/datanode/DataXceiver.java?rev=1085509&r1=1085508&r2=1085509&view=diff
==============================================================================
--- hadoop/hdfs/trunk/src/java/org/apache/hadoop/hdfs/server/datanode/DataXceiver.java (original)
+++ hadoop/hdfs/trunk/src/java/org/apache/hadoop/hdfs/server/datanode/DataXceiver.java Fri Mar 25 18:44:46 2011
@@ -21,6 +21,7 @@ import static org.apache.hadoop.hdfs.pro
import static org.apache.hadoop.hdfs.protocol.DataTransferProtocol.Status.ERROR;
import static org.apache.hadoop.hdfs.protocol.DataTransferProtocol.Status.ERROR_ACCESS_TOKEN;
import static org.apache.hadoop.hdfs.protocol.DataTransferProtocol.Status.SUCCESS;
+import static org.apache.hadoop.hdfs.server.common.Util.now;
import static org.apache.hadoop.hdfs.server.datanode.DataNode.DN_CLIENTTRACE_FORMAT;
import java.io.BufferedInputStream;
@@ -36,13 +37,12 @@ import java.net.SocketException;
import org.apache.commons.logging.Log;
import org.apache.hadoop.hdfs.protocol.Block;
import org.apache.hadoop.hdfs.protocol.DataTransferProtocol;
+import org.apache.hadoop.hdfs.protocol.DataTransferProtocol.BlockConstructionStage;
import org.apache.hadoop.hdfs.protocol.DatanodeInfo;
import org.apache.hadoop.hdfs.protocol.FSConstants;
-import org.apache.hadoop.hdfs.protocol.DataTransferProtocol.BlockConstructionStage;
import org.apache.hadoop.hdfs.security.token.block.BlockTokenIdentifier;
import org.apache.hadoop.hdfs.security.token.block.BlockTokenSecretManager;
import org.apache.hadoop.hdfs.server.common.HdfsConstants;
-import static org.apache.hadoop.hdfs.server.common.Util.now;
import org.apache.hadoop.hdfs.server.datanode.FSDatasetInterface.MetaDataInputStream;
import org.apache.hadoop.io.IOUtils;
import org.apache.hadoop.io.MD5Hash;
@@ -50,8 +50,8 @@ import org.apache.hadoop.io.Text;
import org.apache.hadoop.metrics.util.MetricsTimeVaryingInt;
import org.apache.hadoop.metrics.util.MetricsTimeVaryingRate;
import org.apache.hadoop.net.NetUtils;
-import org.apache.hadoop.security.token.Token;
import org.apache.hadoop.security.token.SecretManager.InvalidToken;
+import org.apache.hadoop.security.token.Token;
import org.apache.hadoop.util.DataChecksum;
import org.apache.hadoop.util.StringUtils;
@@ -236,12 +236,15 @@ class DataXceiver extends DataTransferPr
* Write a block to disk.
*/
@Override
- protected void opWriteBlock(DataInputStream in, Block block,
- int pipelineSize, BlockConstructionStage stage,
- long newGs, long minBytesRcvd, long maxBytesRcvd,
- String client, DatanodeInfo srcDataNode, DatanodeInfo[] targets,
- Token<BlockTokenIdentifier> blockToken) throws IOException {
- updateCurrentThreadName("Receiving block " + block + " client=" + client);
+ protected void opWriteBlock(final DataInputStream in, final Block block,
+ final int pipelineSize, final BlockConstructionStage stage,
+ final long newGs, final long minBytesRcvd, final long maxBytesRcvd,
+ final String clientname, final DatanodeInfo srcDataNode,
+ final DatanodeInfo[] targets, final Token<BlockTokenIdentifier> blockToken
+ ) throws IOException {
+ updateCurrentThreadName("Receiving block " + block + " client=" + clientname);
+ final boolean isDatanode = clientname.length() == 0;
+ final boolean isClient = !isDatanode;
if (LOG.isDebugEnabled()) {
LOG.debug("writeBlock receive buf size " + s.getReceiveBufferSize() +
@@ -268,7 +271,7 @@ class DataXceiver extends DataTransferPr
BlockTokenSecretManager.AccessMode.WRITE);
} catch (InvalidToken e) {
try {
- if (client.length() != 0) {
+ if (isClient) {
ERROR_ACCESS_TOKEN.write(replyOut);
Text.writeString(replyOut, datanode.dnRegistration.getName());
replyOut.flush();
@@ -291,14 +294,14 @@ class DataXceiver extends DataTransferPr
String firstBadLink = ""; // first datanode that failed in connection setup
DataTransferProtocol.Status mirrorInStatus = SUCCESS;
try {
- if (client.length() == 0 ||
+ if (isDatanode ||
stage != BlockConstructionStage.PIPELINE_CLOSE_RECOVERY) {
// open a block receiver
blockReceiver = new BlockReceiver(block, in,
s.getRemoteSocketAddress().toString(),
s.getLocalSocketAddress().toString(),
stage, newGs, minBytesRcvd, maxBytesRcvd,
- client, srcDataNode, datanode);
+ clientname, srcDataNode, datanode);
} else {
datanode.data.recoverClose(block, newGs, minBytesRcvd);
}
@@ -329,7 +332,7 @@ class DataXceiver extends DataTransferPr
// Write header: Copied from DFSClient.java!
DataTransferProtocol.Sender.opWriteBlock(mirrorOut, originalBlock,
- pipelineSize, stage, newGs, minBytesRcvd, maxBytesRcvd, client,
+ pipelineSize, stage, newGs, minBytesRcvd, maxBytesRcvd, clientname,
srcDataNode, targets, blockToken);
if (blockReceiver != null) { // send checksum header
@@ -338,7 +341,7 @@ class DataXceiver extends DataTransferPr
mirrorOut.flush();
// read connect ack (only for clients, not for replication req)
- if (client.length() != 0) {
+ if (isClient) {
mirrorInStatus = DataTransferProtocol.Status.read(mirrorIn);
firstBadLink = Text.readString(mirrorIn);
if (LOG.isDebugEnabled() || mirrorInStatus != SUCCESS) {
@@ -350,7 +353,7 @@ class DataXceiver extends DataTransferPr
}
} catch (IOException e) {
- if (client.length() != 0) {
+ if (isClient) {
ERROR.write(replyOut);
Text.writeString(replyOut, mirrorNode);
replyOut.flush();
@@ -361,7 +364,7 @@ class DataXceiver extends DataTransferPr
mirrorIn = null;
IOUtils.closeSocket(mirrorSock);
mirrorSock = null;
- if (client.length() > 0) {
+ if (isClient) {
throw e;
} else {
LOG.info(datanode.dnRegistration + ":Exception transfering block " +
@@ -373,7 +376,7 @@ class DataXceiver extends DataTransferPr
}
// send connect ack back to source (only for clients)
- if (client.length() != 0) {
+ if (isClient) {
if (LOG.isDebugEnabled() || mirrorInStatus != SUCCESS) {
LOG.info("Datanode " + targets.length +
" forwarding connect ack to upstream firstbadlink is " +
@@ -392,7 +395,7 @@ class DataXceiver extends DataTransferPr
}
// update its generation stamp
- if (client.length() != 0 &&
+ if (isClient &&
stage == BlockConstructionStage.PIPELINE_CLOSE_RECOVERY) {
block.setGenerationStamp(newGs);
block.setNumBytes(minBytesRcvd);
@@ -401,7 +404,7 @@ class DataXceiver extends DataTransferPr
// if this write is for a replication request or recovering
// a failed close for client, then confirm block. For other client-writes,
// the block is finalized in the PacketResponder.
- if ((client.length() == 0 && stage != BlockConstructionStage.TRANSFER_RBW)
+ if ((isDatanode && stage != BlockConstructionStage.TRANSFER_RBW)
||
stage == BlockConstructionStage.PIPELINE_CLOSE_RECOVERY) {
datanode.closeBlock(block, DataNode.EMPTY_DEL_HINT);