incubator-blur-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From amccu...@apache.org
Subject git commit: Fixed BLUR-294
Date Wed, 06 Nov 2013 02:13:26 GMT
Updated Branches:
  refs/heads/master bc07ac3e1 -> e06f3075c


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/e06f3075
Tree: http://git-wip-us.apache.org/repos/asf/incubator-blur/tree/e06f3075
Diff: http://git-wip-us.apache.org/repos/asf/incubator-blur/diff/e06f3075

Branch: refs/heads/master
Commit: e06f3075c1e3c4b8dc51abea97afcd91ed4e1c77
Parents: bc07ac3
Author: Aaron McCurry <amccurry@gmail.com>
Authored: Tue Nov 5 21:12:55 2013 -0500
Committer: Aaron McCurry <amccurry@gmail.com>
Committed: Tue Nov 5 21:13:19 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/e06f3075/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/e06f3075/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/e06f3075/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/e06f3075/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/e06f3075/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/e06f3075/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/e06f3075/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/e06f3075/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/e06f3075/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/e06f3075/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/e06f3075/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/e06f3075/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/e06f3075/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/e06f3075/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/e06f3075/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/e06f3075/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/e06f3075/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/e06f3075/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/e06f3075/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/e06f3075/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


Mime
View raw message