You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@lucene.apache.org by at...@apache.org on 2019/11/28 14:48:27 UTC
[lucene-solr] branch revert-916-revised_LUCENE-8213 created (now
09cd82a)
This is an automated email from the ASF dual-hosted git repository.
atri pushed a change to branch revert-916-revised_LUCENE-8213
in repository https://gitbox.apache.org/repos/asf/lucene-solr.git.
at 09cd82a Revert "LUCENE-8213: Asynchronous Caching in LRUQueryCache (#916)"
This branch includes the following new commits:
new 09cd82a Revert "LUCENE-8213: Asynchronous Caching in LRUQueryCache (#916)"
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
[lucene-solr] 01/01: Revert "LUCENE-8213: Asynchronous Caching in
LRUQueryCache (#916)"
Posted by at...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
atri pushed a commit to branch revert-916-revised_LUCENE-8213
in repository https://gitbox.apache.org/repos/asf/lucene-solr.git
commit 09cd82a6fc51b90275ed326812e1d947c863952e
Author: Atri Sharma <at...@gmail.com>
AuthorDate: Thu Nov 28 20:18:20 2019 +0530
Revert "LUCENE-8213: Asynchronous Caching in LRUQueryCache (#916)"
This reverts commit 2d98a619dd96abfa716098bbc5cac2bc3a1b7d49.
---
lucene/CHANGES.txt | 3 -
.../org/apache/lucene/search/IndexSearcher.java | 10 +-
.../org/apache/lucene/search/LRUQueryCache.java | 76 +--
.../java/org/apache/lucene/search/QueryCache.java | 5 +-
.../apache/lucene/search/TestIndexSearcher.java | 7 +-
.../apache/lucene/search/TestLRUQueryCache.java | 607 +++------------------
6 files changed, 80 insertions(+), 628 deletions(-)
diff --git a/lucene/CHANGES.txt b/lucene/CHANGES.txt
index de61143..5200f8e 100644
--- a/lucene/CHANGES.txt
+++ b/lucene/CHANGES.txt
@@ -55,9 +55,6 @@ Improvements
* LUCENE-8984: MoreLikeThis MLT is biased for uncommon fields (Andy Hind via Anshum Gupta)
-* LUCENE-8213: LRUQueryCache#doCache now uses IndexSearcher's Executor (if present)
- to asynchronously cache heavy queries (Atri Sharma)
-
Bug fixes
* LUCENE-8663: NRTCachingDirectory.slowFileExists may open a file while
diff --git a/lucene/core/src/java/org/apache/lucene/search/IndexSearcher.java b/lucene/core/src/java/org/apache/lucene/search/IndexSearcher.java
index df6c720..66d3b61 100644
--- a/lucene/core/src/java/org/apache/lucene/search/IndexSearcher.java
+++ b/lucene/core/src/java/org/apache/lucene/search/IndexSearcher.java
@@ -181,13 +181,7 @@ public class IndexSearcher {
}
/** Runs searches for each segment separately, using the
- * provided Executor. The passed in Executor will also be
- * used by LRUQueryCache (if enabled) to perform asynchronous
- * query caching.
- * If a task is rejected by the host Executor, the failed task
- * will then be executed on the caller thread. This is done to
- * ensure that a query succeeds, albeit with a higher latency.
- * NOTE:
+ * provided Executor. NOTE:
* if you are using {@link NIOFSDirectory}, do not use
* the shutdownNow method of ExecutorService as this uses
* Thread.interrupt under-the-hood which can silently
@@ -849,7 +843,7 @@ public class IndexSearcher {
final QueryCache queryCache = this.queryCache;
Weight weight = query.createWeight(this, scoreMode, boost);
if (scoreMode.needsScores() == false && queryCache != null) {
- weight = queryCache.doCache(weight, queryCachingPolicy, executor);
+ weight = queryCache.doCache(weight, queryCachingPolicy);
}
return weight;
}
diff --git a/lucene/core/src/java/org/apache/lucene/search/LRUQueryCache.java b/lucene/core/src/java/org/apache/lucene/search/LRUQueryCache.java
index 2079876..da88142 100644
--- a/lucene/core/src/java/org/apache/lucene/search/LRUQueryCache.java
+++ b/lucene/core/src/java/org/apache/lucene/search/LRUQueryCache.java
@@ -28,9 +28,6 @@ import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
-import java.util.concurrent.Executor;
-import java.util.concurrent.FutureTask;
-import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.locks.ReentrantLock;
import java.util.function.Predicate;
@@ -91,6 +88,7 @@ import static org.apache.lucene.util.RamUsageEstimator.QUERY_DEFAULT_RAM_BYTES_U
* @lucene.experimental
*/
public class LRUQueryCache implements QueryCache, Accountable {
+
private final int maxSize;
private final long maxRamBytesUsed;
private final Predicate<LeafReaderContext> leavesToCache;
@@ -273,7 +271,6 @@ public class LRUQueryCache implements QueryCache, Accountable {
assert lock.isHeldByCurrentThread();
assert key instanceof BoostQuery == false;
assert key instanceof ConstantScoreQuery == false;
-
final IndexReader.CacheKey readerKey = cacheHelper.getKey();
final LeafCache leafCache = cache.get(readerKey);
if (leafCache == null) {
@@ -407,15 +404,6 @@ public class LRUQueryCache implements QueryCache, Accountable {
}
}
- // Get original weight from the cached weight
- private Weight getOriginalWeight(Weight weight) {
- while (weight instanceof CachingWrapperWeight) {
- weight = ((CachingWrapperWeight) weight).in;
- }
-
- return weight;
- }
-
// pkg-private for testing
void assertConsistent() {
lock.lock();
@@ -470,10 +458,12 @@ public class LRUQueryCache implements QueryCache, Accountable {
}
@Override
- public Weight doCache(final Weight weight, QueryCachingPolicy policy, Executor executor) {
- Weight originalWeight = getOriginalWeight(weight);
+ public Weight doCache(Weight weight, QueryCachingPolicy policy) {
+ while (weight instanceof CachingWrapperWeight) {
+ weight = ((CachingWrapperWeight) weight).in;
+ }
- return new CachingWrapperWeight(originalWeight, policy, executor);
+ return new CachingWrapperWeight(weight, policy);
}
@Override
@@ -675,13 +665,10 @@ public class LRUQueryCache implements QueryCache, Accountable {
// threads when IndexSearcher is created with threads
private final AtomicBoolean used;
- private final Executor executor;
-
- CachingWrapperWeight(Weight in, QueryCachingPolicy policy, Executor executor) {
+ CachingWrapperWeight(Weight in, QueryCachingPolicy policy) {
super(in.getQuery(), 1f);
this.in = in;
this.policy = policy;
- this.executor = executor;
used = new AtomicBoolean(false);
}
@@ -769,17 +756,6 @@ public class LRUQueryCache implements QueryCache, Accountable {
return supplier.get(leadCost);
}
- boolean cacheSynchronously = executor == null;
- if (cacheSynchronously == false) {
- boolean asyncCachingSucceeded = cacheAsynchronously(context, cacheHelper);
-
- // If async caching failed, synchronous caching will
- // be performed, hence do not return the uncached value
- if (asyncCachingSucceeded) {
- return supplier.get(leadCost);
- }
- }
-
Scorer scorer = supplier.get(Long.MAX_VALUE);
DocIdSet docIdSet = cacheImpl(new DefaultBulkScorer(scorer), context.reader().maxDoc());
putIfAbsent(in.getQuery(), docIdSet, cacheHelper);
@@ -876,19 +852,6 @@ public class LRUQueryCache implements QueryCache, Accountable {
if (docIdSet == null) {
if (policy.shouldCache(in.getQuery())) {
- boolean cacheSynchronously = executor == null;
- // If asynchronous caching is requested, perform the same and return
- // the uncached iterator
- if (cacheSynchronously == false) {
- boolean asyncCachingSucceeded = cacheAsynchronously(context, cacheHelper);
-
- // If async caching failed, we will perform synchronous caching
- // hence do not return the uncached value here
- if (asyncCachingSucceeded) {
- return in.bulkScorer(context);
- }
- }
-
docIdSet = cache(context);
putIfAbsent(in.getQuery(), docIdSet, cacheHelper);
} else {
@@ -908,30 +871,5 @@ public class LRUQueryCache implements QueryCache, Accountable {
return new DefaultBulkScorer(new ConstantScoreScorer(this, 0f, ScoreMode.COMPLETE_NO_SCORES, disi));
}
- // Perform a cache load asynchronously
- // @return true if asynchronous caching succeeded, false otherwise
- private boolean cacheAsynchronously(LeafReaderContext context, IndexReader.CacheHelper cacheHelper) {
- FutureTask<Void> task = new FutureTask<>(() -> {
- // If the reader is being closed -- do nothing
- if (context.reader().tryIncRef()) {
- try {
- DocIdSet localDocIdSet = cache(context);
- putIfAbsent(in.getQuery(), localDocIdSet, cacheHelper);
- } finally {
- context.reader().decRef();
- }
- }
-
- return null;
- });
- try {
- executor.execute(task);
- } catch (RejectedExecutionException e) {
- // Trigger synchronous caching
- return false;
- }
-
- return true;
- }
}
}
diff --git a/lucene/core/src/java/org/apache/lucene/search/QueryCache.java b/lucene/core/src/java/org/apache/lucene/search/QueryCache.java
index 21716ed..94e34ee 100644
--- a/lucene/core/src/java/org/apache/lucene/search/QueryCache.java
+++ b/lucene/core/src/java/org/apache/lucene/search/QueryCache.java
@@ -17,8 +17,6 @@
package org.apache.lucene.search;
-import java.util.concurrent.Executor;
-
/**
* A cache for queries.
*
@@ -32,8 +30,7 @@ public interface QueryCache {
* matching docs per-segment accordingly to the given <code>policy</code>.
* NOTE: The returned weight will only be equivalent if scores are not needed.
* @see Collector#scoreMode()
- * If the Executor is not null, it will be used to perform asynchronous caching
*/
- Weight doCache(Weight weight, QueryCachingPolicy policy, Executor executor);
+ Weight doCache(Weight weight, QueryCachingPolicy policy);
}
diff --git a/lucene/core/src/test/org/apache/lucene/search/TestIndexSearcher.java b/lucene/core/src/test/org/apache/lucene/search/TestIndexSearcher.java
index 1f6f9d4..6617c60 100644
--- a/lucene/core/src/test/org/apache/lucene/search/TestIndexSearcher.java
+++ b/lucene/core/src/test/org/apache/lucene/search/TestIndexSearcher.java
@@ -24,7 +24,6 @@ import java.util.Collection;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
-import java.util.concurrent.Executor;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Future;
import java.util.concurrent.LinkedBlockingQueue;
@@ -35,8 +34,8 @@ import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicInteger;
import org.apache.lucene.document.Document;
-import org.apache.lucene.document.Field;
import org.apache.lucene.document.Field.Store;
+import org.apache.lucene.document.Field;
import org.apache.lucene.document.SortedDocValuesField;
import org.apache.lucene.document.StringField;
import org.apache.lucene.index.IndexReader;
@@ -189,7 +188,7 @@ public class TestIndexSearcher extends LuceneTestCase {
assertEquals(IndexSearcher.getDefaultQueryCache(), searcher.getQueryCache());
QueryCache dummyCache = new QueryCache() {
@Override
- public Weight doCache(Weight weight, QueryCachingPolicy policy, Executor executor) {
+ public Weight doCache(Weight weight, QueryCachingPolicy policy) {
return weight;
}
};
@@ -295,7 +294,7 @@ public class TestIndexSearcher extends LuceneTestCase {
service.shutdown();
}
- public static class RejectingMockExecutor implements ExecutorService {
+ private static class RejectingMockExecutor implements ExecutorService {
public void shutdown() {
}
diff --git a/lucene/core/src/test/org/apache/lucene/search/TestLRUQueryCache.java b/lucene/core/src/test/org/apache/lucene/search/TestLRUQueryCache.java
index 83811cb..68bc8c9 100644
--- a/lucene/core/src/test/org/apache/lucene/search/TestLRUQueryCache.java
+++ b/lucene/core/src/test/org/apache/lucene/search/TestLRUQueryCache.java
@@ -29,18 +29,8 @@ import java.util.HashSet;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
-import java.util.concurrent.Callable;
-import java.util.concurrent.CountDownLatch;
import java.util.Set;
import java.util.concurrent.ExecutionException;
-import java.util.concurrent.Executor;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Future;
-import java.util.concurrent.FutureTask;
-import java.util.concurrent.LinkedBlockingQueue;
-import java.util.concurrent.ThreadPoolExecutor;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
@@ -72,7 +62,6 @@ import org.apache.lucene.store.Directory;
import org.apache.lucene.util.Constants;
import org.apache.lucene.util.IOUtils;
import org.apache.lucene.util.LuceneTestCase;
-import org.apache.lucene.util.NamedThreadFactory;
import org.apache.lucene.util.RamUsageTester;
import org.apache.lucene.util.TestUtil;
@@ -110,9 +99,6 @@ public class TestLRUQueryCache extends LuceneTestCase {
final LRUQueryCache queryCache = new LRUQueryCache(1 + random().nextInt(20), 1 + random().nextInt(10000), context -> random().nextBoolean(), Float.POSITIVE_INFINITY);
Directory dir = newDirectory();
final RandomIndexWriter w = new RandomIndexWriter(random(), dir);
- ExecutorService service = new ThreadPoolExecutor(4, 4, 0L, TimeUnit.MILLISECONDS,
- new LinkedBlockingQueue<Runnable>(),
- new NamedThreadFactory("TestLRUQueryCache"));
final SearcherFactory searcherFactory = new SearcherFactory() {
@Override
public IndexSearcher newSearcher(IndexReader reader, IndexReader previous) throws IOException {
@@ -122,103 +108,87 @@ public class TestLRUQueryCache extends LuceneTestCase {
return searcher;
}
};
- final SearcherFactory concurrentSearcherFactory = new SearcherFactory() {
- @Override
- public IndexSearcher newSearcher(IndexReader reader, IndexReader previous) throws IOException {
- IndexSearcher searcher = new IndexSearcher(reader, service);
- searcher.setQueryCachingPolicy(MAYBE_CACHE_POLICY);
- searcher.setQueryCache(queryCache);
- return searcher;
+ final boolean applyDeletes = random().nextBoolean();
+ final SearcherManager mgr = new SearcherManager(w.w, applyDeletes, false, searcherFactory);
+ final AtomicBoolean indexing = new AtomicBoolean(true);
+ final AtomicReference<Throwable> error = new AtomicReference<>();
+ final int numDocs = atLeast(10000);
+ Thread[] threads = new Thread[3];
+ threads[0] = new Thread() {
+ public void run() {
+ Document doc = new Document();
+ StringField f = new StringField("color", "", Store.NO);
+ doc.add(f);
+ for (int i = 0; indexing.get() && i < numDocs; ++i) {
+ f.setStringValue(RandomPicks.randomFrom(random(), new String[] {"blue", "red", "yellow"}));
+ try {
+ w.addDocument(doc);
+ if ((i & 63) == 0) {
+ mgr.maybeRefresh();
+ if (rarely()) {
+ queryCache.clear();
+ }
+ if (rarely()) {
+ final String color = RandomPicks.randomFrom(random(), new String[] {"blue", "red", "yellow"});
+ w.deleteDocuments(new Term("color", color));
+ }
+ }
+ } catch (Throwable t) {
+ error.compareAndSet(null, t);
+ break;
+ }
+ }
+ indexing.set(false);
}
};
-
- final SearcherFactory[] searcherFactories = {searcherFactory, concurrentSearcherFactory};
-
- for (SearcherFactory currentSearcherFactory : searcherFactories) {
- final boolean applyDeletes = random().nextBoolean();
- final SearcherManager mgr = new SearcherManager(w.w, applyDeletes, false, currentSearcherFactory);
- final AtomicBoolean indexing = new AtomicBoolean(true);
- final AtomicReference<Throwable> error = new AtomicReference<>();
- final int numDocs = atLeast(10000);
- Thread[] threads = new Thread[3];
- threads[0] = new Thread() {
+ for (int i = 1; i < threads.length; ++i) {
+ threads[i] = new Thread() {
+ @Override
public void run() {
- Document doc = new Document();
- StringField f = new StringField("color", "", Store.NO);
- doc.add(f);
- for (int i = 0; indexing.get() && i < numDocs; ++i) {
- f.setStringValue(RandomPicks.randomFrom(random(), new String[]{"blue", "red", "yellow"}));
+ while (indexing.get()) {
try {
- w.addDocument(doc);
- if ((i & 63) == 0) {
- mgr.maybeRefresh();
- if (rarely()) {
- queryCache.clear();
- }
- if (rarely()) {
- final String color = RandomPicks.randomFrom(random(), new String[]{"blue", "red", "yellow"});
- w.deleteDocuments(new Term("color", color));
- }
+ final IndexSearcher searcher = mgr.acquire();
+ try {
+ final String value = RandomPicks.randomFrom(random(), new String[] {"blue", "red", "yellow", "green"});
+ final Query q = new TermQuery(new Term("color", value));
+ TotalHitCountCollector collector = new TotalHitCountCollector();
+ searcher.search(q, collector); // will use the cache
+ final int totalHits1 = collector.getTotalHits();
+ TotalHitCountCollector collector2 = new TotalHitCountCollector();
+ searcher.search(q, new FilterCollector(collector2) {
+ public ScoreMode scoreMode() {
+ return ScoreMode.COMPLETE; // will not use the cache because of scores
+ }
+ });
+ final long totalHits2 = collector2.getTotalHits();
+ assertEquals(totalHits2, totalHits1);
+ } finally {
+ mgr.release(searcher);
}
} catch (Throwable t) {
error.compareAndSet(null, t);
- break;
}
}
- indexing.set(false);
}
};
- for (int i = 1; i < threads.length; ++i) {
- threads[i] = new Thread() {
- @Override
- public void run() {
- while (indexing.get()) {
- try {
- final IndexSearcher searcher = mgr.acquire();
- try {
- final String value = RandomPicks.randomFrom(random(), new String[]{"blue", "red", "yellow", "green"});
- final Query q = new TermQuery(new Term("color", value));
- TotalHitCountCollector collector = new TotalHitCountCollector();
- searcher.search(q, collector); // will use the cache
- final int totalHits1 = collector.getTotalHits();
- TotalHitCountCollector collector2 = new TotalHitCountCollector();
- searcher.search(q, new FilterCollector(collector2) {
- public ScoreMode scoreMode() {
- return ScoreMode.COMPLETE; // will not use the cache because of scores
- }
- });
- final long totalHits2 = collector2.getTotalHits();
- assertEquals(totalHits2, totalHits1);
- } finally {
- mgr.release(searcher);
- }
- } catch (Throwable t) {
- error.compareAndSet(null, t);
- }
- }
- }
- };
- }
-
- for (Thread thread : threads) {
- thread.start();
- }
+ }
- for (Thread thread : threads) {
- thread.join();
- }
+ for (Thread thread : threads) {
+ thread.start();
+ }
- if (error.get() != null) {
- throw error.get();
- }
- queryCache.assertConsistent();
- mgr.close();
+ for (Thread thread : threads) {
+ thread.join();
}
+ if (error.get() != null) {
+ throw error.get();
+ }
+ queryCache.assertConsistent();
+ mgr.close();
w.close();
dir.close();
queryCache.assertConsistent();
- service.shutdown();
}
public void testLRUEviction() throws Exception {
@@ -305,222 +275,6 @@ public class TestLRUQueryCache extends LuceneTestCase {
dir.close();
}
- public void testLRUConcurrentLoadAndEviction() throws Exception {
- Directory dir = newDirectory();
- final RandomIndexWriter w = new RandomIndexWriter(random(), dir);
-
- Document doc = new Document();
- StringField f = new StringField("color", "blue", Store.NO);
- doc.add(f);
- w.addDocument(doc);
- f.setStringValue("red");
- w.addDocument(doc);
- f.setStringValue("green");
- w.addDocument(doc);
- final DirectoryReader reader = w.getReader();
- ExecutorService service = new ThreadPoolExecutor(4, 4, 0L, TimeUnit.MILLISECONDS,
- new LinkedBlockingQueue<Runnable>(),
- new NamedThreadFactory("TestLRUQueryCache"));
-
- IndexSearcher searcher = new IndexSearcher(reader, service);
-
- final CountDownLatch[] latch = {new CountDownLatch(1)};
-
- final LRUQueryCache queryCache = new LRUQueryCache(2, 100000, context -> true, Float.POSITIVE_INFINITY) {
- @Override
- protected void onDocIdSetCache(Object readerCoreKey, long ramBytesUsed) {
- super.onDocIdSetCache(readerCoreKey, ramBytesUsed);
- latch[0].countDown();
- }
- };
-
- final Query blue = new TermQuery(new Term("color", "blue"));
- final Query red = new TermQuery(new Term("color", "red"));
- final Query green = new TermQuery(new Term("color", "green"));
-
- assertEquals(Collections.emptyList(), queryCache.cachedQueries());
-
- searcher.setQueryCache(queryCache);
- // the filter is not cached on any segment: no changes
- searcher.setQueryCachingPolicy(NEVER_CACHE);
- searcher.search(new ConstantScoreQuery(green), 1);
- assertEquals(Collections.emptyList(), queryCache.cachedQueries());
-
- searcher.setQueryCachingPolicy(ALWAYS_CACHE);
-
- // First read should miss
- searcher.search(new ConstantScoreQuery(red), 1);
-
- // Let the cache load be completed
- latch[0].await();
- assertEquals(Collections.singletonList(red), queryCache.cachedQueries());
-
- // Second read should hit
- searcher.search(new ConstantScoreQuery(red), 1);
- assertEquals(Collections.singletonList(red), queryCache.cachedQueries());
- assertEquals(queryCache.getHitCount(), 1);
-
- latch[0] = new CountDownLatch(1);
- searcher.search(new ConstantScoreQuery(green), 1);
-
- // Let the cache load be completed
- latch[0].await();
- assertEquals(Arrays.asList(red, green), queryCache.cachedQueries());
-
- searcher.search(new ConstantScoreQuery(red), 1);
- assertEquals(Arrays.asList(green, red), queryCache.cachedQueries());
- assertEquals(2, queryCache.getCacheCount());
-
- latch[0] = new CountDownLatch(1);
-
- searcher.search(new ConstantScoreQuery(blue), 1);
-
- // Let the cache load be completed
- latch[0].await();
- assertEquals(Arrays.asList(red, blue), queryCache.cachedQueries());
-
- searcher.search(new ConstantScoreQuery(blue), 1);
- assertEquals(Arrays.asList(red, blue), queryCache.cachedQueries());
- assertEquals(3, queryCache.getCacheCount());
-
- latch[0] = new CountDownLatch(1);
-
- searcher.search(new ConstantScoreQuery(green), 1);
-
- // Let the cache load be completed
- latch[0].await();
- assertEquals(Arrays.asList(blue, green), queryCache.cachedQueries());
-
- service.shutdown();
- service.awaitTermination(300, TimeUnit.MILLISECONDS);
-
- searcher.setQueryCachingPolicy(NEVER_CACHE);
- searcher.search(new ConstantScoreQuery(red), 1);
- assertEquals(Arrays.asList(blue, green), queryCache.cachedQueries());
-
- reader.close();
- w.close();
- dir.close();
- }
-
- public void testLRUConcurrentLoadsOfSameQuery() throws Exception {
- Directory dir = newDirectory();
- final RandomIndexWriter w = new RandomIndexWriter(random(), dir);
-
- Document doc = new Document();
- StringField f = new StringField("color", "blue", Store.NO);
- doc.add(f);
- w.addDocument(doc);
- f.setStringValue("red");
- w.addDocument(doc);
- f.setStringValue("green");
- w.addDocument(doc);
- final DirectoryReader reader = w.getReader();
- ExecutorService service = new ThreadPoolExecutor(4, 4, 0L, TimeUnit.MILLISECONDS,
- new LinkedBlockingQueue<Runnable>(),
- new NamedThreadFactory("TestLRUQueryCache"));
-
- ExecutorService stressService = new ThreadPoolExecutor(15, 15, 0L, TimeUnit.MILLISECONDS,
- new LinkedBlockingQueue<Runnable>(),
- new NamedThreadFactory("TestLRUQueryCache2"));
-
- IndexSearcher searcher = new IndexSearcher(reader, service);
-
- final CountDownLatch latch = new CountDownLatch(1);
-
- final LRUQueryCache queryCache = new LRUQueryCache(2, 100000, context -> true, Float.POSITIVE_INFINITY) {
- @Override
- protected void onDocIdSetCache(Object readerCoreKey, long ramBytesUsed) {
- super.onDocIdSetCache(readerCoreKey, ramBytesUsed);
- latch.countDown();
- }
- };
-
- final Query green = new TermQuery(new Term("color", "green"));
-
- assertEquals(Collections.emptyList(), queryCache.cachedQueries());
-
- searcher.setQueryCache(queryCache);
- searcher.setQueryCachingPolicy(ALWAYS_CACHE);
-
- CountDownLatch startLatch = new CountDownLatch(1);
-
- FutureTask<Void> task = new FutureTask<>(() -> {
- startLatch.await();
- searcher.search(new ConstantScoreQuery(green), 1);
- return null;
- });
-
- for (int i = 0; i < 5; i++) {
- stressService.submit(task);
- }
-
- startLatch.countDown();
-
- latch.await();
- assertEquals(Arrays.asList(green), queryCache.cachedQueries());
-
- reader.close();
- w.close();
- dir.close();
- service.shutdown();
- stressService.shutdown();
- }
-
- public void testLRUConcurrentCachingAcrossSegments() throws Exception {
- Directory dir = newDirectory();
- final RandomIndexWriter w = new RandomIndexWriter(random(), dir);
-
- int numDocs = atLeast(150);
- int numIterations = atLeast(3);
-
- for (int i = 0; i < numIterations; i++) {
- for (int j = 0; j < numDocs; j++) {
- Document doc = new Document();
- StringField f = new StringField("color", "blue", Store.NO);
- doc.add(f);
- w.addDocument(doc);
- w.addDocument(doc);
- w.addDocument(doc);
- }
- w.commit();
- }
-
- final DirectoryReader reader = w.getReader();
-
- ExecutorService service = new ThreadPoolExecutor(4, 4, 0L, TimeUnit.MILLISECONDS,
- new LinkedBlockingQueue<Runnable>(),
- new NamedThreadFactory("TestLRUQueryCache"));
-
- IndexSearcher searcher = new IndexSearcher(reader, service) {
- @Override
- protected LeafSlice[] slices(List<LeafReaderContext> leaves) {
- ArrayList<LeafSlice> slices = new ArrayList<>();
- for (LeafReaderContext ctx : leaves) {
- slices.add(new LeafSlice(Arrays.asList(ctx)));
- }
- return slices.toArray(new LeafSlice[0]);
- }
- };
-
- final LRUQueryCache queryCache = new LRUQueryCache(2, 100000, context -> true, Float.POSITIVE_INFINITY);
-
- final Query blue = new TermQuery(new Term("color", "blue"));
-
- assertEquals(Collections.emptyList(), queryCache.cachedQueries());
-
- searcher.setQueryCache(queryCache);
- searcher.setQueryCachingPolicy(ALWAYS_CACHE);
- assert searcher.getSlices().length > 1;
-
- searcher.search(new ConstantScoreQuery(blue), 1);
-
- reader.close();
- w.close();
- dir.close();
- service.shutdown();
- }
-
public void testClearFilter() throws IOException {
Directory dir = newDirectory();
final RandomIndexWriter w = new RandomIndexWriter(random(), dir);
@@ -1447,7 +1201,7 @@ public class TestLRUQueryCache extends LuceneTestCase {
// test that the bulk scorer is propagated when a scorer should not be cached
Weight weight = searcher.createWeight(new MatchAllDocsQuery(), ScoreMode.COMPLETE_NO_SCORES, 1);
weight = new WeightWrapper(weight, scorerCalled, bulkScorerCalled);
- weight = cache.doCache(weight, NEVER_CACHE, null /* executor */);
+ weight = cache.doCache(weight, NEVER_CACHE);
weight.bulkScorer(leaf);
assertEquals(true, bulkScorerCalled.get());
assertEquals(false, scorerCalled.get());
@@ -1484,7 +1238,7 @@ public class TestLRUQueryCache extends LuceneTestCase {
dir.close();
}
- public void testMinSegmentSizePredicate() throws IOException, InterruptedException {
+ public void testMinSegmentSizePredicate() throws IOException {
Directory dir = newDirectory();
IndexWriterConfig iwc = newIndexWriterConfig().setMergePolicy(NoMergePolicy.INSTANCE);
RandomIndexWriter w = new RandomIndexWriter(random(), dir, iwc);
@@ -1496,41 +1250,16 @@ public class TestLRUQueryCache extends LuceneTestCase {
LRUQueryCache cache = new LRUQueryCache(2, 10000, new LRUQueryCache.MinSegmentSizePredicate(2, 0f), Float.POSITIVE_INFINITY);
searcher.setQueryCache(cache);
searcher.count(new DummyQuery());
-
assertEquals(0, cache.getCacheCount());
- final CountDownLatch[] latch = { new CountDownLatch(1)};
- cache = new LRUQueryCache(2, 10000,
- new LRUQueryCache.MinSegmentSizePredicate(1, 0f), Float.POSITIVE_INFINITY) {
- @Override
- protected void onDocIdSetCache(Object readerCoreKey, long ramBytesUsed) {
- super.onDocIdSetCache(readerCoreKey, ramBytesUsed);
- latch[0].countDown();
- }
- };
-
+ cache = new LRUQueryCache(2, 10000, new LRUQueryCache.MinSegmentSizePredicate(1, 0f), Float.POSITIVE_INFINITY);
searcher.setQueryCache(cache);
-
searcher.count(new DummyQuery());
-
- latch[0].await();
assertEquals(1, cache.getCacheCount());
- latch[0] = new CountDownLatch(1);
- cache = new LRUQueryCache(2, 10000,
- new LRUQueryCache.MinSegmentSizePredicate(0, .6f), Float.POSITIVE_INFINITY) {
- @Override
- protected void onDocIdSetCache(Object readerCoreKey, long ramBytesUsed) {
- super.onDocIdSetCache(readerCoreKey, ramBytesUsed);
- latch[0].countDown();
- }
- };
-
+ cache = new LRUQueryCache(2, 10000, new LRUQueryCache.MinSegmentSizePredicate(0, .6f), Float.POSITIVE_INFINITY);
searcher.setQueryCache(cache);
-
searcher.count(new DummyQuery());
-
- latch[0].await();
assertEquals(1, cache.getCacheCount());
w.addDocument(new Document());
@@ -1984,208 +1713,6 @@ public class TestLRUQueryCache extends LuceneTestCase {
t.join();
}
- public void testRejectedExecution() throws IOException {
- ExecutorService service = new TestIndexSearcher.RejectingMockExecutor();
- Directory dir = newDirectory();
- final RandomIndexWriter w = new RandomIndexWriter(random(), dir);
-
- Document doc = new Document();
- StringField f = new StringField("color", "blue", Store.NO);
- doc.add(f);
- w.addDocument(doc);
- f.setStringValue("red");
- w.addDocument(doc);
- f.setStringValue("green");
- w.addDocument(doc);
- final DirectoryReader reader = w.getReader();
-
- final Query red = new TermQuery(new Term("color", "red"));
-
- IndexSearcher searcher = new IndexSearcher(reader, service);
-
- final LRUQueryCache queryCache = new LRUQueryCache(2, 100000, context -> true, Float.POSITIVE_INFINITY);
-
- searcher.setQueryCache(queryCache);
- searcher.setQueryCachingPolicy(ALWAYS_CACHE);
-
- // To ensure that failing ExecutorService still allows query to run
- // successfully
-
- searcher.search(new ConstantScoreQuery(red), 1);
- assertEquals(Collections.singletonList(red), queryCache.cachedQueries());
-
- reader.close();
- w.close();
- dir.close();
- service.shutdown();
- }
-
- public void testFailedAsyncCaching() throws IOException {
- ExecutorService service = new TestIndexSearcher.RejectingMockExecutor();
-
- Directory dir = newDirectory();
- final RandomIndexWriter w = new RandomIndexWriter(random(), dir);
-
- for (int i = 0; i < 5; i++) {
- Document doc = new Document();
- StringField f = new StringField("color", "blue", Store.NO);
- doc.add(f);
- w.addDocument(doc);
- f.setStringValue("red");
- w.addDocument(doc);
- f.setStringValue("green");
- w.addDocument(doc);
- }
-
- w.commit();
-
- final DirectoryReader reader = w.getReader();
-
- final Query red = new TermQuery(new Term("color", "red"));
-
- IndexSearcher searcher = new IndexSearcher(reader, service);
-
- assertEquals(1, searcher.leafContexts.size());
-
- final LRUQueryCache queryCache = new LRUQueryCache(2, 100000, context -> true, Float.POSITIVE_INFINITY);
-
- searcher.setQueryCache(queryCache);
- searcher.setQueryCachingPolicy(ALWAYS_CACHE);
-
- searcher.search(new ConstantScoreQuery(red), 1);
-
- assertEquals(Collections.singletonList(red), queryCache.cachedQueries());
-
- w.close();
- reader.close();
- dir.close();
- service.shutdown();
- }
-
- public void testAsyncCachingHitsClosedReader() throws Exception {
- Directory dir = newDirectory();
- final RandomIndexWriter w = new RandomIndexWriter(random(), dir);
- Document doc = new Document();
- doc.add(new StringField("color", "red", Store.NO));
- w.addDocument(doc);
- IndexReader reader = w.getReader();
- w.close();
-
- // Because our index has a single document, it also has a single segment. In that
- // case, IndexSearcher will use the current thread for searching, so the only thing
- // that runs via the executor is the caching of queries.
- final CountDownLatch[] awaitCaching = new CountDownLatch[1];
- awaitCaching[0] = new CountDownLatch(1);
- final CountDownLatch[] cachingRan = new CountDownLatch[1];
- cachingRan[0] = new CountDownLatch(1);
- final AtomicBoolean success = new AtomicBoolean(false);
- Executor executor = runnable -> {
- new Thread(() -> {
- try {
- awaitCaching[0].await();
- runnable.run();
- success.set(true);
- } catch (InterruptedException e) {
- Thread.currentThread().interrupt();
- } finally {
- cachingRan[0].countDown();
- }
- }).start();
- };
-
- final IndexSearcher searcher = new IndexSearcher(reader, executor);
- final LRUQueryCache queryCache = new LRUQueryCache(2, 100000, context -> true, Float.POSITIVE_INFINITY);
- searcher.setQueryCache(queryCache);
- searcher.setQueryCachingPolicy(ALWAYS_CACHE);
-
- searcher.search(new ConstantScoreQuery(new TermQuery(new Term("color", "red"))), 1);
- assertEquals(Collections.emptyList(), queryCache.cachedQueries());
- awaitCaching[0].countDown();
- cachingRan[0].await();
- assertTrue(success.get());
- assertEquals(Collections.singletonList(new TermQuery(new Term("color", "red"))), queryCache.cachedQueries());
-
- awaitCaching[0] = new CountDownLatch(1);
- cachingRan[0] = new CountDownLatch(1);
- success.set(false);
- queryCache.clear();
-
- searcher.search(new ConstantScoreQuery(new TermQuery(new Term("color", "red"))), 1);
- assertEquals(Collections.emptyList(), queryCache.cachedQueries());
- reader.close();
- awaitCaching[0].countDown();
- cachingRan[0].await();
- assertTrue(success.get());
- assertEquals(Collections.emptyList(), queryCache.cachedQueries());
-
- dir.close();
- }
-
- public static class BlockedMockExecutor implements ExecutorService {
-
- private final CountDownLatch countDownLatch;
-
- public BlockedMockExecutor(final CountDownLatch latch) {
- this.countDownLatch = latch;
- }
-
- public void shutdown() {
- }
-
- public List<Runnable> shutdownNow() {
- throw new UnsupportedOperationException();
- }
-
- public boolean isShutdown() {
- throw new UnsupportedOperationException();
- }
-
- public boolean isTerminated() {
- throw new UnsupportedOperationException();
- }
-
- public boolean awaitTermination(final long l, final TimeUnit timeUnit) throws InterruptedException {
- throw new UnsupportedOperationException();
- }
-
- public <T> Future<T> submit(final Callable<T> tCallable) {
- throw new UnsupportedOperationException();
- }
-
- public <T> Future<T> submit(final Runnable runnable, final T t) {
- throw new UnsupportedOperationException();
- }
-
- public Future<?> submit(final Runnable runnable) {
- throw new UnsupportedOperationException();
- }
-
- public <T> List<Future<T>> invokeAll(final Collection<? extends Callable<T>> callables) throws InterruptedException {
- throw new UnsupportedOperationException();
- }
-
- public <T> List<Future<T>> invokeAll(final Collection<? extends Callable<T>> callables, final long l, final TimeUnit timeUnit) throws InterruptedException {
- throw new UnsupportedOperationException();
- }
-
- public <T> T invokeAny(final Collection<? extends Callable<T>> callables) throws InterruptedException, ExecutionException {
- throw new UnsupportedOperationException();
- }
-
- public <T> T invokeAny(final Collection<? extends Callable<T>> callables, final long l, final TimeUnit timeUnit) throws InterruptedException, ExecutionException, TimeoutException {
- throw new UnsupportedOperationException();
- }
-
- public void execute(final Runnable runnable) {
- try {
- countDownLatch.await();
- runnable.run();
- } catch (InterruptedException e) {
- throw new RuntimeException(e.getMessage());
- }
- }
- }
-
public void testSkipCachingForRangeQuery() throws IOException {
Directory dir = newDirectory();
final RandomIndexWriter w = new RandomIndexWriter(random(), dir);