From f9e6a5783915e834c65c0bec8fca53994980b995 Mon Sep 17 00:00:00 2001 From: Rob Austin Date: Fri, 5 Dec 2014 21:04:03 +0000 Subject: [PATCH] HCOLL-230 - introduced outBytesLock --- .../chronicle/map/StatelessChronicleMap.java | 77 ++++++++++++------- 1 file changed, 51 insertions(+), 26 deletions(-) diff --git a/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java b/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java index 74801d8bc..dfa1ec71e 100644 --- a/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java +++ b/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java @@ -62,8 +62,8 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon private final String name; private final AbstractChronicleMapBuilder chronicleMapBuilder; - private ByteBuffer outBuffer; - private ByteBufferBytes outBytes; + private volatile ByteBuffer outBuffer; + private volatile ByteBufferBytes outBytes; private volatile ByteBuffer inBuffer; private volatile ByteBufferBytes inBytes; @@ -84,10 +84,10 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon private final ReentrantLock inBytesLock = new ReentrantLock(true); + private final ReentrantLock outBytesLock = new ReentrantLock(true); private ExecutorService executorService; - static enum EventId { HEARTBEAT, STATEFUL_UPDATE, @@ -163,9 +163,7 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon return lazyExecutorService().submit(new Callable() { @Override public V call() throws Exception { - - V oldValue = StatelessChronicleMap.this.put(key, value); - return oldValue; + return StatelessChronicleMap.this.put(key, value); } }); } @@ -175,8 +173,7 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon return lazyExecutorService().submit(new Callable() { @Override public V call() throws Exception { - V oldValue = StatelessChronicleMap.this.remove(key); - return oldValue; + return StatelessChronicleMap.this.remove(key); } }); } @@ -245,7 +242,7 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon * java.net.InetSocketAddress)} will attempt to establish the connection when the client make * the first map method call. * - * @param remoteAddress + * @param remoteAddress the Inet Socket Address * @return * @throws java.io.IOException * @see net.openhft.chronicle.map.StatelessChronicleMap#lazyConnect(long, @@ -282,7 +279,7 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon * -127 ( when the server receives this it knows its dealing with a stateless client, receive * back an identifier from the server * - * @param clientChannel + * @param clientChannel clientChannel * @throws java.io.IOException */ private synchronized void doHandShaking(@NotNull final SocketChannel clientChannel) throws IOException { @@ -749,7 +746,8 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon final int numberOfEntries = map.size(); int numberOfEntriesReadSoFar = 0; - synchronized (outBytes) { + outBytesLock.lock(); + try { outBytes.writeStopBit(numberOfEntries); assert outBytes.limit() == outBytes.capacity(); @@ -796,6 +794,8 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon } transactionId = send(sizeLocation, startTime); + } finally { + outBytesLock.unlock(); } @@ -1360,15 +1360,14 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon } private long send(final EventId eventId, final long startTime) { - - // send - synchronized (outBytes) { + outBytesLock.lock(); + try { final long sizeLocation = writeEventAnSkip(eventId); return send(sizeLocation, startTime); + } finally { + outBytesLock.unlock(); } - - } private V readValue(ThreadLocalCopies copies, Bytes in) { @@ -1381,7 +1380,8 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon long transactionId; - synchronized (outBytes) { + outBytesLock.lock(); + try { final long sizeLocation = writeEventAnSkip(eventId); @@ -1390,6 +1390,8 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon copies = writeValue(value, copies); } transactionId = send(sizeLocation, startTime); + } finally { + outBytesLock.unlock(); } @@ -1403,10 +1405,13 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon long transactionId; - synchronized (outBytes) { + outBytesLock.lock(); + try { final long sizeLocation = writeEventAnSkip(eventId); writeValue(value, null); transactionId = send(sizeLocation, startTime); + } finally { + outBytesLock.unlock(); } return readBoolean(transactionId, startTime); @@ -1419,9 +1424,12 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon long transactionId; - synchronized (outBytes) { + outBytesLock.lock(); + try { final long sizeLocation = writeEventAnSkip(eventId); transactionId = send(sizeLocation, startTime); + } finally { + outBytesLock.unlock(); } return readLong(transactionId, startTime); @@ -1433,9 +1441,12 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon long transactionId; - synchronized (outBytes) { + outBytesLock.lock(); + try { final long sizeLocation = writeEventAnSkip(eventId); transactionId = send(sizeLocation, startTime); + } finally { + outBytesLock.unlock(); } return readBoolean(transactionId, startTime); @@ -1448,14 +1459,16 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon long transactionId; - synchronized (outBytes) { + outBytesLock.lock(); + try { final long sizeLocation = writeEventAnSkip(eventId); transactionId = send(sizeLocation, startTime); + } finally { + outBytesLock.unlock(); } readVoid(transactionId, startTime); - } private void readVoid(long transactionId, long startTime) { @@ -1493,9 +1506,12 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon long transactionId; - synchronized (outBytes) { + outBytesLock.lock(); + try { final long sizeLocation = writeEventAnSkip(eventId); transactionId = send(sizeLocation, startTime); + } finally { + outBytesLock.unlock(); } @@ -1510,7 +1526,8 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon long transactionId; - synchronized (outBytes) { + outBytesLock.lock(); + try { final long sizeLocation = eventReturnsNull(eventId) ? writeEvent(eventId) @@ -1521,6 +1538,8 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon copies = writeValue(value, copies); } transactionId = send(sizeLocation, startTime); + } finally { + outBytesLock.unlock(); } @@ -1541,7 +1560,8 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon long transactionId; - synchronized (outBytes) { + outBytesLock.lock(); + try { final long sizeLocation = eventReturnsNull(eventId) ? writeEvent(eventId) @@ -1549,6 +1569,8 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon copies = writeKey(key); transactionId = send(sizeLocation, startTime); + } finally { + outBytesLock.unlock(); } if (eventReturnsNull(eventId)) @@ -1598,11 +1620,14 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon final long startTime = System.currentTimeMillis(); long transactionId; - synchronized (outBytes) { + outBytesLock.lock(); + try { final long sizeLocation = writeEventAnSkip(eventId); writeKey(key); writeObject(object); transactionId = send(sizeLocation, startTime); + } finally { + outBytesLock.unlock(); } if (eventReturnsNull(eventId))