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

unrestricted batches

Browse source
This commit is contained in:
Victor Shyba 2020-12-21 15:33:06 -03:00
parent 21a2e67755
commit c73f64c424
2 changed files with 13 additions and 6 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

17
lbry/wallet/ledger.py
View file

@ -661,24 +661,31 @@ class Ledger(metaclass=LedgerRegistry):
if not batches[-1]: if not batches[-1]:
batches.pop() batches.pop()
for batch in batches: for routine in asyncio.as_completed([self._single_batch(batch, remote_heights) for batch in batches]):
async for tx in self._single_batch(batch, remote_heights): for tx in await routine:
yield tx yield tx
async def request_synced_transactions(self, to_request, remote_history, address): async def request_synced_transactions(self, to_request, remote_history, address):
pending_sync = [] pending_sync = []
async for tx in self.request_transactions(((txid, height) for txid, height in to_request.values())): async for tx in self.request_transactions(((txid, height) for txid, height in to_request.values())):
pending_sync.append(asyncio.ensure_future(self._sync(tx, remote_history))) pending_sync.append(asyncio.ensure_future(self._sync(tx, remote_history)))
log.info("%d/%d", len(pending_sync), len(to_request))
yield tx yield tx
await asyncio.gather(*pending_sync) await asyncio.gather(*pending_sync)
async def _single_batch(self, batch, remote_heights): async def _single_batch(self, batch, remote_heights):
batch_result = await self.network.retriable_call(self.network.get_transaction_batch, batch) heights = {remote_heights[txid] for txid in batch}
unrestriced = 0 < min(heights) < max(heights) < max(self.headers.checkpoints or [0])
batch_result = await self.network.retriable_call(self.network.get_transaction_batch, batch, not unrestriced)
txs = []
pending_verification = []
for txid, (raw, merkle) in batch_result.items(): for txid, (raw, merkle) in batch_result.items():
remote_height = remote_heights[txid] remote_height = remote_heights[txid]
tx = Transaction(unhexlify(raw), height=remote_height) tx = Transaction(unhexlify(raw), height=remote_height)
await self.maybe_verify_transaction(tx, remote_height, merkle) pending_verification.append(self.maybe_verify_transaction(tx, remote_height, merkle))
yield tx txs.append(tx)
await asyncio.gather(*pending_verification)
return txs
async def _sync(self, tx, remote_history): async def _sync(self, tx, remote_history):
check_db_for_txos = {} check_db_for_txos = {}

2
tests/unit/wallet/test_ledger.py
View file

@ -40,7 +40,7 @@ class MockNetwork:
merkle = await self.get_merkle(tx_hash, known_height) merkle = await self.get_merkle(tx_hash, known_height)
return tx, merkle return tx, merkle
async def get_transaction_batch(self, txids): async def get_transaction_batch(self, txids, restricted):
return { return {
txid: await self.get_transaction_and_merkle(txid) txid: await self.get_transaction_and_merkle(txid)
for txid in txids for txid in txids