Repository: hadoop Updated Branches: refs/heads/trunk 265ed1fe8 -> a42bb1cd9
HDFS-8076. Code cleanup for DFSInputStream: use offset instead of LocatedBlock when possible. Contributed by Zhe Zhang. Project: http://git-wip-us.apache.org/repos/asf/hadoop/repo Commit: http://git-wip-us.apache.org/repos/asf/hadoop/commit/a42bb1cd Tree: http://git-wip-us.apache.org/repos/asf/hadoop/tree/a42bb1cd Diff: http://git-wip-us.apache.org/repos/asf/hadoop/diff/a42bb1cd Branch: refs/heads/trunk Commit: a42bb1cd915abe5dc33eda3c01e8c74c64f35748 Parents: 265ed1f Author: Andrew Wang <[email protected]> Authored: Wed Apr 8 15:41:48 2015 -0700 Committer: Andrew Wang <[email protected]> Committed: Wed Apr 8 15:41:48 2015 -0700 ---------------------------------------------------------------------- hadoop-hdfs-project/hadoop-hdfs/CHANGES.txt | 3 ++ .../org/apache/hadoop/hdfs/DFSInputStream.java | 40 ++++++++++---------- 2 files changed, 24 insertions(+), 19 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/hadoop/blob/a42bb1cd/hadoop-hdfs-project/hadoop-hdfs/CHANGES.txt ---------------------------------------------------------------------- diff --git a/hadoop-hdfs-project/hadoop-hdfs/CHANGES.txt b/hadoop-hdfs-project/hadoop-hdfs/CHANGES.txt index 35e9d54..852006d 100644 --- a/hadoop-hdfs-project/hadoop-hdfs/CHANGES.txt +++ b/hadoop-hdfs-project/hadoop-hdfs/CHANGES.txt @@ -396,6 +396,9 @@ Release 2.8.0 - UNRELEASED HDFS-8046. Allow better control of getContentSummary (kihwal) + HDFS-8076. Code cleanup for DFSInputStream: use offset instead of + LocatedBlock when possible. (Zhe Zhang via wang) + OPTIMIZATIONS HDFS-8026. Trace FSOutputSummer#writeChecksumChunks rather than http://git-wip-us.apache.org/repos/asf/hadoop/blob/a42bb1cd/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/DFSInputStream.java ---------------------------------------------------------------------- diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/DFSInputStream.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/DFSInputStream.java index cf8015f..a9f2746 100644 --- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/DFSInputStream.java +++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/DFSInputStream.java @@ -1045,16 +1045,16 @@ implements ByteBufferReadable, CanSetDropBehind, CanSetReadahead, return errMsgr.toString(); } - private void fetchBlockByteRange(LocatedBlock block, long start, long end, + private void fetchBlockByteRange(long blockStartOffset, long start, long end, byte[] buf, int offset, Map<ExtendedBlock, Set<DatanodeInfo>> corruptedBlockMap) throws IOException { - block = getBlockAt(block.getStartOffset()); + LocatedBlock block = getBlockAt(blockStartOffset); while (true) { DNAddrPair addressPair = chooseDataNode(block, null); try { - actualGetFromOneDataNode(addressPair, block, start, end, buf, offset, - corruptedBlockMap); + actualGetFromOneDataNode(addressPair, blockStartOffset, start, end, + buf, offset, corruptedBlockMap); return; } catch (IOException e) { // Ignore. Already processed inside the function. @@ -1064,7 +1064,7 @@ implements ByteBufferReadable, CanSetDropBehind, CanSetReadahead, } private Callable<ByteBuffer> getFromOneDataNode(final DNAddrPair datanode, - final LocatedBlock block, final long start, final long end, + final long blockStartOffset, final long start, final long end, final ByteBuffer bb, final Map<ExtendedBlock, Set<DatanodeInfo>> corruptedBlockMap, final int hedgedReadId) { @@ -1077,8 +1077,8 @@ implements ByteBufferReadable, CanSetDropBehind, CanSetReadahead, TraceScope scope = Trace.startSpan("hedgedRead" + hedgedReadId, parentSpan); try { - actualGetFromOneDataNode(datanode, block, start, end, buf, offset, - corruptedBlockMap); + actualGetFromOneDataNode(datanode, blockStartOffset, start, end, buf, + offset, corruptedBlockMap); return bb; } finally { scope.close(); @@ -1088,7 +1088,7 @@ implements ByteBufferReadable, CanSetDropBehind, CanSetReadahead, } private void actualGetFromOneDataNode(final DNAddrPair datanode, - LocatedBlock block, final long start, final long end, byte[] buf, + long blockStartOffset, final long start, final long end, byte[] buf, int offset, Map<ExtendedBlock, Set<DatanodeInfo>> corruptedBlockMap) throws IOException { DFSClientFaultInjector.get().startFetchFromDatanode(); @@ -1101,7 +1101,7 @@ implements ByteBufferReadable, CanSetDropBehind, CanSetReadahead, // start of the loop. CachingStrategy curCachingStrategy; boolean allowShortCircuitLocalReads; - block = getBlockAt(block.getStartOffset()); + LocatedBlock block = getBlockAt(blockStartOffset); synchronized(infoLock) { curCachingStrategy = cachingStrategy; allowShortCircuitLocalReads = !shortCircuitForbidden(); @@ -1189,7 +1189,7 @@ implements ByteBufferReadable, CanSetDropBehind, CanSetReadahead, * if the first read is taking longer than configured amount of * time. We then wait on which ever read returns first. */ - private void hedgedFetchBlockByteRange(LocatedBlock block, long start, + private void hedgedFetchBlockByteRange(long blockStartOffset, long start, long end, byte[] buf, int offset, Map<ExtendedBlock, Set<DatanodeInfo>> corruptedBlockMap) throws IOException { @@ -1201,7 +1201,7 @@ implements ByteBufferReadable, CanSetDropBehind, CanSetReadahead, ByteBuffer bb = null; int len = (int) (end - start + 1); int hedgedReadId = 0; - block = getBlockAt(block.getStartOffset()); + LocatedBlock block = getBlockAt(blockStartOffset); while (true) { // see HDFS-6591, this metric is used to verify/catch unnecessary loops hedgedReadOpsLoopNumForTesting++; @@ -1213,8 +1213,8 @@ implements ByteBufferReadable, CanSetDropBehind, CanSetReadahead, chosenNode = chooseDataNode(block, ignored); bb = ByteBuffer.wrap(buf, offset, len); Callable<ByteBuffer> getFromDataNodeCallable = getFromOneDataNode( - chosenNode, block, start, end, bb, corruptedBlockMap, - hedgedReadId++); + chosenNode, block.getStartOffset(), start, end, bb, + corruptedBlockMap, hedgedReadId++); Future<ByteBuffer> firstRequest = hedgedService .submit(getFromDataNodeCallable); futures.add(firstRequest); @@ -1251,8 +1251,8 @@ implements ByteBufferReadable, CanSetDropBehind, CanSetReadahead, } bb = ByteBuffer.allocate(len); Callable<ByteBuffer> getFromDataNodeCallable = getFromOneDataNode( - chosenNode, block, start, end, bb, corruptedBlockMap, - hedgedReadId++); + chosenNode, block.getStartOffset(), start, end, bb, + corruptedBlockMap, hedgedReadId++); Future<ByteBuffer> oneMoreRequest = hedgedService .submit(getFromDataNodeCallable); futures.add(oneMoreRequest); @@ -1405,11 +1405,13 @@ implements ByteBufferReadable, CanSetDropBehind, CanSetReadahead, long bytesToRead = Math.min(remaining, blk.getBlockSize() - targetStart); try { if (dfsClient.isHedgedReadsEnabled()) { - hedgedFetchBlockByteRange(blk, targetStart, targetStart + bytesToRead - - 1, buffer, offset, corruptedBlockMap); + hedgedFetchBlockByteRange(blk.getStartOffset(), targetStart, + targetStart + bytesToRead - 1, buffer, offset, + corruptedBlockMap); } else { - fetchBlockByteRange(blk, targetStart, targetStart + bytesToRead - 1, - buffer, offset, corruptedBlockMap); + fetchBlockByteRange(blk.getStartOffset(), targetStart, + targetStart + bytesToRead - 1, buffer, offset, + corruptedBlockMap); } } finally { // Check and report if any block replicas are corrupted.
