From 60b23535d1a443880658bb9c1007a5c4d2ed48c4 Mon Sep 17 00:00:00 2001 From: Rob Austin Date: Thu, 23 Oct 2014 14:52:38 +0100 Subject: [PATCH] HCOLL-167 added support for putAll() --- .../chronicle/map/ChronicleMapBuilder.java | 9 +- .../chronicle/map/StatelessChronicleMap.java | 153 +++++++++++++----- .../openhft/chronicle/map/TcpReplicator.java | 29 ++-- 3 files changed, 136 insertions(+), 55 deletions(-) diff --git a/src/main/java/net/openhft/chronicle/map/ChronicleMapBuilder.java b/src/main/java/net/openhft/chronicle/map/ChronicleMapBuilder.java index 6b5a3a6af..4b1cf7477 100644 --- a/src/main/java/net/openhft/chronicle/map/ChronicleMapBuilder.java +++ b/src/main/java/net/openhft/chronicle/map/ChronicleMapBuilder.java @@ -64,8 +64,8 @@ import static net.openhft.chronicle.map.Objects.builderEquals; * *

Use static {@link #of(Class, Class)} method to obtain a {@code ChronicleMapBuilder} instance. * - *

{@code ChronicleMapBuilder} is mutable, see a note in {@link net.openhft.chronicle.common.ChronicleHashBuilder} interface - * documentation. + *

{@code ChronicleMapBuilder} is mutable, see a note in {@link net.openhft.chronicle.common.ChronicleHashBuilder} + * interface documentation. * *

Later in this documentation, "ChronicleMap" means "ChronicleMaps, created by {@code * ChronicleMapBuilder}", unless specified different, because theoretically someone might provide @@ -819,8 +819,7 @@ public class ChronicleMapBuilder implements Cloneable, if (statelessBuilder != null) return createStatelessMap(statelessBuilder); - return (file != null) ? createWithFile(file) : createWithoutFile(); - + return (file != null) ? createWithFile(file) : createWithoutFile(); } private ChronicleMap createWithoutFile() throws IOException { @@ -837,7 +836,7 @@ public class ChronicleMapBuilder implements Cloneable, final KeyValueSerializer keyValueSerializer = new KeyValueSerializer(keyBuilder, valueBuilder); - return new StatelessChronicleMap(keyValueSerializer, statelessBuilder); + return new StatelessChronicleMap(keyValueSerializer, statelessBuilder, entrySize); } private VanillaChronicleMap newMap() throws IOException { diff --git a/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java b/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java index 7a38c8bb8..e572cf38b 100644 --- a/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java +++ b/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java @@ -32,17 +32,16 @@ class StatelessChronicleMap implements ChronicleMap, Closeable { private final byte[] connectionByte = new byte[1]; private final ByteBuffer connectionOutBuffer = ByteBuffer.wrap(connectionByte); - private final ByteBuffer outBuffer = allocateDirect(1024); - private final ByteBufferBytes out = new ByteBufferBytes(outBuffer.slice()); - - // if you want you could change this later so that "in" has its own buffer. - private ByteBufferBytes in = out; + private ByteBuffer buffer; + private ByteBufferBytes bytes; private final KeyValueSerializer keyValueSerializer; private volatile SocketChannel clientChannel; private CloseablesManager closeables; private final StatelessBuilder builder; + private int maxEntrySize; + private int headroom = 1024; static enum EventId { HEARTBEAT, @@ -61,16 +60,23 @@ class StatelessChronicleMap implements ChronicleMap, Closeable { REPLACE_WITH_OLD_AND_NEW_VALUE, PUT_IF_ABSENT, REMOVE_WITH_VALUE, - TO_STRING + TO_STRING, + PUT_ALL } private long transactionID; StatelessChronicleMap(final KeyValueSerializer keyValueSerializer, - final StatelessBuilder builder) throws IOException { + final StatelessBuilder builder, + int maxEntrySize) throws IOException { this.keyValueSerializer = keyValueSerializer; this.builder = builder; + this.maxEntrySize = maxEntrySize; attemptConnect(builder.remoteAddress()); + headroom = 1024 + maxEntrySize; + buffer = allocateDirect(headroom); + bytes = new ByteBufferBytes(buffer.slice()); + } @@ -339,16 +345,85 @@ class StatelessChronicleMap implements ChronicleMap, Closeable { public synchronized void putAll(Map map) { - for (Map.Entry e : map.entrySet()) { - put(e.getKey(), e.getValue()); + final long sizeLocation = writeEvent(PUT_ALL); + final int numberOfEntries = map.size(); + int numberOfEntriesReadSoFar = 0; + + bytes.writeStopBit(numberOfEntries); + + for (final Map.Entry e : map.entrySet()) { + + numberOfEntriesReadSoFar++; + + // putAll if a bit more complicated than the others + // as the volume of data could cause us to have to resize our buffers + resizeIfRequired(numberOfEntries, numberOfEntriesReadSoFar); + + long start = bytes.position(); + writeKey(e.getKey()); + writeValue(e.getValue()); + int len = (int) (bytes.position() - start); + + if (len > maxEntrySize) + maxEntrySize = len; } + blockingFetch(sizeLocation); + + } + + private void resizeIfRequired(int numberOfEntries, int numberOfEntriesReadSoFar) { + if (bytes.remaining() < maxEntrySize) { + long estimatedRequiredSize = estimateSize(numberOfEntries, numberOfEntriesReadSoFar); + resizeBuffer(estimatedRequiredSize + headroom); + } + } + + /** + * estimates the size based on what been completed so far + * + * @param numberOfEntries + * @param numberOfEntriesReadSoFar + */ + private long estimateSize(int numberOfEntries, int numberOfEntriesReadSoFar) { + // we will back the size estimate on what we have so far + double percentageComplete = numberOfEntries / numberOfEntriesReadSoFar; + return (long) ((double) bytes.position() / percentageComplete); + } + + + void resizeBuffer(long size) { + if (size < buffer.capacity()) + throw new IllegalStateException("it not possible to resize the buffer smaller"); + + ByteBuffer result = ByteBuffer.allocateDirect((int) size); + buffer.clear(); + + int bufferPosition = buffer.position(); + int bufferLimit = buffer.limit(); + + long bytesPosition = bytes.position(); + long bytesLimit = bytes.limit(); + + buffer.position(0); + + for (int i = 0; i < bufferLimit; i++) { + result.put(buffer.get()); + } + + buffer.limit(bufferLimit); + buffer.position(bufferPosition); + + bytes = new ByteBufferBytes(result.slice()); + bytes.limit(bytesLimit); + bytes.position(bytesPosition); + } /* private void writeEntries(Map map) { final HashMap safeCopy = new HashMap(map); - out.writeStopBit(safeCopy.size()); + bytes.writeStopBit(safeCopy.size()); final Set entries = (Set) safeCopy.entrySet(); @@ -419,12 +494,10 @@ class StatelessChronicleMap implements ChronicleMap, Closeable { private long writeEvent(EventId event) { - in.clear(); - out.clear(); + bytes.clear(); + buffer.clear(); - outBuffer.clear(); - - out.write((byte) event.ordinal()); + bytes.write((byte) event.ordinal()); long sizeLocation = markSizeLocation(); return sizeLocation; } @@ -515,30 +588,30 @@ class StatelessChronicleMap implements ChronicleMap, Closeable { writeSizeAt(sizeLocation); // send out all the bytes - send(out, timeoutTime); + send(bytes, timeoutTime); - in.clear(); - in.buffer().clear(); + bytes.clear(); + bytes.buffer().clear(); // the number of bytes in the response int size = receive(2, timeoutTime).readUnsignedShort(); - if (in.capacity() < size) - in = new ByteBufferBytes(allocateDirect(size)); + if (bytes.capacity() < size) + bytes = new ByteBufferBytes(allocateDirect(size)); receive(size, timeoutTime); - boolean isException = in.readBoolean(); - long inTransactionId = in.readLong(); + boolean isException = bytes.readBoolean(); + long inTransactionId = bytes.readLong(); if (inTransactionId != transactionId) throw new IllegalStateException("the received transaction-id=" + inTransactionId + ", does not match the expected transaction-id=" + transactionId); if (isException) - throw (RuntimeException) in.readObject(); + throw (RuntimeException) bytes.readObject(); - return in; + return bytes; } catch (java.nio.channels.ClosedChannelException e) { checkTimeout(timeoutTime); clientChannel = lazyConnect(builder.timeoutMs(), builder.remoteAddress()); @@ -550,31 +623,31 @@ class StatelessChronicleMap implements ChronicleMap, Closeable { } private void write(long transactionId) { - out.writeLong(transactionId); + bytes.writeLong(transactionId); } private ByteBufferBytes receive(int requiredNumberOfBytes, long timeoutTime) throws IOException { - while (in.buffer().position() < requiredNumberOfBytes) { - clientChannel.read(in.buffer()); + while (bytes.buffer().position() < requiredNumberOfBytes) { + clientChannel.read(bytes.buffer()); checkTimeout(timeoutTime); } - in.limit(in.buffer().position()); - return in; + bytes.limit(bytes.buffer().position()); + return bytes; } private void send(final ByteBufferBytes out, long timeoutTime) throws IOException { - outBuffer.limit((int) out.position()); - outBuffer.position(0); + buffer.limit((int) out.position()); + buffer.position(0); - while (outBuffer.remaining() > 0) { - clientChannel.write(outBuffer); + while (buffer.remaining() > 0) { + clientChannel.write(buffer); checkTimeout(timeoutTime); } out.clear(); - outBuffer.clear(); + buffer.clear(); } @@ -584,18 +657,18 @@ class StatelessChronicleMap implements ChronicleMap, Closeable { } private void writeSizeAt(long locationOfSize) { - long size = out.position() - locationOfSize; - out.writeUnsignedShort(locationOfSize, (int) size - 2); + long size = bytes.position() - locationOfSize; + bytes.writeUnsignedShort(locationOfSize, (int) size - 2); } private long markSizeLocation() { - long position = out.position(); - out.skip(2); + long position = bytes.position(); + bytes.skip(2); return position; } private void writeKey(K key) { - keyValueSerializer.writeKey(key, out); + keyValueSerializer.writeKey(key, bytes); } private V readKey(final long sizeLocation) { @@ -603,7 +676,7 @@ class StatelessChronicleMap implements ChronicleMap, Closeable { } private void writeValue(V value) { - keyValueSerializer.writeValue(value, out); + keyValueSerializer.writeValue(value, bytes); } private Set readKeySet(Bytes in) { diff --git a/src/main/java/net/openhft/chronicle/map/TcpReplicator.java b/src/main/java/net/openhft/chronicle/map/TcpReplicator.java index cc162ff8e..0b3fda654 100644 --- a/src/main/java/net/openhft/chronicle/map/TcpReplicator.java +++ b/src/main/java/net/openhft/chronicle/map/TcpReplicator.java @@ -1036,17 +1036,18 @@ class TcpReplicator extends AbstractChannelReplicator implements Closeable { void resizeBuffer(long size) { + if (size < in.capacity()) throw new IllegalStateException("it not possible to resize the buffer smaller"); - ByteBuffer buffer = ByteBuffer.allocateDirect((int) size); + final ByteBuffer buffer = ByteBuffer.allocateDirect((int) size); buffer.clear(); - long outPosition = out.position(); - long outLimit = out.limit(); + final long outPosition = out.position(); + final long outLimit = out.limit(); - int inPosition = in.position(); - int inLimit = in.limit(); + final int inPosition = in.position(); + final int inLimit = in.limit(); in.position(0); @@ -1054,12 +1055,15 @@ class TcpReplicator extends AbstractChannelReplicator implements Closeable { buffer.put(in.get()); } - buffer.position(inPosition); buffer.limit(inLimit); + buffer.position(inPosition); + in = buffer; out = new ByteBufferBytes(in.slice()); - out.position(outPosition); + out.limit(outLimit); + out.position(outPosition); + } /** @@ -1293,6 +1297,10 @@ class StatelessServerConnector { case TO_STRING: return toString(in, out); + case PUT_ALL: + return putAll(in, out); + + default: throw new IllegalStateException("unsupported event=" + event); @@ -1468,6 +1476,7 @@ class StatelessServerConnector { private Work putAll(Bytes in, Bytes out) { final Map m = readEntries(in); final long sizeLocation = reflectTransactionId(in, out); + try { map.putAll(m); } catch (RuntimeException e) { @@ -1674,16 +1683,15 @@ class StatelessServerConnector { private Map readEntries(Bytes in) { - final long size = in.readStopBit(); + final long numberOfEntries = in.readStopBit(); final HashMap result = new HashMap(); - for (long i = 0; i < size; i++) { + for (long i = 0; i < numberOfEntries; i++) { result.put(readKey(in), readValue(in)); } return result; } - } class KeyValueSerializer { @@ -1728,6 +1736,7 @@ class KeyValueSerializer { valueSerializer.writeMarshallable(value, out); } + }