diff --git a/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java b/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java index 4d899f503..f9e3f1989 100644 --- a/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java +++ b/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java @@ -23,7 +23,6 @@ import net.openhft.chronicle.hash.replication.TimeProvider; import net.openhft.chronicle.hash.serialization.BytesReader; import net.openhft.chronicle.hash.serialization.internal.MetaBytesInterop; import net.openhft.chronicle.hash.serialization.internal.MetaBytesWriter; -import net.openhft.chronicle.map.MultiMap; import net.openhft.lang.Maths; import net.openhft.lang.collection.ATSDirectBitSet; import net.openhft.lang.io.Bytes; @@ -291,9 +290,12 @@ final class ReplicatedChronicleMap, ? segment.writeLock() : segment.readLock(); + boolean dirtyReplicationIfKeyPresent = lock instanceof VanillaChronicleMap.Segment.NativeWriteLocked; + V v = segment.acquireWithoutLock(copies, metaKeyWriter, keyWriter, key, usingValue, - segmentHash, create, create ? timeProvider.currentTimeMillis() : 0L); + segmentHash, create, create ? timeProvider.currentTimeMillis() : 0L, + dirtyReplicationIfKeyPresent); checkReallyUsingValue(usingValue, mustReuseValue, v); @@ -650,13 +652,23 @@ final class ReplicatedChronicleMap, } V acquireWithoutLock(ThreadLocalCopies copies, MKI metaKeyInterop, KI keyInterop, K key, V usingValue, - long hash2, boolean create, long timestamp) { + long hash2, boolean create, long timestamp, final boolean + dirtyReplicationIfKeyPresent) { 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); + + V v = onKeyPresentOnAcquire(copies, key, usingValue, offset, entry); + + // require for acquireUsingLocked() when working with native + if (dirtyReplicationIfKeyPresent) + modificationDelegator.dirty(ReplicatedChronicleMap.this, posFromOffset(offset), this); + + return v; + } else { if (!create) return null; @@ -1400,17 +1412,22 @@ final class ReplicatedChronicleMap, LOG.error("", e); } + dirty(map, pos, segment); + } + + /** + * used to dirty the entry for replication + */ + void dirty(ChronicleMap map, long pos, SharedSegment segment) { for (long next = modIterSet.nextSetBit(0); next > 0; next = modIterSet.nextSetBit(next + 1)) { try { final ModificationIterator modificationIterator = modificationIterators.get((int) next); - modificationIterator.onPut(map, entry, metaDataBytes, - added, key, replacedValue, value, pos, segment); + modificationIterator.onPut(map, null, 0, + false, null, null, null, pos, segment); } catch (Exception e) { LOG.error("", e); } } - - } @Override diff --git a/src/main/java/net/openhft/chronicle/map/WriteContext.java b/src/main/java/net/openhft/chronicle/map/WriteContext.java index 275341eb3..9b3537ace 100644 --- a/src/main/java/net/openhft/chronicle/map/WriteContext.java +++ b/src/main/java/net/openhft/chronicle/map/WriteContext.java @@ -21,7 +21,19 @@ package net.openhft.chronicle.map; * @author Rob Austin. */ public interface WriteContext extends Context { + + /** + * Prevents the entry being put back back into the map when the Context is closed + * + * @throws java.lang.IllegalArgumentException if used in a native context + */ void dontPutOnClose(); + + + /** + * this is simular by more efficient than calling map.remove("key") as the entry is already available to + * the WriteContext + */ void removeEntry(); } diff --git a/src/test/java/net/openhft/chronicle/map/ChronicleMapTest.java b/src/test/java/net/openhft/chronicle/map/ChronicleMapTest.java index 34f07812c..314fd8b79 100644 --- a/src/test/java/net/openhft/chronicle/map/ChronicleMapTest.java +++ b/src/test/java/net/openhft/chronicle/map/ChronicleMapTest.java @@ -529,12 +529,6 @@ public class ChronicleMapTest { map.close(); } - - @Test - public void broken() { - fail(); - } - @Test public void testAcquireFromMultipleThreads() throws Exception { int entries = 1000 * 1000; diff --git a/src/test/java/net/openhft/chronicle/map/TCPSocketReplicationBootStrapTests.java b/src/test/java/net/openhft/chronicle/map/TCPSocketReplicationBootStrapTests.java index bad095b27..843431d6d 100644 --- a/src/test/java/net/openhft/chronicle/map/TCPSocketReplicationBootStrapTests.java +++ b/src/test/java/net/openhft/chronicle/map/TCPSocketReplicationBootStrapTests.java @@ -86,9 +86,9 @@ public class TCPSocketReplicationBootStrapTests { @Test public void testBootstrapAndHeartbeat() throws IOException, InterruptedException { - map1 = newTcpSocketShmIntString((byte) 1, 8079, new InetSocketAddress("localhost", 8076)); + map1 = newTcpSocketShmIntString((byte) 1, 8080, new InetSocketAddress("localhost", 8081)); ChronicleMapBuilder map2aBuilder = - newTcpSocketShmBuilder(Integer.class, CharSequence.class, (byte) 2, 8076); + newTcpSocketShmBuilder(Integer.class, CharSequence.class, (byte) 2, 8081); final ReplicatedChronicleMap map2a =