HCOLL-155 added two of peters tests and them implemented the code accordingly - the test are net.openhft.chronicle.map.ChronicleMapTest#testOffheapAcquireUsingLocked and net.openhft.chronicle.map.ChronicleMapTest#testOnheapAcquireUsingLocked

This commit is contained in:
Rob Austin
2014-11-02 00:16:02 +00:00
parent 100d839a8b
commit 118ed1676f
9 changed files with 310 additions and 137 deletions
@@ -142,9 +142,9 @@ public interface ChronicleMap<K, V> extends ConcurrentMap<K, V>, ChronicleHash {
// todo document
SegmentLock acquireUsingReadLocked(@NotNull K key, @NotNull V usingValue);
LockedEntry<K,V> acquireUsingReadLocked(@NotNull K key, @NotNull V usingValue);
// todo document
SegmentLock acquireUsingWriteLocked(@NotNull K key, @NotNull V usingValue);
LockedEntry<K,V> acquireUsingWriteLocked(@NotNull K key, @NotNull V usingValue);
}
@@ -0,0 +1,46 @@
/*
* Copyright 2014 Higher Frequency Trading
*
* http://www.higherfrequencytrading.com
*
* Licensed 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 net.openhft.chronicle.map;
/**
* @author Rob Austin.
*/
public interface LockedEntry<K, V> extends AutoCloseable {
/**
* call this to unlock the maps segment
*/
void close();
/**
* Returns the key corresponding to this entry.
*
* @return the key corresponding to this entry
*/
K key();
/**
* Returns the value corresponding to this entry.
*
* @return the value corresponding to this entry
*/
V value();
}
@@ -45,8 +45,7 @@ import static net.openhft.lang.MemoryUnit.*;
import static net.openhft.lang.collection.DirectBitSet.NOT_FOUND;
/**
* <h2>A Replicating Multi Master HashMap</h2>
* <p>Each remote hash map, mirrors its changes over to
* <h2>A Replicating Multi Master HashMap</h2> <p>Each remote hash map, mirrors its changes over to
* another remote hash map, neither hash map is considered the master store of data, each hash map
* uses timestamps to reconcile changes. We refer to an instance of a remote hash-map as a node. A
* node will be connected to any number of other nodes, for the first implementation the maximum
@@ -58,11 +57,9 @@ import static net.openhft.lang.collection.DirectBitSet.NOT_FOUND;
* nature of this multi master implementation, this return value will only be the old value on the
* nodes local data store. In other words the nodes are only concurrent locally. Its worth realising
* that another node performing exactly the same operation may return a different value. However
* reconciliation will ensure the maps themselves become eventually consistent.
* </p>
* <h2>Reconciliation </h2>
* <p>If two ( or more nodes ) were to receive a change to their maps for the
* same key but different values, say by a user of the maps, calling the put(key, value). Then,
* reconciliation will ensure the maps themselves become eventually consistent. </p>
* <h2>Reconciliation </h2> <p>If two ( or more nodes ) were to receive a change to their maps for
* the same key but different values, say by a user of the maps, calling the put(key, value). Then,
* initially each node will update its local store and each local store will hold a different value,
* but the aim of multi master replication is to provide eventual consistency across the nodes. So,
* with multi master when ever a node is changed it will notify the other nodes of its change. We
@@ -71,8 +68,7 @@ import static net.openhft.lang.collection.DirectBitSet.NOT_FOUND;
* and value. Eventual consistency is achieved by looking at the timestamp from the remote node, if
* for a given key, the remote nodes timestamp is newer than the local nodes timestamp, then the
* event from the remote node will be applied to the local node, otherwise the event will be
* ignored. </p>
* <p>However there is an edge case that we have to concern ourselves with, If two
* ignored. </p> <p>However there is an edge case that we have to concern ourselves with, If two
* nodes update their map at the same time with different values, we have to deterministically
* resolve which update wins, because of eventual consistency both nodes should end up locally
* holding the same data. Although it is rare two remote nodes could receive an update to their maps
@@ -81,8 +77,7 @@ import static net.openhft.lang.collection.DirectBitSet.NOT_FOUND;
* newest timestamp should win, but in this example both timestamps are the same, and the decision
* made to one node should be identical to the decision made to the other. We resolve this simple
* dilemma by using a node identifier, each node will have a unique identifier, the update from the
* node with the smallest identifier wins.
* </p>
* node with the smallest identifier wins. </p>
*
* @param <K> the entries key type
* @param <V> the entries value type
@@ -260,19 +255,41 @@ class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, KI>,
return (Segment) segments[segmentNum];
}
@Override
V lookupUsing(K key, V value, boolean create) {
LockedEntry<K, V> acquireUsing(K key, V value, LockType lockTypeType) {
checkKey(key);
ThreadLocalCopies copies = keyInteropProvider.getCopies(null);
KI keyWriter = keyInteropProvider.get(copies, originalKeyInterop);
KI keyInterop = keyInteropProvider.get(copies, originalKeyInterop);
copies = metaKeyInteropProvider.getCopies(copies);
MKI metaKeyWriter =
metaKeyInteropProvider.get(copies, originalMetaKeyInterop, keyWriter, key);
long hash = metaKeyWriter.hash(keyWriter, key);
MKI metaKeyInterop =
metaKeyInteropProvider.get(copies, originalMetaKeyInterop, keyInterop, key);
long hash = metaKeyInterop.hash(keyInterop, key);
int segmentNum = getSegment(hash);
long segmentHash = segmentHash(hash);
return segment(segmentNum).acquire(copies, metaKeyWriter, keyWriter, key, value,
segmentHash, create, create ? timeProvider.currentTimeMillis() : 0L);
Segment segment = segment(segmentNum);
MutableLockedEntry<K, V, MKI, KI> lock = (lockTypeType == LockType.WRITE_LOCK)
? segment.writeLock()
: segment.readLock();
V v = segment.acquireWithoutLock(copies, metaKeyInterop, keyInterop, key, value,
segmentHash, lockTypeType == LockType.WRITE_LOCK,
(lockTypeType == LockType.WRITE_LOCK) ? timeProvider.currentTimeMillis() : 0L);
// we want to call put on heap objects that don't currently exist in the map
if (v == null && !isNativeValueClass && lockTypeType == LockType.WRITE_LOCK) {
lock.copies = copies;
lock.metaKeyInterop = metaKeyInterop;
lock.keyInterop = keyInterop;
lock.segmentHash = segmentHash;
v = value;
}
lock.value(v);
lock.key(key);
return lock;
}
@Override
@@ -611,50 +628,38 @@ class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, KI>,
return result;
}
V acquireWithoutLock(ThreadLocalCopies copies, MKI metaKeyInterop, KI keyInterop, K key, V usingValue,
long hash2, boolean create, long timestamp) {
/**
* @see VanillaChronicleMap.Segment#acquire
*/
V acquire(ThreadLocalCopies copies, MKI metaKeyInterop, KI keyInterop,
K key, V usingValue, long hash2, boolean create, long timestamp) {
if (create)
writeLock();
else
readLock();
try {
long keySize = metaKeyInterop.size(keyInterop, key);
MultiStoreBytes entry = acquireTmpBytes();
long offset = searchKey(keyInterop, metaKeyInterop, key, keySize, hash2, entry,
hashLookupLiveOnly);
if (offset >= 0L) {
return onKeyPresentOnAcquire(copies, key, usingValue, offset, entry);
long keySize = metaKeyInterop.size(keyInterop, key);
MultiStoreBytes entry = acquireTmpBytes();
long offset = searchKey(keyInterop, metaKeyInterop, key, keySize, hash2, entry,
hashLookupLiveOnly);
if (offset >= 0L) {
return onKeyPresentOnAcquire(copies, key, usingValue, offset, entry);
} else {
if (!create)
return null;
if (defaultValueProvider != null) {
V defaultValue = defaultValueProvider.get(key);
copies = valueWriterProvider.getCopies(copies);
VW valueWriter = valueWriterProvider.get(copies, originalValueWriter);
copies = metaValueWriterProvider.getCopies(copies);
MetaBytesWriter<V, VW> metaValueWriter = metaValueWriterProvider.get(
copies, originalMetaValueWriter, valueWriter, defaultValue);
offset = putEntry(metaKeyInterop, keyInterop, key, keySize, hash2,
localIdentifier, timestamp, hashLookupLiveOnly,
metaValueWriter, valueWriter, defaultValue);
} else {
if (!create)
return null;
if (defaultValueProvider != null) {
V defaultValue = defaultValueProvider.get(key);
copies = valueWriterProvider.getCopies(copies);
VW valueWriter = valueWriterProvider.get(copies, originalValueWriter);
copies = metaValueWriterProvider.getCopies(copies);
MetaBytesWriter<V, VW> metaValueWriter = metaValueWriterProvider.get(
copies, originalMetaValueWriter, valueWriter, defaultValue);
offset = putEntry(metaKeyInterop, keyInterop, key, keySize, hash2,
localIdentifier, timestamp, hashLookupLiveOnly,
metaValueWriter, valueWriter, defaultValue);
} else {
offset = putEntry(metaKeyInterop, keyInterop, key, keySize, hash2,
localIdentifier, timestamp, hashLookupLiveOnly,
prepareValueBytesAsWriter, null, key);
}
return onKeyAbsentOnAcquire(copies, key, keySize, usingValue, offset);
offset = putEntry(metaKeyInterop, keyInterop, key, keySize, hash2,
localIdentifier, timestamp, hashLookupLiveOnly,
prepareValueBytesAsWriter, null, key);
}
} finally {
if (create)
writeUnlock();
else
readUnlock();
return onKeyAbsentOnAcquire(copies, key, keySize, usingValue, offset);
}
}
@Override
@@ -869,8 +874,7 @@ class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, KI>,
/**
* Used only with replication, its sometimes possible to receive an old ( or stale update )
* from a remote map. This method is used to determine if we should ignore such updates.
* <p>We can reject put() and removes() when comparing times stamps with remote
* systems
* <p>We can reject put() and removes() when comparing times stamps with remote systems
*
* @param entry the maps entry
* @param timestamp the time the entry was created or updated
@@ -1321,8 +1325,8 @@ class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, KI>,
}
/**
* Always throws {@code NotSerializableException} since instances of this class are not intended to be
* serializable.
* Always throws {@code NotSerializableException} since instances of this class are not
* intended to be serializable.
*/
private void writeObject(ObjectOutputStream out) throws IOException {
throw new NotSerializableException(getClass().getCanonicalName());
@@ -1341,18 +1345,15 @@ class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, KI>,
}
/**
* <p>Once a change occurs to a map, map replication requires that these changes are picked up by
* another thread, this class provides an iterator like interface to poll for such changes.
* </p>
* <p>In most cases the thread that adds data to the node is unlikely to be the same thread that
* replicates the data over to the other nodes, so data will have to be marshaled between the
* main thread storing data to the map, and the thread running the replication.
* </p>
* <p>One way to perform this marshalling, would be to pipe the data into a queue. However, This class
* <p>Once a change occurs to a map, map replication requires that these changes are picked up
* by another thread, this class provides an iterator like interface to poll for such changes.
* </p> <p>In most cases the thread that adds data to the node is unlikely to be the same thread
* that replicates the data over to the other nodes, so data will have to be marshaled between
* the main thread storing data to the map, and the thread running the replication. </p> <p>One
* way to perform this marshalling, would be to pipe the data into a queue. However, This class
* takes another approach. It uses a bit set, and marks bits which correspond to the indexes of
* the entries that have changed. It then provides an iterator like interface to poll for such
* changes.
* </p>
* changes. </p>
*
* @author Rob Austin.
*/
@@ -1,13 +0,0 @@
package net.openhft.chronicle.map;
/**
* @author Rob Austin.
*/
public interface SegmentLock extends AutoCloseable {
/**
* call this to unlock the segment
*/
void close();
}
@@ -400,12 +400,12 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable {
}
@Override
public SegmentLock acquireUsingReadLocked(K key, V usingValue) {
public LockedEntry acquireUsingReadLocked(K key, V usingValue) {
throw new UnsupportedOperationException();
}
@Override
public SegmentLock acquireUsingWriteLocked(K key, V usingValue) {
public LockedEntry acquireUsingWriteLocked(K key, V usingValue) {
throw new UnsupportedOperationException();
}
@@ -91,6 +91,7 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, KI>,
private final ChronicleHashErrorListener errorListener;
private final int segmentHeaderSize;
private final HashSplitting hashSplitting;
final boolean isNativeValueClass;
transient Provider<BytesReader<K>> keyReaderProvider;
transient Provider<KI> keyInteropProvider;
transient Provider<BytesReader<V>> valueReaderProvider;
@@ -115,6 +116,7 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, KI>,
SerializationBuilder<V> valueBuilder = builder.valueBuilder;
vClass = valueBuilder.eClass;
isNativeValueClass = vClass.getName().endsWith("$$Native");
valueSizeMarshaller = valueBuilder.sizeMarshaller();
originalValueReader = valueBuilder.reader();
@@ -331,7 +333,7 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, KI>,
}
@Override
public V putIfAbsent( K key, V value) {
public V putIfAbsent(K key, V value) {
return put0(key, value, false);
}
@@ -353,21 +355,29 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, KI>,
@Override
public V get(Object key) {
return lookupUsing((K) key, null, false);
try (LockedEntry<K, V> kvLockedEntry = acquireUsing((K) key, null,
LockType.READ_LOCK)) {
return kvLockedEntry.value();
}
}
@Override
public V getUsing(K key, V usingValue) {
return lookupUsing(key, usingValue, false);
try (LockedEntry<K, V> kvLockedEntry = acquireUsing(key, usingValue,
LockType.READ_LOCK)) {
return kvLockedEntry.value();
}
}
@Override
public SegmentLock acquireUsingReadLocked(K key, V usingValue) {
public LockedEntry<K, V> acquireUsingReadLocked(K key, V usingValue) {
return this.acquireUsing(key, usingValue, LockType.READ_LOCK);
}
@Override
public SegmentLock acquireUsingWriteLocked(K key, V usingValue) {
public LockedEntry<K, V> acquireUsingWriteLocked(K key, V usingValue) {
return this.acquireUsing(key, usingValue, LockType.WRITE_LOCK);
//todo call put() on close of a on-heap value
@@ -379,13 +389,16 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, KI>,
// todo remove
@Override
public V acquireUsing(K key, V usingValue) {
return lookupUsing(key, usingValue, true);
try (LockedEntry<K, V> kvLockedEntry = acquireUsing(key, usingValue, LockType.WRITE_LOCK)) {
return kvLockedEntry.value();
}
}
private enum LockType {READ_LOCK, WRITE_LOCK}
enum LockType {READ_LOCK, WRITE_LOCK}
SegmentLock acquireUsing(K key, V value, LockType lockTypeType) {
LockedEntry<K, V> acquireUsing(K key, V value, LockType lockTypeType) {
checkKey(key);
ThreadLocalCopies copies = keyInteropProvider.getCopies(null);
KI keyInterop = keyInteropProvider.get(copies, originalKeyInterop);
@@ -397,16 +410,31 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, KI>,
long segmentHash = segmentHash(hash);
Segment segment = segments[segmentNum];
SegmentLock lock = (lockTypeType == LockType.WRITE_LOCK) ? segment.writeLock() : segment.readLock();
MutableLockedEntry<K, V, MKI, KI> lock = (lockTypeType == LockType.WRITE_LOCK)
? segment.writeLock()
: segment.readLock();
V v = segment.acquireWithoutLock(copies, metaKeyInterop, keyInterop, key, value,
segmentHash, lockTypeType == LockType.WRITE_LOCK);
segment.acquireWithoutLock(copies, metaKeyInterop, keyInterop, key, value,
segmentHash, false);
// we want to call put on heap objects that don't currently exist in the map
if (v == null && !isNativeValueClass && lockTypeType == LockType.WRITE_LOCK) {
lock.copies = copies;
lock.metaKeyInterop = metaKeyInterop;
lock.keyInterop = keyInterop;
lock.segmentHash = segmentHash;
v = value;
}
lock.value(v);
lock.key(key);
return lock;
}
V lookupUsing(K key, V value, boolean create) {
/*
checkKey(key);
ThreadLocalCopies copies = keyInteropProvider.getCopies(null);
KI keyInterop = keyInteropProvider.get(copies, originalKeyInterop);
@@ -418,12 +446,12 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, KI>,
long segmentHash = segmentHash(hash);
Segment segment = segments[segmentNum];
try (SegmentLock lock = create ? segment.writeLock() : segment.readLock()) {
try (LockedEntry lock = create ? segment.writeLock() : segment.readLock()) {
return segment.acquireWithoutLock(copies, metaKeyInterop, keyInterop, key, value,
segmentHash, create);
}
}*/
}
// }
@Override
public boolean containsKey(final Object k) {
@@ -664,20 +692,42 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, KI>,
private MultiMap hashLookup;
private long nextPosToSearchFrom = 0L;
final SegmentLock readUnlock = new SegmentLock() {
final MutableLockedEntry<K, V, MKI, KI> readUnlock = new MutableLockedEntry<K, V, MKI, KI>() {
@Override
public void close() {
Segment.this.readUnlock();
}
};
final SegmentLock writeUnlock = new SegmentLock() {
/**
* close for native values does not call put on close
*/
final MutableLockedEntry<K, V, MKI, KI> nativeWriteUnlock = new MutableLockedEntry<K, V,
MKI, KI>() {
@Override
public void close() {
Segment.this.writeUnlock();
}
};
/**
* close for heap values calls put on close
*/
final MutableLockedEntry<K, V, MKI, KI>
heapWriteUnlock = new MutableLockedEntry<K, V, MKI, KI>() {
@Override
public void close() {
if (copies != null)
putWithoutLock(copies, metaKeyInterop, keyInterop, key(), value(),
segmentHash, true);
copies = null;
Segment.this.writeUnlock();
}
};
/**
* @param index the index of this segment held by the map
*/
@@ -759,7 +809,7 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, KI>,
}
public SegmentLock readLock() throws IllegalStateException {
public MutableLockedEntry<K, V, MKI, KI> readLock() throws IllegalStateException {
while (true) {
final boolean success = segmentHeader.tryRWReadLock(LOCK_OFFSET, lockTimeOutNS);
if (success) return readUnlock;
@@ -772,10 +822,11 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, KI>,
}
}
public SegmentLock writeLock() throws IllegalStateException {
public MutableLockedEntry<K, V, MKI, KI> writeLock() throws IllegalStateException {
while (true) {
final boolean success = segmentHeader.tryRWWriteLock(LOCK_OFFSET, lockTimeOutNS);
if (success) return writeUnlock;
if (success) return isNativeValueClass ? nativeWriteUnlock : heapWriteUnlock;
if (currentThread().isInterrupted()) {
throw new IllegalStateException(new InterruptedException("Unable to obtain lock, interrupted"));
} else {
@@ -926,32 +977,36 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, KI>,
long hash2, boolean replaceIfPresent) {
writeLock();
try {
long keySize = metaKeyInterop.size(keyInterop, key);
hashLookup.startSearch(hash2);
for (long pos; (pos = hashLookup.nextPos()) >= 0L; ) {
long offset = offsetFromPos(pos);
MultiStoreBytes entry = entry(offset);
if (!keyEquals(keyInterop, metaKeyInterop, key, keySize, entry))
continue;
// key is found
entry.skip(keySize);
if (replaceIfPresent) {
return replaceValueOnPut(copies, key, value, entry, pos, offset,
!putReturnsNull, hashLookup);
} else {
return putReturnsNull ? null : readValue(copies, entry, null);
}
}
// key is not found
long offset = putEntry(copies, metaKeyInterop, keyInterop, key, keySize, value);
incrementSize();
notifyPut(offset, true, key, value, posFromOffset(offset));
return null;
return putWithoutLock(copies, metaKeyInterop, keyInterop, key, value, hash2, replaceIfPresent);
} finally {
writeUnlock();
}
}
private V putWithoutLock(ThreadLocalCopies copies, MKI metaKeyInterop, KI keyInterop, K key, V value, long hash2, boolean replaceIfPresent) {
long keySize = metaKeyInterop.size(keyInterop, key);
hashLookup.startSearch(hash2);
for (long pos; (pos = hashLookup.nextPos()) >= 0L; ) {
long offset = offsetFromPos(pos);
MultiStoreBytes entry = entry(offset);
if (!keyEquals(keyInterop, metaKeyInterop, key, keySize, entry))
continue;
// key is found
entry.skip(keySize);
if (replaceIfPresent) {
return replaceValueOnPut(copies, key, value, entry, pos, offset,
!putReturnsNull, hashLookup);
} else {
return putReturnsNull ? null : readValue(copies, entry, null);
}
}
// key is not found
long offset = putEntry(copies, metaKeyInterop, keyInterop, key, keySize, value);
incrementSize();
notifyPut(offset, true, key, value, posFromOffset(offset));
return null;
}
V replaceValueOnPut(ThreadLocalCopies copies, K key, V value, MultiStoreBytes entry, long pos,
long offset, boolean readPrevValue, MultiMap searchedHashLookup) {
long valueSizePos = entry.position();
@@ -1561,4 +1616,32 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, KI>,
return super.setValue(value);
}
}
static abstract class MutableLockedEntry<K, V, MKI, KI> implements LockedEntry<K, V> {
ThreadLocalCopies copies;
MKI metaKeyInterop;
KI keyInterop;
long segmentHash;
private K key;
private V value;
public K key() {
return key;
}
public void key(K key) {
this.key = key;
}
public V value() {
return value;
}
public void value(V value) {
this.value = value;
}
}
}