test: track simulated timers per dispatcher to simplify thread interactions and locking

test/extensions/filters/http/adaptive_concurrency/controller/gradient_controller_test.cc

@@ -57,7 +57,7 @@ class GradientControllerTest : public testing::Test {
                                                              stats_, random_, time_system_);
 
     // Advance time so that the latency sample calculations don't underflow if monotonic time is 0.
-    time_system_.advanceTimeAsync(std::chrono::hours(42));
+    advanceTimeAndLoop(std::chrono::hours(42));
 
     return config;
   }
@@ -107,6 +107,11 @@ class GradientControllerTest : public testing::Test {
             .value());
   }
 
+  template <typename DurationType> void advanceTimeAndLoop(DurationType duration) {
+    time_system_.advanceTimeAsync(duration);
+    dispatcher_->run(Event::Dispatcher::RunType::Block);
+  }
+
   Event::SimulatedTimeSystem time_system_;
   Stats::TestUtil::TestStore stats_;
   NiceMock<Runtime::MockLoader> runtime_;
@@ -258,7 +263,7 @@ TEST_F(GradientControllerTest, MinRTTEpoch) {
   const int min_concurrency = 2;
   auto controller = makeController(yaml);
   const auto min_rtt = std::chrono::milliseconds(1350);
-  time_system_.advanceTimeAsync(min_rtt);
+  advanceTimeAndLoop(min_rtt);
 
   verifyMinRTTActive();
   EXPECT_EQ(controller->concurrencyLimit(), min_concurrency);
@@ -270,7 +275,7 @@ TEST_F(GradientControllerTest, MinRTTEpoch) {
   uint32_t last_limit = controller->concurrencyLimit();
   for (int i = 0; i < 29; ++i) {
     tryForward(controller, true);
-    time_system_.advanceTimeAsync(std::chrono::seconds(1));
+    advanceTimeAndLoop(std::chrono::seconds(1));
     sampleLatency(controller, min_rtt);
     dispatcher_->run(Event::Dispatcher::RunType::Block);
     EXPECT_GT(controller->concurrencyLimit(), last_limit);
@@ -286,8 +291,7 @@ TEST_F(GradientControllerTest, MinRTTEpoch) {
   }
 
   // Move into the next minRTT window while the requests are outstanding.
-  time_system_.advanceTimeAsync(std::chrono::seconds(5));
-  dispatcher_->run(Event::Dispatcher::RunType::Block);
+  advanceTimeAndLoop(std::chrono::seconds(5));
   verifyMinRTTActive();
   EXPECT_EQ(controller->concurrencyLimit(), min_concurrency);
 
@@ -330,7 +334,7 @@ TEST_F(GradientControllerTest, MinRTTLogicTest) {
   }
   tryForward(controller, false);
   tryForward(controller, false);
-  time_system_.advanceTimeAsync(min_rtt);
+  advanceTimeAndLoop(min_rtt);
   for (int i = 0; i < 7; ++i) {
     sampleLatency(controller, min_rtt);
   }
@@ -427,8 +431,7 @@ TEST_F(GradientControllerTest, MinRTTBufferTest) {
       // prevent the concurrency limit from decreasing.
       sampleLatency(controller, std::chrono::milliseconds(6));
     }
-    time_system_.advanceTimeAsync(std::chrono::milliseconds(101));
-    dispatcher_->run(Event::Dispatcher::RunType::Block);
+    advanceTimeAndLoop(std::chrono::milliseconds(101));
     EXPECT_GT(controller->concurrencyLimit(), last_concurrency);
   }
 }
@@ -459,8 +462,7 @@ TEST_F(GradientControllerTest, ConcurrencyLimitBehaviorTestBasic) {
 
   // Ensure that the concurrency window increases on its own due to the headroom calculation with
   // the max gradient.
-  time_system_.advanceTimeAsync(std::chrono::milliseconds(101));
-  dispatcher_->run(Event::Dispatcher::RunType::Block);
+  advanceTimeAndLoop(std::chrono::milliseconds(101));
   EXPECT_GE(controller->concurrencyLimit(), 7);
   EXPECT_LE(controller->concurrencyLimit() / 7.0, 2.0);
 
@@ -472,8 +474,7 @@ TEST_F(GradientControllerTest, ConcurrencyLimitBehaviorTestBasic) {
       tryForward(controller, true);
       sampleLatency(controller, std::chrono::milliseconds(4));
     }
-    time_system_.advanceTimeAsync(std::chrono::milliseconds(101));
-    dispatcher_->run(Event::Dispatcher::RunType::Block);
+    advanceTimeAndLoop(std::chrono::milliseconds(101));
     // Verify the minimum gradient.
     EXPECT_LE(last_concurrency, controller->concurrencyLimit());
     EXPECT_GE(static_cast<double>(last_concurrency) / controller->concurrencyLimit(), 0.5);
@@ -486,8 +487,7 @@ TEST_F(GradientControllerTest, ConcurrencyLimitBehaviorTestBasic) {
       tryForward(controller, true);
       sampleLatency(controller, std::chrono::milliseconds(6));
     }
-    time_system_.advanceTimeAsync(std::chrono::milliseconds(101));
-    dispatcher_->run(Event::Dispatcher::RunType::Block);
+    advanceTimeAndLoop(std::chrono::milliseconds(101));
     EXPECT_LT(controller->concurrencyLimit(), last_concurrency);
     EXPECT_GE(controller->concurrencyLimit(), 7);
   }
@@ -513,7 +513,7 @@ TEST_F(GradientControllerTest, MinRTTReturnToPreviousLimit) {
   // Get initial minRTT measurement out of the way and advance time so request samples are not
   // thought to come from the previous minRTT epoch.
   advancePastMinRTTStage(controller, yaml, std::chrono::milliseconds(5));
-  time_system_.advanceTimeAsync(std::chrono::seconds(1));
+  advanceTimeAndLoop(std::chrono::seconds(1));
 
   // Force the limit calculation to run a few times from some measurements.
   for (int sample_iters = 0; sample_iters < 5; ++sample_iters) {
@@ -522,21 +522,19 @@ TEST_F(GradientControllerTest, MinRTTReturnToPreviousLimit) {
       tryForward(controller, true);
       sampleLatency(controller, std::chrono::milliseconds(4));
     }
-    time_system_.advanceTimeAsync(std::chrono::milliseconds(101));
-    dispatcher_->run(Event::Dispatcher::RunType::Block);
+    advanceTimeAndLoop(std::chrono::milliseconds(101));
     // Verify the value is growing.
     EXPECT_GT(controller->concurrencyLimit(), last_concurrency);
   }
 
   const auto limit_val = controller->concurrencyLimit();
 
   // Wait until the minRTT recalculation is triggered again and verify the limit drops.
-  time_system_.advanceTimeAsync(std::chrono::seconds(31));
-  dispatcher_->run(Event::Dispatcher::RunType::Block);
+  advanceTimeAndLoop(std::chrono::seconds(31));
   EXPECT_EQ(controller->concurrencyLimit(), 3);
 
   // Advance time again for request samples to appear from the current epoch.
-  time_system_.advanceTimeAsync(std::chrono::seconds(1));
+  advanceTimeAndLoop(std::chrono::seconds(1));
 
   // 49 more requests should cause the minRTT to be done calculating.
   for (int i = 0; i < 5; ++i) {
@@ -569,7 +567,7 @@ TEST_F(GradientControllerTest, MinRTTRescheduleTest) {
   // Get initial minRTT measurement out of the way and advance time so request samples are not
   // thought to come from the previous minRTT epoch.
   advancePastMinRTTStage(controller, yaml, std::chrono::milliseconds(5));
-  time_system_.advanceTimeAsync(std::chrono::seconds(1));
+  advanceTimeAndLoop(std::chrono::seconds(1));
 
   // Force the limit calculation to run a few times from some measurements.
   for (int sample_iters = 0; sample_iters < 5; ++sample_iters) {
@@ -578,20 +576,17 @@ TEST_F(GradientControllerTest, MinRTTRescheduleTest) {
       tryForward(controller, true);
       sampleLatency(controller, std::chrono::milliseconds(4));
     }
-    time_system_.advanceTimeAsync(std::chrono::milliseconds(101));
-    dispatcher_->run(Event::Dispatcher::RunType::Block);
+    advanceTimeAndLoop(std::chrono::milliseconds(101));
     // Verify the value is growing.
     EXPECT_GT(controller->concurrencyLimit(), last_concurrency);
   }
 
   // Wait until the minRTT recalculation is triggered again and verify the limit drops.
-  time_system_.advanceTimeAsync(std::chrono::seconds(31));
-  dispatcher_->run(Event::Dispatcher::RunType::Block);
+  advanceTimeAndLoop(std::chrono::seconds(31));
   EXPECT_EQ(controller->concurrencyLimit(), 3);
 
   // Verify sample recalculation doesn't occur during the minRTT window.
-  time_system_.advanceTimeAsync(std::chrono::milliseconds(101));
-  dispatcher_->run(Event::Dispatcher::RunType::Block);
+  advanceTimeAndLoop(std::chrono::milliseconds(101));
   EXPECT_EQ(controller->concurrencyLimit(), 3);
 }
 
@@ -622,17 +617,15 @@ TEST_F(GradientControllerTest, NoSamplesTest) {
       tryForward(controller, true);
       sampleLatency(controller, std::chrono::milliseconds(4));
     }
-    time_system_.advanceTimeAsync(std::chrono::milliseconds(101));
-    dispatcher_->run(Event::Dispatcher::RunType::Block);
+    advanceTimeAndLoop(std::chrono::milliseconds(101));
     // Verify the value is growing.
     EXPECT_GT(controller->concurrencyLimit(), last_concurrency);
   }
 
   // Now we make sure that the limit value doesn't change in the absence of samples.
   for (int sample_iters = 0; sample_iters < 5; ++sample_iters) {
     const auto old_limit = controller->concurrencyLimit();
-    time_system_.advanceTimeAsync(std::chrono::milliseconds(101));
-    dispatcher_->run(Event::Dispatcher::RunType::Block);
+    advanceTimeAndLoop(std::chrono::milliseconds(101));
     EXPECT_EQ(old_limit, controller->concurrencyLimit());
   }
 }
@@ -676,7 +669,7 @@ TEST_F(GradientControllerTest, TimerAccuracyTest) {
   EXPECT_CALL(*sample_timer, enableTimer(std::chrono::milliseconds(123), _));
   for (int i = 0; i < 6; ++i) {
     tryForward(controller, true);
-    time_system_.advanceTimeAsync(std::chrono::milliseconds(5));
+    advanceTimeAndLoop(std::chrono::milliseconds(5));
     sampleLatency(controller, std::chrono::milliseconds(5));
   }
 }
@@ -716,7 +709,7 @@ TEST_F(GradientControllerTest, TimerAccuracyTestNoJitter) {
   EXPECT_CALL(*sample_timer, enableTimer(std::chrono::milliseconds(123), _));
   for (int i = 0; i < 6; ++i) {
     tryForward(controller, true);
-    time_system_.advanceTimeAsync(std::chrono::milliseconds(5));
+    advanceTimeAndLoop(std::chrono::milliseconds(5));
     sampleLatency(controller, std::chrono::milliseconds(5));
   }
 }
@@ -749,8 +742,7 @@ TEST_F(GradientControllerTest, ConsecutiveMinConcurrencyReset) {
 
   // Ensure that the concurrency window increases on its own due to the headroom calculation with
   // the max gradient.
-  time_system_.advanceTimeAsync(std::chrono::milliseconds(101));
-  dispatcher_->run(Event::Dispatcher::RunType::Block);
+  advanceTimeAndLoop(std::chrono::milliseconds(101));
   EXPECT_GE(controller->concurrencyLimit(), 7);
   EXPECT_LE(controller->concurrencyLimit() / 7.0, 2.0);
 
@@ -762,8 +754,7 @@ TEST_F(GradientControllerTest, ConsecutiveMinConcurrencyReset) {
       tryForward(controller, true);
       sampleLatency(controller, elevated_latency);
     }
-    time_system_.advanceTimeAsync(std::chrono::milliseconds(101));
-    dispatcher_->run(Event::Dispatcher::RunType::Block);
+    advanceTimeAndLoop(std::chrono::milliseconds(101));
   }
 
   // Verify that the concurrency limit starts growing with newly measured minRTT.
@@ -773,8 +764,7 @@ TEST_F(GradientControllerTest, ConsecutiveMinConcurrencyReset) {
       tryForward(controller, true);
       sampleLatency(controller, elevated_latency);
     }
-    time_system_.advanceTimeAsync(std::chrono::milliseconds(101));
-    dispatcher_->run(Event::Dispatcher::RunType::Block);
+    advanceTimeAndLoop(std::chrono::milliseconds(101));
     EXPECT_GE(controller->concurrencyLimit(), last_concurrency);
   }
 }

test/extensions/quic_listeners/quiche/active_quic_listener_test.cc

@@ -136,7 +136,8 @@ class ActiveQuicListenerTest : public QuicMultiVersionTest {
     auto proof_source = std::make_unique<TestProofSource>();
     filter_chain_ = &proof_source->filterChain();
     crypto_config_peer.ResetProofSource(std::move(proof_source));
-    simulated_time_system_.advanceTimeWait(std::chrono::milliseconds(100));
+    simulated_time_system_.advanceTimeAsync(std::chrono::milliseconds(100));
+    dispatcher_->run(Event::Dispatcher::RunType::NonBlock);
   }
 
   Network::ActiveUdpListenerFactoryPtr createQuicListenerFactory(const std::string& yaml) {

test/extensions/quic_listeners/quiche/envoy_quic_dispatcher_test.cc

@@ -93,7 +93,8 @@ class EnvoyQuicDispatcherTest : public QuicMultiVersionTest,
 
   void SetUp() override {
     // Advance time a bit because QuicTime regards 0 as uninitialized timestamp.
-    time_system_.advanceTimeWait(std::chrono::milliseconds(100));
+    time_system_.advanceTimeAsync(std::chrono::milliseconds(100));
+    dispatcher_->run(Event::Dispatcher::RunType::NonBlock);
     EXPECT_CALL(listener_config_, perConnectionBufferLimitBytes())
         .WillRepeatedly(Return(1024 * 1024));
   }

test/extensions/quic_listeners/quiche/envoy_quic_server_session_test.cc

@@ -669,7 +669,8 @@ TEST_P(EnvoyQuicServerSessionTest, FlushAndWaitForCloseWithTimeout) {
   EXPECT_EQ(Network::Connection::State::Open, envoy_quic_session_.state());
   // Unblocking the stream shouldn't close the connection as it should be
   // delayed.
-  time_system_.advanceTimeWait(std::chrono::milliseconds(10));
+  time_system_.advanceTimeAsync(std::chrono::milliseconds(10));
+  dispatcher_->run(Event::Dispatcher::RunType::NonBlock);
   envoy_quic_session_.OnCanWrite();
   // delay close alarm should have been rescheduled.
   time_system_.advanceTimeAsync(std::chrono::milliseconds(90));
@@ -700,7 +701,9 @@ TEST_P(EnvoyQuicServerSessionTest, FlusWriteTransitToFlushWriteWithDelay) {
   envoy_quic_session_.close(Network::ConnectionCloseType::FlushWrite);
   EXPECT_EQ(Network::Connection::State::Open, envoy_quic_session_.state());
 
-  time_system_.advanceTimeWait(std::chrono::milliseconds(10));
+  time_system_.advanceTimeAsync(std::chrono::milliseconds(10));
+  dispatcher_->run(Event::Dispatcher::RunType::NonBlock);
+
   // The closing behavior should be changed.
   envoy_quic_session_.close(Network::ConnectionCloseType::FlushWriteAndDelay);
   // Unblocking the stream shouldn't close the connection as it should be
@@ -732,7 +735,8 @@ TEST_P(EnvoyQuicServerSessionTest, FlushAndWaitForCloseWithNoPendingData) {
 
   // Advance the time a bit and try to close again. The delay close timer
   // shouldn't be rescheduled by this call.
-  time_system_.advanceTimeWait(std::chrono::milliseconds(10));
+  time_system_.advanceTimeAsync(std::chrono::milliseconds(10));
+  dispatcher_->run(Event::Dispatcher::RunType::NonBlock);
   envoy_quic_session_.close(Network::ConnectionCloseType::FlushWriteAndDelay);
   EXPECT_EQ(Network::Connection::State::Open, envoy_quic_session_.state());