From a89f31eae61f4797820ec703bd441aaf4ec9a8da Mon Sep 17 00:00:00 2001 From: Rob Austin Date: Fri, 13 Mar 2015 17:20:58 +0000 Subject: [PATCH] improved chunking in putAll() --- .../map/ClientWiredStatelessChronicleMap.java | 24 +++++++++++++------ .../openhft/chronicle/map/MapWireHandler.java | 22 +++++++++++------ 2 files changed, 32 insertions(+), 14 deletions(-) diff --git a/src/main/java/net/openhft/chronicle/map/ClientWiredStatelessChronicleMap.java b/src/main/java/net/openhft/chronicle/map/ClientWiredStatelessChronicleMap.java index 224d227f5..d89882dba 100644 --- a/src/main/java/net/openhft/chronicle/map/ClientWiredStatelessChronicleMap.java +++ b/src/main/java/net/openhft/chronicle/map/ClientWiredStatelessChronicleMap.java @@ -277,6 +277,7 @@ class ClientWiredStatelessChronicleMap implements ChronicleMap, Clon Set> entries = map.entrySet(); Iterator> iterator = entries.iterator(); + OUTER: while (iterator.hasNext()) { hub.outBytesLock().lock(); try { @@ -295,17 +296,26 @@ class ClientWiredStatelessChronicleMap implements ChronicleMap, Clon wire.write(Fields.CHANNEL_ID).int16(channelID); writeField(Fields.METHOD_NAME, "PUT_ALL"); - Map.Entry e = iterator.next(); - - hub.outWire().write(Fields.HAS_NEXT).bool(iterator.hasNext()); - - writeField(Fields.ARG_1, e.getKey()); - writeField(Fields.ARG_2, e.getValue()); - hub.writeSocket(hub.outWire()); + while (iterator.hasNext()) { + + Map.Entry e = iterator.next(); + + + hub.outWire().write(Fields.HAS_NEXT).bool(iterator.hasNext()); + writeField(Fields.ARG_1, e.getKey()); + writeField(Fields.ARG_2, e.getValue()); + + if (wire.bytes().remaining() < wire.bytes().capacity() * 0.5) + break OUTER; + + } + + } finally { + hub.writeSocket(hub.outWire()); hub.outBytesLock().unlock(); } diff --git a/src/main/java/net/openhft/chronicle/map/MapWireHandler.java b/src/main/java/net/openhft/chronicle/map/MapWireHandler.java index 80be4c636..99fe0d7bc 100644 --- a/src/main/java/net/openhft/chronicle/map/MapWireHandler.java +++ b/src/main/java/net/openhft/chronicle/map/MapWireHandler.java @@ -448,12 +448,18 @@ class MapWireHandler implements WireHandler, Consumer { @Override public void run() { - boolean hasNext = inWire.read(HAS_NEXT).bool(); - collectData.put(toBytes(ARG_1), toBytes(ARG_2)); - if (!hasNext) { - // the old code assumed that all the data would fit into a single buffer - // this assumption is invalid - if (!collectData.isEmpty()) { + + + // the old code assumed that ALL of the entries would fit into a single buffer + // this assumption is invalid + boolean hasNext; + for(;;) { + + hasNext = inWire.read(HAS_NEXT).bool(); + + collectData.put(toBytes(ARG_1), toBytes(ARG_2)); + + if (!hasNext) { incompleteWork.remove(transactionId); @@ -463,8 +469,10 @@ class MapWireHandler implements WireHandler, Consumer { bytesMap.delegate.putAll((Map) collectData); return null; }); - + return; } + if (inWire.bytes().remaining() == 0) + return; } } };