Repository: incubator-blur Updated Branches: refs/heads/master 1bbb14b2d -> badd38f47
Updating the shared merge scheduler to have to different pools for small and large pools. Project: http://git-wip-us.apache.org/repos/asf/incubator-blur/repo Commit: http://git-wip-us.apache.org/repos/asf/incubator-blur/commit/fe2a63ca Tree: http://git-wip-us.apache.org/repos/asf/incubator-blur/tree/fe2a63ca Diff: http://git-wip-us.apache.org/repos/asf/incubator-blur/diff/fe2a63ca Branch: refs/heads/master Commit: fe2a63ca04511f474e5d7c255d842edae1e13a3c Parents: 1bbb14b Author: Aaron McCurry <[email protected]> Authored: Tue Nov 4 09:34:46 2014 -0500 Committer: Aaron McCurry <[email protected]> Committed: Tue Nov 4 09:34:46 2014 -0500 ---------------------------------------------------------------------- .../indexserver/DistributedIndexServer.java | 6 +- .../manager/indexserver/LocalIndexServer.java | 2 +- .../manager/writer/SharedMergeScheduler.java | 185 ++++++++++++------- .../blur/thrift/ThriftBlurShardServer.java | 5 +- .../store/BlockCacheDirectoryFactoryV2.java | 4 +- .../blur/store/hdfs_v2/JoinDirectory.java | 2 +- .../org/apache/blur/utils/BlurConstants.java | 3 +- .../src/main/resources/blur-default.properties | 9 +- 8 files changed, 141 insertions(+), 75 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/fe2a63ca/blur-core/src/main/java/org/apache/blur/manager/indexserver/DistributedIndexServer.java ---------------------------------------------------------------------- diff --git a/blur-core/src/main/java/org/apache/blur/manager/indexserver/DistributedIndexServer.java b/blur-core/src/main/java/org/apache/blur/manager/indexserver/DistributedIndexServer.java index 6b2fe8a..c087ed8 100644 --- a/blur-core/src/main/java/org/apache/blur/manager/indexserver/DistributedIndexServer.java +++ b/blur-core/src/main/java/org/apache/blur/manager/indexserver/DistributedIndexServer.java @@ -23,10 +23,10 @@ import java.util.HashMap; import java.util.HashSet; import java.util.List; import java.util.Map; -import java.util.Timer; import java.util.Map.Entry; import java.util.Set; import java.util.SortedSet; +import java.util.Timer; import java.util.TreeSet; import java.util.concurrent.Callable; import java.util.concurrent.ConcurrentHashMap; @@ -120,7 +120,7 @@ public class DistributedIndexServer extends AbstractDistributedIndexServer { BlurFilterCache filterCache, BlockCacheDirectoryFactory blockCacheDirectoryFactory, DistributedLayoutFactory distributedLayoutFactory, String cluster, String nodeName, long safeModeDelay, int shardOpenerThreadCount, int maxMergeThreads, int internalSearchThreads, - int minimumNumberOfNodesBeforeExitingSafeMode, Timer hdfsKeyValueTimer, Timer indexImporterTimer) + int minimumNumberOfNodesBeforeExitingSafeMode, Timer hdfsKeyValueTimer, Timer indexImporterTimer, long smallMergeThreshold) throws KeeperException, InterruptedException { super(clusterStatus, configuration, nodeName, cluster); _indexImporterTimer = indexImporterTimer; @@ -146,7 +146,7 @@ public class DistributedIndexServer extends AbstractDistributedIndexServer { _closer.register(CloseableExecutorService.close(_searchExecutor)); // @TODO allow for configuration of these - _mergeScheduler = _closer.register(new SharedMergeScheduler(maxMergeThreads)); + _mergeScheduler = _closer.register(new SharedMergeScheduler(maxMergeThreads, smallMergeThreshold)); _indexCloser = _closer.register(new BlurIndexCloser()); _timerCacheFlush = setupFlushCacheTimer(); http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/fe2a63ca/blur-core/src/main/java/org/apache/blur/manager/indexserver/LocalIndexServer.java ---------------------------------------------------------------------- diff --git a/blur-core/src/main/java/org/apache/blur/manager/indexserver/LocalIndexServer.java b/blur-core/src/main/java/org/apache/blur/manager/indexserver/LocalIndexServer.java index 3cad08b..7e1d57f 100644 --- a/blur-core/src/main/java/org/apache/blur/manager/indexserver/LocalIndexServer.java +++ b/blur-core/src/main/java/org/apache/blur/manager/indexserver/LocalIndexServer.java @@ -79,7 +79,7 @@ public class LocalIndexServer extends AbstractIndexServer { _timer = new Timer("Index Importer", true); _closer = Closer.create(); _tableContext = TableContext.create(tableDescriptor); - _mergeScheduler = _closer.register(new SharedMergeScheduler(3)); + _mergeScheduler = _closer.register(new SharedMergeScheduler(3, 128 * 1000 * 1000)); _searchExecutor = Executors.newCachedThreadPool(); _closer.register(new CloseableExecutorService(_searchExecutor)); _ramDir = ramDir; http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/fe2a63ca/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 2314a1f..8e33b0a 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 @@ -19,13 +19,14 @@ package org.apache.blur.manager.writer; import static org.apache.blur.metrics.MetricsConstants.LUCENE; import static org.apache.blur.metrics.MetricsConstants.MERGE_THROUGHPUT_BYTES; import static org.apache.blur.metrics.MetricsConstants.ORG_APACHE_BLUR; -import static org.apache.blur.utils.BlurConstants.SHARED_MERGE_SCHEDULER; +import static org.apache.blur.utils.BlurConstants.SHARED_MERGE_SCHEDULER_PREFIX; import java.io.Closeable; import java.io.IOException; -import java.util.concurrent.BlockingQueue; +import java.util.Iterator; +import java.util.UUID; import java.util.concurrent.ExecutorService; -import java.util.concurrent.LinkedBlockingQueue; +import java.util.concurrent.PriorityBlockingQueue; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; @@ -34,106 +35,164 @@ import org.apache.blur.log.Log; import org.apache.blur.log.LogFactory; import org.apache.lucene.index.IndexWriter; import org.apache.lucene.index.MergePolicy; +import org.apache.lucene.index.MergePolicy.OneMerge; import org.apache.lucene.index.MergeScheduler; import com.yammer.metrics.Metrics; import com.yammer.metrics.core.Meter; import com.yammer.metrics.core.MetricName; -public class SharedMergeScheduler implements Runnable, Closeable { +public class SharedMergeScheduler implements Closeable { private static final Log LOG = LogFactory.getLog(SharedMergeScheduler.class); - private static final long ONE_SECOND = TimeUnit.SECONDS.toMillis(1); + private static final Meter _throughputBytes; + + static { + MetricName mergeThoughputBytes = new MetricName(ORG_APACHE_BLUR, LUCENE, MERGE_THROUGHPUT_BYTES); + _throughputBytes = Metrics.newMeter(mergeThoughputBytes, MERGE_THROUGHPUT_BYTES, TimeUnit.SECONDS); + } - private final BlockingQueue<IndexWriter> _writers = new LinkedBlockingQueue<IndexWriter>(); private final AtomicBoolean _running = new AtomicBoolean(true); - private final ExecutorService _service; - private final Meter _throughputBytes; - public SharedMergeScheduler(int threads) { - _service = Executors.newThreadPool(SHARED_MERGE_SCHEDULER, threads, false); - for (int i = 0; i < threads; i++) { - _service.submit(this); + private final ExecutorService _smallMergeService; + private final ExecutorService _largeMergeService; + private final PriorityBlockingQueue<MergeWork> _smallMergeQueue = new PriorityBlockingQueue<MergeWork>(); + private final PriorityBlockingQueue<MergeWork> _largeMergeQueue = new PriorityBlockingQueue<MergeWork>(); + private final long _smallMergeThreshold; + + static class MergeWork implements Comparable<MergeWork> { + + private final String _id; + private final MergePolicy.OneMerge _merge; + private final IndexWriter _writer; + private final long _size; + + public MergeWork(String id, OneMerge merge, IndexWriter writer) throws IOException { + _id = id; + _merge = merge; + _writer = writer; + _size = merge.totalBytesSize(); + } + + @Override + public int compareTo(MergeWork o) { + return Long.compare(_size, o._size); + } + + public void merge() throws IOException { + long s = System.nanoTime(); + _writer.merge(_merge); + long e = System.nanoTime(); + double time = (e - s) / 1000000000.0; + double rate = (_size / 1000 / 1000) / time; + LOG.info("Merge took [{0} s] to complete at rate of [{1} MB/s], input bytes [{2}], segments merged {3}", time, + rate, _size, _merge.segments); + _throughputBytes.mark(_size); + } + + public String getId() { + return _id; } - MetricName mergeThoughputBytes = new MetricName(ORG_APACHE_BLUR, LUCENE, MERGE_THROUGHPUT_BYTES); - _throughputBytes = Metrics.newMeter(mergeThoughputBytes, MERGE_THROUGHPUT_BYTES, TimeUnit.SECONDS); } - private void mergeIndexWriter(IndexWriter writer) { - LOG.debug("Adding writer to merge [{0}]", writer); - _writers.add(writer); + public SharedMergeScheduler(int threads) { + this(threads, 128 * 1000 * 1000); } - private void removeWriter(IndexWriter writer) { - while (_writers.remove(writer)) { - // keep looping until all the references are gone + public SharedMergeScheduler(int threads, long smallMergeThreshold) { + _smallMergeThreshold = smallMergeThreshold; + _smallMergeService = Executors.newThreadPool(SHARED_MERGE_SCHEDULER_PREFIX + "-small", threads, false); + _largeMergeService = Executors.newThreadPool(SHARED_MERGE_SCHEDULER_PREFIX + "-large", threads, false); + for (int i = 0; i < threads; i++) { + _smallMergeService.submit(getMergerRunnable(_smallMergeQueue)); + _largeMergeService.submit(getMergerRunnable(_largeMergeQueue)); } } + private Runnable getMergerRunnable(final PriorityBlockingQueue<MergeWork> queue) { + return new Runnable() { + @Override + public void run() { + while (_running.get()) { + try { + MergeWork mergeWork = queue.take(); + try { + mergeWork.merge(); + } catch (IOException e) { + LOG.error("Unknown error while trying to perform merge on [{0}]", e, mergeWork); + } + } catch (InterruptedException e) { + if (_running.get()) { + LOG.error("Unknown error", e); + } + return; + } + } + } + }; + } + public MergeScheduler getMergeScheduler() { return new MergeScheduler() { - private IndexWriter _writer; + private final String _id = UUID.randomUUID().toString(); @Override public void merge(IndexWriter writer) throws IOException { - _writer = writer; - mergeIndexWriter(writer); + addMerges(_id, writer); } @Override public void close() throws IOException { - removeWriter(_writer); + remove(_id); } }; } - @Override - public void close() throws IOException { - _running.set(false); - _service.shutdownNow(); + protected void addMerges(String id, IndexWriter writer) throws IOException { + OneMerge merge; + while ((merge = writer.getNextMerge()) != null) { + addMerge(id, writer, merge); + } } - @Override - public void run() { - while (_running.get()) { - try { - IndexWriter writer = _writers.poll(); - if (writer == null) { - synchronized (this) { - wait(ONE_SECOND); - } - } else { - performMergeWriter(writer); - } - } catch (InterruptedException e) { - LOG.debug("Merging interrupted, exiting."); - return; - } catch (IOException e) { - LOG.error("Unknown IOException", e); - } + private void addMerge(String id, IndexWriter writer, OneMerge merge) throws IOException { + MergeWork mergeWork = new MergeWork(id, merge, writer); + if (isLargeMerge(merge)) { + _largeMergeQueue.add(mergeWork); + } else { + _smallMergeQueue.add(mergeWork); } } - private void performMergeWriter(IndexWriter writer) throws IOException { - while (true) { - MergePolicy.OneMerge merge = writer.getNextMerge(); - if (merge == null) { - LOG.debug("No merges to run for [{0}]", writer); - return; - } - long s = System.nanoTime(); - writer.merge(merge); - long e = System.nanoTime(); - double time = (e - s) / 1000000000.0; - double rate = (merge.totalBytesSize() / 1000 / 1000) / time; - if (time > 10) { - LOG.info("Merge took [{0} s] to complete at rate of [{1} MB/s]", time, rate); - } else { - LOG.debug("Merge took [{0} s] to complete at rate of [{1} MB/s]", time, rate); + private boolean isLargeMerge(OneMerge merge) throws IOException { + long totalBytesSize = merge.totalBytesSize(); + if (totalBytesSize <= _smallMergeThreshold) { + return false; + } + return true; + } + + protected void remove(String id) { + remove(_smallMergeQueue, id); + remove(_largeMergeQueue, id); + } + + private void remove(PriorityBlockingQueue<MergeWork> queue, String id) { + Iterator<MergeWork> iterator = queue.iterator(); + while (iterator.hasNext()) { + MergeWork mergeWork = iterator.next(); + if (id.equals(mergeWork.getId())) { + iterator.remove(); } - _throughputBytes.mark(merge.totalBytesSize()); } } + @Override + public void close() throws IOException { + _running.set(false); + _smallMergeService.shutdownNow(); + _largeMergeService.shutdownNow(); + } + } http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/fe2a63ca/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 b19f9cb..5201e43 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 @@ -44,6 +44,7 @@ import static org.apache.blur.utils.BlurConstants.BLUR_SHARD_OPENER_THREAD_COUNT import static org.apache.blur.utils.BlurConstants.BLUR_SHARD_SAFEMODEDELAY; import static org.apache.blur.utils.BlurConstants.BLUR_SHARD_SERVER_MINIMUM_BEFORE_SAFEMODE_EXIT; import static org.apache.blur.utils.BlurConstants.BLUR_SHARD_SERVER_THRIFT_THREAD_COUNT; +import static org.apache.blur.utils.BlurConstants.BLUR_SHARD_SMALL_MERGE_THRESHOLD; import static org.apache.blur.utils.BlurConstants.BLUR_SHARD_THRIFT_ACCEPT_QUEUE_SIZE_PER_THREAD; import static org.apache.blur.utils.BlurConstants.BLUR_SHARD_THRIFT_MAX_READ_BUFFER_BYTES; import static org.apache.blur.utils.BlurConstants.BLUR_SHARD_THRIFT_SELECTOR_THREADS; @@ -113,6 +114,7 @@ import sun.misc.VM; public class ThriftBlurShardServer extends ThriftServer { + private static final Log LOG = LogFactory.getLog(ThriftBlurShardServer.class); private static final boolean enableJsonReporter = false; private static final long _64MB = 64 * 1024 * 1024; @@ -209,10 +211,11 @@ public class ThriftBlurShardServer extends ThriftServer { int internalSearchThreads = configuration.getInt(BLUR_SHARD_INTERNAL_SEARCH_THREAD_COUNT, 16); final Timer hdfsKeyValueTimer = new Timer("HDFS KV Store", true); final Timer indexImporterTimer = new Timer("IndexImporter", true); + long smallMergeThreshold = configuration.getLong(BLUR_SHARD_SMALL_MERGE_THRESHOLD, 128 * 1000 * 1000); final DistributedIndexServer indexServer = new DistributedIndexServer(config, zooKeeper, clusterStatus, filterCache, blockCacheDirectoryFactory, distributedLayoutFactory, cluster, nodeName, safeModeDelay, shardOpenerThreadCount, maxMergeThreads, internalSearchThreads, minimumNumberOfNodesBeforeExitingSafeMode, - hdfsKeyValueTimer, indexImporterTimer); + hdfsKeyValueTimer, indexImporterTimer, smallMergeThreshold); BooleanQuery.setMaxClauseCount(configuration.getInt(BLUR_MAX_CLAUSE_COUNT, 1024)); http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/fe2a63ca/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 d1c279f..82e0a93 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 @@ -28,7 +28,7 @@ import static org.apache.blur.utils.BlurConstants.BLUR_SHARD_BLOCK_CACHE_V2_WRIT import static org.apache.blur.utils.BlurConstants.BLUR_SHARD_BLOCK_CACHE_V2_WRITE_NOCACHE_EXT; import static org.apache.blur.utils.BlurConstants.DEFAULT_VALUE; import static org.apache.blur.utils.BlurConstants.OFF_HEAP; -import static org.apache.blur.utils.BlurConstants.SHARED_MERGE_SCHEDULER; +import static org.apache.blur.utils.BlurConstants.SHARED_MERGE_SCHEDULER_PREFIX; import java.io.IOException; import java.util.Arrays; @@ -153,7 +153,7 @@ public class BlockCacheDirectoryFactoryV2 extends BlockCacheDirectoryFactory { public boolean shouldBeQuiet(CacheDirectory directory, String fileName) { Thread thread = Thread.currentThread(); String name = thread.getName(); - if (name.startsWith(SHARED_MERGE_SCHEDULER)) { + if (name.startsWith(SHARED_MERGE_SCHEDULER_PREFIX)) { return true; } return false; http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/fe2a63ca/blur-store/src/main/java/org/apache/blur/store/hdfs_v2/JoinDirectory.java ---------------------------------------------------------------------- diff --git a/blur-store/src/main/java/org/apache/blur/store/hdfs_v2/JoinDirectory.java b/blur-store/src/main/java/org/apache/blur/store/hdfs_v2/JoinDirectory.java index 9e872a3..6619e3f 100644 --- a/blur-store/src/main/java/org/apache/blur/store/hdfs_v2/JoinDirectory.java +++ b/blur-store/src/main/java/org/apache/blur/store/hdfs_v2/JoinDirectory.java @@ -93,7 +93,7 @@ public class JoinDirectory extends Directory implements LastModified, HdfsSymlin @Override public IndexOutput createOutput(String name, IOContext context) throws IOException { - if (Thread.currentThread().getName().startsWith(BlurConstants.SHARED_MERGE_SCHEDULER)) { + if (Thread.currentThread().getName().startsWith(BlurConstants.SHARED_MERGE_SCHEDULER_PREFIX)) { addLongTermSyncFile(name); return _longTermStorage.createOutput(name, context); } http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/fe2a63ca/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 0e0f5b0..cb807cb 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 @@ -114,6 +114,7 @@ public class BlurConstants { public static final String BLUR_SHARD_BLOCK_CACHE_V2_CACHE_BLOCK_SIZE = "blur.shard.block.cache.v2.cacheBlockSize"; public static final String BLUR_SHARD_BLURINDEX_CLASS = "blur.shard.blurindex.class"; public static final String BLUR_SHARD_SERVER_MINIMUM_BEFORE_SAFEMODE_EXIT = "blur.shard.server.minimum.before.safemode.exit"; + public static final String BLUR_SHARD_SMALL_MERGE_THRESHOLD = "blur.shard.small.merge.threshold"; public static final String BLUR_FIELDTYPE = "blur.fieldtype."; @@ -169,7 +170,7 @@ public class BlurConstants { public static final String DELETE_MARKER_VALUE = "delete"; public static final String DELETE_MARKER = "_deletemarker_"; - public static final String SHARED_MERGE_SCHEDULER = "sharedMergeScheduler"; + public static final String SHARED_MERGE_SCHEDULER_PREFIX = "shared-merge-scheduler"; public static final String BLUR_FILTER_ALIAS = "blur.filter.alias."; http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/fe2a63ca/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 999167e..f71c26c 100644 --- a/blur-util/src/main/resources/blur-default.properties +++ b/blur-util/src/main/resources/blur-default.properties @@ -157,15 +157,18 @@ blur.shard.safemodedelay=5000 # The number of shard servers in a cluster that have to be registered before the cluster will exit safemode. blur.shard.server.minimum.before.safemode.exit=0 -# The default time between index commits +# The default time between index commits. blur.shard.time.between.commits=30000 -# The default time between index refreshs +# The default time between index refreshs. blur.shard.time.between.refreshs=3000 -# The max number of threads used during index merges +# The max number of threads used during index merges. blur.shard.merge.thread.count=8 +# The threshold for a small merge in bytes. +blur.shard.small.merge.threshold=128000000 + # The maximum number of clauses in a BooleanQuery blur.max.clause.count=1024
