Added poor man's index warmup.
Project: http://git-wip-us.apache.org/repos/asf/incubator-blur/repo Commit: http://git-wip-us.apache.org/repos/asf/incubator-blur/commit/61b184a4 Tree: http://git-wip-us.apache.org/repos/asf/incubator-blur/tree/61b184a4 Diff: http://git-wip-us.apache.org/repos/asf/incubator-blur/diff/61b184a4 Branch: refs/heads/0.1.5 Commit: 61b184a469a9390e1ebed586d875f1e53e033c8e Parents: ce40829 Author: Aaron McCurry <[email protected]> Authored: Sat Mar 30 21:35:04 2013 -0400 Committer: Aaron McCurry <[email protected]> Committed: Sat Mar 30 21:35:04 2013 -0400 ---------------------------------------------------------------------- .../indexserver/DistributedIndexServer.java | 70 ++++++--------- .../apache/blur/manager/writer/BlurNRTIndex.java | 2 +- .../apache/blur/server/IndexSearcherClosable.java | 12 ++- 3 files changed, 37 insertions(+), 47 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/61b184a4/src/blur-core/src/main/java/org/apache/blur/manager/indexserver/DistributedIndexServer.java ---------------------------------------------------------------------- diff --git a/src/blur-core/src/main/java/org/apache/blur/manager/indexserver/DistributedIndexServer.java b/src/blur-core/src/main/java/org/apache/blur/manager/indexserver/DistributedIndexServer.java index 5dd1114..48c9a9a 100644 --- a/src/blur-core/src/main/java/org/apache/blur/manager/indexserver/DistributedIndexServer.java +++ b/src/blur-core/src/main/java/org/apache/blur/manager/indexserver/DistributedIndexServer.java @@ -80,6 +80,8 @@ import org.apache.hadoop.io.compress.CompressionCodec; import org.apache.lucene.index.IndexReader; import org.apache.lucene.search.similarities.Similarity; import org.apache.lucene.store.Directory; +import org.apache.lucene.store.IOContext; +import org.apache.lucene.store.IndexInput; import org.apache.zookeeper.CreateMode; import org.apache.zookeeper.KeeperException; import org.apache.zookeeper.ZooDefs.Ids; @@ -125,12 +127,12 @@ public class DistributedIndexServer extends AbstractIndexServer { private SharedMergeScheduler _mergeScheduler; private IndexInputCloser _closer = null; private ExecutorService _searchExecutor = null; - + private AtomicLong _tableCount = new AtomicLong(); private AtomicLong _indexCount = new AtomicLong(); private AtomicLong _segmentCount = new AtomicLong(); private AtomicLong _indexMemoryUsage = new AtomicLong(); - + public static interface ReleaseReader { void release() throws IOException; } @@ -140,8 +142,10 @@ public class DistributedIndexServer extends AbstractIndexServer { // EVICTION), EVICTION, TimeUnit.SECONDS); Metrics.newGauge(new MetricName(ORG_APACHE_BLUR, BLUR, TABLE_COUNT, _cluster), new AtomicLongGauge(_tableCount)); Metrics.newGauge(new MetricName(ORG_APACHE_BLUR, BLUR, INDEX_COUNT, _cluster), new AtomicLongGauge(_indexCount)); - Metrics.newGauge(new MetricName(ORG_APACHE_BLUR, BLUR, SEGMENT_COUNT, _cluster), new AtomicLongGauge(_segmentCount)); - Metrics.newGauge(new MetricName(ORG_APACHE_BLUR, BLUR, INDEX_MEMORY_USAGE, _cluster), new AtomicLongGauge(_indexMemoryUsage)); + Metrics + .newGauge(new MetricName(ORG_APACHE_BLUR, BLUR, SEGMENT_COUNT, _cluster), new AtomicLongGauge(_segmentCount)); + Metrics.newGauge(new MetricName(ORG_APACHE_BLUR, BLUR, INDEX_MEMORY_USAGE, _cluster), new AtomicLongGauge( + _indexMemoryUsage)); BlurUtil.setupZookeeper(_zookeeper, _cluster); _openerService = Executors.newThreadPool("shard-opener", _shardOpenerThreadCount); @@ -286,8 +290,8 @@ public class DistributedIndexServer extends AbstractIndexServer { } } - private void updateMetrics(Map<String, BlurIndex> indexes, AtomicLong segmentCount, - AtomicLong indexMemoryUsage) throws IOException { + private void updateMetrics(Map<String, BlurIndex> indexes, AtomicLong segmentCount, AtomicLong indexMemoryUsage) + throws IOException { // @TODO not sure how to do this yet // for (BlurIndex index : indexes.values()) { // IndexReader reader = index.getIndexReader(); @@ -556,7 +560,7 @@ public class DistributedIndexServer extends AbstractIndexServer { private BlurIndex warmUp(BlurIndex index, TableDescriptor table, String shard) throws IOException { final IndexSearcherClosable searcher = index.getIndexReader(); IndexReader reader = searcher.getIndexReader(); - warmUpAllSegments(reader); + warmUpAllSegments(searcher); _warmup.warmBlurIndex(table, shard, reader, index.isClosed(), new ReleaseReader() { @Override public void release() throws IOException { @@ -568,43 +572,21 @@ public class DistributedIndexServer extends AbstractIndexServer { return index; } - private void warmUpAllSegments(IndexReader reader) throws IOException { - LOG.warn("Warm up NOT supported yet."); - // Once the reader warm-up has been re-implemented, this code will change - // accordingly. - - // IndexReader[] indexReaders = reader.getSequentialSubReaders(); - // if (indexReaders != null) { - // for (IndexReader r : indexReaders) { - // warmUpAllSegments(r); - // } - // } - // int maxDoc = reader.maxDoc(); - // int numDocs = reader.numDocs(); - // FieldInfos fieldInfos = ReaderUtil.getMergedFieldInfos(reader); - // Collection<String> fieldNames = new ArrayList<String>(); - // for (FieldInfo fieldInfo : fieldInfos) { - // if (fieldInfo.isIndexed) { - // fieldNames.add(fieldInfo.name); - // } - // } - // int primeDocCount = reader.docFreq(BlurConstants.PRIME_DOC_TERM); - // TermDocs termDocs = reader.termDocs(BlurConstants.PRIME_DOC_TERM); - // termDocs.next(); - // termDocs.close(); - // - // TermPositions termPositions = - // reader.termPositions(BlurConstants.PRIME_DOC_TERM); - // if (termPositions.next()) { - // if (termPositions.freq() > 0) { - // termPositions.nextPosition(); - // } - // } - // termPositions.close(); - // LOG.info("Warmup of indexreader [" + reader + "] complete, maxDocs [" + - // maxDoc + "], numDocs [" + numDocs + "], primeDocumentCount [" + - // primeDocCount + "], fieldCount [" - // + fieldNames.size() + "]"); + private void warmUpAllSegments(IndexSearcherClosable searcher) throws IOException { + LOG.warn("Warm up, stupid impl"); + Directory directory = searcher.getDirectory(); + String[] listAll = directory.listAll(); + byte[] buf = new byte[8192]; + for (String file : listAll) { + LOG.info("Warning up [{0}]", file); + IndexInput input = directory.openInput(file, IOContext.READ); + long length = input.length(); + for (long i = 0; i < length; i += buf.length) { + int len = (int) Math.min(buf.length, length - i); + input.readBytes(buf, 0, len); + } + input.close(); + } } private synchronized Map<String, BlurIndex> openMissingShards(final String table, Set<String> shardsToServe, http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/61b184a4/src/blur-core/src/main/java/org/apache/blur/manager/writer/BlurNRTIndex.java ---------------------------------------------------------------------- diff --git a/src/blur-core/src/main/java/org/apache/blur/manager/writer/BlurNRTIndex.java b/src/blur-core/src/main/java/org/apache/blur/manager/writer/BlurNRTIndex.java index d6c631f..511a6a1 100644 --- a/src/blur-core/src/main/java/org/apache/blur/manager/writer/BlurNRTIndex.java +++ b/src/blur-core/src/main/java/org/apache/blur/manager/writer/BlurNRTIndex.java @@ -90,7 +90,7 @@ public class BlurNRTIndex extends BlurIndex { _searcherFactory = new SearcherFactory() { @Override public IndexSearcher newSearcher(IndexReader reader) throws IOException { - return new IndexSearcherClosable(reader, searchExecutor, _nrtManagerRef); + return new IndexSearcherClosable(reader, searchExecutor, _nrtManagerRef, _directory); } }; http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/61b184a4/src/blur-core/src/main/java/org/apache/blur/server/IndexSearcherClosable.java ---------------------------------------------------------------------- diff --git a/src/blur-core/src/main/java/org/apache/blur/server/IndexSearcherClosable.java b/src/blur-core/src/main/java/org/apache/blur/server/IndexSearcherClosable.java index b91c9c5..64574d8 100644 --- a/src/blur-core/src/main/java/org/apache/blur/server/IndexSearcherClosable.java +++ b/src/blur-core/src/main/java/org/apache/blur/server/IndexSearcherClosable.java @@ -8,14 +8,22 @@ import java.util.concurrent.atomic.AtomicReference; import org.apache.lucene.index.IndexReader; import org.apache.lucene.search.IndexSearcher; import org.apache.lucene.search.NRTManager; +import org.apache.lucene.store.Directory; public class IndexSearcherClosable extends IndexSearcher implements Closeable { - private AtomicReference<NRTManager> _nrtManagerRef; + private final AtomicReference<NRTManager> _nrtManagerRef; + private final Directory _directory; - public IndexSearcherClosable(IndexReader r, ExecutorService executor, AtomicReference<NRTManager> nrtManagerRef) { + public IndexSearcherClosable(IndexReader r, ExecutorService executor, AtomicReference<NRTManager> nrtManagerRef, + Directory directory) { super(r, executor); _nrtManagerRef = nrtManagerRef; + _directory = directory; + } + + public Directory getDirectory() { + return _directory; } @Override
