feat DEAD channel detection, now is everywhere (#612)

* refactring hearbeat mechanism

Signed-off-by: dorjesinpo <[email protected]>

* BrokerResponse w/ hearbeat config

Signed-off-by: dorjesinpo <[email protected]>

* Addressing review

Signed-off-by: dorjesinpo <[email protected]>

---------

Signed-off-by: dorjesinpo <[email protected]>

Author: dorjesinpo

docker/cluster/config/bmqbrkrcfg.json

@@ -73,7 +73,7 @@
         },
         "networkInterfaces": {
             "heartbeats": {
-                "client": 0,
+                "client": 10,
                 "downstreamBroker": 10,
                 "upstreamBroker": 10,
                 "clusterPeer": 10

docker/single-node/config/bmqbrkrcfg.json

@@ -73,7 +73,7 @@
         },
         "networkInterfaces": {
             "heartbeats": {
-                "client": 0,
+                "client": 10,
                 "downstreamBroker": 10,
                 "upstreamBroker": 10,
                 "clusterPeer": 10

src/applications/bmqbrkr/etc/bmqbrkrcfg.json

@@ -73,7 +73,7 @@
         },
         "networkInterfaces": {
             "heartbeats": {
-                "client": 0,
+                "client": 10,
                 "downstreamBroker": 10,
                 "upstreamBroker": 10,
                 "clusterPeer": 10

src/groups/bmq/bmqimp/bmqimp_application.cpp

@@ -74,6 +74,8 @@ namespace {
 const double             k_RECONNECT_INTERVAL_MS = 500;
 const int                k_RECONNECT_COUNT = bsl::numeric_limits<int>::max();
 const bsls::Types::Int64 k_CHANNEL_LOW_WATERMARK = 512 * 1024;
+const int                k_DEFAULT_MAX_MISSED_HEARTBEATS = 10;
+const int                k_DEFAULT_HEARTBEAT_INTERVAL_MS = 1000;
 
 /// Create the StatContextConfiguration to use, from the specified
 /// `options`, and using the specified `allocator` for memory allocations.
@@ -144,6 +146,8 @@ void Application::onChannelDown(const bsl::string&   peerUri,
 {
     // executed by the *IO* thread
 
+    stopHeartbeat();
+
     BALL_LOG_INFO << "Session with '" << peerUri << "' is now DOWN"
                   << " [status: " << status << "]";
 
@@ -159,10 +163,12 @@ void Application::onChannelWatermark(const bsl::string&                peerUri,
     d_brokerSession.handleChannelWatermark(type);
 }
 
-void Application::readCb(const bmqio::Status&                   status,
-                         int*                                   numNeeded,
-                         bdlbb::Blob*                           blob,
-                         const bsl::shared_ptr<bmqio::Channel>& channel)
+void Application::readCb(
+    const bmqio::Status&                           status,
+    int*                                           numNeeded,
+    bdlbb::Blob*                                   blob,
+    const bsl::shared_ptr<bmqio::Channel>&         channel,
+    const bsl::shared_ptr<bmqp::HeartbeatMonitor>& monitor)
 {
     // executed by the *IO* thread
 
@@ -201,10 +207,15 @@ void Application::readCb(const bmqio::Status&                   status,
         return;  // RETURN
     }
 
-    BALL_LOG_TRACE << channel->peerUri() << ": ReadCallback got a blob\n"
-                   << bmqu::BlobStartHexDumper(&readBlob);
+    // Create a raw event with a cloned blob
+    bmqp::Event event(&readBlob, &d_allocator, true);
+
+    if (monitor->checkData(channel.get(), event)) {
+        BALL_LOG_TRACE << channel->peerUri() << ": ReadCallback got a blob\n"
+                       << bmqu::BlobStartHexDumper(&readBlob);
 
-    d_brokerSession.processPacket(readBlob);
+        d_brokerSession.processPacket(event);
+    }
 }
 
 void Application::channelStateCallback(
@@ -238,6 +249,9 @@ void Application::channelStateCallback(
 
         d_brokerSession.setChannel(channel);
 
+        bsl::shared_ptr<bmqp::HeartbeatMonitor> monitor = createMonitor(
+            channel);
+
         // Initiate read flow
         bmqio::Status st;
         channel->read(
@@ -248,7 +262,8 @@ void Application::channelStateCallback(
                                  bdlf::PlaceHolders::_1,  // status
                                  bdlf::PlaceHolders::_2,  // numNeeded
                                  bdlf::PlaceHolders::_3,  // blob
-                                 channel));
+                                 channel,
+                                 monitor));
         if (!st) {
             BALL_LOG_ERROR << "Could not read from channel:"
                            << " [peer: " << channel->peerUri()
@@ -260,6 +275,8 @@ void Application::channelStateCallback(
         // Cancel the timeout event (if the handle is invalid, this will just
         // do nothing)
         d_scheduler.cancelEvent(&d_startTimeoutHandle);
+
+        startHeartbeat(channel, monitor);
     } break;  // BREAK
     case bmqio::ChannelFactoryEvent::e_CONNECT_ATTEMPT_FAILED: {
         BALL_LOG_DEBUG << "Failed an attempt to establish a session with '"
@@ -599,6 +616,7 @@ Application::Application(
 , d_statSnaphotTimerHandle()
 , d_nextStatDump(-1)
 , d_lastAllocatorSnapshot(0)
+, d_heartbeatSchedulerHandle()
 {
     // NOTE:
     //   o The persistent session pool must live longer than the brokerSession
@@ -721,6 +739,8 @@ void Application::stop()
     BALL_LOG_INFO << "::: STOP (SYNC) [state: " << d_brokerSession.state()
                   << "] :::";
 
+    stopHeartbeat();
+
     // Stop the brokerSession
     d_brokerSession.stop();
 }
@@ -730,9 +750,79 @@ void Application::stopAsync()
     BALL_LOG_INFO << "::: STOP (ASYNC) [state: " << d_brokerSession.state()
                   << "] :::";
 
+    stopHeartbeat();
+
     // Stop the brokerSession
     d_brokerSession.stopAsync();
 }
 
+bsl::shared_ptr<bmqp::HeartbeatMonitor>
+Application::createMonitor(const bsl::shared_ptr<bmqio::Channel>& channel)
+{
+    int maxMissedHeartbeats = k_DEFAULT_MAX_MISSED_HEARTBEATS;
+
+    channel->properties().load(
+        &maxMissedHeartbeats,
+        NegotiatedChannelFactory::k_CHANNEL_PROPERTY_MAX_MISSED_HEARTBEATS);
+
+    bsl::shared_ptr<bmqp::HeartbeatMonitor> monitor(
+        new (d_allocator) bmqp::HeartbeatMonitor(maxMissedHeartbeats),
+        &d_allocator);
+
+    return monitor;
+}
+
+void Application::startHeartbeat(
+    const bsl::shared_ptr<bmqio::Channel>&         channel,
+    const bsl::shared_ptr<bmqp::HeartbeatMonitor>& monitor)
+{
+    BSLS_ASSERT_SAFE(monitor);
+
+    if (!monitor->isHearbeatEnabled()) {
+        return;  // RETURN
+    }
+
+    int heartbeatIntervalMs = k_DEFAULT_HEARTBEAT_INTERVAL_MS;
+
+    channel->properties().load(
+        &heartbeatIntervalMs,
+        NegotiatedChannelFactory::k_CHANNEL_PROPERTY_HEARTBEAT_INTERVAL_MS);
+
+    bsls::TimeInterval interval;
+    interval.addMilliseconds(heartbeatIntervalMs);
+
+    d_scheduler.scheduleRecurringEvent(
+        &d_heartbeatSchedulerHandle,
+        interval,
+        bdlf::BindUtil::bind(&Application::onHeartbeatSchedulerEvent,
+                             this,
+                             channel,
+                             monitor));
+}
+void Application::stopHeartbeat()
+{
+    d_scheduler.cancelEventAndWait(&d_heartbeatSchedulerHandle);
+}
+
+void Application::onHeartbeatSchedulerEvent(
+    const bsl::shared_ptr<bmqio::Channel>&         channel,
+    const bsl::shared_ptr<bmqp::HeartbeatMonitor>& monitor)
+{
+    // executed by the *SCHEDULER* thread
+
+    BSLS_ASSERT_SAFE(monitor);
+    BSLS_ASSERT_SAFE(monitor->maxMissedHeartbeats());
+
+    if (!monitor->checkHeartbeat(channel.get())) {
+        BALL_LOG_WARN << "#TCP_DEAD_CHANNEL "
+                      << "Closing unresponsive channel after "
+                      << monitor->maxMissedHeartbeats()
+                      << " missed heartbeats [channel: '" << channel->peerUri()
+                      << "']";
+
+        channel->close();
+    }
+}
+
 }  // close package namespace
 }  // close enterprise namespace

src/groups/bmq/bmqimp/bmqimp_application.h

@@ -40,6 +40,7 @@
 #include <bmqimp_eventqueue.h>
 #include <bmqimp_negotiatedchannelfactory.h>
 #include <bmqp_ctrlmsg_messages.h>
+#include <bmqp_heartbeatmonitor.h>
 #include <bmqt_sessionoptions.h>
 
 #include <bmqio_channel.h>
@@ -159,6 +160,9 @@ class Application {
     // the snapshot was performed on the
     // Counting Allocators context
 
+    /// Scheduler handle of the recurring event to monitor channels heartbeats.
+    bdlmt::EventSchedulerRecurringEventHandle d_heartbeatSchedulerHandle;
+
   private:
     // PRIVATE MANIPULATORS
     void onChannelDown(const bsl::string&   peerUri,
@@ -167,10 +171,11 @@ class Application {
     void onChannelWatermark(const bsl::string&                peerUri,
                             bmqio::ChannelWatermarkType::Enum type);
 
-    void readCb(const bmqio::Status&                   status,
-                int*                                   numNeeded,
-                bdlbb::Blob*                           blob,
-                const bsl::shared_ptr<bmqio::Channel>& channel);
+    void readCb(const bmqio::Status&                           status,
+                int*                                           numNeeded,
+                bdlbb::Blob*                                   blob,
+                const bsl::shared_ptr<bmqio::Channel>&         channel,
+                const bsl::shared_ptr<bmqp::HeartbeatMonitor>& monitor);
 
     void channelStateCallback(const bsl::string&                     endpoint,
                               bmqio::ChannelFactoryEvent::Enum       event,
@@ -212,6 +217,19 @@ class Application {
             bmqimp::BrokerSession::State::Enum    newState,
             bmqimp::BrokerSession::FsmEvent::Enum event);
 
+    /// Recurring scheduler event to check for all `heartbeat-enabled`
+    /// channels : this will send a heartbeat if no data has been received
+    /// on a given channel, or proactively reset the channel if too many
+    /// heartbeats have been missed.
+    void onHeartbeatSchedulerEvent(
+        const bsl::shared_ptr<bmqio::Channel>&         channel,
+        const bsl::shared_ptr<bmqp::HeartbeatMonitor>& monitor);
+    bsl::shared_ptr<bmqp::HeartbeatMonitor>
+    createMonitor(const bsl::shared_ptr<bmqio::Channel>& channel);
+    void startHeartbeat(const bsl::shared_ptr<bmqio::Channel>&         channel,
+                        const bsl::shared_ptr<bmqp::HeartbeatMonitor>& monitor);
+    void stopHeartbeat();
+
   private:
     // NOT IMPLEMENTED
     Application(const Application& other) BSLS_CPP11_DELETED;

src/groups/bmq/bmqimp/bmqimp_brokersession.cpp

@@ -3430,18 +3430,6 @@ void BrokerSession::processControlEvent(const bmqp::Event& event)
     }
 }
 
-void BrokerSession::onHeartbeat()
-{
-    // executed by the *IO* thread
-    // Add to the FSM event queue
-    bsl::shared_ptr<Event> queueEvent = createEvent();
-    queueEvent->configureAsRequestEvent(
-        bdlf::BindUtil::bind(&BrokerSession::doHandleHeartbeat,
-                             this,
-                             bdlf::PlaceHolders::_1));  // eventImpl
-    enqueueFsmEvent(queueEvent);
-}
-
 void BrokerSession::enableMessageRetransmission(
     const bmqp::PutMessageIterator& putIter,
     const bsls::TimeInterval&       sentTime)
@@ -4332,28 +4320,6 @@ void BrokerSession::doHandleChannelWatermark(
     }
 }
 
-void BrokerSession::doHandleHeartbeat(
-    BSLS_ANNOTATION_UNUSED const bsl::shared_ptr<Event>& eventSp)
-{
-    // executed by the FSM thread
-
-    BSLS_ASSERT_SAFE(d_fsmThreadChecker.inSameThread());
-
-    // The broker sent a heartbeat to check on us, simply reply with a
-    // heartbeat response.
-    //
-    // NOTE: the client doesn't check on the broker, therefore it will never
-    //       send 'HEARTBEAT_REQ' and hence we don't have to handle
-    //       'HEARTBEAT_RSP' type.
-    if (d_channel_sp) {
-        d_channel_sp->write(0,  // status
-                            bmqp::ProtocolUtil::heartbeatRspBlob(),
-                            d_sessionOptions.channelHighWatermark());
-        // We explicitly ignore any failure as failure implies issues with the
-        // channel, which is what the heartbeat is trying to expose.
-    }
-}
-
 void BrokerSession::enqueueStateRestoredIfNeeded()
 {
     // executed by the FSM thread
@@ -5851,25 +5817,29 @@ BrokerSession::processPacket(const bdlbb::Blob& packet)
     // executed by the *IO* thread
     // or *APPLICATION* thread
 
-    enum { e_NUM_BYTES_IN_BLOB_TO_DUMP = 256 };
-
     // Create a raw event with a cloned blob
     bmqp::Event event(&packet, d_allocator_p, true);
+
+    return processPacket(event);
+}
+
+bmqt::GenericResult::Enum
+BrokerSession::processPacket(const bmqp::Event& event)
+{
+    // executed by the *IO* thread
+    // or *APPLICATION* thread
+
+    enum { e_NUM_BYTES_IN_BLOB_TO_DUMP = 256 };
+
     if (BSLS_PERFORMANCEHINT_PREDICT_UNLIKELY(!event.isValid())) {
         BSLS_PERFORMANCEHINT_UNLIKELY_HINT;
         BALL_LOG_ERROR << "Received an invalid packet: "
                        << bmqu::BlobStartHexDumper(
-                              &packet,
+                              event.blob(),
                               e_NUM_BYTES_IN_BLOB_TO_DUMP);
         return bmqt::GenericResult::e_INVALID_ARGUMENT;  // RETURN
     }
 
-    if (BSLS_PERFORMANCEHINT_PREDICT_UNLIKELY(event.isHeartbeatReqEvent())) {
-        BSLS_PERFORMANCEHINT_UNLIKELY_HINT;
-        onHeartbeat();
-        return bmqt::GenericResult::e_SUCCESS;  // RETURN
-    }
-
     // Add to event queue
     bsl::shared_ptr<Event> queueEvent = createEvent();
     queueEvent->configureAsRawEvent(event);

src/groups/bmq/bmqimp/bmqimp_brokersession.h

@@ -967,10 +967,6 @@ class BrokerSession BSLS_CPP11_FINAL {
     /// the broker) is available on the channel.
     void processControlEvent(const bmqp::Event& event);
 
-    /// This method gets called each time a new heart beat event (sent by
-    /// the broker) is available on the channel.
-    void onHeartbeat();
-
     void enableMessageRetransmission(const bmqp::PutMessageIterator& putIter,
                                      const bsls::TimeInterval&       sentTime);
 
@@ -1360,10 +1356,6 @@ class BrokerSession BSLS_CPP11_FINAL {
     void doHandleChannelWatermark(bmqio::ChannelWatermarkType::Enum type,
                                   const bsl::shared_ptr<Event>&     eventSp);
 
-    /// Invoked from the FSM thread as a handler to the heartbeat event
-    /// specified as `eventSp` sent by the IO thread.
-    void doHandleHeartbeat(const bsl::shared_ptr<Event>& eventSp);
-
     /// Invoked from the FSM thread to start channel closing.
     void disconnectChannel();
 
@@ -1525,6 +1517,7 @@ class BrokerSession BSLS_CPP11_FINAL {
     /// Return error status in case the packet cannot be handled by any
     /// reason.
     bmqt::GenericResult::Enum processPacket(const bdlbb::Blob& packet);
+    bmqt::GenericResult::Enum processPacket(const bmqp::Event& event);
 
     /// Set the specified `channel` to use for communication with the
     /// bmqbrkr.  If `channel` is non null, this is a newly-established