Skip shard refreshes if shard is search idle

core/src/main/java/org/elasticsearch/action/explain/TransportExplainAction.java

@@ -24,6 +24,8 @@
 import org.elasticsearch.ElasticsearchException;
 import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.action.RoutingMissingException;
+import org.elasticsearch.action.get.GetRequest;
+import org.elasticsearch.action.get.GetResponse;
 import org.elasticsearch.action.support.ActionFilters;
 import org.elasticsearch.action.support.single.shard.TransportSingleShardAction;
 import org.elasticsearch.cluster.ClusterState;
@@ -33,8 +35,10 @@
 import org.elasticsearch.common.inject.Inject;
 import org.elasticsearch.common.lease.Releasables;
 import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.index.IndexService;
 import org.elasticsearch.index.engine.Engine;
 import org.elasticsearch.index.get.GetResult;
+import org.elasticsearch.index.shard.IndexShard;
 import org.elasticsearch.index.shard.ShardId;
 import org.elasticsearch.search.SearchService;
 import org.elasticsearch.search.internal.AliasFilter;
@@ -86,6 +90,19 @@ protected void resolveRequest(ClusterState state, InternalRequest request) {
         }
     }
 
+    @Override
+    protected void shardOperation(ExplainRequest request, ShardId shardId, ActionListener<ExplainResponse> listener) throws IOException {
+        IndexService indexService = searchService.getIndicesService().indexServiceSafe(shardId.getIndex());
+        IndexShard indexShard = indexService.getShard(shardId.id());
+        indexShard.awaitPendingRefresh(b -> {
+            try {
+                super.shardOperation(request, shardId, listener);
+            } catch (IOException ex) {
+                listener.onFailure(ex);
+            }
+        });
+    }
+
     @Override
     protected ExplainResponse shardOperation(ExplainRequest request, ShardId shardId) throws IOException {
         ShardSearchLocalRequest shardSearchLocalRequest = new ShardSearchLocalRequest(shardId,

core/src/main/java/org/elasticsearch/action/get/TransportGetAction.java

@@ -19,9 +19,12 @@
 
 package org.elasticsearch.action.get;
 
+import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.action.RoutingMissingException;
 import org.elasticsearch.action.support.ActionFilters;
 import org.elasticsearch.action.support.single.shard.TransportSingleShardAction;
+import org.elasticsearch.action.termvectors.TermVectorsRequest;
+import org.elasticsearch.action.termvectors.TermVectorsResponse;
 import org.elasticsearch.cluster.ClusterState;
 import org.elasticsearch.cluster.metadata.IndexMetaData;
 import org.elasticsearch.cluster.metadata.IndexNameExpressionResolver;
@@ -38,6 +41,8 @@
 import org.elasticsearch.threadpool.ThreadPool;
 import org.elasticsearch.transport.TransportService;
 
+import java.io.IOException;
+
 /**
  * Performs the get operation.
  */
@@ -76,6 +81,19 @@ protected void resolveRequest(ClusterState state, InternalRequest request) {
         }
     }
 
+    @Override
+    protected void shardOperation(GetRequest request, ShardId shardId, ActionListener<GetResponse> listener) throws IOException {
+        IndexService indexService = indicesService.indexServiceSafe(shardId.getIndex());
+        IndexShard indexShard = indexService.getShard(shardId.id());
+        indexShard.awaitPendingRefresh(b -> {
+            try {
+                super.shardOperation(request, shardId, listener);
+            } catch (IOException ex) {
+                listener.onFailure(ex);
+            }
+        });
+    }
+
     @Override
     protected GetResponse shardOperation(GetRequest request, ShardId shardId) {
         IndexService indexService = indicesService.indexServiceSafe(shardId.getIndex());

core/src/main/java/org/elasticsearch/action/support/single/shard/TransportSingleShardAction.java

@@ -38,6 +38,7 @@
 import org.elasticsearch.common.Nullable;
 import org.elasticsearch.common.logging.LoggerMessageFormat;
 import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.common.util.concurrent.AbstractRunnable;
 import org.elasticsearch.index.shard.ShardId;
 import org.elasticsearch.threadpool.ThreadPool;
 import org.elasticsearch.transport.TransportChannel;
@@ -47,6 +48,8 @@
 import org.elasticsearch.transport.TransportService;
 
 import java.io.IOException;
+import java.io.UncheckedIOException;
+import java.util.concurrent.Executor;
 import java.util.function.Supplier;
 
 import static org.elasticsearch.action.support.TransportActions.isShardNotAvailableException;
@@ -78,7 +81,7 @@ protected TransportSingleShardAction(Settings settings, String actionName, Threa
         if (!isSubAction()) {
             transportService.registerRequestHandler(actionName, request, ThreadPool.Names.SAME, new TransportHandler());
         }
-        transportService.registerRequestHandler(transportShardAction, request, executor, new ShardTransportHandler());
+        transportService.registerRequestHandler(transportShardAction, request, ThreadPool.Names.SAME, new ShardTransportHandler());
     }
 
     /**
@@ -97,6 +100,19 @@ protected void doExecute(Request request, ActionListener<Response> listener) {
 
     protected abstract Response shardOperation(Request request, ShardId shardId) throws IOException;
 
+    protected void shardOperation(Request request, ShardId shardId, ActionListener<Response> listener) throws IOException {
+        threadPool.executor(this.executor).execute(new AbstractRunnable() {
+            @Override
+            public void onFailure(Exception e) {
+                listener.onFailure(e);
+            }
+
+            @Override
+            protected void doRun() throws Exception {
+                listener.onResponse(shardOperation(request, shardId));
+            }
+        });
+    }
     protected abstract Response newResponse();
 
     protected abstract boolean resolveIndex(Request request);
@@ -291,11 +307,27 @@ public void messageReceived(final Request request, final TransportChannel channe
             if (logger.isTraceEnabled()) {
                 logger.trace("executing [{}] on shard [{}]", request, request.internalShardId);
             }
-            Response response = shardOperation(request, request.internalShardId);
-            channel.sendResponse(response);
+            shardOperation(request, request.internalShardId, new ActionListener<Response>() {
+                @Override
+                public void onResponse(Response response) {
+                    try {
+                        channel.sendResponse(response);
+                    } catch (IOException e) {
+                        onFailure(e);
+                    }
+                }
+
+                @Override
+                public void onFailure(Exception e) {
+                    try {
+                        channel.sendResponse(e);
+                    } catch (IOException e1) {
+                        throw new UncheckedIOException(e1);
+                    }
+                }
+            });
         }
     }
-
     /**
      * Internal request class that gets built on each node. Holds the original request plus additional info.
      */

core/src/main/java/org/elasticsearch/action/termvectors/TransportTermVectorsAction.java

@@ -19,6 +19,7 @@
 
 package org.elasticsearch.action.termvectors;
 
+import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.action.RoutingMissingException;
 import org.elasticsearch.action.support.ActionFilters;
 import org.elasticsearch.action.support.single.shard.TransportSingleShardAction;
@@ -37,6 +38,8 @@
 import org.elasticsearch.threadpool.ThreadPool;
 import org.elasticsearch.transport.TransportService;
 
+import java.io.IOException;
+
 /**
  * Performs the get operation.
  */
@@ -82,6 +85,19 @@ protected void resolveRequest(ClusterState state, InternalRequest request) {
         }
     }
 
+    @Override
+    protected void shardOperation(TermVectorsRequest request, ShardId shardId, ActionListener<TermVectorsResponse> listener) throws IOException {
+        IndexService indexService = indicesService.indexServiceSafe(shardId.getIndex());
+        IndexShard indexShard = indexService.getShard(shardId.id());
+        indexShard.awaitPendingRefresh(b -> {
+            try {
+                super.shardOperation(request, shardId, listener);
+            } catch (IOException ex) {
+                listener.onFailure(ex);
+            }
+        });
+    }
+
     @Override
     protected TermVectorsResponse shardOperation(TermVectorsRequest request, ShardId shardId) {
         IndexService indexService = indicesService.indexServiceSafe(shardId.getIndex());

core/src/main/java/org/elasticsearch/common/settings/IndexScopedSettings.java

@@ -36,7 +36,6 @@
 import org.elasticsearch.index.fielddata.IndexFieldDataService;
 import org.elasticsearch.index.mapper.FieldMapper;
 import org.elasticsearch.index.mapper.MapperService;
-import org.elasticsearch.index.seqno.LocalCheckpointTracker;
 import org.elasticsearch.index.similarity.SimilarityService;
 import org.elasticsearch.index.store.FsDirectoryService;
 import org.elasticsearch.index.store.Store;
@@ -135,6 +134,7 @@ public final class IndexScopedSettings extends AbstractScopedSettings {
         IndexSettings.INDEX_TRANSLOG_GENERATION_THRESHOLD_SIZE_SETTING,
         IndexSettings.INDEX_TRANSLOG_RETENTION_AGE_SETTING,
         IndexSettings.INDEX_TRANSLOG_RETENTION_SIZE_SETTING,
+        IndexSettings.INDEX_SEARCH_IDLE_AFTER,
         IndexFieldDataService.INDEX_FIELDDATA_CACHE_KEY,
         FieldMapper.IGNORE_MALFORMED_SETTING,
         FieldMapper.COERCE_SETTING,

core/src/main/java/org/elasticsearch/index/IndexService.java

@@ -689,14 +689,10 @@ private void maybeFSyncTranslogs() {
     private void maybeRefreshEngine() {
         if (indexSettings.getRefreshInterval().millis() > 0) {
             for (IndexShard shard : this.shards.values()) {
-                if (shard.isReadAllowed()) {
-                    try {
-                        if (shard.isRefreshNeeded()) {
-                            shard.refresh("schedule");
-                        }
-                    } catch (IndexShardClosedException | AlreadyClosedException ex) {
-                        // fine - continue;
-                    }
+                try {
+                    shard.scheduledRefresh();
+                } catch (IndexShardClosedException | AlreadyClosedException ex) {
+                    // fine - continue;
                 }
             }
         }

core/src/main/java/org/elasticsearch/index/IndexSettings.java

@@ -62,6 +62,9 @@ public final class IndexSettings {
     public static final Setting<TimeValue> INDEX_TRANSLOG_SYNC_INTERVAL_SETTING =
         Setting.timeSetting("index.translog.sync_interval", TimeValue.timeValueSeconds(5), TimeValue.timeValueMillis(100),
             Property.IndexScope);
+    public static final Setting<TimeValue> INDEX_SEARCH_IDLE_AFTER =
+        Setting.timeSetting("index.search.idle.after", TimeValue.timeValueSeconds(30),
+            TimeValue.timeValueMinutes(0), Property.IndexScope, Property.Dynamic);
     public static final Setting<Translog.Durability> INDEX_TRANSLOG_DURABILITY_SETTING =
         new Setting<>("index.translog.durability", Translog.Durability.REQUEST.name(),
             (value) -> Translog.Durability.valueOf(value.toUpperCase(Locale.ROOT)), Property.Dynamic, Property.IndexScope);
@@ -262,6 +265,8 @@ public final class IndexSettings {
     private volatile int maxNgramDiff;
     private volatile int maxShingleDiff;
     private volatile boolean TTLPurgeDisabled;
+    private volatile TimeValue searchIdleAfter;
+
     /**
      * The maximum number of refresh listeners allows on this shard.
      */
@@ -371,6 +376,7 @@ public IndexSettings(final IndexMetaData indexMetaData, final Settings nodeSetti
         maxSlicesPerScroll = scopedSettings.get(MAX_SLICES_PER_SCROLL);
         this.mergePolicyConfig = new MergePolicyConfig(logger, this);
         this.indexSortConfig = new IndexSortConfig(this);
+        searchIdleAfter = scopedSettings.get(INDEX_SEARCH_IDLE_AFTER);
         singleType = INDEX_MAPPING_SINGLE_TYPE_SETTING.get(indexMetaData.getSettings()); // get this from metadata - it's not registered
         if ((singleType || version.before(Version.V_6_0_0_alpha1)) == false) {
             throw new AssertionError(index.toString()  + "multiple types are only allowed on pre 6.x indices but version is: ["
@@ -411,8 +417,11 @@ public IndexSettings(final IndexMetaData indexMetaData, final Settings nodeSetti
         scopedSettings.addSettingsUpdateConsumer(MAX_REFRESH_LISTENERS_PER_SHARD, this::setMaxRefreshListeners);
         scopedSettings.addSettingsUpdateConsumer(MAX_SLICES_PER_SCROLL, this::setMaxSlicesPerScroll);
         scopedSettings.addSettingsUpdateConsumer(DEFAULT_FIELD_SETTING, this::setDefaultFields);
+        scopedSettings.addSettingsUpdateConsumer(INDEX_SEARCH_IDLE_AFTER, this::setSearchIdleAfter);
     }
 
+    private void setSearchIdleAfter(TimeValue searchIdleAfter) { this.searchIdleAfter = searchIdleAfter; }
+
     private void setTranslogFlushThresholdSize(ByteSizeValue byteSizeValue) {
         this.flushThresholdSize = byteSizeValue;
     }
@@ -752,4 +761,16 @@ public IndexSortConfig getIndexSortConfig() {
     }
 
     public IndexScopedSettings getScopedSettings() { return scopedSettings;}
+
+    /**
+     * Returns true iff the refresh setting exists or in other words is explicitly set.
+     */
+    public boolean isExplicitRefresh() {
+        return INDEX_REFRESH_INTERVAL_SETTING.exists(settings);
+    }
+
+    /**
+     * Returns the time that an index shard becomes search idle unless it's accessed in between
+     */
+    public TimeValue getSearchIdleAfter() { return searchIdleAfter; }
 }

core/src/main/java/org/elasticsearch/index/shard/IndexShard.java

@@ -234,6 +234,9 @@ Runnable getGlobalCheckpointSyncer() {
      */
     private final RefreshListeners refreshListeners;
 
+    private final AtomicLong lastSearcherAccess = new AtomicLong();
+    private final AtomicReference<Translog.Location> pendingRefreshLocation = new AtomicReference<>();
+
     public IndexShard(
             ShardRouting shardRouting,
             IndexSettings indexSettings,
@@ -298,6 +301,7 @@ public IndexShard(
         searcherWrapper = indexSearcherWrapper;
         primaryTerm = indexSettings.getIndexMetaData().primaryTerm(shardId.id());
         refreshListeners = buildRefreshListeners();
+        lastSearcherAccess.set(threadPool.relativeTimeInMillis());
         persistMetadata(path, indexSettings, shardRouting, null, logger);
     }
 
@@ -1120,6 +1124,7 @@ public Engine.Searcher acquireSearcher(String source) {
 
     private Engine.Searcher acquireSearcher(String source, Engine.SearcherScope scope) {
         readAllowed();
+        lastSearcherAccess.lazySet(threadPool.relativeTimeInMillis());
         final Engine engine = getEngine();
         final Engine.Searcher searcher = engine.acquireSearcher(source, scope);
         boolean success = false;
@@ -2418,15 +2423,69 @@ EngineFactory getEngineFactory() {
         return engineFactory;
     }
 
+    /**
+     * Executes a scheduled refresh if necessary
+     */
+    public boolean scheduledRefresh() {
+        if (isReadAllowed() && isRefreshNeeded()) {
+            if (refreshListeners.refreshNeeded() == false // if we have a listener that is waiting for a refresh we need to force it
+                && isSearchIdle() && indexSettings.isExplicitRefresh() == false) {
+                // lets skip this refresh since we are search idle and
+                // don't necessarily need to refresh. the next search execute cause a
+                setRefreshPending();
+                return false;
+            } else {
+                refresh("schedule");
+                return true;
+            }
+        }
+        return false;
+    }
+
     /**
      * Returns <code>true</code> iff one or more changes to the engine are not visible to via the current searcher *or* there are pending
      * refresh listeners.
      * Otherwise <code>false</code>.
      *
      * @throws AlreadyClosedException if the engine or internal indexwriter in the engine is already closed
      */
-    public boolean isRefreshNeeded() {
-        return getEngine().refreshNeeded() || (refreshListeners != null && refreshListeners.refreshNeeded());
+    final boolean isRefreshNeeded() {
+        return refreshListeners.refreshNeeded() // lets check the cheaper one first
+                || getEngine().refreshNeeded();
+    }
+
+    /**
+     * Returns true if this shards is search idle
+     */
+    final boolean isSearchIdle() {
+        return (threadPool.relativeTimeInMillis() - lastSearcherAccess.get()) >= indexSettings.getSearchIdleAfter().getMillis();
+    }
+
+    private void setRefreshPending() {
+        Engine engine = getEngine();
+        if (isSearchIdle()) {
+            acquireSearcher("setRefreshPending").close(); // move the shard into non-search idle
+        }
+        Translog.Location lastWriteLocation = engine.getTranslog().getLastWriteLocation();
+        Translog.Location location;
+        do {
+            location = this.pendingRefreshLocation.get();
+            if (location != null && lastWriteLocation.compareTo(location) <= 0) {
+                break;
+            }
+        } while (pendingRefreshLocation.compareAndSet(location, lastWriteLocation) == false);
+    }
+
+    public void awaitPendingRefresh(Consumer<Boolean> listener) {
+        final Translog.Location location = pendingRefreshLocation.get();
+        if (location != null) {
+            addRefreshListener(location, (b) -> {
+                pendingRefreshLocation.compareAndSet(location, null);
+                listener.accept(true);
+            });
+        } else {
+            listener.accept(false);
+        }
     }
 
     /**