diff --git a/src/main/java/net/openhft/chronicle/map/BytesChronicleMap.java b/src/main/java/net/openhft/chronicle/map/BytesChronicleMap.java index aa867e609..6c23c93be 100644 --- a/src/main/java/net/openhft/chronicle/map/BytesChronicleMap.java +++ b/src/main/java/net/openhft/chronicle/map/BytesChronicleMap.java @@ -26,7 +26,7 @@ import java.util.Set; public class BytesChronicleMap implements AbstractChronicleMap { - private final VanillaChronicleMap delegate; + final VanillaChronicleMap delegate; MapIOBuffer output; public BytesChronicleMap(VanillaChronicleMap delegate) { diff --git a/src/main/java/net/openhft/chronicle/map/StatelessWiredConnector.java b/src/main/java/net/openhft/chronicle/map/StatelessWiredConnector.java index 97879d830..98c8a16f6 100644 --- a/src/main/java/net/openhft/chronicle/map/StatelessWiredConnector.java +++ b/src/main/java/net/openhft/chronicle/map/StatelessWiredConnector.java @@ -44,10 +44,7 @@ import java.nio.BufferOverflowException; import java.nio.ByteBuffer; import java.nio.channels.SelectionKey; import java.nio.channels.SocketChannel; -import java.util.ArrayList; -import java.util.Iterator; -import java.util.List; -import java.util.Map; +import java.util.*; import java.util.function.Consumer; import java.util.function.Function; @@ -171,11 +168,6 @@ class StatelessWiredConnector 4) { @@ -263,6 +255,40 @@ class StatelessWiredConnector incompleteWork = new HashMap<>(); + + + private void writeChunked(long transactionId, + @NotNull final Function> function, + @NotNull final Consumer> c) { + + final BytesChronicleMap m = bytesMap(channelId); + final Iterator iterator = function.apply(m); + + // this allows us to write more data than the buffer will allow + out = () -> { + + // each chunk has its own transaction-id + outWire.write(() -> "TRANSACTION_ID").int64(transactionId); + + write(map -> { + + boolean hasNext = iterator.hasNext(); + outWire.write(() -> "HAS_NEXT").bool(hasNext); + + if (hasNext) + c.accept(iterator); + else + // setting out to NULL denotes that there are no more chunks + out = null; + }); + + }; + + out.run(); + + } + @SuppressWarnings("UnusedReturnValue") void onEvent() { @@ -284,7 +310,6 @@ class StatelessWiredConnector "TRANSACTION_ID").int64(transactionId); @@ -410,11 +441,6 @@ class StatelessWiredConnector> function, - @NotNull final Consumer> c) { + Wire wire = inWire; + ValueIn read = wire.read(() -> fieldName); - final BytesChronicleMap m = bytesMap(channelId); - final Iterator iterator = function.apply(m); + long l = read.readLength(); + if (l > Integer.MAX_VALUE) + throw new BufferOverflowException(); - // this allows us to write more data than the buffer will allow - out = () -> { + int fieldLength = (int) l; - // each chunk has its own transaction-id - outWire.write(() -> "TRANSACTION_ID").int64(transactionId); + long endPos = wire.bytes().position() + fieldLength; + long limit = wire.bytes().limit(); - write(map -> { + try { + byte[] bytes = new byte[fieldLength]; - boolean hasNext = iterator.hasNext(); - outWire.write(() -> "HAS_NEXT").bool(hasNext); + wire.bytes().read(bytes); + return bytes; - if (hasNext) - c.accept(iterator); - else - // setting out to NULL denotes that there are no more chunks - out = null; - }); - - }; - - out.run(); + } finally + { + wire.bytes().position(endPos); + wire.bytes().limit(limit); + } } - - private void putAll() { + private void putAll(long transactionId) { final BytesChronicleMap bytesMap = bytesMap(StatelessWiredConnector.this.channelId); if (bytesMap == null) return; - out = () -> { - if (!inWire.read(() -> "HAS_MORE_CHUNKS").bool()) - // finished reading all the chunks - in = null; + Runnable runnable = incompleteWork.get(transactionId); - bytesMap.output = outMessageAdapter.outBuffer; - final net.openhft.lang.io.Bytes reader = toReader(inWire, "ARG_1", "ARG_2"); - try { - bytesMap.put(reader, reader, timestamp, identifier()); - } catch (Exception e) { - // todo send back exceptions + if (runnable != null) { + runnable.run(); + return; + } + + runnable = new Runnable() { + + // we should try and collect the data and then apply it atomically as quickly possible + final Map collectData = new HashMap<>(); + + @Override + public void run() { + if (inWire.read(() -> "HAS_NEXT").bool()) { + collectData.put(toBytes("ARG_1"), toBytes("ARG_2")); + } else { + // the old code assumed that all the data would fit into a single buffer + // this assumption is invalid + if (!collectData.isEmpty()) { + bytesMap.delegate.putAll((Map) collectData); + incompleteWork.remove(transactionId); + outWire.write(() -> "TRANSACTION_ID").int64(transactionId); + + // todo handle the case where there is an exception + outWire.write(() -> "IS_EXCEPTION").bool(false); + + nofityDataWritten(); + } + } } - }; + + incompleteWork.put(transactionId, runnable); + runnable.run(); + } + private byte identifier() { // if the client provides a remote identifier then we will use that otherwise we will use the local identifier return remoteIdentifier == 0 ? localIdentifer : remoteIdentifier; diff --git a/src/main/java/net/openhft/chronicle/map/WiredStatelessChronicleMap.java b/src/main/java/net/openhft/chronicle/map/WiredStatelessChronicleMap.java index 10bbc4b58..0392ee17a 100644 --- a/src/main/java/net/openhft/chronicle/map/WiredStatelessChronicleMap.java +++ b/src/main/java/net/openhft/chronicle/map/WiredStatelessChronicleMap.java @@ -42,6 +42,17 @@ import static net.openhft.chronicle.map.WiredStatelessChronicleMap.EventId.*; public class WiredStatelessChronicleMap implements ChronicleMap, Cloneable { private static final Logger LOG = LoggerFactory.getLogger(WiredStatelessChronicleMap.class); + + + // todo replace with an ENUM + public static final String METHOD_NAME = "METHOD_NAME"; + public static final String RESULT_IS_NULL = "RESULT_IS_NULL"; + public static final String HAS_NEXT = "HAS_NEXT"; + public static final String RESULT = "RESULT"; + public static final String ARG_1 = "ARG_1"; + public static final String ARG_2 = "ARG_2"; + + private final WiredStatelessClientTcpConnectionHub hub; protected Class kClass; @@ -258,6 +269,60 @@ public class WiredStatelessChronicleMap implements ChronicleMap, Clo return proxyReturnObject(vClass, removeReturnsNull ? REMOVE_WITHOUT_ACC.toString() : REMOVE.toString(), (K) key); } + public void putAll(@NotNull Map map) { + + final long startTime = System.currentTimeMillis(); + long transactionId = hub.nextUniqueTransaction(startTime); + + + Set> entries = map.entrySet(); + Iterator> iterator = entries.iterator(); + + boolean hasNext; + + do { + + hasNext = iterator.hasNext(); + + hub.outBytesLock().lock(); + try { + assert hub.outBytesLock().isHeldByCurrentThread(); + assert !hub.inBytesLock().isHeldByCurrentThread(); + + Wire wire = hub.outWire(); + + assert hub.outBytesLock().isHeldByCurrentThread(); + hub.markSize(wire); + hub.startTime(startTime); + + wire.write(() -> "TRANSACTION_ID").int64(transactionId); + wire.write(() -> "TIME_STAMP").int64(startTime); + wire.write(() -> "CHANNEL_ID").int16(channelID); + + writeField(METHOD_NAME, "PUT_ALL"); + + hub.outWire().write(() -> "HAS_NEXT").bool(hasNext); + + if (hasNext) { + Map.Entry e = iterator.next(); + writeField(ARG_1, e.getKey()); + writeField(ARG_2, e.getValue()); + } + + hub.writeSocket(hub.outWire()); + + } finally { + hub.outBytesLock().unlock(); + } + + } while (hasNext); + + // wait for the transaction id to be received, this will only be received once the + // last chunk has been processed + readVoid(transactionId, startTime); + } + + public V put(K key, V value) { if (key == null || value == null) throw new NullPointerException(); @@ -296,8 +361,8 @@ public class WiredStatelessChronicleMap implements ChronicleMap, Clo final Wire wireIn = hub.proxyReply(timeoutTime, transactionId); - if (wireIn.read(() -> "HAS_NEXT").bool()) { - K k = readK("RESULT", wireIn, null); + if (wireIn.read(() -> HAS_NEXT).bool()) { + K k = readK(RESULT, wireIn, null); result.add(k); } else break; @@ -329,8 +394,8 @@ public class WiredStatelessChronicleMap implements ChronicleMap, Clo final Wire wireIn = hub.proxyReply(timeoutTime, transactionId); - if (wireIn.read(() -> "HAS_NEXT").bool()) { - V v = readV("RESULT", wireIn, null); + if (wireIn.read(() -> HAS_NEXT).bool()) { + V v = readV(RESULT, wireIn, null); result.add(v); } else break; @@ -359,7 +424,7 @@ public class WiredStatelessChronicleMap implements ChronicleMap, Clo try { final Wire wireIn = hub.proxyReply(timeoutTime, transactionId); - if (wireIn.read(() -> "HAS_NEXT").bool()) { + if (wireIn.read(() -> HAS_NEXT).bool()) { result.put( readK("RESULT_KEY", wireIn, null), readV("RESULT_VALUE", wireIn, null)); @@ -380,10 +445,6 @@ public class WiredStatelessChronicleMap implements ChronicleMap, Clo throw new UnsupportedOperationException(); } - public void putAll(@NotNull Map map) { - throw new UnsupportedOperationException(); - } - private long readLong(long transactionId, long startTime) { assert !hub.outBytesLock().isHeldByCurrentThread(); @@ -392,7 +453,7 @@ public class WiredStatelessChronicleMap implements ChronicleMap, Clo // receive hub.inBytesLock().lock(); try { - return hub.proxyReply(timeoutTime, transactionId).read(() -> "RESULT").int64(); + return hub.proxyReply(timeoutTime, transactionId).read(() -> RESULT).int64(); } finally { hub.inBytesLock().unlock(); } @@ -427,7 +488,7 @@ public class WiredStatelessChronicleMap implements ChronicleMap, Clo final Wire wireIn = hub.proxyReply(timeoutTime, transactionId); - if (wireIn.read(() -> "RESULT_IS_NULL").bool()) + if (wireIn.read(() -> RESULT_IS_NULL).bool()) return null; return readV(argName, wireIn, usingValue); @@ -500,7 +561,7 @@ public class WiredStatelessChronicleMap implements ChronicleMap, Clo hub.inBytesLock().lock(); try { final Wire wireIn = hub.proxyReply(timeoutTime, transactionId); - return wireIn.read(() -> "RESULT").bool(); + return wireIn.read(() -> RESULT).bool(); } finally { hub.inBytesLock().unlock(); } @@ -515,7 +576,7 @@ public class WiredStatelessChronicleMap implements ChronicleMap, Clo hub.inBytesLock().lock(); try { final Wire wireIn = hub.proxyReply(timeoutTime, transactionId); - return wireIn.read(() -> "RESULT").int32(); + return wireIn.read(() -> RESULT).int32(); } finally { hub.inBytesLock().unlock(); } @@ -530,9 +591,9 @@ public class WiredStatelessChronicleMap implements ChronicleMap, Clo hub.outBytesLock().lock(); try { transactionId = hub.writeHeader(startTime, channelID, hub.outWire()); - writeField("METHOD_NAME", methodName); - writeField("ARG_1", key); - writeField("ARG_2", value); + writeField(METHOD_NAME, methodName); + writeField(ARG_1, key); + writeField(ARG_2, value); hub.writeSocket(hub.outWire()); } finally { @@ -551,9 +612,9 @@ public class WiredStatelessChronicleMap implements ChronicleMap, Clo hub.outBytesLock().lock(); try { transactionId = hub.writeHeader(startTime, channelID, hub.outWire()); - writeField("METHOD_NAME", methodName); - writeField("ARG_1", value1); - writeField("ARG_2", value2); + writeField(METHOD_NAME, methodName); + writeField(ARG_1, value1); + writeField(ARG_2, value2); hub.writeSocket(hub.outWire()); } finally { hub.outBytesLock().unlock(); @@ -585,7 +646,7 @@ public class WiredStatelessChronicleMap implements ChronicleMap, Clo hub.outBytesLock().lock(); try { transactionId = hub.writeHeader(startTime, channelID, hub.outWire()); - writeField("METHOD_NAME", methodName); + writeField(METHOD_NAME, methodName); hub.writeSocket(hub.outWire()); } finally { hub.outBytesLock().unlock(); @@ -603,7 +664,7 @@ public class WiredStatelessChronicleMap implements ChronicleMap, Clo hub.outBytesLock().lock(); try { transactionId = hub.writeHeader(startTime, channelID, hub.outWire()); - writeField("METHOD_NAME", methodName); + writeField(METHOD_NAME, methodName); hub.writeSocket(hub.outWire()); } finally { hub.outBytesLock().unlock(); @@ -622,7 +683,7 @@ public class WiredStatelessChronicleMap implements ChronicleMap, Clo hub.outBytesLock().lock(); try { transactionId = hub.writeHeader(startTime, channelID, hub.outWire()); - writeField("METHOD_NAME", methodName); + writeField(METHOD_NAME, methodName); hub.writeSocket(hub.outWire()); } finally { hub.outBytesLock().unlock(); @@ -641,7 +702,7 @@ public class WiredStatelessChronicleMap implements ChronicleMap, Clo hub.outBytesLock().lock(); try { transactionId = hub.writeHeader(startTime, channelID, hub.outWire()); - writeField("METHOD_NAME", methodName); + writeField(METHOD_NAME, methodName); hub.writeSocket(hub.outWire()); } finally { hub.outBytesLock().unlock(); @@ -665,9 +726,9 @@ public class WiredStatelessChronicleMap implements ChronicleMap, Clo assert !hub.inBytesLock().isHeldByCurrentThread(); transactionId = hub.writeHeader(startTime, channelID, hub.outWire()); - writeField("METHOD_NAME", methodName); - writeField("ARG_1", key); - writeField("ARG_2", value); + writeField(METHOD_NAME, methodName); + writeField(ARG_1, key); + writeField(ARG_2, value); hub.writeSocket(hub.outWire()); } finally { @@ -678,7 +739,7 @@ public class WiredStatelessChronicleMap implements ChronicleMap, Clo return null; if (resultType == vClass) - return (R) readValue("RESULT", transactionId, startTime, null); + return (R) readValue(RESULT, transactionId, startTime, null); else throw new UnsupportedOperationException("class of type class=" + resultType + " is not " + @@ -697,7 +758,7 @@ public class WiredStatelessChronicleMap implements ChronicleMap, Clo return null; if (rClass == vClass) - return (R) readValue("RESULT", transactionId, startTime, null); + return (R) readValue(RESULT, transactionId, startTime, null); else throw new UnsupportedOperationException("class of type class=" + rClass + " is not " + "supported"); @@ -708,8 +769,8 @@ public class WiredStatelessChronicleMap implements ChronicleMap, Clo hub.outBytesLock().lock(); try { transactionId = hub.writeHeader(startTime, channelID, hub.outWire()); - writeField("METHOD_NAME", methodName); - writeField("ARG_1", arg1); + writeField(METHOD_NAME, methodName); + writeField(ARG_1, arg1); hub.writeSocket(hub.outWire()); } finally { hub.outBytesLock().unlock(); @@ -723,7 +784,7 @@ public class WiredStatelessChronicleMap implements ChronicleMap, Clo hub.outBytesLock().lock(); try { transactionId = hub.writeHeader(startTime, channelID, hub.outWire()); - writeField("METHOD_NAME", methodName); + writeField(METHOD_NAME, methodName); hub.writeSocket(hub.outWire()); } finally { hub.outBytesLock().unlock(); diff --git a/src/main/java/net/openhft/chronicle/map/WiredStatelessClientTcpConnectionHub.java b/src/main/java/net/openhft/chronicle/map/WiredStatelessClientTcpConnectionHub.java index 22bd89721..d3ee5fd11 100644 --- a/src/main/java/net/openhft/chronicle/map/WiredStatelessClientTcpConnectionHub.java +++ b/src/main/java/net/openhft/chronicle/map/WiredStatelessClientTcpConnectionHub.java @@ -381,7 +381,8 @@ public class WiredStatelessClientTcpConnectionHub { intWire.bytes().skip(4); intWire.bytes().limit(messageSize); - System.out.println("\nread:\n" + Bytes.toDebugString(intWire.bytes())); + System.out.println("\n--------------------------------\n" + + "\nclient read:\n" + Bytes.toDebugString(intWire.bytes())); long transactionId0 = intWire.read(() -> "TRANSACTION_ID").int64(); @@ -519,7 +520,7 @@ public class WiredStatelessClientTcpConnectionHub { outBuffer.position(4); System.out.println("\n--------------------------------------------\n" + - "write:\n\n" + Bytes.toDebugString(outBuffer)); + "client wrote:\n\n" + Bytes.toDebugString(outBuffer)); outBuffer.position(0); diff --git a/src/test/java/net/openhft/chronicle/wire/TestWire.java b/src/test/java/net/openhft/chronicle/wire/TestWire.java index ba14f4b51..64db99633 100644 --- a/src/test/java/net/openhft/chronicle/wire/TestWire.java +++ b/src/test/java/net/openhft/chronicle/wire/TestWire.java @@ -24,15 +24,11 @@ import net.openhft.chronicle.hash.replication.TcpTransportAndNetworkConfig; import net.openhft.chronicle.map.ChronicleMap; import net.openhft.chronicle.map.WiredChronicleMapStatelessClientBuilder; import net.openhft.chronicle.map.WiredStatelessChronicleMap; -import org.junit.Assert; import org.junit.Test; import java.io.IOException; import java.net.InetSocketAddress; -import java.util.Collection; -import java.util.Collections; -import java.util.Map; -import java.util.Set; +import java.util.*; import static java.util.concurrent.TimeUnit.SECONDS; import static net.openhft.chronicle.map.ChronicleMapBuilder.of; @@ -319,6 +315,58 @@ public class TestWire { } + @Test + public void testPutAll() throws IOException, InterruptedException { + + { + final TcpTransportAndNetworkConfig tcpConfig = TcpTransportAndNetworkConfig + .of(++SERVER_PORT) + .heartBeatInterval(1, SECONDS); + + hubA = ReplicationHub.builder().tcpTransportAndNetwork(tcpConfig) + .createWithId((byte) 1); + + // this is how you add maps after the custer is created + map1a = of(byte[].class, byte[].class) + .instance().replicatedViaChannel(hubA.createChannel((short) 1)).create(); + + + byte identifier = (byte) 2; + + final WiredChronicleMapStatelessClientBuilder builder = + new WiredChronicleMapStatelessClientBuilder<>( + new InetSocketAddress("localhost", SERVER_PORT), + String.class, + String.class, + (short) 1); + + builder.identifier(identifier); + final ChronicleMap statelessMap = builder.create(); + + + Map map = new HashMap<>(); + map.put("Key1", "Value1"); + map.put("Key2", "Value2"); + + statelessMap.putAll(map); + + Set keys = statelessMap.keySet(); + + assertEquals(2, keys.size()); + assertTrue(keys.contains("Key2")); + assertTrue(keys.contains("Key1")); + + Collection values = statelessMap.values(); + + assertEquals(2, values.size()); + assertTrue(values.contains("Value1")); + assertTrue(values.contains("Value2")); + + map1a.close(); + } + + } + public static class Details implements Marshallable { Class keyClass;