Switch heartbeat implementation back to latches (reset events)

We're back to square one with approaches to heartbeat implementations.
The socket-based solution has two issues:

 * Obscure TLS timeouts.
 * WinRT has incompatible I/O API which makes implementing the same
   socket timeout-based approach becomes a lot more challenging
   (a couple of days spent in an unsuccessful spike).

This reset event-driven approach had bugs in 3.5.0 but is quite
straightforward and I/O layer agnostic, so we're willing to go
back.

Note that we now have a significantly better test coverage in this
area compared to the 3.5.0 development cycle.

Author: michaelklishin

projects/client/RabbitMQ.Client/src/client/impl/Connection.cs

@@ -77,19 +77,28 @@ public class Connection : IConnection
         public EventHandler<EventArgs> m_connectionUnblocked;
         public IConnectionFactory m_factory;
         public IFrameHandler m_frameHandler;
-        public ushort m_heartbeat = 0;
-        public TimeSpan m_heartbeatTimeSpan = TimeSpan.FromSeconds(0);
 
         public Guid m_id = Guid.NewGuid();
-
-        public int m_missedHeartbeats = 0;
         public ModelBase m_model0;
         public volatile bool m_running = true;
         public MainSession m_session0;
         public SessionManager m_sessionManager;
 
         public IList<ShutdownReportEntry> m_shutdownReport = new SynchronizedList<ShutdownReportEntry>(new List<ShutdownReportEntry>());
+
+        //
+        // Heartbeats
+        //
+
+        public ushort m_heartbeat = 0;
+        public TimeSpan m_heartbeatTimeSpan = TimeSpan.FromSeconds(0);
+        public int m_missedHeartbeats = 0;
+
         private Timer _heartbeatWriteTimer;
+        private Timer _heartbeatReadTimer;
+        public AutoResetEvent m_heartbeatRead = new AutoResetEvent(false);
+        public AutoResetEvent m_heartbeatWrite = new AutoResetEvent(false);
+
 
         // true if we haven't finished connection negotiation.
         // In this state socket exceptions are treated as fatal connection
@@ -113,7 +122,7 @@ public Connection(IConnectionFactory factory, bool insist, IFrameHandler frameHa
             StartMainLoop(factory.UseBackgroundThreadsForIO);
             Open(insist);
 
-            StartHeartbeatTimer();
+            StartHeartbeatTimers();
             AppDomain.CurrentDomain.DomainUnload += HandleDomainUnload;
         }
 
@@ -465,8 +474,10 @@ public void EnsureIsOpen()
         public void FinishClose()
         {
             // Notify hearbeat loops that they can leave
+            m_heartbeatRead.Set();
+            m_heartbeatWrite.Set();
             m_closed = true;
-            StopHeartbeatTimer();
+            StopHeartbeatTimers();
 
             m_frameHandler.Close();
             m_model0.SetCloseReason(m_closeReason);
@@ -522,33 +533,6 @@ public bool HardProtocolExceptionHandler(HardProtocolException hpe)
             return false;
         }
 
-        public void HeartbeatWriteTimerCallback(object state)
-        {
-            bool shouldTerminate = false;
-            try
-            {
-                if (!m_closed)
-                {
-                    WriteFrame(m_heartbeatFrame);
-                }
-            }
-            catch (Exception e)
-            {
-                HandleMainLoopException(new ShutdownEventArgs(
-                    ShutdownInitiator.Library,
-                    0,
-                    "End of stream",
-                    e));
-                shouldTerminate = true;
-            }
-
-            if (m_closed || shouldTerminate)
-            {
-                TerminateMainloop();
-                FinishClose();
-            }
-        }
-
         public void InternalClose(ShutdownEventArgs reason)
         {
             if (!SetCloseReason(reason))
@@ -639,96 +623,62 @@ public void MainLoop()
 
         public void MainLoopIteration()
         {
-            try
-            {
-                Frame frame = m_frameHandler.ReadFrame();
+            Frame frame = m_frameHandler.ReadFrame();
 
-                // We have received an actual frame.
-                m_missedHeartbeats = 0;
-                if (frame.Type == Constants.FrameHeartbeat)
-                {
-                    // Ignore it: we've already just reset the heartbeat
-                    // counter.
-                    return;
-                }
+            NotifyHeartbeatListener();
+            // We have received an actual frame.
+            if (frame.Type == Constants.FrameHeartbeat)
+            {
+                // Ignore it: we've already just reset the heartbeat
+                // latch.
+                return;
+            }
 
-                if (frame.Channel == 0)
-                {
-                    // In theory, we could get non-connection.close-ok
-                    // frames here while we're quiescing (m_closeReason !=
-                    // null). In practice, there's a limited number of
-                    // things the server can ask of us on channel 0 -
-                    // essentially, just connection.close. That, combined
-                    // with the restrictions on pipelining, mean that
-                    // we're OK here to handle channel 0 traffic in a
-                    // quiescing situation, even though technically we
-                    // should be ignoring everything except
-                    // connection.close-ok.
-                    m_session0.HandleFrame(frame);
-                }
-                else
+            if (frame.Channel == 0)
+            {
+                // In theory, we could get non-connection.close-ok
+                // frames here while we're quiescing (m_closeReason !=
+                // null). In practice, there's a limited number of
+                // things the server can ask of us on channel 0 -
+                // essentially, just connection.close. That, combined
+                // with the restrictions on pipelining, mean that
+                // we're OK here to handle channel 0 traffic in a
+                // quiescing situation, even though technically we
+                // should be ignoring everything except
+                // connection.close-ok.
+                m_session0.HandleFrame(frame);
+            }
+            else
+            {
+                // If we're still m_running, but have a m_closeReason,
+                // then we must be quiescing, which means any inbound
+                // frames for non-zero channels (and any inbound
+                // commands on channel zero that aren't
+                // Connection.CloseOk) must be discarded.
+                if (m_closeReason == null)
                 {
-                    // If we're still m_running, but have a m_closeReason,
-                    // then we must be quiescing, which means any inbound
-                    // frames for non-zero channels (and any inbound
-                    // commands on channel zero that aren't
-                    // Connection.CloseOk) must be discarded.
-                    if (m_closeReason == null)
+                    // No close reason, not quiescing the
+                    // connection. Handle the frame. (Of course, the
+                    // Session itself may be quiescing this particular
+                    // channel, but that's none of our concern.)
+                    ISession session = m_sessionManager.Lookup(frame.Channel);
+                    if (session == null)
                     {
-                        // No close reason, not quiescing the
-                        // connection. Handle the frame. (Of course, the
-                        // Session itself may be quiescing this particular
-                        // channel, but that's none of our concern.)
-                        ISession session = m_sessionManager.Lookup(frame.Channel);
-                        if (session == null)
-                        {
-                            throw new ChannelErrorException(frame.Channel);
-                        }
-                        else
-                        {
-                            session.HandleFrame(frame);
-                        }
+                        throw new ChannelErrorException(frame.Channel);
+                    }
+                    else
+                    {
+                        session.HandleFrame(frame);
                     }
                 }
             }
-            catch (SocketException ioe)
-            {
-                HandleIOException(ioe);
-            }
-            catch (IOException ioe)
-            {
-                HandleIOException(ioe);
-            }
         }
 
-        // socket receive timeout is configured to be 1/2 of the heartbeat timeout
-        // and the peer must be considered dead after two subsequent missed heartbeats:
-        // terminate after 4 socket timeouts
-        private const int SOCKET_TIMEOUTS_TO_CONSIDER_PEER_UNRESPONSIVE = 4;
-
-        protected void HandleIOException(Exception e)
+        public void NotifyHeartbeatListener()
         {
-            // socket error when in negotiation, throw BrokerUnreachableException
-            // immediately
-            if (m_inConnectionNegotiation)
-            {
-                var cfe = new ConnectFailureException("I/O error before connection negotiation was completed", e);
-                throw new BrokerUnreachableException(cfe);
-            }
-
-            if (++m_missedHeartbeats >= SOCKET_TIMEOUTS_TO_CONSIDER_PEER_UNRESPONSIVE)
+            if (m_heartbeat != 0)
             {
-                var description =
-                    String.Format("Peer missed 2 heartbeats with heartbeat timeout set to {0} seconds",
-                                  m_heartbeat);
-                var eose = new EndOfStreamException(description);
-                m_shutdownReport.Add(new ShutdownReportEntry(description, eose));
-                HandleMainLoopException(new ShutdownEventArgs(ShutdownInitiator.Library,
-                                                              0,
-                                                              "End of stream",
-                                                              eose));
-                TerminateMainloop();
-                FinishClose();
+                m_heartbeatRead.Set();
             }
         }
 
@@ -946,12 +896,15 @@ public bool SetCloseReason(ShutdownEventArgs reason)
             }
         }
 
-        public void StartHeartbeatTimer()
+        public void StartHeartbeatTimers()
         {
             if (Heartbeat != 0)
             {
                 _heartbeatWriteTimer = new Timer(HeartbeatWriteTimerCallback);
-                _heartbeatWriteTimer.Change(TimeSpan.FromMilliseconds(0), m_heartbeatTimeSpan);
+                _heartbeatWriteTimer.Change(TimeSpan.FromMilliseconds(0), TimeSpan.FromMilliseconds(-1));
+
+                _heartbeatReadTimer = new Timer(HeartbeatReadTimerCallback);
+                _heartbeatReadTimer.Change(TimeSpan.FromMilliseconds(0), TimeSpan.FromMilliseconds(-1));
             }
         }
 
@@ -963,9 +916,84 @@ public void StartMainLoop(bool useBackgroundThread)
             mainLoopThread.Start();
         }
 
-        protected void StopHeartbeatTimer()
+        public void HeartbeatReadTimerCallback(object state)
+        {
+            bool shouldTerminate = false;
+            if (!m_closed)
+            {
+                if (!m_heartbeatRead.WaitOne(0, false))
+                {
+                    m_missedHeartbeats++;
+                }
+                else
+                {
+                    m_missedHeartbeats = 0;
+                }
+
+                // Has to miss two full heartbeats to force socket close
+                if (m_missedHeartbeats > 1)
+                {
+                    String description = String.Format("Heartbeat missing with heartbeat == {0} seconds", m_heartbeat);
+                    var eose = new EndOfStreamException(description);
+                    m_shutdownReport.Add(new ShutdownReportEntry(description, eose));
+                    HandleMainLoopException(
+                        new ShutdownEventArgs(ShutdownInitiator.Library, 0, "End of stream", eose));
+                    shouldTerminate = true;
+                }
+             }
+
+            if (shouldTerminate)
+            {
+                TerminateMainloop();
+                FinishClose();
+            }
+            else
+            {
+                _heartbeatReadTimer.Change(Heartbeat * 1000, Timeout.Infinite);
+            }
+        }
+
+        public void HeartbeatWriteTimerCallback(object state)
         {
-            if (_heartbeatWriteTimer != null)
+            bool shouldTerminate = false;
+            try
+            {
+                if (!m_closed)
+                {
+                    if (!m_heartbeatWrite.WaitOne(0, false))
+                    {
+                        WriteFrame(m_heartbeatFrame);
+                    }
+                }
+            }
+            catch (Exception e)
+            {
+                HandleMainLoopException(new ShutdownEventArgs(
+                    ShutdownInitiator.Library,
+                    0,
+                    "End of stream",
+                    e));
+                shouldTerminate = true;
+            }
+
+            if (m_closed || shouldTerminate)
+            {
+                TerminateMainloop();
+                FinishClose();
+            }
+            else
+            {
+                _heartbeatWriteTimer.Change(Heartbeat * 1000, Timeout.Infinite);
+            }
+        }
+
+        protected void StopHeartbeatTimers()
+        {
+            if(_heartbeatReadTimer != null)
+            {
+                _heartbeatReadTimer.Dispose();
+            }
+            if(_heartbeatWriteTimer != null)
             {
                 _heartbeatWriteTimer.Dispose();
             }
@@ -976,7 +1004,7 @@ protected void StopHeartbeatTimer()
         ///</remarks>
         public void TerminateMainloop()
         {
-            StopHeartbeatTimer();
+            StopHeartbeatTimers();
             m_running = false;
         }
 
@@ -988,6 +1016,7 @@ public override string ToString()
         public void WriteFrame(Frame f)
         {
             m_frameHandler.WriteFrame(f);
+            m_heartbeatWrite.Set();
         }
 
         ///<summary>API-side invocation of connection abort.</summary>
@@ -1060,7 +1089,7 @@ public void HandleConnectionUnblocked()
 
         void IDisposable.Dispose()
         {
-            StopHeartbeatTimer();
+            StopHeartbeatTimers();
             Abort();
             if (ShutdownReport.Count > 0)
             {

projects/client/RabbitMQ.Client/src/client/impl/SocketFrameHandler.cs

@@ -94,8 +94,9 @@ public SocketFrameHandler(AmqpTcpEndpoint endpoint,
             }
 
             Stream netstream = m_socket.GetStream();
-            netstream.ReadTimeout = timeout;
-            netstream.WriteTimeout = timeout;
+            // make sure the socket timeout is greater than heartbeat timeout
+            netstream.ReadTimeout = timeout * 4;
+            netstream.WriteTimeout = timeout * 4;
 
             if (endpoint.Ssl.Enabled)
             {
@@ -143,7 +144,8 @@ public int Timeout
                 {
                     if (m_socket.Connected)
                     {
-                        m_socket.ReceiveTimeout = value;
+                        // make sure the socket timeout is greater than heartbeat interval
+                        m_socket.ReceiveTimeout = value * 4;
                     }
                 }
 #pragma warning disable 0168
@@ -163,7 +165,13 @@ public void Close()
                 {
                     try
                     {
-                        m_socket.LingerState = new LingerOption(true, SOCKET_CLOSING_TIMEOUT);
+                        try
+                        {
+                            
+                        } catch (ArgumentException _)
+                        {
+                            // ignore, we are closing anyway
+                        };
                         m_socket.Close();
                     }
                     catch (Exception _)