Fix unbounded memory growth with slow local subscribers

Migration.md

@@ -14,6 +14,19 @@ release will remove the deprecated code.
     * `waitForShutdown()` moved from `Node.hh` / `Node.cc` to `WaitHelpers.hh` / `WaitHelpers.cc`. `Node.hh` re-exports via `#include "gz/transport/WaitHelpers.hh"`, so no code changes are required for existing users of `waitForShutdown()`.
     * `waitUntil()`, `waitForService()`, `waitForTopic()` moved from `Helpers.hh` / `Helpers.cc` to `WaitHelpers.hh` / `WaitHelpers.cc`. Code using these functions must now `#include "gz/transport/WaitHelpers.hh"` directly.
 
+### Modifications
+
+1. The queue used to deliver messages to local (intraprocess) subscribers
+   is now bounded per topic. When a topic has `GZ_TRANSPORT_LOCAL_HWM`
+   messages queued (1000 by default), publishing a new message drops the
+   oldest queued message of that topic instead of letting the queue grow
+   without limit. Previously the queue was unbounded, so a subscriber
+   processing messages slower than the publication rate would make the
+   process run out of memory. Set the `GZ_TRANSPORT_LOCAL_HWM` environment
+   variable to tune the capacity, or to 0 to restore the previous unbounded
+   behavior. The new `localHwm()` function returns the current capacity.
+   * [GitHub issue 926](https://github.com/gazebosim/gz-transport/issues/926)
+
 ### Deprecations
 
 1. The `gzTransportPublish` function in `CIface.h` has been deprecated because

include/gz/transport/Node.hh

@@ -73,6 +73,19 @@ namespace gz::transport
   /// If your buffer reaches the maximum capacity data will be dropped.
   int GZ_TRANSPORT_VISIBLE sndHwm();
 
+  /// \brief Get the capacity (High Water Mark) of the queue that stores
+  /// messages to be delivered to local (intraprocess) subscribers. Note
+  /// that this limit is applied per topic.
+  /// \return The capacity of the local publication queue (units are
+  /// messages). A value of 0 indicates an unlimited queue, which will grow
+  /// until you run out of memory if the local subscribers cannot keep up
+  /// with the publication rate. The default capacity is contained in the
+  /// #kDefaultLocalHwm variable and can be changed with the
+  /// GZ_TRANSPORT_LOCAL_HWM environment variable.
+  /// When a topic reaches this capacity, its oldest queued message is
+  /// dropped to make room for a new one.
+  int GZ_TRANSPORT_VISIBLE localHwm();
+
   /// \class Node Node.hh gz/transport/Node.hh
   /// \brief A class that allows a client to communicate with other peers.
   /// There are two main communication modes: pub/sub messages and service

include/gz/transport/NodeShared.hh

@@ -277,6 +277,16 @@ namespace gz::transport
     /// If your buffer reaches the maximum capacity data will be dropped.
     public: int SndHwm();
 
+    /// \brief Get the capacity (High Water Mark) of the queue that stores
+    /// messages to be delivered to local (intraprocess) subscribers. Note
+    /// that this limit is applied per topic.
+    /// \return The capacity of the local publication queue (units are
+    /// messages). A value of 0 indicates an unlimited queue. The default
+    /// capacity is contained in the #kDefaultLocalHwm variable.
+    /// When a topic reaches this capacity, its oldest queued message is
+    /// dropped to make room for a new one.
+    public: int LocalHwm() const;
+
     /// \brief Turn topic statistics on or off.
     /// \param[in] _topic The name of the topic on which to enable or disable
     /// statistics.

include/gz/transport/TransportTypes.hh

@@ -190,6 +190,12 @@ namespace gz::transport
   /// \brief The high water mark of the send message buffer.
   /// \sa NodeShared::SndHwm
   const int kDefaultSndHwm = 1000;
+
+  /// \brief The high water mark of the queue that stores messages to be
+  /// delivered to local (intraprocess) subscribers. This limit is applied
+  /// per topic.
+  /// \sa NodeShared::LocalHwm
+  const int kDefaultLocalHwm = 1000;
   }
 }
 #endif

src/Node.cc

@@ -57,6 +57,12 @@ int sndHwm()
   return NodeShared::Instance()->SndHwm();
 }
 
+//////////////////////////////////////////////////
+int localHwm()
+{
+  return NodeShared::Instance()->LocalHwm();
+}
+
 //////////////////////////////////////////////////
 /// \internal
 /// \brief Private data for Node::Publisher class.
@@ -459,6 +465,8 @@ bool Node::Publisher::Publish(const ProtoMsg &_msg)
 
     pubMsgDetails->publisherNodeUUID = this->dataPtr->publisher.NUuid();
 
+    pubMsgDetails->fullyQualifiedTopic = this->dataPtr->publisher.Topic();
+
     if (subscribers.haveLocal)
     {
       for (const auto &node : subscribers.localHandlers)
@@ -518,14 +526,7 @@ bool Node::Publisher::Publish(const ProtoMsg &_msg)
 
     // Add the publish message details to the publish queue. The message
     // will be published asynchronously to the local and raw callbacks.
-    {
-      std::unique_lock<std::mutex> queueLock(
-          this->dataPtr->shared->dataPtr->pubThreadMutex);
-      this->dataPtr->shared->dataPtr->pubQueue.push_back(
-          std::move(pubMsgDetails));
-    }
-
-    this->dataPtr->shared->dataPtr->signalNewPub.notify_one();
+    this->dataPtr->shared->dataPtr->EnqueuePubMsg(std::move(pubMsgDetails));
   }
 
   // Handle remote subscribers.

src/NodeShared.cc

@@ -223,6 +223,11 @@ NodeShared::NodeShared()
     this->dataPtr->topicStatsEnabled = (gzStats == "1");
   }
 
+  // Set the capacity of the queue used to deliver messages to local
+  // (intraprocess) subscribers.
+  this->dataPtr->localHwm = this->dataPtr->NonNegativeEnvVar(
+    "GZ_TRANSPORT_LOCAL_HWM", kDefaultLocalHwm);
+
   // My process UUID.
   Uuid uuid;
   this->pUuid = uuid.ToString();
@@ -1348,6 +1353,12 @@ int NodeShared::SndHwm()
   return sndHwm;
 }
 
+/////////////////////////////////////////////////
+int NodeShared::LocalHwm() const
+{
+  return static_cast<int>(this->dataPtr->localHwm);
+}
+
 //////////////////////////////////////////////////
 bool NodeShared::HandlerWrapper::HasSubscriber(
     const std::string &_fullyQualifiedTopic,
@@ -1649,6 +1660,52 @@ void NodeSharedPrivate::AccessControlHandler()
   delete sock;
 }
 
+/////////////////////////////////////////////////
+void NodeSharedPrivate::EnqueuePubMsg(
+    std::unique_ptr<PublishMsgDetails> _msgDetails)
+{
+  std::string droppedTopic;
+  {
+    std::unique_lock<std::mutex> queueLock(this->pubThreadMutex);
+
+    auto &topicIters = this->pubQueueIters[_msgDetails->fullyQualifiedTopic];
+
+    // The queue of this topic is full. Drop its oldest message so that the
+    // queue does not grow in an unbounded way when the local subscription
+    // callbacks are slower than the publication rate.
+    if (this->localHwm > 0 && topicIters.size() >= this->localHwm)
+    {
+      if (this->pubQueueDropWarned.insert(
+            _msgDetails->fullyQualifiedTopic).second)
+      {
+        droppedTopic = _msgDetails->fullyQualifiedTopic;
+      }
+
+      this->pubQueue.erase(topicIters.front());
+      topicIters.pop_front();
+    }
+
+    topicIters.push_back(
+      this->pubQueue.insert(this->pubQueue.end(), std::move(_msgDetails)));
+  }
+
+  this->signalNewPub.notify_one();
+
+  // Warn outside the lock so that slow I/O does not block the publishers
+  // or the delivery thread.
+  if (!droppedTopic.empty())
+  {
+    std::cerr << "The local subscribers of [" << droppedTopic
+              << "] are processing messages slower than the publication "
+              << "rate. Dropping the oldest queued message. This warning "
+              << "will not be repeated for this topic until its queued "
+              << "messages fully drain. Use the GZ_TRANSPORT_LOCAL_HWM "
+              << "environment variable to tune the queue capacity (current "
+              << "capacity: " << this->localHwm << " messages per topic)."
+              << std::endl;
+  }
+}
+
 /////////////////////////////////////////////////
 void NodeSharedPrivate::PublishThread()
 {
@@ -1679,6 +1736,23 @@ void NodeSharedPrivate::PublishThread()
       // Get the message
       msgDetails = std::move(this->pubQueue.front());
       this->pubQueue.pop_front();
+
+      // Keep the per topic bookkeeping in sync with the pubQueue. The
+      // front of the deque of this topic references the entry just popped.
+      auto topicItersIt =
+        this->pubQueueIters.find(msgDetails->fullyQualifiedTopic);
+      if (topicItersIt != this->pubQueueIters.end())
+      {
+        topicItersIt->second.pop_front();
+        if (topicItersIt->second.empty())
+        {
+          this->pubQueueIters.erase(topicItersIt);
+
+          // The backlog of this topic fully drained. Allow the drop
+          // warning to be shown again on the next overload episode.
+          this->pubQueueDropWarned.erase(msgDetails->fullyQualifiedTopic);
+        }
+      }
     }
 
     // Send the message to all the local handlers.

src/NodeSharedPrivate.hh

@@ -21,6 +21,7 @@
 #include <algorithm>
 #include <atomic>
 #include <cstdlib>
+#include <deque>
 #include <filesystem>
 #include <list>
 #include <map>
@@ -342,8 +343,16 @@ namespace gz::transport
 
               /// \brief Publisher's node UUID.
               public: std::string publisherNodeUUID;
+
+              /// \brief Fully qualified topic name. Used to enforce the
+              /// per topic capacity of the pubQueue.
+              public: std::string fullyQualifiedTopic;
             };
 
+    /// \brief Type of the queue that stores messages pending delivery to
+    /// local subscribers.
+    public: using PubQueue = std::list<std::unique_ptr<PublishMsgDetails>>;
+
     /// \brief Publish thread used to process the pubQueue.
     public: std::thread pubThread;
 
@@ -352,7 +361,36 @@ namespace gz::transport
 
     /// \brief List onto which new messages are pushed. The pubThread
     /// will pop off the messages and send them to local subscribers.
-    public: std::list<std::unique_ptr<PublishMsgDetails>> pubQueue;
+    /// Use EnqueuePubMsg to push messages so that the localHwm capacity
+    /// is enforced.
+    public: PubQueue pubQueue;
+
+    /// \brief Iterators to the pubQueue entries of each topic, in queue
+    /// order. Used to track the number of queued messages per topic and to
+    /// drop the oldest message of a topic in constant time when the
+    /// localHwm capacity is reached. Protected by pubThreadMutex.
+    public: std::unordered_map<std::string, std::deque<PubQueue::iterator>>
+              pubQueueIters;
+
+    /// \brief Topics that have already dropped messages from the pubQueue.
+    /// Used to warn only once per overload episode: a topic is removed from
+    /// this set when its backlog fully drains from the pubQueue. Protected
+    /// by pubThreadMutex.
+    public: std::unordered_set<std::string> pubQueueDropWarned;
+
+    /// \brief Maximum number of messages that can be stored in the pubQueue
+    /// per topic. A value of 0 means no limit. Initialized from the
+    /// GZ_TRANSPORT_LOCAL_HWM environment variable before the pubThread
+    /// starts and never modified afterwards.
+    public: std::size_t localHwm = kDefaultLocalHwm;
+
+    /// \brief Add a message to the pubQueue so that it is asynchronously
+    /// delivered to local subscribers. If the topic already has localHwm
+    /// messages stored in the pubQueue, its oldest queued message is
+    /// dropped to make room for the new one.
+    /// \param[in] _msgDetails Details of the message to be published.
+    public: void EnqueuePubMsg(
+              std::unique_ptr<PublishMsgDetails> _msgDetails);
 
     /// \brief used to signal when new work is available
     public: std::condition_variable signalNewPub;

src/Node_TEST.cc

@@ -2564,6 +2564,14 @@ TEST(NodeTest, SndHwm)
     EXPECT_EQ(-1, transport::sndHwm());
 }
 
+//////////////////////////////////////////////////
+/// \brief Check the high water mark of the local publication queue.
+TEST(NodeTest, LocalHwm)
+{
+  // LocalHwm is applicable to all backends.
+  EXPECT_EQ(transport::kDefaultLocalHwm, transport::localHwm());
+}
+
 //////////////////////////////////////////////////
 /// \brief Check that we destruct a Node object before a Node::Publisher.
 TEST(NodePubTest, DestructionOrder)

test/BUILD.bazel

@@ -65,6 +65,7 @@ integration_test_sources = [
     "authPubSub.cc",
     "scopedTopic.cc",
     "callback_scope_TEST.cc",
+    "localHwm.cc",
     "statistics.cc",
     "twoProcsPubSub.cc",
     "twoProcsPubSubStats.cc",

test/integration/CMakeLists.txt

@@ -4,6 +4,7 @@ set(tests
   authPubSub.cc
   scopedTopic.cc
   callback_scope_TEST.cc
+  localHwm.cc
   statistics.cc
   twoProcsPubSub.cc
   twoProcsPubSubStats.cc