SAFE21.io/bal-electrum-plugin
1
0
Fork 0
forked from bitcoinafterlife/bal-electrum-plugin
Code Pull Requests Activity

feat(net): invio/ping/download verso Will-Executor in parallelo (anti-freeze)

Browse Source 
Problema: i server Will-Executor venivano contattati in sequenza e, su
timeout, send_request riprovava 10x con sleep 3s (~130s per server morto).
Un solo server irraggiungibile bloccava l'intera operazione (impallamento UI).

Soluzione:
- ThreadPoolExecutor: ping e push ora in parallelo (tempo ~= server piu' lento,
  non la somma). Un server morto non blocca piu' gli altri.
- Fast-fail per operazioni interattive (ping/info/download): max_retries=0,
  niente retry-storm.
- Feedback live: callback on_each() aggiorna il dialog server-per-server
  (thread-safe via pyqtSignal di BalWaitingDialog.update).
- Push transazioni: parallelo ma con retry per-server mantenuti (no perdita tx).

File:
- bal/core/willexecutors.py: send_request(+max_retries,+retry_sleep),
  get_info_task fast-fail, NEW ping_servers_parallel(), push_transactions_parallel(),
  DEFAULT_TIMEOUT=5.
- bal/gui/qt/window.py: ping_willexecutors_task + push_transactions_to_willexecutors
  riscritti su helper paralleli con feedback live; fetch_will_executors_list fast-fail.
- bal/core/util.py: BUGFIX get_value_amount usava in_output (bool) invece di
  din_output (tupla) -> TypeError. Scoperto dai test ufficiali Gitea.

Test (contro il codice refactor):
- pytest tests/ ufficiali: 117 core + 65 gui = 182 passed.
- smoke/external_zip/windows_overflow/gui_fixes: OK.
- parallel_ping_test (nuovo): 0.50s per 8 server vs ~4.00s sequenziale.
- ruff: nessun nuovo problema introdotto (codice nuovo PEP8-compliant).

Aggiunti i test ufficiali del repo Gitea + REPORT_NETWORKING_PARALLELO.md.
This commit is contained in:
GenSpark AI Developer
2026-06-15 14:29:04 +00:00
parent 3c44a29f84
commit 4c5726571e
16 changed files with 2843 additions and 59 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
bal/core/util.py
View File

@@ -321,7 +321,7 @@ class Util:
value_amount = 0
for outa in outputsa:
same_amount, same_address = Util.in_output(outa, txb.outputs())
same_amount, same_address = Util.din_output(outa, txb.outputs())
if not (same_amount or same_address):
return False
if same_amount and same_address:

143
bal/core/willexecutors.py
View File

@@ -145,8 +145,21 @@ class Willexecutors:
@staticmethod
def send_request(
method, url, data=None, *, timeout=10, handle_response=None, count_reply=0
method, url, data=None, *, timeout=10, handle_response=None, count_reply=0,
max_retries=10, retry_sleep=3,
):
"""Send an HTTP request to a will-executor server.
``max_retries`` / ``retry_sleep`` control the timeout-retry behaviour:
* For *critical* operations (pushing inheritance transactions) the
historical default of up to 10 retries with a 3s back-off is kept, so
a transient network hiccup does not lose a transaction.
* For *interactive* operations (ping / info / list download) callers
should pass ``max_retries=0`` so a dead server fails fast (one short
timeout) instead of blocking the UI for minutes. See
:meth:`ping_servers_parallel`.
"""
network = Network.get_instance()
if not network:
raise Exception("You are offline.")
@@ -178,9 +191,12 @@ class Willexecutors:
else:
raise Exception(f"unexpected {method=!r}")
except TimeoutError:
if count_reply < 10:
_logger.debug(f"timeout({count_reply}) error: retry in 3 sec...")
time.sleep(3)
if count_reply < max_retries:
_logger.debug(
f"timeout({count_reply}) error: retry in {retry_sleep} sec..."
)
if retry_sleep:
time.sleep(retry_sleep)
return Willexecutors.send_request(
method,
url,
@@ -188,6 +204,8 @@ class Willexecutors:
timeout=timeout,
handle_response=handle_response,
count_reply=count_reply + 1,
max_retries=max_retries,
retry_sleep=retry_sleep,
)
else:
_logger.debug(f"Too many timeouts: {count_reply}")
@@ -254,18 +272,28 @@ class Willexecutors:
Willexecutors.get_info_task(url, we)
@staticmethod
def get_info_task(url, willexecutor):
def get_info_task(url, willexecutor, *, timeout=DEFAULT_TIMEOUT,
max_retries=0, retry_sleep=0):
w = None
try:
_logger.info("GETINFO_WILLEXECUTOR")
_logger.debug(url)
w = Willexecutors.send_request("get", url + "/" + chainname + "/info")
# Fast-fail by default (max_retries=0): a dead server returns after a
# single short timeout instead of retrying 10x with sleeps, which
# used to freeze the UI for minutes per unreachable server.
w = Willexecutors.send_request(
"get", url + "/" + chainname + "/info",
timeout=timeout, max_retries=max_retries, retry_sleep=retry_sleep,
)
if isinstance(w, dict):
willexecutor["url"] = url
willexecutor["status"] = 200
willexecutor["base_fee"] = w["base_fee"]
willexecutor["address"] = w["address"]
willexecutor["info"] = w["info"]
else:
# No dict reply (timeout / empty) -> mark as unreachable.
willexecutor["status"] = "KO"
_logger.debug(f"response_data {w}")
except Exception as e:
_logger.error(f"error {e} contacting {url}: {w}")
@@ -274,6 +302,109 @@ class Willexecutors:
willexecutor["last_update"] = datetime.now().timestamp()
return willexecutor
@staticmethod
def ping_servers_parallel(willexecutors, *, on_each=None, max_workers=8,
timeout=DEFAULT_TIMEOUT):
"""Ping every will-executor concurrently and report results as they
arrive.
Network requests run in a thread pool: each ``send_http_on_proxy`` call
schedules its coroutine on Electrum's shared asyncio loop and blocks
only its *own* worker thread, so the total wall-clock time is roughly
that of the slowest server rather than the *sum* of all of them. A
single dead server can no longer stall the whole batch.
Args:
willexecutors: ``{url: we_dict}`` mapping (mutated in place with the
ping result, exactly like the old sequential ``ping_servers``).
on_each: optional ``callback(url, we_dict, ok: bool)`` invoked from a
worker thread each time a server answers (or fails), so the GUI
can update its list live. Must be thread-safe / marshalled to
the GUI thread by the caller.
max_workers: maximum number of concurrent pings.
timeout: per-request timeout in seconds (fast-fail, no retries).
Returns:
The same ``willexecutors`` mapping, updated in place.
"""
from concurrent.futures import ThreadPoolExecutor, as_completed
items = list(willexecutors.items())
if not items:
return willexecutors
def _ping_one(url, we):
we = Willexecutors.get_info_task(
url, we, timeout=timeout, max_retries=0, retry_sleep=0
)
ok = we.get("status") == 200
return url, we, ok
workers = max(1, min(max_workers, len(items)))
with ThreadPoolExecutor(max_workers=workers,
thread_name_prefix="bal-ping") as pool:
futures = [pool.submit(_ping_one, url, we) for url, we in items]
for fut in as_completed(futures):
try:
url, we, ok = fut.result()
except Exception as e: # defensive: never let one server crash all
_logger.error(f"ping_servers_parallel worker error: {e}")
continue
willexecutors[url] = we
if on_each is not None:
try:
on_each(url, we, ok)
except Exception as cb_err:
_logger.error(f"ping on_each callback error: {cb_err}")
return willexecutors
@staticmethod
def push_transactions_parallel(willexecutors, *, on_each=None, max_workers=8):
"""Push transactions to multiple will-executors concurrently.
Like :meth:`ping_servers_parallel` but for the ``pushtxs`` operation.
Each server keeps the historical retry behaviour of
:meth:`push_transactions_to_willexecutor` (which is important so a real
transaction is not lost to a transient hiccup), but the servers are now
contacted in parallel instead of one-after-another, and results are
reported via ``on_each(url, we_dict, ok, exc)`` as they complete.
Returns ``{url: (ok, exception_or_None)}``.
"""
from concurrent.futures import ThreadPoolExecutor, as_completed
targets = [(url, we) for url, we in willexecutors.items() if "txs" in we]
results = {}
if not targets:
return results
def _push_one(url, we):
try:
ok = Willexecutors.push_transactions_to_willexecutor(we)
return url, we, ok, None
except Willexecutors.AlreadyPresentException as ape:
return url, we, False, ape
except Exception as e:
return url, we, False, e
workers = max(1, min(max_workers, len(targets)))
with ThreadPoolExecutor(max_workers=workers,
thread_name_prefix="bal-push") as pool:
futures = [pool.submit(_push_one, url, we) for url, we in targets]
for fut in as_completed(futures):
try:
url, we, ok, exc = fut.result()
except Exception as e:
_logger.error(f"push_transactions_parallel worker error: {e}")
continue
results[url] = (ok, exc)
if on_each is not None:
try:
on_each(url, we, ok, exc)
except Exception as cb_err:
_logger.error(f"push on_each callback error: {cb_err}")
return results
@staticmethod
def initialize_willexecutor(willexecutor, url, status=None, old_willexecutor=None):
old_willexecutor=old_willexecutor if old_willexecutor is not None else {}

145
bal/gui/qt/window.py
View File

@@ -797,48 +797,72 @@ class BalWindow:
msg += f"{url}:\t{willexecutors[url]['broadcast_status']}\n"
return msg
error = False
# Initialise statuses + show the list immediately.
for url in willexecutors:
if self.waiting_dialog._stopping:
return
willexecutor = willexecutors[url]
willexecutors[url].setdefault("broadcast_status", _("waiting..."))
try:
self.waiting_dialog.update(getMsg(willexecutors))
if "txs" in willexecutor:
try:
if Willexecutors.push_transactions_to_willexecutor(
willexecutors[url]
):
for wid in willexecutors[url]["txsids"]:
self.willitems[wid].set_status("PUSHED", True)
willexecutors[url]["broadcast_status"] = _("Success")
else:
for wid in willexecutors[url]["txsids"]:
self.willitems[wid].set_status("PUSH_FAIL", True)
error = True
willexecutors[url]["broadcast_status"] = _("Failed")
del willexecutor["txs"]
except Willexecutors.AlreadyPresentException:
for wid in willexecutor["txsids"]:
if self.waiting_dialog._stopping:
return
self.waiting_dialog.update(
"checking {} - {} : {}".format(
self.willitems[wid].we["url"], wid, "Waiting"
)
)
w = self.willitems[wid]
w.set_check_willexecutor(
Willexecutors.check_transaction(wid, w.we["url"])
)
self.waiting_dialog.update(
"checked {} - {} : {}".format(
self.willitems[wid].we["url"],
wid,
self.willitems[wid].get_status("CHECKED"),
)
)
except Exception:
pass
if error:
error = {"flag": False}
already_present = []
def on_each(url, willexecutor, ok, exc):
# Runs from a worker thread. We only do book-keeping + a thread-safe
# signal-based UI update here; the heavier "already present" check
# path (which itself does network I/O) is handled below in the main
# task thread to keep the original sequential behaviour for it.
if isinstance(exc, Willexecutors.AlreadyPresentException):
already_present.append(url)
willexecutor["broadcast_status"] = _("checking...")
elif ok:
for wid in willexecutor.get("txsids", []):
self.willitems[wid].set_status("PUSHED", True)
willexecutor["broadcast_status"] = _("Success")
else:
for wid in willexecutor.get("txsids", []):
self.willitems[wid].set_status("PUSH_FAIL", True)
error["flag"] = True
willexecutor["broadcast_status"] = _("Failed")
willexecutor.pop("txs", None)
try:
self.waiting_dialog.update(getMsg(willexecutors))
except Exception:
pass
if self.waiting_dialog._stopping:
return
# Push to all servers in parallel (each server keeps its own retry
# behaviour, but a slow/dead server no longer blocks the others).
Willexecutors.push_transactions_parallel(willexecutors, on_each=on_each)
# Handle the "already present" servers: verify each stored tx. This
# keeps the exact original check logic, just executed after the parallel
# push has identified which servers need it.
for url in already_present:
willexecutor = willexecutors[url]
for wid in willexecutor.get("txsids", []):
if self.waiting_dialog._stopping:
return
self.waiting_dialog.update(
"checking {} - {} : {}".format(
self.willitems[wid].we["url"], wid, "Waiting"
)
)
w = self.willitems[wid]
w.set_check_willexecutor(
Willexecutors.check_transaction(wid, w.we["url"])
)
self.waiting_dialog.update(
"checked {} - {} : {}".format(
self.willitems[wid].we["url"],
wid,
self.willitems[wid].get_status("CHECKED"),
)
)
if error["flag"]:
return True
def export_json_file(self, path):
@@ -941,7 +965,13 @@ class BalWindow:
for url in candidates:
_logger.info(f"fetch_will_executors_list: trying {url}")
try:
resp = Willexecutors.send_request("get", url, timeout=20)
# Fast-fail with a couple of short retries instead of the
# default 10x/3s storm: if the user's connection is flaky we
# want to fall back to the next URL (and then show the simple
# error message) quickly, not freeze for minutes.
resp = Willexecutors.send_request(
"get", url, timeout=10, max_retries=1, retry_sleep=1,
)
_logger.info(
f"fetch_will_executors_list: resp type={type(resp).__name__} "
f"len={len(resp) if hasattr(resp, '__len__') else 'n/a'}"
@@ -997,8 +1027,13 @@ class BalWindow:
def ping_willexecutors_task(self, wes):
_logger.info("ping willexecutots task")
pinged = []
failed = []
# Track per-url state for the live status text. Servers are contacted
# in parallel (see Willexecutors.ping_servers_parallel), so a single
# unreachable server no longer blocks all the others: the whole batch
# now takes about as long as the slowest server instead of the sum of
# every server's (possibly timing-out) request.
pinged = set()
failed = set()
def get_title():
msg = _("Ping Will-Executors:")
@@ -1006,26 +1041,32 @@ class BalWindow:
for url in wes:
urlstr = "{:<50}: ".format(url[:50])
if url in pinged:
urlstr += "Ok"
urlstr += _("Ok")
elif url in failed:
urlstr += "Ko"
urlstr += _("Ko")
else:
urlstr += "--"
urlstr += _("waiting...")
urlstr += "\n"
msg += urlstr
return msg
for url, we in wes.items():
def on_each(url, we, ok):
if ok:
pinged.add(url)
else:
failed.add(url)
try:
self.waiting_dialog.update(get_title())
except Exception:
pass
wes[url] = Willexecutors.get_info_task(url, we)
if wes[url]["status"] == "KO":
failed.append(url)
else:
pinged.append(url)
# Show the initial "waiting..." list immediately.
try:
self.waiting_dialog.update(get_title())
except Exception:
pass
Willexecutors.ping_servers_parallel(wes, on_each=on_each)
def ping_willexecutors(self, wes, fn_on_success, fn_on_failure=None):
def on_success(result):