-
Notifications
You must be signed in to change notification settings - Fork 3.7k
[improve][misc] Add LongArrayAckSets for allocation-free ack set operations #26128
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
base: master
Are you sure you want to change the base?
Changes from all commits
5951c36
7ec3f9c
3667187
4ad88cd
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -32,7 +32,6 @@ | |
| import it.unimi.dsi.fastutil.objects.ObjectIntPair; | ||
| import java.time.Instant; | ||
| import java.util.ArrayList; | ||
| import java.util.BitSet; | ||
| import java.util.Collections; | ||
| import java.util.List; | ||
| import java.util.Map; | ||
|
|
@@ -73,7 +72,7 @@ | |
| import org.apache.pulsar.common.stats.Rate; | ||
| import org.apache.pulsar.common.util.DateFormatter; | ||
| import org.apache.pulsar.common.util.FutureUtil; | ||
| import org.apache.pulsar.common.util.collections.BitSetRecyclable; | ||
| import org.apache.pulsar.common.util.LongArrayAckSets; | ||
| import org.apache.pulsar.opentelemetry.OpenTelemetryAttributes; | ||
| import org.apache.pulsar.transaction.common.exception.TransactionConflictException; | ||
|
|
||
|
|
@@ -389,7 +388,7 @@ public Future<Void> sendMessages(final List<? extends Entry> entries, | |
| long[] ackSet = batchIndexesAcks == null ? null : batchIndexesAcks.getAckSet(i); | ||
| int remainingUnacked; | ||
| if (ackSet != null) { | ||
| remainingUnacked = BitSet.valueOf(ackSet).cardinality(); | ||
| remainingUnacked = LongArrayAckSets.cardinality(ackSet); | ||
| unackedMessages -= (batchSize - remainingUnacked); | ||
| } else { | ||
| remainingUnacked = batchSize; | ||
|
|
@@ -804,23 +803,15 @@ private long computeAckedCount(MessageIdData msgId, Position position, Consumer | |
| } | ||
| long[] cursorAckSet = getCursorAckSet(position); | ||
| if (cursorAckSet == null) { | ||
| return batchSize - BitSet.valueOf(ackSets).cardinality(); | ||
| return batchSize - LongArrayAckSets.cardinality(ackSets); | ||
| } | ||
| BitSetRecyclable cursorBitSet = BitSetRecyclable.create().resetWords(cursorAckSet); | ||
| int lastCardinality = cursorBitSet.cardinality(); | ||
| BitSetRecyclable givenBitSet = BitSetRecyclable.create().resetWords(ackSets); | ||
| cursorBitSet.and(givenBitSet); | ||
| givenBitSet.recycle(); | ||
| int currentCardinality = cursorBitSet.cardinality(); | ||
| cursorBitSet.recycle(); | ||
| int lastCardinality = LongArrayAckSets.cardinality(cursorAckSet); | ||
| int currentCardinality = LongArrayAckSets.cardinalityOfIntersection(cursorAckSet, ackSets); | ||
|
Comment on lines
+808
to
+809
Contributor
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. This result can be computed with one scan instead of scanning cursorAckSet twice. For every bit, C - (C & A) is equivalent to C & ~A, so: Here, cursor & ~ack directly represents indexes that were unacked in the cursor and become acked by this acknowledgment. For unequal array lengths, words beyond ackSets.length must still contribute their full cursor cardinality, because the missing ack words are treated as zero by the previous BitSet.and implementation. Could this be implemented as a single-pass helper such as cardinalityOfDifference(cursorAckSet, ackSets)? return LongArrayAckSets.cardinalityOfDifference(cursorAckSet, ackSets);
public static int cardinalityOfDifference(long[] ackSet1, long[] ackSet2) {
int sum = 0;
int commonLength = Math.min(ackSet1.length, ackSet2.length);
for (int i = 0; i < commonLength; i++) {
sum += Long.bitCount(ackSet1[i] & ~ackSet2[i]);
}
for (int i = commonLength; i < ackSet1.length; i++) {
sum += Long.bitCount(ackSet1[i]);
}
return sum;
} |
||
| return lastCardinality - currentCardinality; | ||
| } | ||
|
|
||
| private long getAckedCountForTransactionAck(int batchSize, long[] ackSets) { | ||
| BitSetRecyclable bitset = BitSetRecyclable.create().resetWords(ackSets); | ||
| long ackedCount = batchSize - bitset.cardinality(); | ||
| bitset.recycle(); | ||
| return ackedCount; | ||
| return batchSize - LongArrayAckSets.cardinality(ackSets); | ||
| } | ||
|
|
||
| private void checkAckValidationError(CommandAck ack, Position position) { | ||
|
|
||
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,87 @@ | ||
| /* | ||
| * Licensed to the Apache Software Foundation (ASF) under one | ||
| * or more contributor license agreements. See the NOTICE file | ||
| * distributed with this work for additional information | ||
| * regarding copyright ownership. The ASF licenses this file | ||
| * to you under the Apache License, Version 2.0 (the | ||
| * "License"); you may not use this file except in compliance | ||
| * with the License. You may obtain a copy of the License at | ||
| * | ||
| * http://www.apache.org/licenses/LICENSE-2.0 | ||
| * | ||
| * Unless required by applicable law or agreed to in writing, | ||
| * software distributed under the License is distributed on an | ||
| * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY | ||
| * KIND, either express or implied. See the License for the | ||
| * specific language governing permissions and limitations | ||
| * under the License. | ||
| */ | ||
| package org.apache.pulsar.common.util; | ||
|
|
||
| /** | ||
| * Utility methods for operating on ack sets held in the "long array" bit set representation — a long array | ||
| * containing a little-endian representation of all the bits in a bit set, as produced by | ||
| * {@link java.util.BitSet#toLongArray()} and accepted by {@link java.util.BitSet#valueOf(long[])} — without | ||
| * allocating a {@code BitSet} instance. | ||
| */ | ||
|
Comment on lines
+21
to
+26
Member
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. This could be generalized so that it's about operating on "a long array containing a little-endian representation of all the bits in a bit set". There could be javadoc links to |
||
| public class LongArrayAckSets { | ||
|
|
||
| private LongArrayAckSets() { | ||
| } | ||
|
|
||
|
Member
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Add a private constructor so that this class cannot be instantiated
Contributor
Author
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more.
Thanks for review and suggestion, I'd rather use
Contributor
Author
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. All addressed, feel free to take a look. |
||
| /** | ||
| * Returns the number of bits set to {@code true} in the given ack set. | ||
| * | ||
| * @param ackSet a long array containing a little-endian representation of all the bits in a bit set | ||
| * @return the number of bits set to {@code true} | ||
| */ | ||
| public static int cardinality(long[] ackSet) { | ||
| int sum = 0; | ||
| for (long word : ackSet) { | ||
| sum += Long.bitCount(word); | ||
| } | ||
| return sum; | ||
| } | ||
|
|
||
| /** | ||
| * Returns a new ack set whose words are the bitwise AND of the corresponding words of the two inputs. | ||
| * | ||
| * <p>Extra words in the longer array are implicitly ANDed with zero and therefore contribute no set bits. | ||
| * Trailing all-zero words are trimmed from the result, so it is in the same canonical form produced by | ||
| * {@link java.util.BitSet#toLongArray()}. | ||
| * | ||
| * @param ackSet1 a long array containing a little-endian representation of all the bits in a bit set | ||
| * @param ackSet2 a long array containing a little-endian representation of all the bits in a bit set | ||
| * @return a new array representing the intersection of the two ack sets, with trailing zero words trimmed | ||
| */ | ||
| public static long[] intersect(long[] ackSet1, long[] ackSet2) { | ||
| int len = Math.min(ackSet1.length, ackSet2.length); | ||
| while (len > 0 && (ackSet1[len - 1] & ackSet2[len - 1]) == 0) { | ||
| len--; | ||
| } | ||
| long[] result = new long[len]; | ||
| for (int i = 0; i < len; i++) { | ||
| result[i] = ackSet1[i] & ackSet2[i]; | ||
| } | ||
| return result; | ||
| } | ||
|
|
||
| /** | ||
| * Returns the number of bits set to {@code true} after applying a logical <b>AND</b> to the given ack sets. | ||
| * | ||
| * <p>When the arrays differ in length, extra words in the longer array are treated as zero (i.e. the AND | ||
| * result for those positions is zero and contributes nothing to the cardinality). | ||
| * | ||
| * @param ackSet1 a long array containing a little-endian representation of all the bits in a bit set | ||
| * @param ackSet2 a long array containing a little-endian representation of all the bits in a bit set | ||
| * @return the number of bits set to {@code true} in {@code ackSet1 & ackSet2} | ||
| */ | ||
| public static int cardinalityOfIntersection(long[] ackSet1, long[] ackSet2) { | ||
| int sum = 0; | ||
| int len = Math.min(ackSet1.length, ackSet2.length); | ||
| for (int i = 0; i < len; i++) { | ||
| sum += Long.bitCount(ackSet1[i] & ackSet2[i]); | ||
| } | ||
| return sum; | ||
| } | ||
| } | ||
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 call only needs an emptiness check, computing the full cardinality scans and popcounts every word even after a non-zero word has been found. This is on the dispatcher path and the purpose of the PR is to reduce ack-set overhead. Could LongArrayAckSets expose an isEmpty(long[]) helper that returns false at the first non-zero word, and use that here?