Drop messages when connection send buffer limit has been reached (#126)

**Public-Facing Changes** - Drop messages when connection send buffer limit has been reached **Description** We currently do have the problem that slow clients/connections can cause the server's connection send buffer to grow indefinitely potentially leading to the server being killed as it consumes too much memory. This PR introduces a user configurable send buffer limit. If the limit has been reached, messages that are to be send over this connection will be dropped in order to prevent the send buffer growing indefinitely. See also #116
foxglove · Jan 3, 2023 · e417709 · e417709
1 parent ac3e478
commit e417709
Show file tree

Hide file tree

Showing 4 changed files with 45 additions and 6 deletions.
diff --git a/README.md b/README.md
@@ -70,6 +70,7 @@ Parameters are provided to configure the behavior of the bridge. These parameter
  * __certfile__: Path to the certificate to use for TLS. Required when __tls__ is set to `true`. Defaults to `""`.
  * __keyfile__: Path to the private key to use for TLS. Required when __tls__ is set to `true`. Defaults to `""`.
  * __topic_whitelist__: List of regular expressions ([ECMAScript grammar](https://en.cppreference.com/w/cpp/regex/ecmascript)) of whitelisted topic names. Defaults to `[".*"]`.
+ * __send_buffer_limit__: Connection send buffer limit in bytes. Messages will be dropped when a connection's send buffer reaches this limit to avoid a queue of outdated messages building up. Defaults to `10000000` (10 MB).
  * (ROS 1) __max_update_ms__: The maximum number of milliseconds to wait in between polling `roscore` for new topics, services, or parameters. Defaults to `5000`.
  * (ROS 2) __num_threads__: The number of threads to use for the ROS node executor. This controls the number of subscriptions that can be processed in parallel. 0 means one thread per CPU core. Defaults to `0`.
  * (ROS 2) __max_qos_depth__: Maximum depth used for the QoS profile of subscriptions. Defaults to `10`.

diff --git a/foxglove_bridge_base/include/foxglove_bridge/websocket_server.hpp b/foxglove_bridge_base/include/foxglove_bridge/websocket_server.hpp
@@ -29,8 +29,11 @@ using ConnHandle = websocketpp::connection_hdl;
 using OpCode = websocketpp::frame::opcode::value;
 
 static const websocketpp::log::level APP = websocketpp::log::alevel::app;
+static const websocketpp::log::level WARNING = websocketpp::log::elevel::warn;
 static const websocketpp::log::level RECOVERABLE = websocketpp::log::elevel::rerror;
 
+constexpr size_t DEFAULT_SEND_BUFFER_LIMIT_BYTES = 10000000UL;  // 10 MB
+
 constexpr uint32_t Integer(const std::string_view str) {
   uint32_t result = 0x811C9DC5;  // FNV-1a 32-bit algorithm
   for (char c : str) {
@@ -167,8 +170,9 @@ class Server final : public ServerInterface {
   static bool USES_TLS;
 
   explicit Server(std::string name, LogCallback logger,
-                  const std::vector<std::string>& capabilities, const std::string& certfile = "",
-                  const std::string& keyfile = "");
+                  const std::vector<std::string>& capabilities,
+                  size_t send_buffer_limit_bytes = DEFAULT_SEND_BUFFER_LIMIT_BYTES,
+                  const std::string& certfile = "", const std::string& keyfile = "");
   virtual ~Server();
 
   Server(const Server&) = delete;
@@ -213,6 +217,7 @@ class Server final : public ServerInterface {
   std::string _name;
   LogCallback _logger;
   std::vector<std::string> _capabilities;
+  size_t _send_buffer_limit_bytes;
   std::string _certfile;
   std::string _keyfile;
   ServerType _server;
@@ -248,10 +253,12 @@ class Server final : public ServerInterface {
 template <typename ServerConfiguration>
 inline Server<ServerConfiguration>::Server(std::string name, LogCallback logger,
                                            const std::vector<std::string>& capabilities,
+                                           size_t send_buffer_limit_bytes,
                                            const std::string& certfile, const std::string& keyfile)
     : _name(std::move(name))
     , _logger(logger)
     , _capabilities(capabilities)
+    , _send_buffer_limit_bytes(send_buffer_limit_bytes)
     , _certfile(certfile)
     , _keyfile(keyfile) {
   // Redirect logging
@@ -800,6 +807,17 @@ inline void Server<ServerConfiguration>::broadcastChannels() {
 template <typename ServerConfiguration>
 inline void Server<ServerConfiguration>::sendMessage(ConnHandle clientHandle, ChannelId chanId,
                                                      uint64_t timestamp, std::string_view data) {
+  std::error_code ec;
+  const auto con = _server.get_con_from_hdl(clientHandle, ec);
+  if (ec || !con) {
+    return;
+  }
+
+  const auto bufferSizeinBytes = con->get_buffered_amount();
+  if (bufferSizeinBytes >= _send_buffer_limit_bytes) {
+    return;
+  }
+
   SubscriptionId subId = std::numeric_limits<SubscriptionId>::max();
 
   {

diff --git a/ros1_foxglove_bridge/src/ros1_foxglove_bridge_nodelet.cpp b/ros1_foxglove_bridge/src/ros1_foxglove_bridge_nodelet.cpp
@@ -41,6 +41,8 @@ class FoxgloveBridge : public nodelet::Nodelet {
     auto& nhp = getPrivateNodeHandle();
     const auto address = nhp.param<std::string>("address", DEFAULT_ADDRESS);
     const int port = nhp.param<int>("port", DEFAULT_PORT);
+    const auto send_buffer_limit = static_cast<size_t>(
+      nhp.param<int>("send_buffer_limit", foxglove::DEFAULT_SEND_BUFFER_LIMIT_BYTES));
     const auto useTLS = nhp.param<bool>("tls", false);
     const auto certfile = nhp.param<std::string>("certfile", "");
     const auto keyfile = nhp.param<std::string>("keyfile", "");
@@ -73,10 +75,11 @@ class FoxgloveBridge : public nodelet::Nodelet {
         std::bind(&FoxgloveBridge::logHandler, this, std::placeholders::_1, std::placeholders::_2);
       if (useTLS) {
         _server = std::make_unique<foxglove::Server<foxglove::WebSocketTls>>(
-          "foxglove_bridge", std::move(logHandler), serverCapabilities, certfile, keyfile);
+          "foxglove_bridge", std::move(logHandler), serverCapabilities, send_buffer_limit, certfile,
+          keyfile);
       } else {
         _server = std::make_unique<foxglove::Server<foxglove::WebSocketNoTls>>(
-          "foxglove_bridge", std::move(logHandler), serverCapabilities);
+          "foxglove_bridge", std::move(logHandler), serverCapabilities, send_buffer_limit);
       }
 
       _server->setSubscribeHandler(std::bind(&FoxgloveBridge::subscribeHandler, this,

diff --git a/ros2_foxglove_bridge/src/ros2_foxglove_bridge.cpp b/ros2_foxglove_bridge/src/ros2_foxglove_bridge.cpp
@@ -99,6 +99,20 @@ class FoxgloveBridge : public rclcpp::Node {
     this->declare_parameter<std::vector<std::string>>(
       topicWhiteListDescription.name, std::vector<std::string>({".*"}), topicWhiteListDescription);
 
+    auto sendBufferLimit = rcl_interfaces::msg::ParameterDescriptor{};
+    sendBufferLimit.name = "send_buffer_limit";
+    sendBufferLimit.type = rcl_interfaces::msg::ParameterType::PARAMETER_INTEGER;
+    sendBufferLimit.description =
+      "Connection send buffer limit in bytes. Messages will be dropped when a connection's send "
+      "buffer reaches this limit to avoid a queue of outdated messages building up.";
+    sendBufferLimit.integer_range.resize(1);
+    sendBufferLimit.integer_range[0].from_value = 0;
+    sendBufferLimit.integer_range[0].to_value = std::numeric_limits<int64_t>::max();
+    sendBufferLimit.read_only = true;
+    this->declare_parameter(sendBufferLimit.name,
+                            static_cast<int64_t>(foxglove::DEFAULT_SEND_BUFFER_LIMIT_BYTES),
+                            sendBufferLimit);
+
     const auto regexPatterns =
       this->get_parameter(topicWhiteListDescription.name).as_string_array();
     _topicWhitelistPatterns.reserve(regexPatterns.size());
@@ -112,6 +126,8 @@ class FoxgloveBridge : public rclcpp::Node {
       }
     }
 
+    const auto send_buffer_limit =
+      static_cast<size_t>(this->get_parameter("send_buffer_limit").as_int());
     const auto useTLS = this->get_parameter("tls").as_bool();
     const auto certfile = this->get_parameter("certfile").as_string();
     const auto keyfile = this->get_parameter("keyfile").as_string();
@@ -127,10 +143,11 @@ class FoxgloveBridge : public rclcpp::Node {
 
     if (useTLS) {
       _server = std::make_unique<foxglove::Server<foxglove::WebSocketTls>>(
-        "foxglove_bridge", std::move(logHandler), serverCapabilities, certfile, keyfile);
+        "foxglove_bridge", std::move(logHandler), serverCapabilities, send_buffer_limit, certfile,
+        keyfile);
     } else {
       _server = std::make_unique<foxglove::Server<foxglove::WebSocketNoTls>>(
-        "foxglove_bridge", std::move(logHandler), serverCapabilities);
+        "foxglove_bridge", std::move(logHandler), serverCapabilities, send_buffer_limit);
     }
     _server->setSubscribeHandler(std::bind(&FoxgloveBridge::subscribeHandler, this, _1, _2));
     _server->setUnsubscribeHandler(std::bind(&FoxgloveBridge::unsubscribeHandler, this, _1, _2));