Added basic state tree pruning to chain class

Author: vbuterin

ethereum/common.py

@@ -10,6 +10,8 @@
 from ethereum.messages import apply_transaction
 log = get_logger('eth.block')
 
+SKIP_RECEIPT_ROOT_VALIDATION = False
+
 # Gas limit adjustment algo
 def calc_gaslimit(parent, config=default_config):
     decay = parent.gas_limit // config['GASLIMIT_EMA_FACTOR']
@@ -143,7 +145,7 @@ def verify_execution_results(state, block):
     if block.header.state_root != state.trie.root_hash:
         raise ValueError("State root mismatch: header %s computed %s" %
                          (encode_hex(block.header.state_root), encode_hex(state.trie.root_hash)))
-    if block.header.receipts_root != mk_receipt_sha(state.receipts):
+    if block.header.receipts_root != mk_receipt_sha(state.receipts) and not SKIP_RECEIPT_ROOT_VALIDATION:
         raise ValueError("Receipt root mismatch: header %s computed %s, gas used header %d computed %d, %d receipts" %
                          (encode_hex(block.header.receipts_root), encode_hex(mk_receipt_sha(state.receipts)),
                          block.header.gas_used, state.gas_used, len(state.receipts)))

ethereum/messages.py

@@ -40,7 +40,6 @@
 CREATE_CONTRACT_ADDRESS = b''
 
 # DEV OPTIONS
-SKIP_RECEIPT_ROOT_VALIDATION = False
 SKIP_MEDSTATES = False
 
 def rp(tx, what, actual, target):

ethereum/pow/chain.py

@@ -29,7 +29,7 @@
 class Chain(object):
 
     def __init__(self, genesis=None, env=None, \
-                 new_head_cb=None, reset_genesis=False, localtime=None, **kwargs):
+                 new_head_cb=None, reset_genesis=False, localtime=None, max_history=1000, **kwargs):
         self.env = env or Env()
         # Initialize the state
         if 'head_hash' in self.db:  # new head tag
@@ -82,6 +82,7 @@ def __init__(self, genesis=None, env=None, \
         self.time_queue = []
         self.parent_queue = {}
         self.localtime = time.time() if localtime is None else localtime
+        self.max_history = max_history
 
     # Head (tip) of the chain
     @property
@@ -246,6 +247,7 @@ def add_block(self, block):
                      (now, block.header.timestamp, block.header.timestamp - now))
             return False
         # Is the block being added to the head?
+        self.state.deletes = []
         if block.header.prevhash == self.head_hash:
             log.info('Adding to head', head=encode_hex(block.header.prevhash))
             try:
@@ -260,6 +262,7 @@ def add_block(self, block):
             for i, tx in enumerate(block.transactions):
                 self.db.put(b'txindex:' + tx.hash, rlp.encode([block.number, i]))
             assert self.get_blockhash_by_number(block.header.number) == block.header.hash
+            deletes = self.state.deletes
         # Or is the block being added to a chain that is not currently the head?
         elif block.header.prevhash in self.env.db:
             log.info('Receiving block not on head, adding to secondary post state',
@@ -271,6 +274,7 @@ def add_block(self, block):
                 log.info('Block %s with parent %s invalid, reason: %s' %
                          (encode_hex(block.header.hash), encode_hex(block.header.prevhash), e))
                 return False
+            deletes = temp_state.deletes
             block_score = self.get_score(block)
             # If the block should be the new head, replace the head
             if block_score > self.get_score(self.head):
@@ -313,12 +317,28 @@ def add_block(self, block):
             if block.header.prevhash not in self.parent_queue:
                 self.parent_queue[block.header.prevhash] = []
             self.parent_queue[block.header.prevhash].append(block)
-            log.info('No parent found. Delaying for now')
+            log.info('Got block %d (%s) with prevhash %s, parent not found. Delaying for now' %
+                     (block.number, encode_hex(block.hash), encode_hex(block.prevhash)))
             return False
         self.add_child(block)
         self.db.put('head_hash', self.head_hash)
-        self.db.put(block.header.hash, rlp.encode(block))
+        self.db.put(block.hash, rlp.encode(block))
+        self.db.put(b'deletes:'+block.hash, b''.join(deletes))
+        print('Saved %d trie node deletes for block %d (%s)' % (len(deletes), block.number, utils.encode_hex(block.hash)))
+        # Delete old junk data
+        old_block_hash = self.get_blockhash_by_number(block.number - self.max_history)
+        if old_block_hash:
+            try:
+                deletes = self.db.get(b'deletes:'+old_block_hash)
+                print('Deleting %d trie nodes' % (len(deletes) // 32))
+                for i in range(0, len(deletes), 32):
+                    self.db.delete(deletes[i: i+32])
+                self.db.delete(b'deletes:'+old_block_hash)
+            except KeyError as e:
+                print(e)
+                pass
         self.db.commit()
+        assert (b'deletes:'+block.hash) in self.db
         log.info('Added block %d (%s) with %d txs and %d gas' % \
             (block.header.number, encode_hex(block.header.hash)[:8],
              len(block.transactions), block.header.gas_used))

ethereum/securetrie.py

@@ -41,6 +41,14 @@ def root_hash(self):
     def root_hash(self, value):
         self.trie.root_hash = value
 
+    @property
+    def deletes(self):
+        return self.trie.deletes
+
+    @deletes.setter
+    def deletes(self, value):
+        self.trie.deletes = value
+
     def process_epoch(self, epoch):
         self.trie.process_epoch(epoch)
 

ethereum/state.py

@@ -58,12 +58,13 @@ class Account(rlp.Serializable):
         ('code_hash', hash32)
     ]
 
-    def __init__(self, nonce, balance, storage, code_hash, env):
+    def __init__(self, nonce, balance, storage, code_hash, env, address):
         assert isinstance(env.db, BaseDB)
         self.env = env
+        self.address = address
         super(Account, self).__init__(nonce, balance, storage, code_hash)
         self.storage_cache = {}
-        self.storage_trie = SecureTrie(Trie(self.env.db))
+        self.storage_trie = SecureTrie(Trie(self.env.db, prefix=address))
         self.storage_trie.root_hash = self.storage
         self.touched = False
         self.existent_at_start = True
@@ -102,9 +103,9 @@ def set_storage_data(self, key, value):
         self.storage_cache[key] = value
 
     @classmethod
-    def blank_account(cls, env, initial_nonce=0):
+    def blank_account(cls, env, address, initial_nonce=0):
         env.db.put(BLANK_HASH, b'')
-        o = cls(initial_nonce, 0, trie.BLANK_ROOT, BLANK_HASH, env)
+        o = cls(initial_nonce, 0, trie.BLANK_ROOT, BLANK_HASH, env, address)
         o.existent_at_start = False
         return o
 
@@ -136,6 +137,7 @@ def __init__(self, root=b'', env=Env(), **kwargs):
         self.journal = []
         self.cache = {}
         self.log_listeners = []
+        self.deletes = []
 
     @property
     def db(self):
@@ -160,9 +162,9 @@ def get_and_cache_account(self, address):
             return self.cache[address]
         rlpdata = self.trie.get(address)
         if rlpdata != trie.BLANK_NODE:
-            o = rlp.decode(rlpdata, Account, env=self.env)
+            o = rlp.decode(rlpdata, Account, env=self.env, address=address)
         else:
-            o = Account.blank_account(self.env, self.config['ACCOUNT_INITIAL_NONCE'])
+            o = Account.blank_account(self.env, address, self.config['ACCOUNT_INITIAL_NONCE'])
         self.cache[address] = o
         o._mutable = True
         o._cached_rlp = None
@@ -315,12 +317,15 @@ def commit(self, allow_empties=False):
         for addr, acct in self.cache.items():
             if acct.touched:
                 acct.commit()
+                self.deletes.extend(acct.storage_trie.deletes)
                 if acct.exists or allow_empties or (not self.is_SPURIOUS_DRAGON() and not acct.deleted):
                     # print('upd', encode_hex(addr))
                     self.trie.update(addr, rlp.encode(acct))
                 else:
                     # print('del', encode_hex(addr))
                     self.trie.delete(addr)
+        self.deletes.extend(self.trie.deletes)
+        self.trie.deletes = []
         self.cache = {}
         self.journal = []
 

ethereum/todo_tests/tst_frontier.py

@@ -1,20 +1,23 @@
-from ethereum import parse_genesis_declaration, db
+from ethereum import genesis_helpers, db
 from ethereum.block import Block, BlockHeader
 from ethereum.config import Env
-import ethereum.state_transition as state_transition
-from ethereum import chain
+import ethereum.messages as messages
+import ethereum.common as common
+from pyethapp.leveldb_service import LevelDB
+from ethereum.pow import chain
 import rlp
 import json
 import os
 import sys
 import time
+import random
 
 # from ethereum.slogging import LogRecorder, configure_logging, set_level
 # config_string = ':info,eth.vm.log:trace,eth.vm.op:trace,eth.vm.stack:trace,eth.vm.exit:trace,eth.pb.msg:trace,eth.pb.tx:debug'
 # configure_logging(config_string=config_string)
 
-state_transition.SKIP_MEDSTATES = True
-state_transition.SKIP_RECEIPT_ROOT_VALIDATION = True
+messages.SKIP_MEDSTATES = True
+common.SKIP_RECEIPT_ROOT_VALIDATION = True
 # assert not state_transition.SKIP_MEDSTATES or state_transition.SKIP_RECEIPT_ROOT_VALIDATION
 
 STATE_LOAD_FN = 'saved_state.json'
@@ -33,55 +36,50 @@
 if '--benchmark' in sys.argv:
     BENCHMARK = int(sys.argv[sys.argv.index('--benchmark') + 1])
 
+DB_DIR = '/tmp/%d' % random.randrange(int(time.time() * 1000000))
+if '--db' in sys.argv:
+    DB_DIR = int(sys.argv[sys.argv.index('--db') + 1])
+
 _path, _file = os.path.split(STATE_LOAD_FN)
 if _file in os.listdir(os.path.join(os.getcwd(), _path)):
-    print 'loading state from %s ...' % STATE_LOAD_FN
+    print('loading state from %s ...' % STATE_LOAD_FN)
     c = chain.Chain(json.load(open(STATE_LOAD_FN)), Env())
-    print 'loaded.'
+    print('loaded.')
 elif 'genesis_frontier.json' not in os.listdir(os.getcwd()):
-    print 'Please download genesis_frontier.json from ' + \
-        'http://vitalik.ca/files/genesis_frontier.json'
+    print('Please download genesis_frontier.json from ' + \
+        'http://vitalik.ca/files/genesis_frontier.json')
     sys.exit()
 else:
-    c = chain.Chain(json.load(open('genesis_frontier.json')), Env())
+    c = chain.Chain(json.load(open('genesis_frontier.json')), Env(LevelDB(DB_DIR)))
     assert c.state.trie.root_hash.encode('hex') == \
         'd7f8974fb5ac78d9ac099b9ad5018bedc2ce0a72dad1827a1709da30580f0544'
     assert c.state.prev_headers[0].hash.encode('hex') == \
         'd4e56740f876aef8c010b86a40d5f56745a118d0906a34e69aec8c0db1cb8fa3'
-    print 'state generated from genesis'
-print 'Attempting to open %s' % RLP_BLOCKS_FILE
+    print('state generated from genesis')
+print('Attempting to open %s' % RLP_BLOCKS_FILE)
 _path, _file = os.path.split(RLP_BLOCKS_FILE)
 if not _path or _file not in os.listdir(_path):
-    print 'Please download 200kblocks.rlp from http://vitalik.ca/files/200kblocks.rlp ' + \
-          'and put it in this directory to continue the test'
+    print('Please download 200kblocks.rlp from http://vitalik.ca/files/200kblocks.rlp ' + \
+          'and put it in this directory to continue the test')
     sys.exit()
 
-batch_size = 1024 * 10240  # approximately 10000 blocks
 f = open(RLP_BLOCKS_FILE)
 
-# skip already processed blocks
-skip = c.state.block_number + 1
-print 'Skipping %d' % skip
-count = 0
-block_rlps = f.readlines(batch_size)
-while len(block_rlps) > 0:
-    if len(block_rlps) + count <= skip:
-        count += len(block_rlps)
-        block_rlps = f.readlines(batch_size)
-    else:
-        block_rlps = block_rlps[skip - count:]
-        count = skip
-        break
-print "skipped %d processed blocks" % skip
-
+pos = 0
+block_source_data = f.read()
+block_rlps = []
+while pos < len(block_source_data):
+    _, l1, l2 = rlp.codec.consume_length_prefix(block_source_data, pos)
+    block_rlps.append(block_source_data[pos: l1+l2])
+    pos = l1 + l2
 
 def report(st, num_blks, num_txs, gas_used):
     now = time.time()
     elapsed = now - st
     tps = num_txs / elapsed
     bps = num_blks / elapsed
     gps = gas_used / elapsed
-    print '%.2f >>> elapsed:%d blocks:%d txs:%d gas:%d bps:%d tps:%d gps:%d' % (now, elapsed, num_blks, num_txs, gas_used, bps, tps, gps)
+    print('%.2f >>> elapsed:%d blocks:%d txs:%d gas:%d bps:%d tps:%d gps:%d' % (now, elapsed, num_blks, num_txs, gas_used, bps, tps, gps))
 
 
 def check_snapshot_consistency(snapshot, env=None):
@@ -98,7 +96,7 @@ def check_snapshot_consistency(snapshot, env=None):
 
 
 def snapshot(c, num_blocks):
-    print 'creating snapshot'
+    print('creating snapshot')
     snapshot = c.state.to_snapshot()
     if (num_blocks / SAVE_INTERVAL) % 2 == 1:
         check_snapshot_consistency(snapshot, env=None)
@@ -114,10 +112,11 @@ def snapshot(c, num_blocks):
     open(fn, 'w').write(json.dumps(snapshot, indent=4))
 
 REPORT_INTERVAL = 1000
-SAVE_INTERVAL = 10 * 1000
-SNAPSHOT_INTERVAL = 100 * 1000
+SAVE_INTERVAL = 999999
+SNAPSHOT_INTERVAL = 999999
 
-MANUAL_SNAPSHOTS = [68000, 68382, 68666, 69000, 909330]
+# MANUAL_SNAPSHOTS = [68000, 68382, 68666, 69000, 909330]
+MANUAL_SNAPSHOTS = []
 
 # don't check pow
 BlockHeader.check_pow = lambda *args: True
@@ -127,33 +126,31 @@ def snapshot(c, num_blocks):
 num_blks = 0
 num_txs = 0
 gas_used = 0
-while len(block_rlps) > 0:
-    for block in block_rlps:
-        # print 'prevh:', s.prev_headers
-        block = rlp.decode(block.strip().decode('hex'), Block)
-        assert c.add_block(block)
-        num_blks += 1
-        num_txs += len(block.transactions)
-        gas_used += block.gas_used
-        if BENCHMARK > 0:
-            report(st, num_blks, num_txs, gas_used)
-            if num_blks == BENCHMARK:
-                print "Benchmark completed (%d blocks)." % num_blks
-                sys.exit()
-        else:
-            num_blocks = block.header.number + 1
-            if num_blocks % REPORT_INTERVAL == 0 or num_blocks in MANUAL_SNAPSHOTS:
-                report(st, REPORT_INTERVAL, num_txs, gas_used)
-                st = time.time()
-                num_blks = 0
-                num_txs = 0
-                gas_used = 0
-            if num_blocks % SAVE_INTERVAL == 0 or num_blocks in MANUAL_SNAPSHOTS:
-                snapshot(c, num_blocks)
-                st = time.time()
-                num_blks = 0
-                num_txs = 0
-                gas_used = 0
-    block_rlps = f.readlines(batch_size)
-
-print 'Test successful'
+for block in block_rlps[1:50000]:
+    # print 'prevh:', s.prev_headers
+    block = rlp.decode(block, Block)
+    assert c.add_block(block)
+    num_blks += 1
+    num_txs += len(block.transactions)
+    gas_used += block.gas_used
+    if BENCHMARK > 0:
+        report(st, num_blks, num_txs, gas_used)
+        if num_blks == BENCHMARK:
+            print("Benchmark completed (%d blocks)." % num_blks)
+            sys.exit()
+    else:
+        num_blocks = block.header.number + 1
+        if num_blocks % REPORT_INTERVAL == 0 or num_blocks in MANUAL_SNAPSHOTS:
+            report(st, REPORT_INTERVAL, num_txs, gas_used)
+            st = time.time()
+            num_blks = 0
+            num_txs = 0
+            gas_used = 0
+        if num_blocks % SAVE_INTERVAL == 0 or num_blocks in MANUAL_SNAPSHOTS:
+            snapshot(c, num_blocks)
+            st = time.time()
+            num_blks = 0
+            num_txs = 0
+            gas_used = 0
+
+print('Test successful')