Skip to content

Commit

Permalink
synchronizer: show progress in GUI
Browse files Browse the repository at this point in the history
  • Loading branch information
SomberNight committed May 7, 2019
1 parent e63157c commit 0e6cf15
Show file tree
Hide file tree
Showing 4 changed files with 43 additions and 3 deletions.
6 changes: 6 additions & 0 deletions electrum/address_synchronizer.py
Original file line number Diff line number Diff line change
Expand Up @@ -582,6 +582,12 @@ def set_up_to_date(self, up_to_date):
def is_up_to_date(self):
with self.lock: return self.up_to_date

def get_history_sync_state_details(self) -> Tuple[int, int]:
if self.synchronizer:
return self.synchronizer.num_requests_sent_and_answered()
else:
return 0, 0

@with_transaction_lock
def get_tx_delta(self, tx_hash, address):
"""effect of tx on address"""
Expand Down
13 changes: 12 additions & 1 deletion electrum/gui/kivy/main_window.py
Original file line number Diff line number Diff line change
Expand Up @@ -312,6 +312,9 @@ def __init__(self, **kwargs):
self._trigger_update_status = Clock.create_trigger(self.update_status, .5)
self._trigger_update_history = Clock.create_trigger(self.update_history, .5)
self._trigger_update_interfaces = Clock.create_trigger(self.update_interfaces, .5)

self._periodic_update_status_during_sync = Clock.schedule_interval(self.update_wallet_synchronizing_progress, .5)

# cached dialogs
self._settings_dialog = None
self._password_dialog = None
Expand Down Expand Up @@ -745,7 +748,9 @@ def update_status(self, *dt):
server_height = self.network.get_server_height()
server_lag = self.num_blocks - server_height
if not self.wallet.up_to_date or server_height == 0:
status = _("Synchronizing...")
num_sent, num_answered = self.wallet.get_history_sync_state_details()
status = ("{} [size=18dp]({}/{})[/size]"
.format(_("Synchronizing..."), num_answered, num_sent))
elif server_lag > 1:
status = _("Server is lagging ({} blocks)").format(server_lag)
else:
Expand All @@ -761,6 +766,12 @@ def update_status(self, *dt):
self.balance = str(text.strip()) + ' [size=22dp]%s[/size]'% self.base_unit
self.fiat_balance = self.fx.format_amount(c+u+x) + ' [size=22dp]%s[/size]'% self.fx.ccy

def update_wallet_synchronizing_progress(self, *dt):
if not self.wallet:
return
if not self.wallet.up_to_date:
self._trigger_update_status()

def get_max_amount(self):
from electrum.transaction import TxOutput
if run_hook('abort_send', self):
Expand Down
7 changes: 6 additions & 1 deletion electrum/gui/qt/main_window.py
Original file line number Diff line number Diff line change
Expand Up @@ -740,6 +740,9 @@ def timer_actions(self):
if self.need_update.is_set():
self.need_update.clear()
self.update_wallet()
elif not self.wallet.up_to_date:
# this updates "synchronizing" progress
self.update_status()
# resolve aliases
# FIXME this is a blocking network call that has a timeout of 5 sec
self.payto_e.resolve()
Expand Down Expand Up @@ -822,7 +825,9 @@ def update_status(self):
# until we get a headers subscription request response.
# Display the synchronizing message in that case.
if not self.wallet.up_to_date or server_height == 0:
text = _("Synchronizing...")
num_sent, num_answered = self.wallet.get_history_sync_state_details()
text = ("{} ({}/{})"
.format(_("Synchronizing..."), num_answered, num_sent))
icon = read_QIcon("status_waiting.png")
elif server_lag > 1:
text = _("Server is lagging ({} blocks)").format(server_lag)
Expand Down
20 changes: 19 additions & 1 deletion electrum/synchronizer.py
Original file line number Diff line number Diff line change
Expand Up @@ -24,7 +24,7 @@
# SOFTWARE.
import asyncio
import hashlib
from typing import Dict, List, TYPE_CHECKING
from typing import Dict, List, TYPE_CHECKING, Tuple
from collections import defaultdict

from aiorpcx import TaskGroup, run_in_thread
Expand Down Expand Up @@ -59,12 +59,14 @@ class SynchronizerBase(NetworkJobOnDefaultServer):
def __init__(self, network: 'Network'):
self.asyncio_loop = network.asyncio_loop
NetworkJobOnDefaultServer.__init__(self, network)
self._reset_request_counters()

def _reset(self):
super()._reset()
self.requested_addrs = set()
self.scripthash_to_address = {}
self._processed_some_notifications = False # so that we don't miss them
self._reset_request_counters()
# Queues
self.add_queue = asyncio.Queue()
self.status_queue = asyncio.Queue()
Expand All @@ -79,6 +81,10 @@ async def _start_tasks(self):
# we are being cancelled now
self.session.unsubscribe(self.status_queue)

def _reset_request_counters(self):
self._requests_sent = 0
self._requests_answered = 0

def add(self, addr):
asyncio.run_coroutine_threadsafe(self._add_address(addr), self.asyncio_loop)

Expand All @@ -96,7 +102,9 @@ async def send_subscriptions(self):
async def subscribe_to_address(addr):
h = address_to_scripthash(addr)
self.scripthash_to_address[h] = addr
self._requests_sent += 1
await self.session.subscribe('blockchain.scripthash.subscribe', [h], self.status_queue)
self._requests_answered += 1
self.requested_addrs.remove(addr)

while True:
Expand All @@ -110,6 +118,9 @@ async def handle_status(self):
await self.group.spawn(self._on_address_status, addr, status)
self._processed_some_notifications = True

def num_requests_sent_and_answered(self) -> Tuple[int, int]:
return self._requests_sent, self._requests_answered

async def main(self):
raise NotImplementedError() # implemented by subclasses

Expand Down Expand Up @@ -148,7 +159,9 @@ async def _on_address_status(self, addr, status):
# request address history
self.requested_histories[addr] = status
h = address_to_scripthash(addr)
self._requests_sent += 1
result = await self.network.get_history_for_scripthash(h)
self._requests_answered += 1
self.logger.info(f"receiving history {addr} {len(result)}")
hashes = set(map(lambda item: item['tx_hash'], result))
hist = list(map(lambda item: (item['tx_hash'], item['height']), result))
Expand Down Expand Up @@ -187,6 +200,7 @@ async def _request_missing_txs(self, hist, *, allow_server_not_finding_tx=False)
await group.spawn(self._get_transaction(tx_hash, allow_server_not_finding_tx=allow_server_not_finding_tx))

async def _get_transaction(self, tx_hash, *, allow_server_not_finding_tx=False):
self._requests_sent += 1
try:
result = await self.network.get_transaction(tx_hash)
except UntrustedServerReturnedError as e:
Expand All @@ -196,6 +210,8 @@ async def _get_transaction(self, tx_hash, *, allow_server_not_finding_tx=False):
return
else:
raise
finally:
self._requests_answered += 1
tx = Transaction(result)
try:
tx.deserialize() # see if raises
Expand Down Expand Up @@ -234,6 +250,8 @@ async def main(self):
if (up_to_date != self.wallet.is_up_to_date()
or up_to_date and self._processed_some_notifications):
self._processed_some_notifications = False
if up_to_date:
self._reset_request_counters()
self.wallet.set_up_to_date(up_to_date)
self.wallet.network.trigger_callback('wallet_updated', self.wallet)

Expand Down

0 comments on commit 0e6cf15

Please sign in to comment.