Updated Branches: refs/heads/0.1.5 d37ded638 -> 869a2ad69
Slabs now are lazy to initialize, and the unit test is improved. Project: http://git-wip-us.apache.org/repos/asf/incubator-blur/repo Commit: http://git-wip-us.apache.org/repos/asf/incubator-blur/commit/5ce59476 Tree: http://git-wip-us.apache.org/repos/asf/incubator-blur/tree/5ce59476 Diff: http://git-wip-us.apache.org/repos/asf/incubator-blur/diff/5ce59476 Branch: refs/heads/0.1.5 Commit: 5ce59476bd8f46c0b44ec83841c6a3b4f4a35011 Parents: d37ded6 Author: Aaron McCurry <[email protected]> Authored: Tue Apr 30 20:53:37 2013 -0400 Committer: Aaron McCurry <[email protected]> Committed: Tue Apr 30 20:53:37 2013 -0400 ---------------------------------------------------------------------- .../apache/blur/store/blockcache/BlockCache.java | 126 +++++++++++++-- .../blur/store/blockcache/BlockCacheTest.java | 75 +++++++-- 2 files changed, 168 insertions(+), 33 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/5ce59476/src/blur-store/src/main/java/org/apache/blur/store/blockcache/BlockCache.java ---------------------------------------------------------------------- diff --git a/src/blur-store/src/main/java/org/apache/blur/store/blockcache/BlockCache.java b/src/blur-store/src/main/java/org/apache/blur/store/blockcache/BlockCache.java index 7a3eb99..7777047 100644 --- a/src/blur-store/src/main/java/org/apache/blur/store/blockcache/BlockCache.java +++ b/src/blur-store/src/main/java/org/apache/blur/store/blockcache/BlockCache.java @@ -16,9 +16,19 @@ package org.apache.blur.store.blockcache; * See the License for the specific language governing permissions and * limitations under the License. */ -import static org.apache.blur.metrics.MetricsConstants.*; +import static org.apache.blur.metrics.MetricsConstants.CACHE; +import static org.apache.blur.metrics.MetricsConstants.ENTRIES; +import static org.apache.blur.metrics.MetricsConstants.EVICTION; +import static org.apache.blur.metrics.MetricsConstants.ORG_APACHE_BLUR; +import static org.apache.blur.metrics.MetricsConstants.SIZE; +import java.io.Closeable; +import java.io.IOException; +import java.lang.reflect.Method; import java.nio.ByteBuffer; +import java.security.AccessController; +import java.security.PrivilegedActionException; +import java.security.PrivilegedExceptionAction; import java.util.concurrent.ConcurrentMap; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; @@ -30,7 +40,23 @@ import com.yammer.metrics.core.Gauge; import com.yammer.metrics.core.Meter; import com.yammer.metrics.core.MetricName; -public class BlockCache { +public class BlockCache implements Closeable { + + /** + * <code>true</code>, if this platform supports unmapping mmapped files. + */ + public static final boolean UNMAP_SUPPORTED; + static { + boolean v; + try { + Class.forName("sun.misc.Cleaner"); + Class.forName("java.nio.DirectByteBuffer").getMethod("cleaner"); + v = true; + } catch (Exception e) { + v = false; + } + UNMAP_SUPPORTED = v; + } public static final int _128M = 134217728; public static final int _8K = 8192; @@ -42,7 +68,18 @@ public class BlockCache { private final int _blockSize = _8K; private final int _numberOfBlocksPerSlab; private final int _maxEntries; - private Meter evictions; + private final Meter evictions; + private final int _numberOfSlabs; + private final boolean _directAllocation; + private final ThreadLocal<ByteBuffer[]> _threadLocalSlabs = new ThreadLocal<ByteBuffer[]>() { + @Override + protected ByteBuffer[] initialValue() { + return new ByteBuffer[_numberOfSlabs]; + } + }; + + //This turns the lazy bytebuffer allocation on or off. + private final boolean lazy = true; public BlockCache(boolean directAllocation, long totalMemory) { this(directAllocation, totalMemory, _128M); @@ -50,17 +87,20 @@ public class BlockCache { public BlockCache(boolean directAllocation, long totalMemory, int slabSize) { _numberOfBlocksPerSlab = slabSize / _blockSize; - int numberOfSlabs = (int) (totalMemory / slabSize); - - _slabs = new ByteBuffer[numberOfSlabs]; - _locks = new BlockLocks[numberOfSlabs]; - _lockCounters = new AtomicInteger[numberOfSlabs]; - _maxEntries = (_numberOfBlocksPerSlab * numberOfSlabs) - 1; - for (int i = 0; i < numberOfSlabs; i++) { - if (directAllocation) { - _slabs[i] = ByteBuffer.allocateDirect(_numberOfBlocksPerSlab * _blockSize); - } else { - _slabs[i] = ByteBuffer.allocate(_numberOfBlocksPerSlab * _blockSize); + _numberOfSlabs = (int) (totalMemory / slabSize); + _directAllocation = directAllocation; + + _slabs = new ByteBuffer[_numberOfSlabs]; + _locks = new BlockLocks[_numberOfSlabs]; + _lockCounters = new AtomicInteger[_numberOfSlabs]; + _maxEntries = (_numberOfBlocksPerSlab * _numberOfSlabs) - 1; + for (int i = 0; i < _numberOfSlabs; i++) { + if (!lazy) { + if (_directAllocation) { + _slabs[i] = ByteBuffer.allocateDirect(_numberOfBlocksPerSlab * _blockSize); + } else { + _slabs[i] = ByteBuffer.allocate(_numberOfBlocksPerSlab * _blockSize); + } } _locks[i] = new BlockLocks(_numberOfBlocksPerSlab); _lockCounters[i] = new AtomicInteger(); @@ -197,10 +237,66 @@ public class BlockCache { } private ByteBuffer getSlab(int slabId) { - return _slabs[slabId].duplicate(); + if (!lazy) { + return _slabs[slabId].duplicate(); + } else { + ByteBuffer[] byteBuffers = _threadLocalSlabs.get(); + ByteBuffer byteBuffer = byteBuffers[slabId]; + if (byteBuffer == null) { + synchronized (_slabs) { + ByteBuffer bb = _slabs[slabId]; + if (bb == null) { + if (_directAllocation) { + bb = ByteBuffer.allocateDirect(_numberOfBlocksPerSlab * _blockSize); + } else { + bb = ByteBuffer.allocate(_numberOfBlocksPerSlab * _blockSize); + } + _slabs[slabId] = bb; + } + byteBuffer = bb.duplicate(); + } + byteBuffers[slabId] = byteBuffer; + } + return byteBuffer; + } } public int getSize() { return _cache.size(); } + + public void close() throws IOException { + for (ByteBuffer buffer : this._slabs) { + freeBuffer(buffer); + } + } + + /** + * This code was copied form MMAPDirectory in Lucene. + */ + protected void freeBuffer(final ByteBuffer buffer) throws IOException { + if (buffer == null) { + return; + } + if (UNMAP_SUPPORTED) { + try { + AccessController.doPrivileged(new PrivilegedExceptionAction<Void>() { + @Override + public Void run() throws Exception { + final Method getCleanerMethod = buffer.getClass().getMethod("cleaner"); + getCleanerMethod.setAccessible(true); + final Object cleaner = getCleanerMethod.invoke(buffer); + if (cleaner != null) { + cleaner.getClass().getMethod("clean").invoke(cleaner); + } + return null; + } + }); + } catch (PrivilegedActionException e) { + final IOException ioe = new IOException("unable to unmap the mapped buffer"); + ioe.initCause(e.getCause()); + throw ioe; + } + } + } } http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/5ce59476/src/blur-store/src/test/java/org/apache/blur/store/blockcache/BlockCacheTest.java ---------------------------------------------------------------------- diff --git a/src/blur-store/src/test/java/org/apache/blur/store/blockcache/BlockCacheTest.java b/src/blur-store/src/test/java/org/apache/blur/store/blockcache/BlockCacheTest.java index 98b5c74..4665497 100644 --- a/src/blur-store/src/test/java/org/apache/blur/store/blockcache/BlockCacheTest.java +++ b/src/blur-store/src/test/java/org/apache/blur/store/blockcache/BlockCacheTest.java @@ -17,22 +17,57 @@ package org.apache.blur.store.blockcache; * limitations under the License. */ import static org.junit.Assert.assertTrue; +import static org.junit.Assert.fail; +import java.io.IOException; +import java.util.ArrayList; import java.util.Arrays; +import java.util.List; import java.util.Random; +import java.util.concurrent.Callable; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; import java.util.concurrent.atomic.AtomicLong; import org.junit.Test; public class BlockCacheTest { @Test - public void testBlockCache() { - int blocksInTest = 2000000; - int blockSize = BlockCache._8K; - int slabSize = blockSize * 1024; - long totalMemory = 2 * slabSize; + public void testBlockCache() throws IOException, InterruptedException, ExecutionException { + int blocksPerSlab = 1024; + int slabs = 4; + // Test block are larger than cache size. + final int blocksInTest = (blocksPerSlab * slabs) * 2; + final int blockSize = BlockCache._8K; + int slabSize = blockSize * blocksPerSlab; + long totalMemory = slabs * (long) slabSize; + + final BlockCache blockCache = new BlockCache(true, totalMemory, slabSize); + final int threads = 4; + ExecutorService pool = Executors.newFixedThreadPool(threads); + List<Future<Boolean>> futures = new ArrayList<Future<Boolean>>(); + final int testSpace = blocksInTest / threads; + for (int g = 0; g < threads; g++) { + final int file = g; + futures.add(pool.submit(new Callable<Boolean>() { + @Override + public Boolean call() throws Exception { + return runTest(blockSize, file, testSpace, blockCache); + } + })); + } - BlockCache blockCache = new BlockCache(true, totalMemory, slabSize); + for (Future<Boolean> future : futures) { + if (!future.get()) { + fail(); + } + } + blockCache.close(); + } + + private boolean runTest(int blockSize, int file, int blocksInTest, BlockCache blockCache) { byte[] buffer = new byte[blockSize]; Random random = new Random(); @@ -41,13 +76,12 @@ public class BlockCacheTest { AtomicLong missesInCache = new AtomicLong(); long storeTime = 0; long fetchTime = 0; - int passes = 10000; + int passes = 1000000; BlockCacheKey blockCacheKey = new BlockCacheKey(); for (int j = 0; j < passes; j++) { long block = random.nextInt(blocksInTest); - int file = 0; blockCacheKey.setBlock(block); blockCacheKey.setFile(file); @@ -59,20 +93,25 @@ public class BlockCacheTest { byte[] testData = testData(random, blockSize, newData); long t1 = System.nanoTime(); - blockCache.store(blockCacheKey, 0, testData, 0, blockSize); + boolean store = blockCache.store(blockCacheKey, 0, testData, 0, blockSize); storeTime += (System.nanoTime() - t1); - long t3 = System.nanoTime(); - if (blockCache.fetch(blockCacheKey, buffer)) { - fetchTime += (System.nanoTime() - t3); - assertTrue(Arrays.equals(testData, buffer)); + if (store) { + long t3 = System.nanoTime(); + if (blockCache.fetch(blockCacheKey, buffer)) { + fetchTime += (System.nanoTime() - t3); + if (!Arrays.equals(testData, buffer)) { + return false; + } + } } } System.out.println("Cache Hits = " + hitsInCache.get()); System.out.println("Cache Misses = " + missesInCache.get()); - System.out.println("Store = " + (storeTime / (double) passes) / 1000000.0); - System.out.println("Fetch = " + (fetchTime / (double) passes) / 1000000.0); + System.out.println("Store = avg " + (storeTime / (double) passes) / 1000000.0 + " ms"); + System.out.println("Fetch = avg " + (fetchTime / (double) passes) / 1000000.0 + " ms"); System.out.println("# of Elements = " + blockCache.getSize()); + return true; } /** @@ -89,16 +128,16 @@ public class BlockCacheTest { BlockCacheKey blockCacheKey = new BlockCacheKey(); blockCacheKey.setBlock(0); blockCacheKey.setFile(0); - byte[] newData = new byte[blockSize*3]; + byte[] newData = new byte[blockSize * 3]; byte[] testData = testData(random, blockSize, newData); assertTrue(blockCache.store(blockCacheKey, 0, testData, 0, blockSize)); assertTrue(blockCache.store(blockCacheKey, 0, testData, blockSize, blockSize)); - assertTrue(blockCache.store(blockCacheKey, 0, testData, blockSize*2, blockSize)); + assertTrue(blockCache.store(blockCacheKey, 0, testData, blockSize * 2, blockSize)); assertTrue(blockCache.store(blockCacheKey, 1, testData, 0, blockSize - 1)); assertTrue(blockCache.store(blockCacheKey, 1, testData, blockSize, blockSize - 1)); - assertTrue(blockCache.store(blockCacheKey, 1, testData, blockSize*2, blockSize - 1)); + assertTrue(blockCache.store(blockCacheKey, 1, testData, blockSize * 2, blockSize - 1)); } private static byte[] testData(Random random, int size, byte[] buf) {
