diff --git a/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java b/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java index 23bd41e4a..5e0ab4048 100644 --- a/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java +++ b/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java @@ -24,7 +24,6 @@ import net.openhft.chronicle.hash.serialization.BytesReader; import net.openhft.lang.io.ByteBufferBytes; import net.openhft.lang.io.Bytes; import net.openhft.lang.io.NativeBytes; -import net.openhft.lang.thread.NamedThreadFactory; import net.openhft.lang.threadlocal.ThreadLocalCopies; import org.jetbrains.annotations.NotNull; import org.jetbrains.annotations.Nullable; @@ -40,13 +39,11 @@ import java.nio.ByteBuffer; import java.nio.ByteOrder; import java.nio.channels.SocketChannel; import java.util.*; -import java.util.concurrent.ExecutorService; -import java.util.concurrent.Executors; -import java.util.concurrent.TimeUnit; 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 java.nio.ByteBuffer.allocate; import static net.openhft.chronicle.map.AbstractChannelReplicator.SIZE_OF_SIZE; import static net.openhft.chronicle.map.AbstractChannelReplicator.SIZE_OF_TRANSACTION_ID; import static net.openhft.chronicle.map.StatelessChronicleMap.EventId.*; @@ -66,11 +63,6 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon @NotNull private final AbstractChronicleMapBuilder chronicleMapBuilder; - private volatile ByteBuffer outBuffer; - private volatile ByteBufferBytes outBytes; - - private volatile ByteBuffer inBuffer; - private volatile ByteBufferBytes inBytes; @NotNull private final ReaderWithSize keyReaderWithSize; @@ -80,23 +72,29 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon private final ReaderWithSize valueReaderWithSize; @NotNull private final WriterWithSize valueWriterWithSize; - private volatile SocketChannel clientChannel; + @Nullable private CloseablesManager closeables; @NotNull private final StatelessMapConfig config; - private int maxEntrySize; + private final Class kClass; private final Class vClass; private final boolean putReturnsNull; private final boolean removeReturnsNull; + private ByteBuffer outBuffer; + private ByteBufferBytes outBytes; - private final ReentrantLock inBytesLock = new ReentrantLock(); - private final ReentrantLock outBytesLock = new ReentrantLock(true); + private ByteBuffer inBuffer; + private ByteBufferBytes inBytes; - private ExecutorService executorService; + private SocketChannel clientChannel; + private int maxEntrySize; + + private final ReentrantLock inBytesLock = new ReentrantLock(true); + private final ReentrantReadWriteLock.WriteLock outBytesLock = new ReentrantReadWriteLock().writeLock(); static enum EventId { HEARTBEAT, @@ -159,12 +157,13 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon this.name = chronicleMapBuilder.name(); attemptConnect(config.remoteAddress()); - outBuffer = allocateDirect(maxEntrySize).order(ByteOrder.nativeOrder()); + + outBuffer = allocate(maxEntrySize).order(ByteOrder.nativeOrder()); outBytes = new ByteBufferBytes(outBuffer.slice()); - inBuffer = allocateDirect(maxEntrySize).order(ByteOrder.nativeOrder()); - inBytes = new ByteBufferBytes(outBuffer.slice()); + inBuffer = allocate(maxEntrySize).order(ByteOrder.nativeOrder()); + inBytes = new ByteBufferBytes(inBuffer.slice()); } @@ -188,19 +187,13 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon return VanillaChronicleMap.newInstance(kClass, true); } - private ExecutorService lazyExecutorService() { - if (executorService == null) { - synchronized (this) { - if (executorService == null) - executorService = Executors.newSingleThreadExecutor(new NamedThreadFactory(chronicleMapBuilder.name() + - "-stateless-client-async-" + name, - true)); - } - } - return executorService; + private void checkTimeout(long timeoutTime) { + if (timeoutTime < System.currentTimeMillis()) + throw new RemoteCallTimeoutException(); } + private synchronized SocketChannel lazyConnect(final long timeoutMs, final InetSocketAddress remoteAddress) { @@ -236,14 +229,16 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon } /** - * attempts a single connect without a timeout and eat 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 the first map method - * call. + * attempts a single connect without a timeout and eat 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 + * the first map method call. * * @param remoteAddress the Inet Socket Address * @throws java.io.IOException - * @see net.openhft.chronicle.map.StatelessChronicleMap#lazyConnect(long, java.net.InetSocketAddress) + * @see net.openhft.chronicle.map.StatelessChronicleMap#lazyConnect(long, + * java.net.InetSocketAddress) */ private synchronized void attemptConnect(final InetSocketAddress remoteAddress) { @@ -257,7 +252,9 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon doHandShaking(clientChannel); } catch (IOException e) { if (closeables != null) closeables.closeQuietly(); + } + } /** @@ -272,8 +269,9 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon } /** - * initiates a very simple level of handshaking with the remote server, we send a special ID of -127 ( when the - * server receives this it knows its dealing with a stateless client, receive back an identifier from the server + * initiates a very simple level of handshaking with the remote server, we send a special ID of + * -127 ( when the server receives this it knows its dealing with a stateless client, receive + * back an identifier from the server * * @param clientChannel clientChannel * @throws java.io.IOException @@ -319,16 +317,6 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon closeables.closeQuietly(); closeables = null; - if (executorService != null) { - executorService.shutdown(); - try { - if (!executorService.awaitTermination(20, TimeUnit.SECONDS)) { - executorService.shutdownNow(); - } - } catch (InterruptedException e) { - LOG.error("", e); - } - } } @@ -391,9 +379,10 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon } /** - * calling this method should be avoided at all cost, as the entire {@code object} is serialized. This equals can be - * used to compare map that extends ChronicleMap. So two Chronicle Maps that contain the same data are considered - * equal, even if the instances of the chronicle maps were of different types + * calling this method should be avoided at all cost, as the entire {@code object} is + * serialized. This equals can be used to compare map that extends ChronicleMap. So two + * Chronicle Maps that contain the same data are considered equal, even if the instances of the + * chronicle maps were of different types * * @param object the object that you are comparing against * @return true if the contain the same data @@ -506,6 +495,8 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon private void resizeIfRequired(int numberOfEntries, int numberOfEntriesReadSoFar, final long start) { + assert outBytesLock.isHeldByCurrentThread(); + assert !inBytesLock.isHeldByCurrentThread(); final long remaining = outBytes.remaining(); if (remaining < maxEntrySize) { @@ -518,12 +509,18 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon * estimates the size based on what been completed so far */ private long estimateSize(int numberOfEntries, int numberOfEntriesReadSoFar) { + assert outBytesLock.isHeldByCurrentThread(); + assert !inBytesLock.isHeldByCurrentThread(); // we will back the size estimate on what we have so far final double percentageComplete = (double) numberOfEntriesReadSoFar / (double) numberOfEntries; return (long) ((double) outBytes.position() / percentageComplete); } private void resizeBufferOutBuffer(int size, long start) { + + assert outBytesLock.isHeldByCurrentThread(); + assert !inBytesLock.isHeldByCurrentThread(); + if (LOG.isDebugEnabled()) LOG.debug("resizing buffer to size=" + size); @@ -606,12 +603,21 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon @NotNull public Collection values() { - final long sizeLocation = writeEventAnSkip(VALUES); - final long startTime = System.currentTimeMillis(); + final long timeoutTime; + final long transactionId; - final long timeoutTime = System.currentTimeMillis() + config.timeoutMs(); - final long transactionId = send(sizeLocation, startTime); + outBytesLock.lock(); + try { + + final long sizeLocation = writeEventAnSkip(VALUES); + final long startTime = System.currentTimeMillis(); + + timeoutTime = System.currentTimeMillis() + config.timeoutMs(); + transactionId = send(sizeLocation, startTime); + } finally { + outBytesLock.unlock(); + } // get the data back from the server final Collection result = new ArrayList(); @@ -644,11 +650,18 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon @NotNull public Set> entrySet() { + final long transactionId; + final long timeoutTime; - final long sizeLocation = writeEventAnSkip(ENTRY_SET); - final long startTime = System.currentTimeMillis(); - final long timeoutTime = System.currentTimeMillis() + config.timeoutMs(); - final long transactionId = send(sizeLocation, startTime); + outBytesLock.lock(); + try { + final long sizeLocation = writeEventAnSkip(ENTRY_SET); + final long startTime = System.currentTimeMillis(); + timeoutTime = System.currentTimeMillis() + config.timeoutMs(); + transactionId = send(sizeLocation, startTime); + } finally { + outBytesLock.unlock(); + } // get the data back from the server ThreadLocalCopies copies = keyReaderWithSize.getCopies(null); @@ -688,9 +701,15 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon public void putAll(@NotNull Map map) { + final long sizeLocation; - final long sizeLocation = putReturnsNull ? writeEventAnSkip(PUT_ALL_WITHOUT_ACC) : - writeEventAnSkip(PUT_ALL); + outBytesLock.lock(); + try { + sizeLocation = putReturnsNull ? writeEventAnSkip(PUT_ALL_WITHOUT_ACC) : + writeEventAnSkip(PUT_ALL); + } finally { + outBytesLock.unlock(); + } final long startTime = System.currentTimeMillis(); final long timeoutTime = startTime + config.timeoutMs(); @@ -701,6 +720,9 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon outBytesLock.lock(); try { + assert outBytesLock.isHeldByCurrentThread(); + assert !inBytesLock.isHeldByCurrentThread(); + outBytes.writeStopBit(numberOfEntries); assert outBytes.limit() == outBytes.capacity(); ThreadLocalCopies copies = keyWriterWithSize.getCopies(null); @@ -757,11 +779,19 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon @NotNull public Set keySet() { + final long transactionId; + final long timeoutTime; + + outBytesLock.lock(); + try { + final long sizeLocation = writeEventAnSkip(KEY_SET); + final long startTime = System.currentTimeMillis(); + timeoutTime = startTime + config.timeoutMs(); + transactionId = send(sizeLocation, startTime); + } finally { + outBytesLock.unlock(); + } - final long sizeLocation = writeEventAnSkip(KEY_SET); - final long startTime = System.currentTimeMillis(); - final long timeoutTime = startTime + config.timeoutMs(); - final long transactionId = send(sizeLocation, startTime); final Set result = new HashSet<>(); final BytesReader keyReader = keyReaderWithSize.readerForLoop(null); @@ -809,6 +839,11 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon outBuffer.clear(); outBytes.clear(); + assert event != HEARTBEAT; + + outBytes.clear(); + outBuffer.clear(); + outBytes.write((byte) event.ordinal()); return markSizeLocation(); @@ -818,12 +853,21 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon * skips for the transactionid */ private long writeEventAnSkip(@NotNull EventId event) { + + assert outBytesLock.isHeldByCurrentThread(); + assert !inBytesLock.isHeldByCurrentThread(); + final long sizeLocation = writeEvent(event); + assert outBytes.readByte(0) == event.ordinal(); + + assert outBytes.position() > 0; + // skips for the transaction id outBytes.skip(SIZE_OF_TRANSACTION_ID); outBytes.writeByte(identifier); outBytes.writeInt(0); + assert outBytes.readByte(0) == event.ordinal(); return sizeLocation; } @@ -889,6 +933,10 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon * @return a unique transaction id */ private long send(long sizeLocation, final long startTime) { + + assert outBytesLock.isHeldByCurrentThread(); + assert !inBytesLock.isHeldByCurrentThread(); + long transactionId = nextUniqueTransaction(startTime); final long timeoutTime = startTime + this.config.timeoutMs(); try { @@ -902,7 +950,7 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon if (LOG.isDebugEnabled()) LOG.debug("sending data with transactionId=" + transactionId); - writeSizeAndTransactionIdAt(sizeLocation, transactionId, startTime); + writeSizeAndTransactionIdAt(sizeLocation, transactionId); // send out all the bytes writeBytesToSocket(outBytes, timeoutTime); @@ -1091,6 +1139,8 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon @SuppressWarnings("UnusedReturnValue") private Bytes receive(int requiredNumberOfBytes, long timeoutTime) throws IOException { + inBytes.position(0); + inBytes.limit(requiredNumberOfBytes); inBytes.buffer().position(0); inBytes.buffer().limit(requiredNumberOfBytes); @@ -1123,30 +1173,40 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon out.clear(); outBuffer.clear(); + outBytes.clear(); } - private void checkTimeout(long timeoutTime) { - if (timeoutTime < System.currentTimeMillis()) - throw new RemoteCallTimeoutException(); - } - private void writeSizeAndTransactionIdAt(long locationOfSize, final long transactionId, long startTime) { + private void writeSizeAndTransactionIdAt(long locationOfSize, final long transactionId) { + + assert outBytesLock.isHeldByCurrentThread(); + assert !inBytesLock.isHeldByCurrentThread(); + + assert outBytes.readByte(0) >= LONG_SIZE.ordinal(); + + final long size = outBytes.position() - locationOfSize; final long pos = outBytes.position(); outBytes.position(locationOfSize); + outBuffer.position((int) locationOfSize); + assert outBytes.readByte(0) >= LONG_SIZE.ordinal(); + + assert locationOfSize == 1; int size0 = (int) size - SIZE_OF_SIZE; - assert size0 > 0; outBytes.writeInt(size0); assert transactionId != 0; outBytes.writeLong(transactionId); - + assert outBytes.readByte(0) >= LONG_SIZE.ordinal(); outBytes.position(pos); } private long markSizeLocation() { + assert outBytesLock.isHeldByCurrentThread(); + assert !inBytesLock.isHeldByCurrentThread(); + final long position = outBytes.position(); - outBytes.skip(SIZE_OF_SIZE); + outBytes.writeInt(0); // skip the size return position; } @@ -1156,6 +1216,9 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon @SuppressWarnings("SameParameterValue") private ThreadLocalCopies writeKey(K key, ThreadLocalCopies copies) { + assert outBytesLock.isHeldByCurrentThread(); + assert !inBytesLock.isHeldByCurrentThread(); + long start = outBytes.position(); for (; ; ) { @@ -1176,6 +1239,10 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon @SuppressWarnings("UnusedReturnValue") private ThreadLocalCopies writeKeyInLoop(K key, Object writer, ThreadLocalCopies copies) { + + assert outBytesLock.isHeldByCurrentThread(); + assert !inBytesLock.isHeldByCurrentThread(); + long start = outBytes.position(); for (; ; ) { @@ -1195,6 +1262,9 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon } private ThreadLocalCopies writeValue(V value, ThreadLocalCopies copies) { + assert outBytesLock.isHeldByCurrentThread(); + assert !inBytesLock.isHeldByCurrentThread(); + long start = outBytes.position(); for (; ; ) { try { @@ -1214,6 +1284,8 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon @SuppressWarnings("UnusedReturnValue") private ThreadLocalCopies writeValueInLoop(V value, Object writer, ThreadLocalCopies copies) { + assert outBytesLock.isHeldByCurrentThread(); + assert !inBytesLock.isHeldByCurrentThread(); long start = outBytes.position(); for (; ; ) { try { @@ -1232,6 +1304,8 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon } private void resizeToMessage(long start, @NotNull IllegalArgumentException e) { + assert outBytesLock.isHeldByCurrentThread(); + assert !inBytesLock.isHeldByCurrentThread(); String message = e.getMessage(); if (message.startsWith(START_OF)) { String substring = message.substring("Attempt to write ".length(), message.length()); @@ -1298,10 +1372,16 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon } private long send(@NotNull final EventId eventId, final long startTime) { + + assert outBytesLock.isHeldByCurrentThread(); + assert !inBytesLock.isHeldByCurrentThread(); + + assert eventId.ordinal() != 0; // send outBytesLock.lock(); try { final long sizeLocation = writeEventAnSkip(eventId); + assert sizeLocation == 1; return send(sizeLocation, startTime); } finally { outBytesLock.unlock(); @@ -1468,7 +1548,14 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon @Nullable private O fetchObject(final Class tClass, @NotNull final EventId eventId) { final long startTime = System.currentTimeMillis(); - long transactionId = send(eventId, startTime); + long transactionId; + + outBytesLock.lock(); + try { + transactionId = send(eventId, startTime); + } finally { + outBytesLock.unlock(); + } long timeoutTime = startTime + this.config.timeoutMs(); @@ -1605,6 +1692,8 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon } private void writeObject(@NotNull Object function) { + assert outBytesLock.isHeldByCurrentThread(); + assert !inBytesLock.isHeldByCurrentThread(); long start = outBytes.position(); for (; ; ) { try { diff --git a/src/test/java/net/openhft/chronicle/map/StatelessClientTest.java b/src/test/java/net/openhft/chronicle/map/StatelessClientTest.java index 66fc927b5..a021fea97 100644 --- a/src/test/java/net/openhft/chronicle/map/StatelessClientTest.java +++ b/src/test/java/net/openhft/chronicle/map/StatelessClientTest.java @@ -24,6 +24,8 @@ import net.openhft.lang.io.Bytes; import org.junit.After; import org.junit.Before; import org.junit.Test; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; import java.io.IOException; import java.net.InetSocketAddress; @@ -39,6 +41,7 @@ import static org.junit.Assert.assertTrue; * @author Rob Austin. */ public class StatelessClientTest { + private static final Logger LOG = LoggerFactory.getLogger(StatelessClientTest.class); public static final int SIZE = 2500; static int s_port = 9070; @@ -521,7 +524,7 @@ public class StatelessClientTest { int nThreads = 2; final ExecutorService executorService = Executors.newFixedThreadPool(nThreads); - int count = 100; + int count = 10000; final CountDownLatch latch = new CountDownLatch(count * 2); final AtomicInteger got = new AtomicInteger(); @@ -529,12 +532,12 @@ public class StatelessClientTest { // server try (ChronicleMap server = ChronicleMapBuilder.of(Integer.class, Integer.class) .putReturnsNull(true) - .replication((byte) 1, TcpTransportAndNetworkConfig.of(8054)).create()) { + .replication((byte) 1, TcpTransportAndNetworkConfig.of(8047)).create()) { // stateless client try (ChronicleMap client = ChronicleMapBuilder.of(Integer.class, Integer.class) - .statelessClient(new InetSocketAddress("localhost", 8054)).create()) { + .statelessClient(new InetSocketAddress("localhost", 8047)).create()) { for (int i = 0; i < count; i++) { @@ -544,16 +547,19 @@ public class StatelessClientTest { @Override public void run() { try { + // System.out.print("put("+j+")"); client.put(j, j); latch.countDown(); } catch (Error | Exception e) { - e.printStackTrace(); - throw e; + LOG.error("", e); + executorService.shutdown(); + } } }); } + for (int i = 0; i < count; i++) { final int j = i; @@ -564,8 +570,8 @@ public class StatelessClientTest { Integer result = client.get(j); - if (result == null) { + System.out.print("entry not found so re-submitting"); executorService.submit(this); return; } @@ -579,7 +585,8 @@ public class StatelessClientTest { latch.countDown(); } catch (Error | Exception e) { e.printStackTrace(); - throw e; + LOG.error("", e); + executorService.shutdown(); } } }); @@ -587,7 +594,7 @@ public class StatelessClientTest { } - latch.await(10,TimeUnit.SECONDS); + latch.await(10, TimeUnit.SECONDS); System.out.println("" + count + " messages took " + TimeUnit.MILLISECONDS.toSeconds(System .currentTimeMillis() - startTime) + " seconds, using " + nThreads + "" +