Skip to content
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.

Sign up for GitHub

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

Jump to bottom

[fix][misc] Make ConcurrentBitSet thread safe #22361

Merged
merged 4 commits into from
Mar 27, 2024
Merged

[fix][misc] Make ConcurrentBitSet thread safe #22361

Changes from 1 commit
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
84eb13d
[fix][misc] Make ConcurrentBitSet thread safe
lhotari Mar 26, 2024
bd1d1f1
Fix deadlock
lhotari Mar 26, 2024
48711e5
next* and previous* methods are read-only operations
lhotari Mar 27, 2024
0cf6dc5
Merge remote-tracking branch 'origin/master' into lh-improve-concurre…
lhotari Mar 27, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Jump to
Jump to file
Failed to load files.
Diff view
Diff view
285 changes: 260 additions & 25 deletions pulsar-common/src/main/java/org/apache/pulsar/common/util/collections/ConcurrentBitSet.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,6 +20,7 @@

import java.util.BitSet;
import java.util.concurrent.locks.StampedLock;
import java.util.stream.IntStream;
import lombok.EqualsAndHashCode;

/**
Expand Down Expand Up @@ -65,105 +66,339 @@ public boolean get(int bitIndex) {

@Override
public void set(int bitIndex) {
long stamp = rwLock.writeLock();
try {
super.set(bitIndex);
} finally {
rwLock.unlockWrite(stamp);
}
}

@Override
public void clear(int bitIndex) {
long stamp = rwLock.writeLock();
try {
super.clear(bitIndex);
} finally {
rwLock.unlockWrite(stamp);
}
}

@Override
public void set(int fromIndex, int toIndex) {
long stamp = rwLock.writeLock();
try {
super.set(fromIndex, toIndex);
} finally {
rwLock.unlockWrite(stamp);
}
}

@Override
public void clear(int fromIndex, int toIndex) {
long stamp = rwLock.writeLock();
try {
super.clear(fromIndex, toIndex);
} finally {
rwLock.unlockWrite(stamp);
}
}

@Override
public void clear() {
long stamp = rwLock.writeLock();
try {
super.clear();
} finally {
rwLock.unlockWrite(stamp);
}
}

@Override
public int nextSetBit(int fromIndex) {
long stamp = rwLock.writeLock();
try {
return super.nextSetBit(fromIndex);
} finally {
rwLock.unlockWrite(stamp);
}
}

@Override
public int nextClearBit(int fromIndex) {
long stamp = rwLock.writeLock();
try {
return super.nextClearBit(fromIndex);
} finally {
rwLock.unlockWrite(stamp);
}
}

@Override
public int previousSetBit(int fromIndex) {
long stamp = rwLock.writeLock();
try {
return super.previousSetBit(fromIndex);
} finally {
rwLock.unlockWrite(stamp);
}
}

@Override
public int previousClearBit(int fromIndex) {
long stamp = rwLock.writeLock();
try {
return super.previousClearBit(fromIndex);
} finally {
rwLock.unlockWrite(stamp);
}
}

@Override
public boolean isEmpty() {
long stamp = rwLock.tryOptimisticRead();
super.set(bitIndex);
boolean isEmpty = super.isEmpty();
if (!rwLock.validate(stamp)) {
// Fallback to read lock
stamp = rwLock.readLock();
try {
super.set(bitIndex);
isEmpty = super.isEmpty();
} finally {
rwLock.unlockRead(stamp);
}
}
return isEmpty;
}

@Override
public void set(int fromIndex, int toIndex) {
public int cardinality() {
long stamp = rwLock.tryOptimisticRead();
super.set(fromIndex, toIndex);
int cardinality = super.cardinality();
if (!rwLock.validate(stamp)) {
// Fallback to read lock
stamp = rwLock.readLock();
try {
super.set(fromIndex, toIndex);
cardinality = super.cardinality();
} finally {
rwLock.unlockRead(stamp);
}
}
return cardinality;
}

@Override
public int nextSetBit(int fromIndex) {
public int size() {
long stamp = rwLock.tryOptimisticRead();
int bit = super.nextSetBit(fromIndex);
int size = super.size();
if (!rwLock.validate(stamp)) {
// Fallback to read lock
stamp = rwLock.readLock();
try {
bit = super.nextSetBit(fromIndex);
size = super.size();
} finally {
rwLock.unlockRead(stamp);
}
}
return bit;
return size;
}

@Override
public int nextClearBit(int fromIndex) {
public byte[] toByteArray() {
long stamp = rwLock.tryOptimisticRead();
int bit = super.nextClearBit(fromIndex);
byte[] byteArray = super.toByteArray();
if (!rwLock.validate(stamp)) {
// Fallback to read lock
stamp = rwLock.readLock();
try {
bit = super.nextClearBit(fromIndex);
byteArray = super.toByteArray();
} finally {
rwLock.unlockRead(stamp);
}
}
return bit;
return byteArray;
}

@Override
public int previousSetBit(int fromIndex) {
public long[] toLongArray() {
long stamp = rwLock.tryOptimisticRead();
int bit = super.previousSetBit(fromIndex);
long[] longArray = super.toLongArray();
if (!rwLock.validate(stamp)) {
// Fallback to read lock
stamp = rwLock.readLock();
try {
bit = super.previousSetBit(fromIndex);
longArray = super.toLongArray();
} finally {
rwLock.unlockRead(stamp);
}
}
return bit;
return longArray;
}

@Override
public int previousClearBit(int fromIndex) {
public void flip(int bitIndex) {
long stamp = rwLock.writeLock();
try {
super.flip(bitIndex);
} finally {
rwLock.unlockWrite(stamp);
}
}

@Override
public void flip(int fromIndex, int toIndex) {
long stamp = rwLock.writeLock();
try {
super.flip(fromIndex, toIndex);
} finally {
rwLock.unlockWrite(stamp);
}
}

@Override
public void set(int bitIndex, boolean value) {
long stamp = rwLock.writeLock();
try {
super.set(bitIndex, value);
} finally {
rwLock.unlockWrite(stamp);
}
}

@Override
public void set(int fromIndex, int toIndex, boolean value) {
long stamp = rwLock.writeLock();
try {
super.set(fromIndex, toIndex, value);
} finally {
rwLock.unlockWrite(stamp);
}
}

@Override
public BitSet get(int fromIndex, int toIndex) {
long stamp = rwLock.tryOptimisticRead();
int bit = super.previousClearBit(fromIndex);
BitSet bitSet = super.get(fromIndex, toIndex);
if (!rwLock.validate(stamp)) {
// Fallback to read lock
stamp = rwLock.readLock();
try {
bit = super.previousClearBit(fromIndex);
bitSet = super.get(fromIndex, toIndex);
} finally {
rwLock.unlockRead(stamp);
}
}
return bit;
return bitSet;
}

@Override
public boolean isEmpty() {
public int length() {
long stamp = rwLock.tryOptimisticRead();
boolean isEmpty = super.isEmpty();
int length = super.length();
if (!rwLock.validate(stamp)) {
// Fallback to read lock
stamp = rwLock.readLock();
try {
isEmpty = super.isEmpty();
length = super.length();
} finally {
rwLock.unlockRead(stamp);
}
}
return isEmpty;
return length;
}

@Override
public boolean intersects(BitSet set) {
long stamp = rwLock.writeLock();
try {
return super.intersects(set);
} finally {
rwLock.unlockWrite(stamp);
}
}

@Override
public void and(BitSet set) {
long stamp = rwLock.writeLock();
try {
super.and(set);
} finally {
rwLock.unlockWrite(stamp);
}
}

@Override
public void or(BitSet set) {
long stamp = rwLock.writeLock();
try {
super.or(set);
} finally {
rwLock.unlockWrite(stamp);
}
}

@Override
public void xor(BitSet set) {
long stamp = rwLock.writeLock();
try {
super.xor(set);
} finally {
rwLock.unlockWrite(stamp);
}
}

@Override
public void andNot(BitSet set) {
long stamp = rwLock.writeLock();
try {
super.andNot(set);
} finally {
rwLock.unlockWrite(stamp);
}
}

/**
* Returns the clone of the internal wrapped {@code BitSet}.
* This won't be a clone of the {@code ConcurrentBitSet} object.
*
* @return a clone of the internal wrapped {@code BitSet}
*/
@Override
public Object clone() {
long stamp = rwLock.tryOptimisticRead();
BitSet clonedBitSet = (BitSet) super.clone();
if (!rwLock.validate(stamp)) {
// Fallback to read lock
stamp = rwLock.readLock();
try {
clonedBitSet = (BitSet) super.clone();
} finally {
rwLock.unlockRead(stamp);
}
}
return clonedBitSet;
}

@Override
public String toString() {
long stamp = rwLock.tryOptimisticRead();
String str = super.toString();
if (!rwLock.validate(stamp)) {
// Fallback to read lock
stamp = rwLock.readLock();
try {
str = super.toString();
} finally {
rwLock.unlockRead(stamp);
}
}
return str;
}

/**
* This operation is not supported on {@code ConcurrentBitSet}.
*/
@Override
public IntStream stream() {
throw new UnsupportedOperationException("stream is not supported");
}
}