HBASE-22802 Avoid temp ByteBuffer allocation in FileIOEngine#read

hbase-common/src/main/java/org/apache/hadoop/hbase/nio/ByteBuff.java

@@ -19,6 +19,7 @@
 
 import java.io.IOException;
 import java.nio.ByteBuffer;
+import java.nio.channels.FileChannel;
 import java.nio.channels.ReadableByteChannel;
 import java.util.List;
 
@@ -450,6 +451,16 @@ public byte[] toBytes() {
    */
   public abstract int read(ReadableByteChannel channel) throws IOException;
 
+  /**
+   * Reads bytes from FileChannel into this ByteBuff
+   */
+  public abstract int read(FileChannel channel, long offset) throws IOException;
+
+  /**
+   * Write this ByteBuff's data into target file
+   */
+  public abstract int write(FileChannel channel, long offset) throws IOException;
+
   // static helper methods
   public static int channelRead(ReadableByteChannel channel, ByteBuffer buf) throws IOException {
     if (buf.remaining() <= NIO_BUFFER_LIMIT) {
@@ -475,6 +486,32 @@ public static int channelRead(ReadableByteChannel channel, ByteBuffer buf) throw
     return (nBytes > 0) ? nBytes : ret;
   }
 
+  public static int fileRead(FileChannel channel, ByteBuffer buf, long offset)
+      throws IOException {
+    if (buf.remaining() <= NIO_BUFFER_LIMIT) {
+      return channel.read(buf, offset);
+    }
+    int originalLimit = buf.limit();
+    int initialRemaining = buf.remaining();
+    int ret = 0;
+
+    while (buf.remaining() > 0) {
+      try {
+        int ioSize = Math.min(buf.remaining(), NIO_BUFFER_LIMIT);
+        buf.limit(buf.position() + ioSize);
+        offset += ret;
+        ret = channel.read(buf, offset);
+        if (ret < ioSize) {
+          break;
+        }
+      } finally {
+        buf.limit(originalLimit);
+      }
+    }
+    int nBytes = initialRemaining - buf.remaining();
+    return (nBytes > 0) ? nBytes : ret;
+  }
+
   /**
    * Read integer from ByteBuff coded in 7 bits and increment position.
    * @return Read integer.

hbase-common/src/main/java/org/apache/hadoop/hbase/nio/MultiByteBuff.java

@@ -24,6 +24,7 @@
 import java.nio.BufferUnderflowException;
 import java.nio.ByteBuffer;
 import java.nio.InvalidMarkException;
+import java.nio.channels.FileChannel;
 import java.nio.channels.ReadableByteChannel;
 
 import org.apache.hadoop.hbase.io.ByteBuffAllocator.Recycler;
@@ -1086,6 +1087,48 @@ public int read(ReadableByteChannel channel) throws IOException {
     return total;
   }
 
+  @Override
+  public int read(FileChannel channel, long offset) throws IOException {
+    checkRefCount();
+    int total = 0;
+    while (true) {
+      int len = fileRead(channel, this.curItem, offset);
+      if (len > 0) {
+        total += len;
+        offset += len;
+      }
+      if (this.curItem.hasRemaining()) {
+        break;
+      } else {
+        if (this.curItemIndex >= this.limitedItemIndex) {
+          break;
+        }
+        this.curItemIndex++;
+        this.curItem = this.items[this.curItemIndex];
+      }
+    }
+    return total;
+  }
+
+  @Override
+  public int write(FileChannel channel, long offset) throws IOException {
+    checkRefCount();
+    int total = 0;
+    while (true) {
+      int len = channel.write(curItem, offset);
+      if (len > 0) {
+        total += len;
+        offset += len;
+      }
+      if (this.curItemIndex >= this.limitedItemIndex) {
+        break;
+      }
+      this.curItemIndex++;
+      this.curItem = this.items[this.curItemIndex];
+    }
+    return total;
+  }
+
   @Override
   public ByteBuffer[] nioByteBuffers() {
     checkRefCount();

hbase-common/src/main/java/org/apache/hadoop/hbase/nio/SingleByteBuff.java

@@ -21,6 +21,7 @@
 
 import java.io.IOException;
 import java.nio.ByteBuffer;
+import java.nio.channels.FileChannel;
 import java.nio.channels.ReadableByteChannel;
 
 import org.apache.hadoop.hbase.io.ByteBuffAllocator.Recycler;
@@ -374,6 +375,18 @@ public int read(ReadableByteChannel channel) throws IOException {
     return channelRead(channel, buf);
   }
 
+  @Override
+  public int read(FileChannel channel, long offset) throws IOException {
+    checkRefCount();
+    return fileRead(channel, buf, offset);
+  }
+
+  @Override
+  public int write(FileChannel channel, long offset) throws IOException {
+    checkRefCount();
+    return channel.write(buf, offset);
+  }
+
   @Override
   public ByteBuffer[] nioByteBuffers() {
     checkRefCount();

hbase-server/src/main/java/org/apache/hadoop/hbase/io/hfile/bucket/BucketCache.java

@@ -429,7 +429,9 @@ private void cacheBlockWithWaitInternal(BlockCacheKey cacheKey, Cacheable cached
     if (!cacheEnabled) {
       return;
     }
-    LOG.trace("Caching key={}, item={}", cacheKey, cachedItem);
+    if (LOG.isTraceEnabled()) {
+      LOG.trace("Caching key={}, item={}", cacheKey, cachedItem);
+    }
     // Stuff the entry into the RAM cache so it can get drained to the persistent store
     RAMQueueEntry re =
         new RAMQueueEntry(cacheKey, cachedItem, accessCount.incrementAndGet(), inMemory,
@@ -502,8 +504,10 @@ public Cacheable getBlock(BlockCacheKey key, boolean caching, boolean repeat,
           // block will use the refCnt of bucketEntry, which means if two HFileBlock mapping to
           // the same BucketEntry, then all of the three will share the same refCnt.
           Cacheable cachedBlock = ioEngine.read(bucketEntry);
-          // RPC start to reference, so retain here.
-          cachedBlock.retain();
+          if (ioEngine.usesSharedMemory()) {
+            // RPC start to reference, so retain here.
+            cachedBlock.retain();
+          }
           // Update the cache statistics.
           if (updateCacheMetrics) {
             cacheStats.hit(caching, key.isPrimary(), key.getBlockType());

hbase-server/src/main/java/org/apache/hadoop/hbase/io/hfile/bucket/BucketEntry.java

@@ -80,7 +80,7 @@ class BucketEntry implements HBaseReferenceCounted {
    */
   private final RefCnt refCnt;
   final AtomicBoolean markedAsEvicted;
-  private final ByteBuffAllocator allocator;
+  final ByteBuffAllocator allocator;
 
   /**
    * Time this block was cached. Presumes we are created just before we are added to the cache.
@@ -194,7 +194,10 @@ boolean isRpcRef() {
   }
 
   Cacheable wrapAsCacheable(ByteBuffer[] buffers) throws IOException {
-    ByteBuff buf = ByteBuff.wrap(buffers, this.refCnt);
+    return wrapAsCacheable(ByteBuff.wrap(buffers, this.refCnt));
+  }
+
+  Cacheable wrapAsCacheable(ByteBuff buf) throws IOException {
     return this.deserializerReference().deserialize(buf, allocator);
   }
 

hbase-server/src/main/java/org/apache/hadoop/hbase/io/hfile/bucket/ExclusiveMemoryMmapIOEngine.java

@@ -17,7 +17,6 @@
 package org.apache.hadoop.hbase.io.hfile.bucket;
 
 import java.io.IOException;
-import java.nio.ByteBuffer;
 
 import org.apache.hadoop.hbase.io.hfile.Cacheable;
 import org.apache.hadoop.hbase.nio.ByteBuff;
@@ -35,9 +34,9 @@ public ExclusiveMemoryMmapIOEngine(String filePath, long capacity) throws IOExce
 
   @Override
   public Cacheable read(BucketEntry be) throws IOException {
-    ByteBuff dst = ByteBuff.wrap(ByteBuffer.allocate(be.getLength()));
+    ByteBuff dst = be.allocator.allocate(be.getLength());
     bufferArray.read(be.offset(), dst);
     dst.position(0).limit(be.getLength());
-    return be.wrapAsCacheable(dst.nioByteBuffers());
+    return be.wrapAsCacheable(dst);
   }
 }

hbase-server/src/main/java/org/apache/hadoop/hbase/io/hfile/bucket/FileIOEngine.java

@@ -129,7 +129,7 @@ public Cacheable read(BucketEntry be) throws IOException {
     long offset = be.offset();
     int length = be.getLength();
     Preconditions.checkArgument(length >= 0, "Length of read can not be less than 0.");
-    ByteBuffer dstBuffer = ByteBuffer.allocate(length);
+    ByteBuff dstBuffer = be.allocator.allocate(length);
     if (length != 0) {
       accessFile(readAccessor, dstBuffer, offset);
       // The buffer created out of the fileChannel is formed by copying the data from the file
@@ -142,7 +142,7 @@ public Cacheable read(BucketEntry be) throws IOException {
       }
     }
     dstBuffer.rewind();
-    return be.wrapAsCacheable(new ByteBuffer[] { dstBuffer });
+    return be.wrapAsCacheable(dstBuffer);
   }
 
   @VisibleForTesting
@@ -164,10 +164,7 @@ void closeFileChannels() {
    */
   @Override
   public void write(ByteBuffer srcBuffer, long offset) throws IOException {
-    if (!srcBuffer.hasRemaining()) {
-      return;
-    }
-    accessFile(writeAccessor, srcBuffer, offset);
+    write(ByteBuff.wrap(srcBuffer), offset);
   }
 
   /**
@@ -209,11 +206,13 @@ public void shutdown() {
 
   @Override
   public void write(ByteBuff srcBuffer, long offset) throws IOException {
-    ByteBuffer dup = srcBuffer.asSubByteBuffer(srcBuffer.remaining()).duplicate();
-    write(dup, offset);
+    if (!srcBuffer.hasRemaining()) {
+      return;
+    }
+    accessFile(writeAccessor, srcBuffer, offset);
   }
 
-  private void accessFile(FileAccessor accessor, ByteBuffer buffer,
+  private void accessFile(FileAccessor accessor, ByteBuff buffer,
       long globalOffset) throws IOException {
     int startFileNum = getFileNum(globalOffset);
     int remainingAccessDataLen = buffer.remaining();
@@ -304,23 +303,23 @@ void refreshFileConnection(int accessFileNum, IOException ioe) throws IOExceptio
   }
 
   private interface FileAccessor {
-    int access(FileChannel fileChannel, ByteBuffer byteBuffer, long accessOffset)
+    int access(FileChannel fileChannel, ByteBuff byteBuffer, long accessOffset)
         throws IOException;
   }
 
   private static class FileReadAccessor implements FileAccessor {
     @Override
-    public int access(FileChannel fileChannel, ByteBuffer byteBuffer,
+    public int access(FileChannel fileChannel, ByteBuff byteBuffer,
         long accessOffset) throws IOException {
-      return fileChannel.read(byteBuffer, accessOffset);
+      return byteBuffer.read(fileChannel, accessOffset);
     }
   }
 
   private static class FileWriteAccessor implements FileAccessor {
     @Override
-    public int access(FileChannel fileChannel, ByteBuffer byteBuffer,
+    public int access(FileChannel fileChannel, ByteBuff byteBuffer,
         long accessOffset) throws IOException {
-      return fileChannel.write(byteBuffer, accessOffset);
+      return byteBuffer.write(fileChannel, accessOffset);
     }
   }
 }