diff --git a/hbase-common/src/main/java/org/apache/hadoop/hbase/util/ByteBufferAllocator.java b/hbase-common/src/main/java/org/apache/hadoop/hbase/util/ByteBufferAllocator.java index 8e2ac16..2c5eac8 100644 --- a/hbase-common/src/main/java/org/apache/hadoop/hbase/util/ByteBufferAllocator.java +++ b/hbase-common/src/main/java/org/apache/hadoop/hbase/util/ByteBufferAllocator.java @@ -31,9 +31,8 @@ public interface ByteBufferAllocator { /** * Allocates a bytebuffer * @param size the size of the bytebuffer - * @param directByteBuffer indicator to create a direct bytebuffer * @return the bytebuffer that is created * @throws IOException exception thrown if there is an error while creating the ByteBuffer */ - ByteBuffer allocate(long size, boolean directByteBuffer) throws IOException; + ByteBuffer allocate(long size) throws IOException; } diff --git a/hbase-common/src/main/java/org/apache/hadoop/hbase/util/ByteBufferArray.java b/hbase-common/src/main/java/org/apache/hadoop/hbase/util/ByteBufferArray.java index d1d5e7d..d6c6490 100644 --- a/hbase-common/src/main/java/org/apache/hadoop/hbase/util/ByteBufferArray.java +++ b/hbase-common/src/main/java/org/apache/hadoop/hbase/util/ByteBufferArray.java @@ -63,7 +63,7 @@ public class ByteBufferArray { * @param allocator the ByteBufferAllocator that will create the buffers * @throws IOException throws IOException if there is an exception thrown by the allocator */ - public ByteBufferArray(long capacity, boolean directByteBuffer, ByteBufferAllocator allocator) + public ByteBufferArray(long capacity, ByteBufferAllocator allocator) throws IOException { this.bufferSize = DEFAULT_BUFFER_SIZE; if (this.bufferSize > (capacity / 16)) @@ -71,13 +71,13 @@ public class ByteBufferArray { this.bufferCount = (int) (roundUp(capacity, bufferSize) / bufferSize); LOG.info("Allocating buffers total=" + StringUtils.byteDesc(capacity) + ", sizePerBuffer=" + StringUtils.byteDesc(bufferSize) + ", count=" - + bufferCount + ", direct=" + directByteBuffer); + + bufferCount); buffers = new ByteBuffer[bufferCount + 1]; - createBuffers(directByteBuffer, allocator); + createBuffers(allocator); } @VisibleForTesting - void createBuffers(boolean directByteBuffer, ByteBufferAllocator allocator) + void createBuffers(ByteBufferAllocator allocator) throws IOException { int threadCount = getThreadCount(); ExecutorService service = new ThreadPoolExecutor(threadCount, threadCount, 0L, @@ -90,7 +90,7 @@ public class ByteBufferArray { // Last thread will have to deal with a different number of buffers int buffersToCreate = (i == threadCount - 1) ? lastThreadCount : perThreadCount; futures[i] = service.submit( - new BufferCreatorCallable(bufferSize, directByteBuffer, buffersToCreate, allocator)); + new BufferCreatorCallable(bufferSize, buffersToCreate, allocator)); } int bufferIndex = 0; for (Future future : futures) { @@ -122,14 +122,11 @@ public class ByteBufferArray { */ private static class BufferCreatorCallable implements Callable { private final int bufferCapacity; - private final boolean directByteBuffer; private final int bufferCount; private final ByteBufferAllocator allocator; - BufferCreatorCallable(int bufferCapacity, boolean directByteBuffer, int bufferCount, - ByteBufferAllocator allocator) { + BufferCreatorCallable(int bufferCapacity, int bufferCount, ByteBufferAllocator allocator) { this.bufferCapacity = bufferCapacity; - this.directByteBuffer = directByteBuffer; this.bufferCount = bufferCount; this.allocator = allocator; } @@ -138,7 +135,7 @@ public class ByteBufferArray { public ByteBuffer[] call() throws Exception { ByteBuffer[] buffers = new ByteBuffer[this.bufferCount]; for (int i = 0; i < this.bufferCount; i++) { - buffers[i] = allocator.allocate(this.bufferCapacity, this.directByteBuffer); + buffers[i] = allocator.allocate(this.bufferCapacity); } return buffers; } diff --git a/hbase-common/src/main/resources/hbase-default.xml b/hbase-common/src/main/resources/hbase-default.xml index 28e9b4d..be23dc8 100644 --- a/hbase-common/src/main/resources/hbase-default.xml +++ b/hbase-common/src/main/resources/hbase-default.xml @@ -894,9 +894,10 @@ possible configurations would overwhelm and obscure the important. hbase.bucketcache.ioengine - Where to store the contents of the bucketcache. One of: heap, - offheap, or file. If a file, set it to file:PATH_TO_FILE. See - http://hbase.apache.org/book.html#offheap.blockcache for more information. + Where to store the contents of the bucketcache. One of: offheap, + file, files or mmap. If a file or files, set it to file(s):PATH_TO_FILE. + mmap means the content will be in an mmaped file. Use mmap:PATH_TO_FILE. + See http://hbase.apache.org/book.html#offheap.blockcache for more information. diff --git a/hbase-common/src/test/java/org/apache/hadoop/hbase/util/TestByteBufferArray.java b/hbase-common/src/test/java/org/apache/hadoop/hbase/util/TestByteBufferArray.java index 7077643..4b87945 100644 --- a/hbase-common/src/test/java/org/apache/hadoop/hbase/util/TestByteBufferArray.java +++ b/hbase-common/src/test/java/org/apache/hadoop/hbase/util/TestByteBufferArray.java @@ -38,16 +38,11 @@ public class TestByteBufferArray { int capacity = 4 * 1024 * 1024; ByteBufferAllocator allocator = new ByteBufferAllocator() { @Override - public ByteBuffer allocate(long size, boolean directByteBuffer) - throws IOException { - if (directByteBuffer) { - return ByteBuffer.allocateDirect((int) size); - } else { - return ByteBuffer.allocate((int) size); - } + public ByteBuffer allocate(long size) throws IOException { + return ByteBuffer.allocateDirect((int) size); } }; - ByteBufferArray array = new ByteBufferArray(capacity, false, allocator); + ByteBufferArray array = new ByteBufferArray(capacity, allocator); ByteBuff subBuf = array.asSubByteBuff(0, capacity); subBuf.position(capacity - 1);// Position to the last byte assertTrue(subBuf.hasRemaining()); @@ -61,15 +56,11 @@ public class TestByteBufferArray { int capacity = 470 * 1021 * 1023; ByteBufferAllocator allocator = new ByteBufferAllocator() { @Override - public ByteBuffer allocate(long size, boolean directByteBuffer) throws IOException { - if (directByteBuffer) { - return ByteBuffer.allocateDirect((int) size); - } else { - return ByteBuffer.allocate((int) size); - } + public ByteBuffer allocate(long size) throws IOException { + return ByteBuffer.allocateDirect((int) size); } }; - ByteBufferArray array = new ByteBufferArray(capacity, false, allocator); + ByteBufferArray array = new ByteBufferArray(capacity, allocator); assertEquals(119, array.buffers.length); for (int i = 0; i < array.buffers.length; i++) { if (i == array.buffers.length - 1) { @@ -84,19 +75,15 @@ public class TestByteBufferArray { public void testByteBufferCreation1() throws Exception { ByteBufferAllocator allocator = new ByteBufferAllocator() { @Override - public ByteBuffer allocate(long size, boolean directByteBuffer) throws IOException { - if (directByteBuffer) { - return ByteBuffer.allocateDirect((int) size); - } else { - return ByteBuffer.allocate((int) size); - } + public ByteBuffer allocate(long size) throws IOException { + return ByteBuffer.allocateDirect((int) size); } }; - ByteBufferArray array = new DummyByteBufferArray(7 * 1024 * 1024, false, allocator); + ByteBufferArray array = new DummyByteBufferArray(7 * 1024 * 1024, allocator); // overwrite array.bufferCount = 25; array.buffers = new ByteBuffer[array.bufferCount + 1]; - array.createBuffers(true, allocator); + array.createBuffers(allocator); for (int i = 0; i < array.buffers.length; i++) { if (i == array.buffers.length - 1) { assertEquals(array.buffers[i].capacity(), 0); @@ -108,9 +95,8 @@ public class TestByteBufferArray { private static class DummyByteBufferArray extends ByteBufferArray { - public DummyByteBufferArray(long capacity, boolean directByteBuffer, - ByteBufferAllocator allocator) throws IOException { - super(capacity, directByteBuffer, allocator); + public DummyByteBufferArray(long capacity, ByteBufferAllocator allocator) throws IOException { + super(capacity, allocator); } @Override diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/io/hfile/bucket/BucketCache.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/io/hfile/bucket/BucketCache.java index 6f95b6f..b0011d7 100644 --- a/hbase-server/src/main/java/org/apache/hadoop/hbase/io/hfile/bucket/BucketCache.java +++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/io/hfile/bucket/BucketCache.java @@ -381,14 +381,12 @@ public class BucketCache implements BlockCache, HeapSize { .split(FileIOEngine.FILE_DELIMITER); return new FileIOEngine(capacity, persistencePath != null, filePaths); } else if (ioEngineName.startsWith("offheap")) { - return new ByteBufferIOEngine(capacity, true); - } else if (ioEngineName.startsWith("heap")) { - return new ByteBufferIOEngine(capacity, false); + return new ByteBufferIOEngine(capacity); } else if (ioEngineName.startsWith("mmap:")) { return new FileMmapEngine(ioEngineName.substring(5), capacity); } else { throw new IllegalArgumentException( - "Don't understand io engine name for cache - prefix with file:, heap or offheap"); + "Don't understand io engine name for cache- prefix with file:, files:, mmap: or offheap"); } } diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/io/hfile/bucket/ByteBufferIOEngine.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/io/hfile/bucket/ByteBufferIOEngine.java index 4fefa97..9f4ffba 100644 --- a/hbase-server/src/main/java/org/apache/hadoop/hbase/io/hfile/bucket/ByteBufferIOEngine.java +++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/io/hfile/bucket/ByteBufferIOEngine.java @@ -68,36 +68,28 @@ So said all these, when we read a block it may be possible that the bytes of tha public class ByteBufferIOEngine implements IOEngine { private ByteBufferArray bufferArray; private final long capacity; - private final boolean direct; /** * Construct the ByteBufferIOEngine with the given capacity * @param capacity - * @param direct true if allocate direct buffer * @throws IOException ideally here no exception to be thrown from the allocator */ - public ByteBufferIOEngine(long capacity, boolean direct) + public ByteBufferIOEngine(long capacity) throws IOException { this.capacity = capacity; - this.direct = direct; ByteBufferAllocator allocator = new ByteBufferAllocator() { @Override - public ByteBuffer allocate(long size, boolean directByteBuffer) - throws IOException { - if (directByteBuffer) { - return ByteBuffer.allocateDirect((int) size); - } else { - return ByteBuffer.allocate((int) size); - } + public ByteBuffer allocate(long size) throws IOException { + return ByteBuffer.allocateDirect((int) size); } }; - bufferArray = new ByteBufferArray(capacity, direct, allocator); + bufferArray = new ByteBufferArray(capacity, allocator); } @Override public String toString() { return "ioengine=" + this.getClass().getSimpleName() + ", capacity=" + - String.format("%,d", this.capacity) + ", direct=" + this.direct; + String.format("%,d", this.capacity); } /** diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/io/hfile/bucket/FileMmapEngine.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/io/hfile/bucket/FileMmapEngine.java index 45036e5..4fe39d3 100644 --- a/hbase-server/src/main/java/org/apache/hadoop/hbase/io/hfile/bucket/FileMmapEngine.java +++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/io/hfile/bucket/FileMmapEngine.java @@ -71,20 +71,14 @@ public class FileMmapEngine implements IOEngine { ByteBufferAllocator allocator = new ByteBufferAllocator() { int pos = 0; @Override - public ByteBuffer allocate(long size, boolean directByteBuffer) throws IOException { - ByteBuffer buffer = null; - if (directByteBuffer) { - buffer = fileChannel.map(java.nio.channels.FileChannel.MapMode.READ_WRITE, pos * size, - size); - } else { - throw new IllegalArgumentException( - "Only Direct Bytebuffers allowed with FileMMap engine"); - } + public ByteBuffer allocate(long size) throws IOException { + ByteBuffer buffer = fileChannel.map(java.nio.channels.FileChannel.MapMode.READ_WRITE, + pos * size, size); pos++; return buffer; } }; - bufferArray = new ByteBufferArray(fileSize, true, allocator); + bufferArray = new ByteBufferArray(fileSize, allocator); } private long roundUp(long n, long to) { diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/io/util/MemorySizeUtil.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/io/util/MemorySizeUtil.java index dbfb295..55e4a9a 100644 --- a/hbase-server/src/main/java/org/apache/hadoop/hbase/io/util/MemorySizeUtil.java +++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/io/util/MemorySizeUtil.java @@ -260,21 +260,13 @@ public class MemorySizeUtil { * @return the number of bytes to use for bucket cache, negative if disabled. */ public static long getBucketCacheSize(final Configuration conf) { - final float bucketCachePercentage = conf.getFloat(HConstants.BUCKET_CACHE_SIZE_KEY, 0F); - long bucketCacheSize; - // Values < 1 are treated as % of heap - if (bucketCachePercentage < 1) { - long max = -1L; - final MemoryUsage usage = safeGetHeapMemoryUsage(); - if (usage != null) { - max = usage.getMax(); - } - bucketCacheSize = (long)(max * bucketCachePercentage); - // values >= 1 are treated as # of MiB - } else { - bucketCacheSize = (long)(bucketCachePercentage * 1024 * 1024); + // Size configured in MBs + float bucketCacheSize = conf.getFloat(HConstants.BUCKET_CACHE_SIZE_KEY, 0F); + if (bucketCacheSize < 1) { + throw new IllegalArgumentException("Bucket Cache should be minimum 1 MB in size." + + "Configure 'hbase.bucketcache.size' with > 1 value"); } - return bucketCacheSize; + return (long) (bucketCacheSize * 1024 * 1024); } } diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/io/hfile/bucket/TestBucketCache.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/io/hfile/bucket/TestBucketCache.java index 5a2a51c..1c7f951 100644 --- a/hbase-server/src/test/java/org/apache/hadoop/hbase/io/hfile/bucket/TestBucketCache.java +++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/io/hfile/bucket/TestBucketCache.java @@ -20,14 +20,12 @@ package org.apache.hadoop.hbase.io.hfile.bucket; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertFalse; -import static org.junit.Assert.assertThat; import static org.junit.Assert.assertTrue; import java.io.FileNotFoundException; import java.io.IOException; import java.util.ArrayList; import java.util.Arrays; -import java.util.Collection; import java.util.List; import java.util.Map; import java.util.Random; @@ -92,7 +90,7 @@ public class TestBucketCache { final long capacitySize = 32 * 1024 * 1024; final int writeThreads = BucketCache.DEFAULT_WRITER_THREADS; final int writerQLen = BucketCache.DEFAULT_WRITER_QUEUE_ITEMS; - String ioEngineName = "heap"; + String ioEngineName = "offheap"; String persistencePath = null; private class MockedBucketCache extends BucketCache { diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/io/hfile/bucket/TestBucketWriterThread.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/io/hfile/bucket/TestBucketWriterThread.java index e789b4f..b8949ed 100644 --- a/hbase-server/src/test/java/org/apache/hadoop/hbase/io/hfile/bucket/TestBucketWriterThread.java +++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/io/hfile/bucket/TestBucketWriterThread.java @@ -36,7 +36,6 @@ import java.io.IOException; import java.util.ArrayList; import java.util.List; import java.util.concurrent.BlockingQueue; -import java.util.concurrent.atomic.AtomicLong; import java.util.concurrent.atomic.LongAdder; import static org.hamcrest.CoreMatchers.is; @@ -80,7 +79,7 @@ public class TestBucketWriterThread { // Run with one writer thread only. Means there will be one writer queue only too. We depend // on this in below. final int writerThreadsCount = 1; - this.bc = new MockBucketCache("heap", capacity, 1, new int [] {1}, writerThreadsCount, + this.bc = new MockBucketCache("offheap", capacity, 1, new int [] {1}, writerThreadsCount, capacity, null, 100/*Tolerate ioerrors for 100ms*/); assertEquals(writerThreadsCount, bc.writerThreads.length); assertEquals(writerThreadsCount, bc.writerQueues.size()); diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/io/hfile/bucket/TestByteBufferIOEngine.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/io/hfile/bucket/TestByteBufferIOEngine.java index bf15a59..ab2276a 100644 --- a/hbase-server/src/test/java/org/apache/hadoop/hbase/io/hfile/bucket/TestByteBufferIOEngine.java +++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/io/hfile/bucket/TestByteBufferIOEngine.java @@ -43,7 +43,7 @@ public class TestByteBufferIOEngine { int capacity = 32 * 1024 * 1024; // 32 MB int testNum = 100; int maxBlockSize = 64 * 1024; - ByteBufferIOEngine ioEngine = new ByteBufferIOEngine(capacity, false); + ByteBufferIOEngine ioEngine = new ByteBufferIOEngine(capacity); int testOffsetAtStartNum = testNum / 10; int testOffsetAtEndNum = testNum / 10; for (int i = 0; i < testNum; i++) { @@ -113,7 +113,7 @@ public class TestByteBufferIOEngine { int capacity = 32 * 1024 * 1024; // 32 MB int testNum = 100; int maxBlockSize = 64 * 1024; - ByteBufferIOEngine ioEngine = new ByteBufferIOEngine(capacity, false); + ByteBufferIOEngine ioEngine = new ByteBufferIOEngine(capacity); int testOffsetAtStartNum = testNum / 10; int testOffsetAtEndNum = testNum / 10; for (int i = 0; i < testNum; i++) {