Skip to content

Fix concurrent search and index delete #42621

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 8 commits into from
Jun 6, 2019
Merged

Fix concurrent search and index delete #42621

Show file tree
Hide file tree
Changes from 3 commits
Commits
Show all changes
8 commits
Select commit Hold shift + click to select a range
5f86ea1
Fix concurrent search and index delete
henningandersen May 28, 2019
8edda7b
Removed duplicate catch block.
henningandersen May 29, 2019
ac02fc0
Always call onFree plus test fix
henningandersen Jun 3, 2019
49dfb3c
Close search context if onNewXXX fails.
henningandersen Jun 3, 2019
d7aafcd
Merge remote-tracking branch 'origin/master' into fix_concurrent_sear…
henningandersen Jun 3, 2019
5fe9892
Removed double check for index delete.
henningandersen Jun 4, 2019
b7a24f4
Assert that context is inactive when freeing.
henningandersen Jun 6, 2019
4a48fb8
Merge remote-tracking branch 'origin/master' into fix_concurrent_sear…
henningandersen Jun 6, 2019
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
40 changes: 30 additions & 10 deletions server/src/main/java/org/elasticsearch/search/SearchService.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -548,19 +548,35 @@ final SearchContext createAndPutContext(ShardSearchRequest request) throws IOExc
}

SearchContext context = createContext(request);
onNewContext(context);
boolean success = false;
try {
putContext(context);
if (request.scroll() != null) {
// ensure that if index is deleted concurrently, we free the context immediately, either here or in freeAllContextForIndex
indicesService.indexServiceSafe(request.shardId().getIndex());
Copy link
Contributor

@s1monw s1monw Jun 3, 2019

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I am still not sure why we do this here. Why can't we just let the search go and fail whenever?

Copy link
Contributor Author

@henningandersen henningandersen Jun 3, 2019

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

It might be overkill... The way I understand the code, the primary purpose of afterIndexRemoved is to ensure that we free memory and file system resources as quickly as possible when an index is deleted/closed. Here we try to ensure the same happens also during a race condition. Without this, we theoretically risk holding on to memory/file system resources for a while after the index is deleted/closed. With this, we should free such resources as soon as the index is deleted (and any current search phases have run to completion).

I guess there is a trade-off in that doing this double validation has a small cost and the likelihood of seeing the race condition is very small. I can certainly remove this double validation if you think?

Copy link
Contributor

@s1monw s1monw Jun 4, 2019

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think we should remove the double validation. But the fix you added in this PR is crucial!

Copy link
Contributor Author

@henningandersen henningandersen Jun 4, 2019

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Thanks @s1monw , I have removed the double validation, please have another look at your convenience.

success = true;
return context;
} finally {
if (success == false) {
freeContext(context.id());
}
}
}

private void onNewContext(SearchContext context) {
boolean success = false;
try {
if (context.scrollContext() != null) {
openScrollContexts.incrementAndGet();
context.indexShard().getSearchOperationListener().onNewScrollContext(context);
}
context.indexShard().getSearchOperationListener().onNewContext(context);
success = true;
return context;
} finally {
if (!success) {
freeContext(context.id());
// currently, the concrete listener is CompositeListener, which swallows exceptions, but here we anyway try to do the
// right thing by notifying onFreeXXX in case one of the listeners fails with an exception in the future.
if (success == false) {
onFreeContext(context);
}
}
}
Expand Down Expand Up @@ -648,18 +664,22 @@ private void freeAllContextForIndex(Index index) {
public boolean freeContext(long id) {
try (SearchContext context = removeContext(id)) {
if (context != null) {
assert context.refCount() > 0 : " refCount must be > 0: " + context.refCount();
context.indexShard().getSearchOperationListener().onFreeContext(context);
if (context.scrollContext() != null) {
openScrollContexts.decrementAndGet();
context.indexShard().getSearchOperationListener().onFreeScrollContext(context);
}
onFreeContext(context);
return true;
}
return false;
}
}

private void onFreeContext(SearchContext context) {
Copy link
Contributor

@s1monw s1monw Jun 4, 2019

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

can you add an assertion here that ensures that we removed the context. I think it's trappy that we have an this method and we need to make sure that we never call if if we haven't removed the context. The special case is fina

assert context.refCount() > 0 : " refCount must be > 0: " + context.refCount();
context.indexShard().getSearchOperationListener().onFreeContext(context);
if (context.scrollContext() != null) {
openScrollContexts.decrementAndGet();
context.indexShard().getSearchOperationListener().onFreeScrollContext(context);
}
}

public void freeAllScrollContexts() {
for (SearchContext searchContext : activeContexts.values()) {
if (searchContext.scrollContext() != null) {
Expand Down
18 changes: 16 additions & 2 deletions server/src/test/java/org/elasticsearch/search/SearchServiceTests.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,7 +19,6 @@
package org.elasticsearch.search;

import com.carrotsearch.hppc.IntArrayList;

import org.apache.lucene.search.Query;
import org.apache.lucene.store.AlreadyClosedException;
import org.elasticsearch.ElasticsearchException;
Expand Down Expand Up @@ -51,6 +50,7 @@
import org.elasticsearch.index.query.QueryRewriteContext;
import org.elasticsearch.index.query.QueryShardContext;
import org.elasticsearch.index.query.TermQueryBuilder;
import org.elasticsearch.index.search.stats.SearchStats;
import org.elasticsearch.index.shard.IndexShard;
import org.elasticsearch.index.shard.SearchOperationListener;
import org.elasticsearch.index.shard.ShardId;
Expand Down Expand Up @@ -261,12 +261,16 @@ public void onFailure(Exception e) {
try {
final int rounds = scaledRandomIntBetween(100, 10000);
SearchRequest searchRequest = new SearchRequest().allowPartialSearchResults(true);
SearchRequest scrollSearchRequest = new SearchRequest().allowPartialSearchResults(true)
.scroll(new Scroll(TimeValue.timeValueMinutes(1)));
for (int i = 0; i < rounds; i++) {
try {
try {
PlainActionFuture<SearchPhaseResult> result = new PlainActionFuture<>();
final boolean useScroll = randomBoolean();
service.executeQueryPhase(
new ShardSearchLocalRequest(searchRequest, indexShard.shardId(), 1,
new ShardSearchLocalRequest(useScroll ? scrollSearchRequest : searchRequest,
indexShard.shardId(), 1,
new AliasFilter(null, Strings.EMPTY_ARRAY), 1.0f, -1, null, null),
new SearchTask(123L, "", "", "", null, Collections.emptyMap()), result);
SearchPhaseResult searchPhaseResult = result.get();
Expand All @@ -276,6 +280,9 @@ public void onFailure(Exception e) {
PlainActionFuture<FetchSearchResult> listener = new PlainActionFuture<>();
service.executeFetchPhase(req, new SearchTask(123L, "", "", "", null, Collections.emptyMap()), listener);
listener.get();
if (useScroll) {
service.freeContext(searchPhaseResult.getRequestId());
}
} catch (ExecutionException ex) {
assertThat(ex.getCause(), instanceOf(RuntimeException.class));
throw ((RuntimeException)ex.getCause());
Expand All @@ -293,6 +300,13 @@ public void onFailure(Exception e) {
thread.join();
semaphore.acquire(Integer.MAX_VALUE);
}

assertEquals(0, service.getActiveContexts());

SearchStats.Stats totalStats = indexShard.searchStats().getTotal();
assertEquals(0, totalStats.getQueryCurrent());
assertEquals(0, totalStats.getScrollCurrent());
assertEquals(0, totalStats.getFetchCurrent());
}

public void testTimeout() throws IOException {
Expand Down