From fbe94506f8d5a207e3c3f98b98ee468b2114ab6d Mon Sep 17 00:00:00 2001 From: GenSpark AI Developer Date: Mon, 15 Jun 2026 15:03:40 +0000 Subject: [PATCH] perf(wizard): parallelize Will-Executor broadcast in Building Will wizard The Building Will wizard (BalBuildWillDialog.loop_push) still broadcast the will to will-executors sequentially -- a for-loop calling push_transactions_to_willexecutor one server at a time. This is the slow "Broadcasting your will to executors: Trasmissione" step the user saw: a slow/dead server blocked the whole wizard, just like the non-wizard path did before it was parallelized. Rewrite loop_push to use Willexecutors.push_transactions_parallel (the same helper already used by window.push_transactions_to_willexecutors): - Pre-filter to the user-selected will-executors only. - Push to all selected servers concurrently (ThreadPoolExecutor); each server keeps its own retry behaviour, but a slow/dead server no longer blocks the others. Total time ~= slowest server, not the sum. - on_each callback does thread-safe book-keeping + UI update via msg_edit_row (which emits a pyqtSignal marshalled to the GUI thread). - 'already present' servers are collected and their stored tx verified sequentially afterwards (original check_transaction logic preserved). - Preserve the retry flag and the _stopping cancellation checks. tests/parallel_ping_test.py: add a static check asserting loop_push uses push_transactions_parallel and no longer contains the sequential push loop. Tests: 182 official + smoke/overflow/gui_fixes/parallel/external_zip all pass. ruff: no new issues; new code is PEP8-compliant. --- bal/gui/qt/dialogs.py | 110 +++++++++++++++++++++--------------- tests/parallel_ping_test.py | 19 +++++++ 2 files changed, 84 insertions(+), 45 deletions(-) diff --git a/bal/gui/qt/dialogs.py b/bal/gui/qt/dialogs.py index b48f4e2..aea90c1 100644 --- a/bal/gui/qt/dialogs.py +++ b/bal/gui/qt/dialogs.py @@ -709,54 +709,74 @@ class BalBuildWillDialog(BalDialog): willexecutors = Willexecutors.get_willexecutor_transactions( self.bal_window.willitems ) - for url, willexecutor in willexecutors.items(): - if self._stopping: - return - try: - if Willexecutors.is_selected( - self.bal_window.willexecutors.get(url) - ): - _logger.debug(f"{url}: {willexecutor}") - if not Willexecutors.push_transactions_to_willexecutor( - willexecutor - ): - for wid in willexecutor["txsids"]: - self.bal_window.willitems[wid].set_status( - "PUSH_FAIL", True - ) - retry = True - else: - for wid in willexecutor["txsids"]: - self.bal_window.willitems[wid].set_status( - "PUSHED", True - ) - except Willexecutors.AlreadyPresentException: + + # Only push to the will-executors the user actually selected. We + # filter the mapping up-front so push_transactions_parallel only + # talks to the relevant servers. + selected = { + url: we + for url, we in willexecutors.items() + if Willexecutors.is_selected(self.bal_window.willexecutors.get(url)) + } + + # Servers that report "already present" need their stored tx + # verified afterwards (network I/O); collect them here and process + # them sequentially after the parallel push, keeping the original + # check logic untouched. + already_present = [] + retry_flag = {"value": False} + + def on_each(url, willexecutor, ok, exc): + # Runs from a worker thread. Do only thread-safe book-keeping + # plus a signal-based UI update (msg_edit_row emits a pyqtSignal, + # which is marshalled to the GUI thread). + if isinstance(exc, Willexecutors.AlreadyPresentException): + already_present.append(url) + elif ok: for wid in willexecutor["txsids"]: - if self._stopping: - return - row = self.msg_edit_row( - "checking {} - {} : {}".format( - self.bal_window.willitems[wid].we["url"], wid, "Waiting" - ) - ) - self.bal_plugin = self.bal_window.bal_plugin - w = self.bal_window.willitems[wid] + self.bal_window.willitems[wid].set_status("PUSHED", True) + else: + for wid in willexecutor["txsids"]: + self.bal_window.willitems[wid].set_status("PUSH_FAIL", True) + retry_flag["value"] = True + self.msg_edit_row( + "{} : {}".format(url, "Ok" if ok else "Ko") + ) - w.set_check_willexecutor( - Willexecutors.check_transaction(wid, w.we["url"]) - ) - row = self.msg_edit_row( - "checked {} - {} : {}".format( - self.bal_window.willitems[wid].we["url"], - wid, - self.bal_window.willitems[wid].get_status("CHECKED"), - ), - row, - ) + if self._stopping: + return + # Push to all selected will-executors in parallel: a slow/dead + # server no longer blocks the others, so the wizard's "Broadcasting" + # step is no longer sequential. Each server still keeps its own + # retry behaviour inside push_transactions_to_willexecutor. + Willexecutors.push_transactions_parallel(selected, on_each=on_each) + + retry = retry_flag["value"] + + # Verify the "already present" servers (sequential, original logic). + self.bal_plugin = self.bal_window.bal_plugin + for url in already_present: + for wid in willexecutors[url]["txsids"]: + if self._stopping: + return + row = self.msg_edit_row( + "checking {} - {} : {}".format( + self.bal_window.willitems[wid].we["url"], wid, "Waiting" + ) + ) + w = self.bal_window.willitems[wid] + w.set_check_willexecutor( + Willexecutors.check_transaction(wid, w.we["url"]) + ) + row = self.msg_edit_row( + "checked {} - {} : {}".format( + self.bal_window.willitems[wid].we["url"], + wid, + self.bal_window.willitems[wid].get_status("CHECKED"), + ), + row, + ) - except Exception as e: - _logger.error(f"loop push error:{e}") - raise e if retry: raise Exception("retry") diff --git a/tests/parallel_ping_test.py b/tests/parallel_ping_test.py index 23bbe61..d1a6734 100644 --- a/tests/parallel_ping_test.py +++ b/tests/parallel_ping_test.py @@ -122,6 +122,25 @@ def main(): finally: W.push_transactions_to_willexecutor = orig_push + # ---- 3) the wizard's loop_push must use the parallel helper ---- + # The "Building Will" wizard broadcasts via BalBuildWillDialog.loop_push. + # It previously looped over servers sequentially (one + # push_transactions_to_willexecutor call at a time), which is exactly the + # slow path the user saw at "Broadcasting your will to executors". Make + # sure it now delegates to push_transactions_parallel. + import inspect + dialogs_mod = importlib.import_module(f"{PKG}.gui.qt.dialogs") + loop_push_src = inspect.getsource(dialogs_mod.BalBuildWillDialog.loop_push) + code = "\n".join( + line for line in loop_push_src.splitlines() + if not line.lstrip().startswith("#") + ) + assert "push_transactions_parallel" in code, ( + "wizard loop_push must use push_transactions_parallel (parallel push)") + assert "for url, willexecutor in willexecutors.items()" not in code, ( + "wizard loop_push must not push to servers in a sequential loop") + print("[OK] wizard loop_push uses push_transactions_parallel (not sequential)") + print(f"\n[OK] parallel networking test passed for package {PKG!r}") return 0