moved content of test_cubic to test_recovery_cubic

Aperence · Dec 14, 2023 · fdffd13 · fdffd13
1 parent 38be138
commit fdffd13
Show file tree

Hide file tree

Showing 3 changed files with 53 additions and 68 deletions.
diff --git a/bob.bin b/bob.bin
diff --git a/tests/test_cubic.py b/tests/test_cubic.py
diff --git a/tests/test_recovery_cubic.py b/tests/test_recovery_cubic.py
@@ -19,6 +19,10 @@ def send_probe():
     pass
 
 
+def W_cubic(t, K, W_max):
+    return K_CUBIC_C * (t - K) ** 3 + (W_max)
+
+
 class QuicPacketRecoveryCubicTest(TestCase):
     def setUp(self):
         self.INITIAL_SPACE = QuicPacketSpace()
@@ -186,6 +190,55 @@ def test_log_data(self):
             },
         )
 
+    def test_congestion_avoidance(self):
+        """
+        Check if the cubic implementation respects the mathematical
+        formula defined in the rfc 9438
+        """
+
+        max_datagram_size = 1440
+
+        n = 400  # number of ms to check
+
+        W_max = 5  # starting W_max
+        K = better_cube_root(W_max * (1 - K_CUBIC_LOSS_REDUCTION_FACTOR) / K_CUBIC_C)
+        cwnd = W_max * K_CUBIC_LOSS_REDUCTION_FACTOR
+
+        correct = []
+
+        test_range = range(n)
+
+        for i in test_range:
+            correct.append(W_cubic(i / 1000, K, W_max) * max_datagram_size)
+
+        cubic = CubicCongestionControl(max_datagram_size)
+        cubic.rtt = 0
+        cubic._W_max = W_max * max_datagram_size
+        cubic._starting_congestion_avoidance = True
+        cubic.congestion_window = cwnd * max_datagram_size
+        cubic.ssthresh = cubic.congestion_window
+        cubic._W_est = 0
+
+        results = []
+        for i in test_range:
+            cwnd = cubic.congestion_window // max_datagram_size  # number of segments
+
+            # simulate the reception of cwnd packets (a full window of acks)
+            for _ in range(int(cwnd)):
+                packet = QuicSentPacket(None, True, True, True, 0, 0)
+                packet.sent_bytes = 0  # won't affect results
+
+                cubic.on_packet_acked(packet=packet, now=(i / 1000))
+
+            results.append(cubic.congestion_window)
+
+        for i in test_range:
+            # check if it is almost equal to the value of W_cubic
+            self.assertTrue(
+                correct[i] * 0.99 <= results[i] <= 1.01 * correct[i],
+                f"Error at {i}ms, Result={results[i]}, Expected={correct[i]}",
+            )
+
     def test_reset_idle(self):
         packet = QuicSentPacket(
             epoch=tls.Epoch.ONE_RTT,