diff --git a/server/db.py b/server/db.py index 99c196e..97d558c 100644 --- a/server/db.py +++ b/server/db.py @@ -119,9 +119,14 @@ class UTXOCache(object): hash168s.add(hash168) key = tx_hash + pack('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() + self.wall_time += now - self.last_flush + self.last_flush = now + state = { + 'genesis': self.coin.GENESIS_HASH, + '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 flush(self, daemon_height, flush_utxos=False): + '''Flush out 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 diff --git a/server/env.py b/server/env.py index 5e20586..d17c0a5 100644 --- a/server/env.py +++ b/server/env.py @@ -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): diff --git a/server/server.py b/server/server.py index 1e96080..8efde5f 100644 --- a/server/server.py +++ b/server/server.py @@ -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))