code_public/lidar_rendu
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Rendre effectif le délai de sécurité de 2 h des workers parallèles

Browse Source
This commit is contained in:
Antoine Jacquin
2026-09-18 21:08:25 +02:00
parent 4cdbf5a0e4
commit bb086e308a
1 changed files with 27 additions and 22 deletions
Download Patch File Download Diff File Expand all files Collapse all files

49
lidar_pipeline/pipeline.py
View File

@ -11,7 +11,7 @@ import logging
import multiprocessing import multiprocessing
import shutil import shutil
import time import time
from concurrent.futures import ProcessPoolExecutor, as_completed from concurrent.futures import ProcessPoolExecutor, as_completed, TimeoutError as FuturesTimeoutError
from datetime import datetime from datetime import datetime
from pathlib import Path from pathlib import Path
import subprocess import subprocess
@ -589,27 +589,32 @@ class LidarArchaeoPipeline:
done = 0 done = 0
t_deadline = time.time() + 7200 t_deadline = time.time() + 7200
try: try:
for future in as_completed(future_to_file): # timeout= : sans lui, as_completed bloque entre deux
if time.time() > t_deadline: # complétions et le délai de 2 h n'est jamais évalué si
logger.error("Délai dépassé (2h) — annulation des workers restants") # aucun worker ne rend la main (run figé pour toujours).
for f in future_to_file: try:
f.cancel() futures_iter = as_completed(future_to_file,
break timeout=max(1.0, t_deadline - time.time()))
laz_file = future_to_file[future] for future in futures_iter:
done += 1 laz_file = future_to_file[future]
try: done += 1
success = future.result() try:
results[laz_file.name] = success success = future.result()
status = "✓" if success else "✗" results[laz_file.name] = success
logger.info(f" [{done}/{len(files)}] {status} {laz_file.name}") status = "✓" if success else "✗"
if success and self.incremental_index: logger.info(f" [{done}/{len(files)}] {status} {laz_file.name}")
self._rebuild_index_incremental() if success and self.incremental_index:
except Exception as e: self._rebuild_index_incremental()
logger.error(f" [{done}/{len(files)}] ✗ {laz_file.name}: {e}") except Exception as e:
logger.debug(f" Traceback:", exc_info=True) logger.error(f" [{done}/{len(files)}] ✗ {laz_file.name}: {e}")
report_event(self.output_dir, _file_basename(laz_file), logger.debug(f" Traceback:", exc_info=True)
"tile", "fail", detail=str(e)) report_event(self.output_dir, _file_basename(laz_file),
results[laz_file.name] = False "tile", "fail", detail=str(e))
results[laz_file.name] = False
except FuturesTimeoutError:
logger.error("Délai dépassé (2h) — annulation des workers restants")
for f in future_to_file:
f.cancel()
except KeyboardInterrupt: except KeyboardInterrupt:
logger.info("Interruption — annulation des travaux en cours...") logger.info("Interruption — annulation des travaux en cours...")
for f in future_to_file: for f in future_to_file: