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));
}