Time Memory Leak: Search requests don't eagerly clean the search context, closes #153.

Author: kimchy

modules/elasticsearch/src/main/java/org/elasticsearch/action/search/type/TransportSearchDfsQueryThenFetchAction.java

@@ -69,6 +69,7 @@ private class AsyncAction extends BaseAsyncAction<DfsSearchResult> {
 
         private final Map<SearchShardTarget, FetchSearchResult> fetchResults = searchCache.obtainFetchResults();
 
+        private volatile Map<SearchShardTarget, ExtTIntArrayList> docIdsToLoad;
 
         private AsyncAction(SearchRequest request, ActionListener<SearchResponse> listener) {
             super(request, listener);
@@ -169,9 +170,9 @@ private void executeFetchPhase() {
         private void innerExecuteFetchPhase() {
             sortedShardList = searchPhaseController.sortDocs(queryResults.values());
             final Map<SearchShardTarget, ExtTIntArrayList> docIdsToLoad = searchPhaseController.docIdsToLoad(sortedShardList);
+            this.docIdsToLoad = docIdsToLoad;
 
             if (docIdsToLoad.isEmpty()) {
-                releaseIrrelevantSearchContexts(queryResults, docIdsToLoad);
                 finishHim();
             }
 
@@ -219,8 +220,6 @@ private void innerExecuteFetchPhase() {
                     }
                 }
             }
-
-            releaseIrrelevantSearchContexts(queryResults, docIdsToLoad);
         }
 
         private void executeFetch(final AtomicInteger counter, FetchSearchRequest fetchSearchRequest, DiscoveryNode node) {
@@ -259,6 +258,7 @@ private void innerFinishHim() {
             if (request.scroll() != null) {
                 scrollId = TransportSearchHelper.buildScrollId(request.searchType(), dfsResults);
             }
+            releaseIrrelevantSearchContexts(queryResults, docIdsToLoad);
             searchCache.releaseDfsResults(dfsResults);
             searchCache.releaseQueryResults(queryResults);
             searchCache.releaseFetchResults(fetchResults);

modules/elasticsearch/src/main/java/org/elasticsearch/action/search/type/TransportSearchQueryAndFetchAction.java

@@ -42,7 +42,7 @@
 import static org.elasticsearch.action.search.type.TransportSearchHelper.*;
 
 /**
- * @author kimchy (Shay Banon)
+ * @author kimchy (shay.banon)
  */
 public class TransportSearchQueryAndFetchAction extends TransportSearchTypeAction {
 

modules/elasticsearch/src/main/java/org/elasticsearch/action/search/type/TransportSearchQueryThenFetchAction.java

@@ -63,6 +63,7 @@ private class AsyncAction extends BaseAsyncAction<QuerySearchResult> {
 
         private final Map<SearchShardTarget, FetchSearchResult> fetchResults = searchCache.obtainFetchResults();
 
+        private volatile Map<SearchShardTarget, ExtTIntArrayList> docIdsToLoad;
 
         private AsyncAction(SearchRequest request, ActionListener<SearchResponse> listener) {
             super(request, listener);
@@ -83,9 +84,9 @@ private AsyncAction(SearchRequest request, ActionListener<SearchResponse> listen
         @Override protected void moveToSecondPhase() {
             sortedShardList = searchPhaseController.sortDocs(queryResults.values());
             final Map<SearchShardTarget, ExtTIntArrayList> docIdsToLoad = searchPhaseController.docIdsToLoad(sortedShardList);
+            this.docIdsToLoad = docIdsToLoad;
 
             if (docIdsToLoad.isEmpty()) {
-                releaseIrrelevantSearchContexts(queryResults, docIdsToLoad);
                 finishHim();
             }
 
@@ -134,8 +135,6 @@ private AsyncAction(SearchRequest request, ActionListener<SearchResponse> listen
                     }
                 }
             }
-
-            releaseIrrelevantSearchContexts(queryResults, docIdsToLoad);
         }
 
         private void executeFetch(final AtomicInteger counter, FetchSearchRequest fetchSearchRequest, DiscoveryNode node) {
@@ -174,6 +173,7 @@ private void innerFinishHim() {
             if (request.scroll() != null) {
                 scrollId = TransportSearchHelper.buildScrollId(request.searchType(), queryResults.values());
             }
+            releaseIrrelevantSearchContexts(queryResults, docIdsToLoad);
             searchCache.releaseQueryResults(queryResults);
             searchCache.releaseFetchResults(fetchResults);
             invokeListener(new SearchResponse(internalResponse, scrollId, expectedSuccessfulOps, successulOps.get(), buildShardFailures()));

modules/elasticsearch/src/main/java/org/elasticsearch/action/search/type/TransportSearchTypeAction.java

@@ -279,6 +279,9 @@ protected ShardSearchFailure[] buildShardFailures() {
          */
         protected void releaseIrrelevantSearchContexts(Map<SearchShardTarget, QuerySearchResultProvider> queryResults,
                                                        Map<SearchShardTarget, ExtTIntArrayList> docIdsToLoad) {
+            if (docIdsToLoad == null) {
+                return;
+            }
             for (Map.Entry<SearchShardTarget, QuerySearchResultProvider> entry : queryResults.entrySet()) {
                 if (!docIdsToLoad.containsKey(entry.getKey())) {
                     DiscoveryNode node = nodes.get(entry.getKey().nodeId());

modules/elasticsearch/src/main/java/org/elasticsearch/index/cache/filter/support/AbstractConcurrentMapFilterCache.java

@@ -75,12 +75,16 @@ protected AbstractConcurrentMapFilterCache(Index index, @IndexSettings Settings
 
     private class IndexReaderCleaner implements Runnable {
         @Override public void run() {
+            int totalCount = cache.size();
+            int cleaned = 0;
             for (Iterator<IndexReader> readerIt = cache.keySet().iterator(); readerIt.hasNext();) {
                 IndexReader reader = readerIt.next();
                 if (reader.getRefCount() <= 0) {
                     readerIt.remove();
+                    cleaned++;
                 }
             }
+            logger.trace("Cleaned [{}] out of estimated total [{}]", cleaned, totalCount);
         }
     }
 

modules/elasticsearch/src/main/java/org/elasticsearch/search/SearchService.java

@@ -225,8 +225,8 @@ public QueryFetchSearchResult executeFetchPhase(QuerySearchRequest request) thro
             queryPhase.execute(context);
             shortcutDocIdsToLoad(context);
             fetchPhase.execute(context);
-            if (context.scroll() != null) {
-                activeContexts.put(context.id(), context);
+            if (context.scroll() == null) {
+                freeContext(request.id());
             }
             return new QueryFetchSearchResult(context.queryResult(), context.fetchResult());
         } catch (RuntimeException e) {

modules/elasticsearch/src/main/java/org/elasticsearch/search/action/SearchServiceTransportAction.java

@@ -31,25 +31,45 @@
 import org.elasticsearch.search.query.QuerySearchRequest;
 import org.elasticsearch.search.query.QuerySearchResult;
 import org.elasticsearch.transport.*;
+import org.elasticsearch.util.component.AbstractComponent;
 import org.elasticsearch.util.guice.inject.Inject;
 import org.elasticsearch.util.io.stream.LongStreamable;
 import org.elasticsearch.util.io.stream.VoidStreamable;
+import org.elasticsearch.util.logging.ESLogger;
+import org.elasticsearch.util.settings.Settings;
 
 /**
  * An encapsulation of {@link org.elasticsearch.search.SearchService} operations exposed through
  * transport.
  *
  * @author kimchy (Shay Banon)
  */
-public class SearchServiceTransportAction {
+public class SearchServiceTransportAction extends AbstractComponent {
+
+    static final class FreeContextResponseHandler extends VoidTransportResponseHandler {
+
+        private final ESLogger logger;
+
+        FreeContextResponseHandler(ESLogger logger) {
+            super(false);
+            this.logger = logger;
+        }
+
+        @Override public void handleException(RemoteTransportException exp) {
+            logger.warn("Failed to send release search context", exp);
+        }
+    }
 
     private final TransportService transportService;
 
     private final ClusterService clusterService;
 
     private final SearchService searchService;
 
-    @Inject public SearchServiceTransportAction(TransportService transportService, ClusterService clusterService, SearchService searchService) {
+    private final FreeContextResponseHandler freeContextResponseHandler = new FreeContextResponseHandler(logger);
+
+    @Inject public SearchServiceTransportAction(Settings settings, TransportService transportService, ClusterService clusterService, SearchService searchService) {
+        super(settings);
         this.transportService = transportService;
         this.clusterService = clusterService;
         this.searchService = searchService;
@@ -69,7 +89,7 @@ public void sendFreeContext(DiscoveryNode node, final long contextId) {
         if (clusterService.state().nodes().localNodeId().equals(node.id())) {
             searchService.freeContext(contextId);
         } else {
-            transportService.sendRequest(node, SearchFreeContextTransportHandler.ACTION, new LongStreamable(contextId), VoidTransportResponseHandler.INSTANCE_NOSPAWN);
+            transportService.sendRequest(node, SearchFreeContextTransportHandler.ACTION, new LongStreamable(contextId), freeContextResponseHandler);
         }
     }