Incorporate throttling signal into BBRv1: limit the writable bytes by the available tokens

Summary:
This limits the writableBytes by the available tokens (ie `ThrottlingSignalProvider::bytesToSend`), if the connection is being throttled and that value is provided by the throttling signal provider.

Note that these throttling signals to the transport are provided only for the connections that belong to a specific QE experiment.

Reviewed By: silver23arrow

Differential Revision: D47850261

fbshipit-source-id: 8c52e66198db6d1fee252cacea69b82963a1601a

quic/api/QuicTransportFunctions.cpp

@@ -969,6 +969,19 @@ uint64_t congestionControlWritableBytes(QuicConnectionStateBase& conn) {
   if (conn.congestionController) {
     writableBytes = std::min<uint64_t>(
         writableBytes, conn.congestionController->getWritableBytes());
+
+    if (conn.throttlingSignalProvider &&
+        conn.throttlingSignalProvider->getCurrentThrottlingSignal()
+            .has_value()) {
+      const auto& throttlingSignal =
+          conn.throttlingSignalProvider->getCurrentThrottlingSignal();
+      if (throttlingSignal.value().maybeBytesToSend.has_value()) {
+        // Cap the writable bytes by the amount of tokens available in the
+        // throttler's bucket if one found to be throttling the connection.
+        writableBytes = std::min(
+            throttlingSignal.value().maybeBytesToSend.value(), writableBytes);
+      }
+    }
   }
 
   if (writableBytes == std::numeric_limits<uint64_t>::max()) {

quic/api/test/QuicTransportFunctionsTest.cpp

@@ -4761,5 +4761,36 @@ TEST_F(QuicTransportFunctionsTest, CustomTransportParamTest) {
   EXPECT_EQ(customTransportParameters.size(), 2);
 }
 
+TEST_F(
+    QuicTransportFunctionsTest,
+    StaticCapOnWritableBytesFromThrottlingSignalProvider) {
+  auto conn = createConn();
+  conn->udpSendPacketLen = 2000;
+  auto mockCongestionController =
+      std::make_unique<NiceMock<MockCongestionController>>();
+  auto rawCongestionController = mockCongestionController.get();
+  conn->congestionController = std::move(mockCongestionController);
+
+  auto mockThrottlingSignalProvider =
+      std::make_shared<MockThrottlingSignalProvider>();
+  ThrottlingSignalProvider::ThrottlingSignal expectedSignal;
+  expectedSignal.state =
+      ThrottlingSignalProvider::ThrottlingSignal::State::Throttled;
+  expectedSignal.maybeBytesToSend = 16000;
+  mockThrottlingSignalProvider->useFakeThrottlingSignal(expectedSignal);
+  conn->throttlingSignalProvider = mockThrottlingSignalProvider;
+
+  EXPECT_CALL(*rawCongestionController, getWritableBytes())
+      .WillOnce(Return(10000));
+  EXPECT_EQ(10000, congestionControlWritableBytes(*conn));
+
+  // Since cwnd is larger than available tokens, the writable bytes is capped by
+  // the available tokens
+  EXPECT_CALL(*rawCongestionController, getWritableBytes())
+      .WillOnce(Return(20000));
+  EXPECT_EQ(
+      expectedSignal.maybeBytesToSend, congestionControlWritableBytes(*conn));
+}
+
 } // namespace test
 } // namespace quic

quic/congestion_control/test/ThrottlingSignalProviderTest.cpp

@@ -6,6 +6,9 @@
  */
 
 #include <gtest/gtest.h>
+#include <quic/api/QuicTransportFunctions.h>
+#include <quic/common/test/TestUtils.h>
+#include <quic/congestion_control/SimulatedTBF.h>
 #include <quic/congestion_control/ThrottlingSignalProvider.h>
 #include <quic/state/test/Mocks.h>
 
@@ -13,6 +16,35 @@ using namespace ::testing;
 
 namespace quic::test {
 
+// A simple implementation of throttling signal provider, which has a single
+// static SimTBF that consumes the packets on send.
+class SimpleThrottlingSignalProvider : public PacketProcessor,
+                                       public ThrottlingSignalProvider {
+ public:
+  explicit SimpleThrottlingSignalProvider(SimulatedTBF::Config config)
+      : stbf_(std::move(config)) {}
+  ~SimpleThrottlingSignalProvider() override = default;
+  void onPacketSent(const quic::OutstandingPacketWrapper& packet) override {
+    stbf_.consumeWithBorrowNonBlockingAndUpdateState(
+        packet.metadata.encodedSize, packet.metadata.time);
+  }
+
+  [[nodiscard]] folly::Optional<ThrottlingSignal> getCurrentThrottlingSignal()
+      override {
+    auto availTokens = stbf_.getNumAvailableTokensInBytes(quic::Clock::now());
+    ThrottlingSignal signal = {};
+    signal.state = availTokens > 0 ? ThrottlingSignal::State::Unthrottled
+                                   : ThrottlingSignal::State::Throttled;
+    signal.maybeBytesToSend.assign((uint64_t)availTokens);
+    signal.maybeThrottledRateBytesPerSecond.assign(
+        (uint64_t)stbf_.getRateBytesPerSecond());
+    return signal;
+  }
+
+ private:
+  SimulatedTBF stbf_;
+};
+
 TEST(ThrottlingSignalProviderTest, BasicInitSetGetTest) {
   auto mockThrottlingSignalProvider =
       std::make_shared<MockThrottlingSignalProvider>();
@@ -38,4 +70,51 @@ TEST(ThrottlingSignalProviderTest, BasicInitSetGetTest) {
       expectedSignal.maybeThrottledRateBytesPerSecond);
   EXPECT_FALSE(signal.maybeUnthrottledRateBytesPerSecond.has_value());
 }
+
+TEST(ThrottlingSignalProviderTest, TokenBasedDynamicCapOnWritableBytes) {
+  QuicConnectionStateBase conn(QuicNodeType::Server);
+  conn.udpSendPacketLen = 2000;
+  auto signalProvider =
+      std::make_shared<SimpleThrottlingSignalProvider>(SimulatedTBF::Config{
+          .rateBytesPerSecond = 100 * 1000,
+          .burstSizeBytes = 500 * 1000,
+          .maybeMaxDebtQueueSizeBytes = 50 * 1000,
+          .trackEmptyIntervals = false});
+  conn.throttlingSignalProvider = signalProvider;
+  auto mockCongestionController = std::make_unique<MockCongestionController>();
+  conn.congestionController = std::move(mockCongestionController);
+
+  auto now = Clock::now();
+  uint64_t totalBytesSent = 0;
+  // Send 1000 packets, each with 2000 bytes every 10ms = 200KBps, which is
+  // enough to consume the SimTBF's burst.
+  for (PacketNum pn = 0; pn < 1000; pn++) {
+    auto packet = makeTestingWritePacket(pn, 2000, totalBytesSent, now);
+    signalProvider->onPacketSent(packet);
+    conn.congestionController->onPacketSent(packet);
+    totalBytesSent += 2000;
+
+    // Ack each sent packet after 5ms
+    auto ack = makeAck(
+        pn, 2000, now + std::chrono::milliseconds{5}, packet.metadata.time);
+    conn.congestionController->onPacketAckOrLoss(&ack, nullptr);
+    auto writableBytes = congestionControlWritableBytes(conn);
+    auto maybeSignal = signalProvider->getCurrentThrottlingSignal();
+    ASSERT_TRUE(maybeSignal.has_value());
+    auto maybeBytesToSend = maybeSignal.value().maybeBytesToSend;
+    ASSERT_TRUE(maybeBytesToSend.has_value());
+    // Since congestionControlWritableBytes is rounded to the nearest
+    // multiple of udpSendPacketLen, which is 2000 bytes, do the same for
+    // bytesToSend.
+    auto roundedBytesToSend =
+        (maybeBytesToSend.value() + conn.udpSendPacketLen - 1) /
+        conn.udpSendPacketLen * conn.udpSendPacketLen;
+    // writable bytes must be less than bytes to sends, which is the same as the
+    // number of tokens.
+    EXPECT_GE(roundedBytesToSend, writableBytes);
+
+    now += std::chrono::milliseconds{10};
+  }
+}
+
 } // namespace quic::test