.../hadoop/hbase/util/ByteBufferAllocator.java | 46 ++++++ .../apache/hadoop/hbase/util/ByteBufferArray.java | 25 ++- .../hadoop/hbase/util/TestByteBufferArray.java | 20 ++- .../hadoop/hbase/io/hfile/bucket/BucketCache.java | 2 + .../hbase/io/hfile/bucket/ByteBufferIOEngine.java | 18 ++- .../hbase/io/hfile/bucket/FileMmapEngine.java | 176 +++++++++++++++++++++ .../hbase/io/hfile/bucket/TestFileMmapEngine.java | 68 ++++++++ 7 files changed, 346 insertions(+), 9 deletions(-) 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 new file mode 100644 index 0000000..0fb8eda --- /dev/null +++ b/hbase-common/src/main/java/org/apache/hadoop/hbase/util/ByteBufferAllocator.java @@ -0,0 +1,46 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.hadoop.hbase.util; + +import java.io.IOException; +import java.nio.ByteBuffer; + +import org.apache.hadoop.hbase.classification.InterfaceAudience; + +/** + * Defines the way the ByteBuffers are created + */ +@InterfaceAudience.Private +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; + + /** + * Performs sync operation on the created buffers, if there is any + * @param buffers bytebuffers on which the sync has to be performed + * @throws IOException exception that is thrown if there is an error on sync + */ + void sync(ByteBuffer[] buffers) throws IOException; +} \ No newline at end of file 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 2334cf7..8b3f8d4 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 @@ -18,6 +18,7 @@ */ package org.apache.hadoop.hbase.util; +import java.io.IOException; import java.nio.ByteBuffer; import java.util.concurrent.locks.Lock; import java.util.concurrent.locks.ReentrantLock; @@ -39,20 +40,23 @@ import org.apache.hadoop.util.StringUtils; public final class ByteBufferArray { private static final Log LOG = LogFactory.getLog(ByteBufferArray.class); - static final int DEFAULT_BUFFER_SIZE = 4 * 1024 * 1024; + public static final int DEFAULT_BUFFER_SIZE = 4 * 1024 * 1024; private ByteBuffer buffers[]; private Lock locks[]; private int bufferSize; private int bufferCount; - + private ByteBufferAllocator allocator; /** * We allocate a number of byte buffers as the capacity. In order not to out * of the array bounds for the last byte(see {@link ByteBufferArray#multiple}), * we will allocate one additional buffer with capacity 0; * @param capacity total size of the byte buffer array * @param directByteBuffer true if we allocate direct buffer + * @param allocator the ByteBufferAllocator that will create the buffers + * @throws IOException */ - public ByteBufferArray(long capacity, boolean directByteBuffer) { + public ByteBufferArray(long capacity, boolean directByteBuffer, ByteBufferAllocator allocator) + throws IOException { this.bufferSize = DEFAULT_BUFFER_SIZE; if (this.bufferSize > (capacity / 16)) this.bufferSize = (int) roundUp(capacity / 16, 32768); @@ -62,18 +66,25 @@ public final class ByteBufferArray { + bufferCount + ", direct=" + directByteBuffer); buffers = new ByteBuffer[bufferCount + 1]; locks = new Lock[bufferCount + 1]; + this.allocator = allocator; for (int i = 0; i <= bufferCount; i++) { locks[i] = new ReentrantLock(); if (i < bufferCount) { - buffers[i] = directByteBuffer ? ByteBuffer.allocateDirect(bufferSize) - : ByteBuffer.allocate(bufferSize); + buffers[i] = allocator.allocate(bufferSize, directByteBuffer); } else { - buffers[i] = ByteBuffer.allocate(0); + buffers[i] = allocator.allocate(0, false); } - } } + /** + * Sync the data to IOEngine after writing + * @throws IOException + */ + public void sync() throws IOException { + this.allocator.sync(buffers); + } + private long roundUp(long n, long to) { return ((n + to - 1) / to) * to; } 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 701601d..3d4689f 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 @@ -20,6 +20,9 @@ package org.apache.hadoop.hbase.util; import static org.junit.Assert.assertFalse; import static org.junit.Assert.assertTrue; +import java.io.IOException; +import java.nio.ByteBuffer; + import org.apache.hadoop.hbase.nio.ByteBuff; import org.apache.hadoop.hbase.testclassification.MiscTests; import org.apache.hadoop.hbase.testclassification.SmallTests; @@ -32,7 +35,22 @@ public class TestByteBufferArray { @Test public void testAsSubBufferWhenEndOffsetLandInLastBuffer() throws Exception { int capacity = 4 * 1024 * 1024; - ByteBufferArray array = new ByteBufferArray(capacity, false); + 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); + } + } + @Override + public void sync(ByteBuffer[] buffers) throws IOException { + // do nothing + } + }; + ByteBufferArray array = new ByteBufferArray(capacity, false, allocator); ByteBuff subBuf = array.asSubByteBuff(0, capacity); subBuf.position(capacity - 1);// Position to the last byte assertTrue(subBuf.hasRemaining()); 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 6024958..3315e29 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 @@ -309,6 +309,8 @@ public class BucketCache implements BlockCache, HeapSize { return new ByteBufferIOEngine(capacity, true); else if (ioEngineName.startsWith("heap")) return new ByteBufferIOEngine(capacity, false); + 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"); 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 2227312..7d6061f 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 @@ -26,6 +26,7 @@ import org.apache.hadoop.hbase.io.hfile.Cacheable; import org.apache.hadoop.hbase.io.hfile.CacheableDeserializer; import org.apache.hadoop.hbase.io.hfile.Cacheable.MemoryType; import org.apache.hadoop.hbase.nio.ByteBuff; +import org.apache.hadoop.hbase.util.ByteBufferAllocator; import org.apache.hadoop.hbase.util.ByteBufferArray; /** @@ -48,7 +49,22 @@ public class ByteBufferIOEngine implements IOEngine { throws IOException { this.capacity = capacity; this.direct = direct; - bufferArray = new ByteBufferArray(capacity, 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); + } + } + @Override + public void sync(ByteBuffer[] buffers) throws IOException { + // do nothing + } + }; + bufferArray = new ByteBufferArray(capacity, direct, allocator); } @Override 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 new file mode 100644 index 0000000..1c0580e --- /dev/null +++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/io/hfile/bucket/FileMmapEngine.java @@ -0,0 +1,176 @@ +/** + * Copyright The Apache Software Foundation + * + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with this + * work for additional information regarding copyright ownership. The ASF + * licenses this file to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT + * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the + * License for the specific language governing permissions and limitations + * under the License. + */ +package org.apache.hadoop.hbase.io.hfile.bucket; + +import java.io.IOException; +import java.io.RandomAccessFile; +import java.nio.ByteBuffer; +import java.nio.channels.FileChannel; + +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; +import org.apache.hadoop.hbase.classification.InterfaceAudience; +import org.apache.hadoop.hbase.io.hfile.Cacheable; +import org.apache.hadoop.hbase.io.hfile.Cacheable.MemoryType; +import org.apache.hadoop.hbase.io.hfile.CacheableDeserializer; +import org.apache.hadoop.hbase.nio.ByteBuff; +import org.apache.hadoop.hbase.nio.SingleByteBuff; +import org.apache.hadoop.util.StringUtils; +import org.apache.hadoop.hbase.util.ByteBufferAllocator; +import org.apache.hadoop.hbase.util.ByteBufferArray; + +/** + * IO engine that stores data to a file on the local file system using memory mapping + * mechanism + */ +@InterfaceAudience.Private +public class FileMmapEngine implements IOEngine { + static final Log LOG = LogFactory.getLog(FileMmapEngine.class); + + private final String path; + private long size; + private ByteBufferArray bufferArray; + private final FileChannel fileChannel; + private RandomAccessFile raf = null; + + public FileMmapEngine(String filePath, long capacity) throws IOException { + this.path = filePath; + this.size = capacity; + long fileSize = 0; + try { + raf = new RandomAccessFile(filePath, "rw"); + fileSize = roundUp(capacity, ByteBufferArray.DEFAULT_BUFFER_SIZE); + raf.setLength(fileSize); + fileChannel = raf.getChannel(); + LOG.info("Allocating " + StringUtils.byteDesc(fileSize) + ", on the path:" + filePath); + } catch (java.io.FileNotFoundException fex) { + LOG.error("Can't create bucket cache file " + filePath, fex); + throw fex; + } catch (IOException ioex) { + LOG.error("Can't extend bucket cache file; insufficient space for " + + StringUtils.byteDesc(fileSize), ioex); + shutdown(); + throw ioex; + } + 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 { + buffer = fileChannel.map(java.nio.channels.FileChannel.MapMode.READ_WRITE, pos * size, + 0); + } + pos++; + return buffer; + } + + @Override + public void sync(ByteBuffer[] buffers) throws IOException { + for (ByteBuffer bb : buffers) { + ((java.nio.MappedByteBuffer) bb).force(); + } + } + }; + bufferArray = new ByteBufferArray(fileSize, true, allocator); + } + + private long roundUp(long n, long to) { + return ((n + to - 1) / to) * to; + } + + @Override + public String toString() { + return "ioengine=" + this.getClass().getSimpleName() + ", path=" + this.path + + ", size=" + String.format("%,d", this.size); + } + + /** + * File IO engine is always able to support persistent storage for the cache + * @return true + */ + @Override + public boolean isPersistent() { + return true; + } + + @Override + public Cacheable read(long offset, int length, CacheableDeserializer deserializer) + throws IOException { + // TODO: See if the SHARED mode can be created here + byte[] dst = new byte[length]; + bufferArray.getMultiple(offset, length, dst); + return deserializer.deserialize(new SingleByteBuff(ByteBuffer.wrap(dst)), true, + MemoryType.EXCLUSIVE); + } + + /** + * Transfers data from the given byte buffer to file + * @param srcBuffer the given byte buffer from which bytes are to be read + * @param offset The offset in the file where the first byte to be written + * @throws IOException + */ + @Override + public void write(ByteBuffer srcBuffer, long offset) throws IOException { + assert srcBuffer.hasArray(); + bufferArray.putMultiple(offset, srcBuffer.remaining(), srcBuffer.array(), + srcBuffer.arrayOffset()); + } + + @Override + public void write(ByteBuff srcBuffer, long offset) throws IOException { + // This singleByteBuff can be considered to be array backed + assert srcBuffer.hasArray(); + bufferArray.putMultiple(offset, srcBuffer.remaining(), srcBuffer.array(), + srcBuffer.arrayOffset()); + + } + /** + * Sync the data to file after writing + * @throws IOException + */ + @Override + public void sync() throws IOException { + if (fileChannel != null) { + if(bufferArray != null) { + bufferArray.sync(); + } + } + } + + /** + * Close the file + */ + @Override + public void shutdown() { + try { + fileChannel.close(); + } catch (IOException ex) { + LOG.error("Can't shutdown cleanly", ex); + } + try { + raf.close(); + } catch (IOException ex) { + LOG.error("Can't shutdown cleanly", ex); + } + } +} \ No newline at end of file diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/io/hfile/bucket/TestFileMmapEngine.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/io/hfile/bucket/TestFileMmapEngine.java new file mode 100644 index 0000000..dfc18c7 --- /dev/null +++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/io/hfile/bucket/TestFileMmapEngine.java @@ -0,0 +1,68 @@ +/** + * Copyright The Apache Software Foundation + * + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with this + * work for additional information regarding copyright ownership. The ASF + * licenses this file to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT + * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the + * License for the specific language governing permissions and limitations + * under the License. + */ +package org.apache.hadoop.hbase.io.hfile.bucket; + +import static org.junit.Assert.assertTrue; + +import java.io.File; +import java.io.IOException; +import java.nio.ByteBuffer; + +import org.apache.hadoop.hbase.io.hfile.bucket.TestByteBufferIOEngine.BufferGrabbingDeserializer; +import org.apache.hadoop.hbase.nio.ByteBuff; +import org.apache.hadoop.hbase.testclassification.IOTests; +import org.apache.hadoop.hbase.testclassification.SmallTests; +import org.junit.Test; +import org.junit.experimental.categories.Category; + +/** + * Basic test for {@link FileMmapEngine} + */ +@Category({IOTests.class, SmallTests.class}) +public class TestFileMmapEngine { + @Test + public void testFileMmapEngine() throws IOException { + int size = 2 * 1024 * 1024; // 2 MB + String filePath = "testFileMmapEngine"; + try { + FileMmapEngine fileMmapEngine = new FileMmapEngine(filePath, size); + for (int i = 0; i < 50; i++) { + int len = (int) Math.floor(Math.random() * 100); + long offset = (long) Math.floor(Math.random() * size % (size - len)); + byte[] data1 = new byte[len]; + for (int j = 0; j < data1.length; ++j) { + data1[j] = (byte) (Math.random() * 255); + } + fileMmapEngine.write(ByteBuffer.wrap(data1), offset); + BufferGrabbingDeserializer deserializer = new BufferGrabbingDeserializer(); + fileMmapEngine.read(offset, len, deserializer); + ByteBuff data2 = deserializer.getDeserializedByteBuff(); + for (int j = 0; j < data1.length; ++j) { + assertTrue(data1[j] == data2.get(j)); + } + } + } finally { + File file = new File(filePath); + if (file.exists()) { + file.delete(); + } + } + + } +} \ No newline at end of file