Stream delay results as candidates complete

proxy_monitor_tui.py

@@ -8,7 +8,7 @@ import json
 import os
 from pathlib import Path
 import sys
-from typing import Any
+from typing import Any, Iterator
 import urllib.error
 import urllib.parse
 import urllib.request
@@ -287,7 +287,7 @@ def recommend_fastest(delay_results: dict[str, tuple[int | None, str]]) -> str |
     return min(successful)[1] if successful else None
 
 
-def measure_candidates(
+def iter_measure_candidates(
     client: ProxyMonitorClient,
     target_name: str,
     candidates: list[str],
@@ -295,9 +295,9 @@ def measure_candidates(
     test_url: str,
     timeout_ms: int,
     concurrency: int,
-) -> dict[str, tuple[int | None, str]]:
+) -> Iterator[tuple[str, tuple[int | None, str]]]:
+    """Yield each candidate result as soon as its controller request finishes."""
     unique_candidates = list(dict.fromkeys(candidate for candidate in candidates if candidate))
-    results: dict[str, tuple[int | None, str]] = {}
 
     def measure(candidate: str) -> tuple[int | None, str]:
         try:
@@ -316,9 +316,32 @@ def measure_candidates(
         for future in as_completed(futures):
             candidate = futures[future]
             try:
-                results[candidate] = future.result()
+                yield candidate, future.result()
             except Exception as exc:  # noqa: BLE001
-                results[candidate] = (None, str(exc))
+                yield candidate, (None, str(exc))
+
+
+def measure_candidates(
+    client: ProxyMonitorClient,
+    target_name: str,
+    candidates: list[str],
+    *,
+    test_url: str,
+    timeout_ms: int,
+    concurrency: int,
+) -> dict[str, tuple[int | None, str]]:
+    """Collect all candidate results for non-interactive callers."""
+    unique_candidates = list(dict.fromkeys(candidate for candidate in candidates if candidate))
+    results = dict(
+        iter_measure_candidates(
+            client,
+            target_name,
+            unique_candidates,
+            test_url=test_url,
+            timeout_ms=timeout_ms,
+            concurrency=concurrency,
+        )
+    )
 
     return {candidate: results.get(candidate, (None, "not run")) for candidate in unique_candidates}
 
@@ -431,6 +454,7 @@ def run_textual(
             self.target = target
             self.groups_by_name: dict[str, dict[str, Any]] = {}
             self.delay_results: dict[str, dict[str, tuple[int | None, str]]] = {}
+            self.delay_test_runs: dict[str, int] = {}
             self.selected_group = ""
             self.selected_node = ""
             self.pending_switch: tuple[str, str] | None = None
@@ -537,27 +561,69 @@ def run_textual(
             if not group:
                 self._set_operation("choose a Selector group first")
                 return
-            self._set_operation(f"testing {len(as_list(group.get('candidates')))} candidates...")
-            self.test_selected_group(str(group["name"]), list(as_list(group.get("candidates"))))
+            group_name = str(group["name"])
+            candidates = list(dict.fromkeys(str(candidate) for candidate in as_list(group.get("candidates")) if candidate))
+            run_id = self.delay_test_runs.get(group_name, 0) + 1
+            self.delay_test_runs[group_name] = run_id
+            self.delay_results[group_name] = {}
+            self._show_group(group_name)
+            self._set_operation(
+                f"testing {len(candidates)} candidates; each completed delay appears immediately..."
+            )
+            self.test_selected_group(group_name, candidates, run_id)
 
         @work(thread=True, exclusive=True)
-        def test_selected_group(self, group_name: str, candidates: list[str]) -> None:
-            results = measure_candidates(
+        def test_selected_group(self, group_name: str, candidates: list[str], run_id: int) -> None:
+            total = len(candidates)
+            for candidate, result in iter_measure_candidates(
                 client,
                 str(self.target["name"]),
                 candidates,
                 test_url=test_url,
                 timeout_ms=test_timeout_ms,
                 concurrency=delay_concurrency,
+            ):
+                self.app.call_from_thread(
+                    self._record_delay_result,
+                    group_name,
+                    run_id,
+                    candidate,
+                    result,
+                    total,
+                )
+            self.app.call_from_thread(self._finish_delay_test, group_name, run_id)
+
+        def _record_delay_result(
+            self,
+            group_name: str,
+            run_id: int,
+            candidate: str,
+            result: tuple[int | None, str],
+            total: int,
+        ) -> None:
+            if self.delay_test_runs.get(group_name) != run_id:
+                return
+            results = self.delay_results.setdefault(group_name, {})
+            results[candidate] = result
+            completed = len(results)
+            recommended = recommend_fastest(results)
+            if group_name == self.selected_group:
+                self._show_group(group_name)
+            delay, detail = result
+            latest = f"{delay} ms" if delay is not None else detail
+            self._set_operation(
+                f"testing {completed}/{total}; {candidate}: {latest}; "
+                f"best so far: {recommended or 'none'}"
             )
-            self.app.call_from_thread(self._finish_delay_test, group_name, results)
 
         def _finish_delay_test(
             self,
             group_name: str,
-            results: dict[str, tuple[int | None, str]],
+            run_id: int,
         ) -> None:
-            self.delay_results[group_name] = results
+            if self.delay_test_runs.get(group_name) != run_id:
+                return
+            results = self.delay_results.get(group_name, {})
             recommended = recommend_fastest(results)
             if group_name == self.selected_group:
                 self._show_group(group_name)