From b3d9682aadd834e886f3b543ccf83d3941d1f97a Mon Sep 17 00:00:00 2001 From: Rob Austin Date: Fri, 5 Dec 2014 17:14:20 +0000 Subject: [PATCH] HCOLL-230 - updated entrySet to multi threaded --- .../chronicle/map/StatelessChronicleMap.java | 132 +++++++++++++----- .../chronicle/map/StatelessClientTest.java | 2 +- 2 files changed, 99 insertions(+), 35 deletions(-) diff --git a/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java b/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java index 9380a1c11..cdc243c76 100644 --- a/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java +++ b/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java @@ -503,19 +503,6 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon } - public synchronized void putAll(Map map) { - - final long sizeLocation = putReturnsNull ? writeEvent(PUT_ALL_WITHOUT_ACC) : - writeEventAnSkip(PUT_ALL); - - writeEntriesForPutAll(map); - - if (putReturnsNull) - sendWithoutAcc(outBytes, sizeLocation); - else - blockingFetch(sizeLocation); - } - private synchronized void sendWithoutAcc(final Bytes bytes, long sizeLocation) { writeSizeAt(sizeLocation); long timeoutTime = config.timeoutMs() + System.currentTimeMillis(); @@ -685,9 +672,9 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon inBytes.position(start); } - public synchronized void clear() { - // get the data back from the server - blockingFetch(writeEventAnSkip(CLEAR)); + public void clear() { + + fetchVoid(CLEAR); } @NotNull @@ -697,7 +684,7 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon final long startTime = System.currentTimeMillis(); final long timeoutTime = System.currentTimeMillis() + config.timeoutMs(); - final long transactionId = blockingSend(sizeLocation, startTime); + final long transactionId = send(sizeLocation, startTime); // get the data back from the server final Collection result = new ArrayList(); @@ -734,7 +721,7 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon final long sizeLocation = writeEventAnSkip(ENTRY_SET); final long startTime = System.currentTimeMillis(); final long timeoutTime = System.currentTimeMillis() + config.timeoutMs(); - final long transactionId = blockingSend(sizeLocation, startTime); + final long transactionId = send(sizeLocation, startTime); // get the data back from the server ThreadLocalCopies copies = keyReaderWithSize.getCopies(null); @@ -772,13 +759,31 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon return result.entrySet(); } + public void putAll(Map map) { + + + final long sizeLocation = putReturnsNull ? writeEvent(PUT_ALL_WITHOUT_ACC) : + writeEventAnSkip(PUT_ALL); + + final long startTime = System.currentTimeMillis(); + final long timeoutTime = startTime + config.timeoutMs(); + + + writeEntriesForPutAll(map); + + final long transactionId = send(sizeLocation, startTime); + + if (!putReturnsNull) + blockingFetchReadOnly(timeoutTime, transactionId); + } + @NotNull public Set keySet() { final long sizeLocation = writeEventAnSkip(KEY_SET); final long startTime = System.currentTimeMillis(); final long timeoutTime = startTime + config.timeoutMs(); - final long transactionId = blockingSend(sizeLocation, startTime); + final long transactionId = send(sizeLocation, startTime); final Set result = new HashSet<>(); final BytesReader keyReader = keyReaderWithSize.readerForLoop(null); @@ -898,7 +903,7 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon } } - private Bytes blockingFetch(long sizeLocation) { +/* private Bytes blockingFetch(long sizeLocation) { try { long startTime = System.currentTimeMillis(); return blockingFetchThrowable(sizeLocation, @@ -913,7 +918,7 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon close(); throw e; } - } + }*/ /** * sends data to the server via TCP/IP @@ -922,7 +927,7 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon * @param startTime the current time * @return a unique transaction id */ - private long blockingSend(long sizeLocation, final long startTime) { + private long send(long sizeLocation, final long startTime) { long transactionId = nextUniqueTransaction(startTime); final long timeoutTime = startTime + this.config.timeoutMs(); try { @@ -960,7 +965,7 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon } - private Bytes blockingFetchThrowable(long sizeLocation, final long transactionId, final long timeoutTime) throws IOException, InterruptedException { +/* private Bytes blockingFetchThrowable(long sizeLocation, final long transactionId, final long timeoutTime) throws IOException, InterruptedException { for (; ; ) { if (clientChannel == null) { @@ -988,7 +993,7 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon } } - } + }*/ private Bytes blockingFetchReadOnly(long timeoutTime, final long transactionId) { try { @@ -1231,7 +1236,7 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon } - private void send(final Bytes out, long timeoutTime) throws IOException { + private synchronized void send(final Bytes out, long timeoutTime) throws IOException { outBuffer.limit((int) out.position()); outBuffer.position(0); @@ -1424,7 +1429,7 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon // send synchronized (this) { final long sizeLocation = writeEventAnSkip(eventId); - transactionId = blockingSend(sizeLocation, startTime); + transactionId = send(sizeLocation, startTime); } return transactionId; @@ -1449,7 +1454,7 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon for (V value : values) { copies = writeValue(value, copies); } - transactionId = blockingSend(sizeLocation, startTime); + transactionId = send(sizeLocation, startTime); } @@ -1468,7 +1473,7 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon final long sizeLocation = writeEventAnSkip(eventId); writeValue(value, null); - transactionId = blockingSend(sizeLocation, startTime); + transactionId = send(sizeLocation, startTime); } @@ -1484,7 +1489,7 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon synchronized (this) { final long sizeLocation = writeEventAnSkip(eventId); - transactionId = blockingSend(sizeLocation, startTime); + transactionId = send(sizeLocation, startTime); } return readLong(transactionId, startTime); @@ -1498,7 +1503,7 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon synchronized (this) { final long sizeLocation = writeEventAnSkip(eventId); - transactionId = blockingSend(sizeLocation, startTime); + transactionId = send(sizeLocation, startTime); } @@ -1506,6 +1511,35 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon } + private void fetchVoid(final EventId eventId) { + + final long startTime = System.currentTimeMillis(); + + long transactionId; + + synchronized (this) { + final long sizeLocation = writeEventAnSkip(eventId); + transactionId = send(sizeLocation, startTime); + } + + readVoid(transactionId, startTime); + + + } + + private void readVoid(long transactionId, long startTime) { + long timeoutTime = startTime + this.config.timeoutMs(); + + // receive + inBytesLock.writeLock().lock(); + try { + blockingFetchReadOnly(timeoutTime, transactionId); + return; + } finally { + inBytesLock.writeLock().unlock(); + } + } + private O fetchObject(final Class tClass, final EventId eventId) { final long startTime = System.currentTimeMillis(); @@ -1530,7 +1564,7 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon synchronized (this) { final long sizeLocation = writeEventAnSkip(eventId); - transactionId = blockingSend(sizeLocation, startTime); + transactionId = send(sizeLocation, startTime); } @@ -1555,7 +1589,7 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon for (V value : values) { copies = writeValue(value, copies); } - transactionId = blockingSend(sizeLocation, startTime); + transactionId = send(sizeLocation, startTime); } @@ -1583,7 +1617,7 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon : writeEventAnSkip(eventId); copies = writeKey(key); - transactionId = blockingSend(sizeLocation, startTime); + transactionId = send(sizeLocation, startTime); } if (eventReturnsNull(eventId)) @@ -1596,6 +1630,36 @@ 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 + synchronized (this) { + + final long sizeLocation = (eventReturnsNull(eventId)) + ? writeEvent(eventId) + : writeEventAnSkip(eventId); + + copies = writeKey(key); + for (V value : values) { + copies = writeValue(value, copies); + } + transactionId = send(sizeLocation, startTime); + } + return transactionId; + } + private boolean eventReturnsNull(EventId eventId) { @@ -1637,7 +1701,7 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon final long sizeLocation = writeEventAnSkip(eventId); writeKey(key); writeObject(object); - transactionId = blockingSend(sizeLocation, startTime); + transactionId = send(sizeLocation, startTime); } if (eventReturnsNull(eventId)) diff --git a/src/test/java/net/openhft/chronicle/map/StatelessClientTest.java b/src/test/java/net/openhft/chronicle/map/StatelessClientTest.java index 74183a7f1..a76b0b902 100644 --- a/src/test/java/net/openhft/chronicle/map/StatelessClientTest.java +++ b/src/test/java/net/openhft/chronicle/map/StatelessClientTest.java @@ -567,7 +567,7 @@ public class StatelessClientTest { } - @Ignore + @Ignore("HCOLL-230 this has an issue as the map just gets slower and slower !") @Test public void testThreadSafeness() throws IOException, InterruptedException {