HDFS-4698. Provide client-side metrics for remote reads, local reads, and short-circuit reads. Contributed by Colin Patrick McCabe.

git-svn-id: https://svn.apache.org/repos/asf/hadoop/common/trunk@1481121 13f79535-47bb-0310-9956-ffa450edef68

Author: atm

hadoop-hdfs-project/hadoop-hdfs/CHANGES.txt

@@ -844,6 +844,9 @@ Release 2.0.5-beta - UNRELEASED
     HDFS-4804. WARN when users set the block balanced preference percent below
     0.5 or above 1.0. (Stephen Chu via atm)
 
+    HDFS-4698. Provide client-side metrics for remote reads, local reads, and
+    short-circuit reads. (Colin Patrick McCabe via atm)
+
   OPTIMIZATIONS
 
   BUG FIXES

hadoop-hdfs-project/hadoop-hdfs/src/contrib/libwebhdfs/src/hdfs_web.c

@@ -1197,6 +1197,24 @@ int hdfsFileIsOpenForRead(hdfsFile file)
     return (file->type == INPUT);
 }
 
+int hdfsFileGetReadStatistics(hdfsFile file,
+                              struct hdfsReadStatistics **stats)
+{
+    errno = ENOTSUP;
+    return -1;
+}
+
+int64_t hdfsReadStatisticsGetRemoteBytesRead(
+                            const struct hdfsReadStatistics *stats)
+{
+  return stats->totalBytesRead - stats->totalLocalBytesRead;
+}
+
+void hdfsFileFreeReadStatistics(struct hdfsReadStatistics *stats)
+{
+    free(stats);
+}
+
 int hdfsFileIsOpenForWrite(hdfsFile file)
 {
     return (file->type == OUTPUT);

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/BlockReader.java

@@ -70,4 +70,15 @@ public interface BlockReader extends ByteBufferReadable {
    * filled or the next call will return EOF.
    */
   int readAll(byte[] buf, int offset, int len) throws IOException;
+
+  /**
+   * @return              true only if this is a local read.
+   */
+  boolean isLocal();
+  
+  /**
+   * @return              true only if this is a short-circuit read.
+   *                      All short-circuit reads are also local.
+   */
+  boolean isShortCircuit();
 }

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/BlockReaderLocal.java

@@ -531,4 +531,14 @@ public int available() throws IOException {
     // We never do network I/O in BlockReaderLocal.
     return Integer.MAX_VALUE;
   }
+
+  @Override
+  public boolean isLocal() {
+    return true;
+  }
+  
+  @Override
+  public boolean isShortCircuit() {
+    return true;
+  }
 }

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/BlockReaderLocalLegacy.java

@@ -700,4 +700,14 @@ public int available() throws IOException {
     // We never do network I/O in BlockReaderLocalLegacy.
     return Integer.MAX_VALUE;
   }
+
+  @Override
+  public boolean isLocal() {
+    return true;
+  }
+  
+  @Override
+  public boolean isShortCircuit() {
+    return true;
+  }
 }

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/DFSInputStream.java

@@ -81,7 +81,74 @@ public class DFSInputStream extends FSInputStream implements ByteBufferReadable
   private LocatedBlock currentLocatedBlock = null;
   private long pos = 0;
   private long blockEnd = -1;
+  private final ReadStatistics readStatistics = new ReadStatistics();
 
+  public static class ReadStatistics {
+    public ReadStatistics() {
+      this.totalBytesRead = 0;
+      this.totalLocalBytesRead = 0;
+      this.totalShortCircuitBytesRead = 0;
+    }
+
+    public ReadStatistics(ReadStatistics rhs) {
+      this.totalBytesRead = rhs.getTotalBytesRead();
+      this.totalLocalBytesRead = rhs.getTotalLocalBytesRead();
+      this.totalShortCircuitBytesRead = rhs.getTotalShortCircuitBytesRead();
+    }
+
+    /**
+     * @return The total bytes read.  This will always be at least as
+     * high as the other numbers, since it includes all of them.
+     */
+    public long getTotalBytesRead() {
+      return totalBytesRead;
+    }
+
+    /**
+     * @return The total local bytes read.  This will always be at least
+     * as high as totalShortCircuitBytesRead, since all short-circuit
+     * reads are also local.
+     */
+    public long getTotalLocalBytesRead() {
+      return totalLocalBytesRead;
+    }
+
+    /**
+     * @return The total short-circuit local bytes read.
+     */
+    public long getTotalShortCircuitBytesRead() {
+      return totalShortCircuitBytesRead;
+    }
+
+    /**
+     * @return The total number of bytes read which were not local.
+     */
+    public long getRemoteBytesRead() {
+      return totalBytesRead - totalLocalBytesRead;
+    }
+    
+    void addRemoteBytes(long amt) {
+      this.totalBytesRead += amt;
+    }
+
+    void addLocalBytes(long amt) {
+      this.totalBytesRead += amt;
+      this.totalLocalBytesRead += amt;
+    }
+
+    void addShortCircuitBytes(long amt) {
+      this.totalBytesRead += amt;
+      this.totalLocalBytesRead += amt;
+      this.totalShortCircuitBytesRead += amt;
+    }
+    
+    private long totalBytesRead;
+
+    private long totalLocalBytesRead;
+
+    private long totalShortCircuitBytesRead;
+  }
+  
   private final FileInputStreamCache fileInputStreamCache;
 
   /**
@@ -546,9 +613,25 @@ public synchronized int read() throws IOException {
    * strategy-agnostic.
    */
   private interface ReaderStrategy {
-    public int doRead(BlockReader blockReader, int off, int len) throws ChecksumException, IOException;
+    public int doRead(BlockReader blockReader, int off, int len,
+        ReadStatistics readStatistics) throws ChecksumException, IOException;
   }
 
+  private static void updateReadStatistics(ReadStatistics readStatistics, 
+        int nRead, BlockReader blockReader) {
+    if (nRead <= 0) return;
+    if (blockReader.isShortCircuit()) {
+      readStatistics.totalBytesRead += nRead;
+      readStatistics.totalLocalBytesRead += nRead;
+      readStatistics.totalShortCircuitBytesRead += nRead;
+    } else if (blockReader.isLocal()) {
+      readStatistics.totalBytesRead += nRead;
+      readStatistics.totalLocalBytesRead += nRead;
+    } else {
+      readStatistics.totalBytesRead += nRead;
+    }
+  }
+  
   /**
    * Used to read bytes into a byte[]
    */
@@ -560,8 +643,11 @@ public ByteArrayStrategy(byte[] buf) {
     }
 
     @Override
-    public int doRead(BlockReader blockReader, int off, int len) throws ChecksumException, IOException {      
-        return blockReader.read(buf, off, len);     
+    public int doRead(BlockReader blockReader, int off, int len,
+            ReadStatistics readStatistics) throws ChecksumException, IOException {
+        int nRead = blockReader.read(buf, off, len);
+        updateReadStatistics(readStatistics, nRead, blockReader);
+        return nRead;
     }
   }
 
@@ -575,13 +661,15 @@ private static class ByteBufferStrategy implements ReaderStrategy {
     }
 
     @Override
-    public int doRead(BlockReader blockReader, int off, int len) throws ChecksumException, IOException {
+    public int doRead(BlockReader blockReader, int off, int len,
+        ReadStatistics readStatistics) throws ChecksumException, IOException {
       int oldpos = buf.position();
       int oldlimit = buf.limit();
       boolean success = false;
       try {
         int ret = blockReader.read(buf);
         success = true;
+        updateReadStatistics(readStatistics, ret, blockReader);
         return ret;
       } finally {
         if (!success) {
@@ -613,7 +701,7 @@ private synchronized int readBuffer(ReaderStrategy reader, int off, int len,
     while (true) {
       // retry as many times as seekToNewSource allows.
       try {
-        return reader.doRead(blockReader, off, len);
+        return reader.doRead(blockReader, off, len, readStatistics);
       } catch ( ChecksumException ce ) {
         DFSClient.LOG.warn("Found Checksum error for "
             + getCurrentBlock() + " from " + currentNode
@@ -1275,4 +1363,11 @@ static class DNAddrPair {
       this.addr = addr;
     }
   }
+
+  /**
+   * Get statistics about the reads which this DFSInputStream has done.
+   */
+  public synchronized ReadStatistics getReadStatistics() {
+    return new ReadStatistics(readStatistics);
+  }
 }

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/RemoteBlockReader.java

@@ -40,6 +40,7 @@
 import org.apache.hadoop.hdfs.protocolPB.PBHelper;
 import org.apache.hadoop.hdfs.security.token.block.BlockTokenIdentifier;
 import org.apache.hadoop.io.IOUtils;
+import org.apache.hadoop.net.NetUtils;
 import org.apache.hadoop.security.token.Token;
 import org.apache.hadoop.util.DataChecksum;
 
@@ -78,6 +79,11 @@ public class RemoteBlockReader extends FSInputChecker implements BlockReader {
    * at the beginning so that the read can begin on a chunk boundary.
    */
   private final long bytesNeededToFinish;
+  
+  /**
+   * True if we are reading from a local DataNode.
+   */
+  private final boolean isLocal;
 
   private boolean eos = false;
   private boolean sentStatusCode = false;
@@ -329,6 +335,9 @@ private RemoteBlockReader(String file, String bpid, long blockId,
           checksum.getChecksumSize() > 0? checksum : null, 
           checksum.getBytesPerChecksum(),
           checksum.getChecksumSize());
+
+    this.isLocal = DFSClient.isLocalAddress(NetUtils.
+        createSocketAddr(datanodeID.getXferAddr()));
 
     this.peer = peer;
     this.datanodeID = datanodeID;
@@ -477,4 +486,14 @@ public int available() throws IOException {
     // to us without doing network I/O.
     return DFSClient.TCP_WINDOW_SIZE;
   }
+
+  @Override
+  public boolean isLocal() {
+    return isLocal;
+  }
+  
+  @Override
+  public boolean isShortCircuit() {
+    return false;
+  }
 }

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/RemoteBlockReader2.java

@@ -44,6 +44,7 @@
 import org.apache.hadoop.hdfs.protocolPB.PBHelper;
 import org.apache.hadoop.hdfs.security.token.block.BlockTokenIdentifier;
 import org.apache.hadoop.hdfs.security.token.block.InvalidBlockTokenException;
+import org.apache.hadoop.net.NetUtils;
 import org.apache.hadoop.security.token.Token;
 import org.apache.hadoop.util.DataChecksum;
 
@@ -106,6 +107,11 @@ public class RemoteBlockReader2  implements BlockReader {
    */
   private long bytesNeededToFinish;
 
+  /**
+   * True if we are reading from a local DataNode.
+   */
+  private final boolean isLocal;
+
   private final boolean verifyChecksum;
 
   private boolean sentStatusCode = false;
@@ -255,6 +261,8 @@ protected RemoteBlockReader2(String file, String bpid, long blockId,
       DataChecksum checksum, boolean verifyChecksum,
       long startOffset, long firstChunkOffset, long bytesToRead, Peer peer,
       DatanodeID datanodeID, PeerCache peerCache) {
+    this.isLocal = DFSClient.isLocalAddress(NetUtils.
+        createSocketAddr(datanodeID.getXferAddr()));
     // Path is used only for printing block and file information in debug
     this.peer = peer;
     this.datanodeID = datanodeID;
@@ -431,4 +439,14 @@ public int available() throws IOException {
     // to us without doing network I/O.
     return DFSClient.TCP_WINDOW_SIZE;
   }
+  
+  @Override
+  public boolean isLocal() {
+    return isLocal;
+  }
+  
+  @Override
+  public boolean isShortCircuit() {
+    return false;
+  }
 }

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/client/HdfsDataInputStream.java

@@ -68,4 +68,14 @@ public synchronized List<LocatedBlock> getAllBlocks() throws IOException {
   public long getVisibleLength() throws IOException {
     return ((DFSInputStream) in).getFileLength();
   }
+
+  /**
+   * Get statistics about the reads which this DFSInputStream has done.
+   * Note that because HdfsDataInputStream is buffered, these stats may
+   * be higher than you would expect just by adding up the number of
+   * bytes read through HdfsDataInputStream.
+   */
+  public synchronized DFSInputStream.ReadStatistics getReadStatistics() {
+    return ((DFSInputStream) in).getReadStatistics();
+  }
 }

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/common/JspHelper.java

@@ -214,7 +214,7 @@ public static void streamBlockInAscii(InetSocketAddress addr, String poolId,
         new ExtendedBlock(poolId, blockId, 0, genStamp), blockToken,
         offsetIntoBlock, amtToRead,  true,
         "JspHelper", TcpPeerServer.peerFromSocketAndKey(s, encryptionKey),
-        new DatanodeID(addr.getAddress().toString(),              
+        new DatanodeID(addr.getAddress().getHostAddress(),
             addr.getHostName(), poolId, addr.getPort(), 0, 0), null,
             null, null, false);