Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -21,6 +21,7 @@
import org.apache.bookkeeper.mledger.Position;
import org.apache.bookkeeper.mledger.impl.AckSetState;
import org.apache.bookkeeper.mledger.impl.AckSetStateUtil;
import org.apache.pulsar.common.util.LongArrayAckSets;
import org.apache.pulsar.common.util.collections.BitSetRecyclable;

public class PositionAckSetUtil {
Expand Down Expand Up @@ -54,20 +55,11 @@ public static void andAckSet(Position currentPosition, Position otherPosition) {

//This method is do `and` operation for ack set
public static long[] andAckSet(long[] firstAckSet, long[] secondAckSet) {
BitSetRecyclable thisAckSet = BitSetRecyclable.valueOf(firstAckSet);
BitSetRecyclable otherAckSet = BitSetRecyclable.valueOf(secondAckSet);
thisAckSet.and(otherAckSet);
long[] ackSet = thisAckSet.toLongArray();
thisAckSet.recycle();
otherAckSet.recycle();
return ackSet;
return LongArrayAckSets.intersect(firstAckSet, secondAckSet);
}

public static boolean isAckSetEmpty(long[] ackSet) {
BitSetRecyclable bitSet = BitSetRecyclable.create().resetWords(ackSet);
boolean isEmpty = bitSet.isEmpty();
bitSet.recycle();
return isEmpty;
return LongArrayAckSets.cardinality(ackSet) == 0;

Copy link
Copy Markdown
Contributor

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?

}

//This method is compare two position which position is bigger than another one.
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -49,6 +49,13 @@ public void isAckSetRepeatedTest() {
assertFalse(isAckSetOverlap(thisBitSet.toLongArray(), otherBitSet.toLongArray()));
}

@Test
public void andAckSetTrimsTrailingZeroWordsTest() {
// High words AND to zero → result must be trimmed like BitSet#toLongArray would
assertEquals(andAckSet(new long[]{-1L, 0b01L}, new long[]{-1L, 0b10L}), new long[]{-1L});
assertEquals(andAckSet(new long[]{0b01L}, new long[]{0b10L}), new long[0]);
}

@Test
public void compareToWithAckSetForCumulativeAckTest() {
Position positionOne = PositionFactory.create(1, 1);
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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;

Expand Down Expand Up @@ -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;
Expand Down Expand Up @@ -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

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The 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:

cardinality(cursor) - cardinality(cursor & ack)
= cardinality(cursor & ~ack)

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) {
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -20,8 +20,8 @@


import io.netty.util.Recycler;
import java.util.BitSet;
import org.apache.commons.lang3.tuple.Pair;
import org.apache.pulsar.common.util.LongArrayAckSets;

@SuppressWarnings("unchecked")
public class EntryBatchIndexesAcks {
Expand All @@ -42,7 +42,7 @@ public int getTotalAckedIndexCount() {
for (int i = 0; i < size; i++) {
Pair<Integer, long[]> pair = indexesAcks[i];
if (pair != null) {
count += pair.getLeft() - BitSet.valueOf(pair.getRight()).cardinality();
count += pair.getLeft() - LongArrayAckSets.cardinality(pair.getRight());
}
}
return count;
Expand Down
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

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The 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 java.util.BitSet#toLongArray and java.util.BitSet#valueOf(long[]) as a reference.

public class LongArrayAckSets {

private LongArrayAckSets() {
}

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Add a private constructor so that this class cannot be instantiated

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This class could be called LongArrayBitSets. In some code locations, the concept for a BitSet in long[] word representation is called "long array". I guess this concept comes from BitSet.toLongArray method.

Thanks for review and suggestion, I'd rather use LongArrayAckSets here, because AckSet is a very special usage of BitSet in pulsar, WDYT?

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The 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;
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -872,8 +872,9 @@ public boolean intersects(BitSetRecyclable set) {
*/
public int cardinality() {
int sum = 0;
for (int i = 0; i < wordsInUse; i++)
for (int i = 0; i < wordsInUse; i++) {
sum += Long.bitCount(words[i]);
}
return sum;
}

Expand Down
Loading