diff --git a/tests/all.py b/tests/all.py index ea751954..17332612 100644 --- a/tests/all.py +++ b/tests/all.py @@ -9,5 +9,6 @@ from .hkdf import TestHKDF from .ifac import TestIFAC from .coalesced_transmit import TestTransmitBuffer from .hdlc import TestReceiveBuffer +from .egress import TestHWMLimiter if __name__ == '__main__': unittest.main(verbosity=2) \ No newline at end of file diff --git a/tests/egress.py b/tests/egress.py new file mode 100644 index 00000000..5d0adecc --- /dev/null +++ b/tests/egress.py @@ -0,0 +1,204 @@ +import unittest + +import time + +from RNS.Interfaces.util.TransmitBuffer import TransmitBuffer +from RNS.Interfaces.BackboneInterface import BackboneInterface, BackboneClientInterface + +def framed(payload): return b"\x7e" + payload + b"\x7e" + +class FakeSock: + def send(self, data): return len(data) + def fileno(self): return 999999 + def close(self): return + +class FakeIF: + def __init__(self, tb=None): + self.transmit_buffer = tb if tb is not None else TransmitBuffer() + self._dp_ec_prev_sent = 0 + self._dp_ec_zero_ticks = 0 + self._dp_ec_last_drain = 10.0 + self.tx_stalled = False + self.detached = False + self.socket = FakeSock() + self.torn_down = False + + def receive(self, data): self.torn_down = True + +class TestHWMLimiter(unittest.TestCase): + + def test_01_hard_valve_bounds_memory(self): + print("") + tb = TransmitBuffer() + limit = 4096 + accepted_count = 0 + total_accepted = 0 + rejected = 0 + for _ in range(200): + for size in (500, 700): + frame = framed(b"x" * size) + accepted = tb.append(frame, limit) + if accepted: + accepted_count += 1 + total_accepted += len(frame) + self.assertLessEqual(len(tb), limit) + else: + rejected += 1 + # Rejection implies the frame would breach the limit. + self.assertGreater(total_accepted + len(frame), limit) + + # The limit must never be exceeded, some frames must have been + # rejected, and the buffer must still be fully drainable. + self.assertGreater(rejected, 0) + self.assertLessEqual(len(tb), limit) + drained = tb.drain_to(FakeSock()) + self.assertEqual(drained, total_accepted) + self.assertEqual(len(tb), 0) + + def test_02_oversized_frame_rejected(self): + print("") + tb = TransmitBuffer() + limit = 2048 + self.assertTrue(tb.append(framed(b"x" * 500), limit)) + # A single frame that would exceed the limit is rejected outright, + # even against an empty buffer. + self.assertFalse(tb.append(framed(b"x" * 3000), limit)) + self.assertEqual(len(tb), len(framed(b"x" * 500))) + # The buffer remains fully functional after rejection. + self.assertTrue(tb.append(framed(b"x" * 1000), limit)) + self.assertEqual(len(tb), len(framed(b"x" * 500)) + len(framed(b"x" * 1000))) + drained = tb.drain_to(FakeSock()) + self.assertEqual(drained, len(framed(b"x" * 500)) + len(framed(b"x" * 1000))) + + def test_03_valve_accepts_after_drain(self): + print("") + tb = TransmitBuffer() + limit = 2048 + for _ in range(10): self.assertTrue(tb.append(framed(b"x" * 200), limit)) + self.assertEqual(len(tb), len(framed(b"x" * 200)) * 10) + # Drain most of the buffer; appends are accepted again up to the + # limit, and the total stays bounded. + tb.drain_to(FakeSock()) + for _ in range(10): self.assertTrue(tb.append(framed(b"x" * 200), limit)) + self.assertLessEqual(len(tb), limit) + + def test_04_append_without_limit_unbounded(self): + print("") + tb = TransmitBuffer() + for _ in range(100): self.assertTrue(tb.append(framed(b"x" * 200))) + self.assertEqual(len(tb), len(framed(b"x" * 200)) * 100) + + def test_05_stall_engages_on_zero_drain(self): + print("") + tb = TransmitBuffer() + iface = FakeIF(tb) + # Fill beyond the mid watermark. + while len(tb) <= BackboneInterface.DP_EC_MID_WM: tb.append(framed(b"y" * 4096)) + # No drain progress, fewer than STALL_TICKS evaluations must not stall. + for tick in range(BackboneInterface.DP_EC_STALL_TICKS - 1): + BackboneInterface._dp_ec_evaluate(iface, 11.0 + tick) + self.assertFalse(iface.tx_stalled) + # The STALL_TICKSth zero-drain tick at/above the + # mid watermark engages the gate. + BackboneInterface._dp_ec_evaluate(iface, 15.0) + self.assertTrue(iface.tx_stalled) + + def test_06_stall_releases_below_mid(self): + print("") + tb = TransmitBuffer() + iface = FakeIF(tb) + while len(tb) <= BackboneInterface.DP_EC_MID_WM: tb.append(framed(b"y" * 4096)) + for t in range(BackboneInterface.DP_EC_STALL_TICKS + 1): BackboneInterface._dp_ec_evaluate(iface, 11.0 + t) + self.assertTrue(iface.tx_stalled) + # Drain the buffer completely; the next tick releases the gate. + tb.drain_to(FakeSock()) + self.assertEqual(len(tb), 0) + BackboneInterface._dp_ec_evaluate(iface, 20.0) + self.assertFalse(iface.tx_stalled) + self.assertEqual(iface._dp_ec_zero_ticks, 0) + + def test_07_stall_engages_on_drain_eta(self): + print("") + tb = TransmitBuffer() + iface = FakeIF(tb) + while len(tb) <= BackboneInterface.DP_EC_MID_WM: tb.append(framed(b"y" * 4096)) + # Pretend 512 bytes drained since the last tick, the queue would + # take far longer than DP_EC_MAX_ETA to clear at that rate. + iface._dp_ec_prev_sent = tb._tx_sent - 512 + BackboneInterface._dp_ec_evaluate(iface, 11.0) + self.assertTrue(iface.tx_stalled) + + def test_08_stall_hysteresis_no_flap(self): + print("") + tb = TransmitBuffer() + iface = FakeIF(tb) + while len(tb) <= BackboneInterface.DP_EC_MID_WM: tb.append(framed(b"y" * 4096)) + + # Engage the gate with an ETA above max. + buffered = len(tb) + iface._dp_ec_prev_sent = tb._tx_sent - 128 + BackboneInterface._dp_ec_evaluate(iface, 11.0) + self.assertTrue(iface.tx_stalled) + + # A drain ETA between RELEASE and MAX is the hysteresis band, + # the gate must stay engaged without flapping. + mid_eta = (BackboneInterface.DP_EC_MAX_ETA + BackboneInterface.DP_EC_RELEASE_ETA) / 2 + mid_drain = buffered / mid_eta + iface._dp_ec_prev_sent = tb._tx_sent - mid_drain + BackboneInterface._dp_ec_evaluate(iface, 12.0) + self.assertTrue(iface.tx_stalled) + + # Only when drainage clears the queue well within the release + # ETA is the gate released. + fast_drain = buffered / (BackboneInterface.DP_EC_RELEASE_ETA / 2) + iface._dp_ec_prev_sent = tb._tx_sent - fast_drain + BackboneInterface._dp_ec_evaluate(iface, 13.0) + self.assertFalse(iface.tx_stalled) + + def test_09_sendable_zero_no_stall(self): + print("") + tb = TransmitBuffer() + iface = FakeIF(tb) + while len(tb) <= BackboneInterface.DP_EC_MID_WM: tb.append(framed(b"y" * 4096)) + # A tick with nothing sendable. All data still unsent but none in + # the drainer's visible set is impossible with a full buffer, so + # simulate the empty state instead. An empty buffer must reset + # stall state and the dead-drain clock. + tb.drain_to(FakeSock()) + BackboneInterface._dp_ec_evaluate(iface, 11.0) + self.assertFalse(iface.tx_stalled) + self.assertEqual(iface._dp_ec_last_drain, 11.0) + + def test_10_dead_peer_escalation(self): + print("") + tb = TransmitBuffer() + iface = FakeIF(tb) + while len(tb) <= BackboneInterface.DP_EC_MID_WM: tb.append(framed(b"y" * 4096)) + # Last drain progress a long time ago, no drain since. + iface._dp_ec_last_drain = 10.0 + iface._dp_ec_zero_ticks = BackboneInterface.DP_EC_STALL_TICKS + escalated = BackboneInterface._dp_ec_evaluate(iface, 100.0) + self.assertTrue(escalated) + self.assertTrue(iface.torn_down) + + def test_11_process_outgoing_gate(self): + print("") + # Construct a shell client interface object without full + # initialisation. Process_outgoing only requires the + # egress-control state and the transmit buffer gating. + iface = object.__new__(BackboneClientInterface) + iface.online = True + iface.detached = False + iface.tx_stalled = True + iface.tx_hwm = 1024 + iface.tx_drops = 0 + iface.tx_dropped_bytes = 0 + iface.transmit_buffer = TransmitBuffer() + iface.socket = None + + iface.process_outgoing(b"gated-frame") + self.assertEqual(iface.tx_drops, 1) + self.assertEqual(iface.tx_dropped_bytes, len(framed(b"gated-frame"))) + self.assertEqual(len(iface.transmit_buffer), 0) + +if __name__ == "__main__": unittest.main(verbosity=2) \ No newline at end of file