Skip to content

Commit

Permalink
Switch to curio primitives
Browse files Browse the repository at this point in the history
Gives much clearer code
Neil Booth committed Jul 28, 2018

Verified

This commit was created on GitHub.com and signed with GitHub’s verified signature. The key has expired.
1 parent 55ef1ab commit 751f991
Showing 9 changed files with 213 additions and 304 deletions.
16 changes: 9 additions & 7 deletions electrumx/lib/server_base.py
Original file line number Diff line number Diff line change
@@ -12,6 +12,8 @@
import time
from functools import partial

from aiorpcx import TaskGroup

from electrumx.lib.util import class_logger


@@ -93,18 +95,18 @@ def on_signal(signame):
loop.set_exception_handler(self.on_exception)

shutdown_event = asyncio.Event()
task = loop.create_task(self.serve(shutdown_event))
try:
# Wait for shutdown to be signalled, and log it.
await shutdown_event.wait()
self.logger.info('shutting down')
task.cancel()
await task
async with TaskGroup() as group:
server_task = await group.spawn(self.serve(shutdown_event))
# Wait for shutdown, log on receipt of the event
await shutdown_event.wait()
self.logger.info('shutting down')
server_task.cancel()
finally:
await loop.shutdown_asyncgens()

# Prevent some silly logs
await asyncio.sleep(0)
await asyncio.sleep(0.001)
# Finally, work around an apparent asyncio bug that causes log
# spew on shutdown for partially opened SSL sockets
try:
68 changes: 0 additions & 68 deletions electrumx/lib/tasks.py

This file was deleted.

136 changes: 64 additions & 72 deletions electrumx/server/block_processor.py
Original file line number Diff line number Diff line change
@@ -15,6 +15,8 @@
import time
from functools import partial

from aiorpcx import TaskGroup, run_in_thread

import electrumx
from electrumx.server.daemon import DaemonError
from electrumx.lib.hash import hash_to_hex_str, HASHX_LEN
@@ -44,8 +46,9 @@ def __init__(self, daemon, coin, blocks_event):
# This makes the first fetch be 10 blocks
self.ave_size = self.min_cache_size // 10

async def main_loop(self):
async def main_loop(self, bp_height):
'''Loop forever polling for more blocks.'''
await self.reset_height(bp_height)
while True:
try:
# Sleep a while if there is nothing to prefetch
@@ -153,14 +156,12 @@ class BlockProcessor(electrumx.server.db.DB):
Coordinate backing up in case of chain reorganisations.
'''

def __init__(self, env, tasks, daemon, notifications):
def __init__(self, env, daemon, notifications):
super().__init__(env)

self.tasks = tasks
self.daemon = daemon
self.notifications = notifications

self._caught_up_event = asyncio.Event()
self.blocks_event = asyncio.Event()
self.prefetcher = Prefetcher(daemon, env.coin, self.blocks_event)

@@ -187,16 +188,10 @@ def __init__(self, env, tasks, daemon, notifications):
# If the lock is successfully acquired, in-memory chain state
# is consistent with self.height
self.state_lock = asyncio.Lock()
self.worker_task = None

def add_new_block_callback(self, callback):
'''Add a function called when a new block is found.

If several blocks are processed simultaneously, only called
once. The callback is passed a set of hashXs touched by the
block(s), which is cleared on return.
'''
self.callbacks.append(callback)
async def run_in_thread_shielded(self, func, *args):
async with self.state_lock:
return await asyncio.shield(run_in_thread(func, *args))

async def check_and_advance_blocks(self, raw_blocks):
'''Process the list of raw blocks passed. Detects and handles
@@ -212,14 +207,7 @@ async def check_and_advance_blocks(self, raw_blocks):
chain = [self.tip] + [self.coin.header_hash(h) for h in headers[:-1]]

if hprevs == chain:
start = time.time()
async with self.state_lock:
await self.tasks.run_in_thread(self.advance_blocks, blocks)
if not self.first_sync:
s = '' if len(blocks) == 1 else 's'
self.logger.info('processed {:,d} block{} in {:.1f}s'
.format(len(blocks), s,
time.time() - start))
await self.run_in_thread_shielded(self.advance_blocks, blocks)
if self._caught_up_event.is_set():
await self.notifications.on_block(self.touched, self.height)
self.touched = set()
@@ -244,7 +232,7 @@ async def reorg_chain(self, count=None):
self.logger.info('chain reorg detected')
else:
self.logger.info(f'faking a reorg of {count:,d} blocks')
await self.tasks.run_in_thread(self.flush, True)
await run_in_thread(self.flush, True)

async def get_raw_blocks(last_height, hex_hashes):
heights = range(last_height, last_height - len(hex_hashes), -1)
@@ -260,8 +248,7 @@ async def get_raw_blocks(last_height, hex_hashes):
hashes = [hash_to_hex_str(hash) for hash in reversed(hashes)]
for hex_hashes in chunks(hashes, 50):
raw_blocks = await get_raw_blocks(last, hex_hashes)
async with self.state_lock:
await self.tasks.run_in_thread(self.backup_blocks, raw_blocks)
await self.run_in_thread_shielded(self.backup_blocks, raw_blocks)
last -= len(raw_blocks)
# Truncate header_mc: header count is 1 more than the height.
# Note header_mc is None if the reorg happens at startup.
@@ -468,6 +455,7 @@ def advance_blocks(self, blocks):
It is already verified they correctly connect onto our tip.
'''
start = time.time()
min_height = self.min_undo_height(self.daemon.cached_height())
height = self.height

@@ -492,6 +480,12 @@ def advance_blocks(self, blocks):
self.check_cache_size()
self.next_cache_check = time.time() + 30

if not self.first_sync:
s = '' if len(blocks) == 1 else 's'
self.logger.info('processed {:,d} block{} in {:.1f}s'
.format(len(blocks), s,
time.time() - start))

def advance_txs(self, txs):
self.tx_hashes.append(b''.join(tx_hash for tx, tx_hash in txs))

@@ -744,20 +738,13 @@ def flush_utxos(self, batch):
self.db_height = self.height
self.db_tip = self.tip

async def _process_blocks(self):
async def _process_prefetched_blocks(self):
'''Loop forever processing blocks as they arrive.'''
while True:
if self.height == self.daemon.cached_height():
if not self._caught_up_event.is_set():
self.logger.info(f'caught up to height {self.height}')
await self._first_caught_up()
self._caught_up_event.set()
# Flush everything but with first_sync->False state.
first_sync = self.first_sync
self.first_sync = False
self.flush(True)
if first_sync:
self.logger.info(f'{electrumx.version} synced to '
f'height {self.height:,d}')
await self.blocks_event.wait()
self.blocks_event.clear()
if self.reorg_count:
@@ -767,7 +754,26 @@ async def _process_blocks(self):
blocks = self.prefetcher.get_prefetched_blocks()
await self.check_and_advance_blocks(blocks)

def _on_dbs_opened(self):
async def _first_caught_up(self):
self.logger.info(f'caught up to height {self.height}')
# Flush everything but with first_sync->False state.
first_sync = self.first_sync
self.first_sync = False
self.flush(True)
if first_sync:
self.logger.info(f'{electrumx.version} synced to '
f'height {self.height:,d}')
# Initialise the notification framework
await self.notifications.on_block(set(), self.height)
# Reopen for serving
await self.open_for_serving()
# Populate the header merkle cache
length = max(1, self.height - self.env.reorg_limit)
self.header_mc = MerkleCache(self.merkle, HeaderSource(self), length)
self.logger.info('populated header merkle cache')

async def _first_open_dbs(self):
await self.open_for_sync()
# An incomplete compaction needs to be cancelled otherwise
# restarting it will corrupt the history
self.history.cancel_compaction()
@@ -783,31 +789,32 @@ def _on_dbs_opened(self):

# --- External API

async def catch_up_to_daemon(self):
'''Process and index blocks until we catch up with the daemon.
async def fetch_and_process_blocks(self, caught_up_event):
'''Fetch, process and index blocks from the daemon.
Returns once caught up. Future blocks continue to be
processed in a separate task.
'''
# Open the databases first.
await self.open_for_sync()
self._on_dbs_opened()
# Get the prefetcher running
self.tasks.create_task(self.prefetcher.main_loop())
await self.prefetcher.reset_height(self.height)
# Start our loop that processes blocks as they are fetched
self.worker_task = self.tasks.create_task(self._process_blocks())
# Wait until caught up
await self._caught_up_event.wait()
# Initialise the notification framework
await self.notifications.on_block(set(), self.height)
# Reopen for serving
await self.open_for_serving()
Sets caught_up_event when first caught up. Flushes to disk
and shuts down cleanly if cancelled.
# Populate the header merkle cache
length = max(1, self.height - self.env.reorg_limit)
self.header_mc = MerkleCache(self.merkle, HeaderSource(self), length)
self.logger.info('populated header merkle cache')
This is mainly because if, during initial sync ElectrumX is
asked to shut down when a large number of blocks have been
processed but not written to disk, it should write those to
disk before exiting, as otherwise a significant amount of work
could be lost.
'''
self._caught_up_event = caught_up_event
async with TaskGroup() as group:
await group.spawn(self._first_open_dbs())
# Ensure cached_height is set
await group.spawn(self.daemon.height())
try:
async with TaskGroup() as group:
await group.spawn(self.prefetcher.main_loop(self.height))
await group.spawn(self._process_prefetched_blocks())
finally:
async with self.state_lock:
# Shut down block processing
self.logger.info('flushing to DB for a clean shutdown...')
self.flush(True)

def force_chain_reorg(self, count):
'''Force a reorg of the given number of blocks.
@@ -819,18 +826,3 @@ def force_chain_reorg(self, count):
self.blocks_event.set()
return True
return False

async def shutdown(self):
'''Shutdown cleanly and flush to disk.
If during initial sync ElectrumX is asked to shut down when a
large number of blocks have been processed but not written to
disk, it should write those to disk before exiting, as
otherwise a significant amount of work could be lost.
'''
if self.worker_task:
async with self.state_lock:
# Shut down block processing
self.worker_task.cancel()
self.logger.info('flushing to DB for a clean shutdown...')
self.flush(True)
13 changes: 5 additions & 8 deletions electrumx/server/chain_state.py
Original file line number Diff line number Diff line change
@@ -9,15 +9,16 @@
import asyncio
import pylru

from aiorpcx import run_in_thread


class ChainState(object):
'''Used as an interface by servers to request information about
blocks, transaction history, UTXOs and the mempool.
'''

def __init__(self, env, tasks, daemon, bp, notifications):
def __init__(self, env, daemon, bp, notifications):
self._env = env
self._tasks = tasks
self._daemon = daemon
self._bp = bp
self._history_cache = pylru.lrucache(256)
@@ -64,15 +65,15 @@ def job():

hc = self._history_cache
if hashX not in hc:
hc[hashX] = await self._tasks.run_in_thread(job)
hc[hashX] = await run_in_thread(job)
return hc[hashX]

async def get_utxos(self, hashX):
'''Get UTXOs asynchronously to reduce latency.'''
def job():
return list(self._bp.get_utxos(hashX, limit=None))

return await self._tasks.run_in_thread(job)
return await run_in_thread(job)

def header_branch_and_root(self, length, height):
return self._bp.header_mc.branch_and_root(length, height)
@@ -91,7 +92,3 @@ def raw_header(self, height):
def set_daemon_url(self, daemon_url):
self._daemon.set_urls(self._env.coin.daemon_urls(daemon_url))
return self._daemon.logged_url()

async def shutdown(self):
'''Shut down the block processor to flush chain state to disk.'''
await self._bp.shutdown()
Loading

0 comments on commit 751f991

Please sign in to comment.