diff --git a/Migration.md b/Migration.md index e9f0e821e..2d351c914 100644 --- a/Migration.md +++ b/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 diff --git a/include/gz/transport/Node.hh b/include/gz/transport/Node.hh index 6836eac5c..2cb17fc14 100644 --- a/include/gz/transport/Node.hh +++ b/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 diff --git a/include/gz/transport/NodeShared.hh b/include/gz/transport/NodeShared.hh index 472a0282d..d351e8fce 100644 --- a/include/gz/transport/NodeShared.hh +++ b/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. diff --git a/include/gz/transport/TransportTypes.hh b/include/gz/transport/TransportTypes.hh index 174deb126..ccaab5c50 100644 --- a/include/gz/transport/TransportTypes.hh +++ b/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 diff --git a/src/Node.cc b/src/Node.cc index b34d9d574..ed73bfa52 100644 --- a/src/Node.cc +++ b/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 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. diff --git a/src/NodeShared.cc b/src/NodeShared.cc index 64b192ee6..b5410a027 100644 --- a/src/NodeShared.cc +++ b/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(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 _msgDetails) +{ + std::string droppedTopic; + { + std::unique_lock 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. diff --git a/src/NodeSharedPrivate.hh b/src/NodeSharedPrivate.hh index 689103dc1..dec086d33 100644 --- a/src/NodeSharedPrivate.hh +++ b/src/NodeSharedPrivate.hh @@ -21,6 +21,7 @@ #include #include #include +#include #include #include #include @@ -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>; + /// \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> 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> + 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 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 _msgDetails); /// \brief used to signal when new work is available public: std::condition_variable signalNewPub; diff --git a/src/Node_TEST.cc b/src/Node_TEST.cc index 51d1a7937..f7c1e0848 100644 --- a/src/Node_TEST.cc +++ b/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) diff --git a/test/BUILD.bazel b/test/BUILD.bazel index 1ada6a345..6d4d55d77 100644 --- a/test/BUILD.bazel +++ b/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", diff --git a/test/integration/CMakeLists.txt b/test/integration/CMakeLists.txt index b51dae7a9..c3b04b5b9 100644 --- a/test/integration/CMakeLists.txt +++ b/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 diff --git a/test/integration/localHwm.cc b/test/integration/localHwm.cc new file mode 100644 index 000000000..ddbdf086d --- /dev/null +++ b/test/integration/localHwm.cc @@ -0,0 +1,264 @@ +/* + * Copyright (C) 2026 Open Source Robotics Foundation + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + * +*/ +#include + +#include +#include +#include +#include +#include +#include +#include + +#include "gtest/gtest.h" +#include "gz/transport/Node.hh" + +#include + +#include "test_utils.hh" + +using namespace gz; +using namespace std::chrono_literals; + +// The capacity of the local publication queue used in this test. It is set +// via the GZ_TRANSPORT_LOCAL_HWM environment variable in main(). +static const int kLocalHwm = 5; + +/// \brief Helper that subscribes to a topic and optionally blocks the local +/// delivery thread inside the callback of the first received message until +/// the test releases it. All the received values are recorded. +class BlockingSubscriber +{ + /// \brief Constructor. + /// \param[in] _node Node used to subscribe. + /// \param[in] _topic Topic name. + /// \param[in] _blockFirst True to block the delivery thread inside the + /// callback of the first received message. + public: BlockingSubscriber(transport::Node &_node, + const std::string &_topic, bool _blockFirst = true) + { + this->cb = [this, _blockFirst](const msgs::Int32 &_msg) + { + std::unique_lock lk(this->mutex); + this->received.push_back(_msg.data()); + this->cv.notify_all(); + if (_blockFirst && this->received.size() == 1u) + { + // Block the local delivery thread until ReleaseAndWait() is + // called. + this->cv.wait_for(lk, 30s, [this]{return this->released;}); + } + }; + EXPECT_TRUE(_node.Subscribe(_topic, this->cb)); + } + + /// \brief Wait until the delivery thread is blocked inside the callback + /// of the first received message. + /// \return True if the first message was received before the timeout. + public: bool WaitFirstMsg() + { + std::unique_lock lk(this->mutex); + return this->cv.wait_for(lk, 10s, [this]{return !this->received.empty();}); + } + + /// \brief Wait until _count messages have been received in total. + /// \param[in] _count Expected total number of received messages. + /// \return True if _count messages were received before the timeout. + public: bool WaitReceived(size_t _count) + { + std::unique_lock lk(this->mutex); + return this->cv.wait_for(lk, 10s, + [this, _count]{return this->received.size() >= _count;}); + } + + /// \brief Unblock the delivery thread and wait until _count messages have + /// been received in total. + /// \param[in] _count Expected total number of received messages. + /// \return True if _count messages were received before the timeout. + public: bool ReleaseAndWait(size_t _count) + { + { + std::lock_guard lk(this->mutex); + this->released = true; + } + this->cv.notify_all(); + return this->WaitReceived(_count); + } + + /// \brief Get the values received so far. + /// \return The received values, in reception order. + public: std::vector Received() + { + std::lock_guard lk(this->mutex); + return this->received; + } + + /// \brief Protects all the members below. + private: std::mutex mutex; + + /// \brief Signals new received messages and the release of the callback. + private: std::condition_variable cv; + + /// \brief True when the callback should stop blocking. + private: bool released = false; + + /// \brief Values received so far. + private: std::vector received; + + /// \brief Subscription callback. + private: std::function cb; +}; + +////////////////////////////////////////////////// +/// \brief Check that when a topic exceeds the capacity of the local +/// publication queue, its oldest queued messages are dropped and the +/// newest ones are delivered. +TEST(localHwmTest, DropOldestWhenFull) +{ + ASSERT_EQ(kLocalHwm, transport::localHwm()); + + transport::Node node; + auto pub = node.Advertise("/foo"); + ASSERT_TRUE(pub); + + BlockingSubscriber sub(node, "/foo"); + + // Publish a first message and wait until the delivery thread is blocked + // inside its callback. At this point the local publication queue is empty. + msgs::Int32 msg; + msg.set_data(0); + EXPECT_TRUE(pub.Publish(msg)); + ASSERT_TRUE(sub.WaitFirstMsg()); + + // Publish more messages than the capacity of the queue. The oldest + // messages should be dropped, keeping only the newest kLocalHwm ones. + for (int i = 1; i <= 2 * kLocalHwm; ++i) + { + msg.set_data(i); + EXPECT_TRUE(pub.Publish(msg)); + } + + // Unblock the delivery thread and wait for the queued messages. + ASSERT_TRUE(sub.ReleaseAndWait(1u + kLocalHwm)); + + // Give the delivery thread some time to process unexpected extra messages. + std::this_thread::sleep_for(200ms); + + // We expect the first message and the newest kLocalHwm ones. + std::vector expected = {0, 6, 7, 8, 9, 10}; + EXPECT_EQ(expected, sub.Received()); +} + +////////////////////////////////////////////////// +/// \brief Check that no messages are dropped while a topic stays within the +/// capacity of the local publication queue. +TEST(localHwmTest, NoDropsUnderCapacity) +{ + ASSERT_EQ(kLocalHwm, transport::localHwm()); + + transport::Node node; + auto pub = node.Advertise("/bar"); + ASSERT_TRUE(pub); + + BlockingSubscriber sub(node, "/bar"); + + // Publish a first message and wait until the delivery thread is blocked + // inside its callback. At this point the local publication queue is empty. + msgs::Int32 msg; + msg.set_data(100); + EXPECT_TRUE(pub.Publish(msg)); + ASSERT_TRUE(sub.WaitFirstMsg()); + + // Fill the queue up to its capacity. No messages should be dropped. + for (int i = 101; i <= 100 + kLocalHwm; ++i) + { + msg.set_data(i); + EXPECT_TRUE(pub.Publish(msg)); + } + + // Unblock the delivery thread and wait for the queued messages. + ASSERT_TRUE(sub.ReleaseAndWait(1u + kLocalHwm)); + + std::vector expected = {100, 101, 102, 103, 104, 105}; + EXPECT_EQ(expected, sub.Received()); +} + +////////////////////////////////////////////////// +/// \brief Check that a topic exceeding its queue capacity does not cause +/// drops on other topics, even when their queued messages are older. +TEST(localHwmTest, TopicIsolation) +{ + ASSERT_EQ(kLocalHwm, transport::localHwm()); + + transport::Node node; + auto pubA = node.Advertise("/iso_a"); + auto pubB = node.Advertise("/iso_b"); + ASSERT_TRUE(pubA); + ASSERT_TRUE(pubB); + + BlockingSubscriber subA(node, "/iso_a"); + BlockingSubscriber subB(node, "/iso_b", false); + + // Publish a first message and wait until the delivery thread is blocked + // inside its callback. + msgs::Int32 msg; + msg.set_data(0); + EXPECT_TRUE(pubA.Publish(msg)); + ASSERT_TRUE(subA.WaitFirstMsg()); + + // Queue a few messages of the quiet topic first, so that they become the + // oldest entries of the whole queue. + for (int i = 1; i <= 3; ++i) + { + msg.set_data(i); + EXPECT_TRUE(pubB.Publish(msg)); + } + + // Flood the noisy topic beyond its capacity. Only its own oldest + // messages should be dropped, never the older messages of the quiet + // topic. + for (int i = 1; i <= 2 * kLocalHwm; ++i) + { + msg.set_data(i); + EXPECT_TRUE(pubA.Publish(msg)); + } + + // Unblock the delivery thread and wait for all the expected messages. + ASSERT_TRUE(subA.ReleaseAndWait(1u + kLocalHwm)); + ASSERT_TRUE(subB.WaitReceived(3u)); + + std::vector expectedA = {0, 6, 7, 8, 9, 10}; + std::vector expectedB = {1, 2, 3}; + EXPECT_EQ(expectedA, subA.Received()); + EXPECT_EQ(expectedB, subB.Received()); +} + +////////////////////////////////////////////////// +int main(int argc, char **argv) +{ + // Get a random partition name. + std::string partition = testing::getRandomNumber(); + + // Set the partition name for this process. + gz::utils::setenv("GZ_PARTITION", partition); + + // Use a small local publication queue to make it easy to fill it up. + gz::utils::setenv("GZ_TRANSPORT_LOCAL_HWM", std::to_string(kLocalHwm)); + + ::testing::InitGoogleTest(&argc, argv); + return RUN_ALL_TESTS(); +} diff --git a/tutorials/20_env_variables.md b/tutorials/20_env_variables.md index bcc494648..f0024b280 100644 --- a/tutorials/20_env_variables.md +++ b/tutorials/20_env_variables.md @@ -72,6 +72,18 @@ Below are descriptions of the available environment variables: overrides take priority. Example: `GZ_TRANSPORT_ZENOH_CONFIG_OVERRIDE="transport/link/tx/queue/size/data=8;transport/shared_memory/enabled=true"` * *Available in backend:*: zenoh +* **GZ_TRANSPORT_LOCAL_HWM** + * *Value allowed*: Any non-negative number. + * *Description*: Specifies the capacity (High Water Mark) of the queue + that stores messages to be delivered to local (intraprocess) subscribers. + This limit is applied per topic. When a topic reaches this capacity, its + oldest queued message is dropped to make room for a new one, keeping the + freshest data and preventing the queue from growing without limit when + the local subscription callbacks are slower than the publication rate. + A value of 0 means "infinite" capacity, which will make your process run + out of memory if the local subscribers cannot keep up. + * *Default value*: 1000. + * *Available in backend:*: zeromq, zenoh * **GZ_TRANSPORT_LOG_SQL_PATH** * *Value allowed*: Any path * *Description*: Path to the SQL files used by logging. This does not