lukechilds
/
electrumx
mirror of https://github.com/lukechilds/electrumx.git
1
0
Fork 0
Code Issues Projects Releases Wiki Activity
Browse Source

Merge branch 'hist_sep' into develop

master
Neil Booth 8 years ago
parent
21b25741c0 34096a02e9
commit
75753e8005
3 changed files with 106 additions and 50 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Split View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 147
      server/db.py
  2. 1
      server/env.py
  3. 8
      server/server.py

147
server/db.py
View File

@ -119,9 +119,14 @@ class UTXOCache(object):
hash168s.add(hash168)
key = tx_hash + pack('<H', idx)
if key in self.cache:
self.logger.info('duplicate tx hash {}'
.format(bytes(reversed(tx_hash)).hex()))
# Well-known duplicate coinbases from heights 91722-91880
# that destoyed 100 BTC forever:
# e3bf3d07d4b0375638d5f1db5255fe07ba2c4cb067cd81b84ee974b6585fb468
# d5d27987d2a3dfc724e359870c6644b40e497bdc0589a033220fe15429d88599
#if key in self.cache:
# self.logger.info('duplicate tx hash {}'
# .format(bytes(reversed(tx_hash)).hex()))
# b''.join avoids this: https://bugs.python.org/issue13298
self.cache[key] = b''.join(
@ -286,21 +291,24 @@ class DB(object):
# Meta
self.tx_hash_file_size = 16 * 1024 * 1024
self.cache_MB = env.cache_MB
self.hist_MB = env.hist_MB
self.next_cache_check = 0
self.last_flush = time.time()
self.last_flush_tx_count = 0
self.coin = env.coin
# Chain state (initialize to genesis in case of new DB)
self.height = -1
self.tx_count = 0
self.db_height = -1
self.db_tx_count = 0
self.flush_count = 0
self.utxo_flush_count = 0
self.wall_time = 0
self.tip = self.coin.GENESIS_HASH
# Open DB and metadata files. Record some of its state.
self.db = self.open_db(self.coin)
self.db_tx_count = self.tx_count
self.db_height = self.height
self.tx_count = self.db_tx_count
self.height = self.db_height
# Caches to be flushed later. Headers and tx_hashes have one
# entry per block
@ -310,17 +318,23 @@ class DB(object):
self.history_size = 0
self.utxo_cache = UTXOCache(self, self.db, self.coin)
self.tx_counts = array.array('I')
self.txcount_file.seek(0)
self.tx_counts.fromfile(self.txcount_file, self.height + 1)
# Log state
self.logger.info('{}/{} height: {:,d} tx count: {:,d} '
'flush count: {:,d} sync time: {}'
'flush count: {:,d} utxo flush count: {:,d} '
'sync time: {}'
.format(self.coin.NAME, self.coin.NET, self.height,
self.tx_count, self.flush_count,
self.utxo_flush_count,
formatted_time(self.wall_time)))
self.logger.info('flushing after cache reaches {:,d} MB'
self.logger.info('flushing all after cache reaches {:,d} MB'
.format(self.cache_MB))
self.logger.info('flushing history cache at {:,d} MB'
.format(self.hist_MB))
def open_db(self, coin):
db_name = '{}-{}'.format(coin.NAME, coin.NET)
@ -339,12 +353,56 @@ class DB(object):
else:
self.logger.info('successfully opened database {}'.format(db_name))
self.read_state(db)
self.delete_excess_history(db)
self.headers_file = self.open_file('headers', is_new)
self.txcount_file = self.open_file('txcount', is_new)
return db
def read_state(self, db):
state = db.get(b'state')
state = ast.literal_eval(state.decode('ascii'))
if state['genesis'] != self.coin.GENESIS_HASH:
raise self.Error('DB genesis hash {} does not match coin {}'
.format(state['genesis_hash'],
self.coin.GENESIS_HASH))
self.db_height = state['height']
self.db_tx_count = state['tx_count']
self.tip = state['tip']
self.flush_count = state['flush_count']
self.utxo_flush_count = state['utxo_flush_count']
self.wall_time = state['wall_time']
self.last_flush_tx_count = self.db_tx_count
self.last_flush = time.time()
def delete_excess_history(self, db):
'''Clear history flushed since the most recent UTXO flush.'''
utxo_flush_count = self.utxo_flush_count
diff = self.flush_count - utxo_flush_count
if diff == 0:
return
if diff < 0:
raise self.Error('DB corrupt: flush_count < utxo_flush_count')
self.logger.info('DB not shut down cleanly. Scanning for most '
'recent {:,d} history flushes'.format(diff))
prefix = b'H'
unpack = struct.unpack
keys = []
for key, hist in db.iterator(prefix=prefix):
flush_id, = unpack('>H', key[-2:])
if flush_id > self.utxo_flush_count:
keys.append(key)
self.logger.info('deleting {:,d} history entries'.format(len(keys)))
with db.write_batch(transaction=True) as batch:
for key in keys:
db.delete(key)
self.utxo_flush_count = self.flush_count
self.flush_state(batch)
self.logger.info('deletion complete')
def flush_state(self, batch):
'''Flush chain state to the batch.'''
now = time.time()
@ -352,52 +410,46 @@ class DB(object):
self.last_flush = now
state = {
'genesis': self.coin.GENESIS_HASH,
'height': self.height,
'tx_count': self.tx_count,
'height': self.db_height,
'tx_count': self.db_tx_count,
'tip': self.tip,
'flush_count': self.flush_count,
'utxo_flush_count': self.utxo_flush_count,
'wall_time': self.wall_time,
}
batch.put(b'state', repr(state).encode('ascii'))
def read_state(self, db):
state = db.get(b'state')
state = ast.literal_eval(state.decode('ascii'))
if state['genesis'] != self.coin.GENESIS_HASH:
raise self.Error('DB genesis hash {} does not match coin {}'
.format(state['genesis_hash'],
self.coin.GENESIS_HASH))
self.height = state['height']
self.tx_count = state['tx_count']
self.tip = state['tip']
self.flush_count = state['flush_count']
self.wall_time = state['wall_time']
def flush(self, daemon_height, flush_utxos=False):
'''Flush out cached state.
def flush_all(self, daemon_height):
'''Flush out all cached state.'''
History is always flushed. UTXOs are flushed if flush_utxos.'''
flush_start = time.time()
last_flush = self.last_flush
tx_diff = self.tx_count - self.db_tx_count
height_diff = self.height - self.db_height
self.logger.info('starting flush of {:,d} transactions, {:,d} blocks'
.format(tx_diff, height_diff))
# Write out the files to the FS before flushing to the DB. If
# the DB transaction fails, the files being too long doesn't
# matter. But if writing the files fails we do not want to
# have updated the DB. Flush state last as it reads the wall
# time.
self.flush_to_fs()
if flush_utxos:
# Write out the files to the FS before flushing to the DB.
# If the DB transaction fails, the files being too long
# doesn't matter. But if writing the files fails we do
# not want to have updated the DB. Flush state last as it
# reads the wall time.
self.logger.info('flushing UTXOs: {:,d} txs and {:,d} blocks'
.format(self.tx_count - self.db_tx_count,
self.height - self.db_height))
self.flush_to_fs()
else:
self.logger.info('commencing history flush')
with self.db.write_batch(transaction=True) as batch:
self.utxo_cache.flush(batch)
# History first - fast and frees memory
self.flush_history(batch)
if flush_utxos:
self.utxo_cache.flush(batch)
self.utxo_flush_count = self.flush_count
self.db_tx_count = self.tx_count
self.db_height = self.height
self.flush_state(batch)
self.logger.info('committing transaction...')
# The flush succeeded, so update our record of DB state
self.db_tx_count = self.tx_count
self.db_height = self.height
# Update and put the wall time again - otherwise we drop the
# time it took leveldb to commit the batch
self.flush_state(self.db)
@ -407,6 +459,8 @@ class DB(object):
.format(self.flush_count, self.height, flush_time))
# Log handy stats
tx_diff = self.tx_count - self.last_flush_tx_count
self.last_flush_tx_count = self.tx_count
txs_per_sec = int(self.tx_count / self.wall_time)
this_txs_per_sec = 1 + int(tx_diff / (self.last_flush - last_flush))
if self.height > self.coin.TX_COUNT_HEIGHT:
@ -493,7 +547,7 @@ class DB(object):
file_pos += size
self.tx_hashes = []
def cache_size(self, daemon_height):
def cache_sizes(self, daemon_height):
'''Returns the approximate size of the cache, in MB.'''
# Good average estimates based on traversal of subobjects and
# requesting size from Python (see deep_getsizeof). For
@ -510,14 +564,14 @@ class DB(object):
self.logger.info('cache stats at height {:,d} daemon height: {:,d}'
.format(self.height, daemon_height))
self.logger.info(' entries: UTXO: {:,d} DB: {:,d} '
'hist count: {:,d} hist size: {:,d}'
'hist addrs: {:,d} hist size: {:,d}'
.format(len(self.utxo_cache.cache),
len(self.utxo_cache.db_cache),
len(self.history),
self.history_size))
self.logger.info(' size: {:,d}MB (UTXOs {:,d}MB hist {:,d}MB)'
.format(cache_MB, utxo_MB, hist_MB))
return cache_MB
return cache_MB, hist_MB
def process_block(self, block, daemon_height):
self.headers.append(block[:self.coin.HEADER_LEN])
@ -539,8 +593,9 @@ class DB(object):
now = time.time()
if now > self.next_cache_check:
self.next_cache_check = now + 60
if self.cache_size(daemon_height) > self.cache_MB:
self.flush_all(daemon_height)
cache_MB, hist_MB = self.cache_sizes(daemon_height)
if cache_MB >= self.cache_MB or hist_MB >= self.hist_MB:
self.flush(daemon_height, cache_MB >= self.cache_MB)
def process_tx(self, tx_hash, tx):
cache = self.utxo_cache

1
server/env.py
View File

@ -21,6 +21,7 @@ class Env(object):
self.coin = Coin.lookup_coin_class(coin_name, network)
self.db_dir = self.required('DB_DIRECTORY')
self.cache_MB = self.integer('CACHE_MB', 1000)
self.hist_MB = self.integer('HIST_MB', 250)
self.rpc_url = self.build_rpc_url()
def default(self, envvar, default):

8
server/server.py
View File

@ -79,7 +79,7 @@ class BlockCache(object):
self.logger.info('caught up to height {:d}'
.format(self.daemon_height))
finally:
self.db.flush_all(self.daemon_height)
self.db.flush(self.daemon_height, True)
def cache_used(self):
return sum(len(block) for block in self.blocks)
@ -163,11 +163,11 @@ class BlockCache(object):
if not any(errs):
return tuple(item['result'] for item in result)
if any(err.get('code') == -28 for err in errs):
msg = 'daemon still warming up...'
secs = 10
msg = 'daemon still warming up.'
secs = 30
else:
msg = 'daemon errors: {}'.format(errs)
secs = 1
secs = 3
self.logger.error('{}. Sleeping {:d}s and trying again...'
.format(msg, secs))

Write Preview
Loading…
Cancel
Save