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

This reverts commit 57202cc.

Author: harshith-212

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

@@ -566,52 +566,7 @@ 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 {
     checkStream();

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

@@ -19,7 +19,6 @@
 package org.apache.hadoop.fs;
 
 import java.io.DataInputStream;
-import java.io.EOFException;
 import java.io.FileDescriptor;
 import java.io.FileInputStream;
 import java.io.IOException;
@@ -267,4 +266,4 @@ public void readFully(long position, ByteBuffer buf) throws IOException {
               "unsupported by " + in.getClass().getCanonicalName());
     }
   }
-}
+}

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

@@ -59,18 +59,6 @@ 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.
    */

hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/crypto/CryptoStreamsTestBase.java

@@ -26,7 +26,6 @@
 import java.util.EnumSet;
 import java.util.Random;
 
-import org.apache.hadoop.fs.ByteBufferPositionedReadable;
 import org.apache.hadoop.fs.ByteBufferReadable;
 import org.apache.hadoop.fs.CanUnbuffer;
 import org.apache.hadoop.fs.FSDataOutputStream;
@@ -130,32 +129,6 @@ private void preadCheck(PositionedReadable in) throws Exception {
     Assert.assertArrayEquals(result, expectedData);
   }
 
-  private int byteBufferPreadAll(ByteBufferPositionedReadable in,
-                                 ByteBuffer buf) throws IOException {
-    int n = 0;
-    int total = 0;
-    while (n != -1) {
-      total += n;
-      if (!buf.hasRemaining()) {
-        break;
-      }
-      n = in.read(total, buf);
-    }
-
-    return total;
-  }
-
-  private void byteBufferPreadCheck(ByteBufferPositionedReadable in)
-          throws Exception {
-    ByteBuffer result = ByteBuffer.allocate(dataLen);
-    int n = byteBufferPreadAll(in, result);
-
-    Assert.assertEquals(dataLen, n);
-    ByteBuffer expectedData = ByteBuffer.allocate(n);
-    expectedData.put(data, 0, n);
-    Assert.assertArrayEquals(result.array(), expectedData.array());
-  }
-
   protected OutputStream getOutputStream(int bufferSize) throws IOException {
     return getOutputStream(bufferSize, key, iv);
   }
@@ -315,36 +288,20 @@ private int readAll(InputStream in, long pos, byte[] b, int off, int len)
 
     return total;
   }
-
-  private int readAll(InputStream in, long pos, ByteBuffer buf)
-      throws IOException {
-    int n = 0;
-    int total = 0;
-    while (n != -1) {
-      total += n;
-      if (!buf.hasRemaining()) {
-        break;
-      }
-      n = ((ByteBufferPositionedReadable) in).read(pos + total, buf);
-    }
-
-    return total;
-  }
 
   /** Test positioned read. */
   @Test(timeout=120000)
   public void testPositionedRead() throws Exception {
-    try (OutputStream out = getOutputStream(defaultBufferSize)) {
-      writeData(out);
-    }
+    OutputStream out = getOutputStream(defaultBufferSize);
+    writeData(out);
 
-    try (InputStream in = getInputStream(defaultBufferSize)) {
-      // Pos: 1/3 dataLen
-      positionedReadCheck(in, dataLen / 3);
+    InputStream in = getInputStream(defaultBufferSize);
+    // Pos: 1/3 dataLen
+    positionedReadCheck(in , dataLen / 3);
 
-      // Pos: 1/2 dataLen
-      positionedReadCheck(in, dataLen / 2);
-    }
+    // Pos: 1/2 dataLen
+    positionedReadCheck(in, dataLen / 2);
+    in.close();
   }
 
   private void positionedReadCheck(InputStream in, int pos) throws Exception {
@@ -358,35 +315,6 @@ private void positionedReadCheck(InputStream in, int pos) throws Exception {
     System.arraycopy(data, pos, expectedData, 0, n);
     Assert.assertArrayEquals(readData, expectedData);
   }
-
-  /** Test positioned read with ByteBuffers. */
-  @Test(timeout=120000)
-  public void testPositionedReadWithByteBuffer() throws Exception {
-    try (OutputStream out = getOutputStream(defaultBufferSize)) {
-      writeData(out);
-    }
-
-    try (InputStream in = getInputStream(defaultBufferSize)) {
-      // Pos: 1/3 dataLen
-      positionedReadCheckWithByteBuffer(in, dataLen / 3);
-
-      // Pos: 1/2 dataLen
-      positionedReadCheckWithByteBuffer(in, dataLen / 2);
-    }
-  }
-
-  private void positionedReadCheckWithByteBuffer(InputStream in, int pos)
-          throws Exception {
-    ByteBuffer result = ByteBuffer.allocate(dataLen);
-    int n = readAll(in, pos, result);
-
-    Assert.assertEquals(dataLen, n + pos);
-    byte[] readData = new byte[n];
-    System.arraycopy(result.array(), 0, readData, 0, n);
-    byte[] expectedData = new byte[n];
-    System.arraycopy(data, pos, expectedData, 0, n);
-    Assert.assertArrayEquals(readData, expectedData);
-  }
 
   /** Test read fully. */
   @Test(timeout=120000)
@@ -630,40 +558,12 @@ private void byteBufferReadCheck(InputStream in, ByteBuffer buf,
     System.arraycopy(data, 0, expectedData, 0, n);
     Assert.assertArrayEquals(readData, expectedData);
   }
-
-  private void byteBufferPreadCheck(InputStream in, ByteBuffer buf,
-      int bufPos) throws Exception {
-    // Test reading from position 0
-    buf.position(bufPos);
-    int n = ((ByteBufferPositionedReadable) in).read(0, buf);
-    Assert.assertEquals(bufPos + n, buf.position());
-    byte[] readData = new byte[n];
-    buf.rewind();
-    buf.position(bufPos);
-    buf.get(readData);
-    byte[] expectedData = new byte[n];
-    System.arraycopy(data, 0, expectedData, 0, n);
-    Assert.assertArrayEquals(readData, expectedData);
-
-    // Test reading from half way through the data
-    buf.position(bufPos);
-    n = ((ByteBufferPositionedReadable) in).read(dataLen / 2, buf);
-    Assert.assertEquals(bufPos + n, buf.position());
-    readData = new byte[n];
-    buf.rewind();
-    buf.position(bufPos);
-    buf.get(readData);
-    expectedData = new byte[n];
-    System.arraycopy(data, dataLen / 2, expectedData, 0, n);
-    Assert.assertArrayEquals(readData, expectedData);
-  }
 
   /** Test byte buffer read with different buffer size. */
   @Test(timeout=120000)
   public void testByteBufferRead() throws Exception {
-    try (OutputStream out = getOutputStream(defaultBufferSize)) {
-      writeData(out);
-    }
+    OutputStream out = getOutputStream(defaultBufferSize);
+    writeData(out);
 
     // Default buffer size, initial buffer position is 0
     InputStream in = getInputStream(defaultBufferSize);
@@ -713,53 +613,6 @@ public void testByteBufferRead() throws Exception {
     byteBufferReadCheck(in, buf, 11);
     in.close();
   }
-
-  /** Test byte buffer pread with different buffer size. */
-  @Test(timeout=120000)
-  public void testByteBufferPread() throws Exception {
-    try (OutputStream out = getOutputStream(defaultBufferSize)) {
-      writeData(out);
-    }
-
-    try (InputStream defaultBuf = getInputStream(defaultBufferSize);
-         InputStream smallBuf = getInputStream(smallBufferSize)) {
-
-      ByteBuffer buf = ByteBuffer.allocate(dataLen + 100);
-
-      // Default buffer size, initial buffer position is 0
-      byteBufferPreadCheck(defaultBuf, buf, 0);
-
-      // Default buffer size, initial buffer position is not 0
-      buf.clear();
-      byteBufferPreadCheck(defaultBuf, buf, 11);
-
-      // Small buffer size, initial buffer position is 0
-      buf.clear();
-      byteBufferPreadCheck(smallBuf, buf, 0);
-
-      // Small buffer size, initial buffer position is not 0
-      buf.clear();
-      byteBufferPreadCheck(smallBuf, buf, 11);
-
-      // Test with direct ByteBuffer
-      buf = ByteBuffer.allocateDirect(dataLen + 100);
-
-      // Direct buffer, default buffer size, initial buffer position is 0
-      byteBufferPreadCheck(defaultBuf, buf, 0);
-
-      // Direct buffer, default buffer size, initial buffer position is not 0
-      buf.clear();
-      byteBufferPreadCheck(defaultBuf, buf, 11);
-
-      // Direct buffer, small buffer size, initial buffer position is 0
-      buf.clear();
-      byteBufferPreadCheck(smallBuf, buf, 0);
-
-      // Direct buffer, small buffer size, initial buffer position is not 0
-      buf.clear();
-      byteBufferPreadCheck(smallBuf, buf, 11);
-    }
-  }
 
   @Test(timeout=120000)
   public void testCombinedOp() throws Exception {
@@ -997,23 +850,5 @@ public void testUnbuffer() throws Exception {
         // The close will be called when exiting this try-with-resource block
       }
     }
-
-    // Test ByteBuffer pread
-    try (InputStream in = getInputStream(smallBufferSize)) {
-      if (in instanceof ByteBufferPositionedReadable) {
-        ByteBufferPositionedReadable bbpin = (ByteBufferPositionedReadable) in;
-
-        // Test unbuffer after pread
-        byteBufferPreadCheck(bbpin);
-        ((CanUnbuffer) in).unbuffer();
-
-        // Test pread again after unbuffer
-        byteBufferPreadCheck(bbpin);
-
-        // Test close after unbuffer
-        ((CanUnbuffer) in).unbuffer();
-        // The close will be called when exiting this try-with-resource block
-      }
-    }
   }
 }

hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/crypto/TestCryptoStreams.java

@@ -429,8 +429,6 @@ public boolean hasCapability(String capability) {
       case StreamCapabilities.READAHEAD:
       case StreamCapabilities.DROPBEHIND:
       case StreamCapabilities.UNBUFFER:
-      case StreamCapabilities.READBYTEBUFFER:
-      case StreamCapabilities.PREADBYTEBUFFER:
         return true;
       default:
         return false;

hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/crypto/TestCryptoStreamsNormal.java

@@ -105,32 +105,27 @@ public void testByteBufferReadFully() throws Exception {}
   @Override
   @Test(timeout=10000)
   public void testReadFully() throws IOException {}
-
+  
   @Ignore("Wrapped stream doesn't support Seek")
   @Override
   @Test(timeout=10000)
   public void testSeek() throws IOException {}
-
+  
   @Ignore("Wrapped stream doesn't support ByteBufferRead")
   @Override
   @Test(timeout=10000)
   public void testByteBufferRead() throws IOException {}
-
-  @Ignore("Wrapped stream doesn't support ByteBufferPositionedReadable")
-  @Override
-  @Test(timeout=10000)
-  public void testByteBufferPread() throws IOException {}
-
+  
   @Ignore("Wrapped stream doesn't support ByteBufferRead, Seek")
   @Override
   @Test(timeout=10000)
   public void testCombinedOp() throws IOException {}
-
+  
   @Ignore("Wrapped stream doesn't support SeekToNewSource")
   @Override
   @Test(timeout=10000)
   public void testSeekToNewSource() throws IOException {}
-
+  
   @Ignore("Wrapped stream doesn't support HasEnhancedByteBufferAccess")
   @Override
   @Test(timeout=10000)
@@ -140,4 +135,4 @@ public void testHasEnhancedByteBufferAccess() throws IOException {}
   @Override
   @Test
   public void testUnbuffer() throws Exception {}
-}
+}