udp: limit number of reads per event loop

api/envoy/config/listener/v3/quic_config.proto

@@ -6,9 +6,11 @@ import "envoy/config/core/v3/base.proto";
 import "envoy/config/core/v3/protocol.proto";
 
 import "google/protobuf/duration.proto";
+import "google/protobuf/wrappers.proto";
 
 import "udpa/annotations/status.proto";
 import "udpa/annotations/versioning.proto";
+import "validate/validate.proto";
 
 option java_package = "io.envoyproxy.envoy.config.listener.v3";
 option java_outer_classname = "QuicConfigProto";
@@ -18,6 +20,7 @@ option (udpa.annotations.file_status).package_version_status = ACTIVE;
 // [#protodoc-title: QUIC listener config]
 
 // Configuration specific to the UDP QUIC listener.
+// [#next-free-field: 6]
 message QuicProtocolOptions {
   option (udpa.annotations.versioning).previous_message_type =
       "envoy.api.v2.listener.QuicProtocolOptions";
@@ -35,4 +38,9 @@ message QuicProtocolOptions {
   // Runtime flag that controls whether the listener is enabled or not. If not specified, defaults
   // to enabled.
   core.v3.RuntimeFeatureFlag enabled = 4;
+
+  // A multiplier to number of connections which is used to determine how many packets to read per event loop. A reasonable ratio should allow the listener to process enough payload but not starve TCP and other UDP sockets and also prevent long event loop duration.
+  // The default value is 32. This means if there are N QUIC connections, the total number of packets to read in each READ event will be 32 * N.
+  google.protobuf.UInt32Value packets_received_to_connection_count_ratio = 5
+      [(validate.rules).uint32 = {gte: 1}];
 }

include/envoy/network/listener.h

@@ -330,6 +330,11 @@ class UdpListenerCallbacks {
    * Posts ``data`` to be delivered on this worker.
    */
   virtual void post(Network::UdpRecvData&& data) PURE;
+
+  /**
+   * An estimated number of packets this callback expects to process in current READ event.
+   */
+  virtual size_t numPacketsExpectedPerEventLoop() const PURE;
 };
 
 using UdpListenerCallbacksOptRef = absl::optional<std::reference_wrapper<UdpListenerCallbacks>>;

source/common/network/udp_listener_impl.cc

@@ -75,7 +75,12 @@ void UdpListenerImpl::handleReadCallback() {
   const Api::IoErrorPtr result = Utility::readPacketsFromSocket(
       socket_->ioHandle(), *socket_->addressProvider().localAddress(), *this, time_source_,
       config_.prefer_gro_, packets_dropped_);
-  // TODO(mattklein123): Handle no error when we limit the number of packets read.
+  if (result == nullptr) {
+    // No error. The number of reads was limited by read rate. There are more packets to read.
+    // Register READ for next event loop.
+    socket_->ioHandle().activateFileEvents(Event::FileReadyType::Read);
+    return;
+  }
   if (result->getErrorCode() != Api::IoError::IoErrorCode::Again) {
     // TODO(mattklein123): When rate limited logging is implemented log this at error level
     // on a periodic basis.

source/common/network/udp_listener_impl.h

@@ -46,6 +46,9 @@ class UdpListenerImpl : public BaseListenerImpl,
                      MonotonicTime receive_time) override;
   uint64_t maxDatagramSize() const override { return config_.max_rx_datagram_size_; }
   void onDatagramsDropped(uint32_t dropped) override { cb_.onDatagramsDropped(dropped); }
+  size_t numPacketsExpectedPerEventLoop() const override {
+    return cb_.numPacketsExpectedPerEventLoop();
+  }
 
 protected:
   void handleWriteCallback();

source/common/network/utility.cc

@@ -576,10 +576,10 @@ void passPayloadToProcessor(uint64_t bytes_read, Buffer::InstancePtr buffer,
 Api::IoCallUint64Result Utility::readFromSocket(IoHandle& handle,
                                                 const Address::Instance& local_address,
                                                 UdpPacketProcessor& udp_packet_processor,
-                                                MonotonicTime receive_time, bool prefer_gro,
+                                                MonotonicTime receive_time, bool use_gro,
                                                 uint32_t* packets_dropped) {
 
-  if (prefer_gro && handle.supportsUdpGro()) {
+  if (use_gro) {
     Buffer::InstancePtr buffer = std::make_unique<Buffer::OwnedImpl>();
     IoHandle::RecvMsgOutput output(1, packets_dropped);
 
@@ -696,11 +696,22 @@ Api::IoErrorPtr Utility::readPacketsFromSocket(IoHandle& handle,
                                                UdpPacketProcessor& udp_packet_processor,
                                                TimeSource& time_source, bool prefer_gro,
                                                uint32_t& packets_dropped) {
+  // Make sure at not too many packets will be read in following loop, but at least it will read
+  // once.
+  size_t num_packets_to_read = std::min(MAX_NUM_PACKETS_PER_EVENT_LOOP,
+                                        udp_packet_processor.numPacketsExpectedPerEventLoop());
+  const bool use_gro = prefer_gro && handle.supportsUdpGro();
+  size_t num_reads =
+      use_gro ? (num_packets_to_read / NUM_DATAGRAMS_PER_GRO_RECEIVE)
+              : (handle.supportsMmsg() ? (num_packets_to_read / NUM_DATAGRAMS_PER_MMSG_RECEIVE)
+                                       : num_packets_to_read);
+  // Make sure at least read once.
+  num_reads = std::max(1ul, num_reads);
   do {
     const uint32_t old_packets_dropped = packets_dropped;
     const MonotonicTime receive_time = time_source.monotonicTime();
     Api::IoCallUint64Result result = Utility::readFromSocket(
-        handle, local_address, udp_packet_processor, receive_time, prefer_gro, &packets_dropped);
+        handle, local_address, udp_packet_processor, receive_time, use_gro, &packets_dropped);
 
     if (!result.ok()) {
       // No more to read or encountered a system error.
@@ -723,6 +734,10 @@ Api::IoErrorPtr Utility::readPacketsFromSocket(IoHandle& handle,
           delta);
       udp_packet_processor.onDatagramsDropped(delta);
     }
+    --num_reads;
+    if (num_reads == 0) {
+      return std::move(result.err_);
+    }
   } while (true);
 }
 

source/common/network/utility.h

@@ -61,11 +61,17 @@ class UdpPacketProcessor {
    * the size of datagrams received, they will be dropped.
    */
   virtual uint64_t maxDatagramSize() const PURE;
+
+  /**
+   * An estimated number of packets to read in each READ event.
+   */
+  virtual size_t numPacketsExpectedPerEventLoop() const PURE;
 };
 
 static const uint64_t DEFAULT_UDP_MAX_DATAGRAM_SIZE = 1500;
 static const uint64_t NUM_DATAGRAMS_PER_GRO_RECEIVE = 16;
 static const uint64_t NUM_DATAGRAMS_PER_MMSG_RECEIVE = 16;
+static const uint64_t MAX_NUM_PACKETS_PER_EVENT_LOOP = 6000;
 
 /**
  * Wrapper which resolves UDP socket proto config with defaults.
@@ -362,18 +368,20 @@ class Utility {
   static Api::IoCallUint64Result readFromSocket(IoHandle& handle,
                                                 const Address::Instance& local_address,
                                                 UdpPacketProcessor& udp_packet_processor,
-                                                MonotonicTime receive_time, bool prefer_gro,
+                                                MonotonicTime receive_time, bool use_gro,
                                                 uint32_t* packets_dropped);
 
   /**
-   * Read available packets from a given UDP socket and pass the packet to a given
-   * UdpPacketProcessor.
+   * Read some packets from a given UDP socket and pass the packet to a given
+   * UdpPacketProcessor. Only read no more than MAX_NUM_PACKETS_PER_EVENT_LOOP packets.
    * @param handle is the UDP socket to read from.
    * @param local_address is the socket's local address used to populate port.
    * @param udp_packet_processor is the callback to receive the packets.
    * @param time_source is the time source used to generate the time stamp of the received packets.
    * @param prefer_gro supplies whether to use GRO if the OS supports it.
    * @param packets_dropped is the output parameter for number of packets dropped in kernel.
+   * Return the io error encountered or nullptr if no io error but read stopped
+   * because of MAX_NUM_PACKETS_PER_EVENT_LOOP.
    *
    * TODO(mattklein123): Allow the number of packets read to be limited for fairness. Currently
    *                     this function will always return an error, even if EAGAIN. In the future

source/common/quic/active_quic_listener.cc

@@ -25,25 +25,29 @@ ActiveQuicListener::ActiveQuicListener(
     uint32_t worker_index, uint32_t concurrency, Event::Dispatcher& dispatcher,
     Network::UdpConnectionHandler& parent, Network::ListenerConfig& listener_config,
     const quic::QuicConfig& quic_config, Network::Socket::OptionsSharedPtr options,
-    bool kernel_worker_routing, const envoy::config::core::v3::RuntimeFeatureFlag& enabled)
+    bool kernel_worker_routing, const envoy::config::core::v3::RuntimeFeatureFlag& enabled,
+    uint32_t packets_received_to_connection_count_ratio)
     : ActiveQuicListener(worker_index, concurrency, dispatcher, parent,
                          listener_config.listenSocketFactory().getListenSocket(), listener_config,
-                         quic_config, std::move(options), kernel_worker_routing, enabled) {}
+                         quic_config, std::move(options), kernel_worker_routing, enabled,
+                         packets_received_to_connection_count_ratio) {}
 
 ActiveQuicListener::ActiveQuicListener(
     uint32_t worker_index, uint32_t concurrency, Event::Dispatcher& dispatcher,
     Network::UdpConnectionHandler& parent, Network::SocketSharedPtr listen_socket,
     Network::ListenerConfig& listener_config, const quic::QuicConfig& quic_config,
     Network::Socket::OptionsSharedPtr options, bool kernel_worker_routing,
-    const envoy::config::core::v3::RuntimeFeatureFlag& enabled)
+    const envoy::config::core::v3::RuntimeFeatureFlag& enabled,
+    uint32_t packets_received_to_connection_count_ratio)
     : Server::ActiveUdpListenerBase(
           worker_index, concurrency, parent, *listen_socket,
           dispatcher.createUdpListener(
               listen_socket, *this,
               listener_config.udpListenerConfig()->config().downstream_socket_config()),
           &listener_config),
       dispatcher_(dispatcher), version_manager_(quic::CurrentSupportedVersions()),
-      kernel_worker_routing_(kernel_worker_routing) {
+      kernel_worker_routing_(kernel_worker_routing),
+      packets_received_to_connection_count_ratio_(packets_received_to_connection_count_ratio) {
   // This flag fix a QUICHE issue which may crash Envoy during connection close.
   SetQuicReloadableFlag(quic_single_ack_in_packet2, true);
   // Do not include 32-byte per-entry overhead while counting header size.
@@ -214,9 +218,16 @@ uint32_t ActiveQuicListener::destination(const Network::UdpRecvData& data) const
   return connection_id_snippet % concurrency_;
 }
 
+size_t ActiveQuicListener::numPacketsExpectedPerEventLoop() const {
+  // Expect each session to read 32 packets per READ event.
+  return quic_dispatcher_->NumSessions() * packets_received_to_connection_count_ratio_;
+}
+
 ActiveQuicListenerFactory::ActiveQuicListenerFactory(
     const envoy::config::listener::v3::QuicProtocolOptions& config, uint32_t concurrency)
-    : concurrency_(concurrency), enabled_(config.enabled()) {
+    : concurrency_(concurrency), enabled_(config.enabled()),
+      packets_received_to_connection_count_ratio_(
+          PROTOBUF_GET_WRAPPED_OR_DEFAULT(config, packets_received_to_connection_count_ratio, 32)) {
   uint64_t idle_network_timeout_ms =
       config.has_idle_timeout() ? DurationUtil::durationToMilliseconds(config.idle_timeout())
                                 : 300000;
@@ -299,9 +310,9 @@ Network::ConnectionHandler::ActiveUdpListenerPtr ActiveQuicListenerFactory::crea
   }
 #endif
 
-  return std::make_unique<ActiveQuicListener>(worker_index, concurrency_, disptacher, parent,
-                                              config, quic_config_, std::move(options),
-                                              kernel_worker_routing, enabled_);
+  return std::make_unique<ActiveQuicListener>(
+      worker_index, concurrency_, disptacher, parent, config, quic_config_, std::move(options),
+      kernel_worker_routing, enabled_, packets_received_to_connection_count_ratio_);
 } // namespace Quic
 
 } // namespace Quic

source/common/quic/active_quic_listener.h

@@ -28,13 +28,15 @@ class ActiveQuicListener : public Envoy::Server::ActiveUdpListenerBase,
                      Network::UdpConnectionHandler& parent,
                      Network::ListenerConfig& listener_config, const quic::QuicConfig& quic_config,
                      Network::Socket::OptionsSharedPtr options, bool kernel_worker_routing,
-                     const envoy::config::core::v3::RuntimeFeatureFlag& enabled);
+                     const envoy::config::core::v3::RuntimeFeatureFlag& enabled,
+                     uint32_t packets_received_to_connection_count_ratio);
 
   ActiveQuicListener(uint32_t worker_index, uint32_t concurrency, Event::Dispatcher& dispatcher,
                      Network::UdpConnectionHandler& parent, Network::SocketSharedPtr listen_socket,
                      Network::ListenerConfig& listener_config, const quic::QuicConfig& quic_config,
                      Network::Socket::OptionsSharedPtr options, bool kernel_worker_routing,
-                     const envoy::config::core::v3::RuntimeFeatureFlag& enabled);
+                     const envoy::config::core::v3::RuntimeFeatureFlag& enabled,
+                     uint32_t packets_received_to_connection_count_ratio);
 
   ~ActiveQuicListener() override;
 
@@ -52,6 +54,7 @@ class ActiveQuicListener : public Envoy::Server::ActiveUdpListenerBase,
   Network::UdpPacketWriter& udpPacketWriter() override { return *udp_packet_writer_; }
   void onDataWorker(Network::UdpRecvData&& data) override;
   uint32_t destination(const Network::UdpRecvData& data) const override;
+  size_t numPacketsExpectedPerEventLoop() const override;
 
   // ActiveListenerImplBase
   void pauseListening() override;
@@ -73,6 +76,8 @@ class ActiveQuicListener : public Envoy::Server::ActiveUdpListenerBase,
   // The number of runs of the event loop in which at least one CHLO was buffered.
   // TODO(ggreenway): Consider making this a published stat, or some variation of this information.
   uint64_t event_loops_with_buffered_chlo_for_test_{0};
+
+  uint32_t packets_received_to_connection_count_ratio_;
 };
 
 using ActiveQuicListenerPtr = std::unique_ptr<ActiveQuicListener>;
@@ -97,6 +102,7 @@ class ActiveQuicListenerFactory : public Network::ActiveUdpListenerFactory,
   const uint32_t concurrency_;
   absl::once_flag install_bpf_once_;
   envoy::config::core::v3::RuntimeFeatureFlag enabled_;
+  const uint32_t packets_received_to_connection_count_ratio_;
 };
 
 } // namespace Quic

source/common/quic/envoy_quic_client_connection.cc

@@ -121,7 +121,10 @@ void EnvoyQuicClientConnection::onFileEvent(uint32_t events) {
     Api::IoErrorPtr err = Network::Utility::readPacketsFromSocket(
         connectionSocket()->ioHandle(), *connectionSocket()->addressProvider().localAddress(),
         *this, dispatcher_.timeSource(), true, packets_dropped_);
-    // TODO(danzh): Handle no error when we limit the number of packets read.
+    if (err == nullptr) {
+      connectionSocket()->ioHandle().activateFileEvents(Event::FileReadyType::Read);
+      return;
+    }
     if (err->getErrorCode() != Api::IoError::IoErrorCode::Again) {
       ENVOY_CONN_LOG(error, "recvmsg result {}: {}", *this, static_cast<int>(err->getErrorCode()),
                      err->getErrorDetails());

source/common/quic/envoy_quic_client_connection.h

@@ -53,6 +53,10 @@ class EnvoyQuicClientConnection : public quic::QuicConnection,
   void onDatagramsDropped(uint32_t) override {
     // TODO(mattklein123): Emit a stat for this.
   }
+  size_t numPacketsExpectedPerEventLoop() const override {
+    // Use ~32k to read the same amount as a TCP connection.
+    return 32u;
+  }
 
   // Register file event and apply socket options.
   void setUpConnectionSocket();