Ethereum node query performance benchmark

eth-benchmark.py

import timeit
import threading
import asyncio
import contextlib
import requests

from aiohttp import ClientSession

from web3.providers.base import JSONBaseProvider
from web3.providers import HTTPProvider
from web3 import Web3

@contextlib.contextmanager
def profile(name):
    start_time = timeit.default_timer()
    yield
    print('{}: {:.3f}s'.format(name, timeit.default_timer() - start_time))

# sequential request
def seq(node_address, blocks):
    for block in blocks:
        base_provider = JSONBaseProvider()
        request_data = b'[' + b','.join(
            [base_provider.encode_rpc_request('debug_traceBlockByNumber', [hex(block), {"tracer": "callTracer", "timeout": "60s"}])]
            ) + b']'
        r = requests.post(eth_node_address, data=request_data, headers={'Content-Type': 'application/json'})
        responses = base_provider.decode_rpc_response(r.content)

# threaded request
def threads(node_address, blocks):
    def send_request(block):
        base_provider = JSONBaseProvider()
        request_data = b'[' + b','.join(
            [base_provider.encode_rpc_request('debug_traceBlockByNumber', [hex(block), {"tracer": "callTracer", "timeout": "60s"}])]
            ) + b']'
        r = requests.post(eth_node_address, data=request_data, headers={'Content-Type': 'application/json'})
        responses = base_provider.decode_rpc_response(r.content)

    threads = []
    for i in range(len(blocks)):
        t = threading.Thread(target=send_request, args=[blocks[i]])
        t.start()
        threads.append(t)
    for t in threads:
        t.join()

# asynchronous JSON RPC API request
async def run(node_address, blocks):
    async def async_make_request(session, url, method, params):
        base_provider = JSONBaseProvider()
        request_data = base_provider.encode_rpc_request(method, params)
        async with session.post(url, data=request_data,
                            headers={'Content-Type': 'application/json'}) as response:
            content = await response.read()
        response = base_provider.decode_rpc_response(content)
        return response

    tasks = []

    # Fetch all responses within one Client session,
    # keep connection alive for all requests.
    async with ClientSession() as session:
        for block in blocks:
            task = asyncio.ensure_future(async_make_request(session, node_address,
                                                        'debug_traceBlockByNumber',[hex(block), {"tracer": "callTracer", "timeout": "60s"}]))
            tasks.append(task)

        responses = await asyncio.gather(*tasks)

# batch request
def batch(node_address, blocks):
    base_provider = JSONBaseProvider()
    request_data = b'[' + b','.join(
        [base_provider.encode_rpc_request('debug_traceBlockByNumber', [hex(block), {"tracer": "callTracer", "timeout": "60s"}]) for block in blocks]
        ) + b']'
    r = requests.post(eth_node_address, data=request_data, headers={'Content-Type': 'application/json'})
    responses = base_provider.decode_rpc_response(r.content)

if __name__ == "__main__":
    eth_node_address = "node_endpoint"
    web3 = Web3(HTTPProvider(eth_node_address))

    blocks = []
    for i in range(12722919,12722924):
        blocks.append(i)
        print(i)

    with profile('sequential'):
        seq(eth_node_address, blocks)

    with profile('threaded'):
        threads(eth_node_address, blocks)

    with profile('async'):
        loop = asyncio.get_event_loop()
        future = asyncio.ensure_future(run(eth_node_address, blocks))
        loop.run_until_complete(future)

    with profile('batch'):
        batch(eth_node_address, blocks)