You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@lucene.apache.org by dn...@apache.org on 2019/05/23 16:11:32 UTC
[lucene-solr] branch branch_7x updated: LUCENE-8809: Ensure release
segment states
This is an automated email from the ASF dual-hosted git repository.
dnhatn pushed a commit to branch branch_7x
in repository https://gitbox.apache.org/repos/asf/lucene-solr.git
The following commit(s) were added to refs/heads/branch_7x by this push:
new 4d5edca LUCENE-8809: Ensure release segment states
4d5edca is described below
commit 4d5edca3e32a6b892b8af12186a37095c53e2793
Author: Nhat Nguyen <nh...@elastic.co>
AuthorDate: Thu May 23 08:04:35 2019 -0400
LUCENE-8809: Ensure release segment states
If refresh and rollback happen concurrently, then we can leave segment
states unreleased leads to leaking refCount of some SegmentReaders.
---
.../apache/lucene/index/FrozenBufferedUpdates.java | 25 +++++------
.../org/apache/lucene/index/TestIndexWriter.java | 49 ++++++++++++++++++++++
2 files changed, 62 insertions(+), 12 deletions(-)
diff --git a/lucene/core/src/java/org/apache/lucene/index/FrozenBufferedUpdates.java b/lucene/core/src/java/org/apache/lucene/index/FrozenBufferedUpdates.java
index 121d0a4..c7c034c 100644
--- a/lucene/core/src/java/org/apache/lucene/index/FrozenBufferedUpdates.java
+++ b/lucene/core/src/java/org/apache/lucene/index/FrozenBufferedUpdates.java
@@ -19,7 +19,6 @@ package org.apache.lucene.index;
import java.io.Closeable;
import java.io.IOException;
import java.util.ArrayList;
-import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
@@ -345,21 +344,23 @@ final class FrozenBufferedUpdates {
public static BufferedUpdatesStream.ApplyDeletesResult closeSegmentStates(IndexWriter writer, BufferedUpdatesStream.SegmentState[] segStates, boolean success) throws IOException {
List<SegmentCommitInfo> allDeleted = null;
long totDelCount = 0;
- final List<BufferedUpdatesStream.SegmentState> segmentStates = Arrays.asList(segStates);
- for (BufferedUpdatesStream.SegmentState segState : segmentStates) {
- if (success) {
- totDelCount += segState.rld.getDelCount() - segState.startDelCount;
- int fullDelCount = segState.rld.getDelCount();
- assert fullDelCount <= segState.rld.info.info.maxDoc() : fullDelCount + " > " + segState.rld.info.info.maxDoc();
- if (segState.rld.isFullyDeleted() && writer.getConfig().getMergePolicy().keepFullyDeletedSegment(() -> segState.reader) == false) {
- if (allDeleted == null) {
- allDeleted = new ArrayList<>();
+ try {
+ for (BufferedUpdatesStream.SegmentState segState : segStates) {
+ if (success) {
+ totDelCount += segState.rld.getDelCount() - segState.startDelCount;
+ int fullDelCount = segState.rld.getDelCount();
+ assert fullDelCount <= segState.rld.info.info.maxDoc() : fullDelCount + " > " + segState.rld.info.info.maxDoc();
+ if (segState.rld.isFullyDeleted() && writer.getConfig().getMergePolicy().keepFullyDeletedSegment(() -> segState.reader) == false) {
+ if (allDeleted == null) {
+ allDeleted = new ArrayList<>();
+ }
+ allDeleted.add(segState.reader.getOriginalSegmentInfo());
}
- allDeleted.add(segState.reader.getOriginalSegmentInfo());
}
}
+ } finally {
+ IOUtils.close(segStates);
}
- IOUtils.close(segmentStates);
if (writer.infoStream.isEnabled("BD")) {
writer.infoStream.message("BD", "closeSegmentStates: " + totDelCount + " new deleted documents; pool " + writer.getPendingUpdatesCount()+ " packets; bytesUsed=" + writer.getReaderPoolRamBytesUsed());
}
diff --git a/lucene/core/src/test/org/apache/lucene/index/TestIndexWriter.java b/lucene/core/src/test/org/apache/lucene/index/TestIndexWriter.java
index b877824..1beb237 100644
--- a/lucene/core/src/test/org/apache/lucene/index/TestIndexWriter.java
+++ b/lucene/core/src/test/org/apache/lucene/index/TestIndexWriter.java
@@ -39,6 +39,7 @@ import java.util.Map;
import java.util.Random;
import java.util.Set;
import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.Semaphore;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.function.Predicate;
import java.util.stream.Collectors;
@@ -76,6 +77,8 @@ import org.apache.lucene.search.IndexSearcher;
import org.apache.lucene.search.MatchAllDocsQuery;
import org.apache.lucene.search.PhraseQuery;
import org.apache.lucene.search.ScoreDoc;
+import org.apache.lucene.search.SearcherFactory;
+import org.apache.lucene.search.SearcherManager;
import org.apache.lucene.search.TermQuery;
import org.apache.lucene.search.TopDocs;
import org.apache.lucene.store.AlreadyClosedException;
@@ -3714,4 +3717,50 @@ public class TestIndexWriter extends LuceneTestCase {
dir.close();
}
+ public void testRefreshAndRollbackConcurrently() throws Exception {
+ Directory dir = newDirectory();
+ IndexWriter w = new IndexWriter(dir, newIndexWriterConfig());
+ AtomicBoolean stopped = new AtomicBoolean();
+ Semaphore indexedDocs = new Semaphore(0);
+ Thread indexer = new Thread(() -> {
+ while (stopped.get() == false) {
+ try {
+ String id = Integer.toString(random().nextInt(100));
+ Document doc = new Document();
+ doc.add(new StringField("id", id, Field.Store.YES));
+ w.updateDocument(new Term("id", id), doc);
+ indexedDocs.release(1);
+ } catch (IOException e) {
+ throw new AssertionError(e);
+ } catch (AlreadyClosedException ignored) {
+ return;
+ }
+ }
+ });
+
+ SearcherManager sm = new SearcherManager(w, new SearcherFactory());
+ Thread refresher = new Thread(() -> {
+ while (stopped.get() == false) {
+ try {
+ sm.maybeRefreshBlocking();
+ } catch (IOException e) {
+ throw new AssertionError(e);
+ } catch (AlreadyClosedException ignored) {
+ return;
+ }
+ }
+ });
+
+ try {
+ indexer.start();
+ refresher.start();
+ indexedDocs.acquire(1 + random().nextInt(100));
+ w.rollback();
+ } finally {
+ stopped.set(true);
+ indexer.join();
+ refresher.join();
+ IOUtils.close(sm, dir);
+ }
+ }
}