Conversation
Contributor
|
This pull request has been imported. If you are a Meta employee, you can view this in D109949075. (Because this pull request was imported automatically, there will not be any future comments.) |
The pipeline now treats a `ProcessPoolExecutor` passed to `PipelineBuilder.pipe(op, executor=...)` as a *specification* rather than a live object to drive. At build time it reads the executor's worker count and initializer and stands up its own equivalent worker pool, which it spawns eagerly and shuts down when the pipeline stops. This fixes an unsafe pattern: a stdlib `ProcessPoolExecutor` spawns its worker processes lazily, on demand, as items arrive. In a running pipeline that fork happens mid-load from a process that already has the event-loop thread (and other threads) live, and forking a multi-threaded process can deadlock or corrupt the child. Spawning all workers up front, before the event-loop thread starts, removes that hazard. The in-process build reuses the same eager, fork-safe `_hoist_process_pools` machinery that `run_pipeline_in_subprocess` already used, so the two paths now handle process pools consistently. A `ProcessPoolExecutor` shared across stages maps to a single owned pool, reaped once. `build_pipeline` and `PipelineBuilder.build` gain an `mp_context` argument so callers can select the multiprocessing start method (e.g. `"spawn"` or `"forkserver"`) for those spawned workers — the real lever for fork-safety alongside eager pre-warming. Because the pipeline only reads the executor as a spec, an executor handed over after it has already spawned workers or had work submitted now triggers a `RuntimeWarning` (its own workers are not used and the caller still owns it) instead of being rejected with `ValueError`. This warn-not-reject policy is unified across the in-process, `run_pipeline_in_subprocess`, and stage-fusion paths (`_ensure_executor_unused` becomes `_warn_if_executor_used`). Thread- and interpreter-pool executors are unchanged by this diff; recreating those as pipeline-owned, async-aware workers is a planned follow-up.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
The pipeline now treats a
ProcessPoolExecutorpassed toPipelineBuilder.pipe(op, executor=...)as a specification rather than a live object to drive. At build time it reads the executor's worker count and initializer and stands up its own equivalent worker pool, which it spawns eagerly and shuts down when the pipeline stops.This fixes an unsafe pattern: a stdlib
ProcessPoolExecutorspawns its worker processes lazily, on demand, as items arrive. In a running pipeline that fork happens mid-load from a process that already has the event-loop thread (and other threads) live, and forking a multi-threaded process can deadlock or corrupt the child. Spawning all workers up front, before the event-loop thread starts, removes that hazard. The in-process build reuses the same eager, fork-safe_hoist_process_poolsmachinery thatrun_pipeline_in_subprocessalready used, so the two paths now handle process pools consistently. AProcessPoolExecutorshared across stages maps to a single owned pool, reaped once.build_pipelineandPipelineBuilder.buildgain anmp_contextargument so callers can select the multiprocessing start method (e.g."spawn"or"forkserver") for those spawned workers — the real lever for fork-safety alongside eager pre-warming.Because the pipeline only reads the executor as a spec, an executor handed over after it has already spawned workers or had work submitted now triggers a
RuntimeWarning(its own workers are not used and the caller still owns it) instead of being rejected withValueError. This warn-not-reject policy is unified across the in-process,run_pipeline_in_subprocess, and stage-fusion paths (_ensure_executor_unusedbecomes_warn_if_executor_used).Thread- and interpreter-pool executors are unchanged by this diff; recreating those as pipeline-owned, async-aware workers is a planned follow-up.