Repository: incubator-blur Updated Branches: refs/heads/apache-blur-0.2 02379c667 -> 7aeef6818
Making some updates to allow tables to be created/enabled/disabled/removed faster. Project: http://git-wip-us.apache.org/repos/asf/incubator-blur/repo Commit: http://git-wip-us.apache.org/repos/asf/incubator-blur/commit/7aeef681 Tree: http://git-wip-us.apache.org/repos/asf/incubator-blur/tree/7aeef681 Diff: http://git-wip-us.apache.org/repos/asf/incubator-blur/diff/7aeef681 Branch: refs/heads/apache-blur-0.2 Commit: 7aeef6818758b5b6e1e07a72d4d60c1eaf42c0ba Parents: 02379c6 Author: Aaron McCurry <[email protected]> Authored: Mon Feb 17 22:25:09 2014 -0500 Committer: Aaron McCurry <[email protected]> Committed: Mon Feb 17 22:25:09 2014 -0500 ---------------------------------------------------------------------- .../manager/clusterstatus/ClusterStatus.java | 6 + .../clusterstatus/ZookeeperClusterStatus.java | 15 ++ .../indexserver/DistributedIndexServer.java | 252 ++++++++++++------- 3 files changed, 181 insertions(+), 92 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/7aeef681/blur-core/src/main/java/org/apache/blur/manager/clusterstatus/ClusterStatus.java ---------------------------------------------------------------------- diff --git a/blur-core/src/main/java/org/apache/blur/manager/clusterstatus/ClusterStatus.java b/blur-core/src/main/java/org/apache/blur/manager/clusterstatus/ClusterStatus.java index 4ea5801..0712222 100644 --- a/blur-core/src/main/java/org/apache/blur/manager/clusterstatus/ClusterStatus.java +++ b/blur-core/src/main/java/org/apache/blur/manager/clusterstatus/ClusterStatus.java @@ -23,6 +23,10 @@ import org.apache.blur.thrift.generated.TableDescriptor; public abstract class ClusterStatus { + + public static abstract class Action { + public abstract void action(); + } public abstract List<String> getOnlineShardServers(boolean useCache, String cluster); @@ -72,4 +76,6 @@ public abstract class ClusterStatus { public abstract boolean isOpen(); + public abstract void registerActionOnTableStateChange(Action action); + } http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/7aeef681/blur-core/src/main/java/org/apache/blur/manager/clusterstatus/ZookeeperClusterStatus.java ---------------------------------------------------------------------- diff --git a/blur-core/src/main/java/org/apache/blur/manager/clusterstatus/ZookeeperClusterStatus.java b/blur-core/src/main/java/org/apache/blur/manager/clusterstatus/ZookeeperClusterStatus.java index c55f174..6dc13da 100644 --- a/blur-core/src/main/java/org/apache/blur/manager/clusterstatus/ZookeeperClusterStatus.java +++ b/blur-core/src/main/java/org/apache/blur/manager/clusterstatus/ZookeeperClusterStatus.java @@ -20,6 +20,7 @@ import java.io.Closeable; import java.io.IOException; import java.util.ArrayList; import java.util.Collection; +import java.util.Collections; import java.util.HashSet; import java.util.List; import java.util.Map; @@ -70,6 +71,7 @@ public class ZookeeperClusterStatus extends ClusterStatus { private final ConcurrentMap<String, WatchChildren> _tableWatchers = new ConcurrentHashMap<String, WatchChildren>(); private final Map<String, SafeModeCacheEntry> _clusterToSafeMode = new ConcurrentHashMap<String, ZookeeperClusterStatus.SafeModeCacheEntry>(); private final ConcurrentMap<String, WatchNodeData> _enabledWatchNodeExistance = new ConcurrentHashMap<String, WatchNodeData>(); + private final Set<Action> _tableStateChange = Collections.newSetFromMap(new ConcurrentHashMap<Action, Boolean>()); public ZookeeperClusterStatus(ZooKeeper zooKeeper, BlurConfiguration configuration) { _zk = zooKeeper; @@ -137,6 +139,7 @@ public class ZookeeperClusterStatus extends ClusterStatus { @Override public void action(List<String> tables) { + runActions(); Set<String> newSet = new HashSet<String>(tables); Set<String> oldSet = _tablesPerCluster.put(cluster, newSet); Set<String> newTables = getNewTables(newSet, oldSet); @@ -155,6 +158,7 @@ public class ZookeeperClusterStatus extends ClusterStatus { enabledWatcher.watch(new WatchNodeData.OnChange() { @Override public void action(byte[] data) { + runActions(); _tableDescriptorCache.clear(); } }); @@ -188,6 +192,12 @@ public class ZookeeperClusterStatus extends ClusterStatus { return cluster + "." + table; } + private void runActions() { + for (Action action : _tableStateChange) { + action.action(); + } + } + @Override public List<String> getClusterList(boolean useCache) { if (useCache) { @@ -653,4 +663,9 @@ public class ZookeeperClusterStatus extends ClusterStatus { public boolean isOpen() { return _running.get(); } + + @Override + public void registerActionOnTableStateChange(Action action) { + _tableStateChange.add(action); + } } http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/7aeef681/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 9f4fd20..ebf70dd 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 @@ -26,8 +26,6 @@ import java.util.Map; import java.util.Map.Entry; import java.util.Set; import java.util.SortedSet; -import java.util.Timer; -import java.util.TimerTask; import java.util.TreeSet; import java.util.concurrent.Callable; import java.util.concurrent.ConcurrentHashMap; @@ -44,6 +42,7 @@ import org.apache.blur.log.LogFactory; import org.apache.blur.lucene.store.refcounter.DirectoryReferenceFileGC; import org.apache.blur.manager.BlurFilterCache; import org.apache.blur.manager.clusterstatus.ClusterStatus; +import org.apache.blur.manager.clusterstatus.ClusterStatus.Action; import org.apache.blur.manager.clusterstatus.ZookeeperPathConstants; import org.apache.blur.manager.writer.BlurIndex; import org.apache.blur.manager.writer.BlurIndexCloser; @@ -106,8 +105,10 @@ public class DistributedIndexServer extends AbstractDistributedIndexServer { // set internally private final AtomicBoolean _running = new AtomicBoolean(); - private final Timer _timerTableWarmer; - private final Timer _timerCacheFlush; + private final Thread _timerTableWarmer; + private final Object _warmupLock = new Object(); + private final Thread _timerCacheFlush; + private final Object _cleanupLock = new Object(); private final ExecutorService _openerService; private final DirectoryReferenceFileGC _gc; private final WatchChildren _watchOnlineShards; @@ -121,6 +122,7 @@ public class DistributedIndexServer extends AbstractDistributedIndexServer { private final ShardStateManager _shardStateManager = new ShardStateManager(); private final Closer _closer; private final boolean _warmupDisabled; + private long _shortDelay = 250; public DistributedIndexServer(Configuration configuration, ZooKeeper zookeeper, ClusterStatus clusterStatus, BlurIndexWarmup warmup, BlurFilterCache filterCache, BlockCacheDirectoryFactory blockCacheDirectoryFactory, @@ -128,6 +130,7 @@ public class DistributedIndexServer extends AbstractDistributedIndexServer { int shardOpenerThreadCount, int internalSearchThreads, int warmupThreads, int maxMergeThreads, boolean warmupDisabled) throws KeeperException, InterruptedException { super(clusterStatus, configuration, nodeName, cluster); + _running.set(true); _warmupDisabled = warmupDisabled; _closer = Closer.create(); _shardOpenerThreadCount = shardOpenerThreadCount; @@ -159,6 +162,7 @@ public class DistributedIndexServer extends AbstractDistributedIndexServer { _refresher = _closer.register(new BlurIndexRefresher()); _indexCloser = _closer.register(new BlurIndexCloser()); _timerCacheFlush = setupFlushCacheTimer(); + _timerCacheFlush.start(); registerMyselfAsMemberOfCluster(); String onlineShardsPath = ZookeeperPathConstants.getOnlineShardsPath(_cluster); @@ -171,9 +175,25 @@ public class DistributedIndexServer extends AbstractDistributedIndexServer { TimeUnit.SECONDS, registerNodeTimeOut); safeMode.registerNode(getNodeName(), BlurUtil.getVersion().getBytes()); - _running.set(true); _timerTableWarmer = setupTableWarmer(); + _timerTableWarmer.start(); _watchOnlineShards = watchForShardServerChanges(); + _clusterStatus.registerActionOnTableStateChange(new Action() { + @Override + public void action() { + synchronized (_warmupLock) { + _warmupLock.notifyAll(); + } + } + }); + _clusterStatus.registerActionOnTableStateChange(new Action() { + @Override + public void action() { + synchronized (_cleanupLock) { + _cleanupLock.notifyAll(); + } + } + }); } @Override @@ -182,11 +202,9 @@ public class DistributedIndexServer extends AbstractDistributedIndexServer { _running.set(false); _closer.close(); closeAllIndexes(); - _timerCacheFlush.purge(); - _timerCacheFlush.cancel(); + _timerCacheFlush.interrupt(); _watchOnlineShards.close(); - _timerTableWarmer.purge(); - _timerTableWarmer.cancel(); + _timerTableWarmer.interrupt(); } } @@ -294,55 +312,120 @@ public class DistributedIndexServer extends AbstractDistributedIndexServer { return _closer.register(watchOnlineShards); } - private Timer setupTableWarmer() { - Timer timerTableWarmer = new Timer("Table-Warmer", true); - timerTableWarmer.schedule(new TimerTask() { + private Thread setupTableWarmer() { + Thread thread = new Thread(new Runnable() { @Override public void run() { - try { - warmup(); - } catch (Throwable t) { - if (_running.get()) { - LOG.error("Unknown error", t); - } else { - LOG.debug("Unknown error", t); + while (_running.get()) { + synchronized (_warmupLock) { + try { + _warmupLock.wait(_delay); + } catch (InterruptedException e) { + return; + } + } + for (int i = 0; i < 10; i++) { + runWarmup(); + synchronized (_warmupLock) { + try { + _warmupLock.wait(_shortDelay); + } catch (InterruptedException e) { + return; + } + } } } } + }); + thread.setDaemon(true); + thread.setName("Table-Warmer"); + return thread; + } - private void warmup() { - if (_running.get()) { - List<String> tableList = _clusterStatus.getTableList(false, _cluster); - _tableCount.set(tableList.size()); - long indexCount = 0; - AtomicLong segmentCount = new AtomicLong(); - AtomicLong indexMemoryUsage = new AtomicLong(); - for (String table : tableList) { + private Thread setupFlushCacheTimer() { + Thread thread = new Thread(new Runnable() { + @Override + public void run() { + while (_running.get()) { + synchronized (_cleanupLock) { try { - Map<String, BlurIndex> indexes = getIndexes(table); - int count = indexes.size(); - indexCount += count; - updateMetrics(indexes, segmentCount, indexMemoryUsage); - LOG.debug("Table [{0}] has [{1}] number of shards online in this node.", table, count); - } catch (IOException e) { - LOG.error("Unknown error trying to warm table [{0}]", e, table); + _cleanupLock.wait(_delay); + } catch (InterruptedException e) { + return; + } + } + for (int i = 0; i < 10; i++) { + runCleanup(); + synchronized (_cleanupLock) { + try { + _cleanupLock.wait(_shortDelay); + } catch (InterruptedException e) { + return; + } } } - _indexCount.set(indexCount); - _segmentCount.set(segmentCount.get()); - _indexMemoryUsage.set(indexMemoryUsage.get()); } } + }); + thread.setDaemon(true); + thread.setName("Flush-IndexServer-Caches"); + return thread; + } + + private synchronized void runCleanup() { + try { + cleanup(); + } catch (Throwable t) { + if (_running.get()) { + LOG.error("Unknown error", t); + } else { + LOG.debug("Unknown error", t); + } + } + } - private void updateMetrics(Map<String, BlurIndex> indexes, AtomicLong segmentCount, AtomicLong indexMemoryUsage) - throws IOException { - for (BlurIndex index : indexes.values()) { - indexMemoryUsage.addAndGet(index.getIndexMemoryUsage()); - segmentCount.addAndGet(index.getSegmentCount()); + private synchronized void runWarmup() { + try { + warmupTables(); + } catch (Throwable t) { + if (_running.get()) { + LOG.error("Unknown error", t); + } else { + LOG.debug("Unknown error", t); + } + } + } + + private void warmupTables() { + if (_running.get()) { + List<String> tableList = _clusterStatus.getTableList(false, _cluster); + _tableCount.set(tableList.size()); + long indexCount = 0; + AtomicLong segmentCount = new AtomicLong(); + AtomicLong indexMemoryUsage = new AtomicLong(); + for (String table : tableList) { + try { + Map<String, BlurIndex> indexes = getIndexes(table); + int count = indexes.size(); + indexCount += count; + updateMetrics(indexes, segmentCount, indexMemoryUsage); + LOG.debug("Table [{0}] has [{1}] number of shards online in this node.", table, count); + } catch (IOException e) { + LOG.error("Unknown error trying to warm table [{0}]", e, table); } } - }, _delay, _delay); - return timerTableWarmer; + _indexCount.set(indexCount); + _segmentCount.set(segmentCount.get()); + _indexMemoryUsage.set(indexMemoryUsage.get()); + } + } + + private void updateMetrics(Map<String, BlurIndex> indexes, AtomicLong segmentCount, AtomicLong indexMemoryUsage) + throws IOException { + for (BlurIndex index : indexes.values()) { + indexMemoryUsage.addAndGet(index.getIndexMemoryUsage()); + segmentCount.addAndGet(index.getSegmentCount()); + } } private void registerMyselfAsMemberOfCluster() { @@ -359,59 +442,44 @@ public class DistributedIndexServer extends AbstractDistributedIndexServer { } } - private Timer setupFlushCacheTimer() { - Timer timerCacheFlush = new Timer("Flush-IndexServer-Caches", true); - timerCacheFlush.schedule(new TimerTask() { - @Override - public void run() { - try { - cleanup(); - } catch (Throwable t) { - LOG.error("Unknown error", t); + private void cleanup() { + clearMapOfOldTables(_layout); + boolean closed = false; + Map<String, Map<String, BlurIndex>> oldIndexesThatNeedToBeClosed = clearMapOfOldTables(_indexes); + for (String table : oldIndexesThatNeedToBeClosed.keySet()) { + Map<String, BlurIndex> indexes = oldIndexesThatNeedToBeClosed.get(table); + if (indexes == null) { + continue; + } + for (String shard : indexes.keySet()) { + BlurIndex index = indexes.get(shard); + if (index == null) { + continue; } + close(index, table, shard); + closed = true; } - - private void cleanup() { - clearMapOfOldTables(_layout); - boolean closed = false; - Map<String, Map<String, BlurIndex>> oldIndexesThatNeedToBeClosed = clearMapOfOldTables(_indexes); - for (String table : oldIndexesThatNeedToBeClosed.keySet()) { - Map<String, BlurIndex> indexes = oldIndexesThatNeedToBeClosed.get(table); - if (indexes == null) { - continue; - } - for (String shard : indexes.keySet()) { - BlurIndex index = indexes.get(shard); - if (index == null) { - continue; - } - close(index, table, shard); - closed = true; - } + } + for (String table : _indexes.keySet()) { + Map<String, BlurIndex> shardMap = _indexes.get(table); + if (shardMap != null) { + Set<String> shards = new HashSet<String>(shardMap.keySet()); + Set<String> shardsToServe = getShardsToServe(table); + shards.removeAll(shardsToServe); + if (!shards.isEmpty()) { + LOG.info("Need to close indexes for table [{0}] indexes [{1}]", table, shards); } - for (String table : _indexes.keySet()) { - Map<String, BlurIndex> shardMap = _indexes.get(table); - if (shardMap != null) { - Set<String> shards = new HashSet<String>(shardMap.keySet()); - Set<String> shardsToServe = getShardsToServe(table); - shards.removeAll(shardsToServe); - if (!shards.isEmpty()) { - LOG.info("Need to close indexes for table [{0}] indexes [{1}]", table, shards); - } - for (String shard : shards) { - LOG.info("Closing index for table [{0}] shard [{1}]", table, shard); - BlurIndex index = shardMap.remove(shard); - close(index, table, shard); - closed = true; - } - } - if (closed) { - TableContext.clear(table); - } + for (String shard : shards) { + LOG.info("Closing index for table [{0}] shard [{1}]", table, shard); + BlurIndex index = shardMap.remove(shard); + close(index, table, shard); + closed = true; } } - }, _delay, _delay); - return timerCacheFlush; + if (closed) { + TableContext.clear(table); + } + } } protected void close(BlurIndex index, String table, String shard) {
