HCOLL-197 - improved buffer resizing

src/main/java/net/openhft/chronicle/hash/replication/UdpTransportConfig.java

@@ -25,16 +25,26 @@ import java.net.InetAddress;
 import java.net.NetworkInterface;
 
 public final class UdpTransportConfig {
+    public static final int DEFAULT_UDP_BUFFER_SIZE = 64 * 1024;
     private final InetAddress address;
     private final int port;
-    private final @Nullable NetworkInterface networkInterface;
+
+    @Nullable
+    private final
+    NetworkInterface networkInterface;
+
+    @NotNull
     private final ThrottlingConfig throttlingConfig;
 
+    private final int udpBufferSize;
+
     private UdpTransportConfig(
             InetAddress address,
             int port,
             @Nullable NetworkInterface networkInterface,
-            ThrottlingConfig throttlingConfig) {
+            ThrottlingConfig throttlingConfig,
+            int udpBufferSize) {
+        this.udpBufferSize = udpBufferSize;
         if (address == null) {
             throw new NullPointerException("Null address");
         }
@@ -49,22 +59,23 @@ public final class UdpTransportConfig {
     }
 
     public static UdpTransportConfig multiCast(@NotNull InetAddress address, int port,
-                                                 @NotNull NetworkInterface networkInterface) {
+                                               @Nullable NetworkInterface networkInterface) {
         if (!address.isMulticastAddress() || networkInterface == null)
             throw new IllegalArgumentException();
-        return create(address, port, networkInterface, ThrottlingConfig.noThrottling());
+        return create(address, port, networkInterface, ThrottlingConfig.noThrottling(), DEFAULT_UDP_BUFFER_SIZE);
     }
 
     public static UdpTransportConfig of(@NotNull InetAddress address, int port) {
         if (address.isMulticastAddress())
             throw new IllegalArgumentException();
-        return create(address, port, null, ThrottlingConfig.noThrottling());
+        return create(address, port, null, ThrottlingConfig.noThrottling(), DEFAULT_UDP_BUFFER_SIZE);
     }
 
     static UdpTransportConfig create(InetAddress address, int port,
                                      @Nullable NetworkInterface networkInterface,
-                                     ThrottlingConfig throttlingConfig) {
-        return new UdpTransportConfig(address, port, networkInterface, throttlingConfig);
+                                     ThrottlingConfig throttlingConfig,
+                                     final int udpBufferSize1) {
+        return new UdpTransportConfig(address, port, networkInterface, throttlingConfig, udpBufferSize1);
     }
 
     @NotNull
@@ -129,11 +140,20 @@ public final class UdpTransportConfig {
             @Nullable NetworkInterface networkInterface) {
         if (!address().isMulticastAddress())
             throw new IllegalArgumentException();
-        return create(address(), port(), networkInterface, throttlingConfig());
+        return create(address(), port(), networkInterface, throttlingConfig(), udpBufferSize());
     }
 
     public UdpTransportConfig throttlingConfig(@NotNull ThrottlingConfig throttlingConfig) {
         ThrottlingConfig.checkMillisecondBucketInterval(throttlingConfig, "UDP");
-        return create(address(), port(), networkInterface(), throttlingConfig);
+        return create(address(), port(), networkInterface(), throttlingConfig, udpBufferSize());
     }
+
+    public int udpBufferSize() {
+        return udpBufferSize;
+    }
+
+    public UdpTransportConfig udpBufferSize(int udpBufferSize) {
+        return create(address(), port(), networkInterface(), throttlingConfig(), udpBufferSize);
+    }
+
 }

src/main/java/net/openhft/chronicle/map/AbstractChannelReplicator.java

@@ -410,7 +410,7 @@ abstract class AbstractChannelReplicator implements Closeable {
         }
     }
 
-    static class EntryCallback extends Replica.EntryCallback {
+    static class EntryCallback extends Replica.EntryCallback implements BufferResizer {
 
         private final Replica.EntryExternalizable externalizable;
 
@@ -438,7 +438,7 @@ abstract class AbstractChannelReplicator implements Closeable {
         }
 
 
-        private void resizeBuffer(int size) {
+        public Bytes resizeBuffer(int size) {
 
             if (LOG.isDebugEnabled())
                 LOG.debug("resizing buffer to size=" + size);
@@ -474,7 +474,7 @@ abstract class AbstractChannelReplicator implements Closeable {
             assert out.capacity() == size;
             assert out.capacity() == in.capacity();
             assert in.limit() == in.capacity();
-
+            return in;
         }
 
         public boolean shouldBeIgnored(final Bytes entry, final int chronicleId) {

src/main/java/net/openhft/chronicle/map/ChannelProvider.java

@@ -60,11 +60,14 @@ final class ChannelProvider implements Closeable {
             return channelProvider;
         channelProvider = new ChannelProvider(hub);
         TcpTransportAndNetworkConfig tcpConfig = hub.tcpTransportAndNetwork();
+
+
         if (tcpConfig != null) {
+
             final TcpReplicator tcpReplicator = new TcpReplicator(
                     channelProvider.asReplica,
                     channelProvider.asEntryExternalizable,
-                    tcpConfig, null, hub.remoteNodeValidator());
+                    tcpConfig, hub.remoteNodeValidator(), null);
             channelProvider.add(tcpReplicator);
         }
 

src/main/java/net/openhft/chronicle/map/Replicators.java

@@ -19,43 +19,113 @@ package net.openhft.chronicle.map;
 import net.openhft.chronicle.hash.replication.AbstractReplication;
 import net.openhft.chronicle.hash.replication.TcpTransportAndNetworkConfig;
 import net.openhft.chronicle.hash.replication.UdpTransportConfig;
+import net.openhft.lang.io.ByteBufferBytes;
+import net.openhft.lang.io.Bytes;
 import org.jetbrains.annotations.NotNull;
 
 import java.io.Closeable;
 import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.nio.ByteOrder;
 
 final class Replicators {
 
     static final String ONLY_UDP_WARN_MESSAGE =
             "MISSING TCP REPLICATION : The UdpReplicator only attempts to read data " +
-            "(it does not enforce or guarantee delivery), you should use" +
-            "the UdpReplicator if you have a large number of nodes, and you wish" +
-            "to receive the data before it becomes available on TCP/IP. Since data" +
-            "delivery is not guaranteed, it is recommended that you only use" +
-            "the UDP Replicator in conjunction with a TCP Replicator";
+                    "(it does not enforce or guarantee delivery), you should use" +
+                    "the UdpReplicator if you have a large number of nodes, and you wish" +
+                    "to receive the data before it becomes available on TCP/IP. Since data" +
+                    "delivery is not guaranteed, it is recommended that you only use" +
+                    "the UDP Replicator in conjunction with a TCP Replicator";
 
     private Replicators() {
     }
 
+
+    static class OutBuffer implements BufferResizer {
+
+        @NotNull
+        private ByteBufferBytes in;
+
+        @NotNull
+        private ByteBuffer out;
+
+        @NotNull
+        public ByteBufferBytes in() {
+            return in;
+        }
+
+        @NotNull
+        public ByteBuffer out() {
+            return out;
+        }
+
+
+        OutBuffer(final int tcpBufferSize) {
+            out = ByteBuffer.allocateDirect(tcpBufferSize);
+            in = new ByteBufferBytes(out);
+        }
+
+        @Override
+        public Bytes resizeBuffer(int newCapacity) {
+
+
+            if (newCapacity < out.capacity())
+                throw new IllegalStateException("it not possible to resize the buffer smaller");
+
+            assert newCapacity < Integer.MAX_VALUE;
+
+            final ByteBuffer result = ByteBuffer.allocate(newCapacity).order(ByteOrder.nativeOrder());
+            final long bytesPosition = in.position();
+
+            in = new ByteBufferBytes(result.slice());
+
+            out.position(0);
+            out.limit((int) bytesPosition);
+
+            int numberOfLongs = (int) bytesPosition / 8;
+
+            // chunk in longs first
+            for (int i = 0; i < numberOfLongs; i++) {
+                in.writeLong(out.getLong());
+            }
+
+            for (int i = numberOfLongs * 8; i < bytesPosition; i++) {
+                in.writeByte(out.get());
+            }
+
+            out = result;
+
+            assert out.capacity() == in.capacity();
+
+            assert out.capacity() == newCapacity;
+            assert out.capacity() == in.capacity();
+            assert in.limit() == in.capacity();
+            return in;
+
+        }
+    }
+
     static Replicator tcp(final AbstractReplication replication) {
         return new Replicator() {
+
+
             @Override
             protected Closeable applyTo(@NotNull final ChronicleMapBuilder builder,
                                         @NotNull final Replica replica,
                                         @NotNull final Replica.EntryExternalizable entryExternalizable,
-                                        final ChronicleMap chronicleMap)
-                    throws IOException {
-
-                StatelessServerConnector statelessServer = new StatelessServerConnector(
-                        builder.keyBuilder, builder.valueBuilder,
-                                (VanillaChronicleMap) chronicleMap, builder.entrySize(true));
+                                        final ChronicleMap chronicleMap) throws IOException {
 
                 TcpTransportAndNetworkConfig tcpConfig = replication.tcpTransportAndNetwork();
-                assert tcpConfig != null;
+
+                TcpReplicator.StatelessClientParameters statelessClientParameters = new TcpReplicator.StatelessClientParameters(
+                        (VanillaChronicleMap) chronicleMap,
+                        builder.keyBuilder,
+                        builder.valueBuilder);
+
                 return new TcpReplicator(replica, entryExternalizable,
                         tcpConfig,
-                        statelessServer,
-                        replication.remoteNodeValidator());
+                        replication.remoteNodeValidator(), statelessClientParameters);
             }
         };
     }

src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java

@@ -139,10 +139,22 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable, Clon
 
         this.config = config;
 
-        keyReaderWithSize = new ReaderWithSize<>(chronicleMapBuilder.keyBuilder);
-        keyWriterWithSize = new WriterWithSize<>(chronicleMapBuilder.keyBuilder);
-        valueReaderWithSize = new ReaderWithSize<>(chronicleMapBuilder.valueBuilder);
-        valueWriterWithSize = new WriterWithSize<>(chronicleMapBuilder.valueBuilder);
+        final BufferResizer outBufferResizer = new BufferResizer() {
+            @Override
+            public Bytes resizeBuffer(int newCapacity) {
+                return resizeBufferOutBuffer(newCapacity);
+
+            }
+        };
+
+        keyReaderWithSize = new ReaderWithSize(chronicleMapBuilder.keyBuilder);
+        keyWriterWithSize = new WriterWithSize(chronicleMapBuilder.keyBuilder, outBufferResizer);
+
+        valueReaderWithSize = new ReaderWithSize(chronicleMapBuilder.valueBuilder);
+
+        valueWriterWithSize = new WriterWithSize(chronicleMapBuilder.valueBuilder,
+                outBufferResizer);
+
         this.putReturnsNull = chronicleMapBuilder.putReturnsNull();
         this.removeReturnsNull = chronicleMapBuilder.removeReturnsNull();
         this.maxEntrySize = chronicleMapBuilder.entrySize(true);
@@ -515,21 +527,25 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable, Clon
         return (long) ((double) outBytes.position() / percentageComplete);
     }
 
-    private void resizeBufferOutBuffer(int size, long start) {
+    private Bytes resizeBufferOutBuffer(int newCapacity) {
+        return resizeBufferOutBuffer(newCapacity, outBytes.position());
+    }
+
+    private Bytes resizeBufferOutBuffer(int newCapacity, long start) {
 
 
         assert outBytesLock.isHeldByCurrentThread();
         assert !inBytesLock.isHeldByCurrentThread();
 
         if (LOG.isDebugEnabled())
-            LOG.debug("resizing buffer to size=" + size + " ,name=" + name);
+            LOG.debug("resizing buffer to newCapacity=" + newCapacity + " ,name=" + name);
 
-        if (size < outBuffer.capacity())
+        if (newCapacity < outBuffer.capacity())
             throw new IllegalStateException("it not possible to resize the buffer smaller");
 
-        assert size < Integer.MAX_VALUE;
+        assert newCapacity < Integer.MAX_VALUE;
 
-        final ByteBuffer result = ByteBuffer.allocate(size).order(ByteOrder.nativeOrder());
+        final ByteBuffer result = ByteBuffer.allocate(newCapacity).order(ByteOrder.nativeOrder());
         final long bytesPosition = outBytes.position();
 
         outBytes = new ByteBufferBytes(result.slice());
@@ -552,25 +568,27 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable, Clon
 
         assert outBuffer.capacity() == outBytes.capacity();
 
-        assert outBuffer.capacity() == size;
+        assert outBuffer.capacity() == newCapacity;
         assert outBuffer.capacity() == outBytes.capacity();
         assert outBytes.limit() == outBytes.capacity();
         outBytes.position(start);
+        return outBytes;
     }
 
-    private void resizeBufferInBuffer(int size, long start) {
+
+    private void resizeBufferInBuffer(int newCapacity, long start) {
         assert !outBytesLock.isHeldByCurrentThread();
         assert inBytesLock.isHeldByCurrentThread();
 
         if (LOG.isDebugEnabled())
-            LOG.debug("InBuffer resizing buffer to size=" + size + " ,name=" + name);
+            LOG.debug("InBuffer resizing buffer to newCapacity=" + newCapacity + " ,name=" + name);
 
-        if (size < inBuffer.capacity())
+        if (newCapacity < inBuffer.capacity())
             throw new IllegalStateException("it not possible to resize the buffer smaller");
 
-        assert size < Integer.MAX_VALUE;
+        assert newCapacity < Integer.MAX_VALUE;
 
-        final ByteBuffer result = ByteBuffer.allocate(size).order(ByteOrder.nativeOrder());
+        final ByteBuffer result = ByteBuffer.allocate(newCapacity).order(ByteOrder.nativeOrder());
         final long bytesPosition = inBytes.position();
 
         inBytes = new ByteBufferBytes(result.slice());
@@ -592,7 +610,7 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable, Clon
         inBuffer = result;
 
         assert inBuffer.capacity() == inBytes.capacity();
-        assert inBuffer.capacity() == size;
+        assert inBuffer.capacity() == newCapacity;
         assert inBuffer.capacity() == inBytes.capacity();
         assert inBytes.limit() == inBytes.capacity();
 
@@ -1370,6 +1388,8 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable, Clon
     }
 
     private void resizeToMessageOutBuffer(long start, @NotNull IllegalArgumentException e) {
+
+        e.printStackTrace();
         assert outBytesLock.isHeldByCurrentThread();
         assert !inBytesLock.isHeldByCurrentThread();
         String message = e.getMessage();

src/main/java/net/openhft/chronicle/map/TcpReplicator.java

@@ -62,7 +62,7 @@ interface Work {
  *
  * @author Rob Austin.
  */
-final class TcpReplicator extends AbstractChannelReplicator implements Closeable {
+final class TcpReplicator<K, V> extends AbstractChannelReplicator implements Closeable {
 
     public static final long TIMESTAMP_FACTOR = 10000;
     private static final int STATELESS_CLIENT = -127;
@@ -86,8 +86,8 @@ final class TcpReplicator extends AbstractChannelReplicator implements Closeable
     private final Replica.EntryExternalizable externalizable;
     @NotNull
     private final TcpTransportAndNetworkConfig replicationConfig;
-    @Nullable
-    private final StatelessServerConnector statelessServerConnector;
+
+
     private final
     @Nullable
     RemoteNodeValidator remoteNodeValidator;
@@ -95,19 +95,34 @@ final class TcpReplicator extends AbstractChannelReplicator implements Closeable
 
     private long selectorTimeout;
 
+    StatelessClientParameters<K, V> statelessClientParameters;
+
+    static class StatelessClientParameters<K, V> {
+        public StatelessClientParameters(VanillaChronicleMap<K, ?, ?, V, ?, ?> map,
+                                         SerializationBuilder<K> keySerializationBuilder,
+                                         SerializationBuilder<V> valueSerializationBuilder) {
+            this.map = map;
+            this.keySerializationBuilder = keySerializationBuilder;
+            this.valueSerializationBuilder = valueSerializationBuilder;
+        }
+
+        VanillaChronicleMap<K, ?, ?, V, ?, ?> map;
+        SerializationBuilder<K> keySerializationBuilder;
+        SerializationBuilder<V> valueSerializationBuilder;
+    }
+
     /**
-     * @param statelessServerConnector set to NULL if not required
      * @throws IOException on an io error.
      */
     public TcpReplicator(@NotNull final Replica replica,
                          @NotNull final Replica.EntryExternalizable externalizable,
                          @NotNull final TcpTransportAndNetworkConfig replicationConfig,
-                         @Nullable StatelessServerConnector statelessServerConnector,
-                         @Nullable RemoteNodeValidator remoteNodeValidator) throws IOException {
+                         @Nullable final RemoteNodeValidator remoteNodeValidator,
+                         @Nullable final StatelessClientParameters statelessClientParameters) throws IOException {
 
-        super("TcpSocketReplicator-" + replica.identifier(), replicationConfig.throttlingConfig()
-        );
-        this.statelessServerConnector = statelessServerConnector;
+        super("TcpSocketReplicator-" + replica.identifier(), replicationConfig.throttlingConfig());
+
+        this.statelessClientParameters = statelessClientParameters;
 
         final ThrottlingConfig throttlingConfig = replicationConfig.throttlingConfig();
         long throttleBucketInterval = throttlingConfig.bucketInterval(MILLISECONDS);
@@ -881,7 +896,7 @@ final class TcpReplicator extends AbstractChannelReplicator implements Closeable
     /**
      * Attached to the NIO selection key via methods such as {@link SelectionKey#attach(Object)}
      */
-    class Attached implements Replica.ModificationNotifier {
+    class Attached<K, V> implements Replica.ModificationNotifier {
 
         public TcpSocketChannelEntryReader entryReader;
         public TcpSocketChannelEntryWriter entryWriter;
@@ -901,16 +916,6 @@ final class TcpReplicator extends AbstractChannelReplicator implements Closeable
             return handShakingComplete;
         }
 
-        void clearHandShaking() {
-            handShakingComplete = false;
-
-            remoteIdentifier = Byte.MIN_VALUE;
-            remoteBootstrapTimestamp = Long.MIN_VALUE;
-            remoteHeartbeatInterval = heartBeatIntervalMillis;
-            hasRemoteHeartbeatInterval = false;
-            remoteModificationIterator = null;
-        }
-
         /**
          * called whenever there is a change to the modification iterator
          */
@@ -934,9 +939,17 @@ final class TcpReplicator extends AbstractChannelReplicator implements Closeable
         @Nullable
         public Work uncompletedWork;
         private long lastSentTime;
+        StatelessServerConnector statelessServer;
 
         private TcpSocketChannelEntryWriter() {
             entryCallback = new EntryCallback(externalizable, replicationConfig.tcpBufferSize());
+
+            if (statelessClientParameters != null)
+                statelessServer = new StatelessServerConnector(
+                        statelessClientParameters.map, entryCallback,
+                        replicationConfig.tcpBufferSize(),
+                        statelessClientParameters.keySerializationBuilder,
+                        statelessClientParameters.valueSerializationBuilder);
         }
 
         public boolean isWorkIncomplete() {
@@ -960,6 +973,10 @@ final class TcpReplicator extends AbstractChannelReplicator implements Closeable
             return entryCallback.in();
         }
 
+        private ByteBuffer out() {
+            return entryCallback.out();
+        }
+
         /**
          * sends the identity and timestamp of this node to a remote node
          *
@@ -1071,9 +1088,6 @@ final class TcpReplicator extends AbstractChannelReplicator implements Closeable
             return len;
         }
 
-        private ByteBuffer out() {
-            return entryCallback.out();
-        }
 
         /**
          * used to send an single zero byte if we have not send any data for up to the
@@ -1128,7 +1142,7 @@ final class TcpReplicator extends AbstractChannelReplicator implements Closeable
     /**
      * Reads map entries from a socket, this could be a client or server socket
      */
-    class TcpSocketChannelEntryReader {
+    class TcpSocketChannelEntryReader<K, V> {
         public static final int HEADROOM = 1024;
         public long lastHeartBeatReceived = System.currentTimeMillis();
         ByteBuffer in;
@@ -1238,6 +1252,10 @@ final class TcpReplicator extends AbstractChannelReplicator implements Closeable
                     boolean isStatelessClient = (state != 1);
 
                     if (isStatelessClient) {
+
+                        final StatelessServerConnector statelessServerConnector = attached
+                                .entryWriter.statelessServer;
+
                         if (statelessServerConnector == null) {
                             LOG.error("", new IllegalArgumentException("received an event " +
                                     "from a stateless map, stateless maps are not " +
@@ -1338,26 +1356,33 @@ class StatelessServerConnector<K, V> {
 
     @NotNull
     private final ReaderWithSize<K> keyReaderWithSize;
+
     @NotNull
     private final WriterWithSize<K> keyWriterWithSize;
+
     @NotNull
     private final ReaderWithSize<V> valueReaderWithSize;
+
     @NotNull
     private final WriterWithSize<V> valueWriterWithSize;
+
     @NotNull
     private final VanillaChronicleMap<K, ?, ?, V, ?, ?> map;
-    private final double maxEntrySizeBytes;
+    private final int tcpBufferSize;
+
 
     StatelessServerConnector(
-            SerializationBuilder<K> keySerializationBuilder,
-            SerializationBuilder<V> valueSerializationBuilder,
-            @NotNull VanillaChronicleMap<K, ?, ?, V, ?, ?> map, int maxEntrySizeBytes) {
+            @NotNull VanillaChronicleMap<K, ?, ?, V, ?, ?> map,
+            @NotNull final BufferResizer bufferResizer, int tcpBufferSize,
+            final SerializationBuilder<K> keySerializationBuilder,
+            final SerializationBuilder<V> valueSerializationBuilder) {
+        this.tcpBufferSize = tcpBufferSize;
+
         keyReaderWithSize = new ReaderWithSize<>(keySerializationBuilder);
-        keyWriterWithSize = new WriterWithSize<>(keySerializationBuilder);
+        keyWriterWithSize = new WriterWithSize<>(keySerializationBuilder, bufferResizer);
         valueReaderWithSize = new ReaderWithSize<>(valueSerializationBuilder);
-        valueWriterWithSize = new WriterWithSize<>(valueSerializationBuilder);
+        valueWriterWithSize = new WriterWithSize<>(valueSerializationBuilder, bufferResizer);
         this.map = map;
-        this.maxEntrySizeBytes = maxEntrySizeBytes;
     }
 
     @Nullable
@@ -1784,7 +1809,7 @@ class StatelessServerConnector<K, V> {
                 while (iterator.hasNext()) {
                     // we've filled up the buffer, so lets give another channel a chance to send
                     // some data, we don't know the max key size, we will use the entrySize instead
-                    if (out.remaining() <= maxEntrySizeBytes) {
+                    if (out.position() > tcpBufferSize) {
                         writeHeader(out, sizeLocation, count, true);
                         return false;
                     }
@@ -1826,7 +1851,7 @@ class StatelessServerConnector<K, V> {
                 while (iterator.hasNext()) {
                     // we've filled up the buffer, so lets give another channel a chance to send
                     // some data, we don't know the max key size, we will use the entrySize instead
-                    if (out.remaining() <= maxEntrySizeBytes) {
+                    if (out.position() > tcpBufferSize) {
                         writeHeader(out, sizeLocation, count, true);
                         return false;
                     }
@@ -1859,7 +1884,7 @@ class StatelessServerConnector<K, V> {
         return new Work() {
             @Override
             public boolean doWork(@NotNull Bytes out) {
-                if (out.remaining() <= maxEntrySizeBytes)
+                if (out.position() > tcpBufferSize)
                     return false;
 
                 final long sizeLocation = header(out, transactionId);
@@ -1873,7 +1898,7 @@ class StatelessServerConnector<K, V> {
                 while (iterator.hasNext()) {
                     // we've filled up the buffer, so lets give another channel a chance to send
                     // some data, we don't know the max key size, we will use the entrySize instead
-                    if (out.remaining() <= maxEntrySizeBytes) {
+                    if (out.position() > tcpBufferSize) {
                         writeHeader(out, sizeLocation, count, true);
                         return false;
                     }

src/main/java/net/openhft/chronicle/map/UdpReplicator.java

@@ -39,6 +39,9 @@ final class UdpReplicator extends UdpChannelReplicator implements Replica.Modifi
 
     private static final Logger LOG =
             LoggerFactory.getLogger(UdpReplicator.class.getName());
+    public static final int UPD_BUFFER_SIZE = 64 * 1204;
+    @NotNull
+    private final UdpTransportConfig replicationConfig;
 
     public UdpReplicator(@NotNull final Replica replica,
                          @NotNull final Replica.EntryExternalizable entryExternalizable,
@@ -47,14 +50,16 @@ final class UdpReplicator extends UdpChannelReplicator implements Replica.Modifi
             throws IOException {
 
         super(replicationConfig, serializedEntrySize, replica.identifier());
+        this.replicationConfig = replicationConfig;
 
         Replica.ModificationIterator modificationIterator = replica.acquireModificationIterator(
                 ChronicleMapBuilder.UDP_REPLICATION_MODIFICATION_ITERATOR_ID, this);
 
         setReader(new UdpSocketChannelEntryReader(serializedEntrySize, entryExternalizable));
 
-        setWriter(new UdpSocketChannelEntryWriter(serializedEntrySize, entryExternalizable,
-                modificationIterator, this));
+        setWriter(new UdpSocketChannelEntryWriter(replicationConfig.udpBufferSize(),
+                entryExternalizable,
+                modificationIterator, this, new Replicators.OutBuffer(UPD_BUFFER_SIZE)));
 
         start();
     }
@@ -121,17 +126,19 @@ final class UdpReplicator extends UdpChannelReplicator implements Replica.Modifi
 
         private final EntryCallback entryCallback;
         private final UdpChannelReplicator udpReplicator;
+        private final Replicators.OutBuffer outBuffer;
         private Replica.ModificationIterator modificationIterator;
 
-        UdpSocketChannelEntryWriter(final int serializedEntrySize,
+        UdpSocketChannelEntryWriter(final int updBufferSize,
                                     @NotNull final Replica.EntryExternalizable externalizable,
                                     @NotNull final Replica.ModificationIterator modificationIterator,
-                                    UdpChannelReplicator udpReplicator) {
+                                    UdpChannelReplicator udpReplicator, Replicators.OutBuffer outBuffer) {
             this.udpReplicator = udpReplicator;
 
+            this.outBuffer = outBuffer;
             // we make the buffer twice as large just to give ourselves headroom
 
-            entryCallback = new EntryCallback(externalizable, serializedEntrySize);
+            entryCallback = new EntryCallback(externalizable, updBufferSize);
             this.modificationIterator = modificationIterator;
         }
 
@@ -154,8 +161,8 @@ final class UdpReplicator extends UdpChannelReplicator implements Replica.Modifi
         public int writeAll(@NotNull final DatagramChannel socketChannel)
                 throws InterruptedException, IOException {
 
-            final ByteBufferBytes in = entryCallback.in();
-            final ByteBuffer out = entryCallback.out();
+            final ByteBufferBytes in = outBuffer.in();
+            final ByteBuffer out = outBuffer.out();
 
             out.clear();
             in.clear();

src/main/java/net/openhft/chronicle/map/WriterWithSize.java

@@ -25,14 +25,19 @@ import net.openhft.lang.threadlocal.Provider;
 import net.openhft.lang.threadlocal.ThreadLocalCopies;
 import org.jetbrains.annotations.Nullable;
 
+import java.nio.BufferOverflowException;
+
 final class WriterWithSize<T> {
     private final SizeMarshaller sizeMarshaller;
     private final Object originalWriter;
     private final Provider writerProvider;
     private final MetaBytesInterop originalMetaWriter;
     private final MetaProvider metaWriterProvider;
+    private final BufferResizer bufferResizer;
 
-    WriterWithSize(SerializationBuilder<T> serializationBuilder) {
+    WriterWithSize(SerializationBuilder<T> serializationBuilder,
+                   @Nullable BufferResizer bufferResizer) {
+        this.bufferResizer = bufferResizer;
         sizeMarshaller = serializationBuilder.sizeMarshaller();
         originalWriter = serializationBuilder.interop();
         writerProvider = Provider.of((Class) originalWriter.getClass());
@@ -50,11 +55,25 @@ final class WriterWithSize<T> {
         Object writer = writerProvider.get(copies, originalWriter);
         copies = metaWriterProvider.getCopies(copies);
         MetaBytesWriter metaWriter = metaWriterProvider.get(copies, originalMetaWriter, writer, t);
-        sizeMarshaller.writeSize(out, metaWriter.size(writer, t));
+        long size = metaWriter.size(writer, t);
+        if (bufferResizer != null)
+            out = resizeIfRequired(out, size);
+        sizeMarshaller.writeSize(out, size);
         metaWriter.write(writer, out, t);
         return copies;
     }
 
+    private Bytes resizeIfRequired(Bytes out, long size) {
+        if (out.remaining() < size + 8) {
+            long newCapacity = out.remaining() + size + 8;
+            if (newCapacity > Integer.MAX_VALUE)
+                throw new BufferOverflowException();
+             return bufferResizer.resizeBuffer((int) newCapacity);
+        }
+        return out;
+    }
+
+
     public ThreadLocalCopies writeNullable(Bytes out, T t, @Nullable ThreadLocalCopies copies) {
         out.writeBoolean(t == null);
         if (t == null)
@@ -71,7 +90,10 @@ final class WriterWithSize<T> {
                                          @Nullable ThreadLocalCopies copies) {
         copies = metaWriterProvider.getCopies(copies);
         MetaBytesWriter metaWriter = metaWriterProvider.get(copies, originalMetaWriter, writer, t);
-        sizeMarshaller.writeSize(out, metaWriter.size(writer, t));
+        long size = metaWriter.size(writer, t);
+        if (bufferResizer != null)
+            out = resizeIfRequired(out, size);
+        sizeMarshaller.writeSize(out, size);
         metaWriter.write(writer, out, t);
         return copies;
     }
@@ -84,3 +106,12 @@ final class WriterWithSize<T> {
         return writeInLoop(out, t, writer, copies);
     }
 }
+
+interface BufferResizer {
+
+    /**
+     * @param newCapacity
+     * @return the newly resize buffer
+     */
+    public Bytes resizeBuffer(int newCapacity);
+}