Preperation to support different database backends

neo/Core/Helper.py

@@ -1,9 +1,9 @@
 from base58 import b58decode
 import binascii
 from neo.Blockchain import GetBlockchain, GetStateReader
-from neo.Implementations.Blockchains.LevelDB.CachedScriptTable import CachedScriptTable
-from neo.Implementations.Blockchains.LevelDB.DBCollection import DBCollection
-from neo.Implementations.Blockchains.LevelDB.DBPrefix import DBPrefix
+from neo.Storage.Interface.DBInterface import DBInterface
+from neo.Storage.Common.CachedScriptTable import CachedScriptTable
+from neo.Storage.Common.DBPrefix import DBPrefix
 from neo.Core.State.ContractState import ContractState
 from neo.Core.State.AssetState import AssetState
 from neo.Core.Cryptography.Crypto import Crypto
@@ -201,7 +201,7 @@ def VerifyScripts(verifiable):
                     return False
 
             state_reader = GetStateReader()
-            script_table = CachedScriptTable(DBCollection(blockchain._db, DBPrefix.ST_Contract, ContractState))
+            script_table = CachedScriptTable(DBInterface(blockchain._db, DBPrefix.ST_Contract, ContractState))
 
             engine = ApplicationEngine(TriggerType.Verification, verifiable, script_table, state_reader, Fixed8.Zero())
             engine.LoadScript(verification)

neo/Implementations/Notifications/LevelDB/NotificationDB.py → neo/Implementations/Notifications/NotificationDB.py

@@ -1,6 +1,8 @@
 import plyvel
 from neo.EventHub import events
 from neo.SmartContract.SmartContractEvent import SmartContractEvent, NotifyEvent, NotifyType
+from neo.Storage.Implementation.DBFactory import getNotificationDB
+from neo.Storage.Interface.DBInterface import DBProperties
 from neo.Core.State.ContractState import ContractState
 from neo.Settings import settings
 from neo.Core.Blockchain import Blockchain
@@ -50,7 +52,7 @@ def close():
         Closes the database if it is open
         """
         if NotificationDB.__instance:
-            NotificationDB.__instance.db.close()
+            NotificationDB.__instance.db.closeDB()
             NotificationDB.__instance = None
 
     @property
@@ -69,7 +71,7 @@ def current_events(self):
     def __init__(self, path):
 
         try:
-            self._db = plyvel.DB(path, create_if_missing=True)
+            self._db = getNotificationDB(path)
             logger.info("Created Notification DB At %s " % path)
         except Exception as e:
             logger.info("Notification leveldb unavailable, you may already be running this process: %s " % e)
@@ -126,89 +128,80 @@ def on_persist_completed(self, block):
         """
         if len(self._events_to_write):
 
-            addr_db = self.db.prefixed_db(NotificationPrefix.PREFIX_ADDR)
-            block_db = self.db.prefixed_db(NotificationPrefix.PREFIX_BLOCK)
-            contract_db = self.db.prefixed_db(NotificationPrefix.PREFIX_CONTRACT)
-
-            block_write_batch = block_db.write_batch()
-            contract_write_batch = contract_db.write_batch()
+            addr_db = self.db.getPrefixedDB(NotificationPrefix.PREFIX_ADDR)
+            block_db = self.db.getPrefixedDB(NotificationPrefix.PREFIX_BLOCK)
+            contract_db = self.db.getPrefixedDB(NotificationPrefix.PREFIX_CONTRACT)
 
             block_count = 0
             block_bytes = self._events_to_write[0].block_number.to_bytes(4, 'little')
 
-            for evt in self._events_to_write:  # type:NotifyEvent
-
-                # write the event for both or one of the addresses involved in the transfer
-                write_both = True
-                hash_data = evt.ToByteArray()
-
-                bytes_to = bytes(evt.addr_to.Data)
-                bytes_from = bytes(evt.addr_from.Data)
-
-                if bytes_to == bytes_from:
-                    write_both = False
-
-                total_bytes_to = addr_db.get(bytes_to + NotificationPrefix.PREFIX_COUNT)
-                total_bytes_from = addr_db.get(bytes_from + NotificationPrefix.PREFIX_COUNT)
-
-                if not total_bytes_to:
-                    total_bytes_to = b'\x00'
-
-                if not total_bytes_from:
-                    total_bytes_from = b'x\00'
-
-                addr_to_key = bytes_to + total_bytes_to
-                addr_from_key = bytes_from + total_bytes_from
-
-                with addr_db.write_batch() as b:
-                    b.put(addr_to_key, hash_data)
-                    if write_both:
-                        b.put(addr_from_key, hash_data)
-                    total_bytes_to = int.from_bytes(total_bytes_to, 'little') + 1
-                    total_bytes_from = int.from_bytes(total_bytes_from, 'little') + 1
-                    new_bytes_to = total_bytes_to.to_bytes(4, 'little')
-                    new_bytes_from = total_bytes_from.to_bytes(4, 'little')
-                    b.put(bytes_to + NotificationPrefix.PREFIX_COUNT, new_bytes_to)
-                    if write_both:
-                        b.put(bytes_from + NotificationPrefix.PREFIX_COUNT, new_bytes_from)
-
-                # write the event to the per-block database
-                per_block_key = block_bytes + block_count.to_bytes(4, 'little')
-                block_write_batch.put(per_block_key, hash_data)
-                block_count += 1
-
-                # write the event to the per-contract database
-                contract_bytes = bytes(evt.contract_hash.Data)
-                count_for_contract = contract_db.get(contract_bytes + NotificationPrefix.PREFIX_COUNT)
-                if not count_for_contract:
-                    count_for_contract = b'\x00'
-                contract_event_key = contract_bytes + count_for_contract
-                contract_count_int = int.from_bytes(count_for_contract, 'little') + 1
-                new_contract_count = contract_count_int.to_bytes(4, 'little')
-                contract_write_batch.put(contract_bytes + NotificationPrefix.PREFIX_COUNT, new_contract_count)
-                contract_write_batch.put(contract_event_key, hash_data)
-
-            # finish off the per-block write batch and contract write batch
-            block_write_batch.write()
-            contract_write_batch.write()
+            with block_db.getBatch() as block_write_batch:
+                with contract_db.getBatch() as contract_write_batch:
+                    for evt in self._events_to_write:  # type:NotifyEvent
+                        # write the event for both or one of the addresses involved in the transfer
+                        write_both = True
+                        hash_data = evt.ToByteArray()
+
+                        bytes_to = bytes(evt.addr_to.Data)
+                        bytes_from = bytes(evt.addr_from.Data)
+
+                        if bytes_to == bytes_from:
+                            write_both = False
+
+                        total_bytes_to = addr_db.get(bytes_to + NotificationPrefix.PREFIX_COUNT)
+                        total_bytes_from = addr_db.get(bytes_from + NotificationPrefix.PREFIX_COUNT)
+
+                        if not total_bytes_to:
+                            total_bytes_to = b'\x00'
+
+                        if not total_bytes_from:
+                            total_bytes_from = b'x\00'
+
+                        addr_to_key = bytes_to + total_bytes_to
+                        addr_from_key = bytes_from + total_bytes_from
+
+                        with addr_db.getBatch() as b:
+                            b.put(addr_to_key, hash_data)
+                            if write_both:
+                                b.put(addr_from_key, hash_data)
+                            total_bytes_to = int.from_bytes(total_bytes_to, 'little') + 1
+                            total_bytes_from = int.from_bytes(total_bytes_from, 'little') + 1
+                            new_bytes_to = total_bytes_to.to_bytes(4, 'little')
+                            new_bytes_from = total_bytes_from.to_bytes(4, 'little')
+                            b.put(bytes_to + NotificationPrefix.PREFIX_COUNT, new_bytes_to)
+                            if write_both:
+                                b.put(bytes_from + NotificationPrefix.PREFIX_COUNT, new_bytes_from)
+
+                        # write the event to the per-block database
+                        per_block_key = block_bytes + block_count.to_bytes(4, 'little')
+                        block_write_batch.put(per_block_key, hash_data)
+                        block_count += 1
+
+                        # write the event to the per-contract database
+                        contract_bytes = bytes(evt.contract_hash.Data)
+                        count_for_contract = contract_db.get(contract_bytes + NotificationPrefix.PREFIX_COUNT)
+                        if not count_for_contract:
+                            count_for_contract = b'\x00'
+                        contract_event_key = contract_bytes + count_for_contract
+                        contract_count_int = int.from_bytes(count_for_contract, 'little') + 1
+                        new_contract_count = contract_count_int.to_bytes(4, 'little')
+                        contract_write_batch.put(contract_bytes + NotificationPrefix.PREFIX_COUNT, new_contract_count)
+                        contract_write_batch.put(contract_event_key, hash_data)
 
         self._events_to_write = []
 
         if len(self._new_contracts_to_write):
 
-            token_db = self.db.prefixed_db(NotificationPrefix.PREFIX_TOKEN)
-
-            token_write_batch = token_db.write_batch()
+            token_db = self.db.getPrefixedDB(NotificationPrefix.PREFIX_TOKEN)
 
-            for token_event in self._new_contracts_to_write:
-                try:
-                    hash_data = token_event.ToByteArray()  # used to fail here
-                    hash_key = token_event.contract.Code.ScriptHash().ToBytes()
-                    token_write_batch.put(hash_key, hash_data)
-                except Exception as e:
-                    logger.debug(f"Failed to write new contract, reason: {e}")
-
-            token_write_batch.write()
+            with token_db.getBatch() as token_write_batch:
+                for token_event in self._new_contracts_to_write:
+                    try:
+                        hash_data = token_event.ToByteArray()  # used to fail here
+                        hash_key = token_event.contract.Code.ScriptHash().ToBytes()
+                        token_write_batch.put(hash_key, hash_data)
+                    except Exception as e:
+                        logger.debug(f"Failed to write new contract, reason: {e}")
 
         self._new_contracts_to_write = []
 
@@ -221,12 +214,14 @@ def get_by_block(self, block_number):
         Returns:
             list: a list of notifications
         """
-        blocklist_snapshot = self.db.prefixed_db(NotificationPrefix.PREFIX_BLOCK).snapshot()
+        blocklist_snapshot = self.db.getPrefixedDB(NotificationPrefix.PREFIX_BLOCK).createSnapshot()
+
         block_bytes = block_number.to_bytes(4, 'little')
         results = []
-        for val in blocklist_snapshot.iterator(prefix=block_bytes, include_key=False):
-            event = SmartContractEvent.FromByteArray(val)
-            results.append(event)
+        with blocklist_snapshot.openIter(DBProperties(prefix=block_bytes, include_key=False)) as iterator:
+            for val in iterator:
+                event = SmartContractEvent.FromByteArray(val)
+                results.append(event)
 
         return results
 
@@ -246,16 +241,17 @@ def get_by_addr(self, address):
         if not isinstance(addr, UInt160):
             raise Exception("Incorrect address format")
 
-        addrlist_snapshot = self.db.prefixed_db(NotificationPrefix.PREFIX_ADDR).snapshot()
+        addrlist_snapshot = self.db.getPrefixedDB(NotificationPrefix.PREFIX_ADDR).createSnapshot()
         results = []
 
-        for val in addrlist_snapshot.iterator(prefix=bytes(addr.Data), include_key=False):
-            if len(val) > 4:
-                try:
-                    event = SmartContractEvent.FromByteArray(val)
-                    results.append(event)
-                except Exception as e:
-                    logger.error("could not parse event: %s %s" % (e, val))
+        with addrlist_snapshot.openIter(DBProperties(prefix=bytes(addr.Data), include_key=False)) as iterator:
+            for val in iterator:
+                if len(val) > 4:
+                    try:
+                        event = SmartContractEvent.FromByteArray(val)
+                        results.append(event)
+                    except Exception as e:
+                        logger.error("could not parse event: %s %s" % (e, val))
         return results
 
     def get_by_contract(self, contract_hash):
@@ -274,16 +270,17 @@ def get_by_contract(self, contract_hash):
         if not isinstance(hash, UInt160):
             raise Exception("Incorrect address format")
 
-        contractlist_snapshot = self.db.prefixed_db(NotificationPrefix.PREFIX_CONTRACT).snapshot()
+        contractlist_snapshot = self.db.getPrefixedDB(NotificationPrefix.PREFIX_CONTRACT).createSnapshot()
         results = []
 
-        for val in contractlist_snapshot.iterator(prefix=bytes(hash.Data), include_key=False):
-            if len(val) > 4:
-                try:
-                    event = SmartContractEvent.FromByteArray(val)
-                    results.append(event)
-                except Exception as e:
-                    logger.error("could not parse event: %s %s" % (e, val))
+        with contractlist_snapshot.openIter(DBProperties(prefix=bytes(hash.Data), include_key=False)) as iterator:
+            for val in iterator:
+                if len(val) > 4:
+                    try:
+                        event = SmartContractEvent.FromByteArray(val)
+                        results.append(event)
+                    except Exception as e:
+                        logger.error("could not parse event: %s %s" % (e, val))
         return results
 
     def get_tokens(self):
@@ -292,11 +289,13 @@ def get_tokens(self):
         Returns:
             list: A list of smart contract events with contracts that are NEP5 Tokens
         """
-        tokens_snapshot = self.db.prefixed_db(NotificationPrefix.PREFIX_TOKEN).snapshot()
+        tokens_snapshot = self.db.getPrefixedDB(NotificationPrefix.PREFIX_TOKEN).createSnapshot()
         results = []
-        for val in tokens_snapshot.iterator(include_key=False):
-            event = SmartContractEvent.FromByteArray(val)
-            results.append(event)
+
+        with tokens_snapshot.openIter(DBProperties(include_key=False)) as iterator:
+            for val in iterator:
+                event = SmartContractEvent.FromByteArray(val)
+                results.append(event)
         return results
 
     def get_token(self, hash):
@@ -308,7 +307,7 @@ def get_token(self, hash):
         Returns:
             SmartContractEvent: A smart contract event with a contract that is an NEP5 Token
         """
-        tokens_snapshot = self.db.prefixed_db(NotificationPrefix.PREFIX_TOKEN).snapshot()
+        tokens_snapshot = self.db.getPrefixedDB(DBProperties(NotificationPrefix.PREFIX_TOKEN)).createSnapshot()
 
         try:
             val = tokens_snapshot.get(hash.ToBytes())

neo/Implementations/Notifications/LevelDB/test_notification_db.py → neo/Implementations/Notifications/test_notification_db.py

@@ -8,7 +8,7 @@
 import shutil
 import os
 
-from neo.Implementations.Notifications.LevelDB.NotificationDB import NotificationDB
+from neo.Implementations.Notifications.NotificationDB import NotificationDB
 from neo.Core.BigInteger import BigInteger
 
 

neo/Network/NodeLeader.py

@@ -3,7 +3,6 @@
 from typing import List
 from neo.Core.Block import Block
 from neo.Core.Blockchain import Blockchain as BC
-from neo.Implementations.Blockchains.LevelDB.TestLevelDBBlockchain import TestLevelDBBlockchain
 from neo.Core.TX.Transaction import Transaction
 from neo.Core.TX.MinerTransaction import MinerTransaction
 from neo.Network.NeoNode import NeoNode, HEARTBEAT_BLOCKS
@@ -542,8 +541,7 @@ def RelayDirectly(self, inventory):
             relayed |= peer.Relay(inventory)
 
         if len(self.Peers) == 0:
-            if type(BC.Default()) is TestLevelDBBlockchain:
-                # mock a true result for tests
+            if BC.Default().UT:
                 return True
 
             logger.info("no connected peers")