Updated Branches: refs/heads/apache-blur-0.2 4d5e5ba34 -> 5d026a3dd
Fixed BLUR-294 Project: http://git-wip-us.apache.org/repos/asf/incubator-blur/repo Commit: http://git-wip-us.apache.org/repos/asf/incubator-blur/commit/5d026a3d Tree: http://git-wip-us.apache.org/repos/asf/incubator-blur/tree/5d026a3d Diff: http://git-wip-us.apache.org/repos/asf/incubator-blur/diff/5d026a3d Branch: refs/heads/apache-blur-0.2 Commit: 5d026a3dd1763a2f70119b9f7e68d5e2a026b06b Parents: 4d5e5ba Author: Aaron McCurry <[email protected]> Authored: Tue Nov 5 21:12:55 2013 -0500 Committer: Aaron McCurry <[email protected]> Committed: Tue Nov 5 21:12:55 2013 -0500 ---------------------------------------------------------------------- .../blur/thrift/ThriftBlurShardServer.java | 15 +- .../test/java/org/apache/blur/MiniCluster.java | 2 +- .../blur/manager/writer/IndexImporterTest.java | 35 +++-- .../utils/TableShardCountCollapserTest.java | 3 +- .../mapreduce/lib/BlurOutputFormatTest.java | 4 +- .../blur/store/blockcache/BlockDirectory.java | 7 +- .../store/blockcache_v2/CacheIndexInput.java | 7 +- .../store/blockcache_v2/CacheIndexOutput.java | 7 +- .../apache/blur/store/buffer/BufferStore.java | 138 +++++++++++-------- .../store/buffer/ReusedBufferedIndexInput.java | 7 +- .../store/buffer/ReusedBufferedIndexOutput.java | 7 +- .../org/apache/blur/store/buffer/Store.java | 25 ++++ .../blur/store/BaseDirectoryTestSuite.java | 4 +- .../blur/store/CacheDirectoryTestSuite.java | 3 +- .../store/blockcache/BlockDirectoryTest.java | 3 +- .../store/blockcache_v2/CacheDirectoryTest.java | 3 +- .../blockcache_v2/CacheIndexInputTest.java | 5 +- .../blockcache_v2/CacheIndexOutputTest.java | 3 +- .../apache/blur/metrics/MetricsConstants.java | 4 +- .../src/main/resources/blur-default.properties | 8 +- 20 files changed, 183 insertions(+), 107 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/5d026a3d/blur-core/src/main/java/org/apache/blur/thrift/ThriftBlurShardServer.java ---------------------------------------------------------------------- diff --git a/blur-core/src/main/java/org/apache/blur/thrift/ThriftBlurShardServer.java b/blur-core/src/main/java/org/apache/blur/thrift/ThriftBlurShardServer.java index f46c975..4281e01 100644 --- a/blur-core/src/main/java/org/apache/blur/thrift/ThriftBlurShardServer.java +++ b/blur-core/src/main/java/org/apache/blur/thrift/ThriftBlurShardServer.java @@ -49,6 +49,8 @@ import static org.apache.blur.utils.BlurConstants.BLUR_ZOOKEEPER_TIMEOUT_DEFAULT import static org.apache.blur.utils.BlurUtil.quietClose; import java.lang.reflect.Constructor; +import java.util.Map.Entry; +import java.util.Set; import java.util.concurrent.TimeUnit; import org.apache.blur.BlurConfiguration; @@ -148,9 +150,16 @@ public class ThriftBlurShardServer extends ThriftServer { httpServer = null; } - int _1024Size = configuration.getInt("blur.shard.buffercache.1024", 8192); - int _8192Size = configuration.getInt("blur.shard.buffercache.8192", 8192); - BufferStore.init(_1024Size, _8192Size); + Set<Entry<String, String>> set = configuration.getProperties().entrySet(); + for (Entry<String, String> e : set) { + String key = e.getKey(); + if (key.startsWith("blur.shard.buffercache.")) { + int index = key.lastIndexOf('.'); + int bufferSize = Integer.parseInt(key.substring(index + 1)); + long amount = Long.parseLong(e.getValue()); + BufferStore.initNewBuffer(bufferSize, amount); + } + } BlockCacheDirectoryFactory blockCacheDirectoryFactory; // Alternate BlockCacheDirectoryFactory support currently disabled in 0.2.0, http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/5d026a3d/blur-core/src/test/java/org/apache/blur/MiniCluster.java ---------------------------------------------------------------------- diff --git a/blur-core/src/test/java/org/apache/blur/MiniCluster.java b/blur-core/src/test/java/org/apache/blur/MiniCluster.java index 01b34a4..d8a40fb 100644 --- a/blur-core/src/test/java/org/apache/blur/MiniCluster.java +++ b/blur-core/src/test/java/org/apache/blur/MiniCluster.java @@ -130,7 +130,7 @@ public class MiniCluster { } private void setupBuffers() { - BufferStore.init(16, 16); + BufferStore.initNewBuffer(128, 128 * 128); } public void shutdownBlurCluster() { http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/5d026a3d/blur-core/src/test/java/org/apache/blur/manager/writer/IndexImporterTest.java ---------------------------------------------------------------------- diff --git a/blur-core/src/test/java/org/apache/blur/manager/writer/IndexImporterTest.java b/blur-core/src/test/java/org/apache/blur/manager/writer/IndexImporterTest.java index e7724a5..ac09d89 100644 --- a/blur-core/src/test/java/org/apache/blur/manager/writer/IndexImporterTest.java +++ b/blur-core/src/test/java/org/apache/blur/manager/writer/IndexImporterTest.java @@ -80,10 +80,10 @@ public class IndexImporterTest { TableDescriptor tableDescriptor = new TableDescriptor(); tableDescriptor.setName("test-table"); String uuid = UUID.randomUUID().toString(); - + tableDescriptor.setTableUri(new Path(base, "table-table").toUri().toString()); tableDescriptor.setShardCount(2); - + TableContext tableContext = TableContext.create(tableDescriptor); ShardContext shardContext = ShardContext.create(tableContext, "shard-00000000"); Path tablePath = new Path(base, "table-table"); @@ -98,10 +98,10 @@ public class IndexImporterTest { Analyzer analyzerForIndex = _fieldManager.getAnalyzerForIndex(); IndexWriterConfig conf = new IndexWriterConfig(LUCENE_VERSION, analyzerForIndex); commitWriter = new IndexWriter(commitDirectory, conf); - + mainWriter = new IndexWriter(mainDirectory, conf); - BufferStore.init(128, 128); - + BufferStore.initNewBuffer(128, 128 * 128); + indexImporter = new IndexImporter(new TrackingIndexWriter(mainWriter), new ReentrantReadWriteLock(), shardContext, TimeUnit.MINUTES, 10); } @@ -113,7 +113,6 @@ public class IndexImporterTest { base.getFileSystem(configuration).delete(base, true); } - @Test public void testIndexImporterWithCorrectRowIdShardCombination() throws IOException { List<Field> document = _fieldManager.getFields("1", genRecord("1")); @@ -125,18 +124,18 @@ public class IndexImporterTest { assertFalse(fileSystem.exists(badRowIdsPath)); } -// private void debug(Path file) throws IOException { -// if (!fileSystem.exists(file)) { -// return; -// } -// System.out.println(file); -// if (!fileSystem.isFile(file)) { -// FileStatus[] listStatus = fileSystem.listStatus(file); -// for (FileStatus f : listStatus) { -// debug(f.getPath()); -// } -// } -// } + // private void debug(Path file) throws IOException { + // if (!fileSystem.exists(file)) { + // return; + // } + // System.out.println(file); + // if (!fileSystem.isFile(file)) { + // FileStatus[] listStatus = fileSystem.listStatus(file); + // for (FileStatus f : listStatus) { + // debug(f.getPath()); + // } + // } + // } @Test public void testIndexImporterWithWrongRowIdShardCombination() throws IOException { http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/5d026a3d/blur-core/src/test/java/org/apache/blur/utils/TableShardCountCollapserTest.java ---------------------------------------------------------------------- diff --git a/blur-core/src/test/java/org/apache/blur/utils/TableShardCountCollapserTest.java b/blur-core/src/test/java/org/apache/blur/utils/TableShardCountCollapserTest.java index 28bf718..a2f3be1 100644 --- a/blur-core/src/test/java/org/apache/blur/utils/TableShardCountCollapserTest.java +++ b/blur-core/src/test/java/org/apache/blur/utils/TableShardCountCollapserTest.java @@ -1,4 +1,5 @@ package org.apache.blur.utils; + /** * Licensed to the Apache Software Foundation (ASF) under one or more * contributor license agreements. See the NOTICE file distributed with @@ -48,7 +49,7 @@ public class TableShardCountCollapserTest { @Before public void setup() throws IOException { - BufferStore.init(128, 128); + BufferStore.initNewBuffer(128, 128 * 128); configuration = new Configuration(); path = new Path("./target/tmp-shards-for-testing"); FileSystem fileSystem = path.getFileSystem(configuration); http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/5d026a3d/blur-mapred/src/test/java/org/apache/blur/mapreduce/lib/BlurOutputFormatTest.java ---------------------------------------------------------------------- diff --git a/blur-mapred/src/test/java/org/apache/blur/mapreduce/lib/BlurOutputFormatTest.java b/blur-mapred/src/test/java/org/apache/blur/mapreduce/lib/BlurOutputFormatTest.java index faa3266..0666368 100644 --- a/blur-mapred/src/test/java/org/apache/blur/mapreduce/lib/BlurOutputFormatTest.java +++ b/blur-mapred/src/test/java/org/apache/blur/mapreduce/lib/BlurOutputFormatTest.java @@ -73,7 +73,7 @@ public class BlurOutputFormatTest { } mr = new MiniMRCluster(1, "file:///", 1); jobConf = mr.createJobConf(); - BufferStore.init(128, 128); + BufferStore.initNewBuffer(128, 128 * 128); } @AfterClass @@ -318,7 +318,7 @@ public class BlurOutputFormatTest { } // @TODO this test to fail sometimes due to issues in the MR MiniCluster -// @Test + // @Test public void testBlurOutputFormatCleanupDuringJobKillTest() throws IOException, InterruptedException, ClassNotFoundException { localFs.delete(new Path(TEST_ROOT_DIR + "/in"), true); http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/5d026a3d/blur-store/src/main/java/org/apache/blur/store/blockcache/BlockDirectory.java ---------------------------------------------------------------------- diff --git a/blur-store/src/main/java/org/apache/blur/store/blockcache/BlockDirectory.java b/blur-store/src/main/java/org/apache/blur/store/blockcache/BlockDirectory.java index 041b10d..69a1320 100644 --- a/blur-store/src/main/java/org/apache/blur/store/blockcache/BlockDirectory.java +++ b/blur-store/src/main/java/org/apache/blur/store/blockcache/BlockDirectory.java @@ -24,6 +24,7 @@ import java.util.Set; import org.apache.blur.store.buffer.BufferStore; import org.apache.blur.store.buffer.ReusedBufferedIndexInput; +import org.apache.blur.store.buffer.Store; import org.apache.blur.store.hdfs.DirectoryDecorator; import org.apache.blur.store.hdfs.HdfsDirectory; import org.apache.lucene.store.Directory; @@ -132,6 +133,7 @@ public class BlockDirectory extends Directory implements DirectoryDecorator { private long _fileLength; private String _cacheName; private Cache _cache; + private final Store _store; public CachedIndexInput(IndexInput source, int blockSize, String name, String cacheName, Cache cache, IOContext context) { @@ -141,6 +143,7 @@ public class BlockDirectory extends Directory implements DirectoryDecorator { _fileLength = source.length(); _cacheName = cacheName; _cache = cache; + _store = BufferStore.instance(_blockSize); } @Override @@ -189,11 +192,11 @@ public class BlockDirectory extends Directory implements DirectoryDecorator { int length = (int) Math.min(_blockSize, _fileLength - position); _source.seek(position); - byte[] buf = BufferStore.takeBuffer(_blockSize); + byte[] buf = _store.takeBuffer(_blockSize); _source.readBytes(buf, 0, length); System.arraycopy(buf, blockOffset, b, off, lengthToReadInBlock); _cache.update(_cacheName, blockId, 0, buf, 0, _blockSize); - BufferStore.putBuffer(buf); + _store.putBuffer(buf); } private boolean checkCache(long blockId, int blockOffset, byte[] b, int off, int lengthToReadInBlock) { http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/5d026a3d/blur-store/src/main/java/org/apache/blur/store/blockcache_v2/CacheIndexInput.java ---------------------------------------------------------------------- diff --git a/blur-store/src/main/java/org/apache/blur/store/blockcache_v2/CacheIndexInput.java b/blur-store/src/main/java/org/apache/blur/store/blockcache_v2/CacheIndexInput.java index a91cf2e..95d9689 100644 --- a/blur-store/src/main/java/org/apache/blur/store/blockcache_v2/CacheIndexInput.java +++ b/blur-store/src/main/java/org/apache/blur/store/blockcache_v2/CacheIndexInput.java @@ -20,6 +20,7 @@ package org.apache.blur.store.blockcache_v2; import java.io.IOException; import org.apache.blur.store.buffer.BufferStore; +import org.apache.blur.store.buffer.Store; import org.apache.lucene.store.AlreadyClosedException; import org.apache.lucene.store.IndexInput; @@ -32,6 +33,7 @@ public class CacheIndexInput extends IndexInput { private final CacheDirectory _directory; private final String _fileName; private final Cache _cache; + private final Store _store; private IndexInput _indexInput; private CacheKey _key = new CacheKey(); @@ -57,6 +59,7 @@ public class CacheIndexInput extends IndexInput { _quiet = _cache.shouldBeQuiet(_directory, _fileName); _key.setFileId(_fileId); _isClosed = false; + _store = BufferStore.instance(_bufferSize); } @Override @@ -294,7 +297,7 @@ public class CacheIndexInput extends IndexInput { _cacheValue = _cache.newInstance(_directory, _fileName); long filePosition = getFilePosition(); _indexInput.seek(filePosition); - byte[] buffer = BufferStore.takeBuffer(_bufferSize); + byte[] buffer = _store.takeBuffer(_bufferSize); int len = (int) Math.min(_cacheBlockSize, _fileLength - filePosition); int cachePosition = 0; while (len > 0) { @@ -304,7 +307,7 @@ public class CacheIndexInput extends IndexInput { len -= length; cachePosition += length; } - BufferStore.putBuffer(buffer); + _store.putBuffer(buffer); _cache.put(_key.clone(), _cacheValue); } _cacheValue.incRef(); http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/5d026a3d/blur-store/src/main/java/org/apache/blur/store/blockcache_v2/CacheIndexOutput.java ---------------------------------------------------------------------- diff --git a/blur-store/src/main/java/org/apache/blur/store/blockcache_v2/CacheIndexOutput.java b/blur-store/src/main/java/org/apache/blur/store/blockcache_v2/CacheIndexOutput.java index 959e267..c5dde83 100644 --- a/blur-store/src/main/java/org/apache/blur/store/blockcache_v2/CacheIndexOutput.java +++ b/blur-store/src/main/java/org/apache/blur/store/blockcache_v2/CacheIndexOutput.java @@ -20,6 +20,7 @@ package org.apache.blur.store.blockcache_v2; import java.io.IOException; import org.apache.blur.store.buffer.BufferStore; +import org.apache.blur.store.buffer.Store; import org.apache.lucene.store.IndexOutput; public class CacheIndexOutput extends IndexOutput { @@ -31,6 +32,7 @@ public class CacheIndexOutput extends IndexOutput { private final long _fileId; private final int _fileBufferSize; private final int _cacheBlockSize; + private final Store _store; private long _position; private byte[] _buffer; @@ -45,7 +47,8 @@ public class CacheIndexOutput extends IndexOutput { _cacheBlockSize = _cache.getCacheBlockSize(_directory, _fileName); _fileId = _cache.getFileId(_directory, _fileName); _indexOutput = indexOutput; - _buffer = BufferStore.takeBuffer(_cacheBlockSize); + _store = BufferStore.instance(_cacheBlockSize); + _buffer = _store.takeBuffer(_cacheBlockSize); } @Override @@ -119,7 +122,7 @@ public class CacheIndexOutput extends IndexOutput { flushInternal(); _indexOutput.flush(); _indexOutput.close(); - BufferStore.putBuffer(_buffer); + _store.putBuffer(_buffer); } @Override http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/5d026a3d/blur-store/src/main/java/org/apache/blur/store/buffer/BufferStore.java ---------------------------------------------------------------------- diff --git a/blur-store/src/main/java/org/apache/blur/store/buffer/BufferStore.java b/blur-store/src/main/java/org/apache/blur/store/buffer/BufferStore.java index 2701726..83f47c2 100644 --- a/blur-store/src/main/java/org/apache/blur/store/buffer/BufferStore.java +++ b/blur-store/src/main/java/org/apache/blur/store/buffer/BufferStore.java @@ -20,12 +20,12 @@ import static org.apache.blur.metrics.MetricsConstants.INTERNAL_BUFFERS; import static org.apache.blur.metrics.MetricsConstants.LOST; import static org.apache.blur.metrics.MetricsConstants.LUCENE; import static org.apache.blur.metrics.MetricsConstants.ORG_APACHE_BLUR; -import static org.apache.blur.metrics.MetricsConstants.OTHER_SIZES_ALLOCATED; -import static org.apache.blur.metrics.MetricsConstants._1K_SIZE_ALLOCATED; -import static org.apache.blur.metrics.MetricsConstants._8K_SIZE_ALLOCATED; +import static org.apache.blur.metrics.MetricsConstants.SIZE_ALLOCATED; import java.util.concurrent.ArrayBlockingQueue; import java.util.concurrent.BlockingQueue; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ConcurrentMap; import java.util.concurrent.TimeUnit; import org.apache.blur.log.Log; @@ -35,31 +35,63 @@ import com.yammer.metrics.Metrics; import com.yammer.metrics.core.Meter; import com.yammer.metrics.core.MetricName; -public class BufferStore { +public class BufferStore implements Store { private static final Log LOG = LogFactory.getLog(BufferStore.class); - private static BlockingQueue<byte[]> _1024; - private static BlockingQueue<byte[]> _8192; - - private static Meter _lost; - private static Meter _8K; - private static Meter _1K; - private static Meter _other; - private volatile static boolean setup = false; - - public static void init(int _1KSize, int _8KSize) { - if (!setup) { - _lost = Metrics.newMeter(new MetricName(ORG_APACHE_BLUR, LUCENE, LOST, INTERNAL_BUFFERS), INTERNAL_BUFFERS, TimeUnit.SECONDS); - _8K = Metrics.newMeter(new MetricName(ORG_APACHE_BLUR, LUCENE, _1K_SIZE_ALLOCATED, INTERNAL_BUFFERS), INTERNAL_BUFFERS, TimeUnit.SECONDS); - _1K = Metrics.newMeter(new MetricName(ORG_APACHE_BLUR, LUCENE, _8K_SIZE_ALLOCATED, INTERNAL_BUFFERS), INTERNAL_BUFFERS, TimeUnit.SECONDS); - _other = Metrics.newMeter(new MetricName(ORG_APACHE_BLUR, LUCENE, OTHER_SIZES_ALLOCATED, INTERNAL_BUFFERS), INTERNAL_BUFFERS, TimeUnit.SECONDS); - LOG.info("Initializing the 1024 buffers with [{0}] buffers.", _1KSize); - _1024 = setupBuffers(1024, _1KSize, _1K); - LOG.info("Initializing the 8192 buffers with [{0}] buffers.", _8KSize); - _8192 = setupBuffers(8192, _8KSize, _8K); - setup = true; + private static final Store EMPTY = new Store() { + + private final Meter _emptyCreated; + private final Meter _emptyLost; + + { + _emptyCreated = Metrics.newMeter(new MetricName(ORG_APACHE_BLUR, LUCENE, "Not Configured " + SIZE_ALLOCATED, + INTERNAL_BUFFERS), INTERNAL_BUFFERS, TimeUnit.SECONDS); + _emptyLost = Metrics.newMeter( + new MetricName(ORG_APACHE_BLUR, LUCENE, "Not Configured " + LOST, INTERNAL_BUFFERS), INTERNAL_BUFFERS, + TimeUnit.SECONDS); + } + + @Override + public byte[] takeBuffer(int bufferSize) { + _emptyCreated.mark(); + return new byte[bufferSize]; + } + + @Override + public void putBuffer(byte[] buffer) { + _emptyLost.mark(); } + }; + + private final static ConcurrentMap<Integer, BufferStore> _bufferStores = new ConcurrentHashMap<Integer, BufferStore>(); + + private final BlockingQueue<byte[]> _buffers; + private final Meter _created; + private final Meter _lost; + private final int _bufferSize; + + public synchronized static void initNewBuffer(int bufferSize, long totalAmount) { + BufferStore bufferStore = _bufferStores.get(bufferSize); + if (bufferStore == null) { + long count = totalAmount / bufferSize; + if (count > Integer.MAX_VALUE) { + count = Integer.MAX_VALUE; + } + BufferStore store = new BufferStore(bufferSize, (int) count); + _bufferStores.put(bufferSize, store); + } else { + LOG.warn("Buffer store for size [{0}] already setup.", bufferSize); + } + } + + private BufferStore(int bufferSize, int count) { + _bufferSize = bufferSize; + _created = Metrics.newMeter(new MetricName(ORG_APACHE_BLUR, LUCENE, bufferSize + " " + SIZE_ALLOCATED, + INTERNAL_BUFFERS), INTERNAL_BUFFERS, TimeUnit.SECONDS); + _lost = Metrics.newMeter(new MetricName(ORG_APACHE_BLUR, LUCENE, bufferSize + " " + LOST, INTERNAL_BUFFERS), + INTERNAL_BUFFERS, TimeUnit.SECONDS); + _buffers = setupBuffers(bufferSize, count, _created); } private static BlockingQueue<byte[]> setupBuffers(int bufferSize, int count, Meter meter) { @@ -71,56 +103,46 @@ public class BufferStore { return queue; } - public static byte[] takeBuffer(int bufferSize) { - switch (bufferSize) { - case 1024: - return newBuffer1024(_1024.poll()); - case 8192: - return newBuffer8192(_8192.poll()); - default: - return newBuffer(bufferSize); + public static Store instance(int bufferSize) { + BufferStore bufferStore = _bufferStores.get(bufferSize); + if (bufferStore == null) { + return EMPTY; } + return bufferStore; } - public static void putBuffer(byte[] buffer) { + @Override + public byte[] takeBuffer(int bufferSize) { + if (bufferSize != _bufferSize) { + throw new RuntimeException("Buffer with length [" + bufferSize + "] does not match buffer size of [" + + _bufferSize + "]"); + } + return newBuffer(_buffers.poll()); + } + + @Override + public void putBuffer(byte[] buffer) { if (buffer == null) { return; } - int bufferSize = buffer.length; - switch (bufferSize) { - case 1024: - checkReturn(_1024.offer(buffer)); - return; - case 8192: - checkReturn(_8192.offer(buffer)); - return; + if (buffer.length != _bufferSize) { + throw new RuntimeException("Buffer with length [" + buffer.length + "] does not match buffer size of [" + + _bufferSize + "]"); } + checkReturn(_buffers.offer(buffer)); } - private static void checkReturn(boolean offer) { + private void checkReturn(boolean offer) { if (!offer) { _lost.mark(); } } - private static byte[] newBuffer1024(byte[] buf) { - if (buf != null) { - return buf; - } - _1K.mark(); - return new byte[1024]; - } - - private static byte[] newBuffer8192(byte[] buf) { + private byte[] newBuffer(byte[] buf) { if (buf != null) { return buf; } - _8K.mark(); - return new byte[8192]; - } - - private static byte[] newBuffer(int size) { - _other.mark(); - return new byte[size]; + _created.mark(); + return new byte[_bufferSize]; } } http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/5d026a3d/blur-store/src/main/java/org/apache/blur/store/buffer/ReusedBufferedIndexInput.java ---------------------------------------------------------------------- diff --git a/blur-store/src/main/java/org/apache/blur/store/buffer/ReusedBufferedIndexInput.java b/blur-store/src/main/java/org/apache/blur/store/buffer/ReusedBufferedIndexInput.java index 23c09d3..93f1347 100644 --- a/blur-store/src/main/java/org/apache/blur/store/buffer/ReusedBufferedIndexInput.java +++ b/blur-store/src/main/java/org/apache/blur/store/buffer/ReusedBufferedIndexInput.java @@ -31,7 +31,7 @@ public abstract class ReusedBufferedIndexInput extends IndexInput { @Override public final void close() throws IOException { closeInternal(); - BufferStore.putBuffer(buffer); + store.putBuffer(buffer); buffer = null; } @@ -47,7 +47,7 @@ public abstract class ReusedBufferedIndexInput extends IndexInput { throw new EOFException("read past EOF: " + this); if (buffer == null) { - buffer = BufferStore.takeBuffer(bufferSize); + buffer = store.takeBuffer(bufferSize); seekInternal(bufferStart); } readInternal(buffer, 0, newLength); @@ -82,6 +82,8 @@ public abstract class ReusedBufferedIndexInput extends IndexInput { private int bufferLength = 0; // end of valid bytes private int bufferPosition = 0; // next byte to read + private Store store; + @Override public final byte readByte() throws IOException { if (bufferPosition >= bufferLength) @@ -102,6 +104,7 @@ public abstract class ReusedBufferedIndexInput extends IndexInput { super(resourceDesc); checkBufferSize(bufferSize); this.bufferSize = bufferSize; + this.store = BufferStore.instance(bufferSize); } /** Returns buffer size. @see #setBufferSize */ http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/5d026a3d/blur-store/src/main/java/org/apache/blur/store/buffer/ReusedBufferedIndexOutput.java ---------------------------------------------------------------------- diff --git a/blur-store/src/main/java/org/apache/blur/store/buffer/ReusedBufferedIndexOutput.java b/blur-store/src/main/java/org/apache/blur/store/buffer/ReusedBufferedIndexOutput.java index c4576d5..bcb9511 100644 --- a/blur-store/src/main/java/org/apache/blur/store/buffer/ReusedBufferedIndexOutput.java +++ b/blur-store/src/main/java/org/apache/blur/store/buffer/ReusedBufferedIndexOutput.java @@ -37,6 +37,8 @@ public abstract class ReusedBufferedIndexOutput extends IndexOutput { private int bufferPosition = 0; /** total length of the file */ private long _fileLength = 0; + + private Store store; public ReusedBufferedIndexOutput() { this(BUFFER_SIZE); @@ -45,7 +47,8 @@ public abstract class ReusedBufferedIndexOutput extends IndexOutput { public ReusedBufferedIndexOutput(int bufferSize) { checkBufferSize(bufferSize); this.bufferSize = bufferSize; - buffer = BufferStore.takeBuffer(this.bufferSize); + store = BufferStore.instance(bufferSize); + buffer = store.takeBuffer(this.bufferSize); } protected long getBufferStart() { @@ -80,7 +83,7 @@ public abstract class ReusedBufferedIndexOutput extends IndexOutput { public void close() throws IOException { flushBufferToCache(); closeInternal(); - BufferStore.putBuffer(buffer); + store.putBuffer(buffer); buffer = null; } http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/5d026a3d/blur-store/src/main/java/org/apache/blur/store/buffer/Store.java ---------------------------------------------------------------------- diff --git a/blur-store/src/main/java/org/apache/blur/store/buffer/Store.java b/blur-store/src/main/java/org/apache/blur/store/buffer/Store.java new file mode 100644 index 0000000..541089c --- /dev/null +++ b/blur-store/src/main/java/org/apache/blur/store/buffer/Store.java @@ -0,0 +1,25 @@ +/** + * 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.blur.store.buffer; + +public interface Store { + + byte[] takeBuffer(int bufferSize); + + void putBuffer(byte[] buffer); + +} http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/5d026a3d/blur-store/src/test/java/org/apache/blur/store/BaseDirectoryTestSuite.java ---------------------------------------------------------------------- diff --git a/blur-store/src/test/java/org/apache/blur/store/BaseDirectoryTestSuite.java b/blur-store/src/test/java/org/apache/blur/store/BaseDirectoryTestSuite.java index 2806e39..a70a1fe 100644 --- a/blur-store/src/test/java/org/apache/blur/store/BaseDirectoryTestSuite.java +++ b/blur-store/src/test/java/org/apache/blur/store/BaseDirectoryTestSuite.java @@ -70,7 +70,9 @@ public abstract class BaseDirectoryTestSuite { @Before public void setUp() throws IOException { - BufferStore.init(128, 128); + BufferStore.initNewBuffer(1024, 1024 * 128); + BufferStore.initNewBuffer(8192, 8192 * 128); + file = new File(TMPDIR, "hdfsdirectorytest"); fileControl = new File(TMPDIR, "hdfsdirectorytest-control"); rm(file); http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/5d026a3d/blur-store/src/test/java/org/apache/blur/store/CacheDirectoryTestSuite.java ---------------------------------------------------------------------- diff --git a/blur-store/src/test/java/org/apache/blur/store/CacheDirectoryTestSuite.java b/blur-store/src/test/java/org/apache/blur/store/CacheDirectoryTestSuite.java index 9960d96..c342874 100644 --- a/blur-store/src/test/java/org/apache/blur/store/CacheDirectoryTestSuite.java +++ b/blur-store/src/test/java/org/apache/blur/store/CacheDirectoryTestSuite.java @@ -75,7 +75,8 @@ public abstract class CacheDirectoryTestSuite extends BaseDirectoryTestSuite { getStore()); Directory dir = FSDirectory.open(new File(file, "cache")); - BufferStore.init(128, 128); + BufferStore.initNewBuffer(1024, 1024 * 128); + BufferStore.initNewBuffer(8192, 8192 * 128); return new CacheDirectory("test", "test", wrapLastModified(dir), _cache, null); } http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/5d026a3d/blur-store/src/test/java/org/apache/blur/store/blockcache/BlockDirectoryTest.java ---------------------------------------------------------------------- diff --git a/blur-store/src/test/java/org/apache/blur/store/blockcache/BlockDirectoryTest.java b/blur-store/src/test/java/org/apache/blur/store/blockcache/BlockDirectoryTest.java index 7702b97..b3edee6 100644 --- a/blur-store/src/test/java/org/apache/blur/store/blockcache/BlockDirectoryTest.java +++ b/blur-store/src/test/java/org/apache/blur/store/blockcache/BlockDirectoryTest.java @@ -102,7 +102,8 @@ public class BlockDirectoryTest { @Before public void setUp() throws IOException { - BufferStore.init(128, 128); + BufferStore.initNewBuffer(1024, 1024 * 128); + BufferStore.initNewBuffer(8192, 8192 * 128); file = new File(TMPDIR, "blockdirectorytest"); rm(file); file.mkdirs(); http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/5d026a3d/blur-store/src/test/java/org/apache/blur/store/blockcache_v2/CacheDirectoryTest.java ---------------------------------------------------------------------- diff --git a/blur-store/src/test/java/org/apache/blur/store/blockcache_v2/CacheDirectoryTest.java b/blur-store/src/test/java/org/apache/blur/store/blockcache_v2/CacheDirectoryTest.java index 800191f..c173aae 100644 --- a/blur-store/src/test/java/org/apache/blur/store/blockcache_v2/CacheDirectoryTest.java +++ b/blur-store/src/test/java/org/apache/blur/store/blockcache_v2/CacheDirectoryTest.java @@ -90,7 +90,8 @@ public class CacheDirectoryTest { _cache = new BaseCache(totalNumberOfBytes, fileBufferSize, cacheBlockSize, readFilter, writeFilter, quiet, STORE.ON_HEAP); Directory directory = newDirectory(); - BufferStore.init(128, 128); + BufferStore.initNewBuffer(1024, 1024 * 128); + BufferStore.initNewBuffer(8192, 8192 * 128); _cacheDirectory = new CacheDirectory("test", "test", directory, _cache, null); } http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/5d026a3d/blur-store/src/test/java/org/apache/blur/store/blockcache_v2/CacheIndexInputTest.java ---------------------------------------------------------------------- diff --git a/blur-store/src/test/java/org/apache/blur/store/blockcache_v2/CacheIndexInputTest.java b/blur-store/src/test/java/org/apache/blur/store/blockcache_v2/CacheIndexInputTest.java index 77158bf..7b8ab82 100644 --- a/blur-store/src/test/java/org/apache/blur/store/blockcache_v2/CacheIndexInputTest.java +++ b/blur-store/src/test/java/org/apache/blur/store/blockcache_v2/CacheIndexInputTest.java @@ -47,7 +47,8 @@ public class CacheIndexInputTest { @Before public void setup() { - BufferStore.init(128, 128); + BufferStore.initNewBuffer(1024, 1024 * 128); + BufferStore.initNewBuffer(8192, 8192 * 128); seed = new Random().nextLong(); System.out.println("Using seed [" + seed + "]"); // seed = -265282183286396219l; @@ -258,7 +259,7 @@ public class CacheIndexInputTest { @Override public void close() throws IOException { - + } }; http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/5d026a3d/blur-store/src/test/java/org/apache/blur/store/blockcache_v2/CacheIndexOutputTest.java ---------------------------------------------------------------------- diff --git a/blur-store/src/test/java/org/apache/blur/store/blockcache_v2/CacheIndexOutputTest.java b/blur-store/src/test/java/org/apache/blur/store/blockcache_v2/CacheIndexOutputTest.java index 393e7d8..e470d40 100644 --- a/blur-store/src/test/java/org/apache/blur/store/blockcache_v2/CacheIndexOutputTest.java +++ b/blur-store/src/test/java/org/apache/blur/store/blockcache_v2/CacheIndexOutputTest.java @@ -41,7 +41,8 @@ public class CacheIndexOutputTest { @Before public void setup() { - BufferStore.init(128, 128); + BufferStore.initNewBuffer(1024, 1024 * 128); + BufferStore.initNewBuffer(8192, 8192 * 128); seed = new Random().nextLong(); System.out.println("Using seed [" + seed + "]"); // seed = -265282183286396219l; http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/5d026a3d/blur-util/src/main/java/org/apache/blur/metrics/MetricsConstants.java ---------------------------------------------------------------------- diff --git a/blur-util/src/main/java/org/apache/blur/metrics/MetricsConstants.java b/blur-util/src/main/java/org/apache/blur/metrics/MetricsConstants.java index 48eca37..97b1aa8 100644 --- a/blur-util/src/main/java/org/apache/blur/metrics/MetricsConstants.java +++ b/blur-util/src/main/java/org/apache/blur/metrics/MetricsConstants.java @@ -25,9 +25,7 @@ public class MetricsConstants { public static final String BLUR = "Blur"; public static final String ORG_APACHE_BLUR = "org.apache.blur"; public static final String INTERNAL_BUFFERS = "Internal Buffers"; - public static final String OTHER_SIZES_ALLOCATED = "Other Sizes Allocated"; - public static final String _8K_SIZE_ALLOCATED = "8K Size Allocated"; - public static final String _1K_SIZE_ALLOCATED = "1K Size Allocated"; + public static final String SIZE_ALLOCATED = "Size Allocated"; public static final String LOST = "Lost"; public static final String THRIFT_CALLS = "Thrift Calls in \u00B5s"; public static final String REQUESTS = "Requests"; http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/5d026a3d/blur-util/src/main/resources/blur-default.properties ---------------------------------------------------------------------- diff --git a/blur-util/src/main/resources/blur-default.properties b/blur-util/src/main/resources/blur-default.properties index ffcd783..4e4c997 100644 --- a/blur-util/src/main/resources/blur-default.properties +++ b/blur-util/src/main/resources/blur-default.properties @@ -115,11 +115,11 @@ blur.shard.block.cache.v2.write.nocache.ext=fdt # v2 version of block cache only. This specifies the default behavior if a file type is not specified in the cache or nocache lists during writes. Values true | false blur.shard.block.cache.v2.write.default=true -# The number of 1K byte buffers -blur.shard.buffercache.1024=8192 +# The amount of memory to be used by 8K byte buffers. Note if you change the "blur.shard.block.cache.v2.cacheBlockSize" or "blur.shard.block.cache.v2.fileBufferSize" you should adjust the buffer sizes as well as the total memory allocated. For example if you increased the "blur.shard.block.cache.v2.fileBufferSize" to 64K (65536) then this property should to "blur.shard.buffercache.65536". You can also define as many of these properties as needed. +blur.shard.buffercache.8192=67108864 -# The number of 8K byte buffers -blur.shard.buffercache.8192=8192 +# The amount of memory to be used by 1K byte buffers. Note if you change the "blur.shard.block.cache.v2.cacheBlockSize" or "blur.shard.block.cache.v2.fileBufferSize" you should adjust the buffer sizes as well as the total memory allocated. +blur.shard.buffercache.1024=8388608 # The number of milliseconds to wait for the cluster to settle once changes have ceased blur.shard.safemodedelay=5000
