From 70e8462d86ca847255cd2c3fc5d1d57b929c6d80 Mon Sep 17 00:00:00 2001 From: Jaromir Hamala Date: Fri, 5 Dec 2014 14:05:41 +0200 Subject: [PATCH 1/5] Range checking fixed --- src/main/java/net/openhft/chronicle/map/NodeDiscovery.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/main/java/net/openhft/chronicle/map/NodeDiscovery.java b/src/main/java/net/openhft/chronicle/map/NodeDiscovery.java index 8077e8046..a3c995f63 100644 --- a/src/main/java/net/openhft/chronicle/map/NodeDiscovery.java +++ b/src/main/java/net/openhft/chronicle/map/NodeDiscovery.java @@ -326,7 +326,7 @@ public class NodeDiscovery { "going the grid."); } - if (possible == 128) + if (possible == Byte.MAX_VALUE) possible = 0; else possible++; From 26e2200f22d6cd93f60db11c42643a1b306d58af Mon Sep 17 00:00:00 2001 From: Rob Austin Date: Fri, 5 Dec 2014 15:45:48 +0000 Subject: [PATCH 2/5] fixed compile issue --- .../openhft/chronicle/map/NodeDiscovery.java | 21 ++++++++++--------- 1 file changed, 11 insertions(+), 10 deletions(-) diff --git a/src/main/java/net/openhft/chronicle/map/NodeDiscovery.java b/src/main/java/net/openhft/chronicle/map/NodeDiscovery.java index 8077e8046..dcb5af770 100644 --- a/src/main/java/net/openhft/chronicle/map/NodeDiscovery.java +++ b/src/main/java/net/openhft/chronicle/map/NodeDiscovery.java @@ -273,8 +273,8 @@ public class NodeDiscovery { } /** - * bitwise OR's the two bit sets or put another way, merges the source bitset into the destination bitset - * and returns the destination + * bitwise OR's the two bit sets or put another way, merges the source bitset into the + * destination bitset and returns the destination * * @param source to or * @param destination to or @@ -336,8 +336,8 @@ public class NodeDiscovery { } /** - * Broadcast the nodes host ports and identifiers over UDP, to make it easy to join a grid of remote nodes - * just by name, this functionality requires UDP + * Broadcast the nodes host ports and identifiers over UDP, to make it easy to join a grid of remote + * nodes just by name, this functionality requires UDP * * @author Rob Austin. */ @@ -604,10 +604,11 @@ class AddressAndPort implements Comparable, BytesMarshallable { return port; } + @Override public boolean equals(Object o) { if (this == o) return true; - if (o == null || getClass() != o.getClass()) return false; + if (o == null || AddressAndPort.class != o.getClass()) return false; AddressAndPort that = (AddressAndPort) o; @@ -728,8 +729,8 @@ class DiscoveryNodeBytesMarshallable implements BytesMarshallable { } /** - * this is used to tell nodes that are connecting to us which host and ports are in our grid, along with - * all the identifiers. + * this is used to tell nodes that are connecting to us which host and ports are in our grid, + * along with all the identifiers. */ @Override public void writeMarshallable(@net.openhft.lang.model.constraints.NotNull Bytes out) { @@ -887,7 +888,7 @@ class DiscoveryNodeBytesMarshallable implements BytesMarshallable { @Override public boolean equals(Object o) { if (this == o) return true; - if (o == null || getClass() != o.getClass()) return false; + if (o == null || ProposedNodes.class != o.getClass()) return false; ProposedNodes that = (ProposedNodes) o; @@ -906,8 +907,8 @@ class DiscoveryNodeBytesMarshallable implements BytesMarshallable { } /** - * sends a bootstrap message to the other nodes in the grid, the bootstrap message contains the host:port - * and perhaps even proposed identifier of the node that sent it. + * sends a bootstrap message to the other nodes in the grid, the bootstrap message contains the + * host:port and perhaps even proposed identifier of the node that sent it. * * @param proposedNodes */ From 0d81d29daee1d1ddc7cfa3a331c24c4bd546bf19 Mon Sep 17 00:00:00 2001 From: Peter Lawrey Date: Fri, 5 Dec 2014 17:54:01 +0200 Subject: [PATCH 3/5] Adding multiple clients to configurable test --- .../openhft/chronicle/map/TcpReplicator.java | 29 +- src/test/java/eg/BGChronicleTest.java | 273 +++++++++--------- .../map/example/DistributedSequenceMain.java | 32 +- 3 files changed, 188 insertions(+), 146 deletions(-) diff --git a/src/main/java/net/openhft/chronicle/map/TcpReplicator.java b/src/main/java/net/openhft/chronicle/map/TcpReplicator.java index 33e221791..b02c48739 100644 --- a/src/main/java/net/openhft/chronicle/map/TcpReplicator.java +++ b/src/main/java/net/openhft/chronicle/map/TcpReplicator.java @@ -449,6 +449,7 @@ final class TcpReplicator extends AbstractChannelReplicator implements Closeable attached.entryReader = new TcpSocketChannelEntryReader(); attached.entryWriter = new TcpSocketChannelEntryWriter(); + attached.writeBarrier = true; key.interestOps(OP_WRITE | OP_READ); @@ -487,15 +488,16 @@ final class TcpReplicator extends AbstractChannelReplicator implements Closeable channel.socket().setSoLinger(false, 0); final Attached attached = new Attached(); - channel.register(selector, OP_READ, attached); - - throttle(channel); - attached.entryReader = new TcpSocketChannelEntryReader(); attached.entryWriter = new TcpSocketChannelEntryWriter(); - attached.isServer = true; attached.entryWriter.identifierToBuffer(localIdentifier); + attached.isServer = true; + attached.writeBarrier = true; + + channel.register(selector, OP_READ, attached); + + throttle(channel); } /** @@ -622,17 +624,19 @@ final class TcpReplicator extends AbstractChannelReplicator implements Closeable final long approxTime) throws InterruptedException, IOException { final SocketChannel socketChannel = (SocketChannel) key.channel(); final Attached attached = (Attached) key.attachment(); - - if (attached.entryWriter.isWorkIncomplete()) { - final boolean completed = attached.entryWriter.doWork(); + if (attached == null) throw new NullPointerException("No attached"); + TcpSocketChannelEntryWriter entryWriter = attached.entryWriter; + if (entryWriter == null) throw new NullPointerException("No entryWriter"); + if (entryWriter.isWorkIncomplete()) { + final boolean completed = entryWriter.doWork(); if (completed) - attached.entryWriter.workCompleted(); + entryWriter.workCompleted(); } else if (attached.remoteModificationIterator != null) - attached.entryWriter.entriesToBuffer(attached.remoteModificationIterator, key); + entryWriter.entriesToBuffer(attached.remoteModificationIterator, key); try { - final int len = attached.entryWriter.writeBufferToSocket(socketChannel, + final int len = entryWriter.writeBufferToSocket(socketChannel, approxTime); if (len == -1) @@ -641,7 +645,7 @@ final class TcpReplicator extends AbstractChannelReplicator implements Closeable if (len > 0) contemplateThrottleWrites(len); - if (!attached.entryWriter.hasBytesToWrite() && !attached.entryWriter.isWorkIncomplete()) + if (!entryWriter.hasBytesToWrite() && !entryWriter.isWorkIncomplete()) // TURN OP_WRITE_OFF key.interestOps(key.interestOps() & ~OP_WRITE); } catch (IOException e) { @@ -903,6 +907,7 @@ final class TcpReplicator extends AbstractChannelReplicator implements Closeable public boolean isServer; // the frequency the remote node will send a heartbeat public boolean handShakingComplete; public long remoteHeartbeatInterval = heartBeatIntervalMillis; + public volatile boolean writeBarrier; boolean isHandShakingComplete() { return handShakingComplete; diff --git a/src/test/java/eg/BGChronicleTest.java b/src/test/java/eg/BGChronicleTest.java index df831891b..81a76a60a 100644 --- a/src/test/java/eg/BGChronicleTest.java +++ b/src/test/java/eg/BGChronicleTest.java @@ -5,86 +5,123 @@ import net.openhft.chronicle.map.ChronicleMap; import net.openhft.chronicle.map.ChronicleMapBuilder; import org.junit.Test; +import java.io.File; import java.io.IOException; import java.net.InetSocketAddress; +import java.nio.ByteBuffer; +import java.util.ArrayList; +import java.util.List; import java.util.Random; -import java.util.concurrent.TimeUnit; +import java.util.concurrent.*; + +import static java.lang.System.nanoTime; +import static java.nio.ByteBuffer.wrap; public class BGChronicleTest { private static final int CHRONICLE_HEARTBEAT_SECONDS = 5; - public static final long MAX_NUMBER_OF_EATERIES = 3000000L; + public static final int CLIENTS = Integer.getInteger("clients", 10); + public static final long KEYS = Long.getLong("keys", 100000L); + public static final long MESSAGES = Long.getLong("messages", 200000L); public static final String DEFAULT_HOST = "localhost"; public static final int DEFAULT_PORT = 9050; - - Random rnd = new Random(System.currentTimeMillis()); - byte[] key = new byte[17]; - byte[] value = new byte[11]; + public static final int KEY_SIZE = Integer.getInteger("keySize", 16); + public static final int VALUE_SIZE = Integer.getInteger("valueSize", 16); + private volatile long start; public ChronicleMap startChronicleMapServer(int port) throws InterruptedException, IOException { TcpTransportAndNetworkConfig serverConfig = TcpTransportAndNetworkConfig.of(port) - .heartBeatInterval(CHRONICLE_HEARTBEAT_SECONDS, TimeUnit.SECONDS).name("serverMap") + .heartBeatInterval(CHRONICLE_HEARTBEAT_SECONDS, TimeUnit.SECONDS) + .name("serverMap") .autoReconnectedUponDroppedConnection(true); - + File file = File.createTempFile("bg-test", ".deleteme"); + file.deleteOnExit(); return ChronicleMapBuilder.of(byte[].class, byte[].class) - .entries(MAX_NUMBER_OF_EATERIES) - .constantKeySizeBySample(key) - .constantValueSizeBySample(value) + .entries(KEYS) + .constantKeySizeBySample(new byte[KEY_SIZE]) + .constantValueSizeBySample(new byte[VALUE_SIZE]) .putReturnsNull(true) .removeReturnsNull(true) .replication((byte) 1, serverConfig) - .create(); + .createPersistedTo(file); } /** - * over loop-back on my mac I get around 50,000 messages per second of my local network ( LAN ), - * I get around 730 messages per second - * - * @param hostNameOrIpAddress - * @param port - * @param numberOfMessages - * @throws IOException + * over loop-back on my mac I get around 50,000 messages per second of my local network ( LAN ) */ - public void startChronicleMapClient(String hostNameOrIpAddress, final int port, final long - numberOfMessages) throws IOException { + public void startChronicleMapClient(String hostNameOrIpAddress, int port) throws IOException, InterruptedException { System.out.println("client starting"); + final InetSocketAddress remoteAddress = new InetSocketAddress(hostNameOrIpAddress, port); + start = nanoTime(); - try (ChronicleMap map = ChronicleMapBuilder.of(byte[].class, byte[].class) - .constantKeySizeBySample(key) - .constantValueSizeBySample(value) - .putReturnsNull(true) - .removeReturnsNull(true) - .statelessClient(new InetSocketAddress(hostNameOrIpAddress, port)).create()) { + ExecutorService es = Executors.newFixedThreadPool(CLIENTS); + List> futureList = new ArrayList<>(); + for (int i = 0; i < CLIENTS; i++) { + final int clientId = i; + futureList.add(es.submit(new Callable() { + @Override + public Void call() throws IOException { + try (ChronicleMap map = ChronicleMapBuilder + .of(byte[].class, byte[].class) + .constantKeySizeBySample(new byte[KEY_SIZE]) + .constantValueSizeBySample(new byte[VALUE_SIZE]) + .putReturnsNull(true) + .removeReturnsNull(true) + .statelessClient(remoteAddress) + .create()) { - System.out.println("client started"); + System.out.println(clientId + "/" + CLIENTS + " client started"); - long start = System.nanoTime(); - for (long count = 0; count < numberOfMessages; count++) { + exerciseMap(MESSAGES, map, clientId); + System.out.println(clientId + "/" + CLIENTS + " ... client finished"); + } + return null; + } + })); + } + for (Future future : futureList) { + try { + future.get(); + } catch (ExecutionException e) { + e.printStackTrace(); + } + } + es.shutdown(); + logMessagesPerSecond(MESSAGES, start); + } + + public void exerciseMap(long numberOfMessages, ChronicleMap map, int clientId) { + byte[] key = new byte[KEY_SIZE]; + byte[] value = new byte[VALUE_SIZE]; + Random rnd = new Random(); + ByteBuffer keyBB = wrap(key); + for (long count = 0; count < numberOfMessages; count += CLIENTS * 100) { + // KEYS/100 possible values after first byte. + keyBB.putInt(1, rnd.nextInt((int) (KEYS / 100))); + rnd.nextBytes(value); + for (byte b = 0; b < 100; b++) { + // 100 possible values for the first byte. + key[0] = b; if (rnd.nextBoolean()) { // puts - rnd.nextBytes(key); - rnd.nextBytes(value); + value[0] = b; map.put(key, value); } else { //gets - rnd.nextBytes(key); map.get(key); } - - if (count % 50000 == 0) { - logMessagesPerSecond(count, start); - } - } - logMessagesPerSecond(numberOfMessages, start); + if (clientId == 0 && count > 0 && count % 50000 == 0) { + logMessagesPerSecond(count, start); + } } } @@ -92,72 +129,57 @@ public class BGChronicleTest { /** * over loop-back on my mac I get around 50,000 messages per second of my local network ( LAN ), - * I get around 730 messages per second - * - * @param hostNameOrIpAddress - * @param port - * @param numberOfMessages - * @throws IOException */ - public void startReplicatingChronicleMapClient(String hostNameOrIpAddress, final int port, - final long numberOfMessages) throws - IOException { + public void startReplicatingChronicleMapClient(String hostNameOrIpAddress, final int port) + throws IOException, InterruptedException { - System.out.println("client starting"); + System.out.println("clients starting"); final InetSocketAddress serverHostPort = new InetSocketAddress(hostNameOrIpAddress, port); - TcpTransportAndNetworkConfig clientConfig = TcpTransportAndNetworkConfig.of(port + 1, serverHostPort) + final TcpTransportAndNetworkConfig clientConfig = TcpTransportAndNetworkConfig.of(port + 1, serverHostPort) .heartBeatInterval(CHRONICLE_HEARTBEAT_SECONDS, TimeUnit.SECONDS).name("clientMap") .autoReconnectedUponDroppedConnection(true); - try (ChronicleMap map = ChronicleMapBuilder.of(byte[].class, byte[].class) - .entries(MAX_NUMBER_OF_EATERIES) - .keySize(17) - .valueSize(11) - .putReturnsNull(true) - .removeReturnsNull(true) - .replication((byte) 2, clientConfig) - .create()) { + start = nanoTime(); + ExecutorService es = Executors.newFixedThreadPool(CLIENTS); + for (int i = 0; i < CLIENTS; i++) { + final int clientId = i; + es.submit(new Runnable() { + @Override + public void run() { + try (ChronicleMap map = ChronicleMapBuilder.of(byte[].class, byte[].class) + .entries(KEYS) + .constantKeySizeBySample(new byte[KEY_SIZE]) + .constantValueSizeBySample(new byte[VALUE_SIZE]) + .putReturnsNull(true) + .removeReturnsNull(true) + .replication((byte) (2 + clientId), clientConfig) + .create()) { - System.out.println("client started"); + System.out.println(clientId + "/" + CLIENTS + " client started"); - long start = System.nanoTime(); - for (long count = 0; count < numberOfMessages; count++) { - if (rnd.nextBoolean()) { - // puts - rnd.nextBytes(key); - rnd.nextBytes(value); - map.put(key, value); + exerciseMap(MESSAGES, map, clientId); + System.out.println(clientId + "/" + CLIENTS + " ... client finished"); + } - } else { - //gets - rnd.nextBytes(key); - map.get(key); } - - if (count % 50000 == 0) { - logMessagesPerSecond(count, start); - } - - } - - logMessagesPerSecond(numberOfMessages, start); - + }); } + es.awaitTermination(1, TimeUnit.HOURS); + logMessagesPerSecond(MESSAGES, start); } private void logMessagesPerSecond(double numberOfMessages, long start) { - double messagesPerSecond = numberOfMessages / (double) TimeUnit.NANOSECONDS - .toSeconds(System.nanoTime() - start); - System.out.println("messages per seconds " + messagesPerSecond); + long messagesPerSecond = (long) (numberOfMessages * 1e9 / (System.nanoTime() - start)); + System.out.printf("messages per seconds: %,d%n", messagesPerSecond); } /** * for loop back args = [] - * + *

* for server - eg.BGChronicleTest server for client - eg.BGChronicleTest client * * @param args @@ -167,60 +189,45 @@ public class BGChronicleTest { public static void main(String... args) throws IOException, InterruptedException { final BGChronicleTest o = new BGChronicleTest(); - ChronicleMap serverMap = null; - try { - - // loop-back - if (args.length == 0) { - serverMap = o.startChronicleMapServer(DEFAULT_PORT); - o.startChronicleMapClient(DEFAULT_HOST, DEFAULT_PORT, MAX_NUMBER_OF_EATERIES); - - } else { - - // server - if (args.length > 0 && args[0].equalsIgnoreCase("server")) { - int port = getPort(args, 1); - - serverMap = o.startChronicleMapServer(port); - System.out.println("server ON"); - } else - System.out.println("server if OFF"); - - // client - if (args.length == 0 || (args.length > 1 && args[0].equalsIgnoreCase("client"))) { - - // stateless replicating client - String host = getHost(args, 1); - int port = getPort(args, 2); - System.out.println("client ON...."); - o.startChronicleMapClient(host, port, MAX_NUMBER_OF_EATERIES); - - // replicating client - } else if (args.length > 1 && args[0].equalsIgnoreCase("replicatingClient")) { - String host = getHost(args, 1); - int port = getPort(args, 2); - System.out.println("client ON...."); - o.startReplicatingChronicleMapClient(host, port, MAX_NUMBER_OF_EATERIES); - - } else { - System.out.println("client if OFF"); - for (; ; ) { - Thread.sleep(1000); - } - } - - System.out.println("client if OFF"); - for (; ; ) { - Thread.sleep(1000); - } + // loop-back + if (args.length == 0) { + try (ChronicleMap serverMap = o.startChronicleMapServer(DEFAULT_PORT)) { + o.startChronicleMapClient(DEFAULT_HOST, DEFAULT_PORT); } - } finally { - if (serverMap != null) - serverMap.close(); - } + } else switch (args[0].toLowerCase()) { + case "server": { + int port = getPort(args, 1); + o.startChronicleMapServer(port); + System.out.println("server ON"); + + break; + } + + case "client": { + // stateless replicating client + String host = getHost(args, 1); + int port = getPort(args, 2); + System.out.println("client ON...."); + o.startChronicleMapClient(host, port); + break; + } + + case "replicatingClient": { + String host = getHost(args, 1); + int port = getPort(args, 2); + System.out.println("client ON...."); + o.startReplicatingChronicleMapClient(host, port); + break; + } + default: + System.err.println("Usage: java " + BGChronicleTest.class.getName() + " server {port}"); + System.err.println("\tjava " + BGChronicleTest.class.getName() + " client {hostname} {port}"); + System.err.println("\tjava " + BGChronicleTest.class.getName() + " replicatingClient {hostname} {port}"); + break; + } } private static String getHost(String[] args, int index) { diff --git a/src/test/java/net/openhft/chronicle/map/example/DistributedSequenceMain.java b/src/test/java/net/openhft/chronicle/map/example/DistributedSequenceMain.java index aa1b9b7ec..ba2530470 100644 --- a/src/test/java/net/openhft/chronicle/map/example/DistributedSequenceMain.java +++ b/src/test/java/net/openhft/chronicle/map/example/DistributedSequenceMain.java @@ -25,6 +25,7 @@ import net.openhft.lang.values.LongValue; import java.io.File; import java.io.IOException; +import java.util.Arrays; public class DistributedSequenceMain { @@ -35,18 +36,47 @@ public class DistributedSequenceMain { .entries(128) .actualSegments(1) .createPersistedTo(file)) { + // throughput test. for (int t = 0; t < 5; t++) { LongValue value = DataValueClasses.newDirectReference(LongValue.class); map.acquireUsing("sequence-" + t, value); long start = System.nanoTime(); - int runs = 10000000; + int runs = 1000000; for (int i = 0; i < runs; i++) { long nextId = value.addAtomicValue(1); } double rate = (double) (System.nanoTime() - start) / runs; System.out.printf("Uncontended increment took an average of %.1f ns.%n", rate); } + + // latency test. + LongValue value = DataValueClasses.newDirectReference(LongValue.class); + map.acquireUsing("sequence-X", value); + + int runs = 1000000000, count = 0, wcount = 0; + long[] worst = new long[200000]; + long start = System.nanoTime(); + for (; count < runs; count++) { + long nextId = value.addAtomicValue(1); + long next = System.nanoTime(); + long time = next - start; + if (time > 100) { + worst[wcount++] = time; + if (wcount >= worst.length) + break; + next = System.nanoTime(); + } + start = next; + } + Arrays.sort(worst); + System.out.printf("Worst 1 in 10K, 1 in 100K, 1 in 1000K, worst is %,d / %,d / %,d / %,d %n", + worst[wcount - count / 10000], + worst[wcount - count / 100000], + worst[wcount - count / 1000000], + worst[wcount - 1] + ); } } } + From f02be23f09d8432089f9ca7387152e245ba98024 Mon Sep 17 00:00:00 2001 From: Peter Lawrey Date: Fri, 5 Dec 2014 17:56:07 +0200 Subject: [PATCH 4/5] Adding multiple clients to configurable test --- src/main/java/net/openhft/chronicle/map/TcpReplicator.java | 3 --- 1 file changed, 3 deletions(-) diff --git a/src/main/java/net/openhft/chronicle/map/TcpReplicator.java b/src/main/java/net/openhft/chronicle/map/TcpReplicator.java index b02c48739..bb8e4d6e5 100644 --- a/src/main/java/net/openhft/chronicle/map/TcpReplicator.java +++ b/src/main/java/net/openhft/chronicle/map/TcpReplicator.java @@ -449,7 +449,6 @@ final class TcpReplicator extends AbstractChannelReplicator implements Closeable attached.entryReader = new TcpSocketChannelEntryReader(); attached.entryWriter = new TcpSocketChannelEntryWriter(); - attached.writeBarrier = true; key.interestOps(OP_WRITE | OP_READ); @@ -493,7 +492,6 @@ final class TcpReplicator extends AbstractChannelReplicator implements Closeable attached.entryWriter.identifierToBuffer(localIdentifier); attached.isServer = true; - attached.writeBarrier = true; channel.register(selector, OP_READ, attached); @@ -907,7 +905,6 @@ final class TcpReplicator extends AbstractChannelReplicator implements Closeable public boolean isServer; // the frequency the remote node will send a heartbeat public boolean handShakingComplete; public long remoteHeartbeatInterval = heartBeatIntervalMillis; - public volatile boolean writeBarrier; boolean isHandShakingComplete() { return handShakingComplete; From 184d17e811a9dc977b90fd94656656134321b652 Mon Sep 17 00:00:00 2001 From: Rob Austin Date: Fri, 5 Dec 2014 16:33:12 +0000 Subject: [PATCH 5/5] HCOLL-230 - checking in fixes to the multi threaded stateless client, it now works ( kind of ), unless its put under load, test net.openhft.chronicle.map.StatelessClientTest.testThreadSafeness will gradually slow down - so am still looking into this issue further. --- .../chronicle/map/StatelessChronicleMap.java | 410 +++++++++--------- .../openhft/chronicle/map/TcpReplicator.java | 8 + src/test/java/eg/BGChronicleTest.java | 2 + .../chronicle/map/StatelessClientTest.java | 112 ++++- 4 files changed, 329 insertions(+), 203 deletions(-) diff --git a/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java b/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java index 2fd44961f..9380a1c11 100644 --- a/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java +++ b/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java @@ -41,7 +41,7 @@ import java.nio.channels.SocketChannel; import java.util.*; import java.util.concurrent.*; import java.util.concurrent.atomic.AtomicLong; -import java.util.concurrent.locks.ReentrantLock; +import java.util.concurrent.locks.ReentrantReadWriteLock; import static java.nio.ByteBuffer.allocateDirect; import static net.openhft.chronicle.map.AbstractChannelReplicator.SIZE_OF_SIZE; @@ -82,8 +82,8 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon private final boolean putReturnsNull; private final boolean removeReturnsNull; - private final ReentrantLock inBytesLock = new ReentrantLock(true); - private final ReentrantLock outBytesLock = new ReentrantLock(true); + + private final ReentrantReadWriteLock inBytesLock = new ReentrantReadWriteLock(true); private ExecutorService executorService; @@ -204,8 +204,8 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon return executorService; } - private SocketChannel lazyConnect(final long timeoutMs, - final InetSocketAddress remoteAddress) throws IOException { + private synchronized SocketChannel lazyConnect(final long timeoutMs, + final InetSocketAddress remoteAddress) throws IOException { if (LOG.isDebugEnabled()) LOG.debug("attempting to connect to " + remoteAddress); @@ -252,7 +252,7 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon * java.net.InetSocketAddress) */ - private void attemptConnect(final InetSocketAddress remoteAddress) throws IOException { + private synchronized void attemptConnect(final InetSocketAddress remoteAddress) throws IOException { // ensures that the excising connection are closed closeExisting(); @@ -285,7 +285,7 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon * @param clientChannel * @throws java.io.IOException */ - private void doHandShaking(@NotNull final SocketChannel clientChannel) throws IOException { + private synchronized void doHandShaking(@NotNull final SocketChannel clientChannel) throws IOException { connectionByte[0] = STATELESS_CLIENT_IDENTIFIER; this.connectionOutBuffer.clear(); @@ -428,7 +428,7 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon return fetchObject(String.class, TO_STRING); } - public synchronized boolean isEmpty() { + public boolean isEmpty() { return fetchBoolean(IS_EMPTY); } @@ -649,7 +649,7 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon private void resizeBufferInBuffer(int size, long start) { if (LOG.isDebugEnabled()) - LOG.debug("resizing buffer to size=" + size); + LOG.debug("InBuffer resizing buffer to size=" + size); if (size < inBuffer.capacity()) throw new IllegalStateException("it not possible to resize the buffer smaller"); @@ -704,7 +704,7 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon BytesReader valueReader = valueReaderWithSize.readerForLoop(null); for (; ; ) { - inBytesLock.lock(); + inBytesLock.writeLock().lock(); try { final Bytes in = blockingFetchReadOnly(timeoutTime, transactionId); @@ -722,7 +722,7 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon break; } finally { - inBytesLock.unlock(); + inBytesLock.writeLock().unlock(); } } return result; @@ -745,7 +745,7 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon for (; ; ) { - inBytesLock.lock(); + inBytesLock.writeLock().lock(); try { Bytes in = blockingFetchReadOnly(timeoutTime, transactionId); @@ -765,7 +765,7 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon break; } finally { - inBytesLock.unlock(); + inBytesLock.writeLock().unlock(); } } @@ -784,7 +784,7 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon for (; ; ) { - inBytesLock.lock(); + inBytesLock.writeLock().lock(); try { final Bytes in = blockingFetchReadOnly(timeoutTime, transactionId); @@ -801,7 +801,7 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon break; } finally { - inBytesLock.unlock(); + inBytesLock.writeLock().unlock(); } } @@ -813,11 +813,11 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon final long timeoutTime = startTime + this.config.timeoutMs(); // receive - inBytesLock.lock(); + inBytesLock.writeLock().lock(); try { return blockingFetchReadOnly(timeoutTime, transactionId).readLong(); } finally { - inBytesLock.unlock(); + inBytesLock.writeLock().unlock(); } } @@ -940,6 +940,8 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon // send out all the bytes send(outBytes, timeoutTime); + outBytes.clear(); + outBytes.buffer().clear(); break; } catch (java.nio.channels.ClosedChannelException | ClosedConnectionException e) { @@ -961,27 +963,32 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon private Bytes blockingFetchThrowable(long sizeLocation, final long transactionId, final long timeoutTime) throws IOException, InterruptedException { for (; ; ) { - if (clientChannel == null) - clientChannel = lazyConnect(config.timeoutMs(), config.remoteAddress()); + if (clientChannel == null) { + synchronized (this) { + if (clientChannel == null) { + clientChannel = lazyConnect(config.timeoutMs(), config.remoteAddress()); + } + } - try { + try { - if (LOG.isDebugEnabled()) - LOG.debug("sending data with transactionId=" + transactionId); + if (LOG.isDebugEnabled()) + LOG.debug("sending data with transactionId=" + transactionId); - writeSizeAndTransactionIdAt(sizeLocation, transactionId); + writeSizeAndTransactionIdAt(sizeLocation, transactionId); - // send out all the bytes - send(outBytes, timeoutTime); + // send out all the bytes + send(outBytes, timeoutTime); - return blockingFetchThrowable(timeoutTime, transactionId); - } catch (java.nio.channels.ClosedChannelException | ClosedConnectionException e) { - checkTimeout(timeoutTime); - clientChannel = lazyConnect(config.timeoutMs(), config.remoteAddress()); + return blockingFetchThrowable(timeoutTime, transactionId); + } catch (java.nio.channels.ClosedChannelException | ClosedConnectionException e) { + checkTimeout(timeoutTime); + clientChannel = lazyConnect(config.timeoutMs(), config.remoteAddress()); + } } } - } + } private Bytes blockingFetchReadOnly(long timeoutTime, final long transactionId) { try { @@ -995,6 +1002,9 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon } catch (Exception e) { close(); throw new RuntimeException(e); + } catch (AssertionError e) { + LOG.error("", e); + throw e; } } @@ -1003,80 +1013,139 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon private volatile int parkedRemainingBytes; private volatile long parkedTransactionTimeStamp; - ByteBufferBytes sizeAndTransactionIdBuffer = new ByteBufferBytes(ByteBuffer.allocate + ByteBufferBytes buffer = new ByteBufferBytes(ByteBuffer.allocate (SIZE_OF_SIZE + SIZE_OF_TRANSACTION_ID)); + /** + * reads up to the number of byte in {@code requiredNumberOfBytes} + * + * @param requiredNumberOfBytes the number of bytes to read + * @param timeoutTime timeout in milliseconds + * @return bytes read from the TCP/IP socket + * @throws IOException socket failed to read data + */ + private Bytes receiveBuffer(int requiredNumberOfBytes, long timeoutTime) throws IOException { - private synchronized Bytes blockingFetchThrowable(long timeoutTime, long transactionId) throws IOException, + buffer.clear(); + buffer.buffer().clear(); + /*System.out.println(buffer.position()); + System.out.println(buffer.limit()); + */ + synchronized (this) { + while (buffer.buffer().remaining() > 0) { + assert requiredNumberOfBytes <= buffer.capacity(); + int len; + + // you can't read and write from the buffer a the same time + + len = clientChannel.read(buffer.buffer()); + + + // System.out.println(("receiveBuffer = bytes read=" + len)); + + if (len == -1) + throw new IORuntimeException("Disconnected to remote server"); + + checkTimeout(timeoutTime); + + } + } + buffer.clear(); + + // System.out.println("buffer=" + AbstractBytes.toHex(buffer)); + return buffer; + } + + + private Bytes blockingFetchThrowable(long timeoutTime, long transactionId) throws IOException, InterruptedException { int remainingBytes = 0; - for (; ; ) { - // read the next item from the socket - if (parkedTransactionId == 0) { + try { + for (; ; ) { - assert parkedTransactionTimeStamp == 0; - assert parkedRemainingBytes == 0; + // read the next item from the socket + if (parkedTransactionId == 0) { - receive(SIZE_OF_SIZE + SIZE_OF_TRANSACTION_ID, timeoutTime); - - final int messageSize = inBytes.readInt(); - final int remainingBytes0 = messageSize - (SIZE_OF_SIZE + SIZE_OF_TRANSACTION_ID); - final long transactionId0 = inBytes.readLong(); - assert transactionId0 != 0; + assert parkedTransactionTimeStamp == 0; + assert parkedRemainingBytes == 0; - // if the transaction id is not for this thread, park it and read the next one - if (transactionId0 == transactionId) { + receiveBuffer(SIZE_OF_SIZE + SIZE_OF_TRANSACTION_ID, timeoutTime); - // we have the correct transaction id ! - parkedTransactionId = 0; - remainingBytes = remainingBytes0; - assert remainingBytes > 0; + final int messageSize = buffer.readInt(); + final int remainingBytes0 = messageSize - (SIZE_OF_SIZE + SIZE_OF_TRANSACTION_ID); + final long transactionId0 = buffer.readLong(); + assert transactionId0 != 0; + + + // check the transaction id is reasonable + assert String.valueOf(transactionId0).startsWith("14"); + + + // if the transaction id is not for this thread, park it and read the next one + if (transactionId0 == transactionId) { + + // we have the correct transaction id ! + parkedTransactionId = 0; + remainingBytes = remainingBytes0; + assert remainingBytes > 0; + + clearParked(); + + break; + + } else { + + parkedTransactionTimeStamp = System.currentTimeMillis(); + parkedRemainingBytes = remainingBytes0; + parkedTransactionId = transactionId0; + + pause(); + continue; + } + } + + // the transaction id was read by another thread, but it is for this thread + if (parkedTransactionId == transactionId) { + remainingBytes = parkedRemainingBytes; + clearParked(); + break; + } + + + // time out the old transaction id + if (System.currentTimeMillis() - timeoutTime > + parkedTransactionTimeStamp) { + + LOG.error("", new IllegalStateException("Skipped Message with transaction-id=" + + parkedTransactionTimeStamp + + ", this can occur when you have another thread which has called the " + + "stateless client and terminated abruptly before the message has been " + + "returned from the server")); + + // read the the next message + receive(parkedRemainingBytes, timeoutTime); clearParked(); - break; - } else { - - parkedTransactionTimeStamp = System.currentTimeMillis(); - parkedRemainingBytes = remainingBytes0; - parkedTransactionId = transactionId0; - - pause(); - continue; } + + pause(); } - - // the transaction id was read by another thread, but it is for this thread - if (parkedTransactionId == transactionId) { - remainingBytes = parkedRemainingBytes; - clearParked(); - break; - } - - - // time out the old transaction id - if (System.currentTimeMillis() - timeoutTime > - parkedTransactionTimeStamp) { - - LOG.error("", new IllegalStateException("Skipped Message with transaction-id=" + - parkedTransactionTimeStamp + - ", this can occur when you have another thread which has called the " + - "stateless client and terminated abruptly before the message has been " + - "returned from the server")); - parkedTransactionTimeStamp = 0; - parkedTransactionId = 0; - parkedRemainingBytes = 0; - - // read the the next message - receive(remainingBytes, timeoutTime); - continue; - - } - - pause(); + } catch (Exception e) { + parkedTransactionTimeStamp = 0; + parkedTransactionId = 0; + parkedRemainingBytes = 0; + LOG.error("", e); + throw e; + } catch (Error e) { + parkedTransactionTimeStamp = 0; + parkedTransactionId = 0; + parkedRemainingBytes = 0; + LOG.error("", e); + throw e; } @@ -1089,34 +1158,34 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon resizeBufferInBuffer(minBufferSize, pos); } else inBytes.limit(inBytes.capacity()); + synchronized (this) { + // block until we have received all the bytes in this chunk + receive(remainingBytes, timeoutTime); - // block until we have received all the bytes in this chunk - receive(remainingBytes, timeoutTime); + final boolean isException = inBytes.readBoolean(); - final boolean isException = inBytes.readBoolean(); - - if (isException) { - Throwable throwable = (Throwable) inBytes.readObject(); - try { - Field stackTrace = Throwable.class.getDeclaredField("stackTrace"); - stackTrace.setAccessible(true); - List stes = new ArrayList<>(Arrays.asList((StackTraceElement[]) stackTrace.get(throwable))); - // prune the end of the stack. - for (int i = stes.size() - 1; i > 0 && stes.get(i).getClassName().startsWith("Thread"); i--) { - stes.remove(i); + if (isException) { + Throwable throwable = (Throwable) inBytes.readObject(); + try { + Field stackTrace = Throwable.class.getDeclaredField("stackTrace"); + stackTrace.setAccessible(true); + List stes = new ArrayList<>(Arrays.asList((StackTraceElement[]) stackTrace.get(throwable))); + // prune the end of the stack. + for (int i = stes.size() - 1; i > 0 && stes.get(i).getClassName().startsWith("Thread"); i--) { + stes.remove(i); + } + InetSocketAddress address = config.remoteAddress(); + stes.add(new StackTraceElement("~ remote", "tcp ~", address.getHostName(), address.getPort())); + StackTraceElement[] stackTrace2 = Thread.currentThread().getStackTrace(); + for (int i = 4; i < stackTrace2.length; i++) + stes.add(stackTrace2[i]); + stackTrace.set(throwable, stes.toArray(new StackTraceElement[stes.size()])); + } catch (Exception ignore) { } - InetSocketAddress address = config.remoteAddress(); - stes.add(new StackTraceElement("~ remote", "tcp ~", address.getHostName(), address.getPort())); - StackTraceElement[] stackTrace2 = Thread.currentThread().getStackTrace(); - for (int i = 4; i < stackTrace2.length; i++) - stes.add(stackTrace2[i]); - stackTrace.set(throwable, stes.toArray(new StackTraceElement[stes.size()])); - } catch (Exception ignore) { + NativeBytes.UNSAFE.throwException(throwable); } - NativeBytes.UNSAFE.throwException(throwable); + } - - return inBytes; } @@ -1127,8 +1196,8 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon } private void pause() throws InterruptedException { - inBytesLock.unlock(); - inBytesLock.lock(); + inBytesLock.writeLock().unlock(); + inBytesLock.writeLock().lock(); } @@ -1142,10 +1211,6 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon */ private Bytes receive(int requiredNumberOfBytes, long timeoutTime) throws IOException { - - inBytes.position(0); - inBytes.limit(requiredNumberOfBytes); - inBytes.buffer().position(0); inBytes.buffer().limit(requiredNumberOfBytes); @@ -1189,14 +1254,19 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon final long size = outBytes.position() - locationOfSize; final long pos = outBytes.position(); outBytes.position(locationOfSize); - outBytes.writeInt((int) size - SIZE_OF_SIZE); + int size0 = (int) size - SIZE_OF_SIZE; + assert size0 > 0; + + outBytes.writeInt(size0); outBytes.writeLong(transactionId); outBytes.position(pos); } private void writeSizeAt(long locationOfSize) { final long size = outBytes.position() - locationOfSize; - outBytes.writeInt(locationOfSize, (int) size - SIZE_OF_SIZE); + int size0 = (int) size - SIZE_OF_SIZE; + assert size0 > 0; + outBytes.writeInt(locationOfSize, size0); } private long markSizeLocation() { @@ -1292,7 +1362,6 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon int size = Integer.parseInt(substring.substring(0, i)); long requiresExtra = size - outBytes.remaining(); - System.out.println(size + substring); resizeBufferOutBuffer((int) (outBytes.capacity() + requiresExtra), start); } else resizeBufferOutBuffer((int) (outBytes.capacity() + maxEntrySize), start); @@ -1304,11 +1373,11 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon long timeoutTime = startTime + this.config.timeoutMs(); // receive - inBytesLock.lock(); + inBytesLock.writeLock().lock(); try { return readValue(copies, blockingFetchReadOnly(timeoutTime, transactionId)); } finally { - inBytesLock.unlock(); + inBytesLock.writeLock().unlock(); } } @@ -1316,11 +1385,11 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon long timeoutTime = startTime + this.config.timeoutMs(); // receive - inBytesLock.lock(); + inBytesLock.writeLock().lock(); try { return (O) (blockingFetchReadOnly(timeoutTime, transactionId).readObject()); } finally { - inBytesLock.unlock(); + inBytesLock.writeLock().unlock(); } } @@ -1328,11 +1397,11 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon long timeoutTime = startTime + this.config.timeoutMs(); // receive - inBytesLock.lock(); + inBytesLock.writeLock().lock(); try { return blockingFetchReadOnly(timeoutTime, transactionId).readBoolean(); } finally { - inBytesLock.unlock(); + inBytesLock.writeLock().unlock(); } } @@ -1341,11 +1410,11 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon long timeoutTime = startTime + this.config.timeoutMs(); // receive - inBytesLock.lock(); + inBytesLock.writeLock().lock(); try { return blockingFetchReadOnly(timeoutTime, transactionId).readInt(); } finally { - inBytesLock.unlock(); + inBytesLock.writeLock().unlock(); } } @@ -1353,12 +1422,9 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon long transactionId = nextUniqueTransaction(startTime); // send - outBytesLock.lock(); - try { + synchronized (this) { final long sizeLocation = writeEventAnSkip(eventId); transactionId = blockingSend(sizeLocation, startTime); - } finally { - outBytesLock.unlock(); } return transactionId; @@ -1374,8 +1440,7 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon long transactionId; - outBytesLock.lock(); - try { + synchronized (this) { final long sizeLocation = writeEventAnSkip(eventId); @@ -1385,8 +1450,6 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon copies = writeValue(value, copies); } transactionId = blockingSend(sizeLocation, startTime); - } finally { - outBytesLock.unlock(); } @@ -1400,14 +1463,12 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon long transactionId; - outBytesLock.lock(); - try { + synchronized (this) { + final long sizeLocation = writeEventAnSkip(eventId); writeValue(value, null); transactionId = blockingSend(sizeLocation, startTime); - } finally { - outBytesLock.unlock(); } @@ -1421,14 +1482,9 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon long transactionId; - outBytesLock.lock(); - try { - + synchronized (this) { final long sizeLocation = writeEventAnSkip(eventId); - transactionId = blockingSend(sizeLocation, startTime); - } finally { - outBytesLock.unlock(); } return readLong(transactionId, startTime); @@ -1440,14 +1496,9 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon long transactionId; - outBytesLock.lock(); - try { - + synchronized (this) { final long sizeLocation = writeEventAnSkip(eventId); - transactionId = blockingSend(sizeLocation, startTime); - } finally { - outBytesLock.unlock(); } @@ -1463,11 +1514,11 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon long timeoutTime = startTime + this.config.timeoutMs(); // receive - inBytesLock.lock(); + inBytesLock.writeLock().lock(); try { return (O) blockingFetchReadOnly(timeoutTime, transactionId).readObject(tClass); } finally { - inBytesLock.unlock(); + inBytesLock.writeLock().unlock(); } } @@ -1477,12 +1528,9 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon long transactionId; - outBytesLock.lock(); - try { + synchronized (this) { final long sizeLocation = writeEventAnSkip(eventId); transactionId = blockingSend(sizeLocation, startTime); - } finally { - outBytesLock.unlock(); } @@ -1497,8 +1545,7 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon long transactionId; - outBytesLock.lock(); - try { + synchronized (this) { final long sizeLocation = eventReturnsNull(eventId) ? writeEvent(eventId) @@ -1509,10 +1556,9 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon copies = writeValue(value, copies); } transactionId = blockingSend(sizeLocation, startTime); - } finally { - outBytesLock.unlock(); } + if (eventReturnsNull(eventId)) return null; @@ -1530,8 +1576,7 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon long transactionId; - outBytesLock.lock(); - try { + synchronized (this) { final long sizeLocation = eventReturnsNull(eventId) ? writeEvent(eventId) @@ -1539,8 +1584,6 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon copies = writeKey(key); transactionId = blockingSend(sizeLocation, startTime); - } finally { - outBytesLock.unlock(); } if (eventReturnsNull(eventId)) @@ -1553,38 +1596,6 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon "supported"); } - private O fetchObject(final EventId eventId) { - final long startTime = System.currentTimeMillis(); - long transactionId = send(eventId, startTime); - return readObject(transactionId, startTime); - } - - private O fetchObject(final EventId eventId, K key, V... values) { - final long startTime = System.currentTimeMillis(); - long transactionId = sendAndGetTransactionId(startTime, eventId, key, values); - return (O) readObject(transactionId, startTime); - } - - private long sendAndGetTransactionId(long startTime, final EventId eventId, K key, V... values) { - ThreadLocalCopies copies; - long transactionId;// send - outBytesLock.lock(); - try { - final long sizeLocation = (eventReturnsNull(eventId)) - ? writeEvent(eventId) - : writeEventAnSkip(eventId); - - copies = writeKey(key); - for (V value : values) { - copies = writeValue(value, copies); - } - transactionId = blockingSend(sizeLocation, startTime); - } finally { - outBytesLock.unlock(); - } - return transactionId; - } - private boolean eventReturnsNull(EventId eventId) { @@ -1622,14 +1633,11 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon final long startTime = System.currentTimeMillis(); long transactionId; - outBytesLock.lock(); - try { + synchronized (this) { final long sizeLocation = writeEventAnSkip(eventId); writeKey(key); writeObject(object); transactionId = blockingSend(sizeLocation, startTime); - } finally { - outBytesLock.unlock(); } if (eventReturnsNull(eventId)) diff --git a/src/main/java/net/openhft/chronicle/map/TcpReplicator.java b/src/main/java/net/openhft/chronicle/map/TcpReplicator.java index 33e221791..31b50bff5 100644 --- a/src/main/java/net/openhft/chronicle/map/TcpReplicator.java +++ b/src/main/java/net/openhft/chronicle/map/TcpReplicator.java @@ -1179,6 +1179,9 @@ final class TcpReplicator extends AbstractChannelReplicator implements Closeable // state is used for both heartbeat and stateless state = out.readByte(); sizeInBytes = out.readInt(); + //System.out.println("received size=" + sizeInBytes); + // System.out.println("......sizeInBytes=>" + sizeInBytes); + assert sizeInBytes > 0; // if the buffer is too small to read this payload we will have to grow the // size of the buffer @@ -1200,6 +1203,7 @@ final class TcpReplicator extends AbstractChannelReplicator implements Closeable } final long nextEntryPos = out.position() + sizeInBytes; + assert nextEntryPos > 0; final long limit = out.limit(); out.limit(nextEntryPos); @@ -1522,6 +1526,8 @@ class StatelessServerConnector { writeException(writer, e); writeSizeAndFlags(sizeLocation + SIZE_OF_TRANSACTION_ID, true, writer); + + e.printStackTrace(); return null; } @@ -1866,6 +1872,8 @@ class StatelessServerConnector { long pos = out.position(); long limit = out.limit(); + // System.out.println("Sending with size=" + size); + if (LOG.isDebugEnabled()) { out.position(locationOfSize); out.limit(pos); diff --git a/src/test/java/eg/BGChronicleTest.java b/src/test/java/eg/BGChronicleTest.java index df831891b..ebf5d8c05 100644 --- a/src/test/java/eg/BGChronicleTest.java +++ b/src/test/java/eg/BGChronicleTest.java @@ -3,6 +3,7 @@ package eg; import net.openhft.chronicle.hash.replication.TcpTransportAndNetworkConfig; import net.openhft.chronicle.map.ChronicleMap; import net.openhft.chronicle.map.ChronicleMapBuilder; +import org.junit.Ignore; import org.junit.Test; import java.io.IOException; @@ -236,6 +237,7 @@ public class BGChronicleTest { return port; } + @Ignore @Test public void test() throws IOException, InterruptedException { main(); diff --git a/src/test/java/net/openhft/chronicle/map/StatelessClientTest.java b/src/test/java/net/openhft/chronicle/map/StatelessClientTest.java index b105949ca..74183a7f1 100644 --- a/src/test/java/net/openhft/chronicle/map/StatelessClientTest.java +++ b/src/test/java/net/openhft/chronicle/map/StatelessClientTest.java @@ -18,15 +18,18 @@ package net.openhft.chronicle.map; +import junit.framework.Assert; import net.openhft.chronicle.hash.replication.TcpTransportAndNetworkConfig; import org.junit.After; import org.junit.Before; +import org.junit.Ignore; import org.junit.Test; import java.io.IOException; import java.net.InetSocketAddress; import java.util.*; -import java.util.concurrent.ExecutionException; +import java.util.concurrent.*; +import java.util.concurrent.atomic.AtomicBoolean; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertTrue; @@ -507,6 +510,21 @@ public class StatelessClientTest { } } + @Test(timeout = 10000) + public void testGetAndEntryWeDontHave() throws IOException, + InterruptedException, ExecutionException { + + try (ChronicleMap serverMap = ChronicleMapBuilder.of(Integer.class, CharSequence.class) + .replication((byte) 2, TcpTransportAndNetworkConfig.of(8056)).create()) { + try (ChronicleMap statelessMap = ChronicleMapBuilder.of(Integer + .class, CharSequence.class) + .statelessClient(new InetSocketAddress("localhost", 8056)).create()) { + assertEquals(null, statelessMap.get(3)); + + } + } + } + @Test(timeout = 10000) public void testEquals() throws IOException, InterruptedException { @@ -547,4 +565,94 @@ public class StatelessClientTest { serverMap1.close(); serverMap2.close(); } -} \ No newline at end of file + + + @Ignore + @Test + public void testThreadSafeness() throws IOException, InterruptedException { + + Thread.sleep(5000); + + final ExecutorService executorService = Executors.newFixedThreadPool(3); + + int count = 1000000; + final CountDownLatch latch = new CountDownLatch(count * 2); + + + // server + try (ChronicleMap server = ChronicleMapBuilder.of(Integer.class, Integer.class) + .putReturnsNull(true) + .replication((byte) 1, TcpTransportAndNetworkConfig.of(8059)).create()) { + + // stateless client + try (ChronicleMap client = ChronicleMapBuilder.of(Integer.class, + Integer.class) + .putReturnsNull(true) + .statelessClient(new InetSocketAddress("localhost", 8059)).create()) { + + for (int i = 0; i < count; i++) { + + + final int j = i; + executorService.submit(new Runnable() { + @Override + public void run() { + try { + System.out.println("put " + j + ", Thread" + Thread.currentThread().getName()); + client.put(j, j); + latch.countDown(); + } catch (Error | Exception e) { + e.printStackTrace(); + System.exit(0); + } + } + }); + } + + for (int i = 0; i < count; i++) { + final int j = i; + final AtomicBoolean issue = new AtomicBoolean(); + executorService.submit(new Runnable() { + @Override + public void run() { + try { + System.out.println("get " + j + ", Thread" + Thread.currentThread().getName()); + + while (client.get(j) == null) { + System.out.println("waiting"); + } + + if (j != client.get(j)) + issue.set(true); + + latch.countDown(); + } catch (Error | Exception e) { + e.printStackTrace(); + System.exit(0); + } + } + }); + + + if (issue.get()) + Assert.fail(); + + + } + latch.await(); + + System.out.println("..............................CLosing"); + } + } finally { + executorService.shutdownNow(); + executorService.awaitTermination(100, TimeUnit.SECONDS); + + } + + + } + + +} + +