diff --git a/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java b/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java index 0aeed758c..60df821a5 100644 --- a/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java +++ b/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java @@ -23,7 +23,6 @@ import net.openhft.chronicle.hash.hashing.LongHashFunction; import net.openhft.chronicle.hash.impl.ContextFactory; import net.openhft.chronicle.hash.impl.HashContext; import net.openhft.chronicle.hash.replication.AbstractReplication; -import net.openhft.chronicle.hash.replication.LateUpdateException; import net.openhft.chronicle.hash.replication.TimeProvider; import net.openhft.chronicle.hash.serialization.BytesReader; import net.openhft.chronicle.hash.serialization.internal.BytesBytesInterop; @@ -1009,7 +1008,7 @@ final class ReplicatedChronicleMap) mapContext(); // records the current position of the cursor in the bitset - private long position = -1L; + private volatile long position = -1L; /** * @param bytes the back the bitset, used to mark which entries have changed diff --git a/src/main/java/net/openhft/chronicle/map/TcpReplicator.java b/src/main/java/net/openhft/chronicle/map/TcpReplicator.java index b8e81bd93..4c4eb5ff6 100644 --- a/src/main/java/net/openhft/chronicle/map/TcpReplicator.java +++ b/src/main/java/net/openhft/chronicle/map/TcpReplicator.java @@ -724,8 +724,7 @@ final class TcpReplicator extends AbstractChannelReplicator implements Clo entryWriter.entriesToBuffer(attached.remoteModificationIterator, key); try { - final int len = entryWriter.writeBufferToSocket(socketChannel, - approxTime); + final int len = entryWriter.writeBufferToSocket(socketChannel, approxTime); if (len == -1) socketChannel.close(); @@ -733,8 +732,11 @@ final class TcpReplicator extends AbstractChannelReplicator implements Clo if (len > 0) contemplateThrottleWrites(len); - if (!entryWriter.hasBytesToWrite() && !entryWriter.isWorkIncomplete()) - // TURN OP_WRITE_OFF + if (!entryWriter.hasBytesToWrite() + && !entryWriter.isWorkIncomplete() + && hasNext(attached)) + // 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 key.interestOps(key.interestOps() & ~OP_WRITE); } catch (IOException e) { quietClose(key, e); @@ -744,6 +746,11 @@ final class TcpReplicator extends AbstractChannelReplicator implements Clo } } + private boolean hasNext(Attached attached) { + return attached.remoteModificationIterator != null + && !attached.remoteModificationIterator.hasNext(); + } + /** * called when the selector receives a OP_READ message */ @@ -1150,15 +1157,6 @@ final class TcpReplicator extends AbstractChannelReplicator implements Clo if (entrySize > largestEntrySoFar) largestEntrySoFar = entrySize; - - if (!modificationIterator.hasNext()) { - // 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) - disableWrite(socketChannel, attached); - - } - // we've filled up the buffer lets give another channel a chance to send // some data if (in().remaining() <= largestEntrySoFar || in().position() >