LBRYCommunity/lbry-sdk
1
0
Fork 1
Code Issues Pull requests Projects Releases 282 Packages Wiki Activity

faster header download

Browse source
This commit is contained in:
Victor Shyba 2019-09-27 09:09:17 -03:00 committed by Lex Berezhny
parent 817869b915
commit c30db15efa
2 changed files with 17 additions and 2 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

14
torba/torba/client/baseledger.py
View file

@ -285,6 +285,8 @@ class BaseLedger(metaclass=LedgerRegistry):
first_connection = self.network.on_connected.first first_connection = self.network.on_connected.first
asyncio.ensure_future(self.network.start()) asyncio.ensure_future(self.network.start())
await first_connection await first_connection
async with self._header_processing_lock:
await self.initial_headers_sync()
await self.join_network() await self.join_network()
self.network.on_connected.listen(self.join_network) self.network.on_connected.listen(self.join_network)
@ -303,6 +305,18 @@ class BaseLedger(metaclass=LedgerRegistry):
await self.db.close() await self.db.close()
await self.headers.close() await self.headers.close()
async def initial_headers_sync(self):
target = self.network.remote_height
current = len(self.headers)
get_chunk = partial(self.network.retriable_call, self.network.get_headers, count=2000)
chunks = [asyncio.ensure_future(get_chunk(height)) for height in range(current, target, 2000)]
for chunk in chunks:
headers = await chunk
if not headers:
continue
headers = headers['hex']
await self.update_headers(height=len(self.headers), headers=headers, subscription_update=True)
async def update_headers(self, height=None, headers=None, subscription_update=False): async def update_headers(self, height=None, headers=None, subscription_update=False):
rewound = 0 rewound = 0
while True: while True:

5
torba/torba/client/basenetwork.py
View file

@ -173,9 +173,9 @@ class BaseNetwork:
continue continue
self.client = await self.session_pool.wait_for_fastest_session() self.client = await self.session_pool.wait_for_fastest_session()
log.info("Switching to SPV wallet server: %s:%d", *self.client.server) log.info("Switching to SPV wallet server: %s:%d", *self.client.server)
self._on_connected_controller.add(True)
try: try:
self._update_remote_height((await self.subscribe_headers(),)) self._update_remote_height((await self.subscribe_headers(),))
self._on_connected_controller.add(True)
log.info("Subscribed to headers: %s:%d", *self.client.server) log.info("Subscribed to headers: %s:%d", *self.client.server)
except (asyncio.TimeoutError, ConnectionError): except (asyncio.TimeoutError, ConnectionError):
log.info("Switching to %s:%d timed out, closing and retrying.", *self.client.server) log.info("Switching to %s:%d timed out, closing and retrying.", *self.client.server)
@ -240,7 +240,8 @@ class BaseNetwork:
return self.rpc('blockchain.transaction.get_merkle', [tx_hash, height], restricted) return self.rpc('blockchain.transaction.get_merkle', [tx_hash, height], restricted)
def get_headers(self, height, count=10000): def get_headers(self, height, count=10000):
return self.rpc('blockchain.block.headers', [height, count]) restricted = height >= self.remote_height - 100
return self.rpc('blockchain.block.headers', [height, count], restricted)
# --- Subscribes, history and broadcasts are always aimed towards the master client directly # --- Subscribes, history and broadcasts are always aimed towards the master client directly
def get_history(self, address): def get_history(self, address):