Limits only 16 new QUIC connections can be opened per epoll event.

net/tools/quic/quic_dispatcher.cc

 // Copyright (c) 2012 The Chromium Authors. All rights reserved.
 // Use of this source code is governed by a BSD-style license that can be
 // found in the LICENSE file.
 
 #include "net/tools/quic/quic_dispatcher.h"
 
 #include <utility>
 
 #include "base/debug/stack_trace.h"
 #include "base/logging.h"
 #include "base/macros.h"
 #include "base/stl_util.h"
 #include "net/quic/core/crypto/quic_random.h"
 #include "net/quic/core/quic_bug_tracker.h"
 #include "net/quic/core/quic_flags.h"
 #include "net/quic/core/quic_utils.h"
 #include "net/tools/quic/chlo_extractor.h"
 #include "net/tools/quic/quic_per_connection_packet_writer.h"
 #include "net/tools/quic/quic_simple_server_session.h"
 #include "net/tools/quic/quic_time_wait_list_manager.h"
 #include "net/tools/quic/stateless_rejector.h"
 
 using base::StringPiece;
+using std::list;
 using std::string;
 
 namespace net {
 
 typedef QuicBufferedPacketStore::BufferedPacket BufferedPacket;
 typedef QuicBufferedPacketStore::BufferedPacketList BufferedPacketList;
 typedef QuicBufferedPacketStore::EnqueuePacketResult EnqueuePacketResult;
 
 namespace {
 
@@ skipping 169 matching lines @@
       session_helper_(std::move(session_helper)),
       alarm_factory_(std::move(alarm_factory)),
       delete_sessions_alarm_(
           alarm_factory_->CreateAlarm(new DeleteSessionsAlarm(this))),
       buffered_packets_(this, helper_->GetClock(), alarm_factory_.get()),
       current_packet_(nullptr),
       version_manager_(version_manager),
       framer_(GetSupportedVersions(),
               /*unused*/ QuicTime::Zero(),
               Perspective::IS_SERVER),
-      last_error_(QUIC_NO_ERROR) {
+      last_error_(QUIC_NO_ERROR),
+      new_sessions_allowed_per_event_loop_(0u) {
   framer_.set_visitor(this);
 }
 
 QuicDispatcher::~QuicDispatcher() {
   base::STLDeleteValues(&session_map_);
   base::STLDeleteElements(&closed_session_list_);
 }
 
 void QuicDispatcher::InitializeWithWriter(QuicPacketWriter* writer) {
   DCHECK(writer_ == nullptr);
@@ skipping 373 matching lines @@
   DCHECK(false);
 }
 
 void QuicDispatcher::OnExpiredPackets(
     QuicConnectionId connection_id,
     BufferedPacketList early_arrived_packets) {
   time_wait_list_manager_->AddConnectionIdToTimeWait(
       connection_id, framer_.version(), false, nullptr);
 }
 
+void QuicDispatcher::ProcessBufferedChlos(size_t max_connections_to_create) {
+  // Reset the counter before starting creating connections.
+  new_sessions_allowed_per_event_loop_ = max_connections_to_create;
+  for (; new_sessions_allowed_per_event_loop_ > 0;
+       --new_sessions_allowed_per_event_loop_) {
+    QuicConnectionId connection_id;
+    list<BufferedPacket> packets =
+        buffered_packets_.DeliverPacketsForNextConnection(&connection_id);
+    if (packets.empty()) {
+      return;
+    }
+    QuicServerSessionBase* session =
+        CreateQuicSession(connection_id, packets.front().client_address);
+    DVLOG(1) << "Created new session for " << connection_id;
+    session_map_.insert(std::make_pair(connection_id, session));
+    DeliverPacketsToSession(packets, session);
+  }
+}
+
+bool QuicDispatcher::HasChlosBuffered() const {
+  return buffered_packets_.HasChlosBuffered();
+}
+
 void QuicDispatcher::OnNewConnectionAdded(QuicConnectionId connection_id) {
   VLOG(1) << "Received packet from new connection " << connection_id;
 }
 
 // Return true if there is any packet buffered in the store.
 bool QuicDispatcher::HasBufferedPackets(QuicConnectionId connection_id) {
   return buffered_packets_.HasBufferedPackets(connection_id);
 }
 
 void QuicDispatcher::OnBufferPacketFailure(EnqueuePacketResult result,
@@ skipping 12 matching lines @@
 
 QuicTimeWaitListManager* QuicDispatcher::CreateQuicTimeWaitListManager() {
   return new QuicTimeWaitListManager(writer_.get(), this, helper_.get(),
                                      alarm_factory_.get());
 }
 
 void QuicDispatcher::BufferEarlyPacket(QuicConnectionId connection_id) {
   bool is_new_connection = !buffered_packets_.HasBufferedPackets(connection_id);
   EnqueuePacketResult rs = buffered_packets_.EnqueuePacket(
       connection_id, *current_packet_, current_server_address_,
-      current_client_address_);
+      current_client_address_, /*is_chlo=*/false);
   if (rs != EnqueuePacketResult::SUCCESS) {
     OnBufferPacketFailure(rs, connection_id);
   } else if (is_new_connection) {
     OnNewConnectionAdded(connection_id);
   }
 }
 
 void QuicDispatcher::ProcessChlo() {
+  QUIC_BUG_IF(!FLAGS_quic_buffer_packet_till_chlo &&
+              FLAGS_quic_limit_num_new_sessions_per_epoll_loop)
+      << "Try to limit connection creation per epoll event while not "
+         "supporting packet buffer. "
+         "--quic_limit_num_new_sessions_per_epoll_loop = true "
+         "--quic_buffer_packet_till_chlo = false";
+
+  if (FLAGS_quic_limit_num_new_sessions_per_epoll_loop &&
+      FLAGS_quic_buffer_packet_till_chlo &&
+      new_sessions_allowed_per_event_loop_ <= 0) {
+    // Can't create new session any more. Wait till next event loop.
+    if (!buffered_packets_.HasChloForConnection(current_connection_id_)) {
+      // Only buffer one CHLO per connection.
+      bool is_new_connection =
+          !buffered_packets_.HasBufferedPackets(current_connection_id_);
+      EnqueuePacketResult rs = buffered_packets_.EnqueuePacket(
+          current_connection_id_, *current_packet_, current_server_address_,
+          current_client_address_, /*is_chlo=*/true);
+      if (rs != EnqueuePacketResult::SUCCESS) {
+        OnBufferPacketFailure(rs, current_connection_id_);
+      } else if (is_new_connection) {
+        OnNewConnectionAdded(current_connection_id_);
+      }
+    }
+    return;
+  }
+
   // Creates a new session and process all buffered packets for this connection.
   QuicServerSessionBase* session =
       CreateQuicSession(current_connection_id_, current_client_address_);
   if (FLAGS_quic_enforce_mtu_limit &&
       current_packet().potentially_small_mtu()) {
     session->connection()->set_largest_packet_size_supported(
         kMinimumSupportedPacketSize);
   }
   DVLOG(1) << "Created new session for " << current_connection_id_;
   session_map_.insert(std::make_pair(current_connection_id_, session));
   std::list<BufferedPacket> packets =
       buffered_packets_.DeliverPackets(current_connection_id_);
   // Check if CHLO is the first packet arrived on this connection.
   if (FLAGS_quic_buffer_packet_till_chlo && packets.empty()) {
     OnNewConnectionAdded(current_connection_id_);
   }
   // Process CHLO at first.
   session->ProcessUdpPacket(current_server_address_, current_client_address_,
                             *current_packet_);
 
   // Deliver queued-up packets in the same order as they arrived.
   // Do this even when flag is off because there might be still some packets
   // buffered in the store before flag is turned off.
   DeliverPacketsToSession(packets, session);
+  if (FLAGS_quic_limit_num_new_sessions_per_epoll_loop &&
+      FLAGS_quic_buffer_packet_till_chlo) {
+    --new_sessions_allowed_per_event_loop_;
+  }
 }
 
 bool QuicDispatcher::HandlePacketForTimeWait(
     const QuicPacketPublicHeader& header) {
   if (header.reset_flag) {
     // Public reset packets do not have packet numbers, so ignore the packet.
     return false;
   }
 
   // Switch the framer to the correct version, so that the packet number can
@@ skipping 107 matching lines @@
 void QuicDispatcher::DeliverPacketsToSession(
     const std::list<BufferedPacket>& packets,
     QuicServerSessionBase* session) {
   for (const BufferedPacket& packet : packets) {
     session->ProcessUdpPacket(packet.server_address, packet.client_address,
                               *(packet.packet));
   }
 }
 
 }  // namespace net