From 8d44587c7c3a54fc173d455ef6769cca98a00d4a Mon Sep 17 00:00:00 2001 From: Rob Austin Date: Sat, 20 Dec 2014 13:27:59 +0000 Subject: [PATCH] HCOLL-197 - improved buffer resizing --- .../hash/replication/UdpTransportConfig.java | 38 +++++-- .../map/AbstractChannelReplicator.java | 6 +- .../chronicle/map/ChannelProvider.java | 5 +- .../openhft/chronicle/map/Replicators.java | 98 ++++++++++++++++--- .../chronicle/map/StatelessChronicleMap.java | 52 +++++++--- .../openhft/chronicle/map/TcpReplicator.java | 95 +++++++++++------- .../openhft/chronicle/map/UdpReplicator.java | 21 ++-- .../openhft/chronicle/map/WriterWithSize.java | 37 ++++++- .../openhft/chronicle/map/SerializerTest.java | 6 +- 9 files changed, 267 insertions(+), 91 deletions(-) diff --git a/src/main/java/net/openhft/chronicle/hash/replication/UdpTransportConfig.java b/src/main/java/net/openhft/chronicle/hash/replication/UdpTransportConfig.java index b32325673..b253e7bff 100644 --- a/src/main/java/net/openhft/chronicle/hash/replication/UdpTransportConfig.java +++ b/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); + } + } diff --git a/src/main/java/net/openhft/chronicle/map/AbstractChannelReplicator.java b/src/main/java/net/openhft/chronicle/map/AbstractChannelReplicator.java index 94c8ab798..4834549f2 100644 --- a/src/main/java/net/openhft/chronicle/map/AbstractChannelReplicator.java +++ b/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) { diff --git a/src/main/java/net/openhft/chronicle/map/ChannelProvider.java b/src/main/java/net/openhft/chronicle/map/ChannelProvider.java index b69f8c3bb..e24bf5890 100644 --- a/src/main/java/net/openhft/chronicle/map/ChannelProvider.java +++ b/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); } diff --git a/src/main/java/net/openhft/chronicle/map/Replicators.java b/src/main/java/net/openhft/chronicle/map/Replicators.java index f1c9aa232..e3f510797 100644 --- a/src/main/java/net/openhft/chronicle/map/Replicators.java +++ b/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); } }; } diff --git a/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java b/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java index c3c012792..f8f4cf2c5 100644 --- a/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java +++ b/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java @@ -139,10 +139,22 @@ class StatelessChronicleMap implements ChronicleMap, 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 implements ChronicleMap, 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 implements ChronicleMap, 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 implements ChronicleMap, 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 implements ChronicleMap, Closeable, Clon } private void resizeToMessageOutBuffer(long start, @NotNull IllegalArgumentException e) { + + e.printStackTrace(); assert outBytesLock.isHeldByCurrentThread(); assert !inBytesLock.isHeldByCurrentThread(); String message = e.getMessage(); diff --git a/src/main/java/net/openhft/chronicle/map/TcpReplicator.java b/src/main/java/net/openhft/chronicle/map/TcpReplicator.java index 66bb147d4..b0c3a1786 100644 --- a/src/main/java/net/openhft/chronicle/map/TcpReplicator.java +++ b/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 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 statelessClientParameters; + + static class StatelessClientParameters { + public StatelessClientParameters(VanillaChronicleMap map, + SerializationBuilder keySerializationBuilder, + SerializationBuilder valueSerializationBuilder) { + this.map = map; + this.keySerializationBuilder = keySerializationBuilder; + this.valueSerializationBuilder = valueSerializationBuilder; + } + + VanillaChronicleMap map; + SerializationBuilder keySerializationBuilder; + SerializationBuilder 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 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 { 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 { @NotNull private final ReaderWithSize keyReaderWithSize; + @NotNull private final WriterWithSize keyWriterWithSize; + @NotNull private final ReaderWithSize valueReaderWithSize; + @NotNull private final WriterWithSize valueWriterWithSize; + @NotNull private final VanillaChronicleMap map; - private final double maxEntrySizeBytes; + private final int tcpBufferSize; + StatelessServerConnector( - SerializationBuilder keySerializationBuilder, - SerializationBuilder valueSerializationBuilder, - @NotNull VanillaChronicleMap map, int maxEntrySizeBytes) { + @NotNull VanillaChronicleMap map, + @NotNull final BufferResizer bufferResizer, int tcpBufferSize, + final SerializationBuilder keySerializationBuilder, + final SerializationBuilder 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 { 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 { 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 { 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 { 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; } diff --git a/src/main/java/net/openhft/chronicle/map/UdpReplicator.java b/src/main/java/net/openhft/chronicle/map/UdpReplicator.java index a4cb64688..6060abb14 100644 --- a/src/main/java/net/openhft/chronicle/map/UdpReplicator.java +++ b/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(); diff --git a/src/main/java/net/openhft/chronicle/map/WriterWithSize.java b/src/main/java/net/openhft/chronicle/map/WriterWithSize.java index 4c946e7dd..9164fd5e5 100644 --- a/src/main/java/net/openhft/chronicle/map/WriterWithSize.java +++ b/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 { 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 serializationBuilder) { + WriterWithSize(SerializationBuilder 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 { 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 { @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 { return writeInLoop(out, t, writer, copies); } } + +interface BufferResizer { + + /** + * @param newCapacity + * @return the newly resize buffer + */ + public Bytes resizeBuffer(int newCapacity); +} diff --git a/src/test/java/net/openhft/chronicle/map/SerializerTest.java b/src/test/java/net/openhft/chronicle/map/SerializerTest.java index ce7d73b28..d8bfd6433 100644 --- a/src/test/java/net/openhft/chronicle/map/SerializerTest.java +++ b/src/test/java/net/openhft/chronicle/map/SerializerTest.java @@ -91,7 +91,7 @@ public class SerializerTest { builder.preMapConstruction(false); - WriterWithSize valueWriterWithSize = new WriterWithSize(builder.valueBuilder); + WriterWithSize valueWriterWithSize = new WriterWithSize(builder.valueBuilder,null); ReaderWithSize valueReaderWithSize = new ReaderWithSize(builder.valueBuilder); valueWriterWithSize.write(out, value, null); @@ -114,7 +114,7 @@ public class SerializerTest { builder.preMapConstruction(false); { - WriterWithSize keyWriterWithSize = new WriterWithSize(builder.keyBuilder); + WriterWithSize keyWriterWithSize = new WriterWithSize(builder.keyBuilder, null); ReaderWithSize keyReaderWithSize = new ReaderWithSize(builder.keyBuilder); keyWriterWithSize.write(out, key, null); @@ -207,7 +207,7 @@ public class SerializerTest { @Override public boolean equals(Object o) { if (this == o) return true; - if (o == null || getClass() != o.getClass()) return false; + if (o == null || MyTestClassMarshallable.class != o.getClass()) return false; MyTestClassMarshallable that = (MyTestClassMarshallable) o;