mirror of
https://github.com/OpenHFT/Chronicle-Map.git
synced 2021-09-13 01:26:12 +00:00
Make BiMap implementation dually thread-safe and consistent
This commit is contained in:
@@ -324,12 +324,7 @@ public class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? super
|
||||
|
||||
@Override
|
||||
public void clear() {
|
||||
try (QueryContextInterface<?, ?, ?> cxt = mapContext()) {
|
||||
for (int i = 0; i < actualSegments; i++) {
|
||||
cxt.initTheSegmentIndex(i);
|
||||
cxt.clear();
|
||||
}
|
||||
}
|
||||
forEachEntry(KeyContext::remove);
|
||||
}
|
||||
|
||||
public final long readValueSize(Bytes entry) {
|
||||
|
||||
@@ -4,12 +4,145 @@ import net.openhft.chronicle.hash.Value;
|
||||
import org.jetbrains.annotations.NotNull;
|
||||
import org.junit.Test;
|
||||
|
||||
import java.util.Map;
|
||||
import java.util.Map.Entry;
|
||||
import java.util.concurrent.ExecutionException;
|
||||
import java.util.concurrent.ForkJoinPool;
|
||||
import java.util.concurrent.ThreadLocalRandom;
|
||||
import java.util.function.BiFunction;
|
||||
|
||||
import static net.openhft.chronicle.hash.Value.bytesEquivalent;
|
||||
import static net.openhft.chronicle.map.BiMapTest.DualLockSuccess.FAIL;
|
||||
import static net.openhft.chronicle.map.BiMapTest.DualLockSuccess.SUCCESS;
|
||||
import static org.junit.Assert.assertEquals;
|
||||
import static org.junit.Assert.assertTrue;
|
||||
|
||||
public class BiMapTest {
|
||||
|
||||
static class BiMapEntryOperations<K, V> implements MapEntryOperations<K, V, Void> {
|
||||
enum DualLockSuccess {SUCCESS, FAIL}
|
||||
|
||||
static class BiMapMethods<K, V> implements MapMethods<K, V, DualLockSuccess> {
|
||||
@Override
|
||||
public void remove(MapQueryContext<K, V, DualLockSuccess> q, ReturnValue<V> returnValue) {
|
||||
while (true) {
|
||||
q.updateLock().lock();
|
||||
try {
|
||||
MapEntry<K, V> entry = q.entry();
|
||||
if (entry != null) {
|
||||
returnValue.returnValue(entry.value());
|
||||
if (q.remove(entry) == SUCCESS)
|
||||
return;
|
||||
}
|
||||
} finally {
|
||||
q.readLock().unlock();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
public void put(MapQueryContext<K, V, DualLockSuccess> q, Value<V, ?> value,
|
||||
ReturnValue<V> returnValue) {
|
||||
while (true) {
|
||||
q.updateLock().lock();
|
||||
try {
|
||||
MapEntry<K, V> entry = q.entry();
|
||||
if (entry != null) {
|
||||
throw new IllegalStateException();
|
||||
} else {
|
||||
if (q.insert(q.absentEntry(), value) == SUCCESS)
|
||||
return;
|
||||
}
|
||||
} finally {
|
||||
q.readLock().unlock();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
public void putIfAbsent(MapQueryContext<K, V, DualLockSuccess> q, Value<V, ?> value,
|
||||
ReturnValue<V> returnValue) {
|
||||
while (true) {
|
||||
try {
|
||||
if (q.readLock().tryLock()) {
|
||||
MapEntry<?, V> entry = q.entry();
|
||||
if (entry != null) {
|
||||
returnValue.returnValue(entry.value());
|
||||
return;
|
||||
}
|
||||
// Key is absent
|
||||
q.readLock().unlock();
|
||||
}
|
||||
q.updateLock().lock();
|
||||
MapEntry<?, V> entry = q.entry();
|
||||
if (entry != null) {
|
||||
returnValue.returnValue(entry.value());
|
||||
return;
|
||||
}
|
||||
// Key is absent
|
||||
if (q.insert(q.absentEntry(), value) == SUCCESS)
|
||||
return;
|
||||
} finally {
|
||||
q.readLock().unlock();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
public boolean remove(MapQueryContext<K, V, DualLockSuccess> q, Value<V, ?> value) {
|
||||
while (true) {
|
||||
q.updateLock().lock();
|
||||
MapEntry<K, V> entry = q.entry();
|
||||
try {
|
||||
if (entry != null && bytesEquivalent(entry.value(), value)) {
|
||||
if (q.remove(entry) == SUCCESS) {
|
||||
return true;
|
||||
} else {
|
||||
//noinspection UnnecessaryContinue
|
||||
continue;
|
||||
}
|
||||
} else {
|
||||
return false;
|
||||
}
|
||||
} finally {
|
||||
q.readLock().unlock();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
public void acquireUsing(MapQueryContext<K, V, DualLockSuccess> q,
|
||||
ReturnValue<V> returnValue) {
|
||||
throw new UnsupportedOperationException();
|
||||
}
|
||||
|
||||
@Override
|
||||
public void replace(MapQueryContext<K, V, DualLockSuccess> q, Value<V, ?> value,
|
||||
ReturnValue<V> returnValue) {
|
||||
throw new UnsupportedOperationException();
|
||||
}
|
||||
|
||||
@Override
|
||||
public boolean replace(MapQueryContext<K, V, DualLockSuccess> q, Value<V, ?> oldValue,
|
||||
Value<V, ?> newValue) {
|
||||
throw new UnsupportedOperationException();
|
||||
}
|
||||
|
||||
@Override
|
||||
public void compute(MapQueryContext<K, V, DualLockSuccess> q,
|
||||
BiFunction<? super K, ? super V, ? extends V> remappingFunction,
|
||||
ReturnValue<V> returnValue) {
|
||||
throw new UnsupportedOperationException();
|
||||
}
|
||||
|
||||
@Override
|
||||
public void merge(MapQueryContext<K, V, DualLockSuccess> q, Value<V, ?> value,
|
||||
BiFunction<? super V, ? super V, ? extends V> remappingFunction,
|
||||
ReturnValue<V> returnValue) {
|
||||
throw new UnsupportedOperationException();
|
||||
}
|
||||
}
|
||||
|
||||
static class BiMapEntryOperations<K, V> implements MapEntryOperations<K, V, DualLockSuccess> {
|
||||
ChronicleMap<V, K> reverse;
|
||||
|
||||
public void setReverse(ChronicleMap<V, K> reverse) {
|
||||
@@ -17,14 +150,19 @@ public class BiMapTest {
|
||||
}
|
||||
|
||||
@Override
|
||||
public Void remove(@NotNull MapEntry<K, V> entry) {
|
||||
public DualLockSuccess remove(@NotNull MapEntry<K, V> entry) {
|
||||
try (ExternalMapQueryContext<V, K, ?> rq = reverse.queryContext(entry.value().get())) {
|
||||
rq.updateLock().lock();
|
||||
if (!rq.updateLock().tryLock()) {
|
||||
if (entry.context() instanceof MapQueryContext)
|
||||
return FAIL;
|
||||
throw new IllegalStateException("Concurrent modifications to reverse map " +
|
||||
"during remove during iteration");
|
||||
}
|
||||
MapEntry<V, K> reverseEntry = rq.entry();
|
||||
if (reverseEntry != null) {
|
||||
entry.doRemove();
|
||||
reverseEntry.doRemove();
|
||||
return null;
|
||||
return SUCCESS;
|
||||
} else {
|
||||
throw new IllegalStateException(entry.key() + " maps to " + entry.value() +
|
||||
", but in the reverse map this value is absent");
|
||||
@@ -33,24 +171,27 @@ public class BiMapTest {
|
||||
}
|
||||
|
||||
@Override
|
||||
public Void replaceValue(@NotNull MapEntry<K, V> entry, Value<V, ?> newValue) {
|
||||
public DualLockSuccess replaceValue(@NotNull MapEntry<K, V> entry, Value<V, ?> newValue) {
|
||||
throw new UnsupportedOperationException();
|
||||
}
|
||||
|
||||
@Override
|
||||
public Void insert(@NotNull MapAbsentEntry<K, V> absentEntry, Value<V, ?> value) {
|
||||
public DualLockSuccess insert(@NotNull MapAbsentEntry<K, V> absentEntry,
|
||||
Value<V, ?> value) {
|
||||
try (ExternalMapQueryContext<V, K, ?> rq = reverse.queryContext(value.get())) {
|
||||
rq.updateLock().lock();
|
||||
if (!rq.updateLock().tryLock())
|
||||
return FAIL;
|
||||
MapAbsentEntry<V, K> reverseAbsentEntry = rq.absentEntry();
|
||||
if (reverseAbsentEntry != null) {
|
||||
absentEntry.doInsert(value);
|
||||
reverseAbsentEntry.doInsert(absentEntry.absentKey());
|
||||
return null;
|
||||
return SUCCESS;
|
||||
} else {
|
||||
Value<K, ?> reverseKey = rq.entry().value();
|
||||
if (reverseKey.equals(absentEntry.absentKey())) {
|
||||
// recover
|
||||
return null;
|
||||
absentEntry.doInsert(value);
|
||||
return SUCCESS;
|
||||
}
|
||||
throw new IllegalArgumentException("Try to associate " +
|
||||
absentEntry.absentKey() + " with " + value + ", but in the reverse " +
|
||||
@@ -61,36 +202,40 @@ public class BiMapTest {
|
||||
}
|
||||
|
||||
@Test
|
||||
public void biMapTest() {
|
||||
public void biMapTest() throws InterruptedException, ExecutionException {
|
||||
BiMapEntryOperations<Integer, CharSequence> biMapOps1 = new BiMapEntryOperations<>();
|
||||
ChronicleMap<Integer, CharSequence> map1 = ChronicleMapBuilder
|
||||
.of(Integer.class, CharSequence.class)
|
||||
.entries(100)
|
||||
.actualSegments(1)
|
||||
.averageValueSize(10)
|
||||
.entryOperations(biMapOps1)
|
||||
.mapMethods(new BiMapMethods<>())
|
||||
.create();
|
||||
|
||||
BiMapEntryOperations<CharSequence, Integer> biMapOps2 = new BiMapEntryOperations<>();
|
||||
ChronicleMap<CharSequence, Integer> map2 = ChronicleMapBuilder
|
||||
.of(CharSequence.class, Integer.class)
|
||||
.entries(100)
|
||||
.actualSegments(1)
|
||||
.averageKeySize(10)
|
||||
.entryOperations(biMapOps2)
|
||||
.mapMethods(new BiMapMethods<>())
|
||||
.create();
|
||||
|
||||
biMapOps1.setReverse(map2);
|
||||
biMapOps2.setReverse(map1);
|
||||
|
||||
map1.put(1, "1");
|
||||
assertEquals(1, map2.get("1").intValue());
|
||||
verifyBiMapConsistent(map1, map2);
|
||||
|
||||
map2.remove("1");
|
||||
assertTrue(map2.isEmpty());
|
||||
assertTrue(map1.isEmpty());
|
||||
|
||||
verifyBiMapConsistent(map1, map2);
|
||||
|
||||
map1.put(3, "4");
|
||||
map2.put("5", 6);
|
||||
assertEquals(2, map1.size());
|
||||
verifyBiMapConsistent(map1, map2);
|
||||
|
||||
try (ExternalMapQueryContext<CharSequence, Integer, ?> q = map2.queryContext("4")) {
|
||||
q.updateLock().lock();
|
||||
@@ -112,5 +257,37 @@ public class BiMapTest {
|
||||
}
|
||||
|
||||
map2.put("4", 3); // recover
|
||||
verifyBiMapConsistent(map1, map2);
|
||||
|
||||
map1.clear();
|
||||
verifyBiMapConsistent(map1, map2);
|
||||
|
||||
ForkJoinPool pool = new ForkJoinPool(8);
|
||||
pool.submit(() -> {
|
||||
ThreadLocalRandom.current().ints().limit(10_000).parallel().forEach(i -> {
|
||||
int v = Math.abs(i % 10);
|
||||
if ((i & 1) == 0) {
|
||||
if ((i & 2) == 0) {
|
||||
map1.putIfAbsent(v, "" + v);
|
||||
} else {
|
||||
map1.remove(v, "" + v);
|
||||
}
|
||||
} else {
|
||||
if ((i & 2) == 0) {
|
||||
map2.putIfAbsent("" + v, v);
|
||||
} else {
|
||||
map2.remove("" + v, v);
|
||||
}
|
||||
}
|
||||
});
|
||||
}).get();
|
||||
verifyBiMapConsistent(map1, map2);
|
||||
}
|
||||
|
||||
private static <K, V> void verifyBiMapConsistent(Map<K, V> m1, Map<V, K> m2) {
|
||||
assertEquals(m1.size(), m2.size());
|
||||
for (Entry<K, V> e : m1.entrySet()) {
|
||||
assertEquals(e.getKey(), m2.get(e.getValue()));
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user