From a10f395db7ff10f244b748b60481b1291304fda4 Mon Sep 17 00:00:00 2001 From: Rob Austin Date: Fri, 5 Dec 2014 20:04:29 +0000 Subject: [PATCH] HCOLL-230 - its not possible unfortunaly to read and write a socketConnection between threads, all the access to the socket connections has to be on a single --- .../chronicle/map/StatelessChronicleMap.java | 377 ++++++++++-------- .../java/eg/TestIntrumentVOInterface.java | 45 +++ .../java/eg/TestNesterInterfaceMarshal.java | 110 +++++ .../chronicle/map/StatelessClientTest.java | 47 ++- 4 files changed, 393 insertions(+), 186 deletions(-) create mode 100644 src/test/java/eg/TestIntrumentVOInterface.java create mode 100644 src/test/java/eg/TestNesterInterfaceMarshal.java diff --git a/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java b/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java index cdc243c76..06a1a5381 100644 --- a/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java +++ b/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java @@ -65,8 +65,8 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon private ByteBuffer outBuffer; private ByteBufferBytes outBytes; - private ByteBuffer inBuffer; - private ByteBufferBytes inBytes; + private volatile ByteBuffer inBuffer; + private volatile ByteBufferBytes inBytes; private final ReaderWithSize keyReaderWithSize; private final WriterWithSize keyWriterWithSize; @@ -82,6 +82,8 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon private final boolean putReturnsNull; private final boolean removeReturnsNull; + private ExecutorService ioExecutor = Executors.newSingleThreadExecutor(); + private final ReentrantReadWriteLock inBytesLock = new ReentrantReadWriteLock(true); @@ -204,8 +206,8 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon return executorService; } - private synchronized SocketChannel lazyConnect(final long timeoutMs, - final InetSocketAddress remoteAddress) throws IOException { + private synchronized SocketChannel lazyConnect0(final long timeoutMs, + final InetSocketAddress remoteAddress) throws IOException { if (LOG.isDebugEnabled()) LOG.debug("attempting to connect to " + remoteAddress); @@ -229,9 +231,9 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon break; } catch (IOException e) { closeables.closeQuietly(); - } catch (Exception e) { + } catch (Throwable e) { closeables.closeQuietly(); - throw e; + NativeBytes.UNSAFE.throwException(e); } } @@ -239,7 +241,7 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon } /** - * attempts a single connect without a timeout and eat any IOException used typically in the + * attempts a single connect without a timeout and eats any IOException used typically in the * constructor, its possible the server is not running with this instance is created, {@link * net.openhft.chronicle.map.StatelessChronicleMap#lazyConnect(long, * java.net.InetSocketAddress)} will attempt to establish the connection when the client make @@ -251,19 +253,33 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon * @see net.openhft.chronicle.map.StatelessChronicleMap#lazyConnect(long, * java.net.InetSocketAddress) */ - - private synchronized void attemptConnect(final InetSocketAddress remoteAddress) throws IOException { + private synchronized void attemptConnect(@NotNull final InetSocketAddress remoteAddress) throws + IOException { // ensures that the excising connection are closed closeExisting(); try { - clientChannel = AbstractChannelReplicator.openSocketChannel(closeables); - clientChannel.connect(remoteAddress); - doHandShaking(clientChannel); - } catch (IOException e) { + ioExecutor.submit(new Callable() { + + @Override + public Void call() throws Exception { + StatelessChronicleMap.this.clientChannel = AbstractChannelReplicator + .openSocketChannel(closeables); + clientChannel.connect(remoteAddress); + doHandShaking0(clientChannel); + return null; + } + + }).get(); + + } catch (ExecutionException e) { + NativeBytes.UNSAFE.throwException(e.getCause()); + } catch (Throwable e) { closeables.closeQuietly(); + NativeBytes.UNSAFE.throwException(e); } + } /** @@ -285,7 +301,9 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon * @param clientChannel * @throws java.io.IOException */ - private synchronized void doHandShaking(@NotNull final SocketChannel clientChannel) throws IOException { + private void doHandShaking0(@NotNull final SocketChannel clientChannel) throws + IOException { + connectionByte[0] = STATELESS_CLIENT_IDENTIFIER; this.connectionOutBuffer.clear(); @@ -315,12 +333,53 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon LOG.debug("Attached to a map with a remote identifier=" + remoteIdentifier); } + private synchronized void doHandShaking(@NotNull final SocketChannel clientChannel) throws + Throwable { + + try { + ioExecutor.submit(new Callable() { + + @Override + public Void call() throws Exception { + doHandShaking0(clientChannel); + return null; + } + + }).get(); + + } catch (ExecutionException e) { + throw e.getCause(); + } + + } + + private synchronized SocketChannel lazyConnect(final long timeoutMs, + final InetSocketAddress remoteAddress) throws Throwable { + + try { + return ioExecutor.submit(new Callable() { + + @Override + public SocketChannel call() throws Exception { + return lazyConnect0(timeoutMs, remoteAddress); + } + + }).get(); + + } catch (ExecutionException e) { + throw e.getCause(); + } + + } + + public File file() { throw new UnsupportedOperationException(); } public void close() { + if (closeables != null) closeables.closeQuietly(); closeables = null; @@ -336,6 +395,17 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon } } + + if (ioExecutor != null) { + ioExecutor.shutdown(); + try { + if (!ioExecutor.awaitTermination(20, TimeUnit.SECONDS)) { + ioExecutor.shutdownNow(); + } + } catch (InterruptedException e) { + LOG.error("", e); + } + } } /** @@ -503,28 +573,6 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon } - private synchronized void sendWithoutAcc(final Bytes bytes, long sizeLocation) { - writeSizeAt(sizeLocation); - long timeoutTime = config.timeoutMs() + System.currentTimeMillis(); - try { - for (; ; ) { - if (clientChannel == null) - clientChannel = lazyConnect(config.timeoutMs(), config.remoteAddress()); - - try { - send(bytes, timeoutTime); - return; - } catch (java.nio.channels.ClosedChannelException | ClosedConnectionException e) { - checkTimeout(timeoutTime); - clientChannel = lazyConnect(config.timeoutMs(), config.remoteAddress()); - } - } - } catch (IOException e) { - close(); - throw new IORuntimeException(e); - } - } - private synchronized void writeEntriesForPutAll(Map map) { final int numberOfEntries = map.size(); int numberOfEntriesReadSoFar = 0; @@ -672,7 +720,7 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon inBytes.position(start); } - public void clear() { + public void clear() { fetchVoid(CLEAR); } @@ -957,59 +1005,26 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon } catch (IOException e) { close(); throw new IORuntimeException(e); - } catch (Exception e) { + + } catch (Throwable e) { close(); - throw e; + NativeBytes.UNSAFE.throwException(e); } return transactionId; } -/* private Bytes blockingFetchThrowable(long sizeLocation, final long transactionId, final long timeoutTime) throws IOException, InterruptedException { - - for (; ; ) { - if (clientChannel == null) { - synchronized (this) { - if (clientChannel == null) { - clientChannel = lazyConnect(config.timeoutMs(), config.remoteAddress()); - } - } - - try { - - if (LOG.isDebugEnabled()) - LOG.debug("sending data with transactionId=" + transactionId); - - writeSizeAndTransactionIdAt(sizeLocation, transactionId); - - // 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()); - } - } - } - - }*/ - private Bytes blockingFetchReadOnly(long timeoutTime, final long transactionId) { try { return blockingFetchThrowable(timeoutTime, transactionId); } catch (IOException e) { close(); throw new IORuntimeException(e); - } catch (RuntimeException e) { + } catch (Throwable e) { close(); - throw e; - } catch (Exception e) { - close(); - throw new RuntimeException(e); - } catch (AssertionError e) { LOG.error("", e); - throw e; + NativeBytes.UNSAFE.throwException(e); + return null; } } @@ -1029,32 +1044,43 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon * @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 Bytes receiveBuffer(int requiredNumberOfBytes, long timeoutTime) throws Throwable { + assert requiredNumberOfBytes <= buffer.capacity(); 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 + try { + ioExecutor.submit(new Callable() { - len = clientChannel.read(buffer.buffer()); + @Override + public Void call() throws Exception { + while (buffer.buffer().remaining() > 0) { + + 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)); + // System.out.println(("receiveBuffer = bytes read=" + len)); - if (len == -1) - throw new IORuntimeException("Disconnected to remote server"); + if (len == -1) + throw new IORuntimeException("Disconnected to remote server"); + } + return null; + } - checkTimeout(timeoutTime); + }).get(timeoutTime, TimeUnit.MILLISECONDS); - } + + } catch (TimeoutException e) { + throw new RemoteCallTimeoutException("timed out waiting to send data."); + } catch (ExecutionException e) { + throw e.getCause(); } + buffer.clear(); // System.out.println("buffer=" + AbstractBytes.toHex(buffer)); @@ -1062,8 +1088,51 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon } - private Bytes blockingFetchThrowable(long timeoutTime, long transactionId) throws IOException, - InterruptedException { + /** + * 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 synchronized Bytes receive(int requiredNumberOfBytes, long timeoutTime) throws + Throwable { + + inBytes.buffer().position(0); + inBytes.buffer().limit(requiredNumberOfBytes); + + assert requiredNumberOfBytes <= inBytes.capacity(); + + try { + ioExecutor.submit(new Callable() { + + @Override + public Void call() throws Exception { + while (inBytes.buffer().remaining() > 0) { + + int len = clientChannel.read(inBytes.buffer()); + if (len == -1) + throw new IORuntimeException("Disconnected to remote server"); + } + return null; + } + + }).get(timeoutTime, TimeUnit.MILLISECONDS); + + + } catch (TimeoutException e) { + throw new RemoteCallTimeoutException("timed out waiting to send data."); + } catch (ExecutionException e) { + throw e.getCause(); + } + + inBytes.position(0); + inBytes.limit(requiredNumberOfBytes); + return inBytes; + } + + private Bytes blockingFetchThrowable(long timeoutTime, long transactionId) throws Throwable { int remainingBytes = 0; try { @@ -1084,7 +1153,6 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon assert transactionId0 != 0; - // check the transaction id is reasonable assert String.valueOf(transactionId0).startsWith("14"); @@ -1139,18 +1207,11 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon pause(); } - } catch (Exception e) { + } catch (Throwable 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; + NativeBytes.UNSAFE.throwException(e); } @@ -1163,37 +1224,39 @@ 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); - final boolean isException = inBytes.readBoolean(); + boolean isException; - 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) { + + // block until we have received all the bytes in this chunk + receive(remainingBytes, timeoutTime); + 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); } - NativeBytes.UNSAFE.throwException(throwable); + 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); } + return inBytes; } + private void clearParked() { parkedTransactionId = 0; parkedRemainingBytes = 0; @@ -1206,44 +1269,32 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon } - /** - * 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 receive(int requiredNumberOfBytes, long timeoutTime) throws IOException { - - inBytes.buffer().position(0); - inBytes.buffer().limit(requiredNumberOfBytes); - - while (inBytes.buffer().remaining() > 0) { - assert requiredNumberOfBytes <= inBytes.capacity(); - - int len = clientChannel.read(inBytes.buffer()); - - if (len == -1) - throw new IORuntimeException("Disconnected to remote server"); - - checkTimeout(timeoutTime); - } - - inBytes.position(0); - inBytes.limit(requiredNumberOfBytes); - return inBytes; - } - - - private synchronized void send(final Bytes out, long timeoutTime) throws IOException { + private void send(final Bytes out, final long timeoutTime) throws Throwable { outBuffer.limit((int) out.position()); outBuffer.position(0); - while (outBuffer.remaining() > 0) { - clientChannel.write(outBuffer); - checkTimeout(timeoutTime); + try { + ioExecutor.submit(new Callable() { + + @Override + public Void call() throws Exception { + + while (outBuffer.remaining() > 0) { + int len = clientChannel.write(outBuffer); + if (len == -1) + throw new IORuntimeException("Disconnected to remote server"); + } + + return null; + } + + }).get(timeoutTime, TimeUnit.MILLISECONDS); + + } catch (TimeoutException e) { + throw new RemoteCallTimeoutException("timed out waiting to send data."); + } catch (ExecutionException e) { + throw e.getCause(); } out.clear(); @@ -1424,15 +1475,11 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon } private long send(final EventId eventId, final long startTime) { - long transactionId = nextUniqueTransaction(startTime); // send synchronized (this) { - final long sizeLocation = writeEventAnSkip(eventId); - transactionId = send(sizeLocation, startTime); + return send(writeEventAnSkip(eventId), startTime); } - - return transactionId; } private V readValue(ThreadLocalCopies copies, Bytes in) { diff --git a/src/test/java/eg/TestIntrumentVOInterface.java b/src/test/java/eg/TestIntrumentVOInterface.java new file mode 100644 index 000000000..915080bac --- /dev/null +++ b/src/test/java/eg/TestIntrumentVOInterface.java @@ -0,0 +1,45 @@ +package eg; + +import net.openhft.lang.model.constraints.MaxSize; + +/** + * Created by Vanitha on 12/5/2014. + */ +public interface TestIntrumentVOInterface { + + + public int getSizeOfInstrumentIDArray(); + + public void setSizeOfInstrumentIDArray(int sizeOfInstrumentIDArray); + + + public String getSymbol(); + + public void setSymbol(@MaxSize(20) String symbol); + + + + + public String getCurrencyCode(); + + public void setCurrencyCode(@MaxSize(4) String currencyCode); + + + public void setInstrumentIDAt(@MaxSize(2) int location, TestInstrumentIDVOInterface instrumentID); + + public TestInstrumentIDVOInterface getInstrumentIDAt(int location); + + + interface TestInstrumentIDVOInterface { + + + public String getIdSource(); + + public void setIdSource(@MaxSize(6) String idSource); + + public String getSecurityId(); + + public void setSecurityId(@MaxSize(100) String securityId); + + } +} diff --git a/src/test/java/eg/TestNesterInterfaceMarshal.java b/src/test/java/eg/TestNesterInterfaceMarshal.java new file mode 100644 index 000000000..cf36ad7cc --- /dev/null +++ b/src/test/java/eg/TestNesterInterfaceMarshal.java @@ -0,0 +1,110 @@ +package eg; + +import net.openhft.chronicle.hash.replication.TcpTransportAndNetworkConfig; +import net.openhft.chronicle.map.ChronicleMap; +import net.openhft.chronicle.map.ChronicleMapBuilder; +import net.openhft.lang.model.DataValueClasses; +import org.junit.Assert; +import org.junit.Ignore; +import org.junit.Test; + +import java.net.InetSocketAddress; +import java.util.concurrent.TimeUnit; + +/** + * Created by Vanitha on 12/5/2014. + */ +public class TestNesterInterfaceMarshal { + + + + + /** + * Test that both maps are created first and then populated the first map so that second will get the contents + * @throws Exception + */ + @Test + public void testReplicatedMap() throws Exception { + ChronicleMap map =createReplicatedMap1(); + ChronicleMap replicatedMap=createReplicatedMap2(); + + + //Store some data into MAP1 + TestIntrumentVOInterface intrumentVOInterface= DataValueClasses.newDirectInstance(TestIntrumentVOInterface.class); + intrumentVOInterface.setSymbol("FIXFlyer"); + intrumentVOInterface.setCurrencyCode("USA"); + intrumentVOInterface.setSizeOfInstrumentIDArray(2); + + TestIntrumentVOInterface.TestInstrumentIDVOInterface instrumentIDVOInterface= intrumentVOInterface.getInstrumentIDAt(0); + instrumentIDVOInterface.setIdSource("CUSIP"); + instrumentIDVOInterface.setSecurityId("TEST"); + intrumentVOInterface.setInstrumentIDAt(0,instrumentIDVOInterface); + map.put("KEY1",intrumentVOInterface); + + + + + Assert.assertNotNull(map); + Assert.assertNotNull(replicatedMap); + + + int t = 0; + for (; t < 5000; t++) { + if (map.equals(replicatedMap)) + break; + Thread.sleep(1); + } + + Assert.assertEquals(map, replicatedMap); + Assert.assertTrue(!replicatedMap.isEmpty()); + + System.out.println(map.get("KEY1")); + System.out.println(replicatedMap.get("KEY1")); + + map.close(); + replicatedMap.close(); + + } + + protected ChronicleMap createReplicatedMap1() { + + ChronicleMap chronicleMap = null; + TcpTransportAndNetworkConfig tcpTransportAndNetworkConfig=TcpTransportAndNetworkConfig.of(8076).heartBeatInterval(1L, TimeUnit.SECONDS); + try { + ChronicleMapBuilder builder = ChronicleMapBuilder.of(CharSequence.class, TestIntrumentVOInterface.class) + .entries(5000L).entrySize(2000) + .replication((byte) 1, tcpTransportAndNetworkConfig); + + + chronicleMap = builder.create(); + + } catch (Exception e) { + System.out.println("Error(s) creating instrument cache: " + e); + } + return chronicleMap; + } + + protected ChronicleMap createReplicatedMap2() { + + ChronicleMap chronicleMap = null; + TcpTransportAndNetworkConfig tcpTransportAndNetworkConfig=TcpTransportAndNetworkConfig.of(8077,new InetSocketAddress("127.0.0.1",8076)).heartBeatInterval(1L, TimeUnit.SECONDS); + + try { + ChronicleMapBuilder builder = ChronicleMapBuilder.of(CharSequence.class, TestIntrumentVOInterface.class) + .putReturnsNull(true) + .removeReturnsNull(true) + .entries(5000L).entrySize(2000) + .replication((byte)2,tcpTransportAndNetworkConfig) ; + + + chronicleMap = builder.create(); + + } catch (Exception e) { + System.out.println("*********************Error(s) creating launcher instrument cache: " + e); + } + return chronicleMap; + } + +} + + diff --git a/src/test/java/net/openhft/chronicle/map/StatelessClientTest.java b/src/test/java/net/openhft/chronicle/map/StatelessClientTest.java index a76b0b902..a667ebc29 100644 --- a/src/test/java/net/openhft/chronicle/map/StatelessClientTest.java +++ b/src/test/java/net/openhft/chronicle/map/StatelessClientTest.java @@ -18,7 +18,6 @@ package net.openhft.chronicle.map; -import junit.framework.Assert; import net.openhft.chronicle.hash.replication.TcpTransportAndNetworkConfig; import org.junit.After; import org.junit.Before; @@ -29,7 +28,7 @@ import java.io.IOException; import java.net.InetSocketAddress; import java.util.*; import java.util.concurrent.*; -import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicInteger; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertTrue; @@ -571,14 +570,14 @@ public class StatelessClientTest { @Test public void testThreadSafeness() throws IOException, InterruptedException { - Thread.sleep(5000); + int nThreads = 1; + final ExecutorService executorService = Executors.newFixedThreadPool(nThreads); - final ExecutorService executorService = Executors.newFixedThreadPool(3); - - int count = 1000000; + int count = 100000; final CountDownLatch latch = new CountDownLatch(count * 2); + final AtomicInteger got = new AtomicInteger(); - + long startTime = System.currentTimeMillis(); // server try (ChronicleMap server = ChronicleMapBuilder.of(Integer.class, Integer.class) .putReturnsNull(true) @@ -587,7 +586,6 @@ public class StatelessClientTest { // 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++) { @@ -598,7 +596,6 @@ public class StatelessClientTest { @Override public void run() { try { - System.out.println("put " + j + ", Thread" + Thread.currentThread().getName()); client.put(j, j); latch.countDown(); } catch (Error | Exception e) { @@ -611,19 +608,25 @@ public class StatelessClientTest { 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"); + Integer result = client.get(j); + + + if (result == null) { + executorService.submit(this); + return; } - if (j != client.get(j)) - issue.set(true); + if (result.equals(j)) { + got.incrementAndGet(); + } else { + System.out.println("expected j=" + j + " but got back=" + result); + } latch.countDown(); } catch (Error | Exception e) { @@ -634,14 +637,16 @@ public class StatelessClientTest { }); - if (issue.get()) - Assert.fail(); - - } - latch.await(); - System.out.println("..............................CLosing"); + latch.await(); + System.out.println("" + count + " messages took " + + TimeUnit.MILLISECONDS.toSeconds(System + .currentTimeMillis() - startTime) + " seconds, using " + nThreads + "" + + " threads"); + + assertEquals(count, got.get()); + } } finally { executorService.shutdownNow();