From e1df6dc68f320b1d74e07a6632f54670c9ec4275 Mon Sep 17 00:00:00 2001
From: leventov <leventov@ya.ru>
Date: Fri, 9 Oct 2015 15:11:08 +0300
Subject: [PATCH] Fix a bug in TcpReplicator -- didn't schedule to connect to
 an endpoint later, when channel.socket().setTcpNoDelay(true) throws an
 exception

---
 .../chronicle/map/ReplicatedChronicleMap.java | 10 ++---
 .../openhft/chronicle/map/TcpReplicator.java  | 43 +++++++++++--------
 2 files changed, 31 insertions(+), 22 deletions(-)

diff --git a/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java b/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java
index 1d42d1dd0..b147d00c6 100644
--- a/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java
+++ b/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java
@@ -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);
         }
 
diff --git a/src/main/java/net/openhft/chronicle/map/TcpReplicator.java b/src/main/java/net/openhft/chronicle/map/TcpReplicator.java
index b3ca852b4..da6726da8 100644
--- a/src/main/java/net/openhft/chronicle/map/TcpReplicator.java
+++ b/src/main/java/net/openhft/chronicle/map/TcpReplicator.java
@@ -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();
+            }
+        }
     }
 
     /**