forked from bitcoinafterlife/bal-electrum-plugin
feat(net): parallel Check + timeout counters + GUI tooltips/order
Networking (anti-freeze):
- Add check_transactions_parallel: pressing "Check" now contacts will-executors
concurrently with a fast-fail timeout (CHECK_TIMEOUT=8, 1 retry) and a global
deadline (CHECK_GLOBAL_DEADLINE=30), so a single dead server no longer freezes
the "checking transaction" dialog for ~140s (old default 10s x 10 retries).
- check_transaction now accepts timeout/max_retries/retry_sleep kwargs.
- Rewrite BalWindow.check_transactions_task to use the parallel helper with live
progress + an elapsed-time counter "Checking transactions: 2/5 (4s / 30s)".
Reliable elapsed-time counters (Xs / DEADLINEs):
- Replace the unreliable raw heartbeat thread (whose pyqtSignal emissions were
not marshalled and never repainted) with an on_tick callback driven from the
CALLING thread in push/ping/check parallel helpers.
- Show the maximum wait too (e.g. "3s / 30s") so the user knows when the
operation will give up, in the wizard Broadcasting, Ping and Check dialogs.
- Expose networking constants (PUSH_*/CHECK_*/DEFAULT_TIMEOUT) as Willexecutors
class attributes for a single source of truth in the GUI.
GUI:
- Add hover tooltips: Wizard ("Wizard - Build your will"), Delivery time (truck),
Check Alive (siren), Calendar, Check (refresh).
- Reorder the Will toolbar to: Wizard | Delivery time | Check Alive | Calendar |
Check; tighten layout margins so it all fits the Will window.
Tests:
- parallel_ping_test.py: add coverage for check_transactions_parallel (parallel
timing, global deadline, on_tick from the calling thread) and static checks
that check_transactions_task/loop_push use the parallel helpers + on_tick
counter with the 'Xs / Ns' format.
Verified: ruff (0 new issues), 182 official tests pass, parallel/smoke/gui_fixes/
windows_overflow/external_zip tests pass.
This commit is contained in:
committed by
steal
parent
fbe94506f8
commit
13259e881d
@@ -25,7 +25,36 @@ from electrum.network import Network
|
|||||||
|
|
||||||
from .plugin_base import BalPlugin
|
from .plugin_base import BalPlugin
|
||||||
|
|
||||||
|
# Per-request timeout (seconds) for interactive operations (ping / info /
|
||||||
|
# list download). These fail fast (no retries) so a dead server does not
|
||||||
|
# block the UI.
|
||||||
DEFAULT_TIMEOUT = 5
|
DEFAULT_TIMEOUT = 5
|
||||||
|
|
||||||
|
# Broadcast (pushtxs) timeouts. Broadcasting a will is important, so we keep a
|
||||||
|
# couple of quick retries to survive a transient hiccup -- but far from the old
|
||||||
|
# 10s x 10 retries + 30s sleeps (~140s) that froze the wizard on a dead server.
|
||||||
|
# Worst case per server is now ~ PUSH_TIMEOUT * (1 + PUSH_MAX_RETRIES)
|
||||||
|
# + PUSH_RETRY_SLEEP * PUSH_MAX_RETRIES = 8 * 3 + 1 * 2 = ~26s, and the wizard
|
||||||
|
# also enforces a global deadline on top of this (see push_transactions_parallel).
|
||||||
|
PUSH_TIMEOUT = 8
|
||||||
|
PUSH_MAX_RETRIES = 2
|
||||||
|
PUSH_RETRY_SLEEP = 1
|
||||||
|
|
||||||
|
# Global wall-clock deadline (seconds) for the whole parallel broadcast. Once
|
||||||
|
# it elapses we stop waiting for the still-pending servers, mark them as
|
||||||
|
# "Timeout" and let the wizard proceed instead of appearing stuck.
|
||||||
|
PUSH_GLOBAL_DEADLINE = 30
|
||||||
|
|
||||||
|
# Check (searchtx) timeouts. Used when the user presses "Check" to verify that
|
||||||
|
# each will-executor still holds the transaction. Like the broadcast path, the
|
||||||
|
# old defaults (10s x 10 retries + 30s sleeps ~= 140s per server) froze the
|
||||||
|
# "checking transaction" dialog on a single dead server. Fail fast with one
|
||||||
|
# quick retry, and cap the whole batch with a global deadline.
|
||||||
|
CHECK_TIMEOUT = 8
|
||||||
|
CHECK_MAX_RETRIES = 1
|
||||||
|
CHECK_RETRY_SLEEP = 1
|
||||||
|
CHECK_GLOBAL_DEADLINE = 30
|
||||||
|
|
||||||
_logger = get_logger(__name__)
|
_logger = get_logger(__name__)
|
||||||
|
|
||||||
|
|
||||||
@@ -34,6 +63,20 @@ chainname = BalPlugin.chainname
|
|||||||
|
|
||||||
class Willexecutors:
|
class Willexecutors:
|
||||||
|
|
||||||
|
# Expose the networking constants as class attributes so the GUI layer can
|
||||||
|
# reference them (e.g. to show the "Xs / DEADLINEs" countdown) without
|
||||||
|
# importing module-level names. Single source of truth: the module
|
||||||
|
# constants defined above.
|
||||||
|
DEFAULT_TIMEOUT = DEFAULT_TIMEOUT
|
||||||
|
PUSH_TIMEOUT = PUSH_TIMEOUT
|
||||||
|
PUSH_MAX_RETRIES = PUSH_MAX_RETRIES
|
||||||
|
PUSH_RETRY_SLEEP = PUSH_RETRY_SLEEP
|
||||||
|
PUSH_GLOBAL_DEADLINE = PUSH_GLOBAL_DEADLINE
|
||||||
|
CHECK_TIMEOUT = CHECK_TIMEOUT
|
||||||
|
CHECK_MAX_RETRIES = CHECK_MAX_RETRIES
|
||||||
|
CHECK_RETRY_SLEEP = CHECK_RETRY_SLEEP
|
||||||
|
CHECK_GLOBAL_DEADLINE = CHECK_GLOBAL_DEADLINE
|
||||||
|
|
||||||
@staticmethod
|
@staticmethod
|
||||||
def save(bal_plugin, willexecutors):
|
def save(bal_plugin, willexecutors):
|
||||||
_logger.debug(f"save {willexecutors},{chainname}")
|
_logger.debug(f"save {willexecutors},{chainname}")
|
||||||
@@ -241,7 +284,15 @@ class Willexecutors:
|
|||||||
pass
|
pass
|
||||||
|
|
||||||
@staticmethod
|
@staticmethod
|
||||||
def push_transactions_to_willexecutor(willexecutor):
|
def push_transactions_to_willexecutor(
|
||||||
|
willexecutor, *, timeout=PUSH_TIMEOUT, max_retries=PUSH_MAX_RETRIES,
|
||||||
|
retry_sleep=PUSH_RETRY_SLEEP,
|
||||||
|
):
|
||||||
|
# ``timeout`` / ``max_retries`` / ``retry_sleep`` are forwarded to
|
||||||
|
# send_request so the broadcast fails fast on a dead/slow server instead
|
||||||
|
# of hanging for ~140s (the old default was 10s timeout x 10 retries +
|
||||||
|
# 30s of sleeps). A small number of quick retries still protects
|
||||||
|
# against a transient hiccup without freezing the wizard.
|
||||||
out = True
|
out = True
|
||||||
try:
|
try:
|
||||||
_logger.debug(f"{willexecutor['url']}: {willexecutor['txs']}")
|
_logger.debug(f"{willexecutor['url']}: {willexecutor['txs']}")
|
||||||
@@ -249,6 +300,9 @@ class Willexecutors:
|
|||||||
"post",
|
"post",
|
||||||
willexecutor["url"] + "/" + chainname + "/pushtxs",
|
willexecutor["url"] + "/" + chainname + "/pushtxs",
|
||||||
data=willexecutor["txs"].encode("ascii"),
|
data=willexecutor["txs"].encode("ascii"),
|
||||||
|
timeout=timeout,
|
||||||
|
max_retries=max_retries,
|
||||||
|
retry_sleep=retry_sleep,
|
||||||
):
|
):
|
||||||
willexecutor["broadcast_status"] = _("Success")
|
willexecutor["broadcast_status"] = _("Success")
|
||||||
_logger.debug(f"pushed: {w}")
|
_logger.debug(f"pushed: {w}")
|
||||||
@@ -304,7 +358,8 @@ class Willexecutors:
|
|||||||
|
|
||||||
@staticmethod
|
@staticmethod
|
||||||
def ping_servers_parallel(willexecutors, *, on_each=None, max_workers=8,
|
def ping_servers_parallel(willexecutors, *, on_each=None, max_workers=8,
|
||||||
timeout=DEFAULT_TIMEOUT):
|
timeout=DEFAULT_TIMEOUT, on_tick=None,
|
||||||
|
tick_interval=1.0):
|
||||||
"""Ping every will-executor concurrently and report results as they
|
"""Ping every will-executor concurrently and report results as they
|
||||||
arrive.
|
arrive.
|
||||||
|
|
||||||
@@ -324,10 +379,16 @@ class Willexecutors:
|
|||||||
max_workers: maximum number of concurrent pings.
|
max_workers: maximum number of concurrent pings.
|
||||||
timeout: per-request timeout in seconds (fast-fail, no retries).
|
timeout: per-request timeout in seconds (fast-fail, no retries).
|
||||||
|
|
||||||
|
on_tick: optional ``callback()`` invoked periodically (every
|
||||||
|
``tick_interval`` seconds) **from the calling thread** while
|
||||||
|
waiting for servers, so a Qt caller can refresh an elapsed-time
|
||||||
|
counter from the same thread that drives ``on_each``.
|
||||||
|
|
||||||
Returns:
|
Returns:
|
||||||
The same ``willexecutors`` mapping, updated in place.
|
The same ``willexecutors`` mapping, updated in place.
|
||||||
"""
|
"""
|
||||||
from concurrent.futures import ThreadPoolExecutor, as_completed
|
from concurrent.futures import ThreadPoolExecutor, wait
|
||||||
|
from concurrent.futures import FIRST_COMPLETED
|
||||||
|
|
||||||
items = list(willexecutors.items())
|
items = list(willexecutors.items())
|
||||||
if not items:
|
if not items:
|
||||||
@@ -340,14 +401,28 @@ class Willexecutors:
|
|||||||
ok = we.get("status") == 200
|
ok = we.get("status") == 200
|
||||||
return url, we, ok
|
return url, we, ok
|
||||||
|
|
||||||
|
def _fire_tick():
|
||||||
|
if on_tick is not None:
|
||||||
|
try:
|
||||||
|
on_tick()
|
||||||
|
except Exception as cb_err:
|
||||||
|
_logger.error(f"ping on_tick callback error: {cb_err}")
|
||||||
|
|
||||||
workers = max(1, min(max_workers, len(items)))
|
workers = max(1, min(max_workers, len(items)))
|
||||||
with ThreadPoolExecutor(max_workers=workers,
|
# Manual pool (no ``with``) so we can poll futures in short slices and
|
||||||
thread_name_prefix="bal-ping") as pool:
|
# drive ``on_tick`` from THIS thread between waits (reliable Qt repaint).
|
||||||
futures = [pool.submit(_ping_one, url, we) for url, we in items]
|
pool = ThreadPoolExecutor(max_workers=workers, thread_name_prefix="bal-ping")
|
||||||
for fut in as_completed(futures):
|
futures = {pool.submit(_ping_one, url, we) for url, we in items}
|
||||||
|
try:
|
||||||
|
pending = set(futures)
|
||||||
|
while pending:
|
||||||
|
done, pending = wait(
|
||||||
|
pending, timeout=tick_interval, return_when=FIRST_COMPLETED
|
||||||
|
)
|
||||||
|
for fut in done:
|
||||||
try:
|
try:
|
||||||
url, we, ok = fut.result()
|
url, we, ok = fut.result()
|
||||||
except Exception as e: # defensive: never let one server crash all
|
except Exception as e: # defensive: one server never crashes all
|
||||||
_logger.error(f"ping_servers_parallel worker error: {e}")
|
_logger.error(f"ping_servers_parallel worker error: {e}")
|
||||||
continue
|
continue
|
||||||
willexecutors[url] = we
|
willexecutors[url] = we
|
||||||
@@ -356,22 +431,45 @@ class Willexecutors:
|
|||||||
on_each(url, we, ok)
|
on_each(url, we, ok)
|
||||||
except Exception as cb_err:
|
except Exception as cb_err:
|
||||||
_logger.error(f"ping on_each callback error: {cb_err}")
|
_logger.error(f"ping on_each callback error: {cb_err}")
|
||||||
|
# Drive the elapsed-time counter from the calling thread.
|
||||||
|
_fire_tick()
|
||||||
|
finally:
|
||||||
|
try:
|
||||||
|
pool.shutdown(wait=False, cancel_futures=True)
|
||||||
|
except TypeError:
|
||||||
|
pool.shutdown(wait=False)
|
||||||
return willexecutors
|
return willexecutors
|
||||||
|
|
||||||
@staticmethod
|
@staticmethod
|
||||||
def push_transactions_parallel(willexecutors, *, on_each=None, max_workers=8):
|
def push_transactions_parallel(willexecutors, *, on_each=None, max_workers=8,
|
||||||
|
deadline=PUSH_GLOBAL_DEADLINE, on_timeout=None,
|
||||||
|
on_tick=None, tick_interval=1.0):
|
||||||
"""Push transactions to multiple will-executors concurrently.
|
"""Push transactions to multiple will-executors concurrently.
|
||||||
|
|
||||||
Like :meth:`ping_servers_parallel` but for the ``pushtxs`` operation.
|
Like :meth:`ping_servers_parallel` but for the ``pushtxs`` operation.
|
||||||
Each server keeps the historical retry behaviour of
|
Each server keeps a short retry behaviour
|
||||||
:meth:`push_transactions_to_willexecutor` (which is important so a real
|
(:meth:`push_transactions_to_willexecutor`) so a real transaction is not
|
||||||
transaction is not lost to a transient hiccup), but the servers are now
|
lost to a transient hiccup, but servers are contacted in parallel and
|
||||||
contacted in parallel instead of one-after-another, and results are
|
results are reported via ``on_each(url, we_dict, ok, exc)`` as they
|
||||||
reported via ``on_each(url, we_dict, ok, exc)`` as they complete.
|
complete.
|
||||||
|
|
||||||
Returns ``{url: (ok, exception_or_None)}``.
|
A global wall-clock ``deadline`` (seconds) caps the whole operation: if
|
||||||
|
some servers are still pending when it elapses, we stop waiting, mark
|
||||||
|
them via ``on_timeout(url, we_dict)`` and return, so the caller (the
|
||||||
|
wizard) is never stuck behind one unresponsive server. Pass
|
||||||
|
``deadline=None`` to wait indefinitely (old behaviour).
|
||||||
|
|
||||||
|
``on_tick()`` is invoked periodically (every ``tick_interval`` seconds)
|
||||||
|
**from the calling thread** while waiting for workers. This lets a Qt
|
||||||
|
caller refresh an elapsed-time counter from the same thread that drives
|
||||||
|
``on_each`` (so its pyqtSignal repaints reliably), instead of relying on
|
||||||
|
a separate heartbeat thread whose signal emissions are not marshalled.
|
||||||
|
|
||||||
|
Returns ``{url: (ok, exception_or_None)}`` for the servers that
|
||||||
|
answered in time (timed-out servers are reported via ``on_timeout``).
|
||||||
"""
|
"""
|
||||||
from concurrent.futures import ThreadPoolExecutor, as_completed
|
from concurrent.futures import ThreadPoolExecutor, wait
|
||||||
|
from concurrent.futures import FIRST_COMPLETED
|
||||||
|
|
||||||
targets = [(url, we) for url, we in willexecutors.items() if "txs" in we]
|
targets = [(url, we) for url, we in willexecutors.items() if "txs" in we]
|
||||||
results = {}
|
results = {}
|
||||||
@@ -387,15 +485,48 @@ class Willexecutors:
|
|||||||
except Exception as e:
|
except Exception as e:
|
||||||
return url, we, False, e
|
return url, we, False, e
|
||||||
|
|
||||||
|
def _fire_tick():
|
||||||
|
if on_tick is not None:
|
||||||
|
try:
|
||||||
|
on_tick()
|
||||||
|
except Exception as cb_err:
|
||||||
|
_logger.error(f"push on_tick callback error: {cb_err}")
|
||||||
|
|
||||||
workers = max(1, min(max_workers, len(targets)))
|
workers = max(1, min(max_workers, len(targets)))
|
||||||
with ThreadPoolExecutor(max_workers=workers,
|
# NOTE: we do not use ``with ThreadPoolExecutor(...)`` here because its
|
||||||
thread_name_prefix="bal-push") as pool:
|
# __exit__ calls shutdown(wait=True), which would block on a hung worker
|
||||||
futures = [pool.submit(_push_one, url, we) for url, we in targets]
|
# and defeat the whole point of the global deadline. We shut the pool
|
||||||
for fut in as_completed(futures):
|
# down without waiting once the deadline elapses; the daemon worker(s)
|
||||||
|
# stuck on a dead socket will be torn down when their request finally
|
||||||
|
# times out (PUSH_TIMEOUT), without holding up the wizard.
|
||||||
|
pool = ThreadPoolExecutor(max_workers=workers, thread_name_prefix="bal-push")
|
||||||
|
fut_to_url = {pool.submit(_push_one, url, we): (url, we)
|
||||||
|
for url, we in targets}
|
||||||
|
start = time.time()
|
||||||
|
try:
|
||||||
|
# Poll the futures in short slices so we can call ``on_tick`` from
|
||||||
|
# THIS thread between waits. ``wait(..., timeout=tick_interval)``
|
||||||
|
# returns as soon as a future completes OR the slice elapses,
|
||||||
|
# whichever comes first, so the counter advances ~once per second
|
||||||
|
# while the parallel push runs.
|
||||||
|
pending = set(fut_to_url.keys())
|
||||||
|
while pending:
|
||||||
|
if deadline is not None and (time.time() - start) >= deadline:
|
||||||
|
break
|
||||||
|
slice_timeout = tick_interval
|
||||||
|
if deadline is not None:
|
||||||
|
remaining = deadline - (time.time() - start)
|
||||||
|
slice_timeout = max(0.0, min(tick_interval, remaining))
|
||||||
|
done, pending = wait(
|
||||||
|
pending, timeout=slice_timeout, return_when=FIRST_COMPLETED
|
||||||
|
)
|
||||||
|
for fut in done:
|
||||||
try:
|
try:
|
||||||
url, we, ok, exc = fut.result()
|
url, we, ok, exc = fut.result()
|
||||||
except Exception as e:
|
except Exception as e:
|
||||||
_logger.error(f"push_transactions_parallel worker error: {e}")
|
_logger.error(
|
||||||
|
f"push_transactions_parallel worker error: {e}"
|
||||||
|
)
|
||||||
continue
|
continue
|
||||||
results[url] = (ok, exc)
|
results[url] = (ok, exc)
|
||||||
if on_each is not None:
|
if on_each is not None:
|
||||||
@@ -403,6 +534,139 @@ class Willexecutors:
|
|||||||
on_each(url, we, ok, exc)
|
on_each(url, we, ok, exc)
|
||||||
except Exception as cb_err:
|
except Exception as cb_err:
|
||||||
_logger.error(f"push on_each callback error: {cb_err}")
|
_logger.error(f"push on_each callback error: {cb_err}")
|
||||||
|
# Drive the elapsed-time counter from the calling thread.
|
||||||
|
_fire_tick()
|
||||||
|
# Any server still pending here hit the global deadline.
|
||||||
|
if pending:
|
||||||
|
elapsed = time.time() - start
|
||||||
|
_logger.warning(
|
||||||
|
f"push global deadline ({deadline}s) reached after "
|
||||||
|
f"{elapsed:.1f}s; {len(pending)} server(s) "
|
||||||
|
f"did not answer in time"
|
||||||
|
)
|
||||||
|
for fut in pending:
|
||||||
|
url, we = fut_to_url[fut]
|
||||||
|
if url in results:
|
||||||
|
continue
|
||||||
|
if on_timeout is not None:
|
||||||
|
try:
|
||||||
|
on_timeout(url, we)
|
||||||
|
except Exception as cb_err:
|
||||||
|
_logger.error(
|
||||||
|
f"push on_timeout callback error: {cb_err}"
|
||||||
|
)
|
||||||
|
finally:
|
||||||
|
# Do not block on still-running workers (Python 3.9+: cancel queued).
|
||||||
|
try:
|
||||||
|
pool.shutdown(wait=False, cancel_futures=True)
|
||||||
|
except TypeError:
|
||||||
|
pool.shutdown(wait=False)
|
||||||
|
return results
|
||||||
|
|
||||||
|
@staticmethod
|
||||||
|
def check_transactions_parallel(items, *, on_each=None, max_workers=8,
|
||||||
|
deadline=CHECK_GLOBAL_DEADLINE,
|
||||||
|
on_timeout=None, on_tick=None,
|
||||||
|
tick_interval=1.0):
|
||||||
|
"""Check (searchtx) several will-executors concurrently.
|
||||||
|
|
||||||
|
Same design as :meth:`push_transactions_parallel`, but for the "Check"
|
||||||
|
operation: it verifies that each will-executor still holds its
|
||||||
|
transaction. ``items`` is an iterable of ``(wid, url)`` pairs (one per
|
||||||
|
will-item that has a will-executor).
|
||||||
|
|
||||||
|
Each server is contacted in parallel with a short fail-fast retry
|
||||||
|
(:meth:`check_transaction`), results are reported via
|
||||||
|
``on_each(wid, url, result_or_None, exc)`` as they arrive, ``on_tick()``
|
||||||
|
is called periodically from the calling thread to refresh a counter, and
|
||||||
|
a global ``deadline`` guarantees the dialog never freezes behind one
|
||||||
|
unresponsive server (pending servers are reported via
|
||||||
|
``on_timeout(wid, url)``).
|
||||||
|
|
||||||
|
Returns ``{wid: (result_or_None, exception_or_None)}`` for the servers
|
||||||
|
that answered in time.
|
||||||
|
"""
|
||||||
|
from concurrent.futures import ThreadPoolExecutor, wait
|
||||||
|
from concurrent.futures import FIRST_COMPLETED
|
||||||
|
|
||||||
|
targets = [(wid, url) for wid, url in items if url]
|
||||||
|
results = {}
|
||||||
|
if not targets:
|
||||||
|
return results
|
||||||
|
|
||||||
|
def _check_one(wid, url):
|
||||||
|
try:
|
||||||
|
res = Willexecutors.check_transaction(wid, url)
|
||||||
|
return wid, url, res, None
|
||||||
|
except Exception as e:
|
||||||
|
return wid, url, None, e
|
||||||
|
|
||||||
|
def _fire_tick():
|
||||||
|
if on_tick is not None:
|
||||||
|
try:
|
||||||
|
on_tick()
|
||||||
|
except Exception as cb_err:
|
||||||
|
_logger.error(f"check on_tick callback error: {cb_err}")
|
||||||
|
|
||||||
|
workers = max(1, min(max_workers, len(targets)))
|
||||||
|
# Manual pool (no ``with``): we must not block on a hung worker when the
|
||||||
|
# global deadline elapses (see push_transactions_parallel for details).
|
||||||
|
pool = ThreadPoolExecutor(max_workers=workers, thread_name_prefix="bal-check")
|
||||||
|
fut_to_target = {pool.submit(_check_one, wid, url): (wid, url)
|
||||||
|
for wid, url in targets}
|
||||||
|
start = time.time()
|
||||||
|
try:
|
||||||
|
pending = set(fut_to_target.keys())
|
||||||
|
while pending:
|
||||||
|
if deadline is not None and (time.time() - start) >= deadline:
|
||||||
|
break
|
||||||
|
slice_timeout = tick_interval
|
||||||
|
if deadline is not None:
|
||||||
|
remaining = deadline - (time.time() - start)
|
||||||
|
slice_timeout = max(0.0, min(tick_interval, remaining))
|
||||||
|
done, pending = wait(
|
||||||
|
pending, timeout=slice_timeout, return_when=FIRST_COMPLETED
|
||||||
|
)
|
||||||
|
for fut in done:
|
||||||
|
try:
|
||||||
|
wid, url, res, exc = fut.result()
|
||||||
|
except Exception as e:
|
||||||
|
_logger.error(
|
||||||
|
f"check_transactions_parallel worker error: {e}"
|
||||||
|
)
|
||||||
|
continue
|
||||||
|
results[wid] = (res, exc)
|
||||||
|
if on_each is not None:
|
||||||
|
try:
|
||||||
|
on_each(wid, url, res, exc)
|
||||||
|
except Exception as cb_err:
|
||||||
|
_logger.error(f"check on_each callback error: {cb_err}")
|
||||||
|
# Drive the elapsed-time counter from the calling thread.
|
||||||
|
_fire_tick()
|
||||||
|
# Any server still pending here hit the global deadline.
|
||||||
|
if pending:
|
||||||
|
elapsed = time.time() - start
|
||||||
|
_logger.warning(
|
||||||
|
f"check global deadline ({deadline}s) reached after "
|
||||||
|
f"{elapsed:.1f}s; {len(pending)} server(s) "
|
||||||
|
f"did not answer in time"
|
||||||
|
)
|
||||||
|
for fut in pending:
|
||||||
|
wid, url = fut_to_target[fut]
|
||||||
|
if wid in results:
|
||||||
|
continue
|
||||||
|
if on_timeout is not None:
|
||||||
|
try:
|
||||||
|
on_timeout(wid, url)
|
||||||
|
except Exception as cb_err:
|
||||||
|
_logger.error(
|
||||||
|
f"check on_timeout callback error: {cb_err}"
|
||||||
|
)
|
||||||
|
finally:
|
||||||
|
try:
|
||||||
|
pool.shutdown(wait=False, cancel_futures=True)
|
||||||
|
except TypeError:
|
||||||
|
pool.shutdown(wait=False)
|
||||||
return results
|
return results
|
||||||
|
|
||||||
@staticmethod
|
@staticmethod
|
||||||
@@ -457,11 +721,14 @@ class Willexecutors:
|
|||||||
return {}
|
return {}
|
||||||
|
|
||||||
@staticmethod
|
@staticmethod
|
||||||
def check_transaction(txid, url):
|
def check_transaction(txid, url, *, timeout=CHECK_TIMEOUT,
|
||||||
|
max_retries=CHECK_MAX_RETRIES,
|
||||||
|
retry_sleep=CHECK_RETRY_SLEEP):
|
||||||
_logger.debug(f"{url}:{txid}")
|
_logger.debug(f"{url}:{txid}")
|
||||||
try:
|
try:
|
||||||
w = Willexecutors.send_request(
|
w = Willexecutors.send_request(
|
||||||
"post", url + "/searchtx", data=txid.encode("ascii")
|
"post", url + "/searchtx", data=txid.encode("ascii"),
|
||||||
|
timeout=timeout, max_retries=max_retries, retry_sleep=retry_sleep,
|
||||||
)
|
)
|
||||||
return w
|
return w
|
||||||
except Exception as e:
|
except Exception as e:
|
||||||
|
|||||||
@@ -725,6 +725,20 @@ class BalBuildWillDialog(BalDialog):
|
|||||||
# check logic untouched.
|
# check logic untouched.
|
||||||
already_present = []
|
already_present = []
|
||||||
retry_flag = {"value": False}
|
retry_flag = {"value": False}
|
||||||
|
total = len(selected)
|
||||||
|
done = {"count": 0}
|
||||||
|
|
||||||
|
deadline = Willexecutors.PUSH_GLOBAL_DEADLINE
|
||||||
|
|
||||||
|
def _status_line():
|
||||||
|
# e.g. "Broadcasting your will to executors: 2/3 (5s / 30s)".
|
||||||
|
# The "/ 30s" makes the maximum wait explicit, so the user knows
|
||||||
|
# the wizard will proceed by then (the global deadline) instead
|
||||||
|
# of wondering how long the counter will keep climbing.
|
||||||
|
return "{} {}/{} ({}s / {}s)".format(
|
||||||
|
_("Broadcasting"), done["count"], total,
|
||||||
|
min(int(time.time() - push_start), deadline), deadline,
|
||||||
|
)
|
||||||
|
|
||||||
def on_each(url, willexecutor, ok, exc):
|
def on_each(url, willexecutor, ok, exc):
|
||||||
# Runs from a worker thread. Do only thread-safe book-keeping
|
# Runs from a worker thread. Do only thread-safe book-keeping
|
||||||
@@ -739,18 +753,50 @@ class BalBuildWillDialog(BalDialog):
|
|||||||
for wid in willexecutor["txsids"]:
|
for wid in willexecutor["txsids"]:
|
||||||
self.bal_window.willitems[wid].set_status("PUSH_FAIL", True)
|
self.bal_window.willitems[wid].set_status("PUSH_FAIL", True)
|
||||||
retry_flag["value"] = True
|
retry_flag["value"] = True
|
||||||
|
done["count"] += 1
|
||||||
|
self.msg_edit_row("{} : {}".format(url, "Ok" if ok else "Ko"))
|
||||||
|
self.msg_set_pushing(_status_line())
|
||||||
|
|
||||||
|
def on_timeout(url, willexecutor):
|
||||||
|
# The global deadline elapsed before this server answered. Mark
|
||||||
|
# its txs as failed (so the user can retry later) and show it.
|
||||||
|
for wid in willexecutor.get("txsids", []):
|
||||||
|
self.bal_window.willitems[wid].set_status("PUSH_FAIL", True)
|
||||||
|
retry_flag["value"] = True
|
||||||
self.msg_edit_row(
|
self.msg_edit_row(
|
||||||
"{} : {}".format(url, "Ok" if ok else "Ko")
|
"{} : {}".format(url, self.msg_error(_("Timeout - no answer")))
|
||||||
)
|
)
|
||||||
|
|
||||||
if self._stopping:
|
if self._stopping:
|
||||||
return
|
return
|
||||||
# Push to all selected will-executors in parallel: a slow/dead
|
# Push to all selected will-executors in parallel: a slow/dead
|
||||||
# server no longer blocks the others, so the wizard's "Broadcasting"
|
# server no longer blocks the others, so the wizard's "Broadcasting"
|
||||||
# step is no longer sequential. Each server still keeps its own
|
# step is no longer sequential. Each server keeps a short retry
|
||||||
# retry behaviour inside push_transactions_to_willexecutor.
|
# behaviour, and a global deadline guarantees the wizard always
|
||||||
Willexecutors.push_transactions_parallel(selected, on_each=on_each)
|
# proceeds even if a server never answers.
|
||||||
|
push_start = time.time()
|
||||||
|
self.msg_set_pushing(_status_line())
|
||||||
|
|
||||||
|
# Refresh the elapsed-seconds counter while the (blocking) parallel
|
||||||
|
# push runs, so the user sees time advancing instead of a frozen
|
||||||
|
# "Trasmissione". The tick is driven from THIS (Task) thread by
|
||||||
|
# push_transactions_parallel, the same thread that drives on_each, so
|
||||||
|
# the pyqtSignal repaint is reliable (a separate heartbeat thread's
|
||||||
|
# signal emissions were not being marshalled and never repainted).
|
||||||
|
def on_tick():
|
||||||
|
if self._stopping:
|
||||||
|
return
|
||||||
|
self.msg_set_pushing(_status_line())
|
||||||
|
|
||||||
|
Willexecutors.push_transactions_parallel(
|
||||||
|
selected, on_each=on_each, on_timeout=on_timeout, on_tick=on_tick
|
||||||
|
)
|
||||||
|
|
||||||
|
# Final summary line with the total elapsed time.
|
||||||
|
self.msg_set_pushing(
|
||||||
|
"{}/{} ({}s)".format(done["count"], total,
|
||||||
|
int(time.time() - push_start))
|
||||||
|
)
|
||||||
retry = retry_flag["value"]
|
retry = retry_flag["value"]
|
||||||
|
|
||||||
# Verify the "already present" servers (sequential, original logic).
|
# Verify the "already present" servers (sequential, original logic).
|
||||||
|
|||||||
@@ -451,6 +451,8 @@ class PreviewList(MyTreeView, MessageBoxMixin):
|
|||||||
self.bal_window.bal_plugin.read_file("icons/wizard.png")
|
self.bal_window.bal_plugin.read_file("icons/wizard.png")
|
||||||
)
|
)
|
||||||
)
|
)
|
||||||
|
# Tooltip so the icon is self-explanatory when hovered.
|
||||||
|
wizard.setToolTip(_("Wizard - Build your will"))
|
||||||
wizard.clicked.connect(self.bal_window.init_wizard)
|
wizard.clicked.connect(self.bal_window.init_wizard)
|
||||||
# display = QPushButton(_("Display"))
|
# display = QPushButton(_("Display"))
|
||||||
# display.clicked.connect(self.bal_window.preview_modal_dialog)
|
# display.clicked.connect(self.bal_window.preview_modal_dialog)
|
||||||
@@ -461,13 +463,20 @@ class PreviewList(MyTreeView, MessageBoxMixin):
|
|||||||
self.bal_window.bal_plugin.read_file("icons/reload.png")
|
self.bal_window.bal_plugin.read_file("icons/reload.png")
|
||||||
)
|
)
|
||||||
)
|
)
|
||||||
|
# Tooltip so the icon is self-explanatory when hovered.
|
||||||
|
refresh.setToolTip(_("Check"))
|
||||||
refresh.clicked.connect(self.check)
|
refresh.clicked.connect(self.check)
|
||||||
|
|
||||||
widget = QWidget(self)
|
widget = QWidget(self)
|
||||||
hlayout = QHBoxLayout(widget)
|
hlayout = QHBoxLayout(widget)
|
||||||
|
hlayout.setContentsMargins(0, 0, 0, 0)
|
||||||
self.will_settings_widget = WillSettingsWidget(self.bal_window, self)
|
self.will_settings_widget = WillSettingsWidget(self.bal_window, self)
|
||||||
hlayout.addWidget(self.will_settings_widget)
|
# Toolbar order (left -> right):
|
||||||
|
# Wizard | Delivery time | Check Alive | Calendar | Check (refresh)
|
||||||
|
# The Wizard button goes first (leftmost); the settings widget already
|
||||||
|
# lays out delivery/check-alive/calendar in that order internally.
|
||||||
hlayout.addWidget(wizard)
|
hlayout.addWidget(wizard)
|
||||||
|
hlayout.addWidget(self.will_settings_widget)
|
||||||
hlayout.addWidget(refresh)
|
hlayout.addWidget(refresh)
|
||||||
toolbar.insertWidget(2, widget)
|
toolbar.insertWidget(2, widget)
|
||||||
|
|
||||||
|
|||||||
@@ -150,6 +150,7 @@ class BalTimeEditWidget(QWidget, _LockTimeEditor):
|
|||||||
+ " - y: number of years after currrent day(ex: 1y means one year from today)\n"
|
+ " - y: number of years after currrent day(ex: 1y means one year from today)\n"
|
||||||
)
|
)
|
||||||
label_text = None
|
label_text = None
|
||||||
|
tooltip_text = None
|
||||||
base_field = None
|
base_field = None
|
||||||
|
|
||||||
def __init__(self, bal_window, parent, default_locktime=None):
|
def __init__(self, bal_window, parent, default_locktime=None):
|
||||||
@@ -182,6 +183,10 @@ class BalTimeEditWidget(QWidget, _LockTimeEditor):
|
|||||||
#hbox.addWidget(QLabel(self.label_text))
|
#hbox.addWidget(QLabel(self.label_text))
|
||||||
help_button=HelpButton(self.help_text)
|
help_button=HelpButton(self.help_text)
|
||||||
help_button.setText(self.label_text)
|
help_button.setText(self.label_text)
|
||||||
|
# Show a short label (e.g. "Delivery time" / "Check Alive") when the
|
||||||
|
# user hovers the icon, so the emoji button is self-explanatory.
|
||||||
|
if self.tooltip_text:
|
||||||
|
help_button.setToolTip(_(self.tooltip_text))
|
||||||
#help_button.setStyleSheet("font-size: 155555);
|
#help_button.setStyleSheet("font-size: 155555);
|
||||||
hbox.addWidget(help_button)
|
hbox.addWidget(help_button)
|
||||||
self.combo.currentIndexChanged.connect(self.on_current_index_changed)
|
self.combo.currentIndexChanged.connect(self.on_current_index_changed)
|
||||||
@@ -422,6 +427,7 @@ class ThresholdTimeWidget(BalTimeEditWidget):
|
|||||||
)
|
)
|
||||||
label_text = "🚨"
|
label_text = "🚨"
|
||||||
#label_text = "Check Alive"
|
#label_text = "Check Alive"
|
||||||
|
tooltip_text = "Check Alive"
|
||||||
base_field = "threshold"
|
base_field = "threshold"
|
||||||
|
|
||||||
def __init__(self, bal_window, parent, init_value=None):
|
def __init__(self, bal_window, parent, init_value=None):
|
||||||
@@ -442,6 +448,7 @@ class LockTimeWidget(BalTimeEditWidget):
|
|||||||
)
|
)
|
||||||
label_text = "🚛"
|
label_text = "🚛"
|
||||||
#label_text = "Locktime"
|
#label_text = "Locktime"
|
||||||
|
tooltip_text = "Delivery time"
|
||||||
base_field = "locktime"
|
base_field = "locktime"
|
||||||
|
|
||||||
def __init__(self, bal_window, parent, init_value=None):
|
def __init__(self, bal_window, parent, init_value=None):
|
||||||
@@ -468,6 +475,8 @@ class WillSettingsWidget(QWidget):
|
|||||||
self.bal_window.bal_plugin.read_file("icons/calendar.png")
|
self.bal_window.bal_plugin.read_file("icons/calendar.png")
|
||||||
)
|
)
|
||||||
)
|
)
|
||||||
|
# Tooltip so the icon is self-explanatory when hovered.
|
||||||
|
self.calendar_button.setToolTip(_("Calendar"))
|
||||||
self.calendar_button.clicked.connect(self.open_or_save_calendar)
|
self.calendar_button.clicked.connect(self.open_or_save_calendar)
|
||||||
self.widgets["locktime"] = LockTimeWidget(bal_window, self)
|
self.widgets["locktime"] = LockTimeWidget(bal_window, self)
|
||||||
self.widgets["threshold"] = ThresholdTimeWidget(bal_window, self)
|
self.widgets["threshold"] = ThresholdTimeWidget(bal_window, self)
|
||||||
|
|||||||
@@ -14,6 +14,8 @@ The actual Bitcoin logic lives in :mod:`bal.core`; this class only coordinates
|
|||||||
it with the GUI.
|
it with the GUI.
|
||||||
"""
|
"""
|
||||||
|
|
||||||
|
import threading
|
||||||
|
|
||||||
from .common import *
|
from .common import *
|
||||||
from .common import _, _logger # underscore names are not re-exported by "import *"
|
from .common import _, _logger # underscore names are not re-exported by "import *"
|
||||||
from .widgets import LockTimeWidget, PercAmountEdit, WillSettingsWidget
|
from .widgets import LockTimeWidget, PercAmountEdit, WillSettingsWidget
|
||||||
@@ -898,15 +900,50 @@ class BalWindow:
|
|||||||
|
|
||||||
def check_transactions_task(self, will):
|
def check_transactions_task(self, will):
|
||||||
start = time.time()
|
start = time.time()
|
||||||
for wid, w in will.items():
|
# Servers are now contacted in parallel (see
|
||||||
if self.waiting_dialog._stopping:
|
# Willexecutors.check_transactions_parallel) with a fast-fail timeout and
|
||||||
return
|
# a global deadline, so a single slow/dead will-executor no longer
|
||||||
if w.we:
|
# freezes the "checking transaction" dialog for minutes. The dialog
|
||||||
self.waiting_dialog.update(
|
# shows live progress plus an elapsed-time counter (Xs / DEADLINEs).
|
||||||
"checking transaction: {}\n willexecutor: {}".format(wid, w.we["url"])
|
targets = [(wid, w.we["url"]) for wid, w in will.items() if w.we]
|
||||||
|
total = len(targets)
|
||||||
|
deadline = Willexecutors.CHECK_GLOBAL_DEADLINE
|
||||||
|
done = {"count": 0}
|
||||||
|
|
||||||
|
def _status_line():
|
||||||
|
return "{} {}/{} ({}s / {}s)".format(
|
||||||
|
_("Checking transactions"), done["count"], total,
|
||||||
|
min(int(time.time() - start), deadline), deadline,
|
||||||
)
|
)
|
||||||
|
|
||||||
w.set_check_willexecutor(Willexecutors.check_transaction(wid, w.we["url"]))
|
def on_each(wid, url, res, exc):
|
||||||
|
# Reuse the original per-item logic: set_check_willexecutor handles
|
||||||
|
# both a real response and a None/failure (-> CHECK_FAIL).
|
||||||
|
try:
|
||||||
|
will[wid].set_check_willexecutor(res)
|
||||||
|
except Exception as e:
|
||||||
|
_logger.error(f"check on_each error for {wid}: {e}")
|
||||||
|
done["count"] += 1
|
||||||
|
self.waiting_dialog.update(_status_line())
|
||||||
|
|
||||||
|
def on_timeout(wid, url):
|
||||||
|
# The global deadline elapsed before this server answered: mark the
|
||||||
|
# item as failed (None response) so the user can retry later.
|
||||||
|
try:
|
||||||
|
will[wid].set_check_willexecutor(None)
|
||||||
|
except Exception as e:
|
||||||
|
_logger.error(f"check on_timeout error for {wid}: {e}")
|
||||||
|
|
||||||
|
def on_tick():
|
||||||
|
if getattr(self.waiting_dialog, "_stopping", False):
|
||||||
|
return
|
||||||
|
self.waiting_dialog.update(_status_line())
|
||||||
|
|
||||||
|
if total:
|
||||||
|
self.waiting_dialog.update(_status_line())
|
||||||
|
Willexecutors.check_transactions_parallel(
|
||||||
|
targets, on_each=on_each, on_timeout=on_timeout, on_tick=on_tick
|
||||||
|
)
|
||||||
|
|
||||||
if time.time() - start < 3:
|
if time.time() - start < 3:
|
||||||
time.sleep(3 - (time.time() - start))
|
time.sleep(3 - (time.time() - start))
|
||||||
@@ -1005,8 +1042,43 @@ class BalWindow:
|
|||||||
if fn_on_failure is None:
|
if fn_on_failure is None:
|
||||||
fn_on_failure = log_error
|
fn_on_failure = log_error
|
||||||
|
|
||||||
|
base_msg = _("Downloading will-executors list...")
|
||||||
|
download_start = time.time()
|
||||||
|
# Upper bound shown to the user. fetch_will_executors_list tries up to
|
||||||
|
# two endpoints, each with timeout=10 and one retry (~21s worst case),
|
||||||
|
# so ~45s is a realistic maximum. Showing "Xs / 45s" tells the user how
|
||||||
|
# long they may have to wait instead of an open-ended counter.
|
||||||
|
download_deadline = 45
|
||||||
|
|
||||||
def task():
|
def task():
|
||||||
|
# Heartbeat: show an elapsed-seconds counter (with the max wait made
|
||||||
|
# explicit) while the (blocking) download runs, so the user sees time
|
||||||
|
# advancing instead of a seemingly frozen dialog on a slow link.
|
||||||
|
stop_heartbeat = threading.Event()
|
||||||
|
|
||||||
|
def _heartbeat():
|
||||||
|
while not stop_heartbeat.wait(1.0):
|
||||||
|
if getattr(self.waiting_dialog, "_stopping", False):
|
||||||
|
return
|
||||||
|
try:
|
||||||
|
self.waiting_dialog.update(
|
||||||
|
"{} ({}s / {}s)".format(
|
||||||
|
base_msg,
|
||||||
|
min(int(time.time() - download_start),
|
||||||
|
download_deadline),
|
||||||
|
download_deadline,
|
||||||
|
)
|
||||||
|
)
|
||||||
|
except Exception:
|
||||||
|
return
|
||||||
|
|
||||||
|
hb = threading.Thread(target=_heartbeat, name="bal-dl-hb",
|
||||||
|
daemon=True)
|
||||||
|
hb.start()
|
||||||
|
try:
|
||||||
return self.fetch_will_executors_list(willexecutors)
|
return self.fetch_will_executors_list(willexecutors)
|
||||||
|
finally:
|
||||||
|
stop_heartbeat.set()
|
||||||
|
|
||||||
def on_success(result):
|
def on_success(result):
|
||||||
if result:
|
if result:
|
||||||
@@ -1019,9 +1091,8 @@ class BalWindow:
|
|||||||
_logger.error(f"download_list failed: {exc_info}")
|
_logger.error(f"download_list failed: {exc_info}")
|
||||||
self.show_warning(_(self.DOWNLOAD_FAILED_MESSAGE))
|
self.show_warning(_(self.DOWNLOAD_FAILED_MESSAGE))
|
||||||
|
|
||||||
msg = _("Downloading will-executors list...")
|
|
||||||
self.waiting_dialog = BalWaitingDialog(
|
self.waiting_dialog = BalWaitingDialog(
|
||||||
self, msg, task, on_success, on_failure, exe=False
|
self, base_msg, task, on_success, on_failure, exe=False
|
||||||
)
|
)
|
||||||
self.waiting_dialog.exe()
|
self.waiting_dialog.exe()
|
||||||
|
|
||||||
@@ -1034,9 +1105,23 @@ class BalWindow:
|
|||||||
# every server's (possibly timing-out) request.
|
# every server's (possibly timing-out) request.
|
||||||
pinged = set()
|
pinged = set()
|
||||||
failed = set()
|
failed = set()
|
||||||
|
total = len(wes)
|
||||||
|
ping_start = time.time()
|
||||||
|
|
||||||
|
ping_deadline = Willexecutors.PUSH_GLOBAL_DEADLINE
|
||||||
|
|
||||||
def get_title():
|
def get_title():
|
||||||
|
# Header shows progress + an elapsed-seconds counter with the max
|
||||||
|
# wait made explicit (e.g. "3s / 30s"), so the user sees time
|
||||||
|
# advancing and knows how long it may take, instead of a seemingly
|
||||||
|
# frozen dialog.
|
||||||
|
answered = len(pinged) + len(failed)
|
||||||
msg = _("Ping Will-Executors:")
|
msg = _("Ping Will-Executors:")
|
||||||
|
msg += " {}/{} ({}s / {}s)".format(
|
||||||
|
answered, total,
|
||||||
|
min(int(time.time() - ping_start), ping_deadline),
|
||||||
|
ping_deadline,
|
||||||
|
)
|
||||||
msg += "\n\n"
|
msg += "\n\n"
|
||||||
for url in wes:
|
for url in wes:
|
||||||
urlstr = "{:<50}: ".format(url[:50])
|
urlstr = "{:<50}: ".format(url[:50])
|
||||||
@@ -1066,7 +1151,18 @@ class BalWindow:
|
|||||||
except Exception:
|
except Exception:
|
||||||
pass
|
pass
|
||||||
|
|
||||||
Willexecutors.ping_servers_parallel(wes, on_each=on_each)
|
# Refresh the elapsed-seconds counter while the (blocking) parallel ping
|
||||||
|
# runs. The tick is driven from THIS thread by ping_servers_parallel,
|
||||||
|
# the same thread that drives on_each, so the dialog repaint is reliable.
|
||||||
|
def on_tick():
|
||||||
|
if getattr(self.waiting_dialog, "_stopping", False):
|
||||||
|
return
|
||||||
|
try:
|
||||||
|
self.waiting_dialog.update(get_title())
|
||||||
|
except Exception:
|
||||||
|
pass
|
||||||
|
|
||||||
|
Willexecutors.ping_servers_parallel(wes, on_each=on_each, on_tick=on_tick)
|
||||||
|
|
||||||
def ping_willexecutors(self, wes, fn_on_success, fn_on_failure=None):
|
def ping_willexecutors(self, wes, fn_on_success, fn_on_failure=None):
|
||||||
def on_success(result):
|
def on_success(result):
|
||||||
|
|||||||
@@ -20,6 +20,7 @@ Run with:
|
|||||||
"""
|
"""
|
||||||
import importlib
|
import importlib
|
||||||
import sys
|
import sys
|
||||||
|
import threading
|
||||||
import time
|
import time
|
||||||
|
|
||||||
PKG = sys.argv[1] if len(sys.argv) > 1 else "electrum.plugins.bal"
|
PKG = sys.argv[1] if len(sys.argv) > 1 else "electrum.plugins.bal"
|
||||||
@@ -122,6 +123,167 @@ def main():
|
|||||||
finally:
|
finally:
|
||||||
W.push_transactions_to_willexecutor = orig_push
|
W.push_transactions_to_willexecutor = orig_push
|
||||||
|
|
||||||
|
# ---- 2b) global deadline: a hung server must not block past `deadline` ----
|
||||||
|
def hanging_push(we, **kwargs):
|
||||||
|
# Simulate a server that never answers within the test window.
|
||||||
|
time.sleep(10)
|
||||||
|
return True
|
||||||
|
|
||||||
|
orig_push2 = W.push_transactions_to_willexecutor
|
||||||
|
W.push_transactions_to_willexecutor = staticmethod(hanging_push)
|
||||||
|
try:
|
||||||
|
wes = {
|
||||||
|
"https://fast.example": {
|
||||||
|
"url": "https://fast.example", "txs": "x", "txsids": ["a"],
|
||||||
|
},
|
||||||
|
"https://hang.example": {
|
||||||
|
"url": "https://hang.example", "txs": "y", "txsids": ["b"],
|
||||||
|
},
|
||||||
|
}
|
||||||
|
# fast one answers quickly, hang one never does within the deadline
|
||||||
|
def fast_or_hang(we, **kwargs):
|
||||||
|
if "fast" in we["url"]:
|
||||||
|
return True
|
||||||
|
time.sleep(10)
|
||||||
|
return True
|
||||||
|
W.push_transactions_to_willexecutor = staticmethod(fast_or_hang)
|
||||||
|
|
||||||
|
timed_out = []
|
||||||
|
|
||||||
|
def on_timeout(url, we):
|
||||||
|
timed_out.append(url)
|
||||||
|
|
||||||
|
start = time.time()
|
||||||
|
W.push_transactions_parallel(
|
||||||
|
wes, max_workers=2, deadline=1.0, on_timeout=on_timeout
|
||||||
|
)
|
||||||
|
elapsed = time.time() - start
|
||||||
|
assert elapsed < 3.0, f"deadline not enforced: waited {elapsed:.1f}s"
|
||||||
|
assert "https://hang.example" in timed_out, timed_out
|
||||||
|
print(f"[OK] global deadline enforced: returned in {elapsed:.1f}s, "
|
||||||
|
f"hung server reported via on_timeout")
|
||||||
|
finally:
|
||||||
|
W.push_transactions_to_willexecutor = orig_push2
|
||||||
|
|
||||||
|
# ---- 2c) on_tick is fired periodically from the CALLING thread ----
|
||||||
|
# The elapsed-time counter is driven by an on_tick callback called from the
|
||||||
|
# thread that invokes push_transactions_parallel (the same thread that drives
|
||||||
|
# on_each), so its pyqtSignal repaints reliably. Assert the callback runs
|
||||||
|
# roughly once per tick_interval while the push is in flight, and that it
|
||||||
|
# runs on the calling thread (not on a worker/heartbeat thread).
|
||||||
|
def slow_push2(we, **kwargs):
|
||||||
|
time.sleep(SLOW * 6) # ~3s, long enough for several ticks
|
||||||
|
return True
|
||||||
|
|
||||||
|
orig_push3 = W.push_transactions_to_willexecutor
|
||||||
|
W.push_transactions_to_willexecutor = staticmethod(slow_push2)
|
||||||
|
try:
|
||||||
|
wes = {
|
||||||
|
"https://tick.example": {
|
||||||
|
"url": "https://tick.example", "txs": "x", "txsids": ["a"],
|
||||||
|
},
|
||||||
|
}
|
||||||
|
ticks = []
|
||||||
|
caller_thread = threading.current_thread()
|
||||||
|
tick_threads = set()
|
||||||
|
|
||||||
|
def on_tick():
|
||||||
|
ticks.append(time.time())
|
||||||
|
tick_threads.add(threading.current_thread())
|
||||||
|
|
||||||
|
W.push_transactions_parallel(
|
||||||
|
wes, max_workers=1, on_tick=on_tick, tick_interval=0.5
|
||||||
|
)
|
||||||
|
# ~3s push with 0.5s ticks => at least a few ticks.
|
||||||
|
assert len(ticks) >= 3, f"on_tick fired too few times: {len(ticks)}"
|
||||||
|
assert tick_threads == {caller_thread}, (
|
||||||
|
"on_tick must run on the calling thread, got "
|
||||||
|
f"{[t.name for t in tick_threads]}"
|
||||||
|
)
|
||||||
|
print(f"[OK] on_tick fired {len(ticks)} times from the calling thread")
|
||||||
|
finally:
|
||||||
|
W.push_transactions_to_willexecutor = orig_push3
|
||||||
|
|
||||||
|
# ---- 2d) check_transactions_parallel: parallel + deadline + on_tick ----
|
||||||
|
# Pressing "Check" verifies each will-executor still holds its tx. This used
|
||||||
|
# to be a sequential loop with default (~140s) timeouts, freezing the
|
||||||
|
# "checking transaction" dialog on a dead server. It must now run in
|
||||||
|
# parallel, enforce a global deadline, and drive an on_tick counter from the
|
||||||
|
# calling thread.
|
||||||
|
def slow_check(txid, url, **kwargs):
|
||||||
|
time.sleep(SLOW)
|
||||||
|
return {"tx": "ok"} if "good" in url else None
|
||||||
|
|
||||||
|
orig_check = W.check_transaction
|
||||||
|
W.check_transaction = staticmethod(slow_check)
|
||||||
|
try:
|
||||||
|
targets = []
|
||||||
|
for i in range(N):
|
||||||
|
kind = "good" if i % 2 else "bad"
|
||||||
|
targets.append((f"id{i}", f"https://{kind}-{i}.example"))
|
||||||
|
|
||||||
|
checked = []
|
||||||
|
|
||||||
|
def on_each_check(wid, url, res, exc):
|
||||||
|
checked.append((wid, res))
|
||||||
|
|
||||||
|
start = time.time()
|
||||||
|
results = W.check_transactions_parallel(
|
||||||
|
targets, on_each=on_each_check, max_workers=N
|
||||||
|
)
|
||||||
|
elapsed = time.time() - start
|
||||||
|
sequential = N * SLOW
|
||||||
|
assert elapsed < sequential * 0.6, (
|
||||||
|
f"check not parallel: {elapsed:.2f}s vs {sequential:.2f}s")
|
||||||
|
assert len(results) == N, results
|
||||||
|
print(f"[OK] check parallel: {elapsed:.2f}s for {N} servers "
|
||||||
|
f"(sequential would be ~{sequential:.2f}s)")
|
||||||
|
finally:
|
||||||
|
W.check_transaction = orig_check
|
||||||
|
|
||||||
|
# 2d-bis) global deadline + on_tick from the calling thread
|
||||||
|
def hanging_check(txid, url, **kwargs):
|
||||||
|
if "fast" in url:
|
||||||
|
return {"tx": "ok"}
|
||||||
|
time.sleep(10)
|
||||||
|
return {"tx": "ok"}
|
||||||
|
|
||||||
|
orig_check2 = W.check_transaction
|
||||||
|
W.check_transaction = staticmethod(hanging_check)
|
||||||
|
try:
|
||||||
|
targets = [
|
||||||
|
("idf", "https://fast.example"),
|
||||||
|
("idh", "https://hang.example"),
|
||||||
|
]
|
||||||
|
timed_out = []
|
||||||
|
ticks = []
|
||||||
|
caller_thread = threading.current_thread()
|
||||||
|
tick_threads = set()
|
||||||
|
|
||||||
|
def on_timeout_check(wid, url):
|
||||||
|
timed_out.append(wid)
|
||||||
|
|
||||||
|
def on_tick_check():
|
||||||
|
ticks.append(time.time())
|
||||||
|
tick_threads.add(threading.current_thread())
|
||||||
|
|
||||||
|
start = time.time()
|
||||||
|
W.check_transactions_parallel(
|
||||||
|
targets, max_workers=2, deadline=2.0,
|
||||||
|
on_timeout=on_timeout_check, on_tick=on_tick_check,
|
||||||
|
tick_interval=0.5,
|
||||||
|
)
|
||||||
|
elapsed = time.time() - start
|
||||||
|
assert elapsed < 4.0, f"check deadline not enforced: {elapsed:.1f}s"
|
||||||
|
assert "idh" in timed_out, timed_out
|
||||||
|
assert len(ticks) >= 2, f"check on_tick fired too few times: {len(ticks)}"
|
||||||
|
assert tick_threads == {caller_thread}, (
|
||||||
|
"check on_tick must run on the calling thread")
|
||||||
|
print(f"[OK] check global deadline enforced ({elapsed:.1f}s), on_tick "
|
||||||
|
f"fired {len(ticks)}x from the calling thread")
|
||||||
|
finally:
|
||||||
|
W.check_transaction = orig_check2
|
||||||
|
|
||||||
# ---- 3) the wizard's loop_push must use the parallel helper ----
|
# ---- 3) the wizard's loop_push must use the parallel helper ----
|
||||||
# The "Building Will" wizard broadcasts via BalBuildWillDialog.loop_push.
|
# The "Building Will" wizard broadcasts via BalBuildWillDialog.loop_push.
|
||||||
# It previously looped over servers sequentially (one
|
# It previously looped over servers sequentially (one
|
||||||
@@ -141,6 +303,48 @@ def main():
|
|||||||
"wizard loop_push must not push to servers in a sequential loop")
|
"wizard loop_push must not push to servers in a sequential loop")
|
||||||
print("[OK] wizard loop_push uses push_transactions_parallel (not sequential)")
|
print("[OK] wizard loop_push uses push_transactions_parallel (not sequential)")
|
||||||
|
|
||||||
|
# The wizard counter must be driven via on_tick from the calling thread, NOT
|
||||||
|
# via a separate heartbeat thread (whose pyqtSignal emissions never
|
||||||
|
# repainted the dialog -> the counter was invisible during "Broadcasting").
|
||||||
|
assert "on_tick" in code, (
|
||||||
|
"wizard loop_push must drive the counter via on_tick (calling thread)")
|
||||||
|
assert "threading.Thread" not in code, (
|
||||||
|
"wizard loop_push must not use a heartbeat thread for the counter "
|
||||||
|
"(its pyqtSignal emissions are not marshalled / never repaint)")
|
||||||
|
print("[OK] wizard loop_push drives the counter via on_tick (no heartbeat "
|
||||||
|
"thread)")
|
||||||
|
|
||||||
|
# The counter must show the maximum wait too ("Xs / DEADLINEs"), so the user
|
||||||
|
# knows when the wizard will give up waiting, not just an open-ended number.
|
||||||
|
assert "PUSH_GLOBAL_DEADLINE" in code, (
|
||||||
|
"wizard counter must reference the global deadline so it can show "
|
||||||
|
"'Xs / DEADLINEs'")
|
||||||
|
assert "{}s / {}s" in code or "s / {}s" in code, (
|
||||||
|
"wizard counter must render the elapsed time AND the deadline "
|
||||||
|
"(e.g. '3s / 30s')")
|
||||||
|
print("[OK] wizard counter shows elapsed time AND the max deadline "
|
||||||
|
"(Xs / 30s)")
|
||||||
|
|
||||||
|
# ---- 4) the "Check" dialog must use check_transactions_parallel ----
|
||||||
|
# Pressing "Check" runs BalWindow.check_transactions_task. It used to loop
|
||||||
|
# over will-items sequentially calling check_transaction (default ~140s
|
||||||
|
# timeouts), freezing the "checking transaction" dialog. It must now use the
|
||||||
|
# parallel helper and show the elapsed-time counter.
|
||||||
|
window_mod = importlib.import_module(f"{PKG}.gui.qt.window")
|
||||||
|
check_src = inspect.getsource(window_mod.BalWindow.check_transactions_task)
|
||||||
|
check_code = "\n".join(
|
||||||
|
line for line in check_src.splitlines()
|
||||||
|
if not line.lstrip().startswith("#")
|
||||||
|
)
|
||||||
|
assert "check_transactions_parallel" in check_code, (
|
||||||
|
"check_transactions_task must use check_transactions_parallel")
|
||||||
|
assert "on_tick" in check_code, (
|
||||||
|
"check dialog must drive its counter via on_tick (calling thread)")
|
||||||
|
assert "{}s / {}s" in check_code, (
|
||||||
|
"check dialog counter must render elapsed time AND the deadline")
|
||||||
|
print("[OK] check_transactions_task uses check_transactions_parallel "
|
||||||
|
"with on_tick counter (Xs / 30s)")
|
||||||
|
|
||||||
print(f"\n[OK] parallel networking test passed for package {PKG!r}")
|
print(f"\n[OK] parallel networking test passed for package {PKG!r}")
|
||||||
return 0
|
return 0
|
||||||
|
|
||||||
|
|||||||
Reference in New Issue
Block a user