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

Fix a bug in TcpReplicator -- didn't schedule to connect to an endpoint later, when channel.socket().setTcpNoDelay(true) throws an exception

Browse Source
This commit is contained in:
leventov
2015-10-09 15:11:08 +03:00
parent cad3d3abfe
commit e1df6dc68f
2 changed files with 31 additions and 22 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java
+5 -5
View File
@@ -550,19 +550,19 @@ public class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? sup
private int iterationMainSegmentsAreaOrTierBulk = -1;
private long tierBulkBitSetAddr = 0L;
public ModificationIterator(int identifier) {
public ModificationIterator(int remoteIdentifier) {
long bitsPerSegment = bitsPerSegmentInModIterBitSet();
segmentIndexShift = Long.numberOfTrailingZeros(bitsPerSegment);
posMask = bitsPerSegment - 1L;
mainSegmentsChangesBitSetAddr = ms.address() + startOfModificationIterators +
identifier * modIterBitSetSizeInBytes();
remoteIdentifier * modIterBitSetSizeInBytes();
nativeAccess().zeroOut(null, mainSegmentsChangesBitSetAddr, modIterBitSetSizeInBytes());
offsetToBitSetWithinATierBulk =
identifier * tierBulkModIterBitSetSizeInBytes(numberOfTiersInBulk);
remoteIdentifier * tierBulkModIterBitSetSizeInBytes(numberOfTiersInBulk);
}
public ModificationIterator(int identifier, ModificationNotifier notifier) {
this(identifier);
public ModificationIterator(int remoteIdentifier, ModificationNotifier notifier) {
this(remoteIdentifier);
setModificationNotifier(notifier);
}
src/main/java/net/openhft/chronicle/map/TcpReplicator.java
+26 -17
View File
@@ -452,7 +452,13 @@ public final class TcpReplicator<K, V> extends AbstractChannelReplicator impleme
if (!channel.finishConnect()) {
return;
}
} catch (SocketException e) {
channel.configureBlocking(false);
channel.socket().setTcpNoDelay(true);
channel.socket().setSoTimeout(0);
channel.socket().setSoLinger(false, 0);
} catch (IOException e) {
quietClose(key, e);
// when node discovery is used ( by nodes broadcasting out their host:port over UDP ),
@@ -474,11 +480,6 @@ public final class TcpReplicator<K, V> extends AbstractChannelReplicator impleme
LOG.debug("successfully connected to {}, local-id={}",
channel.socket().getInetAddress(), localIdentifier);
channel.configureBlocking(false);
channel.socket().setTcpNoDelay(true);
channel.socket().setSoTimeout(0);
channel.socket().setSoLinger(false, 0);
attached.entryReader = new TcpSocketChannelEntryReader();
attached.entryWriter = new TcpSocketChannelEntryWriter();
@@ -820,6 +821,8 @@ public final class TcpReplicator<K, V> extends AbstractChannelReplicator impleme
if (len == -1) {
socketChannel.register(selector, 0, attached);
if (replicationConfig.autoReconnectedUponDroppedConnection()) {
// TODO why != null check here, absent everywhere else? on connectLater() calls?
// TODO shouldn't check attached.isServer, same as in catch (IOE) clause below?
AbstractConnector connector = attached.connector;
if (connector != null)
connector.connectLater();
@@ -1018,6 +1021,7 @@ public final class TcpReplicator<K, V> extends AbstractChannelReplicator impleme
} catch (ClosedChannelException e) {
if (socketChannel.isOpen())
LOG.error("", e);
onFail(socketChannel);
}
}
});
@@ -1027,20 +1031,25 @@ public final class TcpReplicator<K, V> extends AbstractChannelReplicator impleme
return socketChannel;
} finally {
if (!success) {
try {
try {
socketChannel.socket().close();
} catch (Exception e) {
LOG.error("", e);
}
socketChannel.close();
} catch (IOException e) {
LOG.error("", e);
}
this.connectLater();
onFail(socketChannel);
}
}
}
private void onFail(SocketChannel socketChannel) {
try {
try {
socketChannel.socket().close();
} catch (Exception e) {
LOG.error("", e);
}
socketChannel.close();
} catch (IOException e) {
LOG.error("", e);
} finally {
this.connectLater();
}
}
}
/**