Lazy initialize checkpoint tracker bit sets #27179
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
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -19,13 +19,12 @@ | |
|
|
||
| package org.elasticsearch.index.seqno; | ||
|
|
||
| import com.carrotsearch.hppc.LongObjectHashMap; | ||
| import org.apache.lucene.util.FixedBitSet; | ||
| import org.elasticsearch.common.SuppressForbidden; | ||
| import org.elasticsearch.common.settings.Setting; | ||
| import org.elasticsearch.index.IndexSettings; | ||
|
|
||
| /** | ||
| * This class generates sequences numbers and keeps track of the so-called "local checkpoint" which is the highest number for which all | ||
| * previous sequence numbers have been processed (inclusive). | ||
|
|
@@ -40,21 +39,16 @@ public class LocalCheckpointTracker { | |
| Setting.intSetting("index.seq_no.checkpoint.bit_arrays_size", 1024, 4, Setting.Property.IndexScope); | ||
|
|
||
| /** | ||
| * A collection of bit arrays representing pending sequence numbers. Each sequence number is mapped to a bit array by dividing by the | ||
| * bit set size. | ||
| */ | ||
| final LongObjectHashMap<FixedBitSet> processedSeqNo = new LongObjectHashMap<>(); | ||
|
|
||
| /** | ||
| * The size of each bit set representing processed sequence numbers. | ||
| */ | ||
| private final int bitArraysSize; | ||
|
|
||
| /** | ||
| * The current local checkpoint, i.e., all sequence numbers no more than this number have been completed. | ||
| */ | ||
|
|
@@ -85,7 +79,6 @@ public LocalCheckpointTracker(final IndexSettings indexSettings, final long maxS | |
| "max seq. no. must be non-negative or [" + SequenceNumbers.NO_OPS_PERFORMED + "] but was [" + maxSeqNo + "]"); | ||
| } | ||
| bitArraysSize = SETTINGS_BIT_ARRAYS_SIZE.get(indexSettings.getSettings()); | ||
| nextSeqNo = maxSeqNo == SequenceNumbers.NO_OPS_PERFORMED ? 0 : maxSeqNo + 1; | ||
| checkpoint = localCheckpoint; | ||
| } | ||
|
|
@@ -113,9 +106,9 @@ public synchronized void markSeqNoAsCompleted(final long seqNo) { | |
| // this is possible during recovery where we might replay an operation that was also replicated | ||
| return; | ||
| } | ||
| final FixedBitSet bitArray = getBitArrayForSeqNo(seqNo); | ||
| final int offset = seqNoToBitArrayOffset(seqNo); | ||
| bitArray.set(offset); | ||
| if (seqNo == checkpoint + 1) { | ||
| updateCheckpoint(); | ||
| } | ||
|
|
@@ -130,7 +123,6 @@ synchronized void resetCheckpoint(final long checkpoint) { | |
| assert checkpoint != SequenceNumbers.UNASSIGNED_SEQ_NO; | ||
| assert checkpoint <= this.checkpoint; | ||
| processedSeqNo.clear(); | ||
| this.checkpoint = checkpoint; | ||
| } | ||
|
|
||
|
|
@@ -183,26 +175,30 @@ synchronized void waitForOpsToComplete(final long seqNo) throws InterruptedExcep | |
| @SuppressForbidden(reason = "Object#notifyAll") | ||
| private void updateCheckpoint() { | ||
| assert Thread.holdsLock(this); | ||
| assert getBitArrayForSeqNo(checkpoint + 1).get(seqNoToBitArrayOffset(checkpoint + 1)) : | ||
| "updateCheckpoint is called but the bit following the checkpoint is not set"; | ||
| try { | ||
| // keep it simple for now, get the checkpoint one by one; in the future we can optimize and read words | ||
| long bitArrayKey = getBitArrayKey(checkpoint); | ||
| FixedBitSet current = processedSeqNo.get(bitArrayKey); | ||
| if (current == null) { | ||
|
There was a problem hiding this comment. when can this happen? it seems we only clean bit sets here? There was a problem hiding this comment. When we clean the set (e.g., the checkpoint is equal to |
||
| // the bit set corresponding to the checkpoint has already been removed, set ourselves up for the next bit set | ||
| assert checkpoint % bitArraysSize == bitArraysSize - 1; | ||
| current = processedSeqNo.get(++bitArrayKey); | ||
| } | ||
| do { | ||
| checkpoint++; | ||
| /* | ||
| * The checkpoint always falls in the current bit set or we have already cleaned it; if it falls on the last bit of the | ||
| * current bit set, we can clean it. | ||
| */ | ||
| if (checkpoint == (1 + bitArrayKey) * bitArraysSize - 1) { | ||
|
There was a problem hiding this comment. can we maybe cache |
||
| assert current != null; | ||
| final FixedBitSet removed = processedSeqNo.remove(bitArrayKey); | ||
| assert removed == current; | ||
| current = processedSeqNo.get(++bitArrayKey); | ||
| } | ||
| } while (current != null && current.get(seqNoToBitArrayOffset(checkpoint + 1))); | ||
| } finally { | ||
| // notifies waiters in waitForOpsToComplete | ||
| this.notifyAll(); | ||
|
|
@@ -215,31 +211,30 @@ assert getBitSetForSeqNo(checkpoint + 1).get(seqNoToBitSetOffset(checkpoint + 1) | |
| * @param seqNo the sequence number to obtain the bit array for | ||
| * @return the bit array corresponding to the provided sequence number | ||
| */ | ||
| private long getBitArrayKey(final long seqNo) { | ||
| assert Thread.holdsLock(this); | ||
| return seqNo / bitArraysSize; | ||
| } | ||
|
|
||
| private FixedBitSet getBitArrayForSeqNo(final long seqNo) { | ||
|
There was a problem hiding this comment. maybe use final int slot = processedSeqNo.indexOf(bitArrayKey);
if (processedSeqNo.indexExists(slot) == false) {
processedSeqNo.indexInsert(slot, bitArrayKey, new FixedBitSet(bitArraysSize));
}
return processedSeqNo.indexGet(slot);There was a problem hiding this comment. Thanks @martijnvg, I pushed cabda87. There was a problem hiding this comment. Actually the There was a problem hiding this comment. ah, I didn't realize that. Maybe we should do this to make optimal use of the final int slot = processedSeqNo.indexOf(bitArrayKey);
if (processedSeqNo.indexExists(slot) == false) {
FixedBitSet bitSet = new FixedBitSet(bitArraysSize));
processedSeqNo.indexInsert(slot, bitArrayKey, bitSet);
return bitSet;
} else {
return processedSeqNo.indexGet(slot);
}There was a problem hiding this comment. When you see a chance to avoid I pushed f4f0dae. |
||
| assert Thread.holdsLock(this); | ||
| final long bitArrayKey = getBitArrayKey(seqNo); | ||
| if (processedSeqNo.containsKey(bitArrayKey) == false) { | ||
| processedSeqNo.put(bitArrayKey, new FixedBitSet(bitArraysSize)); | ||
| } | ||
| return processedSeqNo.get(bitArrayKey); | ||
| } | ||
|
|
||
| /** | ||
| * Obtain the position in the bit array corresponding to the provided sequence number. The bit array corresponding to the sequence | ||
| * number can be obtained via {@link #getBitArrayForSeqNo(long)}. | ||
| * | ||
| * @param seqNo the sequence number to obtain the position for | ||
| * @return the position in the bit array corresponding to the provided sequence number | ||
| */ | ||
| private int seqNoToBitArrayOffset(final long seqNo) { | ||
| assert Thread.holdsLock(this); | ||
| return Math.toIntExact(seqNo % bitArraysSize); | ||
| } | ||
|
|
||
| } | ||
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
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.
since this is a
FixedBitSetI wonder why renamed everything to use bitArray? I don't mind. Just curious.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.
Because I did not like the inconsistency that in some places it's referred to as a "bit array", and in some it's referred to as a "bit set". I wanted to make them all "bit set" but the problem is the setting
index.seq_no.checkpoint.bit_arrays_size(currently never released to the world) refers to "bit array". If, and only if, you would be okay with changing the name of this setting toindex.seq_no.checkpoint.bit_sets_size(in 6.0) then I would be okay with normalizing everything to "bit set".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'm fine with renaming the setting. I think I added it just be able to control things from tests and as an extra escape hatch if the default size proved wrong. We might want to just remove it. We can solve the test part differently and I'm less concerned now about the default size (I'm also thinking that 1024 is maybe too small)..
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 opened #27191.