Alawn/Chronicle-Map
1
0
Fork 0
mirror of https://github.com/OpenHFT/Chronicle-Map.git synced 2021-09-13 01:26:12 +00:00
Code Releases Activity

HCOLL-329 Fix Chronicle Map Cluster replication.

Browse Source
This commit is contained in:
Rob Austin
2015-05-07 19:26:11 +01:00
parent d7f5411294
commit dea1abd833
1 changed files with 18 additions and 34 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/main/java/net/openhft/chronicle/map/TcpReplicator.java
+18 -34
View File
@@ -694,9 +694,9 @@ final class TcpReplicator<K, V> extends AbstractChannelReplicator implements Clo
attached.hasRemoteHeartbeatInterval = true;
// now we're finished we can get on with reading the entries
attached.handShakingComplete = true;
attached.remoteModificationIterator.dirtyEntries(attached.remoteBootstrapTimestamp);
reader.entriesFromBuffer(attached, key);
attached.handShakingComplete = true;
}
}
@@ -738,10 +738,18 @@ final class TcpReplicator<K, V> extends AbstractChannelReplicator implements Clo
if (!entryWriter.hasBytesToWrite()
&& !entryWriter.isWorkIncomplete()
&& hasNext(attached))
&& !hasNext(attached)
&& attached.isHandShakingComplete()) {
// 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
// The OP_WRITE will get added back in as soon as we have data to write
if (LOG.isDebugEnabled())
LOG.debug("Disabling OP_WRITE to remoteIdentifier=" +
attached.remoteIdentifier +
", localIdentifier=" + localIdentifier);
key.interestOps(key.interestOps() & ~OP_WRITE);
}
} catch (IOException e) {
quietClose(key, e);
if (!attached.isServer)
@@ -752,7 +760,7 @@ final class TcpReplicator<K, V> extends AbstractChannelReplicator implements Clo
private boolean hasNext(Attached attached) {
return attached.remoteModificationIterator != null
&& !attached.remoteModificationIterator.hasNext();
&& attached.remoteModificationIterator.hasNext();
}
/**
@@ -891,6 +899,9 @@ final class TcpReplicator<K, V> extends AbstractChannelReplicator implements Clo
final ServerSocketChannel serverChannel = openServerSocketChannel();
if (serverChannel == null)
return null;
serverChannel.socket().setReceiveBufferSize(BUFFER_SIZE);
serverChannel.configureBlocking(false);
serverChannel.register(TcpReplicator.this.selector, 0);
@@ -903,8 +914,10 @@ final class TcpReplicator<K, V> extends AbstractChannelReplicator implements Clo
closeables.add(serverSocket);
}
serverSocket.setReuseAddress(true);
if (serverSocket == null)
return null;
serverSocket.setReuseAddress(true);
serverSocket.bind(details.address());
// these can be run on this thread
@@ -1014,6 +1027,7 @@ final class TcpReplicator<K, V> extends AbstractChannelReplicator implements Clo
@Nullable
public Replica.ModificationIterator remoteModificationIterator;
// used for connect later
public AbstractConnector connector;
public long remoteBootstrapTimestamp = Long.MIN_VALUE;
public byte remoteIdentifier = Byte.MIN_VALUE;
@@ -1138,12 +1152,6 @@ final class TcpReplicator<K, V> extends AbstractChannelReplicator implements Clo
void entriesToBuffer(@NotNull final Replica.ModificationIterator modificationIterator,
@NotNull final SelectionKey selectionKey) throws InterruptedException {
final SocketChannel socketChannel = (SocketChannel) selectionKey.channel();
final Attached attached = (Attached) selectionKey.attachment();
// this can occur when new map are added to a channel
final boolean handShakingComplete = attached.isHandShakingComplete();
int entriesWritten = 0;
try {
for (; ; entriesWritten++) {
@@ -1237,30 +1245,6 @@ final class TcpReplicator<K, V> extends AbstractChannelReplicator implements Clo
in().writeLong(localHeartbeatInterval);
}
/**
* removes back in the OP_WRITE from the selector, otherwise it'll spin loop. The OP_WRITE
* will get added back in as soon as we have data to write
*
* @param socketChannel the socketChannel we wish to stop writing to
* @param attached data associated with the socketChannels key
*/
public synchronized void disableWrite(@NotNull final SocketChannel socketChannel,
@NotNull final Attached attached) {
try {
SelectionKey key = socketChannel.keyFor(selector);
if (key != null) {
if (attached.isHandShakingComplete() && selector.isOpen()) {
if (LOG.isDebugEnabled())
LOG.debug("Disabling OP_WRITE to remoteIdentifier=" +
attached.remoteIdentifier +
", localIdentifier=" + localIdentifier);
key.interestOps(key.interestOps() & ~OP_WRITE);
}
}
} catch (Exception e) {
LOG.error("", e);
}
}
public boolean doWork() {
return uncompletedWork != null && uncompletedWork.doWork(in());