Introduce translog generation rolling #23606
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -55,6 +55,7 @@ | |
| import java.nio.file.StandardOpenOption; | ||
| import java.util.ArrayList; | ||
| import java.util.List; | ||
| import java.util.Locale; | ||
| import java.util.Optional; | ||
| import java.util.Set; | ||
| import java.util.concurrent.atomic.AtomicBoolean; | ||
|
|
@@ -127,6 +128,7 @@ public class Translog extends AbstractIndexShardComponent implements IndexShardC | |
| private final TranslogConfig config; | ||
| private final LongSupplier globalCheckpointSupplier; | ||
| private final String translogUUID; | ||
| private final AtomicBoolean rollingGeneration = new AtomicBoolean(); | ||
|
|
||
| /** | ||
| * Creates a new Translog instance. This method will create a new transaction log unless the given {@link TranslogGeneration} is | ||
|
|
@@ -329,7 +331,7 @@ public Path location() { | |
| * Returns the generation of the current transaction log. | ||
| */ | ||
| public long currentFileGeneration() { | ||
| try (ReleasableLock ignored = readLock.acquire()) { | ||
| return current.getGeneration(); | ||
| } | ||
| } | ||
|
|
@@ -409,20 +411,40 @@ TranslogWriter createWriter(long fileGeneration) throws IOException { | |
| public Location add(final Operation operation) throws IOException { | ||
| final ReleasableBytesStreamOutput out = new ReleasableBytesStreamOutput(bigArrays); | ||
| try { | ||
| final long start = out.position(); | ||
| out.skip(Integer.BYTES); | ||
| writeOperationNoSize(new BufferedChecksumStreamOutput(out), operation); | ||
| final long end = out.position(); | ||
| final int operationSize = (int) (end - Integer.BYTES - start); | ||
| out.seek(start); | ||
| out.writeInt(operationSize); | ||
| out.seek(end); | ||
| final ReleasablePagedBytesReference bytes = out.bytes(); | ||
| final Location location; | ||
| final boolean shouldRollGeneration; | ||
| try (ReleasableLock ignored = readLock.acquire()) { | ||
| ensureOpen(); | ||
| location = current.add(bytes, operation.seqNo()); | ||
| // check if we should roll under the read lock | ||
| shouldRollGeneration = | ||
| shouldRollGeneration() && rollingGeneration.compareAndSet(false, true); | ||
| } | ||
| if (shouldRollGeneration) { | ||
| try (ReleasableLock ignored = writeLock.acquire()) { | ||
| /* | ||
| * We have to check the condition again lest we could roll twice if another | ||
| * thread committed the translog (which rolls the generation) between us | ||
| * releasing the read lock and acquiring the write lock. | ||
| */ | ||
| if (shouldRollGeneration()) { | ||
| this.rollGeneration(); | ||
| } | ||
| } finally { | ||
| final boolean wasRolling = rollingGeneration.getAndSet(false); | ||
|
There was a problem hiding this comment. we have a similar logic in |
||
| assert wasRolling; | ||
| } | ||
| } | ||
| return location; | ||
| } catch (final AlreadyClosedException | IOException ex) { | ||
| try { | ||
| closeOnTragicEvent(ex); | ||
|
|
@@ -442,6 +464,19 @@ public Location add(final Operation operation) throws IOException { | |
| } | ||
| } | ||
|
|
||
| /** | ||
| * Tests whether or not the current generation of the translog should be rolled into a new | ||
| * generation. This test is based on the size of the current generation compared to the | ||
| * configured generation threshold size. | ||
| * | ||
| * @return {@code true} if the current generation should be rolled into a new generation | ||
| */ | ||
| private boolean shouldRollGeneration() { | ||
| final long size = this.current.sizeInBytes(); | ||
| final long threshold = this.indexSettings.getGenerationThresholdSize().getBytes(); | ||
| return size > threshold; | ||
| } | ||
|
|
||
| /** | ||
| * The a {@linkplain Location} that will sort after the {@linkplain Location} returned by the last write but before any locations which | ||
| * can be returned by the next write. | ||
|
|
@@ -1322,44 +1357,62 @@ public static void writeOperationNoSize(BufferedChecksumStreamOutput out, Transl | |
| out.writeInt((int) checksum); | ||
| } | ||
|
|
||
| /** | ||
| * Roll the current translog generation into a new generation. This does not commit the | ||
| * translog. The translog write lock must be held by the current thread. | ||
| * | ||
| * @throws IOException if an I/O exception occurred during any file operations | ||
| */ | ||
| void rollGeneration() throws IOException { | ||
| assert writeLock.isHeldByCurrentThread(); | ||
|
There was a problem hiding this comment. maybe allow this assert to give us a message? |
||
| try { | ||
| final TranslogReader reader = current.closeIntoReader(); | ||
| readers.add(reader); | ||
| final Path checkpoint = location.resolve(CHECKPOINT_FILE_NAME); | ||
| assert Checkpoint.read(checkpoint).generation == current.getGeneration(); | ||
| final Path generationCheckpoint = | ||
| location.resolve(getCommitCheckpointFileName(current.getGeneration())); | ||
| Files.copy(checkpoint, generationCheckpoint); | ||
| IOUtils.fsync(generationCheckpoint, false); | ||
| IOUtils.fsync(generationCheckpoint.getParent(), true); | ||
| // create a new translog file; this will sync it and update the checkpoint data; | ||
| current = createWriter(current.getGeneration() + 1); | ||
| logger.trace("current translog set to [{}]", current.getGeneration()); | ||
| } catch (final Exception e) { | ||
| IOUtils.closeWhileHandlingException(this); // tragic event | ||
| throw e; | ||
| } | ||
| } | ||
|
|
||
| @Override | ||
| public long prepareCommit() throws IOException { | ||
| try (ReleasableLock ignored = writeLock.acquire()) { | ||
| ensureOpen(); | ||
| if (currentCommittingGeneration != NOT_SET_GENERATION) { | ||
| final String message = String.format( | ||
| Locale.ROOT, | ||
| "already committing a translog with generation [%d]", | ||
| currentCommittingGeneration); | ||
| throw new IllegalStateException(message); | ||
| } | ||
| currentCommittingGeneration = current.getGeneration(); | ||
| rollGeneration(); | ||
| } | ||
| return 0; | ||
| } | ||
|
|
||
| @Override | ||
| public long commit() throws IOException { | ||
| try (ReleasableLock ignored = writeLock.acquire()) { | ||
| ensureOpen(); | ||
| if (currentCommittingGeneration == NOT_SET_GENERATION) { | ||
| prepareCommit(); | ||
| } | ||
| assert currentCommittingGeneration != NOT_SET_GENERATION; | ||
| assert readers.stream().anyMatch(r -> r.getGeneration() == currentCommittingGeneration) | ||
| : "readers missing committing generation [" + currentCommittingGeneration + "]"; | ||
| // set the last committed generation otherwise old files will not be cleaned up | ||
| lastCommittedTranslogFileGeneration = current.getGeneration(); | ||
|
There was a problem hiding this comment. I think this is wrong - if we end up rolling during a commit we need to set There was a problem hiding this comment. Good catch (adding multiple rolls between pre-commit and commit catches it too)! I pushed 1669224. |
||
| currentCommittingGeneration = NOT_SET_GENERATION; | ||
| trimUnreferencedReaders(); | ||
| } | ||
|
|
||
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
nit: wondering if this should just be
index.translog.generation_sizeThere was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
I picked
index.translog.generation_threshold_sizeto be consistent withindex.translog.flush_threshold_size. Do you still wonder if it should be changed?There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
We can keep as is then. I don't mind much.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
can you maybe document what happens if that size is exceeded or add a link to the explain?
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
I wanna understand why this is
64MBwhy can't we useINDEX_TRANSLOG_FLUSH_THRESHOLD_SIZE_SETTINGor it's default instead rather than an arbitrary value? I would love to get some insight that we also can document? Then there is also the question why we can't simply flush this from the outside viaIndexShard#maybeFlush, that would simplify the translog potentially, one property that I liked about theTranslog#add()operation was that it would never acquire a write lock. We are now entering potentially dangerous territory here, locking can be a beast.There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Jason can say why he chose 64MB - I think any value we choose now will be arbitrary so I was good with it for now and tweak later if needed.
We want a number that's much smaller than the flush size. We're heading towards keeping the generations that are needed to recover all seq# ops after a certain point. That means that we won't always clean previous generation when flushing. If we use the flush size for generations we can end up in a poisonous where we repeatedly try to flush but the translog is not trimmed.
Do you mean doing both flushing and potentially opening a new generation from the same method that is called after indexing? i.e., call it maybeFlushAndRoll ( :) ) ? I would be good with that (though prefer the current approach).
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
I was looking into having a
Translog#rollover()method we can from themaybeFlushwe can renamemaybeFlushtoonAfterOpterationto not necessarily yield impl details. I am a bit concerned about the write lock, which might not be a problem today but maybe tomorrow.makes sense. lets document this. it's not clear from reviewing the PR
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
I pushed dd14ba8 to move the control of rolling to index shard. Would @nik9000, @bleskes, and @s1monw please take another look?