Skip to content

Commit 885dd78

Browse files
authored
feat: parallel execution verbs with reliable progress bars (#327)
* feat(parallel): shared plumbing for the parallel verbs * feat(parallel): display layer with keep-alive plain display * feat(parallel): MultiBar-backed per-task display * feat(parallel): sync engine and ordered map over threads * feat(parallel): pin error, timeout and interrupt contracts * test(parallel): process and interpreter pools with auto-chunking * feat(parallel): lazy ordered imap and completion-order imap_unordered * feat(parallel): starmap, tqdm-style aliases and as_completed * feat(parallel): reusable Pool layer * feat(parallel): asyncio engine and amap * feat(parallel): aimap, gather drop-in and AsyncPool * feat(parallel): run() shell helper * feat(parallel): @parallel decorator * feat(parallel): public API exports * docs(parallel): howto, reference and README section * test(parallel): close coverage gaps to 100% branch * fix(ci): version-gate coverage, subinterpreter pickling and Windows argv quoting * test(perf): guard import laziness and parallel coordinator scaling * fix(lint): parenthesize the laziness probe for ISC004 (ruff 0.16)
1 parent 69cc258 commit 885dd78

27 files changed

Lines changed: 4604 additions & 12 deletions

‎README.rst‎

Lines changed: 27 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -119,6 +119,33 @@ Multiple bars
119119
if __name__ == '__main__':
120120
main()
121121
122+
Parallel execution
123+
==============================================================================
124+
125+
Run a function over a batch of items -- threads, processes, or asyncio --
126+
with a live progress bar, in one call:
127+
128+
.. code:: python
129+
130+
import progressbar
131+
132+
results = progressbar.map(fetch, urls, workers=8) # threads
133+
results = progressbar.map(crunch, files, pool='process') # processes
134+
results = await progressbar.amap(fetch, urls) # asyncio
135+
136+
# A progress-bar'd xargs -P:
137+
progressbar.run('gzip -k {}', files, workers=4)
138+
139+
# An overall bar plus one bar per in-flight task:
140+
progressbar.map(crunch, files, workers=4, bar='multi')
141+
142+
Results come back in input order; ``imap``/``imap_unordered`` stream
143+
them instead, ``gather`` is a drop-in ``asyncio.gather`` with a bar,
144+
and the bar keeps animating even while long tasks are running. See the
145+
`parallel execution guide
146+
<https://progressbar-2.readthedocs.io/en/latest/howto/parallel-execution.html>`_
147+
for errors, timeouts, pools, and the decorator form.
148+
122149
Unknown length and animated bars
123150
==============================================================================
124151

‎docs/howto/index.rst‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -21,3 +21,4 @@ already read the :doc:`tutorial <../tutorial/index>`.
2121
unknown-length
2222
multibar
2323
multibar-line-offset
24+
parallel-execution

‎docs/howto/parallel-execution.rst‎

Lines changed: 236 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,236 @@
1+
======================================
2+
Run a batch in parallel with progress
3+
======================================
4+
5+
One call runs a function over a batch of items -- on threads,
6+
processes, or asyncio -- and renders a live progress bar while it
7+
happens:
8+
9+
.. code-block:: python
10+
11+
import progressbar
12+
13+
results = progressbar.map(fetch, urls, workers=8)
14+
15+
``progressbar.map`` mirrors the builtin ``map``: results come back in
16+
input order, multiple iterables zip (``progressbar.map(pow, bases,
17+
exps)``), and the bar counts completed items. The bar keeps animating
18+
-- ETA, timers, spinners -- even while long tasks are running with
19+
nothing finishing.
20+
21+
Choosing where the work runs
22+
============================
23+
24+
.. code-block:: python
25+
26+
# Threads (default): I/O-bound work, no pickling requirements.
27+
progressbar.map(fetch, urls, workers=8)
28+
29+
# Processes: CPU-bound work; fn and items must be picklable and
30+
# fn must be a module-level function.
31+
progressbar.map(crunch, files, pool='process')
32+
33+
# tqdm-style spellings of the same two calls:
34+
progressbar.thread_map(fetch, urls)
35+
progressbar.process_map(crunch, files)
36+
37+
# An executor you already have (never shut down for you):
38+
progressbar.map(fetch, urls, pool=my_executor)
39+
40+
Process pools accept the executor construction keywords you would pass
41+
to ``ProcessPoolExecutor`` -- ``initializer``/``initargs`` for
42+
per-worker setup (database connections, loaded models),
43+
``mp_context``, and ``max_tasks_per_child`` (Python 3.11+). Thread
44+
pools accept ``thread_name_prefix``. On Python 3.14+,
45+
``pool='interpreter'`` runs on an ``InterpreterPoolExecutor``.
46+
47+
For many small items on a process pool, items are automatically
48+
submitted in chunks to amortize the per-task overhead (the bar then
49+
advances a chunk at a time); pass ``chunksize=`` to tune it.
50+
51+
Streaming results as they arrive
52+
================================
53+
54+
.. code-block:: python
55+
56+
# Input order, lazily -- multiprocessing.Pool.imap semantics:
57+
for result in progressbar.imap(fetch, urls, workers=8):
58+
handle(result)
59+
60+
# Completion order, as (item, result) pairs -- the pair restores
61+
# the correspondence that completion order loses:
62+
for url, result in progressbar.imap_unordered(fetch, urls):
63+
print(f'{url} done')
64+
65+
Breaking out of either loop cancels the not-yet-submitted work and
66+
shuts the run down. For deterministic cleanup wrap the iterator in
67+
``contextlib.closing`` (``contextlib.aclosing`` for the async
68+
variants).
69+
70+
Async code
71+
==========
72+
73+
.. code-block:: python
74+
75+
# fn may be async -- or plain sync, which runs via asyncio.to_thread:
76+
results = await progressbar.amap(fetch, urls, concurrency=8)
77+
78+
# Lazy variants, mirroring the sync pair:
79+
async for result in progressbar.aimap(fetch, urls, concurrency=8):
80+
...
81+
async for url, result in progressbar.aimap_unordered(fetch, urls):
82+
...
83+
84+
# A drop-in asyncio.gather with a bar (same signature, same
85+
# ordering, same return_exceptions keyword):
86+
results = await progressbar.gather(*coroutines)
87+
88+
``concurrency=None`` (the default) creates every task up front, exactly
89+
like ``asyncio.gather``; pass a limit for large batches so tasks are
90+
created lazily in a window of that size.
91+
92+
Shell commands: a progress-bar'd ``xargs -P``
93+
=============================================
94+
95+
.. code-block:: python
96+
97+
# {} (or {item}) is replaced per item; no placeholder appends the
98+
# item as the last argument, like xargs:
99+
procs = progressbar.run('gzip -k {}', files, workers=4)
100+
101+
# List and callable templates for full control:
102+
progressbar.run(['ffmpeg', '-i', '{}', '{}.mp4'], videos)
103+
progressbar.run(lambda p: ['convert', p, p.with_suffix('.png')], images)
104+
105+
Substitution replaces only the exact placeholder tokens -- never
106+
``str.format`` -- so commands containing literal braces (``awk '{print
107+
$1}'``) work, and an item containing spaces stays a single argument.
108+
Each command's output is captured into its
109+
:py:class:`subprocess.CompletedProcess` (so child output can't corrupt
110+
the bar), and a non-zero exit raises
111+
:py:class:`subprocess.CalledProcessError` by default (``check=False``
112+
to collect exit codes instead). ``shell=True`` is available for the
113+
string form but substitutes items into a shell command line -- only
114+
use it with items you trust.
115+
116+
Sub-task bars with ``bar='multi'``
117+
==================================
118+
119+
.. code-block:: python
120+
121+
progressbar.map(crunch, files, workers=4, bar='multi')
122+
123+
Instead of one aggregate bar, a ``MultiBar`` shows an overall bar plus
124+
one line per in-flight task, labeled with the item. Inside a worker
125+
(threads and asyncio), ``progressbar.current_task_bar()`` returns that
126+
task's own bar for sub-progress:
127+
128+
.. code-block:: python
129+
130+
def crunch(path):
131+
bar = progressbar.current_task_bar()
132+
for i, block in enumerate(read_blocks(path)):
133+
process(block)
134+
if bar is not None:
135+
bar.update(i)
136+
137+
``'multi'`` suits modest worker counts -- the display occupies one
138+
terminal row per in-flight task plus one for the total. Process
139+
workers can't reach their bar this way yet
140+
(``current_task_bar()`` returns ``None`` there): the bar object cannot
141+
cross the process boundary in this release.
142+
143+
Errors, timeouts and Ctrl-C
144+
===========================
145+
146+
.. code-block:: python
147+
148+
# Default: fail fast. First exception cancels the pending work and
149+
# re-raises; running tasks finish first.
150+
progressbar.map(crunch, files)
151+
152+
# Collect instead: exceptions come back in place of results.
153+
results = progressbar.map(crunch, files, on_error='return')
154+
for item, result in zip(files, results):
155+
if isinstance(result, Exception):
156+
print(f'{item} failed: {result}')
157+
158+
# An overall deadline; expiry cancels pending work and raises
159+
# TimeoutError without waiting for stragglers.
160+
progressbar.map(crunch, files, timeout=60)
161+
162+
On ``KeyboardInterrupt`` the pending work is cancelled, the bar's
163+
final state is left on its own line, and the interrupt re-raises --
164+
no garbled terminal. One honest limitation applies everywhere:
165+
*running* tasks cannot be killed. A cancelled process task runs to
166+
completion in the background, and a sync function inside ``amap`` is
167+
abandoned in its thread, not interrupted.
168+
169+
Reusing a pool across batches
170+
=============================
171+
172+
.. code-block:: python
173+
174+
with progressbar.Pool(8) as pool: # Pool(8, 'process') for processes
175+
thumbnails = pool.map(thumbnail, images)
176+
uploads = pool.map(upload, thumbnails)
177+
pool.run('touch {}', markers)
178+
179+
async with progressbar.AsyncPool(8) as pool:
180+
results = await pool.map(fetch, urls)
181+
182+
Constructor keywords (``bar=``, ``on_error=``, ...) become per-call
183+
defaults; the executor is created lazily on first use and
184+
``pool.executor`` exposes it for direct ``submit()`` calls.
185+
186+
The decorator spelling
187+
======================
188+
189+
.. code-block:: python
190+
191+
@progressbar.parallel(workers=4, pool='process')
192+
def crunch(path): ...
193+
194+
crunch(one_path) # still an ordinary call
195+
crunch.map(paths) # parallel + bar, decorator config applied
196+
await crunch.amap(paths)
197+
198+
The decorator needs a plain module-level function (that's what
199+
``pool='process'`` can pickle); it returns the same function object
200+
with ``.map``/``.imap``/``.imap_unordered``/``.starmap``/``.amap``/
201+
``.aimap``/``.aimap_unordered`` attached.
202+
203+
Progress for futures you already have
204+
=====================================
205+
206+
.. code-block:: python
207+
208+
futures = [executor.submit(work, item) for item in items]
209+
for future in progressbar.as_completed(futures):
210+
handle(future.result())
211+
212+
A superset of :py:func:`concurrent.futures.as_completed` -- same
213+
yield order and ``timeout`` behavior, plus the bar.
214+
215+
Semantics worth knowing
216+
=======================
217+
218+
- **Bar options pass through.** Every verb forwards unknown keywords
219+
to the bar: ``prefix=``/``desc=``, ``widgets=``, ``max_value=``, ...
220+
A typo raises ``TypeError`` instead of being silently ignored.
221+
- **Bar modes.** ``bar='plain'`` (default), ``bar='multi'``,
222+
``bar=False`` (run silently), or pass your own configured
223+
``ProgressBar``/``MultiBar`` instance to be driven.
224+
- **One poll knob.** ``poll_interval`` (default 0.1s) is both how
225+
often the coordinator wakes and how often the bar redraws with no
226+
progress -- the keep-alive cadence.
227+
- **Bounded memory.** Items are pulled from the input lazily and at
228+
most ``buffersize`` tasks (default ``4 × workers``) are unfinished
229+
at once, so million-item and generator inputs run in flat memory.
230+
This deviates deliberately from ``Executor.map(buffersize=None)``
231+
on Python 3.14, where ``None`` means unbounded.
232+
- **imap_unordered yields pairs.** ``multiprocessing.Pool``'s version
233+
yields bare results; ours yields ``(item, result)`` because
234+
completion order loses the correspondence.
235+
- **Totals.** ``len()`` where available, ``operator.length_hint``
236+
otherwise; without either the bar runs in unknown-length mode.

‎docs/reference/index.rst‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -9,5 +9,6 @@ Complete descriptions of the public API surface.
99

1010
progressbar
1111
multibar
12+
parallel
1213
cli
1314
Full module autodoc <../progressbar>

0 commit comments

Comments
 (0)