Removing index warm up logic from Blur, it has been disabled for some time now.
Project: http://git-wip-us.apache.org/repos/asf/incubator-blur/repo Commit: http://git-wip-us.apache.org/repos/asf/incubator-blur/commit/e9dce540 Tree: http://git-wip-us.apache.org/repos/asf/incubator-blur/tree/e9dce540 Diff: http://git-wip-us.apache.org/repos/asf/incubator-blur/diff/e9dce540 Branch: refs/heads/blur-384-random-port-cleanup Commit: e9dce5402d2d2ff78c04432fb1a14b073e4f110d Parents: 13cef3b Author: Aaron McCurry <[email protected]> Authored: Mon Oct 6 09:10:28 2014 -0400 Committer: Aaron McCurry <[email protected]> Committed: Mon Oct 6 09:10:28 2014 -0400 ---------------------------------------------------------------------- .../manager/indexserver/BlurIndexWarmup.java | 88 ---- .../indexserver/DefaultBlurIndexWarmup.java | 166 ------ .../indexserver/DistributedIndexServer.java | 48 +- .../manager/indexserver/LocalIndexServer.java | 4 +- .../blur/manager/indexserver/SafeMode.java | 2 +- .../apache/blur/manager/writer/BlurIndex.java | 3 +- .../blur/manager/writer/BlurIndexReadOnly.java | 2 +- .../manager/writer/BlurIndexReaderWarmer.java | 60 --- .../manager/writer/BlurIndexSimpleWriter.java | 12 +- .../org/apache/blur/server/TableContext.java | 7 +- .../blur/thrift/ThriftBlurShardServer.java | 14 +- .../blur/command/ShardCommandManagerTest.java | 2 +- .../indexserver/DefaultBlurIndexWarmupTest.java | 162 ------ .../writer/BlurIndexSimpleWriterTest.java | 5 +- .../blur/manager/writer/IndexImporterTest.java | 2 +- .../org/apache/blur/command/test1/howto.txt | 1 + .../apache/blur/lucene/warmup/IndexTracer.java | 141 ------ .../blur/lucene/warmup/IndexTracerResult.java | 218 -------- .../apache/blur/lucene/warmup/IndexWarmup.java | 502 ------------------- .../apache/blur/lucene/warmup/NotSupported.java | 28 -- .../blur/lucene/warmup/ThrottledIndexInput.java | 164 ------ .../blur/lucene/warmup/TraceableDirectory.java | 152 ------ .../blur/lucene/warmup/TraceableIndexInput.java | 93 ---- .../blur/lucene/warmup/IndexWarmupTest.java | 183 ------- .../blur/lucene/warmup/SlowAccessDirectory.java | 149 ------ .../org/apache/blur/utils/BlurConstants.java | 5 +- .../src/main/resources/blur-default.properties | 15 +- 27 files changed, 26 insertions(+), 2202 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/e9dce540/blur-core/src/main/java/org/apache/blur/manager/indexserver/BlurIndexWarmup.java ---------------------------------------------------------------------- diff --git a/blur-core/src/main/java/org/apache/blur/manager/indexserver/BlurIndexWarmup.java b/blur-core/src/main/java/org/apache/blur/manager/indexserver/BlurIndexWarmup.java deleted file mode 100644 index 2bd394b..0000000 --- a/blur-core/src/main/java/org/apache/blur/manager/indexserver/BlurIndexWarmup.java +++ /dev/null @@ -1,88 +0,0 @@ -package org.apache.blur.manager.indexserver; - -/** - * 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. - */ -import static org.apache.blur.utils.BlurConstants.BLUR_SHARD_INDEX_WARMUP_CLASS; -import static org.apache.blur.utils.BlurConstants.BLUR_SHARD_INDEX_WARMUP_THROTTLE; -import static org.apache.blur.utils.BlurConstants.BLUR_SHARD_WARMUP_THREAD_COUNT; - -import java.io.IOException; -import java.lang.reflect.Constructor; -import java.util.concurrent.atomic.AtomicBoolean; -import java.util.concurrent.atomic.AtomicLong; - -import org.apache.blur.BlurConfiguration; -import org.apache.blur.log.Log; -import org.apache.blur.log.LogFactory; -import org.apache.blur.manager.indexserver.DistributedIndexServer.ReleaseReader; -import org.apache.blur.thrift.generated.TableDescriptor; -import org.apache.lucene.index.IndexReader; - -public abstract class BlurIndexWarmup { - - private static final Log LOG = LogFactory.getLog(BlurIndexWarmup.class); - - protected long _warmupBandwidthThrottleBytesPerSec; - - public BlurIndexWarmup(long warmupBandwidthThrottleBytesPerSec) { - _warmupBandwidthThrottleBytesPerSec = warmupBandwidthThrottleBytesPerSec; - } - - public static BlurIndexWarmup getIndexWarmup(BlurConfiguration configuration) { - long totalThrottle = configuration.getLong(BLUR_SHARD_INDEX_WARMUP_THROTTLE, 30000000); - int totalThreadCount = configuration.getInt(BLUR_SHARD_WARMUP_THREAD_COUNT, 30000000); - long warmupBandwidthThrottleBytesPerSec = totalThrottle / totalThreadCount; - if (warmupBandwidthThrottleBytesPerSec <= 0) { - LOG.warn("Invalid values of either [{0} = {1}] or [{2} = {3}], needs to be greater then 0", - BLUR_SHARD_INDEX_WARMUP_THROTTLE, totalThrottle, BLUR_SHARD_WARMUP_THREAD_COUNT, totalThreadCount); - } - - String blurFilterCacheClass = configuration.get(BLUR_SHARD_INDEX_WARMUP_CLASS); - if (blurFilterCacheClass != null && blurFilterCacheClass.isEmpty()) { - if (!blurFilterCacheClass.equals("org.apache.blur.manager.indexserver.DefaultBlurIndexWarmup")) { - try { - Class<?> clazz = Class.forName(blurFilterCacheClass); - Constructor<?> constructor = clazz.getConstructor(new Class[]{Long.TYPE}); - return (BlurIndexWarmup) constructor.newInstance(warmupBandwidthThrottleBytesPerSec); - } catch (Exception e) { - throw new RuntimeException(e); - } - } - } - return new DefaultBlurIndexWarmup(warmupBandwidthThrottleBytesPerSec); - } - - /** - * Once the reader has be warmed up, release() must be called on the - * ReleaseReader even if an exception occurs. - * - * @param table - * the table descriptor. - * @param shard - * the shard name. - * @param reader - * thread reader itself. - * @param isClosed - * to check if the shard has been migrated to another node. - * @param releaseReader - * to release the handle on the reader. - * @throws IOException - */ - public abstract void warmBlurIndex(TableDescriptor table, String shard, IndexReader reader, AtomicBoolean isClosed, - ReleaseReader releaseReader, AtomicLong pause) throws IOException; - -} http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/e9dce540/blur-core/src/main/java/org/apache/blur/manager/indexserver/DefaultBlurIndexWarmup.java ---------------------------------------------------------------------- diff --git a/blur-core/src/main/java/org/apache/blur/manager/indexserver/DefaultBlurIndexWarmup.java b/blur-core/src/main/java/org/apache/blur/manager/indexserver/DefaultBlurIndexWarmup.java deleted file mode 100644 index b8a0dde..0000000 --- a/blur-core/src/main/java/org/apache/blur/manager/indexserver/DefaultBlurIndexWarmup.java +++ /dev/null @@ -1,166 +0,0 @@ -package org.apache.blur.manager.indexserver; - -/** - * 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. - */ -import java.io.IOException; -import java.lang.reflect.Field; -import java.util.Collections; -import java.util.HashMap; -import java.util.HashSet; -import java.util.List; -import java.util.Map; -import java.util.Set; -import java.util.TreeSet; -import java.util.concurrent.TimeUnit; -import java.util.concurrent.atomic.AtomicBoolean; -import java.util.concurrent.atomic.AtomicLong; - -import org.apache.blur.log.Log; -import org.apache.blur.log.LogFactory; -import org.apache.blur.lucene.warmup.IndexTracerResult; -import org.apache.blur.lucene.warmup.IndexWarmup; -import org.apache.blur.manager.indexserver.DistributedIndexServer.ReleaseReader; -import org.apache.blur.thrift.generated.TableDescriptor; -import org.apache.blur.utils.GCAction; -import org.apache.blur.utils.GCWatcher; -import org.apache.lucene.index.AtomicReader; -import org.apache.lucene.index.AtomicReaderContext; -import org.apache.lucene.index.FilterDirectoryReader; -import org.apache.lucene.index.IndexReader; -import org.apache.lucene.index.IndexReaderContext; -import org.apache.lucene.index.SegmentReader; -import org.apache.lucene.util.OpenBitSet; - -public class DefaultBlurIndexWarmup extends BlurIndexWarmup { - - private static final Log LOG = LogFactory.getLog(DefaultBlurIndexWarmup.class); - - private Set<AtomicBoolean> _stops = Collections.synchronizedSet(new HashSet<AtomicBoolean>()); - - public DefaultBlurIndexWarmup(long warmupBandwidthThrottleBytesPerSec) { - super(warmupBandwidthThrottleBytesPerSec); - GCWatcher.registerAction(new GCAction() { - @Override - public void takeAction() throws Exception { - synchronized (_stops) { - for (AtomicBoolean stop : _stops) { - stop.set(true); - } - } - } - }); - } - - @Override - public void warmBlurIndex(final TableDescriptor table, final String shard, IndexReader reader, - AtomicBoolean isClosed, ReleaseReader releaseReader, AtomicLong pauseWarmup) throws IOException { - LOG.info("Running warmup for reader [{0}]", reader); - long s = System.nanoTime(); - AtomicBoolean stop = new AtomicBoolean(); - try { - if (reader instanceof FilterDirectoryReader) { - reader = getBase((FilterDirectoryReader) reader); - } - int maxSampleSize = 1000; - synchronized (_stops) { - _stops.add(stop); - } - IndexWarmup indexWarmup = new IndexWarmup(isClosed, stop, maxSampleSize, _warmupBandwidthThrottleBytesPerSec); - String context = table.getName() + "/" + shard; - Map<String, List<IndexTracerResult>> sampleIndex = indexWarmup.sampleIndex(reader, context); - Iterable<String> preCacheCols = table.getPreCacheCols(); - if (preCacheCols == null) { - // All fields. - preCacheCols = getFields(reader); - } - boolean _oldWay = false; - if (_oldWay) { - warm(reader, preCacheCols, indexWarmup, sampleIndex, context, isClosed, pauseWarmup); - } else { - int blockSize = 8192;// @TODO - int bufferSize = 1024 * 1024;// @TODO - Map<String, OpenBitSet> filePartsToWarm = new HashMap<String, OpenBitSet>(); - for (String fieldName : preCacheCols) { - indexWarmup.getFilePositionsToWarm(reader, sampleIndex, fieldName, context, filePartsToWarm, blockSize); - } - indexWarmup.warmFile(reader, filePartsToWarm, context, blockSize, bufferSize); - } - - } finally { - synchronized (_stops) { - _stops.remove(stop); - } - releaseReader.release(); - long e = System.nanoTime(); - LOG.info("Warmup completed for table [{0}] shard [{1}] in [{2} ms]", table.name, shard, (e - s) / 1000000.0); - } - } - - private IndexReader getBase(FilterDirectoryReader reader) { - try { - Field field = FilterDirectoryReader.class.getDeclaredField("in"); - field.setAccessible(true); - return (IndexReader) field.get(reader); - } catch (Exception e) { - LOG.error("Unknown error trying to get base reader from [{0}]", e, reader); - return reader; - } - } - - private Iterable<String> getFields(IndexReader reader) throws IOException { - Set<String> fields = new TreeSet<String>(); - for (IndexReaderContext ctext : reader.getContext().leaves()) { - AtomicReaderContext atomicReaderContext = (AtomicReaderContext) ctext; - AtomicReader atomicReader = atomicReaderContext.reader(); - if (atomicReader instanceof SegmentReader) { - for (String f : atomicReader.fields()) { - fields.add(f); - } - } - } - return fields; - } - - private void warm(IndexReader reader, Iterable<String> preCacheCols, IndexWarmup indexWarmup, - Map<String, List<IndexTracerResult>> sampleIndex, String context, AtomicBoolean isClosed, AtomicLong pauseWarmup) { - for (String field : preCacheCols) { - maybePause(pauseWarmup); - try { - indexWarmup.warmFile(reader, sampleIndex, field, context); - } catch (IOException e) { - LOG.error("Context [{0}] unknown error trying to warmup the [{1}] field", e, context, field); - } - if (isClosed.get()) { - LOG.info("Context [{0}] index closed", context); - return; - } - } - } - - private void maybePause(AtomicLong pauseWarmup) { - while (pauseWarmup.get() > 0) { - synchronized (this) { - try { - this.wait(TimeUnit.SECONDS.toMillis(5)); - } catch (InterruptedException e) { - return; - } - } - } - } - -} http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/e9dce540/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 56d3d04..47b8aed 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 @@ -46,7 +46,6 @@ import org.apache.blur.manager.writer.BlurIndex; import org.apache.blur.manager.writer.BlurIndexCloser; import org.apache.blur.manager.writer.BlurIndexReadOnly; import org.apache.blur.manager.writer.SharedMergeScheduler; -import org.apache.blur.server.IndexSearcherClosable; import org.apache.blur.server.ShardContext; import org.apache.blur.server.TableContext; import org.apache.blur.store.BlockCacheDirectoryFactory; @@ -62,11 +61,8 @@ import org.apache.blur.zookeeper.WatchChildren.OnChange; import org.apache.blur.zookeeper.ZookeeperPathConstants; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.fs.Path; -import org.apache.lucene.index.IndexReader; import org.apache.lucene.store.Directory; -import org.apache.zookeeper.CreateMode; import org.apache.zookeeper.KeeperException; -import org.apache.zookeeper.ZooDefs.Ids; import org.apache.zookeeper.ZooKeeper; import org.apache.zookeeper.data.Stat; @@ -95,9 +91,7 @@ public class DistributedIndexServer extends AbstractDistributedIndexServer { private final BlockCacheDirectoryFactory _blockCacheDirectoryFactory; private final ZooKeeper _zookeeper; private final int _internalSearchThreads; - private final int _warmupThreads; private final long _safeModeDelay; - private final BlurIndexWarmup _warmup; private final BlurFilterCache _filterCache; private final DistributedLayoutFactory _distributedLayoutFactory; @@ -112,33 +106,27 @@ public class DistributedIndexServer extends AbstractDistributedIndexServer { private final SharedMergeScheduler _mergeScheduler; private final ExecutorService _searchExecutor; private final BlurIndexCloser _indexCloser; - private final ExecutorService _warmupExecutor; private final ConcurrentMap<String, LayoutEntry> _layout = new ConcurrentHashMap<String, LayoutEntry>(); 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; private long _shortDelay = 250; private final int _minimumNumberOfNodes; public DistributedIndexServer(Configuration configuration, ZooKeeper zookeeper, ClusterStatus clusterStatus, - BlurIndexWarmup warmup, BlurFilterCache filterCache, BlockCacheDirectoryFactory blockCacheDirectoryFactory, + BlurFilterCache filterCache, BlockCacheDirectoryFactory blockCacheDirectoryFactory, DistributedLayoutFactory distributedLayoutFactory, String cluster, String nodeName, long safeModeDelay, - int shardOpenerThreadCount, int internalSearchThreads, int warmupThreads, int maxMergeThreads, - boolean warmupDisabled, int minimumNumberOfNodesBeforeExitingSafeMode) throws KeeperException, - InterruptedException { + int shardOpenerThreadCount, int maxMergeThreads, int internalSearchThreads, + int minimumNumberOfNodesBeforeExitingSafeMode) throws KeeperException, InterruptedException { super(clusterStatus, configuration, nodeName, cluster); _minimumNumberOfNodes = minimumNumberOfNodesBeforeExitingSafeMode; _running.set(true); - _warmupDisabled = warmupDisabled; _closer = Closer.create(); _shardOpenerThreadCount = shardOpenerThreadCount; _zookeeper = zookeeper; _filterCache = filterCache; _safeModeDelay = safeModeDelay; - _warmup = warmup == null ? new DefaultBlurIndexWarmup(1000000) : warmup; _internalSearchThreads = internalSearchThreads; - _warmupThreads = warmupThreads; _blockCacheDirectoryFactory = blockCacheDirectoryFactory; _distributedLayoutFactory = distributedLayoutFactory; @@ -147,11 +135,9 @@ public class DistributedIndexServer extends AbstractDistributedIndexServer { BlurUtil.setupZookeeper(_zookeeper, _cluster); _openerService = Executors.newThreadPool("shard-opener", _shardOpenerThreadCount); _searchExecutor = Executors.newThreadPool("internal-search", _internalSearchThreads); - _warmupExecutor = Executors.newThreadPool("warmup", _warmupThreads); _closer.register(CloseableExecutorService.close(_openerService)); _closer.register(CloseableExecutorService.close(_searchExecutor)); - _closer.register(CloseableExecutorService.close(_warmupExecutor)); // @TODO allow for configuration of these _mergeScheduler = _closer.register(new SharedMergeScheduler(maxMergeThreads)); @@ -527,41 +513,15 @@ public class DistributedIndexServer extends AbstractDistributedIndexServer { } BlurIndex index = tableContext.newInstanceBlurIndex(shardContext, directory, _mergeScheduler, _searchExecutor, - _indexCloser, _warmup); + _indexCloser); if (_clusterStatus.isReadOnly(true, _cluster, table)) { index = new BlurIndexReadOnly(index); } _filterCache.opening(table, shard, index); - TableDescriptor tableDescriptor = _clusterStatus.getTableDescriptor(true, _cluster, table); - warmUp(index, tableDescriptor, shard); return index; } - 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() { - try { - final IndexSearcherClosable searcher = index.getIndexSearcher(); - IndexReader reader = searcher.getIndexReader(); - _warmup.warmBlurIndex(table, shard, reader, index.isClosed(), new ReleaseReader() { - @Override - public void release() throws IOException { - // this will allow for closing of index - searcher.close(); - } - }, _pauseWarmup); - } catch (Exception e) { - LOG.error("Unknown error while trying to warmup index [" + index + "]", e); - } - } - }); - } - private synchronized Map<String, BlurIndex> openMissingShards(final String table, Set<String> shardsToServe, final Map<String, BlurIndex> tableIndexes) { Map<String, Future<BlurIndex>> opening = new HashMap<String, Future<BlurIndex>>(); http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/e9dce540/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 8ffa392..1fd82b6 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 @@ -66,7 +66,6 @@ public class LocalIndexServer extends AbstractIndexServer { private final TableContext _tableContext; private final Closer _closer; private final boolean _ramDir; - private final BlurIndexWarmup _indexWarmup; private final BlurIndexCloser _indexCloser; public LocalIndexServer(TableDescriptor tableDescriptor) throws IOException { @@ -82,7 +81,6 @@ public class LocalIndexServer extends AbstractIndexServer { _ramDir = ramDir; _indexCloser = _closer.register(new BlurIndexCloser()); getIndexes(_tableContext.getTable()); - _indexWarmup = BlurIndexWarmup.getIndexWarmup(_tableContext.getBlurConfiguration()); } @Override @@ -159,7 +157,7 @@ public class LocalIndexServer extends AbstractIndexServer { private BlurIndex openIndex(String table, String shard, Directory dir) throws CorruptIndexException, IOException { ShardContext shardContext = ShardContext.create(_tableContext, shard); BlurIndexSimpleWriter index = new BlurIndexSimpleWriter(shardContext, dir, _mergeScheduler, _searchExecutor, - _indexCloser, _indexWarmup); + _indexCloser); return index; } http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/e9dce540/blur-core/src/main/java/org/apache/blur/manager/indexserver/SafeMode.java ---------------------------------------------------------------------- diff --git a/blur-core/src/main/java/org/apache/blur/manager/indexserver/SafeMode.java b/blur-core/src/main/java/org/apache/blur/manager/indexserver/SafeMode.java index 31af8f6..81465cb 100644 --- a/blur-core/src/main/java/org/apache/blur/manager/indexserver/SafeMode.java +++ b/blur-core/src/main/java/org/apache/blur/manager/indexserver/SafeMode.java @@ -93,7 +93,7 @@ public class SafeMode extends ZooKeeperLockManager { List<String> children = new ArrayList<String>(_zooKeeper.getChildren(_nodePath, _watcher)); Collections.sort(children); if (children.equals(prev) && children.size() >= _minimumNumberOfNodes) { - LOG.info("Clustered has settled."); + LOG.info("Cluster has settled."); return; } else { prev = children; http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/e9dce540/blur-core/src/main/java/org/apache/blur/manager/writer/BlurIndex.java ---------------------------------------------------------------------- diff --git a/blur-core/src/main/java/org/apache/blur/manager/writer/BlurIndex.java b/blur-core/src/main/java/org/apache/blur/manager/writer/BlurIndex.java index f8ff71a..99dbd6c 100644 --- a/blur-core/src/main/java/org/apache/blur/manager/writer/BlurIndex.java +++ b/blur-core/src/main/java/org/apache/blur/manager/writer/BlurIndex.java @@ -22,7 +22,6 @@ import java.util.concurrent.ExecutorService; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; -import org.apache.blur.manager.indexserver.BlurIndexWarmup; import org.apache.blur.server.IndexSearcherClosable; import org.apache.blur.server.ShardContext; import org.apache.blur.thrift.generated.RowMutation; @@ -42,7 +41,7 @@ public abstract class BlurIndex { protected ShardContext _shardContext; public BlurIndex(ShardContext shardContext, Directory directory, SharedMergeScheduler mergeScheduler, - ExecutorService searchExecutor, BlurIndexCloser indexCloser, BlurIndexWarmup indexWarmup) throws IOException { + ExecutorService searchExecutor, BlurIndexCloser indexCloser) throws IOException { _shardContext = shardContext; } http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/e9dce540/blur-core/src/main/java/org/apache/blur/manager/writer/BlurIndexReadOnly.java ---------------------------------------------------------------------- diff --git a/blur-core/src/main/java/org/apache/blur/manager/writer/BlurIndexReadOnly.java b/blur-core/src/main/java/org/apache/blur/manager/writer/BlurIndexReadOnly.java index c2aee75..12f194f 100644 --- a/blur-core/src/main/java/org/apache/blur/manager/writer/BlurIndexReadOnly.java +++ b/blur-core/src/main/java/org/apache/blur/manager/writer/BlurIndexReadOnly.java @@ -28,7 +28,7 @@ public class BlurIndexReadOnly extends BlurIndex { private final BlurIndex _blurIndex; public BlurIndexReadOnly(BlurIndex blurIndex) throws IOException { - super(null, null, null, null, null, null); + super(null, null, null, null, null); _blurIndex = blurIndex; } http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/e9dce540/blur-core/src/main/java/org/apache/blur/manager/writer/BlurIndexReaderWarmer.java ---------------------------------------------------------------------- diff --git a/blur-core/src/main/java/org/apache/blur/manager/writer/BlurIndexReaderWarmer.java b/blur-core/src/main/java/org/apache/blur/manager/writer/BlurIndexReaderWarmer.java deleted file mode 100644 index f2a0881..0000000 --- a/blur-core/src/main/java/org/apache/blur/manager/writer/BlurIndexReaderWarmer.java +++ /dev/null @@ -1,60 +0,0 @@ -package org.apache.blur.manager.writer; - -/** - * 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. - */ -import java.io.IOException; -import java.util.concurrent.atomic.AtomicBoolean; -import java.util.concurrent.atomic.AtomicLong; - -import org.apache.blur.log.Log; -import org.apache.blur.log.LogFactory; -import org.apache.blur.manager.indexserver.BlurIndexWarmup; -import org.apache.blur.manager.indexserver.DistributedIndexServer.ReleaseReader; -import org.apache.blur.server.ShardContext; -import org.apache.blur.thrift.generated.TableDescriptor; -import org.apache.lucene.index.AtomicReader; -import org.apache.lucene.index.IndexWriter.IndexReaderWarmer; - -public class BlurIndexReaderWarmer extends IndexReaderWarmer { - - private static final Log LOG = LogFactory.getLog(BlurIndexReaderWarmer.class); - - private final AtomicBoolean _isClosed; - private final AtomicLong _pause = new AtomicLong(); - private final BlurIndexWarmup _indexWarmup; - private final TableDescriptor _table; - private final String _shard; - - public BlurIndexReaderWarmer(ShardContext shardContext, AtomicBoolean isClosed, BlurIndexWarmup indexWarmup) { - _isClosed = isClosed; - _table = shardContext.getTableContext().getDescriptor(); - _shard = shardContext.getShard(); - _indexWarmup = indexWarmup; - } - - @Override - public void warm(AtomicReader reader) throws IOException { - LOG.debug("Warming reader [{0}]", reader); - ReleaseReader releaseReader = new ReleaseReader() { - @Override - public void release() throws IOException { - - } - }; - _indexWarmup.warmBlurIndex(_table, _shard, reader, _isClosed, releaseReader, _pause); - } -} http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/e9dce540/blur-core/src/main/java/org/apache/blur/manager/writer/BlurIndexSimpleWriter.java ---------------------------------------------------------------------- diff --git a/blur-core/src/main/java/org/apache/blur/manager/writer/BlurIndexSimpleWriter.java b/blur-core/src/main/java/org/apache/blur/manager/writer/BlurIndexSimpleWriter.java index fd9f22c..bbc57d4 100644 --- a/blur-core/src/main/java/org/apache/blur/manager/writer/BlurIndexSimpleWriter.java +++ b/blur-core/src/main/java/org/apache/blur/manager/writer/BlurIndexSimpleWriter.java @@ -41,8 +41,6 @@ import org.apache.blur.index.IndexDeletionPolicyReader; import org.apache.blur.log.Log; import org.apache.blur.log.LogFactory; import org.apache.blur.lucene.codec.Blur024Codec; -import org.apache.blur.lucene.warmup.TraceableDirectory; -import org.apache.blur.manager.indexserver.BlurIndexWarmup; import org.apache.blur.server.IndexSearcherClosable; import org.apache.blur.server.ShardContext; import org.apache.blur.server.TableContext; @@ -90,9 +88,8 @@ public class BlurIndexSimpleWriter extends BlurIndex { private final MutationQueueProcessor _mutationQueueProcessor; public BlurIndexSimpleWriter(ShardContext shardContext, Directory directory, SharedMergeScheduler mergeScheduler, - final ExecutorService searchExecutor, BlurIndexCloser indexCloser, BlurIndexWarmup indexWarmup) - throws IOException { - super(shardContext, directory, mergeScheduler, searchExecutor, indexCloser, indexWarmup); + final ExecutorService searchExecutor, BlurIndexCloser indexCloser) throws IOException { + super(shardContext, directory, mergeScheduler, searchExecutor, indexCloser); _searchThreadPool = searchExecutor; _shardContext = shardContext; _tableContext = _shardContext.getTableContext(); @@ -103,7 +100,6 @@ public class BlurIndexSimpleWriter extends BlurIndex { _conf.setWriteLockTimeout(TimeUnit.MINUTES.toMillis(5)); _conf.setCodec(new Blur024Codec(_tableContext.getBlurConfiguration())); _conf.setSimilarity(_tableContext.getSimilarity()); - _conf.setMergedSegmentWarmer(new BlurIndexReaderWarmer(shardContext, _isClosed, indexWarmup)); TieredMergePolicy mergePolicy = (TieredMergePolicy) _conf.getMergePolicy(); mergePolicy.setUseCompoundFile(false); _conf.setMergeScheduler(mergeScheduler.getMergeScheduler()); @@ -119,9 +115,7 @@ public class BlurIndexSimpleWriter extends BlurIndex { new BlurIndexWriter(directory, _conf).close(); } - // This directory allows for warm up by adding tracing ability. - TraceableDirectory dir = new TraceableDirectory(directory); - _directory = dir; + _directory = directory; _indexCloser = indexCloser; _indexReader.set(wrap(DirectoryReader.open(_directory))); http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/e9dce540/blur-core/src/main/java/org/apache/blur/server/TableContext.java ---------------------------------------------------------------------- diff --git a/blur-core/src/main/java/org/apache/blur/server/TableContext.java b/blur-core/src/main/java/org/apache/blur/server/TableContext.java index 788d34f..372b5b9 100644 --- a/blur-core/src/main/java/org/apache/blur/server/TableContext.java +++ b/blur-core/src/main/java/org/apache/blur/server/TableContext.java @@ -45,7 +45,6 @@ import org.apache.blur.log.Log; import org.apache.blur.log.LogFactory; import org.apache.blur.lucene.search.FairSimilarity; import org.apache.blur.manager.ReadInterceptor; -import org.apache.blur.manager.indexserver.BlurIndexWarmup; import org.apache.blur.manager.writer.BlurIndex; import org.apache.blur.manager.writer.BlurIndexCloser; import org.apache.blur.manager.writer.BlurIndexSimpleWriter; @@ -325,7 +324,7 @@ public class TableContext { @SuppressWarnings("unchecked") public BlurIndex newInstanceBlurIndex(ShardContext shardContext, Directory dir, SharedMergeScheduler mergeScheduler, - ExecutorService searchExecutor, BlurIndexCloser indexCloser, BlurIndexWarmup indexWarmup) throws IOException { + ExecutorService searchExecutor, BlurIndexCloser indexCloser) throws IOException { String className = _blurConfiguration.get(BLUR_SHARD_BLURINDEX_CLASS, BlurIndexSimpleWriter.class.getName()); @@ -337,7 +336,7 @@ public class TableContext { } Constructor<? extends BlurIndex> constructor = findConstructor(clazz); try { - return constructor.newInstance(shardContext, dir, mergeScheduler, searchExecutor, indexCloser, indexWarmup); + return constructor.newInstance(shardContext, dir, mergeScheduler, searchExecutor, indexCloser); } catch (InstantiationException e) { throw new IOException(e); } catch (IllegalAccessException e) { @@ -352,7 +351,7 @@ public class TableContext { private Constructor<? extends BlurIndex> findConstructor(Class<? extends BlurIndex> clazz) throws IOException { try { return clazz.getConstructor(new Class[] { ShardContext.class, Directory.class, SharedMergeScheduler.class, - ExecutorService.class, BlurIndexCloser.class, BlurIndexWarmup.class }); + ExecutorService.class, BlurIndexCloser.class }); } catch (NoSuchMethodException e) { throw new IOException(e); } catch (SecurityException e) { http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/e9dce540/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 bf556d7..faf3cf0 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 @@ -38,6 +38,7 @@ import static org.apache.blur.utils.BlurConstants.BLUR_SHARD_DEEP_PAGING_CACHE_S import static org.apache.blur.utils.BlurConstants.BLUR_SHARD_FETCHCOUNT; import static org.apache.blur.utils.BlurConstants.BLUR_SHARD_FILTER_CACHE_CLASS; import static org.apache.blur.utils.BlurConstants.BLUR_SHARD_HOSTNAME; +import static org.apache.blur.utils.BlurConstants.BLUR_SHARD_INTERNAL_SEARCH_THREAD_COUNT; import static org.apache.blur.utils.BlurConstants.BLUR_SHARD_MERGE_THREAD_COUNT; import static org.apache.blur.utils.BlurConstants.BLUR_SHARD_OPENER_THREAD_COUNT; import static org.apache.blur.utils.BlurConstants.BLUR_SHARD_SAFEMODEDELAY; @@ -46,8 +47,6 @@ 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_DEFAULT_MAX_FRAME_SIZE; import static org.apache.blur.utils.BlurConstants.BLUR_THRIFT_MAX_FRAME_SIZE; import static org.apache.blur.utils.BlurConstants.BLUR_TMP_PATH; @@ -75,7 +74,6 @@ import org.apache.blur.manager.BlurQueryChecker; import org.apache.blur.manager.DefaultBlurFilterCache; import org.apache.blur.manager.IndexManager; import org.apache.blur.manager.clusterstatus.ZookeeperClusterStatus; -import org.apache.blur.manager.indexserver.BlurIndexWarmup; import org.apache.blur.manager.indexserver.BlurServerShutDown; import org.apache.blur.manager.indexserver.BlurServerShutDown.BlurShutdown; import org.apache.blur.manager.indexserver.DistributedIndexServer; @@ -195,23 +193,19 @@ public class ThriftBlurShardServer extends ThriftServer { final BlurIndexRefresher refresher = new BlurIndexRefresher(); BlurFilterCache filterCache = getFilterCache(configuration); - BlurIndexWarmup indexWarmup = BlurIndexWarmup.getIndexWarmup(configuration); DistributedLayoutFactory distributedLayoutFactory = DistributedLayoutFactoryImpl.getDistributedLayoutFactory( configuration, cluster, zooKeeper); long safeModeDelay = configuration.getLong(BLUR_SHARD_SAFEMODEDELAY, 60000); int shardOpenerThreadCount = configuration.getInt(BLUR_SHARD_OPENER_THREAD_COUNT, 16); - 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); int minimumNumberOfNodesBeforeExitingSafeMode = configuration.getInt( BLUR_SHARD_SERVER_MINIMUM_BEFORE_SAFEMODE_EXIT, 0); + int internalSearchThreads = configuration.getInt(BLUR_SHARD_INTERNAL_SEARCH_THREAD_COUNT, 16); final DistributedIndexServer indexServer = new DistributedIndexServer(config, zooKeeper, clusterStatus, - indexWarmup, filterCache, blockCacheDirectoryFactory, distributedLayoutFactory, cluster, nodeName, - safeModeDelay, shardOpenerThreadCount, internalSearchThreads, warmupThreads, maxMergeThreads, warmupDisabled, - minimumNumberOfNodesBeforeExitingSafeMode); + filterCache, blockCacheDirectoryFactory, distributedLayoutFactory, cluster, nodeName, safeModeDelay, + shardOpenerThreadCount, maxMergeThreads,internalSearchThreads, minimumNumberOfNodesBeforeExitingSafeMode); BooleanQuery.setMaxClauseCount(configuration.getInt(BLUR_MAX_CLAUSE_COUNT, 1024)); http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/e9dce540/blur-core/src/test/java/org/apache/blur/command/ShardCommandManagerTest.java ---------------------------------------------------------------------- diff --git a/blur-core/src/test/java/org/apache/blur/command/ShardCommandManagerTest.java b/blur-core/src/test/java/org/apache/blur/command/ShardCommandManagerTest.java index 29e3e8b..5b67c6d 100644 --- a/blur-core/src/test/java/org/apache/blur/command/ShardCommandManagerTest.java +++ b/blur-core/src/test/java/org/apache/blur/command/ShardCommandManagerTest.java @@ -322,7 +322,7 @@ public class ShardCommandManagerTest { protected BlurIndex getNullBlurIndex(String shard) throws IOException { ShardContext shardContext = ShardContext.create(getTableContextFactory().getTableContext("test"), shard); - return new BlurIndex(shardContext, null, null, null, null, null) { + return new BlurIndex(shardContext, null, null, null, null) { @Override public void removeSnapshot(String name) throws IOException { http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/e9dce540/blur-core/src/test/java/org/apache/blur/manager/indexserver/DefaultBlurIndexWarmupTest.java ---------------------------------------------------------------------- diff --git a/blur-core/src/test/java/org/apache/blur/manager/indexserver/DefaultBlurIndexWarmupTest.java b/blur-core/src/test/java/org/apache/blur/manager/indexserver/DefaultBlurIndexWarmupTest.java deleted file mode 100644 index 4f69e60..0000000 --- a/blur-core/src/test/java/org/apache/blur/manager/indexserver/DefaultBlurIndexWarmupTest.java +++ /dev/null @@ -1,162 +0,0 @@ -/** - * 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.manager.indexserver; - -import java.io.File; -import java.io.IOException; -import java.util.Random; -import java.util.concurrent.atomic.AtomicBoolean; -import java.util.concurrent.atomic.AtomicLong; - -import org.apache.blur.lucene.warmup.TraceableDirectory; -import org.apache.blur.manager.indexserver.DistributedIndexServer.ReleaseReader; -import org.apache.blur.thrift.generated.TableDescriptor; -import org.apache.lucene.analysis.standard.StandardAnalyzer; -import org.apache.lucene.document.Document; -import org.apache.lucene.document.Field.Store; -import org.apache.lucene.document.TextField; -import org.apache.lucene.index.AtomicReader; -import org.apache.lucene.index.DirectoryReader; -import org.apache.lucene.index.IndexReader; -import org.apache.lucene.index.IndexWriter; -import org.apache.lucene.index.IndexWriterConfig; -import org.apache.lucene.index.IndexableField; -import org.apache.lucene.index.TieredMergePolicy; -import org.apache.lucene.store.Directory; -import org.apache.lucene.store.FSDirectory; -import org.apache.lucene.util.Version; -import org.junit.Test; - -public class DefaultBlurIndexWarmupTest { - - private Random _random = new Random(); - private int _numberOfFields = 10; - - @Test - public void testDefaultBlurIndexWarmupTestFullIndexReader() throws IOException { - File file = new File("./target/tmp/DefaultBlurIndexWarmupTest-test"); - Directory dir = FSDirectory.open(file); - - Directory directory = new TraceableDirectory(dir); - IndexReader indexReader = getIndexReader(directory); - - DefaultBlurIndexWarmup indexWarmup = new DefaultBlurIndexWarmup(10000000); - AtomicBoolean isClosed = new AtomicBoolean(); - - AtomicLong pauseWarmup = new AtomicLong(); - ReleaseReader releaseReader = new ReleaseReader() { - @Override - public void release() throws IOException { - - } - }; - String shard = "shard"; - TableDescriptor table = new TableDescriptor(); - table.setName("test"); - - long t1 = System.nanoTime(); - indexWarmup.warmBlurIndex(table, shard, indexReader, isClosed, releaseReader, pauseWarmup); - long t2 = System.nanoTime(); - System.out.println((t2 - t1) / 1000000.0); - } - - @Test - public void testDefaultBlurIndexWarmupTestOneSegment() throws IOException { - File file = new File("./target/tmp/DefaultBlurIndexWarmupTest-test"); - Directory dir = FSDirectory.open(file); - - Directory directory = new TraceableDirectory(dir); - IndexReader indexReader = getIndexReader(directory); - - DefaultBlurIndexWarmup indexWarmup = new DefaultBlurIndexWarmup(10000000); - AtomicBoolean isClosed = new AtomicBoolean(); - - AtomicLong pauseWarmup = new AtomicLong(); - ReleaseReader releaseReader = new ReleaseReader() { - @Override - public void release() throws IOException { - - } - }; - String shard = "shard"; - TableDescriptor table = new TableDescriptor(); - table.setName("test"); - - AtomicReader reader = indexReader.leaves().iterator().next().reader(); - - long t1 = System.nanoTime(); - indexWarmup.warmBlurIndex(table, shard, reader, isClosed, releaseReader, pauseWarmup); - long t2 = System.nanoTime(); - System.out.println((t2 - t1) / 1000000.0); - } - - private IndexReader getIndexReader(Directory directory) throws IOException { - if (!DirectoryReader.indexExists(directory)) { - long t1 = System.nanoTime(); - populate(directory); - long t2 = System.nanoTime(); - System.out.println((t2 - t1) / 1000000.0); - } - return DirectoryReader.open(directory); - } - - private void populate(Directory directory) throws IOException { - IndexWriterConfig conf = new IndexWriterConfig(Version.LUCENE_43, new StandardAnalyzer(Version.LUCENE_43)); - TieredMergePolicy mergePolicy = (TieredMergePolicy) conf.getMergePolicy(); - mergePolicy.setUseCompoundFile(false); - IndexWriter writer = new IndexWriter(directory, conf); - addDocs(writer); - writer.close(); - } - - private void addDocs(IndexWriter writer) throws IOException { - for (int i = 0; i < 2000; i++) { - writer.addDocument(getDoc()); - } - } - - private Iterable<? extends IndexableField> getDoc() { - Document document = new Document(); - document.add(new TextField(getFieldName(), getText(), Store.YES)); - return document; - } - - private String getText() { - StringBuilder builder = new StringBuilder(); - for (int i = 0; i < 1000; i++) { - builder.append(getWord()).append(' '); - } - return builder.toString(); - } - - private String getWord() { - StringBuilder builder = new StringBuilder(); - for (int i = 0; i < 10; i++) { - builder.append(getChar()); - } - return builder.toString(); - } - - private char getChar() { - return (char) (_random.nextInt(26) + 'a'); - } - - private String getFieldName() { - return "test" + _random.nextInt(_numberOfFields); - } - -} http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/e9dce540/blur-core/src/test/java/org/apache/blur/manager/writer/BlurIndexSimpleWriterTest.java ---------------------------------------------------------------------- diff --git a/blur-core/src/test/java/org/apache/blur/manager/writer/BlurIndexSimpleWriterTest.java b/blur-core/src/test/java/org/apache/blur/manager/writer/BlurIndexSimpleWriterTest.java index 4808911..5d54b75 100644 --- a/blur-core/src/test/java/org/apache/blur/manager/writer/BlurIndexSimpleWriterTest.java +++ b/blur-core/src/test/java/org/apache/blur/manager/writer/BlurIndexSimpleWriterTest.java @@ -32,7 +32,6 @@ import java.util.concurrent.ExecutorService; import org.apache.blur.BlurConfiguration; import org.apache.blur.concurrent.Executors; -import org.apache.blur.manager.indexserver.DefaultBlurIndexWarmup; import org.apache.blur.server.IndexSearcherClosable; import org.apache.blur.server.ShardContext; import org.apache.blur.server.TableContext; @@ -77,7 +76,6 @@ public class BlurIndexSimpleWriterTest { private SharedMergeScheduler _mergeScheduler; private String uuid; private BlurIndexCloser _closer; - private DefaultBlurIndexWarmup _indexWarmup; @Before public void setup() throws IOException { @@ -91,7 +89,6 @@ public class BlurIndexSimpleWriterTest { _configuration = new Configuration(); _service = Executors.newThreadPool("test", 10); _closer = new BlurIndexCloser(); - _indexWarmup = new DefaultBlurIndexWarmup(1000000); } private void setupWriter(Configuration configuration) throws IOException { @@ -117,7 +114,7 @@ public class BlurIndexSimpleWriterTest { path.mkdirs(); FSDirectory directory = FSDirectory.open(path); ShardContext shardContext = ShardContext.create(tableContext, "test-shard-" + uuid); - _writer = new BlurIndexSimpleWriter(shardContext, directory, _mergeScheduler, _service, _closer, _indexWarmup); + _writer = new BlurIndexSimpleWriter(shardContext, directory, _mergeScheduler, _service, _closer); } @After http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/e9dce540/blur-core/src/test/java/org/apache/blur/manager/writer/IndexImporterTest.java ---------------------------------------------------------------------- diff --git a/blur-core/src/test/java/org/apache/blur/manager/writer/IndexImporterTest.java b/blur-core/src/test/java/org/apache/blur/manager/writer/IndexImporterTest.java index 4800865..756cc08 100644 --- a/blur-core/src/test/java/org/apache/blur/manager/writer/IndexImporterTest.java +++ b/blur-core/src/test/java/org/apache/blur/manager/writer/IndexImporterTest.java @@ -120,7 +120,7 @@ public class IndexImporterTest { } private BlurIndex getBlurIndex(ShardContext shardContext, final Directory mainDirectory) throws IOException { - return new BlurIndex(shardContext, mainDirectory, null, null, null, null) { + return new BlurIndex(shardContext, mainDirectory, null, null, null) { @Override public void removeSnapshot(String name) throws IOException { http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/e9dce540/blur-core/src/test/resources/org/apache/blur/command/test1/howto.txt ---------------------------------------------------------------------- diff --git a/blur-core/src/test/resources/org/apache/blur/command/test1/howto.txt b/blur-core/src/test/resources/org/apache/blur/command/test1/howto.txt index 35aff76..36414ca 100644 --- a/blur-core/src/test/resources/org/apache/blur/command/test1/howto.txt +++ b/blur-core/src/test/resources/org/apache/blur/command/test1/howto.txt @@ -17,3 +17,4 @@ # Running this directory. javac -cp ../../../../../../../../target/blur-core-*.jar TestCommand.java jar -cf test1.jar META-INF/ TestCommand.class +rm TestCommand.class http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/e9dce540/blur-store/src/main/java/org/apache/blur/lucene/warmup/IndexTracer.java ---------------------------------------------------------------------- diff --git a/blur-store/src/main/java/org/apache/blur/lucene/warmup/IndexTracer.java b/blur-store/src/main/java/org/apache/blur/lucene/warmup/IndexTracer.java deleted file mode 100644 index 84ce8bb..0000000 --- a/blur-store/src/main/java/org/apache/blur/lucene/warmup/IndexTracer.java +++ /dev/null @@ -1,141 +0,0 @@ -package org.apache.blur.lucene.warmup; - -/** - * 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. - */ -import java.io.IOException; - -import org.apache.blur.lucene.warmup.IndexTracerResult.FILE_TYPE; -import org.apache.lucene.index.DocsAndPositionsEnum; -import org.apache.lucene.index.DocsEnum; -import org.apache.lucene.index.SegmentReader; -import org.apache.lucene.index.Terms; -import org.apache.lucene.index.TermsEnum; -import org.apache.lucene.util.Bits; - -public class IndexTracer { - - private boolean _hasOffsets; - private boolean _hasPayloads; - private boolean _hasPositions; - private SegmentReader _segmentReader; - private Bits _liveDocs; - private IndexTracerResult _result; - private TraceableDirectory _traceableDirectory; - private final int _maxSampleSize; - - public IndexTracer(TraceableDirectory dir, int maxSampleSize) { - _traceableDirectory = dir; - _maxSampleSize = maxSampleSize; - } - - public void initTrace(SegmentReader segmentReader, String field, boolean hasPositions, boolean hasPayloads, - boolean hasOffsets) { - _segmentReader = segmentReader; - _hasPositions = hasPositions; - _hasPayloads = hasPayloads; - _hasOffsets = hasOffsets; - _liveDocs = _segmentReader.getLiveDocs(); - _result = new IndexTracerResult(segmentReader.getSegmentName(), field); - } - - public void trace(String name, long filePointer) { - int index = name.lastIndexOf('.'); - String ext = name.substring(index); - if (ext.endsWith(".tim")) { - if (!_result.isFilePositionCaptured(FILE_TYPE.TIM)) { - _result.setPosition(filePointer, name, FILE_TYPE.TIM); - } - } else if (ext.endsWith(".doc")) { - if (!_result.isFilePositionCaptured(FILE_TYPE.DOC)) { - _result.setPosition(filePointer, name, FILE_TYPE.DOC); - } - } else if (ext.endsWith(".pos")) { - if (!_result.isFilePositionCaptured(FILE_TYPE.POS)) { - _result.setPosition(filePointer, name, FILE_TYPE.POS); - } - } else { - throw new RuntimeException("Not Implemented"); - } - } - - public IndexTracerResult runTrace(Terms terms) throws IOException { - IndexWarmup.enableRunTrace(); - _traceableDirectory.setIndexTracer(this); - _traceableDirectory.setTrace(true); - try { - TermsEnum termsEnum = terms.iterator(null); - int sampleCount = 0; - while (termsEnum.next() != null) { - if (_hasPositions || _hasOffsets || _hasPayloads) { - DocsAndPositionsEnum docsAndPositions = termsEnum.docsAndPositions(_liveDocs, null); - int nextDoc; - do { - nextDoc = docsAndPositions.nextDoc(); - int freq = docsAndPositions.freq(); - for (int i = 0; i < freq; i++) { - docsAndPositions.nextPosition(); - if (_hasPayloads) { - docsAndPositions.getPayload(); - } - if (traceComplete()) { - return getResult(); - } - } - } while (nextDoc != DocsEnum.NO_MORE_DOCS); - } else { - DocsEnum docsEnum = termsEnum.docs(_liveDocs, null); - int nextDoc; - do { - nextDoc = docsEnum.nextDoc(); - if (traceComplete()) { - return getResult(); - } - } while (nextDoc != DocsEnum.NO_MORE_DOCS); - } - sampleCount++; - if (sampleCount >= _maxSampleSize) { - break; - } - } - return getResult(); - } finally { - _traceableDirectory.setTrace(false); - IndexWarmup.disableRunTrace(); - } - } - - private boolean traceComplete() { - if (!_result.isFilePositionCaptured(FILE_TYPE.TIM)) { - return false; - } - if (!_result.isFilePositionCaptured(FILE_TYPE.DOC)) { - return false; - } - if (_hasPositions && !_result.isFilePositionCaptured(FILE_TYPE.POS)) { - return false; - } - if (_hasPayloads && !_result.isFilePositionCaptured(FILE_TYPE.PAY)) { - return false; - } - return true; - } - - IndexTracerResult getResult() { - return _result; - } - -} http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/e9dce540/blur-store/src/main/java/org/apache/blur/lucene/warmup/IndexTracerResult.java ---------------------------------------------------------------------- diff --git a/blur-store/src/main/java/org/apache/blur/lucene/warmup/IndexTracerResult.java b/blur-store/src/main/java/org/apache/blur/lucene/warmup/IndexTracerResult.java deleted file mode 100644 index 0853f5b..0000000 --- a/blur-store/src/main/java/org/apache/blur/lucene/warmup/IndexTracerResult.java +++ /dev/null @@ -1,218 +0,0 @@ -package org.apache.blur.lucene.warmup; - -/** - * 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. - */ -import java.io.IOException; - -import org.apache.lucene.store.IndexInput; -import org.apache.lucene.store.IndexOutput; - -public class IndexTracerResult { - - public enum FILE_TYPE { - TIM, DOC, POS, PAY - } - - private boolean _timCaptured; - private long _timPosition; - private String _timFileName; - - private boolean _docCaptured; - private long _docPosition; - private String _docFileName; - - private boolean _posCaptured; - private long _posPosition; - private String _posFileName; - - private boolean _payCaptured; - private long _payPosition; - private String _payFileName; - - private String _segmentName; - private String _field; - - IndexTracerResult() { - - } - - public static IndexTracerResult read(IndexInput input) throws IOException { - - IndexTracerResult result = new IndexTracerResult(); - - result._field = input.readString(); - result._segmentName = input.readString(); - - result._timCaptured = readBoolean(input); - if (result._timCaptured) { - result._timPosition = input.readVLong(); - result._timFileName = input.readString(); - } - - result._docCaptured = readBoolean(input); - if (result._docCaptured) { - result._docPosition = input.readVLong(); - result._docFileName = input.readString(); - } - - result._posCaptured = readBoolean(input); - if (result._posCaptured) { - result._posPosition = input.readVLong(); - result._posFileName = input.readString(); - } - - result._payCaptured = readBoolean(input); - if (result._payCaptured) { - result._payPosition = input.readVLong(); - result._payFileName = input.readString(); - } - return result; - } - - public void write(IndexOutput output) throws IOException { - output.writeString(_field); - output.writeString(_segmentName); - - writeBoolean(output, _timCaptured); - if (_timCaptured) { - output.writeVLong(_timPosition); - output.writeString(_timFileName); - } - - writeBoolean(output, _docCaptured); - if (_docCaptured) { - output.writeVLong(_docPosition); - output.writeString(_docFileName); - } - - writeBoolean(output, _posCaptured); - if (_posCaptured) { - output.writeVLong(_posPosition); - output.writeString(_posFileName); - } - - writeBoolean(output, _payCaptured); - if (_payCaptured) { - output.writeVLong(_payPosition); - output.writeString(_payFileName); - } - } - - private static boolean readBoolean(IndexInput input) throws IOException { - return input.readVInt() == 1; - } - - private static void writeBoolean(IndexOutput output, boolean b) throws IOException { - output.writeVInt(b ? 1 : 0); - } - - public IndexTracerResult(String segmentName, String field) { - _segmentName = segmentName; - _field = field; - } - - public void setPosition(long position, String fileName, FILE_TYPE type) { - switch (type) { - case TIM: - _timPosition = position; - _timCaptured = true; - _timFileName = fileName; - break; - case DOC: - _docPosition = position; - _docCaptured = true; - _docFileName = fileName; - break; - case PAY: - _payPosition = position; - _payCaptured = true; - _payFileName = fileName; - break; - case POS: - _posPosition = position; - _posCaptured = true; - _posFileName = fileName; - break; - default: - throw new NotSupported(type); - } - - } - - public long getPosition(FILE_TYPE type) { - switch (type) { - case TIM: - return _timPosition; - case DOC: - return _docPosition; - case PAY: - return _payPosition; - case POS: - return _posPosition; - default: - throw new NotSupported(type); - } - } - - public boolean isFilePositionCaptured(FILE_TYPE type) { - switch (type) { - case TIM: - return _timCaptured; - case DOC: - return _docCaptured; - case PAY: - return _payCaptured; - case POS: - return _posCaptured; - default: - throw new NotSupported(type); - } - } - - public String getFileName(FILE_TYPE type) { - switch (type) { - case TIM: - return _timFileName; - case DOC: - return _docFileName; - case PAY: - return _payFileName; - case POS: - return _posFileName; - default: - throw new NotSupported(type); - } - } - - public String getSegmentName() { - return _segmentName; - } - - public String getField() { - return _field; - } - - @Override - public String toString() { - return "IndexTracerResult [_docCaptured=" + _docCaptured + ", _docPosition=" + _docPosition + ", _docFileName=" - + _docFileName + ", _posCaptured=" + _posCaptured + ", _posPosition=" + _posPosition + ", _posFileName=" - + _posFileName + ", _payCaptured=" + _payCaptured + ", _payPosition=" + _payPosition + ", _payFileName=" - + _payFileName + ", _timCaptured=" + _timCaptured + ", _timPosition=" + _timPosition + ", _timFileName=" - + _timFileName + ", _segmentName=" + _segmentName + ", _field=" + _field + "]"; - } - -} http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/e9dce540/blur-store/src/main/java/org/apache/blur/lucene/warmup/IndexWarmup.java ---------------------------------------------------------------------- diff --git a/blur-store/src/main/java/org/apache/blur/lucene/warmup/IndexWarmup.java b/blur-store/src/main/java/org/apache/blur/lucene/warmup/IndexWarmup.java deleted file mode 100644 index 5433432..0000000 --- a/blur-store/src/main/java/org/apache/blur/lucene/warmup/IndexWarmup.java +++ /dev/null @@ -1,502 +0,0 @@ -package org.apache.blur.lucene.warmup; - -/** - * 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. - */ -import java.io.FileNotFoundException; -import java.io.IOException; -import java.util.ArrayList; -import java.util.HashMap; -import java.util.List; -import java.util.Map; -import java.util.Map.Entry; -import java.util.concurrent.TimeUnit; -import java.util.concurrent.atomic.AtomicBoolean; - -import org.apache.blur.index.ExitableReader.ExitableFilterAtomicReader; -import org.apache.blur.log.Log; -import org.apache.blur.log.LogFactory; -import org.apache.blur.lucene.warmup.IndexTracerResult.FILE_TYPE; -import org.apache.lucene.index.AtomicReader; -import org.apache.lucene.index.AtomicReaderContext; -import org.apache.lucene.index.Fields; -import org.apache.lucene.index.IndexReader; -import org.apache.lucene.index.IndexReaderContext; -import org.apache.lucene.index.SegmentReader; -import org.apache.lucene.index.Terms; -import org.apache.lucene.store.Directory; -import org.apache.lucene.store.IOContext; -import org.apache.lucene.store.IndexInput; -import org.apache.lucene.store.IndexOutput; -import org.apache.lucene.util.OpenBitSet; - -/** - * IndexWarmup is used to pre-read portions of the index by field. Usage:<br/> - * <br/> - * int maxSampleSize = 1000; <br/> - * Directory dir = FSDirectory.open(new File("/path/index"));<br/> - * dir = new TraceableDirectory(dir);<br/> - * DirectoryReader reader = DirectoryReader.open(dir);<br/> - * <br/> - * IndexWarmup indexWarmup = new IndexWarmup(new AtomicBoolean());<br/> - * Map<String, List<IndexTracerResult>> sampleIndex = - * indexWarmup.sampleIndex(reader, "");<br/> - * indexWarmup.warm(reader, sampleIndex, "uuid", null);<br/> - * indexWarmup.warm(reader, sampleIndex, "test", "test");<br/> - * indexWarmup.warm(reader, sampleIndex, "nothing", null);<br/> - * indexWarmup.warm(reader, sampleIndex, "id2", "tst");<br/> - */ -public class IndexWarmup { - - private static final Log LOG = LogFactory.getLog(IndexWarmup.class); - private static final String SAMPLE_EXT = ".sample"; - private static final long _5_SECONDS = TimeUnit.SECONDS.toNanos(5); - private static final int DEFAULT_THROTTLE = 2000000; - - private final AtomicBoolean _isClosed; - private final int _maxSampleSize; - private final long _maxBytesPerSec; - private final AtomicBoolean _stop; - - public IndexWarmup(AtomicBoolean isClosed, AtomicBoolean stop, int maxSampleSize, long maxBytesPerSec) { - _isClosed = isClosed; - _stop = stop; - _maxSampleSize = maxSampleSize; - _maxBytesPerSec = maxBytesPerSec; - } - - public IndexWarmup(AtomicBoolean isClosed, AtomicBoolean stop, int maxSampleSize) { - this(isClosed, stop, maxSampleSize, DEFAULT_THROTTLE); - } - - private static ThreadLocal<Boolean> runTrace = new ThreadLocal<Boolean>() { - @Override - protected Boolean initialValue() { - return Boolean.FALSE; - } - }; - - public static boolean isRunTrace() { - return runTrace.get(); - } - - public static void enableRunTrace() { - runTrace.set(Boolean.TRUE); - } - - public static void disableRunTrace() { - runTrace.set(Boolean.FALSE); - } - - public void warmFile(IndexReader reader, Map<String, List<IndexTracerResult>> sampleIndex, String fieldName, - String context) throws IOException { - for (Entry<String, List<IndexTracerResult>> segment : sampleIndex.entrySet()) { - warmFile(reader, segment.getValue(), fieldName, context); - } - } - - public void getFilePositionsToWarm(IndexReader reader, Map<String, List<IndexTracerResult>> sampleIndex, - String fieldName, String context, Map<String, OpenBitSet> filePartsToWarm, int blockSize) throws IOException { - for (Entry<String, List<IndexTracerResult>> segment : sampleIndex.entrySet()) { - getFilePositionsToWarm(reader, segment.getValue(), fieldName, context, filePartsToWarm, blockSize); - } - } - - public void getFilePositionsToWarm(IndexReader reader, List<IndexTracerResult> traces, String fieldName, - String context, Map<String, OpenBitSet> filePartsToWarm, int blockSize) throws IOException { - int index = find(traces, fieldName); - if (index < 0) { - // not found - return; - } - IndexTracerResult trace = traces.get(index); - for (FILE_TYPE type : FILE_TYPE.values()) { - if (trace.isFilePositionCaptured(type)) { - long startingPosition = trace.getPosition(type); - long endingPosition = Long.MAX_VALUE; - int nextIndex = findNextSetTrace(traces, index + 1, type); - if (nextIndex >= 0) { - IndexTracerResult next = traces.get(nextIndex); - endingPosition = next.getPosition(type); - } - String fileName = trace.getFileName(type); - String segmentName = trace.getSegmentName(); - getFilePositionsToWarm(reader, segmentName, fileName, fieldName, startingPosition, endingPosition, context, - filePartsToWarm, blockSize); - } - } - } - - private void getFilePositionsToWarm(IndexReader reader, String segmentName, String fileName, String fieldName, - long startingPosition, long endingPosition, String context, Map<String, OpenBitSet> filePartsToWarm, int blockSize) - throws IOException { - Directory dir = getDirectory(reader, segmentName, context); - long fileLength = dir.fileLength(fileName); - if (endingPosition == Long.MAX_VALUE) { - endingPosition = fileLength - 1; - } - OpenBitSet openBitSet = filePartsToWarm.get(fileName); - if (openBitSet == null) { - openBitSet = new OpenBitSet((long) Math.ceil(fileLength / (double) blockSize)); - filePartsToWarm.put(fileName, openBitSet); - } - long startingBlock = startingPosition / blockSize; - long endingBlock = endingPosition / blockSize; - openBitSet.set(startingBlock, endingBlock); - } - - void warmFile(IndexReader reader, List<IndexTracerResult> traces, String fieldName, String context) - throws IOException { - int index = find(traces, fieldName); - if (index < 0) { - // not found - return; - } - IndexTracerResult trace = traces.get(index); - for (FILE_TYPE type : FILE_TYPE.values()) { - if (trace.isFilePositionCaptured(type)) { - long startingPosition = trace.getPosition(type); - long endingPosition = Long.MAX_VALUE; - int nextIndex = findNextSetTrace(traces, index + 1, type); - if (nextIndex >= 0) { - IndexTracerResult next = traces.get(nextIndex); - endingPosition = next.getPosition(type); - } - String fileName = trace.getFileName(type); - String segmentName = trace.getSegmentName(); - warmFile(reader, segmentName, fileName, fieldName, startingPosition, endingPosition, context); - } - } - } - - private void warmFile(IndexReader reader, String segmentName, String fileName, String fieldName, - long startingPosition, long endingPosition, String context) throws IOException { - Directory dir = getDirectory(reader, segmentName, context); - if (dir == null) { - LOG.info("Context [{0}] cannot find segment [{1}]", context, segmentName); - return; - } - if (endingPosition == Long.MAX_VALUE) { - endingPosition = dir.fileLength(fileName) - 1; - } - if (_isClosed.get() || _stop.get()) { - LOG.info("Context [{0}] index closed", context); - return; - } - long length = endingPosition - startingPosition; - final long totalLength = length; - IndexInput input = new ThrottledIndexInput(dir.openInput(fileName, IOContext.READ), _maxBytesPerSec); - try { - input.seek(startingPosition); - byte[] buf = new byte[8192]; - long start = System.nanoTime(); - long bytesReadPerPass = 0; - while (length > 0) { - long now = System.nanoTime(); - if (start + _5_SECONDS < now) { - double seconds = (now - start) / 1000000000.0; - double rateMbPerSec = (bytesReadPerPass / seconds) / 1000 / 1000; - double complete = (((double) totalLength - (double) length) / (double) totalLength) * 100.0; - LOG.debug("Context [{3}] warming field [{0}] in file [{1}] is [{2}%] complete at rate of [{4} MB/s]", - fieldName, fileName, complete, context, rateMbPerSec); - start = System.nanoTime(); - bytesReadPerPass = 0; - if (_isClosed.get()) { - LOG.info("Context [{0}] index closed", context); - return; - } - } - int len = (int) Math.min(length, buf.length); - input.readBytes(buf, 0, len, false); - length -= len; - bytesReadPerPass += len; - } - long now = System.nanoTime(); - double seconds = (now - start) / 1000000000.0; - if (seconds < 1) { - seconds = 1; - } - double rateMbPerSec = (bytesReadPerPass / seconds) / 1000 / 1000; - LOG.info("Context [{3}] warming field [{0}] in file [{1}] is [{2}%] complete at rate of [{4} MB/s]", fieldName, - fileName, 100, context, rateMbPerSec); - } finally { - input.close(); - } - } - - private Directory getDirectory(IndexReader reader, String segmentName, String context) { - if (reader instanceof AtomicReader) { - return getDirectory((AtomicReader) reader, segmentName, context); - } - for (IndexReaderContext ctext : reader.getContext().leaves()) { - if (_isClosed.get() || _stop.get()) { - LOG.info("Context [{0}] index closed", context); - return null; - } - AtomicReaderContext atomicReaderContext = (AtomicReaderContext) ctext; - AtomicReader atomicReader = atomicReaderContext.reader(); - if (atomicReader instanceof SegmentReader) { - SegmentReader segmentReader = (SegmentReader) atomicReader; - if (segmentReader.getSegmentName().equals(segmentName)) { - return segmentReader.directory(); - } - } - } - return null; - } - - private Directory getDirectory(IndexReader reader, String context) { - if (reader instanceof AtomicReader) { - return getDirectory((AtomicReader) reader, context); - } - for (IndexReaderContext ctext : reader.getContext().leaves()) { - if (_isClosed.get() || _stop.get()) { - LOG.info("Context [{0}] index closed", context); - return null; - } - AtomicReaderContext atomicReaderContext = (AtomicReaderContext) ctext; - AtomicReader atomicReader = atomicReaderContext.reader(); - if (atomicReader instanceof SegmentReader) { - SegmentReader segmentReader = (SegmentReader) atomicReader; - return segmentReader.directory(); - } - } - return null; - } - - private Directory getDirectory(AtomicReader reader, String context) { - if (reader instanceof ExitableFilterAtomicReader) { - return getDirectory(((ExitableFilterAtomicReader) reader).getOriginalReader(), context); - } else if (reader instanceof SegmentReader) { - return ((SegmentReader) reader).directory(); - } - return null; - } - - private Directory getDirectory(AtomicReader atomicReader, String segmentName, String context) { - if (atomicReader instanceof SegmentReader) { - SegmentReader segmentReader = (SegmentReader) atomicReader; - if (segmentReader.getSegmentName().equals(segmentName)) { - return segmentReader.directory(); - } - } - return null; - } - - private int findNextSetTrace(List<IndexTracerResult> traces, int startingIndex, FILE_TYPE type) { - int size = traces.size(); - for (int i = startingIndex; i < size; i++) { - IndexTracerResult trace = traces.get(i); - if (trace.isFilePositionCaptured(type)) { - return i; - } - } - return -1; - } - - private int find(List<IndexTracerResult> traces, String fieldName) { - int index = 0; - for (IndexTracerResult trace : traces) { - if (trace.getField().equals(fieldName)) { - return index; - } - index++; - } - return -1; - } - - public Map<String, List<IndexTracerResult>> sampleIndex(IndexReader reader, String context) throws IOException { - Map<String, List<IndexTracerResult>> results = new HashMap<String, List<IndexTracerResult>>(); - for (IndexReaderContext ctext : reader.getContext().leaves()) { - if (_isClosed.get() || _stop.get()) { - LOG.info("Context [{0}] index closed", context); - return null; - } - AtomicReaderContext atomicReaderContext = (AtomicReaderContext) ctext; - AtomicReader atomicReader = atomicReaderContext.reader(); - results.putAll(sampleIndex(atomicReader, context)); - } - return results; - } - - public Map<String, List<IndexTracerResult>> sampleIndex(AtomicReader atomicReader, String context) throws IOException { - Map<String, List<IndexTracerResult>> results = new HashMap<String, List<IndexTracerResult>>(); - if (atomicReader instanceof SegmentReader) { - SegmentReader segmentReader = (SegmentReader) atomicReader; - Directory directory = segmentReader.directory(); - if (!(directory instanceof TraceableDirectory)) { - LOG.info("Context [{1}] cannot warmup directory [{0}] needs to be a TraceableDirectory.", directory, context); - return results; - } - IndexTracer tracer = new IndexTracer((TraceableDirectory) directory, _maxSampleSize); - String fileName = getSampleFileName(segmentReader.getSegmentName()); - List<IndexTracerResult> segmentTraces = new ArrayList<IndexTracerResult>(); - if (directory.fileExists(fileName)) { - IndexInput input = directory.openInput(fileName, IOContext.READONCE); - segmentTraces = read(input); - input.close(); - } else { - Fields fields = atomicReader.fields(); - for (String field : fields) { - LOG.debug("Context [{1}] sampling field [{0}].", field, context); - Terms terms = fields.terms(field); - boolean hasOffsets = terms.hasOffsets(); - boolean hasPayloads = terms.hasPayloads(); - boolean hasPositions = terms.hasPositions(); - - tracer.initTrace(segmentReader, field, hasPositions, hasPayloads, hasOffsets); - IndexTracerResult result = tracer.runTrace(terms); - segmentTraces.add(result); - } - if (_isClosed.get() || _stop.get()) { - LOG.info("Context [{0}] index closed", context); - return null; - } - IndexOutput output = directory.createOutput(fileName, IOContext.DEFAULT); - write(segmentTraces, output); - output.close(); - } - results.put(segmentReader.getSegmentName(), segmentTraces); - } - return results; - } - - public static String getSampleFileName(String segmentName) { - return segmentName + SAMPLE_EXT; - } - - private List<IndexTracerResult> read(IndexInput input) throws IOException { - int count = input.readVInt(); - List<IndexTracerResult> results = new ArrayList<IndexTracerResult>(count); - for (int i = 0; i < count; i++) { - results.add(IndexTracerResult.read(input)); - } - return results; - } - - private void write(List<IndexTracerResult> segmentTraces, IndexOutput output) throws IOException { - output.writeVInt(segmentTraces.size()); - for (IndexTracerResult r : segmentTraces) { - r.write(output); - } - } - - public void warmFile(IndexReader indexReader, Map<String, OpenBitSet> filePartsToWarm, String context, int blockSize, - int bufferSize) throws IOException { - Directory directory = getDirectory(indexReader, context); - ThrottledIndexInput input = null; - long s = System.nanoTime(); - double _targetThrouhput = _maxBytesPerSec / 1000.0 / 1000.0; - for (Entry<String, OpenBitSet> e : filePartsToWarm.entrySet()) { - try { - input = warmFile(directory, e.getKey(), e.getValue(), blockSize, context, bufferSize, input); - if (input != null) { - long end = System.nanoTime(); - double seconds = (end - s) / 1000000000.0; - double rateMbPerSec = (input.getTotalBytesRead() / seconds) / 1000 / 1000; - LOG.debug("Context [{0}] warming file [{1}] is [{2}%] complete " - + "at rate of [{3} MB/s] target was [{4} MB/s]", context, e.getKey(), 100, rateMbPerSec, _targetThrouhput); - } - } catch (FileNotFoundException ex) { - LOG.debug("File [{0}] no longer exists.", e.getKey()); - } - } - if (input != null) { - long end = System.nanoTime(); - double seconds = (end - s) / 1000000000.0; - double rateMbPerSec = (input.getTotalBytesRead() / seconds) / 1000 / 1000; - LOG.info("Finished warming context [{2}] [{0} MB/s] target was [{1} MB/s]", rateMbPerSec, _targetThrouhput, - context); - } - } - - private ThrottledIndexInput warmFile(Directory dir, String fileName, OpenBitSet bitSet, int blockSize, - String context, int bufferSize, ThrottledIndexInput lastInput) throws IOException { - if (bufferSize < blockSize) { - throw new IOException("Buffer Size [" + bufferSize + "] cannot be less than Block Size [" + blockSize + "]"); - } - if (_isClosed.get() || _stop.get()) { - LOG.info("Context [{0}] index closed", context); - return null; - } - ThrottledIndexInput input; - if (lastInput == null) { - input = new ThrottledIndexInput(dir.openInput(fileName, IOContext.READ), _maxBytesPerSec); - } else { - input = new ThrottledIndexInput(dir.openInput(fileName, IOContext.READ), _maxBytesPerSec, - lastInput.getTotalBytesRead(), lastInput.getTotalSleepTime(), lastInput.getStartTime()); - } - try { - long length = input.length(); - final long totalAmountToBeRead = bitSet.capacity() * blockSize; - byte[] buf = new byte[bufferSize]; - int maxNumberOfContiguousBlocks = bufferSize / blockSize; - long start = System.nanoTime(); - final long originalStart = start; - long bytesReadPerPass = 0; - long totalBytesRead = 0; - long blockPosition = 0; - while ((blockPosition = bitSet.nextSetBit(blockPosition)) != -1) { - long now = System.nanoTime(); - if (start + _5_SECONDS < now) { - double seconds = (now - start) / 1000000000.0; - double rateMbPerSec = (bytesReadPerPass / seconds) / 1000 / 1000; - double complete = (((double) totalAmountToBeRead - (double) totalBytesRead) / (double) totalAmountToBeRead) * 100.0; - LOG.debug("Context [{0}] warming file [{1}] is [{2}%] complete at rate of [{3} MB/s]", context, fileName, - complete, rateMbPerSec); - start = System.nanoTime(); - bytesReadPerPass = 0; - if (_isClosed.get()) { - LOG.info("Context [{0}] index closed", context); - return null; - } - } - long currentPosition = input.getFilePointer(); - long newPosition = blockPosition * blockSize; - if (newPosition != currentPosition) { - input.seek(newPosition); - } - int numberOfContiguousBlocks = getNumberOfContiguousBlock(maxNumberOfContiguousBlocks, bitSet, blockPosition); - int len = (int) Math.min(blockSize * numberOfContiguousBlocks, length - newPosition); - input.readBytes(buf, 0, len); - bytesReadPerPass += len; - totalBytesRead += len; - blockPosition += numberOfContiguousBlocks; - if (_isClosed.get()) { - LOG.info("Context [{0}] index closed", context); - return null; - } - } - long now = System.nanoTime(); - double seconds = (now - originalStart) / 1000000000.0; - double rateMbPerSec = (totalBytesRead / seconds) / 1000 / 1000; - LOG.debug("Context [{0}] warming file [{1}] is [{2}%] complete at rate of [{3} MB/s]", context, fileName, 100, - rateMbPerSec); - return input; - } finally { - input.close(); - } - } - - private int getNumberOfContiguousBlock(int maxNumberOfContiguousBlocks, OpenBitSet bitSet, long blockPosition) { - for (int i = 0; i < maxNumberOfContiguousBlocks; i++) { - if (!bitSet.get(i + blockPosition)) { - return i; - } - } - return maxNumberOfContiguousBlocks; - } -} http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/e9dce540/blur-store/src/main/java/org/apache/blur/lucene/warmup/NotSupported.java ---------------------------------------------------------------------- diff --git a/blur-store/src/main/java/org/apache/blur/lucene/warmup/NotSupported.java b/blur-store/src/main/java/org/apache/blur/lucene/warmup/NotSupported.java deleted file mode 100644 index a9e2f0e..0000000 --- a/blur-store/src/main/java/org/apache/blur/lucene/warmup/NotSupported.java +++ /dev/null @@ -1,28 +0,0 @@ -package org.apache.blur.lucene.warmup; -/** - * 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. - */ -import org.apache.blur.lucene.warmup.IndexTracerResult.FILE_TYPE; - -public class NotSupported extends RuntimeException { - - private static final long serialVersionUID = 5988131795588013735L; - - public NotSupported(FILE_TYPE type) { - super("FILE_TYPE of [" + type.name() + "] not supported"); - } - -}
