From 40e7b1d53cc660244624e58d116dc30ec0e5d269 Mon Sep 17 00:00:00 2001 From: Rob Austin Date: Mon, 27 Apr 2015 13:06:17 +0100 Subject: [PATCH] HCOLL-325 Issue when bootstrapping with a large number of entries, and removed a few @Ignores on tests that should work. --- .../chronicle/map/ReplicatedChronicleMap.java | 2 +- .../openhft/chronicle/map/TcpReplicator.java | 24 ++++---- .../chronicle/map/BooleanValuesTest.java | 1 - .../chronicle/map/CHMUseCasesTest.java | 4 +- .../chronicle/map/ChronicleMapTest.java | 1 - .../chronicle/map/TcpReplicationSoakTest.java | 1 - .../chronicle/map/TestReplication.java | 59 +++++++++++++++++++ 7 files changed, 73 insertions(+), 19 deletions(-) create mode 100644 src/test/java/net/openhft/chronicle/map/TestReplication.java diff --git a/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java b/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java index 663992ed4..85b7e9ca7 100644 --- a/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java +++ b/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java @@ -1401,7 +1401,7 @@ final class ReplicatedChronicleMap extends AbstractChannelReplicator implements Clo entryWriter.entriesToBuffer(attached.remoteModificationIterator, key); try { - final int len = entryWriter.writeBufferToSocket(socketChannel, - approxTime); + final int len = entryWriter.writeBufferToSocket(socketChannel, approxTime); if (len == -1) socketChannel.close(); @@ -737,8 +736,11 @@ final class TcpReplicator extends AbstractChannelReplicator implements Clo if (len > 0) contemplateThrottleWrites(len); - if (!entryWriter.hasBytesToWrite() && !entryWriter.isWorkIncomplete()) - // TURN OP_WRITE_OFF + if (!entryWriter.hasBytesToWrite() + && !entryWriter.isWorkIncomplete() + && hasNext(attached)) + // if we have no more data to write to the socket then we will + // un-register OP_WRITE on the selector, until more data becomes available key.interestOps(key.interestOps() & ~OP_WRITE); } catch (IOException e) { quietClose(key, e); @@ -748,6 +750,11 @@ final class TcpReplicator extends AbstractChannelReplicator implements Clo } } + private boolean hasNext(Attached attached) { + return attached.remoteModificationIterator != null + && !attached.remoteModificationIterator.hasNext(); + } + /** * called when the selector receives a OP_READ message */ @@ -1157,15 +1164,6 @@ final class TcpReplicator extends AbstractChannelReplicator implements Clo if (entrySize > largestEntrySoFar) largestEntrySoFar = entrySize; - - if (!modificationIterator.hasNext()) { - // if we have no more data to write to the socket then we will - // un-register OP_WRITE on the selector, until more data becomes available - if (in().position() == 0 && handShakingComplete) - disableWrite(socketChannel, attached); - - } - // we've filled up the buffer lets give another channel a chance to send // some data if (in().remaining() <= largestEntrySoFar || in().position() > diff --git a/src/test/java/net/openhft/chronicle/map/BooleanValuesTest.java b/src/test/java/net/openhft/chronicle/map/BooleanValuesTest.java index cbc38e6ed..216d74f3e 100644 --- a/src/test/java/net/openhft/chronicle/map/BooleanValuesTest.java +++ b/src/test/java/net/openhft/chronicle/map/BooleanValuesTest.java @@ -33,7 +33,6 @@ public class BooleanValuesTest { * see issue http://stackoverflow.com/questions/26219313/strange-npe-from-chronicle-map-toy-code */ @Test - @Ignore public void testTestBooleanValues() throws IOException, InterruptedException { try (ChronicleMap map = ChronicleMapBuilder.of(Integer.class, Boolean.class).create()) { diff --git a/src/test/java/net/openhft/chronicle/map/CHMUseCasesTest.java b/src/test/java/net/openhft/chronicle/map/CHMUseCasesTest.java index f49508edf..b579185b2 100644 --- a/src/test/java/net/openhft/chronicle/map/CHMUseCasesTest.java +++ b/src/test/java/net/openhft/chronicle/map/CHMUseCasesTest.java @@ -415,7 +415,7 @@ public class CHMUseCasesTest { } - @Ignore + @Ignore @Test public void testByteArrayArrayValue() throws ExecutionException, InterruptedException, IOException { @@ -2538,7 +2538,7 @@ public class CHMUseCasesTest { } } - @Ignore + @Test public void testMapStringIntegerValue() throws IOException { diff --git a/src/test/java/net/openhft/chronicle/map/ChronicleMapTest.java b/src/test/java/net/openhft/chronicle/map/ChronicleMapTest.java index f17bfcf43..cee2b57e7 100755 --- a/src/test/java/net/openhft/chronicle/map/ChronicleMapTest.java +++ b/src/test/java/net/openhft/chronicle/map/ChronicleMapTest.java @@ -43,7 +43,6 @@ import static net.openhft.chronicle.map.StatelessClientTest.localClient; import static org.junit.Assert.*; @SuppressWarnings({"unchecked", "ResultOfMethodCallIgnored"}) -@Ignore public class ChronicleMapTest { static long count = 0; diff --git a/src/test/java/net/openhft/chronicle/map/TcpReplicationSoakTest.java b/src/test/java/net/openhft/chronicle/map/TcpReplicationSoakTest.java index 13a4bbf25..17efee1c5 100644 --- a/src/test/java/net/openhft/chronicle/map/TcpReplicationSoakTest.java +++ b/src/test/java/net/openhft/chronicle/map/TcpReplicationSoakTest.java @@ -20,7 +20,6 @@ import java.util.concurrent.TimeUnit; /** * @author Rob Austin. */ -@Ignore public class TcpReplicationSoakTest { diff --git a/src/test/java/net/openhft/chronicle/map/TestReplication.java b/src/test/java/net/openhft/chronicle/map/TestReplication.java new file mode 100644 index 000000000..65b918806 --- /dev/null +++ b/src/test/java/net/openhft/chronicle/map/TestReplication.java @@ -0,0 +1,59 @@ +package net.openhft.chronicle.map; + +import net.openhft.chronicle.hash.replication.TcpTransportAndNetworkConfig; +import net.openhft.chronicle.map.ChronicleMap; +import net.openhft.chronicle.map.ChronicleMapBuilder; +import org.junit.Assert; +import org.junit.Test; + +import java.net.InetSocketAddress; + +/** + * Created by Rob Austin + */ +public class TestReplication { + + + @Test + public void testAllDataGetsReplicated() throws InterruptedException { + + + TcpTransportAndNetworkConfig tcpConfigServer1 = + TcpTransportAndNetworkConfig.of(8082); + + TcpTransportAndNetworkConfig tcpConfigServer2 = + TcpTransportAndNetworkConfig.of(8083, new InetSocketAddress("localhost", + 8082)); + + final ChronicleMap map2 = ChronicleMapBuilder.of(Integer.class, + Integer.class) + .replication((byte) 2, tcpConfigServer2) + .create(); + + + final ChronicleMap map1 = ChronicleMapBuilder.of(Integer.class, + Integer.class) + .replication((byte) 1, tcpConfigServer1) + .create(); + + + for (int i = 0; i < 7000; i++) { + map1.put(i, i); + } + + + for (int i = 0; i < 10; i++) { + + Thread.sleep(100); + System.out.println(map2.size()); + } + + + Assert.assertEquals(map1.size(),map2.size()); + + } + + + + +}