Merge pull request #61 from pipermerriam/piper/make-ethtesterclient-work-async

Make ethtester async

Author: pipermerriam

populus/ethtester_client.py

@@ -4,6 +4,10 @@
 
 https://github.com/ConsenSys/eth-testrpc
 """
+import time
+import Queue
+import threading
+import uuid
 
 
 from ethereum import utils as ethereum_utils
@@ -121,10 +125,34 @@ class EthTesterClient(object):
     Stand-in replacement for the rpc client that speaks directly to the
     `ethereum.tester` facilities.
     """
-    def __init__(self):
+    def __init__(self, async=True, async_timeout=10):
         self.evm = t.state()
         self.evm.mine()
 
+        self.is_async = async
+        self.async_timeout = async_timeout
+
+        if self.is_async:
+            self.request_queue = Queue.Queue()
+            self.results = {}
+
+            self.request_thread = threading.Thread(target=self.process_requests)
+            self.request_thread.daemon = True
+            self.request_thread.start()
+
+    def process_requests(self):
+        while True:
+            id, args, kwargs = self.request_queue.get()
+            mine = kwargs.pop('_mine', False)
+            try:
+                self._send_transaction(*args, **kwargs)
+                if mine:
+                    self.evm.mine()
+                response = self.evm.last_tx.hash
+            except ValueError as e:
+                response = e
+            self.results[id] = response
+
     def wait_for_block(self, block_number, max_wait=0):
         while self.evm.block.number < block_number:
             self.evm.mine()
@@ -177,9 +205,34 @@ def _send_transaction(self, _from=None, to=None, gas=None, gas_price=None,
         return self.evm.send(sender=sender, to=to, value=value, evmdata=data)
 
     def send_transaction(self, *args, **kwargs):
-        self._send_transaction(*args, **kwargs)
-        self.evm.mine()
-        return self.evm.last_tx.hash
+        if self.is_async:
+            kwargs['_mine'] = True
+            request_id = uuid.uuid4()
+            self.request_queue.put((request_id, args, kwargs))
+            start = time.time()
+            while time.time() - start < self.async_timeout:
+                if request_id in self.results:
+                    return self.results.pop(request_id)
+            raise ValueError("Timeout waiting for {0}".format(request_id))
+        else:
+            self._send_transaction(*args, **kwargs)
+            self.evm.mine()
+            return self.evm.last_tx.hash
+
+    def make_ipc_request(self, *args, **kwargs):
+        if self.is_async:
+            request_id = uuid.uuid4()
+            self.request_queue.put((request_id, args, kwargs))
+            start = time.time()
+            while time.time() - start < 10:
+                if request_id in self.results:
+                    result = self.results.pop(request_id)
+                    if isinstance(result, Exception):
+                        raise result
+                    return result
+            raise ValueError("Timeout waiting for {0}".format(request_id))
+        else:
+            return self._make_ipc_request(*args, **kwargs)
 
     def _get_transaction_by_hash(self, txn_hash):
         txn_hash = strip_0x(txn_hash)

tests/ethtester/test_ethtester_async.py

@@ -0,0 +1,36 @@
+import pytest
+import threading
+
+from ethereum import tester
+
+from populus.ethtester_client import EthTesterClient
+
+
+def test_async_requests():
+    client = EthTesterClient(async=True, async_timeout=60)
+
+    threads = []
+    errors = []
+    to_addr = tester.encode_hex(tester.accounts[1])
+
+    def spam_block_number():
+        for i in range(5):
+            try:
+                client.send_transaction(
+                    to=to_addr,
+                    value=1,
+                )
+            except Exception as e:
+                errors.append(e)
+                pytest.fail(e.message)
+
+    for i in range(5):
+        thread = threading.Thread(target=spam_block_number)
+        thread.daemon = True
+        threads.append(thread)
+
+    [thread.start() for thread in threads]
+
+    [thread.join() for thread in threads]
+
+    assert not errors