95 lines
3 KiB
Python
95 lines
3 KiB
Python
"""Unit test for electrum-0001: receive_history_callback list membership O(N^2).
|
|
|
|
The defect is in AddressSynchronizer.receive_history_callback where
|
|
`(tx_hash, height) not in hist` performs O(N) linear scan on a list
|
|
for each entry in old_hist, creating O(old_hist * hist) = O(N^2).
|
|
|
|
Fix: convert hist to a set before the loop for O(1) membership checks.
|
|
"""
|
|
|
|
import time
|
|
|
|
|
|
def _receive_history_callback_BEFORE(old_hist, hist):
|
|
"""Original: linear scan on list for each old entry."""
|
|
removed = []
|
|
for tx_hash, height in old_hist.items():
|
|
if (tx_hash, height) not in hist: # O(N) scan on list
|
|
removed.append(tx_hash)
|
|
return removed
|
|
|
|
|
|
def _receive_history_callback_AFTER(old_hist, hist):
|
|
"""Fixed: convert to set for O(1) membership checks."""
|
|
hist_set = set(hist)
|
|
removed = []
|
|
for tx_hash, height in old_hist.items():
|
|
if (tx_hash, height) not in hist_set: # O(1) lookup in set
|
|
removed.append(tx_hash)
|
|
return removed
|
|
|
|
|
|
def make_test_data(n):
|
|
"""Create test data simulating address history.
|
|
|
|
old_hist: dict of txid -> height (what wallet has)
|
|
hist: list of (txid, height) tuples (what server says)
|
|
We make half the entries match and half not match.
|
|
"""
|
|
old_hist = {}
|
|
hist = []
|
|
for i in range(n):
|
|
txid = f"{'%064x' % i}"
|
|
height = 700000 + i
|
|
old_hist[txid] = height
|
|
# server history includes only even-numbered entries
|
|
if i % 2 == 0:
|
|
hist.append((txid, height))
|
|
return old_hist, hist
|
|
|
|
|
|
def test_correctness():
|
|
"""Both implementations must return the same result."""
|
|
old_hist, hist = make_test_data(200)
|
|
removed_before = sorted(_receive_history_callback_BEFORE(old_hist, hist))
|
|
removed_after = sorted(_receive_history_callback_AFTER(old_hist, hist))
|
|
assert removed_before == removed_after, "Results differ!"
|
|
# Half should be removed (odd-numbered entries)
|
|
assert len(removed_before) == 100
|
|
print("PASS: correctness")
|
|
|
|
|
|
def test_performance():
|
|
"""Measure O(N^2) vs O(N) performance."""
|
|
N = 2000
|
|
old_hist, hist = make_test_data(N)
|
|
|
|
# Warm up
|
|
_receive_history_callback_BEFORE(old_hist, hist)
|
|
_receive_history_callback_AFTER(old_hist, hist)
|
|
|
|
# Benchmark BEFORE (O(N^2))
|
|
t0 = time.perf_counter()
|
|
iterations = 5
|
|
for _ in range(iterations):
|
|
_receive_history_callback_BEFORE(old_hist, hist)
|
|
t_before = (time.perf_counter() - t0) / iterations
|
|
|
|
# Benchmark AFTER (O(N))
|
|
t0 = time.perf_counter()
|
|
for _ in range(iterations):
|
|
_receive_history_callback_AFTER(old_hist, hist)
|
|
t_after = (time.perf_counter() - t0) / iterations
|
|
|
|
ratio = t_before / t_after if t_after > 0 else float('inf')
|
|
print(f"N={N}")
|
|
print(f" BEFORE: {t_before*1000:.3f} ms")
|
|
print(f" AFTER: {t_after*1000:.3f} ms")
|
|
print(f" Ratio: {ratio:.1f}x")
|
|
assert ratio > 2.0, f"Expected at least 2x speedup, got {ratio:.1f}x"
|
|
print("PASS: performance")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
test_correctness()
|
|
test_performance()
|