204 lines
8.4 KiB
Python
204 lines
8.4 KiB
Python
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) |