Updated Branches: refs/heads/master b5916ac38 -> 783f83455
Fixed BLUR-245 Project: http://git-wip-us.apache.org/repos/asf/incubator-blur/repo Commit: http://git-wip-us.apache.org/repos/asf/incubator-blur/commit/6b000703 Tree: http://git-wip-us.apache.org/repos/asf/incubator-blur/tree/6b000703 Diff: http://git-wip-us.apache.org/repos/asf/incubator-blur/diff/6b000703 Branch: refs/heads/master Commit: 6b000703457e64d5c9334426ed012c027a359eb3 Parents: b5916ac Author: Aaron McCurry <[email protected]> Authored: Sat Sep 28 09:56:06 2013 -0400 Committer: Aaron McCurry <[email protected]> Committed: Sat Sep 28 09:56:06 2013 -0400 ---------------------------------------------------------------------- .../org/apache/blur/manager/IndexManager.java | 64 ++++++++------------ .../blur/thrift/ThriftBlurShardServer.java | 2 + .../apache/blur/manager/IndexManagerTest.java | 1 + .../org/apache/blur/utils/BlurConstants.java | 1 + .../src/main/resources/blur-default.properties | 3 + 5 files changed, 32 insertions(+), 39 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/6b000703/blur-core/src/main/java/org/apache/blur/manager/IndexManager.java ---------------------------------------------------------------------- diff --git a/blur-core/src/main/java/org/apache/blur/manager/IndexManager.java b/blur-core/src/main/java/org/apache/blur/manager/IndexManager.java index a734d78..2aff9fd 100644 --- a/blur-core/src/main/java/org/apache/blur/manager/IndexManager.java +++ b/blur-core/src/main/java/org/apache/blur/manager/IndexManager.java @@ -143,6 +143,7 @@ public class IndexManager { private Timer _fetchTimer; private int _fetchCount = 100; private int _maxHeapPerRowFetch = 10000000; + private int _mutateThreadCount; public static AtomicBoolean DEBUG_RUN_SLOW = new AtomicBoolean(false); @@ -164,10 +165,14 @@ public class IndexManager { "Internal Queries/s", TimeUnit.SECONDS); _fetchTimer = Metrics.newTimer(new MetricName(ORG_APACHE_BLUR, BLUR, "Fetch Timer"), TimeUnit.MICROSECONDS, TimeUnit.SECONDS); - + if (_threadCount == 0) { + throw new RuntimeException("Thread Count cannot be 0."); + } + if (_mutateThreadCount == 0) { + throw new RuntimeException("Mutate Thread Count cannot be 0."); + } _executor = Executors.newThreadPool("index-manager", _threadCount); - // @TODO give the mutate it's own thread pool - _mutateExecutor = Executors.newThreadPool("index-manager-mutate", _threadCount); + _mutateExecutor = Executors.newThreadPool("index-manager-mutate", _mutateThreadCount); _statusManager.init(); LOG.info("Init Complete"); @@ -722,11 +727,11 @@ public class IndexManager { List<String> terms = new ArrayList<String>(size); AtomicReader areader = BlurUtil.getAtomicReader(reader); Terms termsAll = areader.terms(term.field()); - - if(termsAll == null) { - return terms; + + if (termsAll == null) { + return terms; } - + TermsEnum termEnum = termsAll.iterator(null); termEnum.seekCeil(term.bytes()); @@ -795,40 +800,17 @@ public class IndexManager { } public void mutate(final RowMutation mutation) throws BlurException, IOException { - Future<Void> future = _mutateExecutor.submit(new Callable<Void>() { - @Override - public Void call() throws Exception { - doMutate(mutation); - return null; - } - }); - try { - future.get(); - } catch (InterruptedException e) { - throw new BException("Unknown error during mutation", e); - } catch (ExecutionException e) { - throw new BException("Unknown error during mutation", e.getCause()); - } + long s = System.nanoTime(); + doMutate(mutation); + long e = System.nanoTime(); + LOG.debug("doMutate took [{0} ms] to complete", (e - s) / 1000000.0); } public void mutate(final List<RowMutation> mutations) throws BlurException, IOException { - Future<Void> future = _mutateExecutor.submit(new Callable<Void>() { - @Override - public Void call() throws Exception { - long s = System.nanoTime(); - doMutates(mutations); - long e = System.nanoTime(); - LOG.debug("doMutates took [" + (e - s) / 1000000.0 + " ms] to complete"); - return null; - } - }); - try { - future.get(); - } catch (InterruptedException e) { - throw new BException("Unknown error during mutation", e); - } catch (ExecutionException e) { - throw new BException("Unknown error during mutation", e.getCause()); - } + long s = System.nanoTime(); + doMutates(mutations); + long e = System.nanoTime(); + LOG.debug("doMutates took [{0} ms] to complete", (e - s) / 1000000.0); } private void doMutates(List<RowMutation> mutations) throws BlurException, IOException { @@ -1149,9 +1131,13 @@ public class IndexManager { } public void setThreadCount(int threadCount) { - this._threadCount = threadCount; + _threadCount = threadCount; } + public void setMutateThreadCount(int mutateThreadCount) { + _mutateThreadCount = mutateThreadCount; + } + public void setFilterCache(BlurFilterCache filterCache) { _filterCache = filterCache; } http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/6b000703/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 1807a70..4a89cff 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 @@ -21,6 +21,7 @@ import static org.apache.blur.utils.BlurConstants.BLUR_CLUSTER_NAME; import static org.apache.blur.utils.BlurConstants.BLUR_CONTROLLER_BIND_PORT; import static org.apache.blur.utils.BlurConstants.BLUR_GUI_CONTROLLER_PORT; import static org.apache.blur.utils.BlurConstants.BLUR_GUI_SHARD_PORT; +import static org.apache.blur.utils.BlurConstants.BLUR_INDEXMANAGER_MUTATE_THREAD_COUNT; import static org.apache.blur.utils.BlurConstants.BLUR_INDEXMANAGER_SEARCH_THREAD_COUNT; import static org.apache.blur.utils.BlurConstants.BLUR_MAX_CLAUSE_COUNT; import static org.apache.blur.utils.BlurConstants.BLUR_MAX_HEAP_PER_ROW_FETCH; @@ -233,6 +234,7 @@ public class ThriftBlurShardServer extends ThriftServer { indexManager.setIndexServer(indexServer); indexManager.setMaxClauseCount(configuration.getInt(BLUR_MAX_CLAUSE_COUNT, 1024)); indexManager.setThreadCount(configuration.getInt(BLUR_INDEXMANAGER_SEARCH_THREAD_COUNT, 32)); + indexManager.setMutateThreadCount(configuration.getInt(BLUR_INDEXMANAGER_MUTATE_THREAD_COUNT, 32)); indexManager.setFilterCache(filterCache); indexManager.setClusterStatus(clusterStatus); indexManager.setFetchCount(configuration.getInt(BLUR_SHARD_FETCHCOUNT, 100)); http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/6b000703/blur-core/src/test/java/org/apache/blur/manager/IndexManagerTest.java ---------------------------------------------------------------------- diff --git a/blur-core/src/test/java/org/apache/blur/manager/IndexManagerTest.java b/blur-core/src/test/java/org/apache/blur/manager/IndexManagerTest.java index d6ff887..b7639cd 100644 --- a/blur-core/src/test/java/org/apache/blur/manager/IndexManagerTest.java +++ b/blur-core/src/test/java/org/apache/blur/manager/IndexManagerTest.java @@ -107,6 +107,7 @@ public class IndexManagerTest { indexManager.setStatusCleanupTimerDelay(1000); indexManager.setIndexServer(server); indexManager.setThreadCount(1); + indexManager.setMutateThreadCount(1); indexManager.setClusterStatus(new ClusterStatus() { @Override http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/6b000703/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 fbe86d0..1439aa1 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 @@ -61,6 +61,7 @@ public class BlurConstants { public static final String BLUR_SHARD_FILTER_CACHE_CLASS = "blur.shard.filter.cache.class"; public static final String BLUR_SHARD_INDEX_WARMUP_CLASS = "blur.shard.index.warmup.class"; public static final String BLUR_INDEXMANAGER_SEARCH_THREAD_COUNT = "blur.indexmanager.search.thread.count"; + public static final String BLUR_INDEXMANAGER_MUTATE_THREAD_COUNT = "blur.indexmanager.mutate.thread.count"; public static final String BLUR_SHARD_DATA_FETCH_THREAD_COUNT = "blur.shard.data.fetch.thread.count"; public static final String BLUR_SHARD_INTERNAL_SEARCH_THREAD_COUNT = "blur.shard.internal.search.thread.count"; public static final String BLUR_SHARD_WARMUP_THREAD_COUNT = "blur.shard.warmup.thread.count"; http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/6b000703/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 4a1ef8b..b96c6d9 100644 --- a/blur-util/src/main/resources/blur-default.properties +++ b/blur-util/src/main/resources/blur-default.properties @@ -94,6 +94,9 @@ blur.max.clause.count=1024 # The number of thread used for parallel searching in the index manager blur.indexmanager.search.thread.count=8 +# The number of thread used for parallel mutating in the index manager +blur.indexmanager.mutate.thread.count=8 + # The number of threads used for parallel searching in the index searchers blur.shard.internal.search.thread.count=8
