HDFS-3246: pRead equivalent for direct read path (apache#597)

HDFS-3246: pRead equivalent for direct read path

Contributed by Sahil Takiar

Author: deepakdamri

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/crypto/CryptoInputStream.java

@@ -33,18 +33,7 @@
 import com.google.common.base.Preconditions;
 import org.apache.hadoop.classification.InterfaceAudience;
 import org.apache.hadoop.classification.InterfaceStability;
-import org.apache.hadoop.fs.ByteBufferReadable;
-import org.apache.hadoop.fs.CanSetDropBehind;
-import org.apache.hadoop.fs.CanSetReadahead;
-import org.apache.hadoop.fs.CanUnbuffer;
-import org.apache.hadoop.fs.FSExceptionMessages;
-import org.apache.hadoop.fs.HasEnhancedByteBufferAccess;
-import org.apache.hadoop.fs.HasFileDescriptor;
-import org.apache.hadoop.fs.PositionedReadable;
-import org.apache.hadoop.fs.ReadOption;
-import org.apache.hadoop.fs.Seekable;
-import org.apache.hadoop.fs.StreamCapabilities;
-import org.apache.hadoop.fs.StreamCapabilitiesPolicy;
+import org.apache.hadoop.fs.*;
 import org.apache.hadoop.io.ByteBufferPool;
 import org.apache.hadoop.util.StringUtils;
 
@@ -64,7 +53,7 @@
 public class CryptoInputStream extends FilterInputStream implements 
     Seekable, PositionedReadable, ByteBufferReadable, HasFileDescriptor, 
     CanSetDropBehind, CanSetReadahead, HasEnhancedByteBufferAccess, 
-    ReadableByteChannel, CanUnbuffer, StreamCapabilities {
+    ReadableByteChannel, CanUnbuffer, StreamCapabilities{
   private final byte[] oneByteBuf = new byte[1];
   private final CryptoCodec codec;
   private final Decryptor decryptor;
@@ -565,6 +554,51 @@ private void decrypt(ByteBuffer buf, int n, int start)
     }
     buf.position(pos);
   }
+
+  private void decrypt(long filePosition, ByteBuffer buf, int length, int start)
+          throws IOException {
+    ByteBuffer localInBuffer = null;
+    ByteBuffer localOutBuffer = null;
+
+    // Duplicate the buffer so we don't have to worry about resetting the
+    // original position and limit at the end of the method
+    buf = buf.duplicate();
+
+    int decryptedBytes = 0;
+    Decryptor localDecryptor = null;
+    try {
+      localInBuffer = getBuffer();
+      localOutBuffer = getBuffer();
+      localDecryptor = getDecryptor();
+      byte[] localIV = initIV.clone();
+      updateDecryptor(localDecryptor, filePosition, localIV);
+      byte localPadding = getPadding(filePosition);
+      // Set proper filePosition for inputdata.
+      localInBuffer.position(localPadding);
+
+      while (decryptedBytes < length) {
+        buf.position(start + decryptedBytes);
+        buf.limit(start + decryptedBytes +
+                Math.min(length - decryptedBytes, localInBuffer.remaining()));
+        localInBuffer.put(buf);
+        // Do decryption
+        try {
+          decrypt(localDecryptor, localInBuffer, localOutBuffer, localPadding);
+          buf.position(start + decryptedBytes);
+          buf.limit(start + length);
+          decryptedBytes += localOutBuffer.remaining();
+          buf.put(localOutBuffer);
+        } finally {
+          localPadding = afterDecryption(localDecryptor, localInBuffer,
+                                         filePosition + length, localIV);
+        }
+      }
+    } finally {
+      returnBuffer(localInBuffer);
+      returnBuffer(localOutBuffer);
+      returnDecryptor(localDecryptor);
+    }
+  }
 
   @Override
   public int available() throws IOException {

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/FSDataInputStream.java

@@ -41,29 +41,29 @@ public class FSDataInputStream extends DataInputStream
       HasEnhancedByteBufferAccess, CanUnbuffer, StreamCapabilities,
       ByteBufferPositionedReadable {
   /**
-   * Map ByteBuffers that we have handed out to readers to ByteBufferPool 
+   * Map ByteBuffers that we have handed out to readers to ByteBufferPool
    * objects
    */
   private final IdentityHashStore<ByteBuffer, ByteBufferPool>
-    extendedReadBuffers
-      = new IdentityHashStore<ByteBuffer, ByteBufferPool>(0);
+          extendedReadBuffers
+          = new IdentityHashStore<ByteBuffer, ByteBufferPool>(0);
 
   public FSDataInputStream(InputStream in) {
     super(in);
-    if( !(in instanceof Seekable) || !(in instanceof PositionedReadable) ) {
+    if (!(in instanceof Seekable) || !(in instanceof PositionedReadable)) {
       throw new IllegalArgumentException(in.getClass().getCanonicalName() +
-          " is not an instance of Seekable or PositionedReadable");
+              " is not an instance of Seekable or PositionedReadable");
     }
   }
-  
+
   /**
    * Seek to the given offset.
    *
    * @param desired offset to seek to
    */
   @Override
   public void seek(long desired) throws IOException {
-    ((Seekable)in).seek(desired);
+    ((Seekable) in).seek(desired);
   }
 
   /**
@@ -73,65 +73,65 @@ public void seek(long desired) throws IOException {
    */
   @Override
   public long getPos() throws IOException {
-    return ((Seekable)in).getPos();
+    return ((Seekable) in).getPos();
   }
-  
+
   /**
    * Read bytes from the given position in the stream to the given buffer.
    *
-   * @param position  position in the input stream to seek
-   * @param buffer    buffer into which data is read
-   * @param offset    offset into the buffer in which data is written
-   * @param length    maximum number of bytes to read
+   * @param position position in the input stream to seek
+   * @param buffer   buffer into which data is read
+   * @param offset   offset into the buffer in which data is written
+   * @param length   maximum number of bytes to read
    * @return total number of bytes read into the buffer, or <code>-1</code>
-   *         if there is no more data because the end of the stream has been
-   *         reached
+   * if there is no more data because the end of the stream has been
+   * reached
    */
   @Override
   public int read(long position, byte[] buffer, int offset, int length)
-    throws IOException {
-    return ((PositionedReadable)in).read(position, buffer, offset, length);
+          throws IOException {
+    return ((PositionedReadable) in).read(position, buffer, offset, length);
   }
 
   /**
    * Read bytes from the given position in the stream to the given buffer.
    * Continues to read until <code>length</code> bytes have been read.
    *
-   * @param position  position in the input stream to seek
-   * @param buffer    buffer into which data is read
-   * @param offset    offset into the buffer in which data is written
-   * @param length    the number of bytes to read
-   * @throws IOException IO problems
+   * @param position position in the input stream to seek
+   * @param buffer   buffer into which data is read
+   * @param offset   offset into the buffer in which data is written
+   * @param length   the number of bytes to read
+   * @throws IOException  IO problems
    * @throws EOFException If the end of stream is reached while reading.
    *                      If an exception is thrown an undetermined number
-   *                      of bytes in the buffer may have been written. 
+   *                      of bytes in the buffer may have been written.
    */
   @Override
   public void readFully(long position, byte[] buffer, int offset, int length)
-    throws IOException {
-    ((PositionedReadable)in).readFully(position, buffer, offset, length);
+          throws IOException {
+    ((PositionedReadable) in).readFully(position, buffer, offset, length);
   }
-  
+
   /**
    * See {@link #readFully(long, byte[], int, int)}.
    */
   @Override
   public void readFully(long position, byte[] buffer)
-    throws IOException {
-    ((PositionedReadable)in).readFully(position, buffer, 0, buffer.length);
+          throws IOException {
+    ((PositionedReadable) in).readFully(position, buffer, 0, buffer.length);
   }
-  
+
   /**
    * Seek to the given position on an alternate copy of the data.
    *
-   * @param  targetPos  position to seek to
+   * @param targetPos position to seek to
    * @return true if a new source is found, false otherwise
    */
   @Override
   public boolean seekToNewSource(long targetPos) throws IOException {
-    return ((Seekable)in).seekToNewSource(targetPos); 
+    return ((Seekable) in).seekToNewSource(targetPos);
   }
-  
+
   /**
    * Get a reference to the wrapped input stream. Used by unit tests.
    *
@@ -145,7 +145,7 @@ public InputStream getWrappedStream() {
   @Override
   public int read(ByteBuffer buf) throws IOException {
     if (in instanceof ByteBufferReadable) {
-      return ((ByteBufferReadable)in).read(buf);
+      return ((ByteBufferReadable) in).read(buf);
     }
 
     throw new UnsupportedOperationException("Byte-buffer read unsupported " +
@@ -165,37 +165,36 @@ public FileDescriptor getFileDescriptor() throws IOException {
 
   @Override
   public void setReadahead(Long readahead)
-      throws IOException, UnsupportedOperationException {
+          throws IOException, UnsupportedOperationException {
     try {
-      ((CanSetReadahead)in).setReadahead(readahead);
+      ((CanSetReadahead) in).setReadahead(readahead);
     } catch (ClassCastException e) {
       throw new UnsupportedOperationException(in.getClass().getCanonicalName() +
-          " does not support setting the readahead caching strategy.");
+              " does not support setting the readahead caching strategy.");
     }
   }
 
   @Override
   public void setDropBehind(Boolean dropBehind)
-      throws IOException, UnsupportedOperationException {
+          throws IOException, UnsupportedOperationException {
     try {
-      ((CanSetDropBehind)in).setDropBehind(dropBehind);
+      ((CanSetDropBehind) in).setDropBehind(dropBehind);
     } catch (ClassCastException e) {
       throw new UnsupportedOperationException("this stream does not " +
-          "support setting the drop-behind caching setting.");
+              "support setting the drop-behind caching setting.");
     }
   }
 
   @Override
   public ByteBuffer read(ByteBufferPool bufferPool, int maxLength,
-      EnumSet<ReadOption> opts) 
+                         EnumSet<ReadOption> opts)
           throws IOException, UnsupportedOperationException {
     try {
-      return ((HasEnhancedByteBufferAccess)in).read(bufferPool,
-          maxLength, opts);
-    }
-    catch (ClassCastException e) {
+      return ((HasEnhancedByteBufferAccess) in).read(bufferPool,
+              maxLength, opts);
+    } catch (ClassCastException e) {
       ByteBuffer buffer = ByteBufferUtil.
-          fallbackRead(this, bufferPool, maxLength);
+              fallbackRead(this, bufferPool, maxLength);
       if (buffer != null) {
         extendedReadBuffers.put(buffer, bufferPool);
       }
@@ -204,23 +203,22 @@ public ByteBuffer read(ByteBufferPool bufferPool, int maxLength,
   }
 
   private static final EnumSet<ReadOption> EMPTY_READ_OPTIONS_SET =
-      EnumSet.noneOf(ReadOption.class);
+          EnumSet.noneOf(ReadOption.class);
 
   final public ByteBuffer read(ByteBufferPool bufferPool, int maxLength)
           throws IOException, UnsupportedOperationException {
     return read(bufferPool, maxLength, EMPTY_READ_OPTIONS_SET);
   }
-  
+
   @Override
   public void releaseBuffer(ByteBuffer buffer) {
     try {
-      ((HasEnhancedByteBufferAccess)in).releaseBuffer(buffer);
-    }
-    catch (ClassCastException e) {
-      ByteBufferPool bufferPool = extendedReadBuffers.remove( buffer);
+      ((HasEnhancedByteBufferAccess) in).releaseBuffer(buffer);
+    } catch (ClassCastException e) {
+      ByteBufferPool bufferPool = extendedReadBuffers.remove(buffer);
       if (bufferPool == null) {
         throw new IllegalArgumentException("tried to release a buffer " +
-            "that was not created by this stream.");
+                "that was not created by this stream.");
       }
       bufferPool.putBuffer(buffer);
     }
@@ -241,6 +239,7 @@ public boolean hasCapability(String capability) {
 
   /**
    * String value. Includes the string value of the inner stream
+   *
    * @return the stream
    */
   @Override
@@ -254,17 +253,7 @@ public int read(long position, ByteBuffer buf) throws IOException {
       return ((ByteBufferPositionedReadable) in).read(position, buf);
     }
     throw new UnsupportedOperationException("Byte-buffer pread unsupported " +
-        "by " + in.getClass().getCanonicalName());
-  }
-
-  @Override
-  public void readFully(long position, ByteBuffer buf) throws IOException {
-    if (in instanceof ByteBufferPositionedReadable) {
-      ((ByteBufferPositionedReadable) in).readFully(position, buf);
-    } else {
-      throw new UnsupportedOperationException("Byte-buffer pread " +
-              "unsupported by " + in.getClass().getCanonicalName());
-    }
+            "by " + in.getClass().getCanonicalName());
   }
 
   @Override

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/StreamCapabilities.java

@@ -59,6 +59,18 @@ public interface StreamCapabilities {
    */
   String UNBUFFER = "in:unbuffer";
 
+  /**
+   * Stream read(ByteBuffer) capability implemented by
+   * {@link ByteBufferReadable#read(java.nio.ByteBuffer)}.
+   */
+  String READBYTEBUFFER = "in:readbytebuffer";
+
+  /**
+   * Stream read(long, ByteBuffer) capability implemented by
+   * {@link ByteBufferPositionedReadable#read(long, java.nio.ByteBuffer)}.
+   */
+  String PREADBYTEBUFFER = "in:preadbytebuffer";
+
   /**
    * Capabilities that a stream can support and be queried for.
    */