diff --git a/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java b/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java index 0ce3d5567..eebe5cc7a 100644 --- a/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java +++ b/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java @@ -387,6 +387,11 @@ final class ReplicatedChronicleMap.Segment segment, long pos) { + onRelocation(segment, pos); // erase modification bits + } + @Override void onRemove(VanillaChronicleMap.Segment segment, long pos) { for (long next = modIterSet.nextSetBit(0L); next > 0L; @@ -399,6 +404,11 @@ final class ReplicatedChronicleMap.Segment segment, long pos) { + onRelocation(segment, pos); // erase modification bits + } + @Override void onRelocation(VanillaChronicleMap.Segment segment, long pos) { for (long next = modIterSet.nextSetBit(0L); next > 0L; @@ -931,8 +941,7 @@ final class ReplicatedChronicleMap, // do nothing } + void onRemoteRemove(Segment segment, long pos) { + // do nothing + } + void onPut(Segment segment, long pos) { // do nothing } + void onRemotePut(Segment segment, long pos) { + // do nothing + } + void onRelocation(Segment segment, long pos) { // do nothing } @@ -1940,8 +1948,7 @@ class VanillaChronicleMap, metaValueInterop, valueInterop, value, valueSize, searchedHashLookup); // put callbacks - if (!remote) - onPut(this, segmentState.pos); + onPutMaybeRemote(segmentState.pos, remote); if (bytesEventListener != null) bytesEventListener.onPut(entry, 0L, metaDataBytes, valueSizePos, false); if (eventListener != null) { @@ -1952,6 +1959,14 @@ class VanillaChronicleMap, return resultUnused ? null : prevValue; } + final void onPutMaybeRemote(long pos, boolean remote) { + if (remote) { + onRemotePut(this, pos); + } else { + onPut(this, pos); + } + } + /** * Returns value size, writes the entry (key, value, sizes) to the entry, after this method * call entry positioned after value bytes written (i. e. at the end of entry), sets entry @@ -2169,8 +2184,7 @@ class VanillaChronicleMap, decrementSize(); // remove callbacks - if (!remote) - onRemove(this, pos); + onRemoveMaybeRemote(pos, remote); if (bytesEventListener != null) bytesEventListener.onRemove(entry, 0L, metaDataBytes, valueSizePos); if (eventListener != null) { @@ -2183,6 +2197,14 @@ class VanillaChronicleMap, return booleanResult ? Boolean.TRUE : removedValue; } + final void onRemoveMaybeRemote(long pos, boolean remote) { + if (remote) { + onRemoteRemove(this, pos); + } else { + onRemove(this, pos); + } + } + private > boolean containsKey(ThreadLocalCopies copies, MKBI metaKeyInterop, KBI keyInterop, KB key, long keySize, long hash2) { @@ -2337,15 +2359,15 @@ class VanillaChronicleMap, MetaBytesWriter metaElemWriter, EW elemWriter, E newElem, long newElemSize, MultiMap searchedHashLookup) { - long valueSizeAddr = entry.address() + valueSizePos; + long entryStartAddr = entry.address(); + long valueSizeAddr = entryStartAddr + valueSizePos; long newValueAddr = alignment.alignAddr( valueSizeAddr + valueSizeMarshaller.sizeEncodingSize(newElemSize)); long newEntryEndAddr = newValueAddr + newElemSize; - long entryStartAddr = entry.address(); - long oldEntrySize = entryEndAddr - entryStartAddr; - int oldSizeInBlocks = inBlocks(oldEntrySize); newValueDoesNotFit: if (newEntryEndAddr != entryEndAddr) { + long oldEntrySize = entryEndAddr - entryStartAddr; + int oldSizeInBlocks = inBlocks(oldEntrySize); int newSizeInBlocks = inBlocks(newEntryEndAddr - entryStartAddr); if (newSizeInBlocks > oldSizeInBlocks) { if (newSizeInBlocks > MAX_ENTRY_OVERSIZE_FACTOR) { diff --git a/src/test/java/net/openhft/chronicle/map/TCPSocketReplicationIntValueTest.java b/src/test/java/net/openhft/chronicle/map/TCPSocketReplicationIntValueTest.java index 6a907c631..fc45a19ed 100644 --- a/src/test/java/net/openhft/chronicle/map/TCPSocketReplicationIntValueTest.java +++ b/src/test/java/net/openhft/chronicle/map/TCPSocketReplicationIntValueTest.java @@ -161,14 +161,13 @@ public class TCPSocketReplicationIntValueTest { for (; t < timeOutMs; t++) { if (map1.equals(map2)) break; - Thread.sleep(25); + Thread.sleep(1); } assertEquals(map1, map2); } // TODO test this with larger sizes. @Test(timeout = 12000) - @Ignore("Doesn't work, maps need to check equality.") public void testSoakTestWithRandomData() throws IOException, InterruptedException { final long start = System.currentTimeMillis(); System.out.print("SoakTesting "); diff --git a/src/test/java/net/openhft/chronicle/map/TcpReplicationSoakTest.java b/src/test/java/net/openhft/chronicle/map/TcpReplicationSoakTest.java index 825f7b2da..14dd1dcfe 100644 --- a/src/test/java/net/openhft/chronicle/map/TcpReplicationSoakTest.java +++ b/src/test/java/net/openhft/chronicle/map/TcpReplicationSoakTest.java @@ -12,6 +12,7 @@ import java.io.IOException; import java.net.InetSocketAddress; import java.nio.ByteBuffer; import java.util.*; +import java.util.concurrent.TimeUnit; /** * @author Rob Austin. @@ -33,7 +34,8 @@ public class TcpReplicationSoakTest { { final TcpTransportAndNetworkConfig tcpConfig1 = TcpTransportAndNetworkConfig.of(s_port, - endpoint).autoReconnectedUponDroppedConnection(true).name(" map1"); + endpoint).autoReconnectedUponDroppedConnection(true).name(" map1") + .heartBeatInterval(1, TimeUnit.SECONDS); map1 = ChronicleMapBuilder.of(Integer.class, CharSequence.class) @@ -45,7 +47,8 @@ public class TcpReplicationSoakTest { } { final TcpTransportAndNetworkConfig tcpConfig2 = TcpTransportAndNetworkConfig.of - (s_port + 1).autoReconnectedUponDroppedConnection(true).name("map2"); + (s_port + 1).autoReconnectedUponDroppedConnection(true).name("map2") + .heartBeatInterval(1, TimeUnit.SECONDS); map2 = ChronicleMapBuilder.of(Integer.class, CharSequence.class) .entries(Builder.SIZE + Builder.SIZE) @@ -86,7 +89,6 @@ public class TcpReplicationSoakTest { @Test - @Ignore public void testSoakTestWithRandomData() throws IOException, InterruptedException { System.out.print("SoakTesting "); @@ -130,7 +132,7 @@ public class TcpReplicationSoakTest { map1UnChanged = new HashMap(map1); map2UnChanged = new HashMap(map2); } - Thread.sleep(500); + Thread.sleep(1); if (numberOfTimesTheSame == 100) { System.out.println("same"); break;