Adding a global switch for index warmup. Also a preconnecting thread in the controller to remove the inital connection time for searches.
Project: http://git-wip-us.apache.org/repos/asf/incubator-blur/repo Commit: http://git-wip-us.apache.org/repos/asf/incubator-blur/commit/76c55cc0 Tree: http://git-wip-us.apache.org/repos/asf/incubator-blur/tree/76c55cc0 Diff: http://git-wip-us.apache.org/repos/asf/incubator-blur/diff/76c55cc0 Branch: refs/heads/apache-blur-0.2 Commit: 76c55cc0194c78b8056891308ed348aa9d24666c Parents: bd28c8a Author: Aaron McCurry <[email protected]> Authored: Thu Jan 16 13:14:46 2014 -0500 Committer: Aaron McCurry <[email protected]> Committed: Thu Jan 16 13:14:46 2014 -0500 ---------------------------------------------------------------------- .../indexserver/DistributedIndexServer.java | 9 ++++- .../blur/thrift/BlurControllerServer.java | 40 ++++++++++++++++++++ .../blur/thrift/ThriftBlurShardServer.java | 4 +- .../org/apache/blur/utils/BlurConstants.java | 1 + .../src/main/resources/blur-default.properties | 3 ++ 5 files changed, 54 insertions(+), 3 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/76c55cc0/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 4ab6188..2e04109 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 @@ -120,13 +120,15 @@ public class DistributedIndexServer extends AbstractDistributedIndexServer { private final ConcurrentMap<String, Map<String, BlurIndex>> _indexes = new ConcurrentHashMap<String, Map<String, BlurIndex>>(); private final ShardStateManager _shardStateManager = new ShardStateManager(); private final Closer _closer; + private final boolean _warmupDisabled; public DistributedIndexServer(Configuration configuration, ZooKeeper zookeeper, ClusterStatus clusterStatus, BlurIndexWarmup warmup, BlurFilterCache filterCache, BlockCacheDirectoryFactory blockCacheDirectoryFactory, DistributedLayoutFactory distributedLayoutFactory, String cluster, String nodeName, long safeModeDelay, - int shardOpenerThreadCount, int internalSearchThreads, int warmupThreads, int maxMergeThreads) - throws KeeperException, InterruptedException { + int shardOpenerThreadCount, int internalSearchThreads, int warmupThreads, int maxMergeThreads, + boolean warmupDisabled) throws KeeperException, InterruptedException { super(clusterStatus, configuration, nodeName, cluster); + _warmupDisabled = warmupDisabled; _closer = Closer.create(); _shardOpenerThreadCount = shardOpenerThreadCount; _zookeeper = zookeeper; @@ -504,6 +506,9 @@ public class DistributedIndexServer extends AbstractDistributedIndexServer { } private void warmUp(final BlurIndex index, final TableDescriptor table, final String shard) throws IOException { + if (_warmupDisabled) { + return; + } _warmupExecutor.submit(new Runnable() { @Override public void run() { http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/76c55cc0/blur-core/src/main/java/org/apache/blur/thrift/BlurControllerServer.java ---------------------------------------------------------------------- diff --git a/blur-core/src/main/java/org/apache/blur/thrift/BlurControllerServer.java b/blur-core/src/main/java/org/apache/blur/thrift/BlurControllerServer.java index 696b947..f529b0a 100644 --- a/blur-core/src/main/java/org/apache/blur/thrift/BlurControllerServer.java +++ b/blur-core/src/main/java/org/apache/blur/thrift/BlurControllerServer.java @@ -28,8 +28,10 @@ import java.util.HashSet; import java.util.IdentityHashMap; import java.util.List; import java.util.Map; +import java.util.TimerTask; import java.util.Map.Entry; import java.util.Set; +import java.util.Timer; import java.util.TreeSet; import java.util.UUID; import java.util.concurrent.Callable; @@ -182,6 +184,7 @@ public class BlurControllerServer extends TableAdmin implements Iface { private ConcurrentMap<String, WatchNodeExistance> _watchForOnlineShardsPerClusterExistance = new ConcurrentHashMap<String, WatchNodeExistance>(); private ConcurrentMap<String, WatchChildren> _watchForTablesPerCluster = new ConcurrentHashMap<String, WatchChildren>(); private ConcurrentMap<String, WatchChildren> _watchForOnlineShardsPerCluster = new ConcurrentHashMap<String, WatchChildren>(); + private Timer _preconnectTimer; public void init() throws KeeperException, InterruptedException { setupZookeeper(); @@ -194,6 +197,43 @@ public class BlurControllerServer extends TableAdmin implements Iface { watchForLayoutChanges(cluster); } updateLayout(); + startPreconnectTimer(); + } + + private void startPreconnectTimer() { + _preconnectTimer = new Timer("controller preconnect clients", true); + _preconnectTimer.schedule(new TimerTask() { + @Override + public void run() { + try { + preconnectClients(); + } catch (Exception e) { + LOG.error("Unknown error while trying to preconnect to shard servers.", e); + } + } + }, TimeUnit.SECONDS.toMillis(5), TimeUnit.SECONDS.toMillis(5)); + } + + private void preconnectClients() { + if (_clusterStatus != null) { + List<String> clusterList = _clusterStatus.getClusterList(true); + for (String cluster : clusterList) { + List<String> onlineShardServers = _clusterStatus.getOnlineShardServers(true, cluster); + for (String shardServer : onlineShardServers) { + preconnectClients(shardServer); + } + } + } + } + + private void preconnectClients(String shardServer) { + try { + Iface client = org.apache.blur.thrift.BlurClient.getClient(shardServer); + client.ping(); + LOG.debug("Pinging shard server [{0}]", shardServer); + } catch (Exception e) { + LOG.error("Error while trying to ping shard server [{0}]", shardServer); + } } private void setupZookeeper() throws KeeperException, InterruptedException { http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/76c55cc0/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 d6f31ef..7ee5f07 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 @@ -42,6 +42,7 @@ import static org.apache.blur.utils.BlurConstants.BLUR_SHARD_SERVER_THRIFT_THREA 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; +import static org.apache.blur.utils.BlurConstants.BLUR_SHARD_WARMUP_DISABLED; import static org.apache.blur.utils.BlurConstants.BLUR_SHARD_WARMUP_THREAD_COUNT; import static org.apache.blur.utils.BlurConstants.BLUR_THRIFT_MAX_FRAME_SIZE; import static org.apache.blur.utils.BlurConstants.BLUR_ZOOKEEPER_CONNECTION; @@ -206,9 +207,10 @@ public class ThriftBlurShardServer extends ThriftServer { int internalSearchThreads = configuration.getInt(BLUR_SHARD_WARMUP_THREAD_COUNT, 16); int warmupThreads = configuration.getInt(BLUR_SHARD_WARMUP_THREAD_COUNT, 16); int maxMergeThreads = configuration.getInt(BLUR_SHARD_MERGE_THREAD_COUNT, 3); + boolean warmupDisabled = configuration.getBoolean(BLUR_SHARD_WARMUP_DISABLED, false); final DistributedIndexServer indexServer = new DistributedIndexServer(config, zooKeeper, clusterStatus, indexWarmup, filterCache, blockCacheDirectoryFactory, distributedLayoutFactory, cluster, nodeName, - safeModeDelay, shardOpenerThreadCount, internalSearchThreads, warmupThreads, maxMergeThreads); + safeModeDelay, shardOpenerThreadCount, internalSearchThreads, warmupThreads, maxMergeThreads, warmupDisabled); BooleanQuery.setMaxClauseCount(configuration.getInt(BLUR_MAX_CLAUSE_COUNT, 1024)); http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/76c55cc0/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 e8307d9..cca9885 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 @@ -87,6 +87,7 @@ public class BlurConstants { public static final String BLUR_SHARD_THRIFT_MAX_READ_BUFFER_BYTES = "blur.shard.thrift.max.read.buffer.bytes"; public static final String BLUR_SHARD_THRIFT_ACCEPT_QUEUE_SIZE_PER_THREAD = "blur.shard.thrift.accept.queue.size.per.thread"; public static final String BLUR_SHARD_DISTRIBUTED_LAYOUT_FACTORY_CLASS = "blur.shard.distributed.layout.factory.class"; + public static final String BLUR_SHARD_WARMUP_DISABLED = "blur.shard.warmup.disabled"; public static final String BLUR_SHARD_BLOCK_CACHE_V2_READ_CACHE_EXT = "blur.shard.block.cache.v2.read.cache.ext"; http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/76c55cc0/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 c0f71c3..7e46952 100644 --- a/blur-util/src/main/resources/blur-default.properties +++ b/blur-util/src/main/resources/blur-default.properties @@ -75,6 +75,9 @@ blur.shard.cache.max.timetolive=60000 # Default implementation of the blur cache filter, which is a pass through filter that does nothing blur.shard.filter.cache.class=org.apache.blur.manager.DefaultBlurFilterCache +# Globally disable index warmup. +blur.shard.warmup.disabled=false + # Default Blur index warmup class that warms the fields provided in the table descriptor blur.shard.index.warmup.class=org.apache.blur.manager.indexserver.DefaultBlurIndexWarmup
