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 extends K, ? extends V> map) {
- for (Map.Entry extends K, ? extends V> 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 extends K, ? extends V> 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 extends K, ? extends V> 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);
}
+
}