From 118ed1676f752e54d2cfaeb7e9d3bbf9e8125c95 Mon Sep 17 00:00:00 2001 From: Rob Austin Date: Sun, 2 Nov 2014 00:16:02 +0000 Subject: [PATCH] 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 --- .../openhft/chronicle/map/ChronicleMap.java | 4 +- .../openhft/chronicle/map/LockedEntry.java | 46 +++++ .../chronicle/map/ReplicatedChronicleMap.java | 147 ++++++++-------- .../openhft/chronicle/map/SegmentLock.java | 13 -- .../chronicle/map/StatelessChronicleMap.java | 4 +- .../chronicle/map/VanillaChronicleMap.java | 165 +++++++++++++----- .../chronicle/map/ChronicleMapTest.java | 58 +++++- .../chronicle/map/ReplicationCheckingMap.java | 4 +- .../jrs166/map/StatelessChronicleMapTest.java | 6 +- 9 files changed, 310 insertions(+), 137 deletions(-) create mode 100644 src/main/java/net/openhft/chronicle/map/LockedEntry.java delete mode 100644 src/main/java/net/openhft/chronicle/map/SegmentLock.java diff --git a/src/main/java/net/openhft/chronicle/map/ChronicleMap.java b/src/main/java/net/openhft/chronicle/map/ChronicleMap.java index 76b1f66b8..79f60e5cd 100644 --- a/src/main/java/net/openhft/chronicle/map/ChronicleMap.java +++ b/src/main/java/net/openhft/chronicle/map/ChronicleMap.java @@ -142,9 +142,9 @@ public interface ChronicleMap extends ConcurrentMap, ChronicleHash { // todo document - SegmentLock acquireUsingReadLocked(@NotNull K key, @NotNull V usingValue); + LockedEntry acquireUsingReadLocked(@NotNull K key, @NotNull V usingValue); // todo document - SegmentLock acquireUsingWriteLocked(@NotNull K key, @NotNull V usingValue); + LockedEntry acquireUsingWriteLocked(@NotNull K key, @NotNull V usingValue); } diff --git a/src/main/java/net/openhft/chronicle/map/LockedEntry.java b/src/main/java/net/openhft/chronicle/map/LockedEntry.java new file mode 100644 index 000000000..5cc3b807c --- /dev/null +++ b/src/main/java/net/openhft/chronicle/map/LockedEntry.java @@ -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 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(); + +} + diff --git a/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java b/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java index 0ca0d6ee9..32a31e82e 100644 --- a/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java +++ b/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java @@ -45,8 +45,7 @@ import static net.openhft.lang.MemoryUnit.*; import static net.openhft.lang.collection.DirectBitSet.NOT_FOUND; /** - *

A Replicating Multi Master HashMap

- *

Each remote hash map, mirrors its changes over to + *

A Replicating Multi Master HashMap

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. - *

- *

Reconciliation

- *

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.

+ *

Reconciliation

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.

- *

However there is an edge case that we have to concern ourselves with, If two + * ignored.

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. - *

+ * node with the smallest identifier wins.

* * @param the entries key type * @param the entries value type @@ -260,19 +255,41 @@ class ReplicatedChronicleMap, return (Segment) segments[segmentNum]; } - @Override - V lookupUsing(K key, V value, boolean create) { + LockedEntry 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 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, 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 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 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, /** * 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. - *

We can reject put() and removes() when comparing times stamps with remote - * systems + *

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, } /** - * 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, } /** - *

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. - *

- *

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. - *

- *

One way to perform this marshalling, would be to pipe the data into a queue. However, This class + *

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. + *

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.

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. - *

+ * changes.

* * @author Rob Austin. */ diff --git a/src/main/java/net/openhft/chronicle/map/SegmentLock.java b/src/main/java/net/openhft/chronicle/map/SegmentLock.java deleted file mode 100644 index bd140fcaa..000000000 --- a/src/main/java/net/openhft/chronicle/map/SegmentLock.java +++ /dev/null @@ -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(); - -} diff --git a/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java b/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java index 1d99bb2dc..a1da659d4 100644 --- a/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java +++ b/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java @@ -400,12 +400,12 @@ class StatelessChronicleMap implements ChronicleMap, 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(); } diff --git a/src/main/java/net/openhft/chronicle/map/VanillaChronicleMap.java b/src/main/java/net/openhft/chronicle/map/VanillaChronicleMap.java index de8dd53ba..d8dd18848 100644 --- a/src/main/java/net/openhft/chronicle/map/VanillaChronicleMap.java +++ b/src/main/java/net/openhft/chronicle/map/VanillaChronicleMap.java @@ -91,6 +91,7 @@ class VanillaChronicleMap, private final ChronicleHashErrorListener errorListener; private final int segmentHeaderSize; private final HashSplitting hashSplitting; + final boolean isNativeValueClass; transient Provider> keyReaderProvider; transient Provider keyInteropProvider; transient Provider> valueReaderProvider; @@ -115,6 +116,7 @@ class VanillaChronicleMap, SerializationBuilder valueBuilder = builder.valueBuilder; vClass = valueBuilder.eClass; + isNativeValueClass = vClass.getName().endsWith("$$Native"); valueSizeMarshaller = valueBuilder.sizeMarshaller(); originalValueReader = valueBuilder.reader(); @@ -331,7 +333,7 @@ class VanillaChronicleMap, } @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, @Override public V get(Object key) { - return lookupUsing((K) key, null, false); + + try (LockedEntry 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 kvLockedEntry = acquireUsing(key, usingValue, + LockType.READ_LOCK)) { + return kvLockedEntry.value(); + } } @Override - public SegmentLock acquireUsingReadLocked(K key, V usingValue) { + public LockedEntry acquireUsingReadLocked(K key, V usingValue) { return this.acquireUsing(key, usingValue, LockType.READ_LOCK); } @Override - public SegmentLock acquireUsingWriteLocked(K key, V usingValue) { + public LockedEntry 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, // todo remove @Override public V acquireUsing(K key, V usingValue) { - return lookupUsing(key, usingValue, true); + + try (LockedEntry 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 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, long segmentHash = segmentHash(hash); Segment segment = segments[segmentNum]; - SegmentLock lock = (lockTypeType == LockType.WRITE_LOCK) ? segment.writeLock() : segment.readLock(); + MutableLockedEntry 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, 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, private MultiMap hashLookup; private long nextPosToSearchFrom = 0L; - final SegmentLock readUnlock = new SegmentLock() { + final MutableLockedEntry readUnlock = new MutableLockedEntry() { @Override public void close() { Segment.this.readUnlock(); } }; - final SegmentLock writeUnlock = new SegmentLock() { + /** + * close for native values does not call put on close + */ + final MutableLockedEntry nativeWriteUnlock = new MutableLockedEntry() { @Override public void close() { Segment.this.writeUnlock(); } }; + /** + * close for heap values calls put on close + */ + final MutableLockedEntry + heapWriteUnlock = new MutableLockedEntry() { + @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, } - public SegmentLock readLock() throws IllegalStateException { + public MutableLockedEntry readLock() throws IllegalStateException { while (true) { final boolean success = segmentHeader.tryRWReadLock(LOCK_OFFSET, lockTimeOutNS); if (success) return readUnlock; @@ -772,10 +822,11 @@ class VanillaChronicleMap, } } - public SegmentLock writeLock() throws IllegalStateException { + public MutableLockedEntry 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, 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, return super.setValue(value); } } + + static abstract class MutableLockedEntry implements LockedEntry { + 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; + } + + } } + diff --git a/src/test/java/net/openhft/chronicle/map/ChronicleMapTest.java b/src/test/java/net/openhft/chronicle/map/ChronicleMapTest.java index 397d057df..1d9091615 100644 --- a/src/test/java/net/openhft/chronicle/map/ChronicleMapTest.java +++ b/src/test/java/net/openhft/chronicle/map/ChronicleMapTest.java @@ -38,6 +38,7 @@ import java.util.concurrent.atomic.AtomicInteger; import static net.openhft.chronicle.map.Alignment.*; import static org.junit.Assert.*; +import static org.junit.Assert.assertEquals; @SuppressWarnings({"unchecked", "ResultOfMethodCallIgnored"}) public class ChronicleMapTest { @@ -469,7 +470,7 @@ public class ChronicleMapTest { @Test public void testGetWithoutAcquireFirst() throws Exception { - ChronicleMap map = getSharedMap(10 , 1, 24); + ChronicleMap map = getSharedMap(10, 1, 24); assertNull(map.getUsing("key", DataValueClasses.newDirectReference(LongValue.class))); map.close(); @@ -1249,4 +1250,59 @@ public class ChronicleMapTest { } } } + + @Test + public void testOffheapAcquireUsingLocked() throws IOException { + OffHeapUpdatableChronicleMapBuilder builder = OffHeapUpdatableChronicleMapBuilder + .of(CharSequence.class, LongValue.class) + .entries(1000) + .entrySize(16); + File tmpFile = File.createTempFile("testAcquireUsingLocked", ".deleteme"); + tmpFile.deleteOnExit(); + final ChronicleMap map = builder.createWithFile(tmpFile); + + LongValue value = nativeLongValue(); + try (LockedEntry lockedEntry = map.acquireUsingWriteLocked("one", value)) { + lockedEntry.value().addValue(1); + } + + try (LockedEntry lockedEntry = map.acquireUsingReadLocked("two", value)) { + assertEquals(null, lockedEntry.value()); + } + + try (LockedEntry lockedEntry = map.acquireUsingReadLocked("one", value)) { + assertEquals(1, lockedEntry.value().getValue()); + } + map.close(); + } + + @Test + public void testOnheapAcquireUsingLocked() throws IOException { + ChronicleMapBuilder builder = ChronicleMapBuilder + .of(CharSequence.class, LongValue.class) + .entries(1000) + .entrySize(40); + File tmpFile = File.createTempFile("testAcquireUsingLocked", ".deleteme"); + tmpFile.deleteOnExit(); + final ChronicleMap map = builder.createWithFile(tmpFile); + + LongValue value = DataValueClasses.newInstance(LongValue.class); + + + try (LockedEntry lockedEntry = map.acquireUsingWriteLocked("one", value)) { + lockedEntry.value().addValue(1); + } + + try (LockedEntry lockedEntry = map.acquireUsingReadLocked("two", value)) { + assertEquals(null, lockedEntry.value()); + } + + try (LockedEntry lockedEntry = map.acquireUsingReadLocked("one", value)) { + assertEquals(1, lockedEntry.value().getValue()); + } + + map.close(); + } } + + diff --git a/src/test/java/net/openhft/chronicle/map/ReplicationCheckingMap.java b/src/test/java/net/openhft/chronicle/map/ReplicationCheckingMap.java index 2771aebf0..9971a8f46 100644 --- a/src/test/java/net/openhft/chronicle/map/ReplicationCheckingMap.java +++ b/src/test/java/net/openhft/chronicle/map/ReplicationCheckingMap.java @@ -252,12 +252,12 @@ public class ReplicationCheckingMap implements ChronicleMap { } @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(); } diff --git a/src/test/java/net/openhft/chronicle/map/jrs166/map/StatelessChronicleMapTest.java b/src/test/java/net/openhft/chronicle/map/jrs166/map/StatelessChronicleMapTest.java index 51a70cd46..5bcae4591 100644 --- a/src/test/java/net/openhft/chronicle/map/jrs166/map/StatelessChronicleMapTest.java +++ b/src/test/java/net/openhft/chronicle/map/jrs166/map/StatelessChronicleMapTest.java @@ -21,7 +21,7 @@ package net.openhft.chronicle.map.jrs166.map; import net.openhft.chronicle.map.ChronicleMap; import net.openhft.chronicle.map.ChronicleMapBuilder; import net.openhft.chronicle.hash.TcpReplicationConfig; -import net.openhft.chronicle.map.SegmentLock; +import net.openhft.chronicle.map.LockedEntry; import net.openhft.chronicle.map.jrs166.JSR166TestCase; import org.jetbrains.annotations.NotNull; import org.junit.Test; @@ -151,12 +151,12 @@ public class StatelessChronicleMapTest extends JSR166TestCase { } @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(); }