From 184d17e811a9dc977b90fd94656656134321b652 Mon Sep 17 00:00:00 2001 From: Rob Austin Date: Fri, 5 Dec 2014 16:33:12 +0000 Subject: [PATCH] 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); + + } + + + } + + +} + +