Skip to content

fix(util): honour flush_at_end and progress_bar in parallelize - #154

Open
LukasGold wants to merge 2 commits into
mainfrom
test/parallelize-coverage
Open

fix(util): honour flush_at_end and progress_bar in parallelize#154
LukasGold wants to merge 2 commits into
mainfrom
test/parallelize-coverage

Conversation

@LukasGold

@LukasGold LukasGold commented Sep 1, 2026

Copy link
Copy Markdown
Contributor

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: honours flush_at_end again, replaying the collected output in the order of iterable after the batch, including when the batch raised.
  • parallelize: honours progress_bar again, selecting asyncio.gather over tqdm.gather when it is off.
  • Removed the unreachable dask branch, which held the only remaining uses of either parameter behind a hardcoded MODE = "asyncio".
  • 22 offline tests in 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) and if progress_bar: in the dask branch it was replacing, while the new asyncio branch has neither and calls tqdm.gather unconditionally. 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-global sys.stdout. Asyncio dispatches the tasks onto a ThreadPoolExecutor, so concurrent redirects would overwrite each other and the calling thread's stream. The existing redirect_print_explicitly already carries the symptoms as comments: "does not store the messages until parallel execution finishes", and writes being "sometimes redirected in parallel execution as if print(line, file=None) was used".

Routing by thread fixes that at the source. sys.stdout is replaced once per call by an object that dispatches on threading.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_end is set, and printed only when it is. That restores the semantics the dask path had and that every call site already assumes: each one passes flush_at_end=debug, against a debug field documented as "If True, debug messages will be printed". Before this change the asyncio path printed task output unconditionally, so debug=False was not quiet.

Error reporting is unaffected: WtSite.get_page and friends signal failures through warnings.warn and raised exceptions, neither of which goes through stdout.

Verification

tests/utils/parallelize_test.py passes offline (22 tests), as does the rest of the offline suite (76 passed, 1 skipped).

- 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
@github-actions

github-actions Bot commented Sep 1, 2026

Copy link
Copy Markdown
Contributor

Release preview

Merging this PR would release v2.0.3 (current: v2.0.2).

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
@LukasGold LukasGold changed the title test: cover osw.utils.util.parallelize fix(util): honour flush_at_end and progress_bar in parallelize Sep 1, 2026
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

Write tests for parallelized functions

1 participant