Start the process pool
(self)
| 440 | } |
| 441 | |
| 442 | def start(self) -> None: |
| 443 | """Start the process pool""" |
| 444 | # Convert config to dict for pickling |
| 445 | # We need to be careful with nested dataclasses |
| 446 | config_dict = self._serialize_config(self.config) |
| 447 | |
| 448 | # Pass current environment to worker processes (to keep API keys, etc.) |
| 449 | import os |
| 450 | import sys |
| 451 | |
| 452 | current_env = dict(os.environ) |
| 453 | |
| 454 | # Configure executor |
| 455 | executor_kwargs = { |
| 456 | "max_workers": self.num_workers, |
| 457 | "initializer": _worker_init, |
| 458 | "initargs": (config_dict, self.evaluation_file, current_env), |
| 459 | } |
| 460 | |
| 461 | # Handle Python version differences for multiprocessing |
| 462 | if sys.version_info >= (3, 11): |
| 463 | logger.info(f"Set max {self.config.max_tasks_per_child} tasks per child") |
| 464 | executor_kwargs["max_tasks_per_child"] = self.config.max_tasks_per_child |
| 465 | elif self.config.max_tasks_per_child is not None: |
| 466 | logger.warn( |
| 467 | "max_tasks_per_child is only supported in Python 3.11+. " |
| 468 | "Ignoring max_tasks_per_child and using spawn start method." |
| 469 | ) |
| 470 | # Fallback to spawn context |
| 471 | executor_kwargs["mp_context"] = mp.get_context("spawn") |
| 472 | |
| 473 | # Create process pool with initializer |
| 474 | self.executor = ProcessPoolExecutor(**executor_kwargs) |
| 475 | logger.info(f"Started process pool with {self.num_workers} processes") |
| 476 | |
| 477 | def stop(self) -> None: |
| 478 | """Stop the process pool""" |