Perform batch RPC requests to ethereum node

[Uxio0]

• Version: 4.2.0
• Python: 3.6
• OS: osx/linux/win

What was wrong?

There's currently not support for batch requests, when using HTTPProvider. Even using IPC/WS connectors I think a speed up could be possible using batch requests.

How can it be fixed?

Implementing batch requests

[carver]

I don't understand what you mean by batching, here. IPC is a streaming connection. Can you elaborate?

[pipermerriam]

@carver web3js at least historically had a feature for asynchronous request batching via an API that let you stack up some JSON-RPC requests, and then submit them as a batch. The requests would be issued concurrently. (feature may still exist, not sure).

I'm good with exploring this alongside #657

[Uxio0]

I'm using this feature with HTTP requests to the JSON RPC API (no web3 involved) and is working very fast (compared to doing everything in multiple requests or event using a ThreadPoolExecutor), so I think it can't harm web3 to at least have that option

[carver]

web3js at least historically had a feature for asynchronous request batching via an API that let you stack up some JSON-RPC requests, and then submit them as a batch

Right, that makes sense in web3js, because the HTTP request to your node has a lot of overhead (HTTP headers) for each message. But there is roughly 0 overhead for each message in an IPC stream. So I'm skeptical that it will provide any benefit in that context.

I'm less sure about the internals of websockets. There might be some benefit there.

Batching over an HTTP connection is likely to have benefits. 👍

[Uxio0]

Hi @carver,

I agree about that, but:

• I'm talking about HTTP, not in every environment is possible to use IPC/WS.
• Even using IPC/WS, it's possible that Geth/Parity and other nodes have optimizations depending on if they are receiving a batch request or multiple streamed requests. At least for me is faster to send batchs than to send synchronous IPC requests (python 3.6 at least with web3 4.2.0).

Hope I have explained myself better :)

[carver]

Ah yes, thanks. I updated the issue to indicate that HTTP is the primary target. 👍

I will be happy if we find that it boosts speed for everything else, too!

[pipermerriam]

@carver having now thought about this for a few minutes, I think this is going to be easy if we build it on top of #657 since everything under web3.async.* will be coroutine based and thus, batching them to run concurrently (in the asyncio version of concurrency) should be really easy to expose .

Here's some imagined API

w3 = Web3(...)

batch = web3.async.create_batch()
batch.eth.getBalance(addr_a)
batch.eth.getBalance(addr_b)
batch.eth.getBalance(addr_c)
bal_a, bal_b, bal_c = await batch.execute()

[dangell7]

Interesting idea. What happens when one tries to send multiple tranascations? Cuz in my system, I have to wait till the first transaction clears before send the next. Prob because getTransactionCount doesn’t take the pending transaction into consideration. I didn’t really dive into the problem, but wanted to bring it up.

[pipermerriam]

@dangell7 good point. We may choose to start with a minimal initial implementation which only allows batching requests that are read-only or that don't modify state.

[boneyard93501]

@pipermerriam anybody working on this? it's not all that far off the poc i did a while ago. might be able to resurrect that without too many changes as a wip starting point.

[carver]

@boneyard93501 not that I've seen

[boneyard93501]

@carver cool. if you don't mind, i'll take a stab at it starting wednesday.

[pipermerriam]

@boneyard93501 are you looking at starting into the web3.async API? If not, can you provide a proposal for what you're looking to build? Basic API and a basic description of how it'll work.

[boneyard93501]

@pipermerriam @carver
if a solution is supposed to be provider agnostic, i think a thread-based pub-sub solution is probably the most straight forward (and could be wrapped in an event-loop, if needed). quick and dirty outline, untested, unoptimized/un-toolzed:

import time
import datetime
import queue
import threading


WORKER_TIMEOUT = 5.0


def batch_request_handler(rpc_request_queue, provider, response_processing_fn):
    '''  '''
    last_job = None
    while 1:
        try:
            if (datetime.datetime.utcnow() - last_job).total_seconds() > WORKER_TIMEOUT:
                response_processing_fn(('','', 'TimeoutError'))
                break
            job = rpc_request_queue.get()
            if job is None:
                break
            last_job = datetime.datetime.utcnow()
            try:
                result = provider._make_request(job)
                response_processing_fn((job,result, ''))
            except Exception as exc:
                response_processing_fn((job,'',exc))
        except (KeyboardInterrupt, SystemExit):
            pass
        finally:
            rpc_request_queue.task_done()
        time.sleep(0.1)


def batch_request(provider, payload, response_processing_fn, max_request_threads=10, max_request_q=200):
    '''
        could change provider to provider_type and then create provider in the worker to
        avoid any possible thread-safety issues now of down the road.
        also should take threading.Event, adjust worker while condition accordingly, so user can kill it all at will
   '''

    rpc_reqeuest_queue = queue.Queue()

    request_args = (rpc_reqeuest_queue, provider, response_processing_fn)
    req_threads = []

    for i in range(max_request_threads):
        t = threading.Thread(target=batch_request_handler, args=request_args)
        t.setDaemon(True)  # we don't really need it since we're joining but it helps to avoid zombies
        t.name = f'batch request thread {i}'
        req_threads.append(t)

    [t.start() for t in req_threads]

    for job in payload:
        rpc_reqeuest_queue.put(job)

    time.sleep(WORKER_TIMEOUT + 1.0)
    [t.join() for t in req_threads]

 

# user side
class SuperSimplClientBatchProcessingClass:
    def __init__(self, batch_size):
        ''' '''
        self.expected_results = batch_size
        self.jobs_processed = 0

    def job_update(self, data):
        ''' only requirement from web3py side is to be able to accept a (named) tuple '''
        self.jobs_processed += 1
        if data[1]:
            pass  # do something with successes
        else:
            pass  # do something with failures including TimeoutError

    @property
    def done(self):
        if len(self.jobs_processed) == self.jobs_processed:
            return True
        else:
            return False

    @property
    def progress(self):
        return {'batch-size': self.batch_size, 'processed_jobs': self.jobs_processed}