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 |
|---|---|---|
|
|
@@ -329,7 +329,7 @@ public Path location() { | |
| * Returns the generation of the current transaction log. | ||
| */ | ||
| public long currentFileGeneration() { | ||
| try (ReleasableLock ignored = readLock.acquire()) { | ||
| return current.getGeneration(); | ||
| } | ||
| } | ||
|
|
@@ -399,6 +399,8 @@ TranslogWriter createWriter(long fileGeneration) throws IOException { | |
| return newFile; | ||
| } | ||
|
|
||
| final AtomicBoolean foldingGeneration = new AtomicBoolean(); | ||
|
There was a problem hiding this comment. Putting this down here rather than up with the rest of the members will make it much harder to find. Even if this is the one place it is used. |
||
|
|
||
| /** | ||
| * Adds an operation to the transaction log. | ||
| * | ||
|
|
@@ -409,20 +411,31 @@ 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; | ||
| try (ReleasableLock ignored = readLock.acquire()) { | ||
| ensureOpen(); | ||
| location = current.add(bytes, operation.seqNo()); | ||
| } | ||
| try (ReleasableLock ignored = writeLock.acquire()) { | ||
| if (shouldFoldGeneration(this) && foldingGeneration.compareAndSet(false, true)) { | ||
| // we have to check the condition again lest we could fold twice in a race | ||
|
There was a problem hiding this comment. If you already have the writeLock can this race? There was a problem hiding this comment. My fingers did not do what they suppose to do here. 😉 I pushed 152cae7. |
||
| if (shouldFoldGeneration(this)) { | ||
| this.foldGeneration(current.getGeneration()); | ||
| } | ||
| final boolean wasFoldingGeneration = foldingGeneration.getAndSet(false); | ||
|
There was a problem hiding this comment. I would feel better if this was in a finally clause (although if we have any exception here, we will fail the translog anyway). Thinking about this more - shall we move this logic into it's own "foldIfNeeded" method? |
||
| assert wasFoldingGeneration; | ||
| } | ||
| } | ||
| return location; | ||
| } catch (final AlreadyClosedException | IOException ex) { | ||
| try { | ||
| closeOnTragicEvent(ex); | ||
|
|
@@ -442,6 +455,20 @@ public Location add(final Operation operation) throws IOException { | |
| } | ||
| } | ||
|
|
||
| /** | ||
| * Tests whether or not the current generation of the translog should be folded into a new | ||
| * generation. This test is based on the size of the current generation compared to the | ||
| * configured generation threshold size. | ||
| * | ||
| * @param translog the translog | ||
| * @return {@code true} if the current generation should be folded into a new generation | ||
| */ | ||
| private static boolean shouldFoldGeneration(final Translog translog) { | ||
|
There was a problem hiding this comment. I'd wouldn't have declared this as static. I'm really curious why you did. I expect I have something to learn from this.... |
||
| final long size = translog.current.sizeInBytes(); | ||
| final long threshold = translog.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,27 +1349,46 @@ public static void writeOperationNoSize(BufferedChecksumStreamOutput out, Transl | |
| out.writeInt((int) checksum); | ||
| } | ||
|
|
||
| /** | ||
| * Fold 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. | ||
| * | ||
| * @param generation the current translog generation | ||
| * @throws IOException if an I/O exception occurred during any file operations | ||
| */ | ||
| void foldGeneration(final long generation) throws IOException { | ||
|
There was a problem hiding this comment. s/generation/currentGeneration/? |
||
| 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 == generation; | ||
| final Path generationCheckpoint = | ||
| location.resolve(getCommitCheckpointFileName(generation)); | ||
| 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(generation + 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) { | ||
| throw new IllegalStateException("already committing a translog with generation: " + | ||
| currentCommittingGeneration); | ||
| } | ||
| final long generation = current.getGeneration(); | ||
| currentCommittingGeneration = generation; | ||
| foldGeneration(generation); | ||
| } catch (final Exception e) { | ||
|
There was a problem hiding this comment. Is this catch block still needed? |
||
| IOUtils.closeWhileHandlingException(this); // tragic event | ||
| throw e; | ||
| } | ||
|
|
||
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
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?