HCOLL-325 Issue when bootstrapping with a large number of entries, and removed a few @Ignores on tests that should work.

This commit is contained in:
Rob Austin
2015-04-27 13:06:17 +01:00
parent b35a7581fd
commit 40e7b1d53c
7 changed files with 73 additions and 19 deletions
@@ -1401,7 +1401,7 @@ final class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? supe
new EntryModifiableCallback();
// records the current position of the cursor in the bitset
private long position = -1L;
private volatile long position = -1L;
// todo get rid of this
private final MultiStoreBytes tmpBytes = new MultiStoreBytes();
@@ -728,8 +728,7 @@ final class TcpReplicator<K, V> 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<K, V> 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<K, V> 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<K, V> 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() >
@@ -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<Integer, Boolean> map =
ChronicleMapBuilder.of(Integer.class, Boolean.class).create()) {
@@ -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 {
@@ -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;
@@ -20,7 +20,6 @@ import java.util.concurrent.TimeUnit;
/**
* @author Rob Austin.
*/
@Ignore
public class TcpReplicationSoakTest {
@@ -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<Integer, Integer> map2 = ChronicleMapBuilder.of(Integer.class,
Integer.class)
.replication((byte) 2, tcpConfigServer2)
.create();
final ChronicleMap<Integer, Integer> 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());
}
}