From 005f4d7a7e31c7a91fb35ff8e8dc4e8288a46eba Mon Sep 17 00:00:00 2001 From: Rob Austin Date: Tue, 16 Dec 2014 15:28:52 +0000 Subject: [PATCH] HCOLL-246 Fixed - Tcp Replication fails with large values or entries sizes that don't included internal replication bytes" --- .../TcpTransportAndNetworkConfig.java | 2 +- .../map/AbstractChannelReplicator.java | 93 ++++++++++++++--- .../chronicle/map/ChannelProvider.java | 25 ++++- .../net/openhft/chronicle/map/Replica.java | 99 +++++++++++-------- .../chronicle/map/ReplicatedChronicleMap.java | 38 ++++++- .../openhft/chronicle/map/Replicators.java | 2 +- .../openhft/chronicle/map/TcpReplicator.java | 77 ++++++++------- .../chronicle/map/UdpChannelReplicator.java | 4 +- .../openhft/chronicle/map/UdpReplicator.java | 21 ++-- .../chronicle/map/TcpReplicationSoakTest.java | 33 ++++--- 10 files changed, 276 insertions(+), 118 deletions(-) diff --git a/src/main/java/net/openhft/chronicle/hash/replication/TcpTransportAndNetworkConfig.java b/src/main/java/net/openhft/chronicle/hash/replication/TcpTransportAndNetworkConfig.java index a0b9f6261..926401cfe 100644 --- a/src/main/java/net/openhft/chronicle/hash/replication/TcpTransportAndNetworkConfig.java +++ b/src/main/java/net/openhft/chronicle/hash/replication/TcpTransportAndNetworkConfig.java @@ -134,7 +134,7 @@ public final class TcpTransportAndNetworkConfig implements Serializable { heartBeatIntervalUnit); } - public int packetSize() { + public int tcpBufferSize() { return tcpBufferSize; } diff --git a/src/main/java/net/openhft/chronicle/map/AbstractChannelReplicator.java b/src/main/java/net/openhft/chronicle/map/AbstractChannelReplicator.java index 8f87caed2..390337a06 100644 --- a/src/main/java/net/openhft/chronicle/map/AbstractChannelReplicator.java +++ b/src/main/java/net/openhft/chronicle/map/AbstractChannelReplicator.java @@ -32,6 +32,8 @@ import java.io.Closeable; import java.io.IOException; import java.lang.reflect.Field; import java.net.InetSocketAddress; +import java.nio.ByteBuffer; +import java.nio.ByteOrder; import java.nio.channels.*; import java.security.AccessController; import java.security.PrivilegedAction; @@ -76,8 +78,7 @@ abstract class AbstractChannelReplicator implements Closeable { ThreadLocalCopies copies; VanillaChronicleMap.SegmentState segmentState; - AbstractChannelReplicator(String name, ThrottlingConfig throttlingConfig, - int maxEntrySizeBytes) + AbstractChannelReplicator(String name, ThrottlingConfig throttlingConfig) throws IOException { executorService = Executors.newSingleThreadExecutor( new NamedThreadFactory(name, true) { @@ -90,7 +91,7 @@ abstract class AbstractChannelReplicator implements Closeable { throttler = throttlingConfig.throttling(DAYS) > 0 ? new Throttler(selector, throttlingConfig.bucketInterval(MILLISECONDS), - maxEntrySizeBytes, throttlingConfig.throttling(DAYS)) : null; + throttlingConfig.throttling(DAYS)) : null; startedHere = new Throwable("Started here"); } @@ -317,12 +318,11 @@ abstract class AbstractChannelReplicator implements Closeable { Throttler(@NotNull Selector selector, long throttleIntervalInMillis, - long serializedEntrySize, long bitsPerDay) { this.selector = selector; this.throttleInterval = throttleIntervalInMillis; double bytesPerMs = ((double) bitsPerDay) / DAYS.toMillis(1) / BITS_IN_A_BYTE; - this.maxBytesInInterval = round(bytesPerMs * throttleInterval) - serializedEntrySize; + this.maxBytesInInterval = round(bytesPerMs * throttleInterval); } public void add(SelectableChannel selectableChannel) { @@ -413,16 +413,74 @@ abstract class AbstractChannelReplicator implements Closeable { static class EntryCallback extends Replica.EntryCallback { private final Replica.EntryExternalizable externalizable; - private final ByteBufferBytes in; + + @NotNull + private ByteBufferBytes in; + + @NotNull + private ByteBuffer out; EntryCallback(@NotNull final Replica.EntryExternalizable externalizable, - @NotNull final ByteBufferBytes in) { + final int tcpBufferSize) { this.externalizable = externalizable; - this.in = in; + out = ByteBuffer.allocateDirect(tcpBufferSize); + in = new ByteBufferBytes(out); } + @NotNull + public ByteBufferBytes in() { + return in; + } + + @NotNull + public ByteBuffer out() { + return out; + } + + + private void resizeBuffer(int size) { + + if (LOG.isDebugEnabled()) + LOG.debug("resizing buffer to size=" + size); + + if (size < out.capacity()) + throw new IllegalStateException("it not possible to resize the buffer smaller"); + + assert size < Integer.MAX_VALUE; + + final ByteBuffer result = ByteBuffer.allocate(size).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() == size; + assert out.capacity() == in.capacity(); + assert in.limit() == in.capacity(); + + } + + @Override public boolean onEntry(final Bytes entry, final int chronicleId) { + long pos0 = in.position(); // used to denote that this is not a stateless map event @@ -433,7 +491,16 @@ abstract class AbstractChannelReplicator implements Closeable { // this is where we will store the size of the entry in.skip(SIZE_OF_SIZE); + int entrySize = externalizable.sizeOfEntry(entry, chronicleId); + + if (entrySize > in.remaining()) { + long newSize = in.position() + entrySize; + assert newSize < Integer.MAX_VALUE; + resizeBuffer((int) newSize); + } + final long start = in.position(); + externalizable.writeExternalEntry(entry, in, chronicleId); if (in.position() == start) { @@ -443,16 +510,16 @@ abstract class AbstractChannelReplicator implements Closeable { // write the length of the entry, just before the start, so when we read it back // we read the length of the entry first and hence know how many preceding writer to read - final long entrySize = (int) (in.position() - start); - if (entrySize > Integer.MAX_VALUE) + final long bytesWritten = (int) (in.position() - start); + + if (bytesWritten > Integer.MAX_VALUE) throw new IllegalStateException("entry too large, the entry size=" + entrySize + ", " + "entries are limited to a size of " + Integer.MAX_VALUE); - int entrySize1 = (int) entrySize; if (LOG.isDebugEnabled()) - LOG.debug("sending entry of entrySize=" + entrySize1); + LOG.debug("sending entry of entrySize=" + (int) bytesWritten); - in.writeInt(sizeLocation, entrySize1); + in.writeInt(sizeLocation, (int) bytesWritten); return true; } diff --git a/src/main/java/net/openhft/chronicle/map/ChannelProvider.java b/src/main/java/net/openhft/chronicle/map/ChannelProvider.java index bedc095cb..addfb502e 100644 --- a/src/main/java/net/openhft/chronicle/map/ChannelProvider.java +++ b/src/main/java/net/openhft/chronicle/map/ChannelProvider.java @@ -64,7 +64,7 @@ final class ChannelProvider implements Closeable { final TcpReplicator tcpReplicator = new TcpReplicator( channelProvider.asReplica, channelProvider.asEntryExternalizable, - tcpConfig, hub.maxEntrySize(), null, hub.remoteNodeValidator()); + tcpConfig, null, hub.remoteNodeValidator()); channelProvider.add(tcpReplicator); } @@ -85,6 +85,18 @@ final class ChannelProvider implements Closeable { private static final byte BOOTSTRAP_MESSAGE = 'B'; final EntryExternalizable asEntryExternalizable = new EntryExternalizable() { + @Override + public int sizeOfEntry(@NotNull Bytes entry, int chronicleChannel) { + channelDataLock.readLock().lock(); + try { + + return channelEntryExternalizables[chronicleChannel] + .sizeOfEntry(entry, chronicleChannel); + } finally { + channelDataLock.readLock().unlock(); + } + } + /** * writes the entry to the chronicle channel provided * @@ -116,7 +128,7 @@ final class ChannelProvider implements Closeable { final int chronicleId = (int) source.readStopBit(); if (chronicleId < chronicleChannels.length) { // this channel is has not currently been created so it updates will be ignored - if (channelEntryExternalizables[chronicleId] != null) + if (channelEntryExternalizables[chronicleId] != null) channelEntryExternalizables[chronicleId] .readExternalEntry(copies, segmentState, source); } else @@ -398,8 +410,8 @@ final class ChannelProvider implements Closeable { } /** - * used to send system messages such as bootstrap from one remote node to another, it also can be used in - * a broadcast context + * used to send system messages such as bootstrap from one remote node to another, it also can + * be used in a broadcast context */ class SystemQueue { @@ -467,6 +479,11 @@ final class ChannelProvider implements Closeable { } }; final EntryExternalizable asEntryExternalizable = new EntryExternalizable() { + @Override + public int sizeOfEntry(@NotNull Bytes entry, int chronicleId) { + return (int) entry.remaining(); + } + @Override public void writeExternalEntry(@NotNull Bytes entry, @NotNull Bytes destination, int na) { diff --git a/src/main/java/net/openhft/chronicle/map/Replica.java b/src/main/java/net/openhft/chronicle/map/Replica.java index 7ceb3b66b..9f328e74f 100644 --- a/src/main/java/net/openhft/chronicle/map/Replica.java +++ b/src/main/java/net/openhft/chronicle/map/Replica.java @@ -30,39 +30,43 @@ import java.io.Closeable; interface Replica extends Closeable { /** - * Provides the unique Identifier associated with this map instance.

An identifier is used to - * determine which replicating node made the change.

If two nodes update their map at the same time - * with different values, we have to deterministically resolve which update wins, because of eventual - * consistency both nodes should end up locally holding the same data. Although it is rare two remote - * nodes could receive an update to their maps at exactly the same time for the same key, we have to - * handle this edge case, its therefore important not to rely on timestamps alone to reconcile the - * updates. Typically the update with the newest timestamp should win, but in this example both - * timestamps are the same, and the decision made to one node should be identical to the decision made to - * the other. We resolve this simple dilemma by using a node identifier, each node will have a unique - * identifier, the update from the node with the smallest identifier wins. + * Provides the unique Identifier associated with this map instance.

An identifier is used + * to determine which replicating node made the change.

If two nodes update their map at the + * same time with different values, we have to deterministically resolve which update wins, + * because of eventual consistency both nodes should end up locally holding the same data. + * Although it is rare two remote nodes could receive an update to their maps at exactly the + * same time for the same key, we have to handle this edge case, its therefore important not to + * rely on timestamps alone to reconcile the updates. Typically the update with the newest + * timestamp should win, but in this example both timestamps are the same, and the decision + * made to one node should be identical to the decision made to the other. We resolve this + * simple dilemma by using a node identifier, each node will have a unique identifier, the + * update from the node with the smallest identifier wins. * * @return the unique Identifier associated with this map instance */ byte identifier(); /** - * Gets (if it does not exist, creates) an instance of ModificationIterator associated with a remote node, - * this weak associated is bound using the {@code identifier}. + * Gets (if it does not exist, creates) an instance of ModificationIterator associated with a + * remote node, this weak associated is bound using the {@code identifier}. * * @param remoteIdentifier the identifier of the remote node - * @param modificationNotifier called when ever there is a change applied to the modification iterator - * @return the ModificationIterator dedicated for replication to the remote node with the given identifier + * @param modificationNotifier called when ever there is a change applied to the modification + * iterator + * @return the ModificationIterator dedicated for replication to the remote node with the given + * identifier * @see #identifier() */ ModificationIterator acquireModificationIterator(byte remoteIdentifier, ModificationNotifier modificationNotifier); /** - * Returns the timestamp of the last change from the specified remote node, already replicated to this - * Replica.

Used in conjunction with replication, to back fill data from a remote node. This node - * may have missed updates while it was not been running or connected via TCP. + * Returns the timestamp of the last change from the specified remote node, already replicated + * to this Replica.

Used in conjunction with replication, to back fill data from a remote + * node. This node may have missed updates while it was not been running or connected via TCP. * - * @param remoteIdentifier the identifier of the remote node to check last replicated update time from + * @param remoteIdentifier the identifier of the remote node to check last replicated update + * time from * @return a timestamp of the last modification to an entry, or 0 if there are no entries. * @see #identifier() */ @@ -97,20 +101,21 @@ interface Replica extends Closeable { boolean hasNext(); /** - * A non-blocking call that provides the entry that has changed to {@code callback.onEntry()}. + * A non-blocking call that provides the entry that has changed to {@code + * callback.onEntry()}. * * @param callback a callback which will be called when a new entry becomes available. * @param chronicleId only assigned when using chronicle channels - * @return {@code true} if the entry was accepted by the {@code callback.onEntry()} method, {@code - * false} if the entry was not accepted or was not available + * @return {@code true} if the entry was accepted by the {@code callback.onEntry()} method, + * {@code false} if the entry was not accepted or was not available */ boolean nextEntry(@NotNull final EntryCallback callback, final int chronicleId); /** - * Dirties all entries with a modification time equal to {@code fromTimeStamp} or newer. It means all - * these entries will be considered as "new" by this ModificationIterator and iterated once again no - * matter if they have already been.

This functionality is used to publish recently modified - * entries to a new remote node as it connects. + * Dirties all entries with a modification time equal to {@code fromTimeStamp} or newer. It + * means all these entries will be considered as "new" by this ModificationIterator and + * iterated once again no matter if they have already been.

This functionality is used + * to publish recently modified entries to a new remote node as it connects. * * @param fromTimeStamp the timestamp from which all entries should be dirty */ @@ -122,25 +127,35 @@ interface Replica extends Closeable { */ interface EntryExternalizable { + /** + * @param entry an entry in the map + * @param chronicleId is the channel id used to identify the canonical map or queue + * @return the size of the entry + */ + int sizeOfEntry(@NotNull Bytes entry, int chronicleId); + /** * The map implements this method to save its contents. * * @param entry the byte location of the entry to be stored - * @param destination a buffer the entry will be written to, the segment may reject this operation and - * add zeroBytes, if the identifier in the entry did not match the maps local + * @param destination a buffer the entry will be written to, the segment may reject this + * operation and add zeroBytes, if the identifier in the entry did not + * match the maps local * @param chronicleId is the channel id used to identify the canonical map or queue */ - void writeExternalEntry(@NotNull Bytes entry, @NotNull Bytes destination, int chronicleId); + void writeExternalEntry(@NotNull Bytes entry, @NotNull Bytes destination, int + chronicleId); /** - * The map implements this method to restore its contents. This method must read the values in the - * same sequence and with the same types as were written by {@code writeExternalEntry()}. This method - * is typically called when we receive a remote replication event, this event could originate from - * either a remote {@code put(K key, V value)} or {@code remove(Object key)} + * The map implements this method to restore its contents. This method must read the values + * in the same sequence and with the same types as were written by {@code + * writeExternalEntry()}. This method is typically called when we receive a remote + * replication event, this event could originate from either a remote {@code put(K key, V + *value)} or {@code remove(Object key)} * * @param copies * @param segmentState - * @param source bytes to read an entry from + * @param source bytes to read an entry from */ void readExternalEntry( @NotNull ThreadLocalCopies copies, @@ -148,8 +163,8 @@ interface Replica extends Closeable { } /** - * provides a key and value from NativeBytes, this can be used in conjunction with the modification - * iterator to get the key and value out of the NativeBytes + * provides a key and value from NativeBytes, this can be used in conjunction with the + * modification iterator to get the key and value out of the NativeBytes */ interface EntryResolver { @@ -175,20 +190,20 @@ interface Replica extends Closeable { } /** - * Implemented typically by a replicator, This interface provides the event, which will get called - * whenever a put() or remove() has occurred to the map + * Implemented typically by a replicator, This interface provides the event, which will get + * called whenever a put() or remove() has occurred to the map */ abstract class EntryCallback { /** * Called whenever a put() or remove() has occurred to a replicating map. * - * @param entry the entry you will receive, this does not have to be locked, as locking is - * already provided from the caller. + * @param entry the entry you will receive, this does not have to be locked, as + * locking is already provided from the caller. * @param chronicleId only assigned when clustering - * @return {@code false} if this entry should be ignored because the identifier of the source node is - * not from one of our changes, WARNING even though we check the identifier in the - * ModificationIterator the entry may have been updated. + * @return {@code false} if this entry should be ignored because the identifier of the + * source node is not from one of our changes, WARNING even though we check the identifier + * in the ModificationIterator the entry may have been updated. */ public abstract boolean onEntry(final Bytes entry, final int chronicleId); diff --git a/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java b/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java index 1f5da6bd6..ab5473a06 100644 --- a/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java +++ b/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java @@ -422,6 +422,41 @@ final class ReplicatedChronicleMap= 0L; ) { long offset = offsetFromPos(pos); if (entry == null) { diff --git a/src/main/java/net/openhft/chronicle/map/Replicators.java b/src/main/java/net/openhft/chronicle/map/Replicators.java index 1db4347a9..f1c9aa232 100644 --- a/src/main/java/net/openhft/chronicle/map/Replicators.java +++ b/src/main/java/net/openhft/chronicle/map/Replicators.java @@ -54,7 +54,7 @@ final class Replicators { assert tcpConfig != null; return new TcpReplicator(replica, entryExternalizable, tcpConfig, - builder.entrySize(true), statelessServer, + statelessServer, replication.remoteNodeValidator()); } }; diff --git a/src/main/java/net/openhft/chronicle/map/TcpReplicator.java b/src/main/java/net/openhft/chronicle/map/TcpReplicator.java index 5c4def7b3..b09c2effe 100644 --- a/src/main/java/net/openhft/chronicle/map/TcpReplicator.java +++ b/src/main/java/net/openhft/chronicle/map/TcpReplicator.java @@ -76,11 +76,12 @@ final class TcpReplicator extends AbstractChannelReplicator implements Closeable private final KeyInterestUpdater opWriteUpdater = new KeyInterestUpdater(OP_WRITE, selectionKeysStore); private final BitSet activeKeys = new BitSet(selectionKeysStore.length); private final long heartBeatIntervalMillis; + private long largestEntrySoFar = 128; @NotNull private final Replica replica; private final byte localIdentifier; - private final int maxEntrySizeBytes; + @NotNull private final Replica.EntryExternalizable externalizable; @NotNull @@ -95,23 +96,17 @@ final class TcpReplicator extends AbstractChannelReplicator implements Closeable private long selectorTimeout; /** - * @param maxEntrySizeBytes used to check that the last entry will fit into the buffer, - * it can not be smaller than the size of and entry, if it is - * set smaller the buffer will over flow, it can be larger then - * the entry, but setting it too large reduces the workable - * space in the buffer. * @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, - final int maxEntrySizeBytes, @Nullable StatelessServerConnector statelessServerConnector, @Nullable RemoteNodeValidator remoteNodeValidator) throws IOException { - super("TcpSocketReplicator-" + replica.identifier(), replicationConfig.throttlingConfig(), - maxEntrySizeBytes); + super("TcpSocketReplicator-" + replica.identifier(), replicationConfig.throttlingConfig() + ); this.statelessServerConnector = statelessServerConnector; final ThrottlingConfig throttlingConfig = replicationConfig.throttlingConfig(); @@ -123,7 +118,7 @@ final class TcpReplicator extends AbstractChannelReplicator implements Closeable this.replica = replica; this.localIdentifier = replica.identifier(); - this.maxEntrySizeBytes = maxEntrySizeBytes; + this.externalizable = externalizable; this.replicationConfig = replicationConfig; @@ -767,7 +762,7 @@ final class TcpReplicator extends AbstractChannelReplicator implements Closeable serverChannel.socket().setReceiveBufferSize(BUFFER_SIZE); serverChannel.configureBlocking(false); - serverChannel.register(TcpReplicator.this.selector,0); + serverChannel.register(TcpReplicator.this.selector, 0); ServerSocket serverSocket = null; try { @@ -926,10 +921,7 @@ final class TcpReplicator extends AbstractChannelReplicator implements Closeable */ private class TcpSocketChannelEntryWriter { - @NotNull - final ByteBufferBytes in; - @NotNull - private final ByteBuffer out; + @NotNull private final EntryCallback entryCallback; // if uncompletedWork is set ( not null ) , this must be completed before any further work @@ -939,9 +931,7 @@ final class TcpReplicator extends AbstractChannelReplicator implements Closeable private long lastSentTime; private TcpSocketChannelEntryWriter() { - out = ByteBuffer.allocateDirect(replicationConfig.packetSize() + maxEntrySizeBytes); - in = new ByteBufferBytes(out); - entryCallback = new EntryCallback(externalizable, in); + entryCallback = new EntryCallback(externalizable, replicationConfig.tcpBufferSize()); } public boolean isWorkIncomplete() { @@ -958,7 +948,11 @@ final class TcpReplicator extends AbstractChannelReplicator implements Closeable * @param localIdentifier the current nodes identifier */ void identifierToBuffer(final byte localIdentifier) { - in.writeByte(localIdentifier); + in().writeByte(localIdentifier); + } + + private Bytes in() { + return entryCallback.in(); } /** @@ -967,9 +961,10 @@ final class TcpReplicator extends AbstractChannelReplicator implements Closeable * @param timeStampOfLastMessage the last timestamp we received a message from that node */ void writeRemoteBootstrapTimestamp(final long timeStampOfLastMessage) { - in.writeLong(timeStampOfLastMessage); + in().writeLong(timeStampOfLastMessage); } + /** * writes all the entries that have changed, to the buffer which will later be written to * TCP/IP @@ -989,12 +984,19 @@ final class TcpReplicator extends AbstractChannelReplicator implements Closeable int entriesWritten = 0; try { for (; ; entriesWritten++) { - final boolean wasDataRead = modificationIterator.nextEntry(entryCallback, 0); - if (!wasDataRead) { + long start = in().position(); + final boolean isMoreDataAvailable = modificationIterator.nextEntry(entryCallback, 0); + long entrySize = in().position() - start; + + if (entrySize > largestEntrySoFar) + largestEntrySoFar = entrySize; + + + if (!isMoreDataAvailable) { // if we have no more data to write to the socket then we will // un-register OP_WRITE on the selector, until more data becomes available - if (in.position() == 0 && handShakingComplete) + if (in().position() == 0 && handShakingComplete) disableWrite(socketChannel, attached); return; @@ -1002,7 +1004,7 @@ final class TcpReplicator extends AbstractChannelReplicator implements Closeable // we've filled up the buffer lets give another channel a chance to send // some data - if (in.remaining() <= maxEntrySizeBytes) + if (in().remaining() <= largestEntrySoFar) return; // if we have space in the buffer to write more data and we just wrote data @@ -1024,6 +1026,9 @@ final class TcpReplicator extends AbstractChannelReplicator implements Closeable private int writeBufferToSocket(@NotNull final SocketChannel socketChannel, final long approxTime) throws IOException { + final Bytes in = in(); + final ByteBuffer out = out(); + if (in.position() == 0) return 0; @@ -1031,6 +1036,7 @@ final class TcpReplicator extends AbstractChannelReplicator implements Closeable lastSentTime = approxTime; assert in.position() <= Integer.MAX_VALUE; int size = (int) in.position(); + out.limit(size); final int len = socketChannel.write(out); @@ -1051,20 +1057,24 @@ 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 * localHeartbeatInterval */ private void writeHeartbeatToBuffer() { // denotes the state - 0 for a heartbeat - in.writeByte(HEARTBEAT.ordinal()); + in().writeByte(HEARTBEAT.ordinal()); // denotes the size in bytes - in.writeInt(0); + in().writeInt(0); } private void writeRemoteHeartbeatInterval(long localHeartbeatInterval) { - in.writeLong(localHeartbeatInterval); + in().writeLong(localHeartbeatInterval); } /** @@ -1093,11 +1103,11 @@ final class TcpReplicator extends AbstractChannelReplicator implements Closeable } public boolean doWork() { - return uncompletedWork != null && uncompletedWork.doWork(in); + return uncompletedWork != null && uncompletedWork.doWork(in()); } public boolean hasBytesToWrite() { - return in.position() > 0; + return in().position() > 0; } } @@ -1113,7 +1123,7 @@ final class TcpReplicator extends AbstractChannelReplicator implements Closeable private byte state; private TcpSocketChannelEntryReader() { - in = ByteBuffer.allocateDirect(replicationConfig.packetSize() + maxEntrySizeBytes); + in = ByteBuffer.allocateDirect(replicationConfig.tcpBufferSize()); out = new ByteBufferBytes(in.slice()); out.limit(0); in.clear(); @@ -1222,7 +1232,7 @@ final class TcpReplicator extends AbstractChannelReplicator implements Closeable final Work futureWork = statelessServerConnector.processStatelessEvent(state, - attached.entryWriter.in, attached.entryReader.out); + attached.entryWriter.in(), attached.entryReader.out); // turn the OP_WRITE on key.interestOps(key.interestOps() | OP_WRITE); @@ -1232,7 +1242,8 @@ final class TcpReplicator extends AbstractChannelReplicator implements Closeable // the buffer is no longer full, and as such is treated as future work if (futureWork != null) { try { // we will complete what we can for now - boolean isComplete = futureWork.doWork(attached.entryWriter.in); + boolean isComplete = futureWork.doWork(attached.entryWriter + .in()); if (!isComplete) attached.entryWriter.uncompletedWork = futureWork; } catch (Exception e) { @@ -1264,7 +1275,7 @@ final class TcpReplicator extends AbstractChannelReplicator implements Closeable // the maxEntrySizeBytes used here may not be the maximum size of the entry in its serialized form // however, its only use as an indication that the buffer is becoming full and should be compacted // the buffer can be compacted at any time - if (in.position() == 0 || in.remaining() > maxEntrySizeBytes) + if (in.position() == 0 || in.remaining() > largestEntrySoFar) return; in.limit(in.position()); diff --git a/src/main/java/net/openhft/chronicle/map/UdpChannelReplicator.java b/src/main/java/net/openhft/chronicle/map/UdpChannelReplicator.java index 0ec606d45..fd3f2a247 100644 --- a/src/main/java/net/openhft/chronicle/map/UdpChannelReplicator.java +++ b/src/main/java/net/openhft/chronicle/map/UdpChannelReplicator.java @@ -73,8 +73,8 @@ class UdpChannelReplicator extends AbstractChannelReplicator implements Replica. final byte localIdentifier) throws IOException { - super("UdpReplicator-" + localIdentifier, replicationConfig.throttlingConfig(), - maxPayloadSize); + super("UdpReplicator-" + localIdentifier, replicationConfig.throttlingConfig() + ); this.localIdentifier = localIdentifier; diff --git a/src/main/java/net/openhft/chronicle/map/UdpReplicator.java b/src/main/java/net/openhft/chronicle/map/UdpReplicator.java index 970eeaa2d..a4cb64688 100644 --- a/src/main/java/net/openhft/chronicle/map/UdpReplicator.java +++ b/src/main/java/net/openhft/chronicle/map/UdpReplicator.java @@ -30,10 +30,10 @@ import java.nio.ByteBuffer; import java.nio.channels.DatagramChannel; /** - * The UdpReplicator attempts to read the data ( but it does not enforce or grantee delivery ), typically, 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. In order to not miss data, UdpReplicator should be used in conjunction with - * the TCP Replicator. + * The UdpReplicator attempts to read the data ( but it does not enforce or grantee delivery ), + * typically, 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. In order to not miss data, UdpReplicator + * should be used in conjunction with the TCP Replicator. */ final class UdpReplicator extends UdpChannelReplicator implements Replica.ModificationNotifier, Closeable { @@ -106,7 +106,7 @@ final class UdpReplicator extends UdpChannelReplicator implements Replica.Modifi // check the the first 4 bytes are the inverted len followed by the len // we do this to check that this is a valid start of entry, otherwise we throw it away - if ( ~size != invertedSize) + if (~size != invertedSize) return; if (out.remaining() != size) @@ -118,8 +118,7 @@ final class UdpReplicator extends UdpChannelReplicator implements Replica.Modifi private static class UdpSocketChannelEntryWriter implements EntryWriter { - private final ByteBuffer out; - private final ByteBufferBytes in; + private final EntryCallback entryCallback; private final UdpChannelReplicator udpReplicator; private Replica.ModificationIterator modificationIterator; @@ -131,9 +130,8 @@ final class UdpReplicator extends UdpChannelReplicator implements Replica.Modifi this.udpReplicator = udpReplicator; // we make the buffer twice as large just to give ourselves headroom - out = ByteBuffer.allocateDirect(serializedEntrySize * 2); - in = new ByteBufferBytes(out); - entryCallback = new EntryCallback(externalizable, in); + + entryCallback = new EntryCallback(externalizable, serializedEntrySize); this.modificationIterator = modificationIterator; } @@ -156,6 +154,9 @@ 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(); + out.clear(); in.clear(); in.skip(SIZE_OF_SIZE); diff --git a/src/test/java/net/openhft/chronicle/map/TcpReplicationSoakTest.java b/src/test/java/net/openhft/chronicle/map/TcpReplicationSoakTest.java index 187a743d1..b02814f16 100644 --- a/src/test/java/net/openhft/chronicle/map/TcpReplicationSoakTest.java +++ b/src/test/java/net/openhft/chronicle/map/TcpReplicationSoakTest.java @@ -1,7 +1,10 @@ package net.openhft.chronicle.map; +import junit.framework.Assert; import net.openhft.chronicle.hash.replication.TcpTransportAndNetworkConfig; -import org.junit.*; +import org.junit.After; +import org.junit.Before; +import org.junit.Test; import java.io.Closeable; import java.io.IOException; @@ -22,14 +25,15 @@ public class TcpReplicationSoakTest { private ChronicleMap map2; static int s_port = 8010; - int valueSize = 1000000; - - char[] value = new char[valueSize]; + int valueSize = 1_00_000; // 1MB + char[] valueX = new char[valueSize - 100]; + char[] valueY = new char[valueSize - 100]; @Before public void setup() throws IOException { - Arrays.fill(value, 'X'); + Arrays.fill(valueX, 'X'); + Arrays.fill(valueX, 'Y'); int keySize = 4; int entrySize = keySize + valueSize; @@ -44,7 +48,8 @@ public class TcpReplicationSoakTest { map1 = ChronicleMapBuilder.of(Integer.class, CharSequence.class) - .entries(entrySize) + .entrySize(entrySize) + .entries(2) .actualSegments(1) .replication((byte) 1, tcpConfig1) .instance() @@ -58,7 +63,8 @@ public class TcpReplicationSoakTest { .tcpBufferSize(1024 * 64); map2 = ChronicleMapBuilder.of(Integer.class, CharSequence.class) - .entries(entrySize) + .entrySize(entrySize) + .entries(2) .replication((byte) 2, tcpConfig2) .instance() .name("map2") @@ -95,17 +101,22 @@ public class TcpReplicationSoakTest { StatelessClientTest.checkThreadsShutdown(threads); } - @Ignore("HCOLL-246 Tcp Replication fails with large values or entries sizes that don't " + - "included internal replication bytes") + @Test public void testLargeValues() throws IOException, InterruptedException { - map1.put(1, new String(value)); + String xString = new String(valueX); + String yString = new String(valueY); + + map1.put(1, xString); + map1.put(2, yString); + System.out.println("\nwaiting till equal"); waitTillEqual(10000); - Assert.assertEquals(map1, map2); + Assert.assertEquals(xString, map2.get(1)); + Assert.assertEquals(yString, map2.get(2)); }