Changeset View
Changeset View
Standalone View
Standalone View
test/functional/p2p_tx_download.py
Show All 34 Lines | def __init__(self): | ||||
self.tx_getdata_count = 0 | self.tx_getdata_count = 0 | ||||
def on_getdata(self, message): | def on_getdata(self, message): | ||||
for i in message.inv: | for i in message.inv: | ||||
if i.type & MSG_TYPE_MASK == MSG_TX: | if i.type & MSG_TYPE_MASK == MSG_TX: | ||||
self.tx_getdata_count += 1 | self.tx_getdata_count += 1 | ||||
# Constants from net_processing | class NetConstants: | ||||
GETDATA_TX_INTERVAL = 60 # seconds | """Constants from net_processing""" | ||||
INBOUND_PEER_TX_DELAY = 2 # seconds | |||||
OVERLOADED_PEER_DELAY = 2 # seconds | def __init__(self, | ||||
MAX_GETDATA_IN_FLIGHT = 100 | getdata_interval, | ||||
MAX_PEER_TX_ANNOUNCEMENTS = 5000 | inbound_peer_delay, | ||||
overloaded_peer_delay, | |||||
max_getdata_in_flight, | |||||
max_peer_announcements, | |||||
): | |||||
self.getdata_interval = getdata_interval | |||||
self.inbound_peer_delay = inbound_peer_delay | |||||
self.overloaded_peer_delay = overloaded_peer_delay | |||||
self.max_getdata_in_flight = max_getdata_in_flight | |||||
self.max_peer_announcements = max_peer_announcements | |||||
self.max_getdata_inbound_wait = self.getdata_interval + self.inbound_peer_delay | |||||
class TestContext: | |||||
def __init__(self, constants): | |||||
self.constants = constants | |||||
TX_TEST_CONTEXT = TestContext( | |||||
NetConstants( | |||||
getdata_interval=60, # seconds | |||||
inbound_peer_delay=2, # seconds | |||||
overloaded_peer_delay=2, # seconds | |||||
max_getdata_in_flight=100, | |||||
max_peer_announcements=5000, | |||||
), | |||||
) | |||||
# Python test constants | # Python test constants | ||||
NUM_INBOUND = 10 | NUM_INBOUND = 10 | ||||
MAX_GETDATA_INBOUND_WAIT = GETDATA_TX_INTERVAL + INBOUND_PEER_TX_DELAY | |||||
class TxDownloadTest(BitcoinTestFramework): | class TxDownloadTest(BitcoinTestFramework): | ||||
def set_test_params(self): | def set_test_params(self): | ||||
self.setup_clean_chain = False | self.setup_clean_chain = False | ||||
self.num_nodes = 2 | self.num_nodes = 2 | ||||
def test_tx_requests(self): | def test_tx_requests(self, context): | ||||
self.log.info( | self.log.info( | ||||
"Test that we request transactions from all our peers, eventually") | "Test that we request transactions from all our peers, eventually") | ||||
txid = 0xdeadbeef | txid = 0xdeadbeef | ||||
self.log.info("Announce the txid from each incoming peer to node 0") | self.log.info("Announce the txid from each incoming peer to node 0") | ||||
msg = msg_inv([CInv(t=MSG_TX, h=txid)]) | msg = msg_inv([CInv(t=MSG_TX, h=txid)]) | ||||
for p in self.nodes[0].p2ps: | for p in self.nodes[0].p2ps: | ||||
p.send_and_ping(msg) | p.send_and_ping(msg) | ||||
outstanding_peer_index = [i for i in range(len(self.nodes[0].p2ps))] | outstanding_peer_index = [i for i in range(len(self.nodes[0].p2ps))] | ||||
def getdata_found(peer_index): | def getdata_found(peer_index): | ||||
p = self.nodes[0].p2ps[peer_index] | p = self.nodes[0].p2ps[peer_index] | ||||
with p2p_lock: | with p2p_lock: | ||||
return p.last_message.get( | return p.last_message.get( | ||||
"getdata") and p.last_message["getdata"].inv[-1].hash == txid | "getdata") and p.last_message["getdata"].inv[-1].hash == txid | ||||
node_0_mocktime = int(time.time()) | node_0_mocktime = int(time.time()) | ||||
while outstanding_peer_index: | while outstanding_peer_index: | ||||
node_0_mocktime += MAX_GETDATA_INBOUND_WAIT | node_0_mocktime += context.constants.max_getdata_inbound_wait | ||||
self.nodes[0].setmocktime(node_0_mocktime) | self.nodes[0].setmocktime(node_0_mocktime) | ||||
wait_until(lambda: any(getdata_found(i) | wait_until(lambda: any(getdata_found(i) | ||||
for i in outstanding_peer_index)) | for i in outstanding_peer_index)) | ||||
for i in outstanding_peer_index: | for i in outstanding_peer_index: | ||||
if getdata_found(i): | if getdata_found(i): | ||||
outstanding_peer_index.remove(i) | outstanding_peer_index.remove(i) | ||||
self.nodes[0].setmocktime(0) | self.nodes[0].setmocktime(0) | ||||
self.log.info("All outstanding peers received a getdata") | self.log.info("All outstanding peers received a getdata") | ||||
def test_inv_block(self): | def test_inv_block(self, context): | ||||
self.log.info("Generate a transaction on node 0") | self.log.info("Generate a transaction on node 0") | ||||
tx = self.nodes[0].createrawtransaction( | tx = self.nodes[0].createrawtransaction( | ||||
inputs=[{ | inputs=[{ | ||||
# coinbase | # coinbase | ||||
"txid": self.nodes[0].getblock(self.nodes[0].getblockhash(1))['tx'][0], | "txid": self.nodes[0].getblock(self.nodes[0].getblockhash(1))['tx'][0], | ||||
"vout": 0 | "vout": 0 | ||||
}], | }], | ||||
outputs={ADDRESS_BCHREG_UNSPENDABLE: 50 - 0.00025}, | outputs={ADDRESS_BCHREG_UNSPENDABLE: 50 - 0.00025}, | ||||
Show All 18 Lines | def test_inv_block(self, context): | ||||
# should get the tx within a timeout. (Assuming that node 0 | # should get the tx within a timeout. (Assuming that node 0 | ||||
# announced the tx within the timeout) | # announced the tx within the timeout) | ||||
# The timeout is the sum of | # The timeout is the sum of | ||||
# * the worst case until the tx is first requested from an inbound | # * the worst case until the tx is first requested from an inbound | ||||
# peer, plus | # peer, plus | ||||
# * the first time it is re-requested from the outbound peer, plus | # * the first time it is re-requested from the outbound peer, plus | ||||
# * 2 seconds to avoid races | # * 2 seconds to avoid races | ||||
assert self.nodes[1].getpeerinfo()[0]['inbound'] is False | assert self.nodes[1].getpeerinfo()[0]['inbound'] is False | ||||
timeout = 2 + INBOUND_PEER_TX_DELAY + GETDATA_TX_INTERVAL | timeout = 2 + context.constants.inbound_peer_delay + \ | ||||
context.constants.getdata_interval | |||||
self.log.info( | self.log.info( | ||||
"Tx should be received at node 1 after {} seconds".format(timeout)) | "Tx should be received at node 1 after {} seconds".format(timeout)) | ||||
self.sync_mempools(timeout=timeout) | self.sync_mempools(timeout=timeout) | ||||
def test_in_flight_max(self): | def test_in_flight_max(self, context): | ||||
max_getdata_in_flight = context.constants.max_getdata_in_flight | |||||
max_inbound_delay = context.constants.inbound_peer_delay + \ | |||||
context.constants.overloaded_peer_delay | |||||
self.log.info("Test that we don't load peers with more than {} transaction requests immediately".format( | self.log.info("Test that we don't load peers with more than {} transaction requests immediately".format( | ||||
MAX_GETDATA_IN_FLIGHT)) | max_getdata_in_flight)) | ||||
txids = [i for i in range(MAX_GETDATA_IN_FLIGHT + 2)] | txids = [i for i in range(max_getdata_in_flight + 2)] | ||||
p = self.nodes[0].p2ps[0] | p = self.nodes[0].p2ps[0] | ||||
with p2p_lock: | with p2p_lock: | ||||
p.tx_getdata_count = 0 | p.tx_getdata_count = 0 | ||||
mock_time = int(time.time() + 1) | mock_time = int(time.time() + 1) | ||||
self.nodes[0].setmocktime(mock_time) | self.nodes[0].setmocktime(mock_time) | ||||
for i in range(MAX_GETDATA_IN_FLIGHT): | for i in range(max_getdata_in_flight): | ||||
p.send_message(msg_inv([CInv(t=MSG_TX, h=txids[i])])) | p.send_message(msg_inv([CInv(t=MSG_TX, h=txids[i])])) | ||||
p.sync_with_ping() | p.sync_with_ping() | ||||
mock_time += INBOUND_PEER_TX_DELAY | mock_time += context.constants.inbound_peer_delay | ||||
self.nodes[0].setmocktime(mock_time) | self.nodes[0].setmocktime(mock_time) | ||||
p.wait_until(lambda: p.tx_getdata_count >= MAX_GETDATA_IN_FLIGHT) | p.wait_until(lambda: p.tx_getdata_count >= max_getdata_in_flight) | ||||
for i in range(MAX_GETDATA_IN_FLIGHT, len(txids)): | for i in range(max_getdata_in_flight, len(txids)): | ||||
p.send_message(msg_inv([CInv(t=MSG_TX, h=txids[i])])) | p.send_message(msg_inv([CInv(t=MSG_TX, h=txids[i])])) | ||||
p.sync_with_ping() | p.sync_with_ping() | ||||
self.log.info( | self.log.info( | ||||
"No more than {} requests should be seen within {} seconds after announcement".format( | "No more than {} requests should be seen within {} seconds after announcement".format( | ||||
MAX_GETDATA_IN_FLIGHT, | max_getdata_in_flight, | ||||
INBOUND_PEER_TX_DELAY + | max_inbound_delay - 1)) | ||||
OVERLOADED_PEER_DELAY - | |||||
1)) | |||||
self.nodes[0].setmocktime( | self.nodes[0].setmocktime( | ||||
mock_time + | mock_time + | ||||
INBOUND_PEER_TX_DELAY + | max_inbound_delay - 1) | ||||
OVERLOADED_PEER_DELAY - | |||||
1) | |||||
p.sync_with_ping() | p.sync_with_ping() | ||||
with p2p_lock: | with p2p_lock: | ||||
assert_equal(p.tx_getdata_count, MAX_GETDATA_IN_FLIGHT) | assert_equal(p.tx_getdata_count, max_getdata_in_flight) | ||||
self.log.info( | self.log.info( | ||||
"If we wait {} seconds after announcement, we should eventually get more requests".format( | "If we wait {} seconds after announcement, we should eventually get more requests".format( | ||||
INBOUND_PEER_TX_DELAY + | max_inbound_delay)) | ||||
OVERLOADED_PEER_DELAY)) | |||||
self.nodes[0].setmocktime( | self.nodes[0].setmocktime( | ||||
mock_time + | mock_time + | ||||
INBOUND_PEER_TX_DELAY + | max_inbound_delay) | ||||
OVERLOADED_PEER_DELAY) | |||||
p.wait_until(lambda: p.tx_getdata_count == len(txids)) | p.wait_until(lambda: p.tx_getdata_count == len(txids)) | ||||
def test_expiry_fallback(self): | def test_expiry_fallback(self, context): | ||||
self.log.info( | self.log.info( | ||||
'Check that expiry will select another peer for download') | 'Check that expiry will select another peer for download') | ||||
TXID = 0xffaa | TXID = 0xffaa | ||||
peer1 = self.nodes[0].add_p2p_connection(TestP2PConn()) | peer1 = self.nodes[0].add_p2p_connection(TestP2PConn()) | ||||
peer2 = self.nodes[0].add_p2p_connection(TestP2PConn()) | peer2 = self.nodes[0].add_p2p_connection(TestP2PConn()) | ||||
for p in [peer1, peer2]: | for p in [peer1, peer2]: | ||||
p.send_message(msg_inv([CInv(t=MSG_TX, h=TXID)])) | p.send_message(msg_inv([CInv(t=MSG_TX, h=TXID)])) | ||||
# One of the peers is asked for the tx | # One of the peers is asked for the tx | ||||
peer2.wait_until( | peer2.wait_until( | ||||
lambda: sum( | lambda: sum( | ||||
p.tx_getdata_count for p in [ | p.tx_getdata_count for p in [ | ||||
peer1, peer2]) == 1) | peer1, peer2]) == 1) | ||||
with p2p_lock: | with p2p_lock: | ||||
peer_expiry, peer_fallback = ( | peer_expiry, peer_fallback = ( | ||||
peer1, peer2) if peer1.tx_getdata_count == 1 else ( | peer1, peer2) if peer1.tx_getdata_count == 1 else ( | ||||
peer2, peer1) | peer2, peer1) | ||||
assert_equal(peer_fallback.tx_getdata_count, 0) | assert_equal(peer_fallback.tx_getdata_count, 0) | ||||
# Wait for request to peer_expiry to expire | # Wait for request to peer_expiry to expire | ||||
self.nodes[0].setmocktime(int(time.time()) + GETDATA_TX_INTERVAL + 1) | self.nodes[0].setmocktime( | ||||
int(time.time()) + context.constants.getdata_interval + 1) | |||||
peer_fallback.wait_until( | peer_fallback.wait_until( | ||||
lambda: peer_fallback.tx_getdata_count >= 1, timeout=1) | lambda: peer_fallback.tx_getdata_count >= 1, timeout=1) | ||||
with p2p_lock: | with p2p_lock: | ||||
assert_equal(peer_fallback.tx_getdata_count, 1) | assert_equal(peer_fallback.tx_getdata_count, 1) | ||||
# reset mocktime | # reset mocktime | ||||
self.restart_node(0) | self.restart_node(0) | ||||
def test_disconnect_fallback(self): | def test_disconnect_fallback(self, context): | ||||
self.log.info( | self.log.info( | ||||
'Check that disconnect will select another peer for download') | 'Check that disconnect will select another peer for download') | ||||
TXID = 0xffbb | TXID = 0xffbb | ||||
peer1 = self.nodes[0].add_p2p_connection(TestP2PConn()) | peer1 = self.nodes[0].add_p2p_connection(TestP2PConn()) | ||||
peer2 = self.nodes[0].add_p2p_connection(TestP2PConn()) | peer2 = self.nodes[0].add_p2p_connection(TestP2PConn()) | ||||
for p in [peer1, peer2]: | for p in [peer1, peer2]: | ||||
p.send_message(msg_inv([CInv(t=MSG_TX, h=TXID)])) | p.send_message(msg_inv([CInv(t=MSG_TX, h=TXID)])) | ||||
# One of the peers is asked for the tx | # One of the peers is asked for the tx | ||||
peer2.wait_until( | peer2.wait_until( | ||||
lambda: sum( | lambda: sum( | ||||
p.tx_getdata_count for p in [ | p.tx_getdata_count for p in [ | ||||
peer1, peer2]) == 1) | peer1, peer2]) == 1) | ||||
with p2p_lock: | with p2p_lock: | ||||
peer_disconnect, peer_fallback = ( | peer_disconnect, peer_fallback = ( | ||||
peer1, peer2) if peer1.tx_getdata_count == 1 else ( | peer1, peer2) if peer1.tx_getdata_count == 1 else ( | ||||
peer2, peer1) | peer2, peer1) | ||||
assert_equal(peer_fallback.tx_getdata_count, 0) | assert_equal(peer_fallback.tx_getdata_count, 0) | ||||
peer_disconnect.peer_disconnect() | peer_disconnect.peer_disconnect() | ||||
peer_disconnect.wait_for_disconnect() | peer_disconnect.wait_for_disconnect() | ||||
peer_fallback.wait_until( | peer_fallback.wait_until( | ||||
lambda: peer_fallback.tx_getdata_count >= 1, timeout=1) | lambda: peer_fallback.tx_getdata_count >= 1, timeout=1) | ||||
with p2p_lock: | with p2p_lock: | ||||
assert_equal(peer_fallback.tx_getdata_count, 1) | assert_equal(peer_fallback.tx_getdata_count, 1) | ||||
def test_notfound_fallback(self): | def test_notfound_fallback(self, context): | ||||
self.log.info( | self.log.info( | ||||
'Check that notfounds will select another peer for download immediately') | 'Check that notfounds will select another peer for download immediately') | ||||
TXID = 0xffdd | TXID = 0xffdd | ||||
peer1 = self.nodes[0].add_p2p_connection(TestP2PConn()) | peer1 = self.nodes[0].add_p2p_connection(TestP2PConn()) | ||||
peer2 = self.nodes[0].add_p2p_connection(TestP2PConn()) | peer2 = self.nodes[0].add_p2p_connection(TestP2PConn()) | ||||
for p in [peer1, peer2]: | for p in [peer1, peer2]: | ||||
p.send_message(msg_inv([CInv(t=MSG_TX, h=TXID)])) | p.send_message(msg_inv([CInv(t=MSG_TX, h=TXID)])) | ||||
# One of the peers is asked for the tx | # One of the peers is asked for the tx | ||||
peer2.wait_until( | peer2.wait_until( | ||||
lambda: sum( | lambda: sum( | ||||
p.tx_getdata_count for p in [ | p.tx_getdata_count for p in [ | ||||
peer1, peer2]) == 1) | peer1, peer2]) == 1) | ||||
with p2p_lock: | with p2p_lock: | ||||
peer_notfound, peer_fallback = ( | peer_notfound, peer_fallback = ( | ||||
peer1, peer2) if peer1.tx_getdata_count == 1 else ( | peer1, peer2) if peer1.tx_getdata_count == 1 else ( | ||||
peer2, peer1) | peer2, peer1) | ||||
assert_equal(peer_fallback.tx_getdata_count, 0) | assert_equal(peer_fallback.tx_getdata_count, 0) | ||||
# Send notfound, so that fallback peer is selected | # Send notfound, so that fallback peer is selected | ||||
peer_notfound.send_and_ping(msg_notfound(vec=[CInv(MSG_TX, TXID)])) | peer_notfound.send_and_ping(msg_notfound(vec=[CInv(MSG_TX, TXID)])) | ||||
peer_fallback.wait_until( | peer_fallback.wait_until( | ||||
lambda: peer_fallback.tx_getdata_count >= 1, timeout=1) | lambda: peer_fallback.tx_getdata_count >= 1, timeout=1) | ||||
with p2p_lock: | with p2p_lock: | ||||
assert_equal(peer_fallback.tx_getdata_count, 1) | assert_equal(peer_fallback.tx_getdata_count, 1) | ||||
def test_preferred_inv(self): | def test_preferred_inv(self, context): | ||||
self.log.info( | self.log.info( | ||||
'Check that invs from preferred peers are downloaded immediately') | 'Check that invs from preferred peers are downloaded immediately') | ||||
self.restart_node(0, extra_args=['-whitelist=noban@127.0.0.1']) | self.restart_node(0, extra_args=['-whitelist=noban@127.0.0.1']) | ||||
peer = self.nodes[0].add_p2p_connection(TestP2PConn()) | peer = self.nodes[0].add_p2p_connection(TestP2PConn()) | ||||
peer.send_message(msg_inv([CInv(t=MSG_TX, h=0xff00ff00)])) | peer.send_message(msg_inv([CInv(t=MSG_TX, h=0xff00ff00)])) | ||||
peer.wait_until(lambda: peer.tx_getdata_count >= 1, timeout=1) | peer.wait_until(lambda: peer.tx_getdata_count >= 1, timeout=1) | ||||
with p2p_lock: | with p2p_lock: | ||||
assert_equal(peer.tx_getdata_count, 1) | assert_equal(peer.tx_getdata_count, 1) | ||||
def test_large_inv_batch(self): | def test_large_inv_batch(self, context): | ||||
max_peer_announcements = context.constants.max_peer_announcements | |||||
self.log.info( | self.log.info( | ||||
'Test how large inv batches are handled with relay permission') | 'Test how large inv batches are handled with relay permission') | ||||
self.restart_node(0, extra_args=['-whitelist=relay@127.0.0.1']) | self.restart_node(0, extra_args=['-whitelist=relay@127.0.0.1']) | ||||
peer = self.nodes[0].add_p2p_connection(TestP2PConn()) | peer = self.nodes[0].add_p2p_connection(TestP2PConn()) | ||||
peer.send_message(msg_inv([CInv(t=MSG_TX, h=txid) | peer.send_message(msg_inv([CInv(t=MSG_TX, h=txid) | ||||
for txid in range(MAX_PEER_TX_ANNOUNCEMENTS + 1)])) | for txid in range(max_peer_announcements + 1)])) | ||||
peer.wait_until(lambda: peer.tx_getdata_count == | peer.wait_until(lambda: peer.tx_getdata_count == | ||||
MAX_PEER_TX_ANNOUNCEMENTS + 1) | max_peer_announcements + 1) | ||||
self.log.info( | self.log.info( | ||||
'Test how large inv batches are handled without relay permission') | 'Test how large inv batches are handled without relay permission') | ||||
self.restart_node(0) | self.restart_node(0) | ||||
peer = self.nodes[0].add_p2p_connection(TestP2PConn()) | peer = self.nodes[0].add_p2p_connection(TestP2PConn()) | ||||
peer.send_message(msg_inv([CInv(t=MSG_TX, h=txid) | peer.send_message(msg_inv([CInv(t=MSG_TX, h=txid) | ||||
for txid in range(MAX_PEER_TX_ANNOUNCEMENTS + 1)])) | for txid in range(max_peer_announcements + 1)])) | ||||
peer.wait_until(lambda: peer.tx_getdata_count == | peer.wait_until(lambda: peer.tx_getdata_count == | ||||
MAX_PEER_TX_ANNOUNCEMENTS) | max_peer_announcements) | ||||
peer.sync_with_ping() | peer.sync_with_ping() | ||||
with p2p_lock: | with p2p_lock: | ||||
assert_equal(peer.tx_getdata_count, MAX_PEER_TX_ANNOUNCEMENTS) | assert_equal(peer.tx_getdata_count, max_peer_announcements) | ||||
def test_spurious_notfound(self): | def test_spurious_notfound(self, context): | ||||
self.log.info('Check that spurious notfound is ignored') | self.log.info('Check that spurious notfound is ignored') | ||||
self.nodes[0].p2ps[0].send_message(msg_notfound(vec=[CInv(MSG_TX, 1)])) | self.nodes[0].p2ps[0].send_message(msg_notfound(vec=[CInv(MSG_TX, 1)])) | ||||
def run_test(self): | def run_test(self): | ||||
context = TX_TEST_CONTEXT | |||||
# Run tests without mocktime that only need one peer-connection first, | # Run tests without mocktime that only need one peer-connection first, | ||||
# to avoid restarting the nodes | # to avoid restarting the nodes | ||||
self.test_expiry_fallback() | self.test_expiry_fallback(context) | ||||
self.test_disconnect_fallback() | self.test_disconnect_fallback(context) | ||||
self.test_notfound_fallback() | self.test_notfound_fallback(context) | ||||
self.test_preferred_inv() | self.test_preferred_inv(context) | ||||
self.test_large_inv_batch() | self.test_large_inv_batch(context) | ||||
self.test_spurious_notfound() | self.test_spurious_notfound(context) | ||||
# Run each test against new bitcoind instances, as setting mocktimes has long-term effects on when | # Run each test against new bitcoind instances, as setting mocktimes has long-term effects on when | ||||
# the next trickle relay event happens. | # the next trickle relay event happens. | ||||
for test in [self.test_in_flight_max, | for test in [self.test_in_flight_max, | ||||
self.test_inv_block, self.test_tx_requests]: | self.test_inv_block, self.test_tx_requests]: | ||||
self.stop_nodes() | self.stop_nodes() | ||||
self.start_nodes() | self.start_nodes() | ||||
self.connect_nodes(1, 0) | self.connect_nodes(1, 0) | ||||
# Setup the p2p connections | # Setup the p2p connections | ||||
self.peers = [] | self.peers = [] | ||||
for node in self.nodes: | for node in self.nodes: | ||||
for _ in range(NUM_INBOUND): | for _ in range(NUM_INBOUND): | ||||
self.peers.append(node.add_p2p_connection(TestP2PConn())) | self.peers.append( | ||||
node.add_p2p_connection( | |||||
TestP2PConn())) | |||||
self.log.info( | self.log.info( | ||||
"Nodes are setup with {} incoming connections each".format(NUM_INBOUND)) | "Nodes are setup with {} incoming connections each".format(NUM_INBOUND)) | ||||
test() | test(context) | ||||
if __name__ == '__main__': | if __name__ == '__main__': | ||||
TxDownloadTest().main() | TxDownloadTest().main() |