fix(util): honour flush_at_end and progress_bar in parallelize - #154
Open
LukasGold wants to merge 2 commits into
Open
fix(util): honour flush_at_end and progress_bar in parallelize#154LukasGold wants to merge 2 commits into
LukasGold wants to merge 2 commits into
Conversation
- result ordering, kwargs forwarding, empty and non-list iterables - concurrency guarded by a barrier instead of wall-clock timing - both failure modes: propagate first exception, or collect in place - the already-running event loop path used from notebooks - closes #25
Contributor
Release previewMerging this PR would release v2.0.3 (current: Changelog preview (truncated)## v2.0.3 (2026-09-01)
### Bug Fixes
- **util**: Honour flush_at_end and progress_bar in parallelize
([`2794c9b`](https://github.com/OpenSemanticLab/osw-python/commit/2794c9b0869963886c692947732ca910291e1b4f))
### Testing
- Cover osw.utils.util.parallelize
([`fb73aaf`](https://github.com/OpenSemanticLab/osw-python/commit/fb73aafe410fa6c71f6b8f306061b2f02fd965df))
- Rename oold.py to oold_test.py so its tests are collected
([`20072a9`](https://github.com/OpenSemanticLab/osw-python/commit/20072a9249cd97126a222c62a70f84e0433343ef))
Preview via python-semantic-release and conventional commits. |
- route each worker thread's stdout to its own buffer instead of rebinding sys.stdout - flush_at_end replays the collected output in input order once the batch ends - progress_bar=False now selects asyncio.gather over tqdm.gather - drop the unreachable dask branch that still held the only uses of both - cover parallelize with 22 offline tests, closes #25 and #154
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.
Closes #25 and #154.
Changes
ThreadRoutedStdout: a stdout stand-in that gives each worker thread its own buffer and passes everything else straight through to the wrapped stream.capture_output: wraps a task so what it prints lands in its own slot.parallelize: honoursflush_at_endagain, replaying the collected output in the order ofiterableafter the batch, including when the batch raised.parallelize: honoursprogress_baragain, selectingasyncio.gatherovertqdm.gatherwhen it is off.MODE = "asyncio".tests/utils/parallelize_test.py, covering ordering, kwargs forwarding, concurrency, both exception modes, the already-running-loop case, and both parameters.Rationale
Both parameters went inert in 0e6c04b, "refactor: replace dask with asyncio". That commit kept
with MessageBuffer(flush_at_end)andif progress_bar:in the dask branch it was replacing, while the new asyncio branch has neither and callstqdm.gatherunconditionally. No comment explains the omission, so it reads as an oversight in the port rather than a decision.Restoring the old mechanism was not an option. The dask path captured output with
contextlib.redirect_stdout, which rebinds the process-globalsys.stdout. Asyncio dispatches the tasks onto aThreadPoolExecutor, so concurrent redirects would overwrite each other and the calling thread's stream. The existingredirect_print_explicitlyalready carries the symptoms as comments: "does not store the messages until parallel execution finishes", and writes being "sometimes redirected in parallel execution as ifprint(line, file=None)was used".Routing by thread fixes that at the source.
sys.stdoutis replaced once per call by an object that dispatches onthreading.local(), so tasks never contend for it, and the progress bar stays legible because task output no longer interleaves with it.Behaviour change
Task output is now captured whether or not
flush_at_endis set, and printed only when it is. That restores the semantics the dask path had and that every call site already assumes: each one passesflush_at_end=debug, against adebugfield documented as "If True, debug messages will be printed". Before this change the asyncio path printed task output unconditionally, sodebug=Falsewas not quiet.Error reporting is unaffected:
WtSite.get_pageand friends signal failures throughwarnings.warnand raised exceptions, neither of which goes through stdout.Verification
tests/utils/parallelize_test.pypasses offline (22 tests), as does the rest of the offline suite (76 passed, 1 skipped).