feat: parallel execution verbs with reliable progress bars - #327
Conversation
![github-advanced-security[bot]](https://avatars.githubusercontent.com/in/57789?s=60&v=4){kind=small}
|
|
||
| def __call__(self, *args: typing.Any, **kwargs: typing.Any) -> typing.Any: | ||
| """The original, undecorated call.""" | ||
| ... |
| self, *iterables: typing.Any, **kwargs: typing.Any | ||
| ) -> list[typing.Any]: | ||
| """Parallel ordered map over the iterables; see the module verb.""" | ||
| ... |
| self, *iterables: typing.Any, **kwargs: typing.Any | ||
| ) -> typing.Generator[typing.Any, None, None]: | ||
| """Lazy ordered results; see the module verb.""" | ||
| ... |
| self, *iterables: typing.Any, **kwargs: typing.Any | ||
| ) -> typing.Generator[tuple[typing.Any, typing.Any], None, None]: | ||
| """Completion-order pairs; see the module verb.""" | ||
| ... |
| self, iterable: typing.Any, **kwargs: typing.Any | ||
| ) -> list[typing.Any]: | ||
| """Parallel map over pre-tupled arguments; see the module verb.""" | ||
| ... |
| if self._external is not None: | ||
| return self._external | ||
| if self._executor is None: | ||
| self._executor, _owned, _workers = resolve_executor( |
| if self._external is not None: | ||
| return self._external | ||
| if self._executor is None: | ||
| self._executor, _owned, _workers = resolve_executor( |
| ) -> list[typing.Any]: | ||
| """`run` a shell command per item on this pool's executor.""" | ||
| # Deferred import: _shell imports this module. | ||
| from . import _shell |
|
|
||
| import pytest | ||
|
|
||
| import progressbar |
|
|
||
| def test_bound_method_rejected(self) -> None: | ||
| class _Thing: | ||
| def method(self) -> None: ... |
There was a problem hiding this comment.
Pull request overview
Adds a new parallel-execution feature set to progressbar, providing sync (concurrent.futures) and asyncio engines that drive progress bars reliably (including keep-alive ticking), plus a shell run() helper, reusable Pool/AsyncPool, and a @parallel decorator. This integrates into the public API via lazy exports while keeping star-import-unsafe names out of __all__, and is backed by extensive tests and new documentation.
Changes:
- Introduces sync + asyncio parallel engines under
progressbar/_parallel/with a shared display contract (plain/multi/False/instance). - Adds public-facing helpers (
Pool,AsyncPool,parallel,current_task_bar, plus namespace-only verbs likemap,amap,run, etc.) and updates API/export snapshot. - Adds comprehensive tests and new docs/howto + reference material for parallel execution.
Reviewed changes
Copilot reviewed 26 out of 26 changed files in this pull request and generated 2 comments.
Show a summary per file
| File | Description |
|---|---|
| tests/test_parallel_shell.py | Tests for brace-safe shell templating and run() behavior. |
| tests/test_parallel_process.py | Process/interpreter pool behavior, chunking, and executor passthrough tests. |
| tests/test_parallel_pool.py | Tests for reusable sync Pool lifecycle, defaults, and verbs. |
| tests/test_parallel_map.py | Tests for ordered sync map execution and bar behavior. |
| tests/test_parallel_imap.py | Tests for lazy imap and completion-order imap_unordered, including early-break cleanup. |
| tests/test_parallel_errors.py | Tests for fail-fast, on_error, timeout, and interrupt contracts in sync engine. |
| tests/test_parallel_display.py | Tests for display backends (plain, multi, False, instance) and keep-alive ticks. |
| tests/test_parallel_decorator.py | Tests for @parallel decorator binding, async/sync verbs, and picklability. |
| tests/test_parallel_common.py | Tests for shared helpers (totals, chunking, bar-kwargs validation, task-bar context). |
| tests/test_parallel_async.py | Tests for asyncio engine (amap/aimap/gather), keep-alive, cancellation, and call strategy. |
| tests/test_parallel_aliases.py | Tests for starmap, tqdm-style aliases, and progress-aware as_completed. |
| tests/test_init_exports.py | Extends export-list consistency checks to include namespace-only exports. |
| tests/api_surface_snapshot.json | Updates snapshot to include new public API surface additions. |
| README.rst | Adds a “Parallel execution” section with examples and link to docs. |
| progressbar/_parallel/_sync.py | Implements sync parallel coordination, executor resolution, verbs, and reusable Pool. |
| progressbar/_parallel/_shell.py | Implements brace-safe command templating and run() helper on threads. |
| progressbar/_parallel/_display.py | Implements display protocol + backends (PlainDisplay, MultiDisplay, NullDisplay). |
| progressbar/_parallel/_decorator.py | Implements @parallel decorator that attaches batch verbs to functions. |
| progressbar/_parallel/_common.py | Shared utilities (chunking, totals, defaults, bar kwarg validation, task-bar contextvar). |
| progressbar/_parallel/_async.py | Implements asyncio coordination engine plus AsyncPool, amap/aimap/gather. |
| progressbar/_parallel/init.py | Aggregates parallel implementation exports for lazy re-exporting. |
| progressbar/init.py | Adds lazy exports + namespace-only mapping for parallel verbs and types. |
| docs/reference/parallel.rst | Adds API reference page for parallel execution surface and shared keywords. |
| docs/reference/index.rst | Links new parallel reference page into docs index. |
| docs/howto/parallel-execution.rst | Adds how-to guide covering usage patterns, errors/timeouts, pools, decorator, and run(). |
| docs/howto/index.rst | Links new parallel how-to page into docs index. |
💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.
| _sync, | ||
| ) | ||
|
|
||
| #: A tiny portable command: exit with the given code. |
| def _pool_kind(pool: typing.Any) -> str: | ||
| """Map a `pool=` argument to 'thread'/'process'/'interpreter'.""" | ||
| if isinstance(pool, str): | ||
| return pool | ||
| if isinstance(pool, concurrent.futures.ProcessPoolExecutor): | ||
| return 'process' | ||
| return 'thread' |
![chatgpt-codex-connector[bot]](https://avatars.githubusercontent.com/in/1144995?s=60&v=4){kind=small}
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 18f7bcf218
ℹ️ About Codex in GitHub
Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".
| for task in self.in_flight: | ||
| task.cancel() |
There was a problem hiding this comment.
When progressbar.gather(..., return_exceptions=False) encounters one failed awaitable, execute_async enters its finally block and this loop cancels every remaining child. Native asyncio.gather propagates the first exception while allowing the other submitted awaitables to continue, so the advertised drop-in replacement can unexpectedly abort side-effecting sibling operations; the gather path needs different exception cleanup from amap.
Useful? React with 👍 / 👎.
| self._check_deadline() | ||
| future = self._next_done() |
There was a problem hiding this comment.
When a task finishes after timeout but before the current poll_interval wait expires, the deadline was checked only before blocking, so the completed future is accepted; if it was the final future, the loop exits successfully without another check. For example, a 50 ms task with timeout=10 ms and poll_interval=100 ms returns normally instead of timing out. The async loop has the same ordering around _next_done(), so both paths should bound the wait by the remaining deadline or recheck immediately afterward.
Useful? React with 👍 / 👎.
| @functools.wraps(verb, assigned=('__doc__',), updated=()) | ||
| def bound(*iterables: typing.Any, **kwargs: typing.Any) -> typing.Any: | ||
| return verb(fn, *iterables, **{**config, **kwargs}) |
There was a problem hiding this comment.
A decorator configured as documented with @parallel(workers=4, pool='process') forwards pool to .amap() as well; execute_async then treats it as a bar keyword and raises TypeError instead of running. Conversely, an async-only default such as concurrency breaks the attached sync verbs. Since every decorated function exposes both verb families, backend-specific defaults need to be filtered or translated before dispatch.
Useful? React with 👍 / 👎.
| else: | ||
| while len(self.in_flight) < self.window and self._launch_one(): | ||
| pass |
There was a problem hiding this comment.
With a nonempty input and concurrency=0 or a negative value, this condition launches no tasks, self.in_flight remains empty, and amap silently returns [] as though the batch succeeded. The synchronous loop behaves identically for buffersize <= 0; these limits should be validated as positive rather than dropping every input item.
Useful? React with 👍 / 👎.
Summary
One verb family runs a function -- or a shell command -- over a batch of items on threads, processes, or asyncio, with a progress bar that keeps animating even while long tasks are in flight:
Full surface:
map/imap/imap_unordered/starmap/thread_map/process_map/as_completed/amap/aimap/aimap_unordered/gather/run/@parallel/Pool/AsyncPool/current_task_bar.Design
concurrent.futuresengine and an asyncio engine share aDisplayprotocol (plainaggregate bar, MultiBar-backedmultiwith auto-managed per-task bars,False, or a caller-supplied instance).imapis ordered/lazy/results-only likemultiprocessing.Pool.imap;gatherkeeps asyncio's exact contract includingreturn_exceptions;as_completedis a superset of the stdlib function; executor construction kwargs (initializer,initargs,mp_context,max_tasks_per_child,thread_name_prefix) pass through.poll_intervalknob drives both the coordinator and the bar.buffersize, default 4x workers) -- million-item and generator inputs run flat. Process pools auto-chunk (~16 chunks/worker, capped) so tiny items aren't pathological.on_error='return'opt-out; overalltimeoutthat cancels without waiting for stragglers; Ctrl-C cancels pending work, leaves the bar's final state on its own line, and re-raises. Documented honestly: running tasks cannot be killed.TypeErrorinstead of being swallowed by the mixin chain.progressbar/_parallel/) becauseprogressbar.parallelis the public decorator; namespace-only names (map,gather, ...) stay out of__all__so star-imports never shadow builtins, guarded by extendedtest_init_exports.Testing
progressbar/_parallel/; ruff and pyright clean.break, process-pool pickling (incl. decorated functions), duplicate-label multi-bar keys, and brace-saferun()templates (awk '{print $1}') are all pinned by tests.docs/howto/parallel-execution.rst), reference page, README section; sphinx-Wbuild clean.Design spec and adversarial-review history (internal + tqdm/alive-progress/mpire comparison) in
docs/superpowers/specs/2026-08-14-parallel-execution-design.md(untracked by convention).Phase 2 (seams reserved, no API breakage): process-worker sub-bars via an
initializer-installed queue reporter;ExceptionGroupcollect mode once the floor is 3.11.