Engine updates - #30
Merged
Merged
Conversation
Correctness fixes in the pipeline AST: - Command.__eq__ no longer mixes in self.proc (runtime state), so a command's identity is stable once its job runs. exec_kwargs (env, cwd) now participates in __eq__/__hash__ (canonicalised for order-independence) and renders into str() as a conventional bash prefix, so two jobs differing only in env/cwd no longer collide. - ProcSub.__eq__ requires the same subclass: <(x) != >(x). - Pipeline file/stream conflicts raise ValueError instead of calling sys.exit(), and the guards test `is not None` so fd 0 is not skipped. Fix a deadlock when a proc-sub FIFO is never opened: if the outer command failed to spawn or exited early, the background task's blocking open() never returned and Context.cleanup() hung the whole run, including the Ctrl-C path. cleanup() now unblocks still-unopened FIFOs (holding both ends non-blocking), lets the unblocked task return without spawning, and re-raises task exceptions only after all resources are released. Replace the injected `perl -MFcntl` pipe-buffer command with a native Pipeline(pipe_size=...) option. The injected command received the inherited stdout rather than the internal pipe it was meant to grow, so F_SETPIPE_SZ hit a non-pipe fd and the pipe kept its default 64 KiB size. The pipes are created with os.pipe() here, so they are now sized directly (best-effort; logged fallback on non-Linux or over fs.pipe-max-size) before the children inherit them. skip_pipe existed only for the injection and is removed; the bwa alignment pipelines in command_strings.py request the intended 256 MiB via pipe_size. Co-Authored-By: Claude Fable 5 <[email protected]>
Replace the undocumented send(None)/send(job) generator protocol between the DAG, scheduler, and executors with an explicit, typed interface: DAG.mark_finished(job) returns the jobs a completion unblocked, and BaseScheduler (new ABC) exposes start() / job_finished(job) with the contract documented and covered by a conformance-test harness. Make interrupt handling asyncio-correct and opt-in. The old handler ran time.sleep(10) and asyncio.run(context.cleanup()) inside a signal handler while the executor's event loop was running -- asyncio.run raises RuntimeError from within a running loop, and the blocking sleep froze it. Handlers (SIGINT/SIGTERM) are now installed on the running loop via loop.add_signal_handler only for the duration of a run (and only on the main thread), the previous handlers are restored afterwards, and shutdown is fully async: SIGTERM -> bounded grace period -> SIGKILL -> awaited cleanup. BasePipeline opts in, preserving Ctrl-C behavior. Executor correctness fixes: - A job that fails to launch is recorded in jobs_with_errors and stops new launches, instead of being dropped silently; the catch is narrowed to OSError so genuine bugs still propagate. - A later pipeline stage failing to spawn now terminates the stages that already started (e.g. `sleep 600 | no_such_cmd` no longer leaks sleep past the end of the run). - A proc-sub inner command that fails to spawn fails its own job instead of aborting the whole run mid-flight; teardown paths log (rather than re-raise) launch errors surfaced from cleanup so sibling contexts are still released. - A sub-command killed by SIGPIPE (downstream reader exited early) is no longer treated as a job failure, matching non-pipefail bash. Scheduler/DAG validation: - Jobs that can never be scheduled (threads over budget, or a managed resource request over capacity) raise DagExecutionError up front instead of stalling the run silently. - Adding a second job with an identical pipeline raises instead of silently collapsing onto the first. - DAG.add_job accepts any iterable of dependencies; a list or generator used to pass validation but was silently discarded, leaving the job immediately ready. Jobs may now declare inputs/outputs metadata: declared inputs are verified before a run and declared outputs after (StorageProvider seam, local presence checks), and a failed or interrupted job's declared outputs are deleted so a partial file cannot be mistaken for a good one. DAG.skip_satisfied plus the has_all_outputs predicate can prune already-satisfied jobs before a run, groundwork for a future resume option. The long-dead Job.run() method is removed. Co-Authored-By: Claude Fable 5 <[email protected]>
- main() created the temp dir and only removed it after build_dag() and
run() returned, so any raise in between (an infeasible DAG, a launch
failure, a KeyboardInterrupt) leaked the directory. Wrap
build/run/check in try/finally so the temp dir is always removed, and
honor retain_tmpdir on failure too so a failed run can be inspected.
- check_execution raised a bare ValueError("Execution failed"); raise
DagExecutionError naming the failed jobs instead, to aid post-mortems.
The unexecuted-jobs check uses DagExecutionError as well.
- setup_logging set the level on the logger's parent, which resolved to
the root logger -- so -v/--debug turned on DEBUG for every third-party
library. Add logging.set_level(), which scopes the level to the
sentieon_cli package logger; the module loggers are at NOTSET and
inherit from it, and the root logger is left untouched.
- add_arguments mutated the shared class-level params specs (deleting
"flags" and injecting inferred "type" entries), so registering a
pipeline class on a second parser lost its short flags (e.g. -r, -t).
Build a fresh kwargs dict per argument instead.
Co-Authored-By: Claude Fable 5 <[email protected]>
Remove engine capabilities that nothing in the CLI uses: - DAG.skip_satisfied() and the has_all_outputs predicate; no pipeline offers a resume option, so the pre-run prune had no callers. - The storage module (StorageProvider/LocalStorageProvider) and the executor's input/output staging built on it, along with LocalExecutor._remove_outputs(); pipelines do not declare per-job files, so the staging and failed-output removal were dead paths. - Job.inputs/outputs/image; Job is back to (pipeline, name, threads, resources). Drop the tests that covered the removed functionality and condense some overly verbose code comments. Co-Authored-By: Claude Fable 5 <[email protected]>
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
1 participant
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.
Updates and improvements to the sentieon-cli workflow engine.