HDFS-8076. Code cleanup for DFSInputStream: use offset instead of Loc…

…atedBlock when possible. Contributed by Zhe Zhang.
apache · Apr 8, 2015 · a42bb1c · a42bb1c
1 parent 265ed1f
commit a42bb1c
Show file tree

Hide file tree

Showing 2 changed files with 24 additions and 19 deletions.
diff --git 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

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
@@ -1045,16 +1045,16 @@ private static String getBestNodeDNAddrPairErrorString(
     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 @@ private void fetchBlockByteRange(LocatedBlock block, long start, long end,
   }
 
   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 @@ public ByteBuffer call() throws Exception {
         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 @@ public ByteBuffer call() throws Exception {
   }
 
   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 @@ private void actualGetFromOneDataNode(final DNAddrPair datanode,
       // 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 @@ private void actualGetFromOneDataNode(final DNAddrPair datanode,
    * 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 @@ private void hedgedFetchBlockByteRange(LocatedBlock block, long start,
     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 @@ private void hedgedFetchBlockByteRange(LocatedBlock block, long start,
         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 @@ private void hedgedFetchBlockByteRange(LocatedBlock block, long start,
           }
           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 @@ private int pread(long position, byte[] buffer, int offset, int length)
       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.