From 100d839a8bf637169486e0a7825c2d8a51b024d8 Mon Sep 17 00:00:00 2001 From: Rob Austin Date: Sat, 1 Nov 2014 15:53:47 +0000 Subject: [PATCH] HCOLL-155 uncompleted, checked in order that Peter can review the changes so far of - chm.acquireUsingReadLocked and chm.acquireUsingWriteLocked --- pom.xml | 2 +- .../map/AbstractChronicleMapBuilder.java | 2 +- .../openhft/chronicle/map/ChronicleMap.java | 12 +- .../openhft/chronicle/map/SegmentLock.java | 13 ++ .../openhft/chronicle/map/SharedSegment.java | 4 +- .../chronicle/map/StatelessChronicleMap.java | 180 +++++++++++++---- .../openhft/chronicle/map/TcpReplicator.java | 53 ++++- .../chronicle/map/VanillaChronicleMap.java | 188 ++++++++++++------ .../chronicle/map/ReplicationCheckingMap.java | 10 + .../chronicle/map/StatelessClientTest.java | 127 ++++++++++++ .../jrs166/map/StatelessChronicleMapTest.java | 11 + 11 files changed, 495 insertions(+), 107 deletions(-) create mode 100644 src/main/java/net/openhft/chronicle/map/SegmentLock.java diff --git a/pom.xml b/pom.xml index 0cae561ea..a74a0acf7 100755 --- a/pom.xml +++ b/pom.xml @@ -46,7 +46,7 @@ net.openhft lang - 6.4.11 + 6.4.12-SNAPSHOT diff --git a/src/main/java/net/openhft/chronicle/map/AbstractChronicleMapBuilder.java b/src/main/java/net/openhft/chronicle/map/AbstractChronicleMapBuilder.java index 08a44d6ea..1e4fce4fa 100644 --- a/src/main/java/net/openhft/chronicle/map/AbstractChronicleMapBuilder.java +++ b/src/main/java/net/openhft/chronicle/map/AbstractChronicleMapBuilder.java @@ -843,7 +843,7 @@ public abstract class AbstractChronicleMapBuilder newMap() throws IOException { diff --git a/src/main/java/net/openhft/chronicle/map/ChronicleMap.java b/src/main/java/net/openhft/chronicle/map/ChronicleMap.java index 50524b87e..76b1f66b8 100644 --- a/src/main/java/net/openhft/chronicle/map/ChronicleMap.java +++ b/src/main/java/net/openhft/chronicle/map/ChronicleMap.java @@ -23,6 +23,7 @@ import net.openhft.chronicle.hash.serialization.BytesReader; import net.openhft.lang.io.Bytes; import net.openhft.lang.io.serialization.BytesMarshaller; import net.openhft.lang.model.Byteable; +import org.jetbrains.annotations.NotNull; import java.io.Closeable; import java.util.Map; @@ -100,7 +101,7 @@ public interface ChronicleMap extends ConcurrentMap, ChronicleHash { * @see #acquireUsing(Object, Object) * @see ChronicleMapBuilder#valueMarshaller(BytesMarshaller) */ - V getUsing(K key, V usingValue); + V getUsing(K key, V usingValue); /** * Acquire a value for a key, creating if absent. @@ -137,6 +138,13 @@ public interface ChronicleMap extends ConcurrentMap, ChronicleHash { * @return value to which the given key is mapping after this call, either found or created * @see #getUsing(Object, Object) */ - V acquireUsing(K key, V usingValue); + public V acquireUsing(K key, V usingValue) ; + + // todo document + SegmentLock acquireUsingReadLocked(@NotNull K key, @NotNull V usingValue); + + // todo document + SegmentLock acquireUsingWriteLocked(@NotNull K key, @NotNull V usingValue); } + diff --git a/src/main/java/net/openhft/chronicle/map/SegmentLock.java b/src/main/java/net/openhft/chronicle/map/SegmentLock.java new file mode 100644 index 000000000..bd140fcaa --- /dev/null +++ b/src/main/java/net/openhft/chronicle/map/SegmentLock.java @@ -0,0 +1,13 @@ +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/SharedSegment.java b/src/main/java/net/openhft/chronicle/map/SharedSegment.java index bb2c5113b..b0fd84c40 100644 --- a/src/main/java/net/openhft/chronicle/map/SharedSegment.java +++ b/src/main/java/net/openhft/chronicle/map/SharedSegment.java @@ -28,9 +28,9 @@ import java.util.Map; */ interface SharedSegment { - void readLock(); + AutoCloseable readLock(); - void writeLock(); + AutoCloseable writeLock(); Map.Entry getEntry(long pos); diff --git a/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java b/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java index 1be7354ef..1d99bb2dc 100644 --- a/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java +++ b/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java @@ -65,6 +65,8 @@ class StatelessChronicleMap implements ChronicleMap, Closeable { private int maxEntrySize; private final Class kClass; private final Class vClass; + private final boolean putReturnsNull; + private final boolean removeReturnsNull; static enum EventId { @@ -74,8 +76,11 @@ class StatelessChronicleMap implements ChronicleMap, Closeable { IS_EMPTY, CONTAINS_KEY, CONTAINS_VALUE, - GET, PUT, + GET, + PUT, + PUT_WITHOUT_ACC, REMOVE, + REMOVE_WITHOUT_ACC, CLEAR, KEY_SET, VALUES, @@ -86,6 +91,7 @@ class StatelessChronicleMap implements ChronicleMap, Closeable { REMOVE_WITH_VALUE, TO_STRING, PUT_ALL, + PUT_ALL_WITHOUT_ACC, HASH_CODE } @@ -93,12 +99,16 @@ class StatelessChronicleMap implements ChronicleMap, Closeable { StatelessChronicleMap(final KeyValueSerializer keyValueSerializer, final StatelessBuilder builder, - int maxEntrySize, Class kClass, Class vClass) throws IOException { + int maxEntrySize, Class kClass, Class vClass, + boolean putReturnsNull, + boolean removeReturnsNull) throws IOException { this.keyValueSerializer = keyValueSerializer; this.builder = builder; this.maxEntrySize = maxEntrySize; this.kClass = kClass; this.vClass = vClass; + this.putReturnsNull = putReturnsNull; + this.removeReturnsNull = removeReturnsNull; attemptConnect(builder.remoteAddress()); buffer = allocateDirect(maxEntrySize).order(ByteOrder.nativeOrder()); @@ -242,7 +252,7 @@ class StatelessChronicleMap implements ChronicleMap, Closeable { public synchronized V putIfAbsent(K key, V value) { - final long sizeLocation = writeEvent(PUT_IF_ABSENT); + final long sizeLocation = writeEventAnSkip(PUT_IF_ABSENT); final ThreadLocalCopies local = keyValueSerializer.threadLocalCopies(); writeKey(key, local); writeValue(value, local); @@ -255,7 +265,7 @@ class StatelessChronicleMap implements ChronicleMap, Closeable { if (key == null) throw new NullPointerException("key can not be null"); - final long sizeLocation = writeEvent(REMOVE_WITH_VALUE); + final long sizeLocation = writeEventAnSkip(REMOVE_WITH_VALUE); final ThreadLocalCopies local = keyValueSerializer.threadLocalCopies(); writeKey((K) key, local); writeValue((V) value, local); @@ -269,7 +279,7 @@ class StatelessChronicleMap implements ChronicleMap, Closeable { if (key == null) throw new NullPointerException("key can not be null"); - final long sizeLocation = writeEvent(REPLACE_WITH_OLD_AND_NEW_VALUE); + final long sizeLocation = writeEventAnSkip(REPLACE_WITH_OLD_AND_NEW_VALUE); final ThreadLocalCopies local = keyValueSerializer.threadLocalCopies(); writeKey(key, local); writeValue(oldValue, local); @@ -284,7 +294,7 @@ class StatelessChronicleMap implements ChronicleMap, Closeable { if (key == null) throw new NullPointerException("key can not be null"); - final long sizeLocation = writeEvent(REPLACE); + final long sizeLocation = writeEventAnSkip(REPLACE); final ThreadLocalCopies local = keyValueSerializer.threadLocalCopies(); writeKey(key, local); writeValue(value, local); @@ -329,31 +339,23 @@ class StatelessChronicleMap implements ChronicleMap, Closeable { } - public synchronized void putAll(Map map) { - - final long sizeLocation = writeEvent(PUT_ALL); - writeEntries(map); - blockingFetch(sizeLocation); - } - - @Override public synchronized int hashCode() { - return blockingFetch(writeEvent(HASH_CODE)).readInt(); + return blockingFetch(writeEventAnSkip(HASH_CODE)).readInt(); } public synchronized String toString() { - return blockingFetch(writeEvent(TO_STRING)).readObject(String.class); + return blockingFetch(writeEventAnSkip(TO_STRING)).readObject(String.class); } public synchronized boolean isEmpty() { - return blockingFetch(writeEvent(IS_EMPTY)).readBoolean(); + return blockingFetch(writeEventAnSkip(IS_EMPTY)).readBoolean(); } public synchronized boolean containsKey(Object key) { - final long sizeLocation = writeEvent(CONTAINS_KEY); + final long sizeLocation = writeEventAnSkip(CONTAINS_KEY); writeKey((K) key); // get the data back from the server @@ -363,7 +365,7 @@ class StatelessChronicleMap implements ChronicleMap, Closeable { public synchronized boolean containsValue(Object value) { - final long sizeLocation = writeEvent(CONTAINS_VALUE); + final long sizeLocation = writeEventAnSkip(CONTAINS_VALUE); writeValue((V) value); // get the data back from the server @@ -372,12 +374,12 @@ class StatelessChronicleMap implements ChronicleMap, Closeable { public synchronized long longSize() { - return blockingFetch(writeEvent(LONG_SIZE)).readLong(); + return blockingFetch(writeEventAnSkip(LONG_SIZE)).readLong(); } public synchronized V get(Object key) { - final long sizeLocation = writeEvent(GET); + final long sizeLocation = writeEventAnSkip(GET); final ThreadLocalCopies local = keyValueSerializer.threadLocalCopies(); writeKey((K) key, local); @@ -387,35 +389,102 @@ class StatelessChronicleMap implements ChronicleMap, Closeable { } public synchronized V getUsing(K key, V usingValue) { - throw new UnsupportedOperationException("getUsing is not supported for stateless clients"); + throw new UnsupportedOperationException("getUsing() is not supported for stateless " + + "clients"); } public synchronized V acquireUsing(K key, V usingValue) { - throw new UnsupportedOperationException("getUsing is not supported for stateless clients"); + throw new UnsupportedOperationException("acquireUsing() is not supported for stateless " + + "clients"); } - public synchronized V put(K key, V value) { - - final long sizeLocation = writeEvent(PUT); - final ThreadLocalCopies local = keyValueSerializer.threadLocalCopies(); - writeKey(key, local); - writeValue(value, local); - - // get the data back from the server - return readKey(sizeLocation, local); + @Override + public SegmentLock acquireUsingReadLocked(K key, V usingValue) { + throw new UnsupportedOperationException(); } + @Override + public SegmentLock acquireUsingWriteLocked(K key, V usingValue) { + throw new UnsupportedOperationException(); + } + + public synchronized V remove(Object key) { if (key == null) throw new NullPointerException("key can not be null"); - final long sizeLocation = writeEvent(REMOVE); + final long sizeLocation = (removeReturnsNull) ? writeEvent(REMOVE_WITHOUT_ACC) : writeEventAnSkip(REMOVE); + + writeKey((K) key); - // get the data back from the server - return readKey(sizeLocation); + if (removeReturnsNull) { + sendWithoutAcc(bytes, sizeLocation); + return null; + } else + // get the data back from the server + return readKey(sizeLocation); + } + + + public synchronized V put(K key, V value) { + + final ThreadLocalCopies local = keyValueSerializer.threadLocalCopies(); + + final long sizeLocation = (putReturnsNull) ? writeEvent(PUT_WITHOUT_ACC) : writeEventAnSkip(PUT); + + writeKey(key, local); + writeValue(value, local); + + + if (putReturnsNull) { + sendWithoutAcc(bytes, sizeLocation); + return null; + } else + // get the data back from the server + return readKey(sizeLocation, local); + + } + + + public synchronized void putAll(Map map) { + + final long sizeLocation = (putReturnsNull) ? writeEvent(PUT_ALL_WITHOUT_ACC) : writeEventAnSkip(PUT_ALL); + + writeEntries(map); + + if (putReturnsNull) + sendWithoutAcc(bytes, sizeLocation); + else + blockingFetch(sizeLocation); + + } + + + private void sendWithoutAcc(final Bytes bytes, long sizeLocation) { + writeSizeAt(sizeLocation); + long timeoutTime = builder.timeoutMs() + System.currentTimeMillis(); + try { + for (; ; ) { + + if (clientChannel == null) + clientChannel = lazyConnect(builder.timeoutMs(), builder.remoteAddress()); + + try { + + send(bytes, timeoutTime); + return; + } catch (java.nio.channels.ClosedChannelException | ClosedConnectionException e) { + checkTimeout(timeoutTime); + clientChannel = lazyConnect(builder.timeoutMs(), builder.remoteAddress()); + } + } + } catch (IOException e) { + close(); + throw new IORuntimeException(e); + } } @@ -534,13 +603,13 @@ class StatelessChronicleMap implements ChronicleMap, Closeable { public synchronized void clear() { // get the data back from the server - blockingFetch(writeEvent(CLEAR)); + blockingFetch(writeEventAnSkip(CLEAR)); } @NotNull public synchronized Collection values() { - final long sizeLocation = writeEvent(VALUES); + final long sizeLocation = writeEventAnSkip(VALUES); final long startTime = System.currentTimeMillis(); final long transactionId = nextUniqueTransaction(startTime); @@ -575,7 +644,7 @@ class StatelessChronicleMap implements ChronicleMap, Closeable { @NotNull public synchronized Set> entrySet() { - final long sizeLocation = writeEvent(ENTRY_SET); + final long sizeLocation = writeEventAnSkip(ENTRY_SET); final long startTime = System.currentTimeMillis(); final long transactionId = nextUniqueTransaction(startTime); @@ -623,7 +692,7 @@ class StatelessChronicleMap implements ChronicleMap, Closeable { @NotNull public synchronized Set keySet() { - final long sizeLocation = writeEvent(KEY_SET); + final long sizeLocation = writeEventAnSkip(KEY_SET); final long startTime = System.currentTimeMillis(); final long transactionId = nextUniqueTransaction(startTime); @@ -656,9 +725,7 @@ class StatelessChronicleMap implements ChronicleMap, Closeable { return result; } - private long writeEvent(EventId event) { - buffer.clear(); bytes.clear(); @@ -666,9 +733,28 @@ class StatelessChronicleMap implements ChronicleMap, Closeable { bytes.write((byte) event.ordinal()); long sizeLocation = markSizeLocation(); + + return sizeLocation; } + + /** + * skips for the transactionid + * + * @param event + * @return + */ + private long writeEventAnSkip(EventId event) { + + long sizeLocation = writeEvent(event); + + // skips for the transaction id + bytes.skip(SIZE_OF_TRANSACTIONID); + return sizeLocation; + } + + class Entry implements Map.Entry { final K key; @@ -828,7 +914,7 @@ class StatelessChronicleMap implements ChronicleMap, Closeable { } - private ByteBufferBytes receive(int requiredNumberOfBytes, long timeoutTime) throws IOException { + private Bytes receive(int requiredNumberOfBytes, long timeoutTime) throws IOException { while (bytes.buffer().position() < requiredNumberOfBytes) { clientChannel.read(bytes.buffer()); @@ -839,7 +925,7 @@ class StatelessChronicleMap implements ChronicleMap, Closeable { return bytes; } - private void send(final ByteBufferBytes out, long timeoutTime) throws IOException { + private void send(final Bytes out, long timeoutTime) throws IOException { buffer.limit((int) out.position()); buffer.position(0); @@ -869,10 +955,16 @@ class StatelessChronicleMap implements ChronicleMap, Closeable { } + + private void writeSizeAt(long locationOfSize) { + final long size = bytes.position() - locationOfSize; + bytes.writeInt(locationOfSize, (int) size - SIZE_OF_SIZE); + } + + private long markSizeLocation() { final long position = bytes.position(); bytes.skip(SIZE_OF_SIZE); - bytes.skip(SIZE_OF_TRANSACTIONID); return position; } diff --git a/src/main/java/net/openhft/chronicle/map/TcpReplicator.java b/src/main/java/net/openhft/chronicle/map/TcpReplicator.java index bd8a018d7..b92a09d27 100644 --- a/src/main/java/net/openhft/chronicle/map/TcpReplicator.java +++ b/src/main/java/net/openhft/chronicle/map/TcpReplicator.java @@ -1261,6 +1261,16 @@ class StatelessServerConnector { case ENTRY_SET: return entrySet(reader, writer); + case PUT_WITHOUT_ACC: + return put(reader); + + + case PUT_ALL_WITHOUT_ACC: + return putAll(reader); + + case REMOVE_WITHOUT_ACC: + return remove(reader); + } final long sizeLocation = reflectTransactionId(reader, writer); @@ -1484,6 +1494,20 @@ class StatelessServerConnector { } + + private Work put(Bytes reader) { + if (MAP_SUPPORTS_BYTES) { + map.put(reader); + } else { + final ThreadLocalCopies local = keyValueSerializer.threadLocalCopies(); + final K k = readKey(reader, local); + final V v = readValue(reader, local); + map.put(k, v); + } + + return null; + } + private Work put(Bytes reader, Bytes writer, final long sizeLocation) { if (MAP_SUPPORTS_BYTES) { map.put(reader, writer); @@ -1502,6 +1526,19 @@ class StatelessServerConnector { return null; } + private Work remove(Bytes reader) { + if (MAP_SUPPORTS_BYTES) { + map.removeWithBytes(reader); + } else { + final ThreadLocalCopies local = keyValueSerializer.threadLocalCopies(); + final K key = readKey(reader, local); + map.remove(key); + } + + return null; + } + + private Work remove(Bytes reader, Bytes writer, final long sizeLocation) { if (MAP_SUPPORTS_BYTES) { map.remove(reader, writer); @@ -1540,6 +1577,18 @@ class StatelessServerConnector { return null; } + private Work putAll(Bytes reader) { + + if (MAP_SUPPORTS_BYTES) { + map.putAll(reader); + } else { + final ThreadLocalCopies local = keyValueSerializer.threadLocalCopies(); + map.putAll(readEntries(reader, local)); + } + + return null; + } + private Work clear(Bytes reader, Bytes writer, final long sizeLocation) { @@ -1788,7 +1837,9 @@ class StatelessServerConnector { final HashMap result = new HashMap(); for (long i = 0; i < numberOfEntries; i++) { - result.put(readKey(reader, local), readValue(reader, local)); + K key = readKey(reader, local); + V value = readValue(reader, local); + result.put(key, value); } return result; } diff --git a/src/main/java/net/openhft/chronicle/map/VanillaChronicleMap.java b/src/main/java/net/openhft/chronicle/map/VanillaChronicleMap.java index 3abecdc90..de8dd53ba 100644 --- a/src/main/java/net/openhft/chronicle/map/VanillaChronicleMap.java +++ b/src/main/java/net/openhft/chronicle/map/VanillaChronicleMap.java @@ -29,6 +29,8 @@ import net.openhft.lang.io.serialization.impl.VanillaBytesMarshallerFactory; import net.openhft.lang.threadlocal.Provider; import net.openhft.lang.threadlocal.ThreadLocalCopies; import org.jetbrains.annotations.NotNull; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; import java.io.File; import java.io.IOException; @@ -48,6 +50,10 @@ import static net.openhft.lang.MemoryUnit.*; class VanillaChronicleMap, V, VW, MVW extends MetaBytesWriter> extends AbstractMap implements ChronicleMap, Serializable { + + + private static final Logger LOG = LoggerFactory.getLogger(VanillaChronicleMap.class); + /** * Because DirectBitSet implementations couldn't find more than 64 continuous clear or set * bits. @@ -325,11 +331,12 @@ class VanillaChronicleMap, } @Override - public V putIfAbsent(K key, V value) { + public V putIfAbsent( K key, V value) { return put0(key, value, false); } - private V put0(K key, V value, boolean replaceIfPresent) { + private V put0(@net.openhft.lang.model.constraints.NotNull K key, V value, + boolean replaceIfPresent) { checkKey(key); checkValue(value); ThreadLocalCopies copies = keyInteropProvider.getCopies(null); @@ -354,11 +361,51 @@ class VanillaChronicleMap, return lookupUsing(key, usingValue, false); } + @Override + public SegmentLock acquireUsingReadLocked(K key, V usingValue) { + return this.acquireUsing(key, usingValue, LockType.READ_LOCK); + } + + @Override + public SegmentLock acquireUsingWriteLocked(K key, V usingValue) { + return this.acquireUsing(key, usingValue, LockType.WRITE_LOCK); + + //todo call put() on close of a on-heap value + + //todo set dirty when using replication on + } + + + // todo remove @Override public V acquireUsing(K key, V usingValue) { return lookupUsing(key, usingValue, true); } + + private enum LockType {READ_LOCK, WRITE_LOCK} + + SegmentLock acquireUsing(K key, V value, LockType lockTypeType) { + checkKey(key); + ThreadLocalCopies copies = keyInteropProvider.getCopies(null); + KI keyInterop = keyInteropProvider.get(copies, originalKeyInterop); + copies = metaKeyInteropProvider.getCopies(copies); + MKI metaKeyInterop = + metaKeyInteropProvider.get(copies, originalMetaKeyInterop, keyInterop, key); + long hash = metaKeyInterop.hash(keyInterop, key); + int segmentNum = getSegment(hash); + long segmentHash = segmentHash(hash); + + Segment segment = segments[segmentNum]; + SegmentLock lock = (lockTypeType == LockType.WRITE_LOCK) ? segment.writeLock() : segment.readLock(); + + + segment.acquireWithoutLock(copies, metaKeyInterop, keyInterop, key, value, + segmentHash, false); + return lock; + + } + V lookupUsing(K key, V value, boolean create) { checkKey(key); ThreadLocalCopies copies = keyInteropProvider.getCopies(null); @@ -369,8 +416,13 @@ class VanillaChronicleMap, long hash = metaKeyInterop.hash(keyInterop, key); int segmentNum = getSegment(hash); long segmentHash = segmentHash(hash); - return segments[segmentNum].acquire(copies, metaKeyInterop, keyInterop, key, value, - segmentHash, create); + Segment segment = segments[segmentNum]; + + try (SegmentLock lock = create ? segment.writeLock() : segment.readLock()) { + return segment.acquireWithoutLock(copies, metaKeyInterop, keyInterop, key, value, + segmentHash, create); + } + } @Override @@ -400,6 +452,7 @@ class VanillaChronicleMap, return (entrySet != null) ? entrySet : (entrySet = new EntrySet()); } + /** * @throws NullPointerException if the specified key is null */ @@ -418,6 +471,7 @@ class VanillaChronicleMap, return removeIfValueIs(key, (V) value) != null; } + /** * removes ( if there exists ) an entry from the map, if the {@param key} and {@param * expectedValue} match that of a maps.entry. If the {@param expectedValue} equals null then ( @@ -451,6 +505,7 @@ class VanillaChronicleMap, return oldValue.equals(replaceIfValueIs(key, oldValue, newValue)); } + /** * @throws NullPointerException if the specified key or value is null */ @@ -569,6 +624,10 @@ class VanillaChronicleMap, throw new UnsupportedOperationException("todo"); } + public void putAll(Bytes reader) { + throw new UnsupportedOperationException("todo"); + } + long[] segmentSizes() { long[] sizes = new long[segments.length]; for (int i = 0; i < segments.length; i++) @@ -576,6 +635,14 @@ class VanillaChronicleMap, return sizes; } + public void put(Bytes reader) { + throw new UnsupportedOperationException("todo"); + } + + public void removeWithBytes(Bytes reader) { + throw new UnsupportedOperationException("todo"); + } + // these methods should be package local, not public or private. class Segment implements SharedSegment { /* @@ -597,6 +664,19 @@ class VanillaChronicleMap, private MultiMap hashLookup; private long nextPosToSearchFrom = 0L; + final SegmentLock readUnlock = new SegmentLock() { + @Override + public void close() { + Segment.this.readUnlock(); + } + }; + + final SegmentLock writeUnlock = new SegmentLock() { + @Override + public void close() { + Segment.this.writeUnlock(); + } + }; /** * @param index the index of this segment held by the map @@ -679,10 +759,10 @@ class VanillaChronicleMap, } - public void readLock() throws IllegalStateException { + public SegmentLock readLock() throws IllegalStateException { while (true) { final boolean success = segmentHeader.tryRWReadLock(LOCK_OFFSET, lockTimeOutNS); - if (success) return; + if (success) return readUnlock; if (currentThread().isInterrupted()) { throw new IllegalStateException(new InterruptedException("Unable to obtain lock, interrupted")); } else { @@ -692,10 +772,10 @@ class VanillaChronicleMap, } } - public void writeLock() throws IllegalStateException { + public SegmentLock writeLock() throws IllegalStateException { while (true) { final boolean success = segmentHeader.tryRWWriteLock(LOCK_OFFSET, lockTimeOutNS); - if (success) return; + if (success) return writeUnlock; if (currentThread().isInterrupted()) { throw new IllegalStateException(new InterruptedException("Unable to obtain lock, interrupted")); } else { @@ -762,51 +842,44 @@ class VanillaChronicleMap, return (int) (sizeInBytes / entrySize) + 1; } - V acquire(ThreadLocalCopies copies, MKI metaKeyInterop, KI keyInterop, K key, V usingValue, - long hash2, boolean create) { + V acquireWithoutLock(ThreadLocalCopies copies, MKI metaKeyInterop, KI keyInterop, K key, V usingValue, + long hash2, boolean create) { V v = acquire0(copies, metaKeyInterop, keyInterop, key, usingValue, hash2, false); if (create && v == null) - v = acquire0(copies, metaKeyInterop, keyInterop, key, usingValue, hash2, true); + v = acquire0(copies, metaKeyInterop, keyInterop, key, usingValue, hash2, true); return v; } - V acquire0(ThreadLocalCopies copies, MKI metaKeyInterop, KI keyInterop, K key, V usingValue, - long hash2, boolean create) { - if (create) - writeLock(); - else - readLock(); - try { - long keySize = metaKeyInterop.size(keyInterop, key); - MultiStoreBytes entry = acquireTmpBytes(); - long offset = searchKey(keyInterop, metaKeyInterop, key, keySize, hash2, entry, - hashLookup); - 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, - metaValueWriter, valueWriter, defaultValue); - } else { - offset = putEntry(metaKeyInterop, keyInterop, key, keySize, - prepareValueBytesAsWriter, null, key); - } - return onKeyAbsentOnAcquire(copies, key, keySize, usingValue, offset); + V acquire0(ThreadLocalCopies copies, MKI metaKeyInterop, KI keyInterop, K key, V usingValue, + long hash2, boolean create) { + + + long keySize = metaKeyInterop.size(keyInterop, key); + MultiStoreBytes entry = acquireTmpBytes(); + long offset = searchKey(keyInterop, metaKeyInterop, key, keySize, hash2, entry, + hashLookup); + 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, + metaValueWriter, valueWriter, defaultValue); + } else { + offset = putEntry(metaKeyInterop, keyInterop, key, keySize, + prepareValueBytesAsWriter, null, key); } - } finally { - if (create) - writeUnlock(); - else - readUnlock(); + return onKeyAbsentOnAcquire(copies, key, keySize, usingValue, offset); } + } long searchKey(KI keyInterop, MKI metaKeyInterop, K key, long keySize, long hash2, @@ -1007,12 +1080,12 @@ class VanillaChronicleMap, /** * Removes a key (or key-value pair) from the Segment. * - *

The entry will only be removed if {@code expectedValue} equals to {@code null} or the value previously - * corresponding to the specified key. + *

The entry will only be removed if {@code expectedValue} equals to {@code null} or the + * value previously corresponding to the specified key. * * @param hash2 a hash code related to the {@code keyBytes} - * @return the value of the entry that was removed if the entry corresponding to the {@code keyBytes} exists and - * {@link #removeReturnsNull} is {@code false}, {@code null} otherwise + * @return the value of the entry that was removed if the entry corresponding to the {@code + * keyBytes} exists and {@link #removeReturnsNull} is {@code false}, {@code null} otherwise */ V remove(ThreadLocalCopies copies, MKI metaKeyInterop, KI keyInterop, K key, V expectedValue, long hash2) { @@ -1068,8 +1141,9 @@ class VanillaChronicleMap, } /** - * Replaces the specified value for the key with the given value. {@code newValue} is set only if the existing - * value corresponding to the specified key is equal to {@code expectedValue} or {@code expectedValue == null}. + * Replaces the specified value for the key with the given value. {@code newValue} is set + * only if the existing value corresponding to the specified key is equal to {@code + * expectedValue} or {@code expectedValue == null}. * * @param hash2 a hash code related to the {@code keyBytes} * @return the replaced value or {@code null} if the value was not replaced @@ -1145,15 +1219,17 @@ class VanillaChronicleMap, } /** - * Replaces value in existing entry. May cause entry relocation, because there may be not enough space for new - * value in location already allocated for this entry. + * Replaces value in existing entry. May cause entry relocation, because there may be not + * enough space for new value in location already allocated for this entry. * * @param pos index of the first block occupied by the entry - * @param offset relative offset of the entry in Segment bytes (before, i. e. including metaData) + * @param offset relative offset of the entry in Segment bytes (before, i. e. + * including metaData) * @param entry relative pointer in Segment bytes * @param valueSizePos relative position of value size in entry * @param entryEndAddr absolute address of the entry end - * @return relative offset of the entry in Segment bytes after putting value (that may cause entry relocation) + * @return relative offset of the entry in Segment bytes after putting value (that may cause + * entry relocation) */ long putValue(long pos, long offset, NativeBytes entry, long valueSizePos, long entryEndAddr, ThreadLocalCopies copies, V value, Bytes valueBytes, diff --git a/src/test/java/net/openhft/chronicle/map/ReplicationCheckingMap.java b/src/test/java/net/openhft/chronicle/map/ReplicationCheckingMap.java index 255308c93..2771aebf0 100644 --- a/src/test/java/net/openhft/chronicle/map/ReplicationCheckingMap.java +++ b/src/test/java/net/openhft/chronicle/map/ReplicationCheckingMap.java @@ -251,6 +251,16 @@ public class ReplicationCheckingMap implements ChronicleMap { return map1.acquireUsing(key, usingValue); } + @Override + public SegmentLock acquireUsingReadLocked(K key, V usingValue) { + throw new UnsupportedOperationException(); + } + + @Override + public SegmentLock acquireUsingWriteLocked(K key, V usingValue) { + throw new UnsupportedOperationException(); + } + @Override public File file() { throw new UnsupportedOperationException(); diff --git a/src/test/java/net/openhft/chronicle/map/StatelessClientTest.java b/src/test/java/net/openhft/chronicle/map/StatelessClientTest.java index fd413aa1c..e3495a017 100644 --- a/src/test/java/net/openhft/chronicle/map/StatelessClientTest.java +++ b/src/test/java/net/openhft/chronicle/map/StatelessClientTest.java @@ -270,6 +270,133 @@ public class StatelessClientTest { } + @Test + public void testBufferOverFlowPutAllWherePutReturnsNull() throws IOException, + InterruptedException { + + final ChronicleMap serverMap; + final ChronicleMap statelessMap; + + + // stateless client + { + statelessMap = ChronicleMapBuilder.of(Integer + .class, CharSequence.class) + .putReturnsNull(true) + .stateless(remoteAddress(new InetSocketAddress("localhost", 8086))).create(); + } + + // server + { + serverMap = ChronicleMapBuilder.of(Integer.class, CharSequence.class) + .replicators((byte) 2, TcpReplicationConfig.of(8086)) + .putReturnsNull(true) + .create(); + } + + + Map payload = new HashMap(); + + for (int i = 0; i < SIZE; i++) { + payload.put(i, "some value=" + i); + } + + + statelessMap.putAll(payload); + + int value = SIZE - 10; + + Assert.assertEquals("some value=" + value, statelessMap.get(value)); + Assert.assertEquals(SIZE, statelessMap.size()); + + serverMap.close(); + statelessMap.close(); + + } + + + @Test + public void testPutWherePutReturnsNull() throws IOException, + InterruptedException { + + final ChronicleMap serverMap; + final ChronicleMap statelessMap; + + + // stateless client + { + statelessMap = ChronicleMapBuilder.of(Integer + .class, CharSequence.class) + .putReturnsNull(true) + .removeReturnsNull(true) + .stateless(remoteAddress(new InetSocketAddress("localhost", 8086))).create(); + } + + // server + { + serverMap = ChronicleMapBuilder.of(Integer.class, CharSequence.class) + .replicators((byte) 2, TcpReplicationConfig.of(8086)) + .putReturnsNull(true) + .removeReturnsNull(true) + .create(); + } + + + statelessMap.put(1, "some value"); + + Assert.assertEquals("some value", statelessMap.get(1)); + Assert.assertEquals(1, statelessMap.size()); + + + serverMap.close(); + statelessMap.close(); + + } + + + + @Test + public void testRemoveWhereRemoveReturnsNull() throws IOException, + InterruptedException { + + final ChronicleMap serverMap; + final ChronicleMap statelessMap; + + + // stateless client + { + statelessMap = ChronicleMapBuilder.of(Integer + .class, CharSequence.class) + .putReturnsNull(true) + .removeReturnsNull(true) + .stateless(remoteAddress(new InetSocketAddress("localhost", 8086))).create(); + } + + // server + { + serverMap = ChronicleMapBuilder.of(Integer.class, CharSequence.class) + .replicators((byte) 2, TcpReplicationConfig.of(8086)) + .putReturnsNull(true) + .removeReturnsNull(true) + .create(); + } + + + statelessMap.put(1, "some value"); + + Assert.assertEquals("some value", statelessMap.get(1)); + Assert.assertEquals(1, statelessMap.size()); + + statelessMap.remove(1); + + Assert.assertEquals(null, statelessMap.get(1)); + Assert.assertEquals(0, statelessMap.size()); + + serverMap.close(); + statelessMap.close(); + + } + @Test public void testEquals() throws IOException, InterruptedException { 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 28f850b0d..51a70cd46 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,6 +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.jrs166.JSR166TestCase; import org.jetbrains.annotations.NotNull; import org.junit.Test; @@ -149,6 +150,16 @@ public class StatelessChronicleMapTest extends JSR166TestCase { return d.acquireUsing(key, usingValue); } + @Override + public SegmentLock acquireUsingReadLocked(K key, V usingValue) { + throw new UnsupportedOperationException(); + } + + @Override + public SegmentLock acquireUsingWriteLocked(K key, V usingValue) { + throw new UnsupportedOperationException(); + } + @Override public V putIfAbsent(K key, V value) { return d.putIfAbsent(key, value);