Move HexaryTrieSync from py-trie here, and make it async (#1124)

Author: gsalgado

tests/trinity/core/integration_test_helpers.py

@@ -6,7 +6,10 @@
 from eth.chains.base import (
     MiningChain,
 )
+from eth.db.backends.level import LevelDB
+from eth.db.backends.memory import MemoryDB
 
+from trinity.db.base import AsyncBaseDB
 from trinity.db.chain import AsyncChainDB
 from trinity.db.header import AsyncHeaderDB
 
@@ -31,6 +34,16 @@ async def passthrough_method(self, *args, **kwargs):
     return passthrough_method
 
 
+class FakeAsyncMemoryDB(MemoryDB, AsyncBaseDB):
+    coro_set = async_passthrough('set')
+    coro_exists = async_passthrough('exists')
+
+
+class FakeAsyncLevelDB(LevelDB, AsyncBaseDB):
+    coro_set = async_passthrough('set')
+    coro_exists = async_passthrough('exists')
+
+
 class FakeAsyncHeaderDB(AsyncHeaderDB):
     coro_get_canonical_block_hash = async_passthrough('get_canonical_block_hash')
     coro_get_canonical_block_header_by_number = async_passthrough('get_canonical_block_header_by_number')  # noqa: E501

tests/trinity/core/p2p-proto/test_state_sync.py

@@ -2,12 +2,75 @@
 import random
 import time
 
+import pytest
+
+from hypothesis import (
+    given,
+    settings,
+    strategies,
+    example,
+)
+from hypothesis.types import RandomWithSeed
+
+from trie import HexaryTrie
+
 from eth.db.backends.memory import MemoryDB
 from eth.db.account import AccountDB
 from eth.utils.logging import TraceLogger
 
+from trinity.sync.full.hexary_trie import HexaryTrieSync
 from trinity.sync.full.state import StateSync, TrieNodeRequestTracker
 
+from tests.trinity.core.integration_test_helpers import FakeAsyncMemoryDB
+
+
+# produces a branch node with an extention node who's encoding is less than 32
+# bytes in length so it is inlined.
+EXAMPLE_37968 = 37968
+
+# produces an top level extension node who's encoding is less than 32 bytes in
+# length so it gets inlined.
+EXAMPLE_809368 = 809368
+
+
+def make_random_trie(random):
+    trie = HexaryTrie({})
+    contents = {}
+    for _ in range(1000):
+        key_length = random.randint(2, 32)
+        key = bytes([random.randint(0, 255) for _ in range(key_length)])
+        value_length = random.randint(2, 64)
+        value = bytes([random.randint(0, 255) for _ in range(value_length)])
+        trie[key] = value
+        contents[key] = value
+    return trie, contents
+
+
+@given(random=strategies.randoms())
+@settings(max_examples=10)
+@example(random=RandomWithSeed(EXAMPLE_37968))
+@example(random=RandomWithSeed(EXAMPLE_809368))
+def test_trie_sync(random, event_loop):
+
+    # Apparently hypothesis tests cannot be used in conjunction with pytest-asyncio yet, so do it
+    # like this for now. https://github.com/HypothesisWorks/hypothesis/pull/1343
+    async def _test_trie_sync():
+        src_trie, contents = make_random_trie(random)
+        dest_db = FakeAsyncMemoryDB()
+        scheduler = HexaryTrieSync(src_trie.root_hash, dest_db, TraceLogger("test"))
+        requests = scheduler.next_batch()
+        while len(requests) > 0:
+            results = []
+            for request in requests:
+                results.append([request.node_key, src_trie.db[request.node_key]])
+            await scheduler.process(results)
+            requests = scheduler.next_batch(10)
+        dest_trie = HexaryTrie(dest_db, src_trie.root_hash)
+        for key, value in contents.items():
+            assert dest_trie[key] == value
+
+    event_loop.run_until_complete(_test_trie_sync())
+
 
 def make_random_state(n):
     raw_db = MemoryDB()
@@ -29,16 +92,17 @@ def make_random_state(n):
     return raw_db, account_db.state_root, contents
 
 
-def test_state_sync():
+@pytest.mark.asyncio
+async def test_state_sync():
     raw_db, state_root, contents = make_random_state(1000)
-    dest_db = MemoryDB()
-    scheduler = StateSync(state_root, dest_db)
+    dest_db = FakeAsyncMemoryDB()
+    scheduler = StateSync(state_root, dest_db, TraceLogger('test'))
     requests = scheduler.next_batch(10)
     while requests:
         results = []
         for request in requests:
             results.append([request.node_key, raw_db[request.node_key]])
-        scheduler.process(results)
+        await scheduler.process(results)
         requests = scheduler.next_batch(10)
 
     result_account_db = AccountDB(dest_db, state_root)

tests/trinity/core/p2p-proto/test_sync.py

@@ -6,7 +6,6 @@
 from eth_utils import decode_hex
 
 from eth import constants
-from eth.db.backends.memory import MemoryDB
 from eth.vm.forks.frontier import FrontierVM, _PoWMiningVM
 
 
@@ -20,6 +19,7 @@
     FakeAsyncChain,
     FakeAsyncChainDB,
     FakeAsyncHeaderDB,
+    FakeAsyncMemoryDB,
 )
 from tests.trinity.core.peer_helpers import (
     get_directly_linked_peers,
@@ -137,7 +137,7 @@ def finalizer():
 
 @pytest.fixture
 def chaindb_20():
-    chain = PoWMiningChain.from_genesis(MemoryDB(), GENESIS_PARAMS, GENESIS_STATE)
+    chain = PoWMiningChain.from_genesis(FakeAsyncMemoryDB(), GENESIS_PARAMS, GENESIS_STATE)
     for i in range(20):
         tx = chain.create_unsigned_transaction(
             nonce=i,
@@ -154,7 +154,7 @@ def chaindb_20():
 
 @pytest.fixture
 def chaindb_fresh():
-    chain = PoWMiningChain.from_genesis(MemoryDB(), GENESIS_PARAMS, GENESIS_STATE)
+    chain = PoWMiningChain.from_genesis(FakeAsyncMemoryDB(), GENESIS_PARAMS, GENESIS_STATE)
     assert chain.chaindb.get_canonical_head().block_number == 0
     return chain.chaindb
 

trinity/db/base.py

@@ -4,6 +4,10 @@
     BaseProxy,
 )
 
+from eth.db.backends.base import BaseDB
+
+from trinity.utils.mp import async_method
+
 
 class DBProxy(BaseProxy):
     _exposed_ = (
@@ -15,7 +19,11 @@ class DBProxy(BaseProxy):
         'exists',
         'get',
         'set',
+        'coro_set',
+        'coro_exists',
     )
+    coro_set = async_method('set')
+    coro_exists = async_method('exists')
 
     def get(self, key: bytes) -> bytes:
         return self._callmethod('get', (key,))
@@ -40,3 +48,12 @@ def exists(self, key: bytes) -> bool:
 
     def __contains__(self, key: bytes) -> bool:
         return self._callmethod('__contains__', (key,))
+
+
+class AsyncBaseDB(BaseDB):
+
+    async def coro_set(self, key: bytes, value: bytes) -> None:
+        raise NotImplementedError()
+
+    async def coro_exists(self, key: bytes) -> bool:
+        raise NotImplementedError()

trinity/exceptions.py

@@ -30,3 +30,10 @@ class AlreadyWaiting(BaseTrinityError):
     peer when there is already an active wait for that message type.
     """
     pass
+
+
+class SyncRequestAlreadyProcessed(BaseTrinityError):
+    """
+    Raised when a trie SyncRequest has already been processed.
+    """
+    pass

trinity/server.py

@@ -16,7 +16,6 @@
 from cancel_token import CancelToken, OperationCancelled
 
 from eth.chains import AsyncChain
-from eth.db.backends.base import BaseDB
 
 from p2p.auth import (
     decode_authentication,
@@ -53,6 +52,7 @@
 )
 from p2p.service import BaseService
 
+from trinity.db.base import AsyncBaseDB
 from trinity.db.chain import AsyncChainDB
 from trinity.db.header import BaseAsyncHeaderDB
 from trinity.protocol.eth.peer import ETHPeer
@@ -76,7 +76,7 @@ def __init__(self,
                  chain: AsyncChain,
                  chaindb: AsyncChainDB,
                  headerdb: BaseAsyncHeaderDB,
-                 base_db: BaseDB,
+                 base_db: AsyncBaseDB,
                  network_id: int,
                  max_peers: int = DEFAULT_MAX_PEERS,
                  peer_class: Type[BasePeer] = ETHPeer,
@@ -295,16 +295,15 @@ def _test() -> None:
     from pathlib import Path
     import signal
 
-    from eth.db.backends.level import LevelDB
     from eth.chains.ropsten import RopstenChain, ROPSTEN_GENESIS_HEADER
 
     from p2p import ecies
     from p2p.constants import ROPSTEN_BOOTNODES
 
     from trinity.utils.chains import load_nodekey
 
-    from tests.p2p.integration_test_helpers import FakeAsyncHeaderDB
-    from tests.p2p.integration_test_helpers import FakeAsyncChainDB, FakeAsyncRopstenChain
+    from tests.p2p.integration_test_helpers import (
+        FakeAsyncLevelDB, FakeAsyncHeaderDB, FakeAsyncChainDB, FakeAsyncRopstenChain)
 
     parser = argparse.ArgumentParser()
     parser.add_argument('-db', type=str, required=True)
@@ -322,7 +321,7 @@ def _test() -> None:
     logging.getLogger('p2p.server.Server').setLevel(log_level)
 
     loop = asyncio.get_event_loop()
-    db = LevelDB(args.db)
+    db = FakeAsyncLevelDB(args.db)
     headerdb = FakeAsyncHeaderDB(db)
     chaindb = FakeAsyncChainDB(db)
     chaindb.persist_header(ROPSTEN_GENESIS_HEADER)