Fixed BLUR-262
Project: http://git-wip-us.apache.org/repos/asf/incubator-blur/repo Commit: http://git-wip-us.apache.org/repos/asf/incubator-blur/commit/3cedeb54 Tree: http://git-wip-us.apache.org/repos/asf/incubator-blur/tree/3cedeb54 Diff: http://git-wip-us.apache.org/repos/asf/incubator-blur/diff/3cedeb54 Branch: refs/heads/master Commit: 3cedeb54309c81c8ec1995b9fa3e911615b7fcb7 Parents: baf45fb Author: Aaron McCurry <[email protected]> Authored: Thu Oct 10 20:45:48 2013 -0400 Committer: Aaron McCurry <[email protected]> Committed: Thu Oct 10 20:46:43 2013 -0400 ---------------------------------------------------------------------- .../manager/writer/SharedMergeScheduler.java | 7 ++--- .../store/BlockCacheDirectoryFactoryV2.java | 28 +++++++++++++++----- .../blur/store/blockcache_v2/BaseCache.java | 14 +++++++++- .../apache/blur/store/blockcache_v2/Cache.java | 21 +++++++++++++++ .../store/blockcache_v2/CacheIndexInput.java | 12 ++++++++- .../apache/blur/store/blockcache_v2/Quiet.java | 21 +++++++++++++++ .../blur/store/CacheDirectoryTestSuite.java | 10 +++++-- .../store/blockcache_v2/CacheDirectoryTest.java | 9 +++++-- .../blockcache_v2/CacheIndexInputTest.java | 10 +++++++ .../org/apache/blur/utils/BlurConstants.java | 2 ++ 10 files changed, 118 insertions(+), 16 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/3cedeb54/blur-core/src/main/java/org/apache/blur/manager/writer/SharedMergeScheduler.java ---------------------------------------------------------------------- diff --git a/blur-core/src/main/java/org/apache/blur/manager/writer/SharedMergeScheduler.java b/blur-core/src/main/java/org/apache/blur/manager/writer/SharedMergeScheduler.java index 1d99907..e53f464 100644 --- a/blur-core/src/main/java/org/apache/blur/manager/writer/SharedMergeScheduler.java +++ b/blur-core/src/main/java/org/apache/blur/manager/writer/SharedMergeScheduler.java @@ -16,6 +16,8 @@ package org.apache.blur.manager.writer; * See the License for the specific language governing permissions and * limitations under the License. */ +import static org.apache.blur.utils.BlurConstants.SHARED_MERGE_SCHEDULER; + import java.io.Closeable; import java.io.IOException; import java.util.concurrent.BlockingQueue; @@ -33,7 +35,6 @@ import org.apache.lucene.index.MergeScheduler; public class SharedMergeScheduler implements Runnable, Closeable { private static final Log LOG = LogFactory.getLog(SharedMergeScheduler.class); - private static final long ONE_SECOND = 1000; private BlockingQueue<IndexWriter> _writers = new LinkedBlockingQueue<IndexWriter>(); @@ -42,7 +43,7 @@ public class SharedMergeScheduler implements Runnable, Closeable { public SharedMergeScheduler() { int threads = 3; - _service = Executors.newThreadPool("sharedMergeScheduler", threads, false); + _service = Executors.newThreadPool(SHARED_MERGE_SCHEDULER, threads, false); for (int i = 0; i < threads; i++) { _service.submit(this); } @@ -56,7 +57,7 @@ public class SharedMergeScheduler implements Runnable, Closeable { } } } - + private void removeWriter(IndexWriter writer) { synchronized (_writers) { _writers.remove(writer); http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/3cedeb54/blur-store/src/main/java/org/apache/blur/store/BlockCacheDirectoryFactoryV2.java ---------------------------------------------------------------------- diff --git a/blur-store/src/main/java/org/apache/blur/store/BlockCacheDirectoryFactoryV2.java b/blur-store/src/main/java/org/apache/blur/store/BlockCacheDirectoryFactoryV2.java index 9e44253..6d1a47e 100644 --- a/blur-store/src/main/java/org/apache/blur/store/BlockCacheDirectoryFactoryV2.java +++ b/blur-store/src/main/java/org/apache/blur/store/BlockCacheDirectoryFactoryV2.java @@ -16,6 +16,8 @@ */ package org.apache.blur.store; +import static org.apache.blur.utils.BlurConstants.SHARED_MERGE_SCHEDULER; + import java.io.IOException; import java.util.Set; @@ -25,6 +27,7 @@ import org.apache.blur.store.blockcache_v2.BaseCache.STORE; import org.apache.blur.store.blockcache_v2.Cache; import org.apache.blur.store.blockcache_v2.CacheDirectory; import org.apache.blur.store.blockcache_v2.FileNameFilter; +import org.apache.blur.store.blockcache_v2.Quiet; import org.apache.blur.store.blockcache_v2.Size; import org.apache.lucene.store.Directory; @@ -36,21 +39,21 @@ public class BlockCacheDirectoryFactoryV2 extends BlockCacheDirectoryFactory { final int fileBufferSizeInt = 8192; final int cacheBlockSizeInt = 8192; final STORE store = STORE.OFF_HEAP; - + Size fileBufferSize = new Size() { @Override public int getSize(String directoryName, String fileName) { return fileBufferSizeInt; } }; - + Size cacheBlockSize = new Size() { @Override public int getSize(String directoryName, String fileName) { return cacheBlockSizeInt; } }; - + FileNameFilter readFilter = new FileNameFilter() { @Override public boolean accept(String directoryName, String fileName) { @@ -60,7 +63,7 @@ public class BlockCacheDirectoryFactoryV2 extends BlockCacheDirectoryFactory { return true; } }; - + FileNameFilter writeFilter = new FileNameFilter() { @Override public boolean accept(String directoryName, String fileName) { @@ -70,9 +73,20 @@ public class BlockCacheDirectoryFactoryV2 extends BlockCacheDirectoryFactory { return true; } }; - - _cache = new BaseCache(totalNumberOfBytes, fileBufferSize, cacheBlockSize, readFilter, writeFilter, - store); + + Quiet quiet = new Quiet() { + @Override + public boolean shouldBeQuiet(String directoryName, String fileName) { + Thread thread = Thread.currentThread(); + String name = thread.getName(); + if (name.startsWith(SHARED_MERGE_SCHEDULER)) { + return true; + } + return false; + } + }; + + _cache = new BaseCache(totalNumberOfBytes, fileBufferSize, cacheBlockSize, readFilter, writeFilter, quiet, store); } @Override http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/3cedeb54/blur-store/src/main/java/org/apache/blur/store/blockcache_v2/BaseCache.java ---------------------------------------------------------------------- diff --git a/blur-store/src/main/java/org/apache/blur/store/blockcache_v2/BaseCache.java b/blur-store/src/main/java/org/apache/blur/store/blockcache_v2/BaseCache.java index 35f7d83..84f4b32 100644 --- a/blur-store/src/main/java/org/apache/blur/store/blockcache_v2/BaseCache.java +++ b/blur-store/src/main/java/org/apache/blur/store/blockcache_v2/BaseCache.java @@ -65,9 +65,10 @@ public class BaseCache extends Cache { private final Size _fileBufferSize; private final Map<FileIdKey, Long> _fileNameToId = new ConcurrentHashMap<FileIdKey, Long>(); private final AtomicLong _fileId = new AtomicLong(); + private final Quiet _quiet; public BaseCache(long totalNumberOfBytes, Size fileBufferSize, Size cacheBlockSize, FileNameFilter readFilter, - FileNameFilter writeFilter, STORE store) { + FileNameFilter writeFilter, Quiet quiet, STORE store) { _cacheMap = new ConcurrentLinkedHashMap.Builder<CacheKey, CacheValue>().weigher(new BaseCacheWeigher()) .maximumWeightedCapacity(totalNumberOfBytes).listener(new BaseCacheEvictionListener()).build(); _fileBufferSize = fileBufferSize; @@ -75,6 +76,7 @@ public class BaseCache extends Cache { _writeFilter = writeFilter; _store = store; _cacheBlockSize = cacheBlockSize; + _quiet = quiet; } private void addToReleaseQueue(CacheValue value) { @@ -86,6 +88,11 @@ public class BaseCache extends Cache { LOG.debug("CacheValue was not released [{0}]", value); } } + + @Override + public boolean shouldBeQuiet(CacheDirectory directory, String fileName) { + return _quiet.shouldBeQuiet(directory.getDirectoryName(), fileName); + } @Override public CacheValue newInstance(CacheDirectory directory, String fileName, int cacheBlockSize) { @@ -146,6 +153,11 @@ public class BaseCache extends Cache { public CacheValue get(CacheKey key) { return _cacheMap.get(key); } + + @Override + public CacheValue getQuietly(CacheKey key) { + return _cacheMap.getQuietly(key); + } @Override public void put(CacheKey key, CacheValue value) { http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/3cedeb54/blur-store/src/main/java/org/apache/blur/store/blockcache_v2/Cache.java ---------------------------------------------------------------------- diff --git a/blur-store/src/main/java/org/apache/blur/store/blockcache_v2/Cache.java b/blur-store/src/main/java/org/apache/blur/store/blockcache_v2/Cache.java index 552d3db..bdc6b75 100644 --- a/blur-store/src/main/java/org/apache/blur/store/blockcache_v2/Cache.java +++ b/blur-store/src/main/java/org/apache/blur/store/blockcache_v2/Cache.java @@ -123,6 +123,16 @@ public abstract class Cache { public abstract CacheValue get(CacheKey key); /** + * Gets the cache value for the given key. Null if missing. NOTE: This method + * will not effect the priority of the cache. + * + * @param key + * the key. + * @return the cache value or null. + */ + public abstract CacheValue getQuietly(CacheKey key); + + /** * Puts the cache entry into the cache. * * @param key @@ -151,4 +161,15 @@ public abstract class Cache { */ public abstract void releaseDirectory(String directoryName); + /** + * Determines if the reader should be quiet or not. + * + * @param directory + * the directory. + * @param fileName + * the file name. + * @return boolean. + */ + public abstract boolean shouldBeQuiet(CacheDirectory directory, String fileName); + } http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/3cedeb54/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 2f61f2d..e9bd601 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 @@ -38,6 +38,7 @@ public class CacheIndexInput extends IndexInput { private long _position; private int _blockPosition; + private boolean _quiet; public CacheIndexInput(CacheDirectory directory, String fileName, IndexInput indexInput, Cache cache) throws IOException { @@ -51,6 +52,7 @@ public class CacheIndexInput extends IndexInput { _fileId = _cache.getFileId(_directory, _fileName); _cacheBlockSize = _cache.getCacheBlockSize(_directory, _fileName); _bufferSize = _cache.getFileBufferSize(_directory, _fileName); + _quiet = _cache.shouldBeQuiet(_directory, _fileName); _key.setFileId(_fileId); } @@ -134,7 +136,7 @@ public class CacheIndexInput extends IndexInput { private void fill() throws IOException { _key.setBlockId(getBlockId()); - _cacheValue = _cache.get(_key); + _cacheValue = get(_key); if (_cacheValue == null) { _cacheValue = _cache.newInstance(_directory, _fileName); long filePosition = getFilePosition(); @@ -156,6 +158,13 @@ public class CacheIndexInput extends IndexInput { _blockPosition = getBlockPosition(); } + private CacheValue get(CacheKey key) { + if (_quiet) { + return _cache.getQuietly(key); + } + return _cache.get(key); + } + private int getBlockPosition() { return (int) (_position % _cacheBlockSize); } @@ -222,6 +231,7 @@ public class CacheIndexInput extends IndexInput { if (clone._cacheValue != null) { clone._cacheValue.incRef(); } + clone._quiet = _cache.shouldBeQuiet(_directory, _fileName); return clone; } http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/3cedeb54/blur-store/src/main/java/org/apache/blur/store/blockcache_v2/Quiet.java ---------------------------------------------------------------------- diff --git a/blur-store/src/main/java/org/apache/blur/store/blockcache_v2/Quiet.java b/blur-store/src/main/java/org/apache/blur/store/blockcache_v2/Quiet.java new file mode 100644 index 0000000..e7c9fb7 --- /dev/null +++ b/blur-store/src/main/java/org/apache/blur/store/blockcache_v2/Quiet.java @@ -0,0 +1,21 @@ +/** + * 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.blockcache_v2; + +public interface Quiet { + boolean shouldBeQuiet(String directoryName, String fileName); +} http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/3cedeb54/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 db59271..ca51d9b 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 @@ -24,6 +24,7 @@ import org.apache.blur.store.blockcache_v2.BaseCache.STORE; import org.apache.blur.store.blockcache_v2.Cache; import org.apache.blur.store.blockcache_v2.CacheDirectory; import org.apache.blur.store.blockcache_v2.FileNameFilter; +import org.apache.blur.store.blockcache_v2.Quiet; import org.apache.blur.store.blockcache_v2.Size; import org.apache.blur.store.buffer.BufferStore; import org.apache.lucene.store.Directory; @@ -63,8 +64,13 @@ public abstract class CacheDirectoryTestSuite extends BaseDirectoryTestSuite { return true; } }; - - Cache cache = new BaseCache(totalNumberOfBytes, fileBufferSize, cacheBlockSize, readFilter, writeFilter, + Quiet quiet = new Quiet() { + @Override + public boolean shouldBeQuiet(String directoryName, String fileName) { + return false; + } + }; + Cache cache = new BaseCache(totalNumberOfBytes, fileBufferSize, cacheBlockSize, readFilter, writeFilter,quiet, getStore()); Directory dir = FSDirectory.open(new File(file, "cache")); http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/3cedeb54/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 97d8adc..cad9476 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 @@ -79,8 +79,13 @@ public class CacheDirectoryTest { return true; } }; - - Cache cache = new BaseCache(totalNumberOfBytes, fileBufferSize, cacheBlockSize, readFilter, writeFilter, + Quiet quiet = new Quiet() { + @Override + public boolean shouldBeQuiet(String directoryName, String fileName) { + return false; + } + }; + Cache cache = new BaseCache(totalNumberOfBytes, fileBufferSize, cacheBlockSize, readFilter, writeFilter, quiet, STORE.ON_HEAP); Directory directory = newDirectory(); BufferStore.init(128, 128); http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/3cedeb54/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 4716d2c..4d488d4 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 @@ -246,6 +246,16 @@ public class CacheIndexInputTest { } + @Override + public CacheValue getQuietly(CacheKey key) { + return cache.getQuietly(key); + } + + @Override + public boolean shouldBeQuiet(CacheDirectory directory, String fileName) { + return false; + } + }; return cacheFactory; } http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/3cedeb54/blur-util/src/main/java/org/apache/blur/utils/BlurConstants.java ---------------------------------------------------------------------- diff --git a/blur-util/src/main/java/org/apache/blur/utils/BlurConstants.java b/blur-util/src/main/java/org/apache/blur/utils/BlurConstants.java index 993208b..fb5a0e6 100644 --- a/blur-util/src/main/java/org/apache/blur/utils/BlurConstants.java +++ b/blur-util/src/main/java/org/apache/blur/utils/BlurConstants.java @@ -110,6 +110,8 @@ public class BlurConstants { public static final long ZK_WAIT_TIME = TimeUnit.SECONDS.toMillis(5); public static final String DELETE_MARKER_VALUE = "delete"; public static final String DELETE_MARKER = "_deletemarker_"; + + public static final String SHARED_MERGE_SCHEDULER = "sharedMergeScheduler"; static { try {
