Measure speed of `addr` response to `getaddr` message on Bitcoin network

getaddr_profile.py

# Depends on python-bitcoinlib only

import socket, time, bitcoin
from bitcoin.messages import msg_version, msg_verack, msg_addr, msg_getaddr, MsgSerializable, msg_pong
from bitcoin.net import CAddress
from statistics import mean
from threading import Thread

from concurrent.futures import ThreadPoolExecutor


PORT = 8333

bitcoin.SelectParams('mainnet') 

DNS_SEEDS = [
    'dnsseed.bitcoin.dashjr.org',
    'dnsseed.bluematt.me',
    'seed.bitcoin.sipa.be',
    'seed.bitcoinstats.com',
    'seed.bitcoin.sprovoost.nl',
]


def version_pkt(client_ip, server_ip):
    msg = msg_version()
    msg.nVersion = 70002
    msg.addrTo.ip = server_ip
    msg.addrTo.port = PORT
    msg.addrFrom.ip = client_ip
    msg.addrFrom.port = PORT

    return msg

def addr_pkt( str_addrs ):
    msg = msg_addr()
    addrs = []
    for i in str_addrs:
        addr = CAddress()
        addr.port = 8333
        addr.nTime = int(time.time())
        addr.ip = i

        addrs.append( addr )
    msg.addrs = addrs
    return msg


server_ip = "162.229.209.252"
client_ip = "127.0.0.1"


def fetch_ips(dns_seed):
    ip_list = []
    ais = socket.getaddrinfo(dns_seed, 0, 0, 0, 0)
    for result in ais:
        ip_list.append(result[-1][0])
    return list(set(ip_list))


def fetch_addresses():
    # FIXME: this needs a better name. just confused it with db queries ...
    result = []
    for dns_seed in DNS_SEEDS:
        try:
            ips = fetch_ips(dns_seed)
            addresses = [(ip, 8333) for ip in ips]
            result.extend(addresses)
        except:
            logger.info(f"Error fetching addresses from {dns_seed}")
            continue
    return result


def connect(address):
    sock = socket.create_connection(address, timeout=60)
    stream = sock.makefile('rb')

    # Send Version packet
    sock.send( version_pkt(client_ip, server_ip).to_bytes() )

    # Get Version reply
    version = MsgSerializable.stream_deserialize(stream)

    # Send Verack
    sock.send( msg_verack().to_bytes() )

    # Get Verack
    verack = MsgSerializable.stream_deserialize(stream)

    return sock, stream


def measure_time_to_addr(address, send_getaddr, results):
    sock, stream = connect(address)

    # Send out addr
    sock.send(addr_pkt([server_ip]).to_bytes())

    # Send getaddr if flag set
    if send_getaddr:
        sock.send(msg_getaddr().to_bytes())


    start = time.time()
    while time.time() - start < 60:
        msg = MsgSerializable.stream_deserialize(stream)
        if msg.command == b'addr':
            if len(msg.addrs) > 1 or msg.addrs[0] != address:
                results.append(time.time() - start)
                return
        if msg.command == b'ping':
            sock.send(msg_pong().to_bytes())
    # Timeout ...
    results.append(61)

def run_profile(addresses, send_getaddr):
    trials = []

    for address in addresses:
        duration = measure_time_to_addr(address, send_getaddr)
        trials.append(duration)

        average = sum(trials) / float(len(trials))
        print(f'Average duration: {average}')

    return trials

def run_threaded_profile(addresses, send_getaddr):
    with ThreadPoolExecutor(max_workers=5) as executor:
        return  [executor.submit(measure_time_to_addr, address, send_getaddr)
                 for address in addresses]

def threaded_profile(addresses, send_getaddr):
    results = []
    threads = [Thread(target=measure_time_to_addr, args=(address, send_getaddr, results))
               for address in addresses]
    for thread in threads:
        thread.start()
    for thread in threads:
        thread.join()
    return results


def printout(trials):
    total = len(trials)
    for t in [1, 5, 20, 60]:
        under_time = len([trial for trial in trials if trial < t])
        percent_under_time = under_time / total
        print(f'{percent_under_time} made it in {t}')


def run_profiles():
    start = time.time()
    addresses = fetch_addresses()
    with_getaddr = threaded_profile(addresses, send_getaddr=True)
    without_getaddr = threaded_profile(addresses, send_getaddr=False)

    print('With "getaddr":')
    printout(with_getaddr)
    print('Without "getaddr":')
    printout(without_getaddr)


if __name__ == '__main__':
    run_profiles()