@mytec: iter2.4.2 start

backend/app/services/parallel_coverage_service.py

@@ -28,6 +28,7 @@ import threading
 import multiprocessing as mp
 from typing import List, Dict, Tuple, Any, Optional, Callable
 import numpy as np
+import psutil
 
 
 # ── Cancellation token ──
@@ -46,6 +47,46 @@ class CancellationToken:
         return self._event.is_set()
 
 
+# ── Worker process cleanup ──
+
+def _kill_worker_processes() -> int:
+    """Kill all child processes of the current process.
+
+    Uses psutil to find and terminate/kill child processes that may be
+    orphaned after ProcessPoolExecutor timeout or cancellation.
+    Returns the number of children killed.
+    """
+    try:
+        current = psutil.Process(os.getpid())
+        children = current.children(recursive=True)
+    except (psutil.NoSuchProcess, psutil.AccessDenied):
+        return 0
+
+    if not children:
+        return 0
+
+    count = len(children)
+
+    # First: graceful terminate
+    for child in children:
+        try:
+            child.terminate()
+        except (psutil.NoSuchProcess, psutil.AccessDenied):
+            pass
+
+    # Wait up to 3 seconds for graceful exit
+    gone, alive = psutil.wait_procs(children, timeout=3)
+
+    # Force kill survivors
+    for p in alive:
+        try:
+            p.kill()
+        except (psutil.NoSuchProcess, psutil.AccessDenied):
+            pass
+
+    return count
+
+
 # ── Try to import Ray ──
 
 RAY_AVAILABLE = False
@@ -426,10 +467,12 @@ def _calculate_with_process_pool(
 
     t_calc = time.time()
     all_results: List[Dict] = []
+    pool = None
 
-    with ProcessPoolExecutor(max_workers=num_workers) as executor:
+    try:
+        pool = ProcessPoolExecutor(max_workers=num_workers)
         futures = {
-            executor.submit(
+            pool.submit(
                 _pool_worker_process_chunk,
                 (chunk, terrain_cache, buildings, osm_data, config),
             ): i
@@ -460,6 +503,17 @@ def _calculate_with_process_pool(
             log_fn(f"Progress: {completed_chunks}/{len(chunks)} chunks ({pct}%) — "
                    f"{pts} pts, {rate:.0f} pts/s, ETA {eta:.0f}s")
 
+    except Exception as e:
+        log_fn(f"ProcessPool error: {e}")
+
+    finally:
+        # CRITICAL: Always cleanup pool and orphaned workers
+        if pool:
+            pool.shutdown(wait=False, cancel_futures=True)
+        killed = _kill_worker_processes()
+        if killed > 0:
+            log_fn(f"Killed {killed} orphaned worker processes")
+
     calc_time = time.time() - t_calc
     log_fn(f"ProcessPool done: {calc_time:.1f}s, {len(all_results)} results "
            f"({calc_time / max(1, total_points) * 1000:.1f}ms/point)")