Merge pull request OpenDDS#4928 from OpenDDS/rtps-relay-synchronous-output

RtpsRelay output queue may be ill-behaved

Author: jrw972

docs/devguide/internet_enabled_rtps.rst

@@ -339,6 +339,10 @@ The command-line options for the RtpsRelay:
 
   Use a thread pool with this many threads (default 1) to handle input/output/timer events.
 
+.. option:: -SynchronousOutput 0|1
+
+  Send messages immediately, defaults to 0 (disabled).
+
 .. _internet_enabled_rtps--deployment-considerations:
 
 Deployment Considerations

docs/news.d/rtpsrelay-synchronous-output.rst

@@ -0,0 +1,5 @@
+.. news-prs: 4928
+
+.. news-start-section: Additions
+- Added :option:`RtpsRelay -SynchronousOutput` option to RtpsRelay.
+.. news-end-section

tests/DCPS/RtpsRelay/Smoke/run_test.pl

@@ -75,7 +75,8 @@ sub get_relay_args {
     "-VerticalAddress ${port_digit}444",
     "-HorizontalAddress 127.0.0.1:11${port_digit}44",
     "-MetaDiscoveryAddress 127.0.0.1:808${n}",
-    "-ORBVerboseLogging 1"
+    "-ORBVerboseLogging 1",
+    "-SynchronousOutput 1"
   );
 }
 

tools/rtpsrelay/Config.h

@@ -38,6 +38,7 @@ class Config {
     , admission_max_participants_high_water_(0)
     , admission_max_participants_low_water_(0)
     , handler_threads_(1)
+    , synchronous_output_(false)
   {}
 
   void relay_id(const std::string& value)
@@ -360,6 +361,16 @@ class Config {
     return handler_threads_;
   }
 
+  void synchronous_output(bool flag)
+  {
+    synchronous_output_ = flag;
+  }
+
+  bool synchronous_output() const
+  {
+    return synchronous_output_;
+  }
+
 private:
   std::string relay_id_;
   OpenDDS::DCPS::GUID_t application_participant_guid_;
@@ -393,6 +404,7 @@ class Config {
   size_t admission_max_participants_high_water_;
   size_t admission_max_participants_low_water_;
   size_t handler_threads_;
+  bool synchronous_output_;
 };
 
 }

tools/rtpsrelay/RelayHandler.cpp

@@ -175,39 +175,16 @@ int RelayHandler::handle_output(ACE_HANDLE)
 
   if (!outgoing_.empty()) {
     const auto& out = outgoing_.front();
+    size_t total_bytes;
 
-    const int BUFFERS_SIZE = 2;
-    iovec buffers[BUFFERS_SIZE];
-    size_t total_bytes = 0;
-
-    int idx = 0;
-    for (ACE_Message_Block* block = out.message_block.get(); block && idx < BUFFERS_SIZE; block = block->cont(), ++idx) {
-      buffers[idx].iov_base = block->rd_ptr();
-#ifdef _MSC_VER
-#pragma warning(push)
-      // iov_len is 32-bit on 64-bit VC++, but we don't want a cast here
-      // since on other platforms iov_len is 64-bit
-#pragma warning(disable : 4267)
-#endif
-      buffers[idx].iov_len = block->length();
-      total_bytes += buffers[idx].iov_len;
-#ifdef _MSC_VER
-#pragma warning(pop)
-#endif
-    }
-
-    const auto bytes = socket_.send(buffers, idx, out.address, 0);
-
-    if (bytes < 0) {
+    if (send_i(out, total_bytes) < 0) {
       HANDLER_ERROR((LM_ERROR, "(%P|%t) ERROR: RelayHandler::handle_output %C failed to send to %C: %m\n",
                      name_.c_str(), OpenDDS::DCPS::LogAddr(out.address).c_str()));
       const auto new_now = OpenDDS::DCPS::MonotonicTimePoint::now();
-      stats_reporter_.dropped_message(
-        total_bytes, new_now - now, new_now - out.timestamp, now, out.type);
+      stats_reporter_.dropped_message(total_bytes, new_now - now, new_now - out.timestamp, now, out.type);
     } else {
       const auto new_now = OpenDDS::DCPS::MonotonicTimePoint::now();
-      stats_reporter_.output_message(
-        total_bytes, new_now - now, new_now - out.timestamp, now, out.type);
+      stats_reporter_.output_message(total_bytes, new_now - now, new_now - out.timestamp, now, out.type);
     }
 
     outgoing_.pop();
@@ -225,15 +202,55 @@ void RelayHandler::enqueue_message(const ACE_INET_Addr& addr,
                                    const OpenDDS::DCPS::MonotonicTimePoint& now,
                                    MessageType type)
 {
-  ACE_GUARD(ACE_Thread_Mutex, g, outgoing_mutex_);
+  const Element out(addr, msg, now, type);
+  if (config_.synchronous_output()) {
+    size_t total_bytes;
+
+    if (send_i(out, total_bytes) < 0) {
+      HANDLER_ERROR((LM_ERROR, "(%P|%t) ERROR: RelayHandler::enqueue_message %C failed to send to %C: %m\n",
+                     name_.c_str(), OpenDDS::DCPS::LogAddr(out.address).c_str()));
+      stats_reporter_.dropped_message(total_bytes, OpenDDS::DCPS::TimeDuration::zero_value, OpenDDS::DCPS::TimeDuration::zero_value, now, out.type);
+    } else {
+      stats_reporter_.output_message(total_bytes, OpenDDS::DCPS::TimeDuration::zero_value, OpenDDS::DCPS::TimeDuration::zero_value, now, out.type);
+    }
+
+  } else {
+    ACE_GUARD(ACE_Thread_Mutex, g, outgoing_mutex_);
+
+    const auto empty = outgoing_.empty();
+
+    outgoing_.push(out);
+    stats_reporter_.max_queue_size(outgoing_.size(), now);
+    if (empty) {
+      reactor()->register_handler(this, WRITE_MASK);
+    }
+  }
+}
 
-  const auto empty = outgoing_.empty();
+ssize_t RelayHandler::send_i(const Element& out,
+                             size_t& total_bytes)
+{
+  const int BUFFERS_SIZE = 2;
+  iovec buffers[BUFFERS_SIZE];
+  total_bytes = 0;
 
-  outgoing_.push(Element(addr, msg, now, type));
-  stats_reporter_.max_queue_size(outgoing_.size(), now);
-  if (empty) {
-    reactor()->register_handler(this, WRITE_MASK);
+  int idx = 0;
+  for (ACE_Message_Block* block = out.message_block.get(); block && idx < BUFFERS_SIZE; block = block->cont(), ++idx) {
+    buffers[idx].iov_base = block->rd_ptr();
+#ifdef _MSC_VER
+#pragma warning(push)
+    // iov_len is 32-bit on 64-bit VC++, but we don't want a cast here
+    // since on other platforms iov_len is 64-bit
+#pragma warning(disable : 4267)
+#endif
+    buffers[idx].iov_len = block->length();
+    total_bytes += buffers[idx].iov_len;
+#ifdef _MSC_VER
+#pragma warning(pop)
+#endif
   }
+
+  return socket_.send(buffers, idx, out.address, 0);
 }
 
 VerticalHandler::VerticalHandler(const Config& config,
@@ -595,7 +612,7 @@ CORBA::ULong VerticalHandler::send(GuidAddrSet::Proxy& proxy,
   CORBA::ULong sent = 0;
   for (const auto& addr : address_set) {
     if (addr != horizontal_address_) {
-      horizontal_handler_->enqueue_message(addr, to_partitions, to_guids, msg, now);
+      horizontal_handler_->enqueue_or_send_message(addr, to_partitions, to_guids, msg, now);
       ++sent;
     } else {
       // Local recipients.
@@ -641,8 +658,6 @@ size_t VerticalHandler::send(const ACE_INET_Addr& addr,
   message.block = block.get();
   serializer << message;
   RelayHandler::enqueue_message(addr, block, now, type);
-  const auto new_now = OpenDDS::DCPS::MonotonicTimePoint::now();
-  stats_reporter_.output_message(length, new_now - now, new_now - now, now, type);
   return length;
 }
 
@@ -663,11 +678,11 @@ HorizontalHandler::HorizontalHandler(const Config& config,
   , vertical_handler_(nullptr)
 {}
 
-void HorizontalHandler::enqueue_message(const ACE_INET_Addr& addr,
-                                        const StringSet& to_partitions,
-                                        const GuidSet& to_guids,
-                                        const OpenDDS::DCPS::Lockable_Message_Block_Ptr& msg,
-                                        const OpenDDS::DCPS::MonotonicTimePoint& now)
+void HorizontalHandler::enqueue_or_send_message(const ACE_INET_Addr& addr,
+                                                const StringSet& to_partitions,
+                                                const GuidSet& to_guids,
+                                                const OpenDDS::DCPS::Lockable_Message_Block_Ptr& msg,
+                                                const OpenDDS::DCPS::MonotonicTimePoint& now)
 {
   using namespace OpenDDS::DCPS;
 
@@ -984,7 +999,7 @@ int SpdpHandler::handle_exception(ACE_HANDLE /*fd*/)
       const auto pos = proxy.find(fan_in_from_guid);
       if (pos != proxy.end() && pos->second.spdp_message) {
         // Send the SPDP message horizontally.  We may be sending to ourselves which is okay.
-        horizontal_handler_->enqueue_message(fan_in_replay_address, StringSet(), fan_in_to_guid_set, pos->second.spdp_message, now);
+        horizontal_handler_->enqueue_or_send_message(fan_in_replay_address, StringSet(), fan_in_to_guid_set, pos->second.spdp_message, now);
       }
     }
 

tools/rtpsrelay/RelayHandler.h

@@ -75,6 +75,8 @@ class RelayHandler : public ACE_Event_Handler {
       , type(type)
     {}
   };
+  ssize_t send_i(const Element& out,
+                 size_t& total_bytes);
   using OutgoingType = std::queue<Element>;
   OutgoingType outgoing_;
   mutable ACE_Thread_Mutex outgoing_mutex_;
@@ -199,11 +201,11 @@ class HorizontalHandler : public RelayHandler {
 
   void vertical_handler(VerticalHandler* vertical_handler) { vertical_handler_ = vertical_handler; }
 
-  void enqueue_message(const ACE_INET_Addr& addr,
-                       const StringSet& to_partitions,
-                       const GuidSet& to_guids,
-                       const OpenDDS::DCPS::Lockable_Message_Block_Ptr& msg,
-                       const OpenDDS::DCPS::MonotonicTimePoint& now);
+  void enqueue_or_send_message(const ACE_INET_Addr& addr,
+                               const StringSet& to_partitions,
+                               const GuidSet& to_guids,
+                               const OpenDDS::DCPS::Lockable_Message_Block_Ptr& msg,
+                               const OpenDDS::DCPS::MonotonicTimePoint& now);
 
 private:
   const GuidPartitionTable& guid_partition_table_;

tools/rtpsrelay/RtpsRelay.cpp

@@ -235,6 +235,9 @@ int run(int argc, ACE_TCHAR* argv[])
     } else if ((arg = args.get_the_parameter("-HandlerThreads"))) {
       config.handler_threads(std::atoi(arg));
       args.consume_arg();
+    } else if ((arg = args.get_the_parameter("-SynchronousOutput"))) {
+      config.synchronous_output(ACE_OS::atoi(arg));
+      args.consume_arg();
     } else if ((arg = args.get_the_parameter("-MaxIpsPerClient"))) {
       config.max_ips_per_client(ACE_OS::atoi(arg));
       args.consume_arg();