Improve reftests logging (#4664)

* Use editable installs so we can test changes.
* Remove "Filtered" prints which can be like 100k prints.
* Print a status every 5 minutes:
* Fix residual python processes after ctrl-C'ing reference test
generation.
  * A side effect of this is these prints. Unavoidable:

Author: jtraglia

Makefile

@@ -189,11 +189,7 @@ help-verbose:
 ###############################################################################
 
 VENV = .venv
-
-# Use editable installs for all non-generation targets, but use non-editable
-# installs for generators. More details: ethereum/consensus-specs#4633.
-UV_RUN    = uv run
-UV_RUN_NE = uv run --no-editable
+UV_RUN = uv run
 
 # Sync dependencies using uv.
 _sync: MAYBE_VERBOSE := $(if $(filter true,$(verbose)),--verbose)
@@ -334,7 +330,7 @@ reftests: MAYBE_TESTS := $(if $(k),--cases $(subst ${COMMA}, ,$(k)))
 reftests: MAYBE_FORKS := $(if $(fork),--forks $(subst ${COMMA}, ,$(fork)))
 reftests: MAYBE_PRESETS := $(if $(preset),--presets $(subst ${COMMA}, ,$(preset)))
 reftests: _pyspec
-	@$(UV_RUN_NE) python -m tests.generators.main \
+	@$(UV_RUN) python -m tests.generators.main \
 		--output $(TEST_VECTOR_DIR) \
 		$(MAYBE_VERBOSE) \
 		$(MAYBE_THREADS) \
@@ -350,7 +346,7 @@ comptests: MAYBE_FORKS := $(if $(fork),--forks $(subst ${COMMA}, ,$(fork)))
 comptests: MAYBE_PRESETS := $(if $(preset),--presets $(subst ${COMMA}, ,$(preset)))
 comptests: MAYBE_SEED := $(if $(seed),--fc-gen-seed $(seed))
 comptests: _pyspec
-	@$(UV_RUN_NE) python -m tests.generators.compliance_runners.fork_choice.test_gen \
+	@$(UV_RUN) python -m tests.generators.compliance_runners.fork_choice.test_gen \
 		--output $(COMP_TEST_VECTOR_DIR) \
 		--fc-gen-config $(FC_GEN_CONFIG) \
 		$(MAYBE_THREADS) \

pyproject.toml

@@ -48,6 +48,7 @@ generator = [
   "filelock==3.19.1",
   "minizinc==0.10.0",
   "pathos==0.3.4",
+  "psutil==7.1.0",
   "pytest==8.4.2",
   "python-snappy==0.7.3",
   "rich==14.1.0",

tests/core/pyspec/eth2spec/gen_helpers/gen_base/gen_runner.py

@@ -6,6 +6,7 @@
 from collections.abc import Iterable
 from typing import Any
 
+import psutil
 from pathos.multiprocessing import ProcessingPool as Pool
 from rich import box
 from rich.console import Console
@@ -121,7 +122,7 @@ def run_generator(input_test_cases: Iterable[TestCase], args=None):
     def debug_print(msg):
         """Only print if verbose is enabled."""
         if args.verbose:
-            print(msg)
+            print(msg, flush=True)
 
     console = Console()
     dumper = Dumper()
@@ -135,16 +136,12 @@ def debug_print(msg):
         total_found += 1
         # Check if the test case should be filtered out
         if len(args.runners) != 0 and test_case.runner_name not in args.runners:
-            debug_print(f"Filtered: {test_case.get_identifier()}")
             continue
         if len(args.presets) != 0 and test_case.preset_name not in args.presets:
-            debug_print(f"Filtered: {test_case.get_identifier()}")
             continue
         if len(args.forks) != 0 and test_case.fork_name not in args.forks:
-            debug_print(f"Filtered: {test_case.get_identifier()}")
             continue
         if len(args.cases) != 0 and not any(s in test_case.case_name for s in args.cases):
-            debug_print(f"Filtered: {test_case.get_identifier()}")
             continue
 
         # Set the output dir and add this to out list
@@ -166,17 +163,60 @@ def worker_function(data):
         """Execute a test case and update active tests."""
         test_case, active_tests = data
         key = (uuid.uuid4(), test_case.get_identifier())
-        active_tests[key] = time.time()
+        test_start = time.time()
+        active_tests[key] = test_start
+
+        debug_print(f"Starting: {test_case.get_identifier()}")
+
         try:
             execute_test(test_case, dumper)
-            debug_print(f"Generated: {test_case.get_identifier()}")
+            elapsed = time.time() - test_start
+            debug_print(f"Generated: {test_case.get_identifier()} (took {elapsed:.2f}s)")
             return "generated"
         except SkippedTest:
-            debug_print(f"Skipped: {test_case.get_identifier()}")
+            elapsed = time.time() - test_start
+            debug_print(f"Skipped: {test_case.get_identifier()} (took {elapsed:.2f}s)")
             return "skipped"
         finally:
             del active_tests[key]
 
+    def periodic_status_print(active_tests, total_tasks, completed, skipped, interval=300):
+        """Print status updates periodically in verbose mode."""
+        process = psutil.Process()
+        while completed.value < total_tasks:
+            time.sleep(interval)
+            remaining = total_tasks - completed.value
+            if remaining > 0:
+                active_count = len(active_tests)
+                # Get system-wide and process memory stats
+                vm = psutil.virtual_memory()
+                total_memory_mb = vm.total / 1024 / 1024
+                system_used_mb = vm.used / 1024 / 1024
+                # Include main process + all child processes (worker pool)
+                process_rss_mb = process.memory_info().rss / 1024 / 1024
+                for child in process.children(recursive=True):
+                    try:
+                        process_rss_mb += child.memory_info().rss / 1024 / 1024
+                    except (psutil.NoSuchProcess, psutil.AccessDenied):
+                        pass
+
+                debug_print(
+                    f"Progress: {completed.value}/{total_tasks} completed, "
+                    f"{skipped.value} skipped, {active_count} active, "
+                    f"{remaining} remaining, elapsed {time_since(start_time)}"
+                )
+                debug_print(
+                    f"Memory: "
+                    f"this process {process_rss_mb:.0f}MB, "
+                    f"all processes {system_used_mb:.0f}MB, "
+                    f"total available {total_memory_mb:.0f}MB"
+                )
+                if active_tests:
+                    for key, start_time_test in list(active_tests.items()):
+                        debug_print(
+                            f"  - Active: {key[1]} (running for {time_since(start_time_test)})"
+                        )
+
     def display_active_tests(active_tests, total_tasks, completed, skipped, width):
         """Display a table of active tests."""
         with Live(console=console) as live:
@@ -211,47 +251,69 @@ def display_active_tests(active_tests, total_tasks, completed, skipped, width):
                 time.sleep(0.25)
 
     # Generate all of the test cases
-    with multiprocessing.Manager() as manager:
-        active_tests = manager.dict()
-        completed = manager.Value("i", 0)
-        skipped = manager.Value("i", 0)
-        width = max([len(t.get_identifier()) for t in selected_test_cases])
-
-        if not args.verbose:
-            display_thread = threading.Thread(
-                target=display_active_tests,
-                args=(active_tests, len(selected_test_cases), completed, skipped, width),
-                daemon=True,
-            )
-            display_thread.start()
-
-        # Map each test case to a thread worker
-        inputs = [(t, active_tests) for t in selected_test_cases]
-
-        if args.threads == 1:
-            for input in inputs:
-                result = worker_function(input)
-                if result == "skipped":
-                    skipped.value += 1
-                completed.value += 1
-        else:
-            for result in Pool(processes=args.threads).uimap(worker_function, inputs):
-                if result == "skipped":
-                    skipped.value += 1
-                completed.value += 1
-
-        if not args.verbose:
-            display_thread.join()
-
-        elapsed = round(time.time() - start_time, 2)
-
-        # Display final summary using rich
-        total_selected = len(selected_test_cases)
-        total_completed = completed.value - skipped.value
-        total_skipped = skipped.value
+    try:
+        with multiprocessing.Manager() as manager:
+            active_tests = manager.dict()
+            completed = manager.Value("i", 0)
+            skipped = manager.Value("i", 0)
+            width = max([len(t.get_identifier()) for t in selected_test_cases])
+
+            if not args.verbose:
+                display_thread = threading.Thread(
+                    target=display_active_tests,
+                    args=(active_tests, len(selected_test_cases), completed, skipped, width),
+                    daemon=True,
+                )
+                display_thread.start()
+            else:
+                # Start periodic status printing in verbose mode
+                status_thread = threading.Thread(
+                    target=periodic_status_print,
+                    args=(active_tests, len(selected_test_cases), completed, skipped),
+                    daemon=True,
+                )
+                status_thread.start()
 
-    display_test_summary(
-        console, total_found, total_selected, total_completed, total_skipped, elapsed
-    )
+            # Map each test case to a thread worker
+            inputs = [(t, active_tests) for t in selected_test_cases]
 
-    debug_print(f"Completed generation of {tests_prefix} in {elapsed} seconds")
+            if args.threads == 1:
+                for input in inputs:
+                    result = worker_function(input)
+                    if result == "skipped":
+                        skipped.value += 1
+                    completed.value += 1
+            else:
+                pool = Pool(processes=args.threads)
+                try:
+                    for result in pool.uimap(worker_function, inputs):
+                        if result == "skipped":
+                            skipped.value += 1
+                        completed.value += 1
+                except KeyboardInterrupt:
+                    # Terminate pool immediately on interrupt
+                    pool.terminate()
+                    pool.join()
+                    raise
+                else:
+                    # Normal cleanup when completed
+                    pool.close()
+                    pool.join()
+
+            if not args.verbose:
+                display_thread.join()
+
+            elapsed = round(time.time() - start_time, 2)
+
+            # Display final summary using rich
+            total_selected = len(selected_test_cases)
+            total_completed = completed.value - skipped.value
+            total_skipped = skipped.value
+
+        display_test_summary(
+            console, total_found, total_selected, total_completed, total_skipped, elapsed
+        )
+
+        debug_print(f"Completed generation of {tests_prefix} in {elapsed} seconds")
+    except KeyboardInterrupt:
+        return

tests/core/pyspec/eth2spec/gen_helpers/gen_base/utils.py

@@ -1,17 +1,16 @@
 import functools
-import os
 import signal
 import time
 
 from rich.console import Console
 
 
 def install_sigint_handler(console: Console) -> None:
-    """On Ctrl-C, show the cursor and exit immediately."""
+    """On Ctrl-C, show the cursor and allow cleanup to run."""
 
     def _handle_sigint(signum, frame):
         console.show_cursor()
-        os._exit(0)
+        raise KeyboardInterrupt
 
     signal.signal(signal.SIGINT, _handle_sigint)