honeymoose
/
OpenSearch
mirror of https://github.com/honeymoose/OpenSearch.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Acquire ReadLock before reading SegmentInfos in RobinEngine 

Before reading segmentinfos we should aquire the read lock to prevent
reading the segment infos on an already closed robin engine.
This commit is contained in:
Simon Willnauer 2013-09-24 13:56:08 +02:00
parent 5b1f263569
commit df0112358d
1 changed files with 27 additions and 31 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

58
src/main/java/org/elasticsearch/index/engine/robin/RobinEngine.java
View File

@ -278,9 +278,7 @@ public class RobinEngine extends AbstractIndexShardComponent implements Engine {
}
translog.newTranslog(translogIdGenerator.get());
this.searcherManager = buildSearchManager(indexWriter);
SegmentInfos infos = new SegmentInfos();
infos.read(store.directory());
lastCommittedSegmentInfos = infos;
readLastCommittedSegmentsInfo();
} catch (IOException e) {
try {
indexWriter.rollback();
@ -296,6 +294,12 @@ public class RobinEngine extends AbstractIndexShardComponent implements Engine {
}
}
private void readLastCommittedSegmentsInfo() throws IOException {
SegmentInfos infos = new SegmentInfos();
infos.read(store.directory());
lastCommittedSegmentInfos = infos;
}
@Override
public TimeValue defaultRefreshInterval() {
return new TimeValue(1, TimeUnit.SECONDS);
@ -750,9 +754,7 @@ public class RobinEngine extends AbstractIndexShardComponent implements Engine {
@Override
public void flush(Flush flush) throws EngineException {
if (indexWriter == null) {
throw new EngineClosedException(shardId, failedEngine);
}
ensureOpen();
if (flush.type() == Flush.Type.NEW_WRITER || flush.type() == Flush.Type.COMMIT_TRANSLOG) {
// check outside the lock as well so we can check without blocking on the write lock
if (onGoingRecoveries.get() > 0) {
@ -770,9 +772,7 @@ public class RobinEngine extends AbstractIndexShardComponent implements Engine {
if (flush.type() == Flush.Type.NEW_WRITER) {
rwl.writeLock().lock();
try {
if (indexWriter == null) {
throw new EngineClosedException(shardId, failedEngine);
}
ensureOpen();
if (onGoingRecoveries.get() > 0) {
throw new FlushNotAllowedEngineException(shardId, "Recovery is in progress, flush is not allowed");
}
@ -820,9 +820,7 @@ public class RobinEngine extends AbstractIndexShardComponent implements Engine {
} else if (flush.type() == Flush.Type.COMMIT_TRANSLOG) {
rwl.readLock().lock();
try {
if (indexWriter == null) {
throw new EngineClosedException(shardId, failedEngine);
}
ensureOpen();
if (onGoingRecoveries.get() > 0) {
throw new FlushNotAllowedEngineException(shardId, "Recovery is in progress, flush is not allowed");
}
@ -862,9 +860,7 @@ public class RobinEngine extends AbstractIndexShardComponent implements Engine {
// of that translog
rwl.readLock().lock();
try {
if (indexWriter == null) {
throw new EngineClosedException(shardId, failedEngine);
}
ensureOpen();
// we allow to *just* commit if there is an ongoing recovery happening...
// its ok to use this, only a flush will cause a new translogId, and we are locked here from
// other flushes use flushLock
@ -892,21 +888,30 @@ public class RobinEngine extends AbstractIndexShardComponent implements Engine {
}
// reread the last committed segment infos
rwl.readLock().lock();
try {
SegmentInfos infos = new SegmentInfos();
infos.read(store.directory());
lastCommittedSegmentInfos = infos;
ensureOpen();
readLastCommittedSegmentsInfo();
} catch (Throwable e) {
if (!closed) {
logger.warn("failed to read latest segment infos on flush", e);
}
} finally {
rwl.readLock().unlock();
}
} finally {
flushLock.unlock();
flushing.decrementAndGet();
}
}
private void ensureOpen() {
if (indexWriter == null) {
throw new EngineClosedException(shardId, failedEngine);
}
}
private void refreshVersioningTable(long time) {
// we need to refresh in order to clear older version values
refresh(new Refresh("version_table").force(true));
@ -939,9 +944,7 @@ public class RobinEngine extends AbstractIndexShardComponent implements Engine {
possibleMergeNeeded = false;
rwl.readLock().lock();
try {
if (indexWriter == null) {
throw new EngineClosedException(shardId, failedEngine);
}
ensureOpen();
indexWriter.maybeMerge();
} catch (OutOfMemoryError e) {
failEngine(e);
@ -966,9 +969,7 @@ public class RobinEngine extends AbstractIndexShardComponent implements Engine {
if (optimizeMutex.compareAndSet(false, true)) {
rwl.readLock().lock();
try {
if (indexWriter == null) {
throw new EngineClosedException(shardId, failedEngine);
}
ensureOpen();
if (optimize.onlyExpungeDeletes()) {
indexWriter.forceMergeDeletes(false);
} else if (optimize.maxNumSegments() <= 0) {
@ -1031,9 +1032,7 @@ public class RobinEngine extends AbstractIndexShardComponent implements Engine {
rwl.readLock().lock();
try {
flush(new Flush().type(Flush.Type.COMMIT).waitIfOngoing(true));
if (indexWriter == null) {
throw new EngineClosedException(shardId, failedEngine);
}
ensureOpen();
return deletionPolicy.snapshot();
} catch (IOException e) {
throw new SnapshotFailedEngineException(shardId, e);
@ -1121,10 +1120,7 @@ public class RobinEngine extends AbstractIndexShardComponent implements Engine {
public List<Segment> segments() {
rwl.readLock().lock();
try {
IndexWriter indexWriter = this.indexWriter;
if (indexWriter == null) {
throw new EngineClosedException(shardId, failedEngine);
}
ensureOpen();
Map<String, Segment> segments = new HashMap<String, Segment>();
// first, go over and compute the search ones...