Pools¶
mqdm provides convenient interfaces for pooled work with progress bars.
mqdm.pool(func, items)runs the items and returns a list of results.mqdm.ipool(func, items)does the same, but yields results as they complete.
It wraps around concurrent.futures (builtin library) and provides a progress bar for monitoring the execution of the task queue.
Additionally, worker functions can have their own progress bars, which integrate seamlessly with the main pool progress display.
Simple sequential¶
You can run both sequential and parallel workloads using the same shape code with mqdm.pool.
import mqdm
import time
def process_fn(xs):
for _ in mqdm.mqdm(xs, desc="batch"):
time.sleep(0.03)
return len(xs)
def main():
batches = [
list(range(18)),
list(range(24)),
list(range(30)),
list(range(36)),
]
mqdm.pool(process_fn, batches, n_workers=0)
if __name__ == "__main__":
main()
The pool will run tasks sequentially when either n_workers=0 or pool_mode='sequential'.
Simple parallel processes¶
To run tasks in parallel, provide the number of workers via the n_workers argument. The default pool_mode is "process".
import mqdm
import time
def process_fn(xs):
for _ in mqdm.mqdm(xs, desc="batch"):
time.sleep(0.1)
return len(xs)
def main():
batches = [
list(range(18)),
list(range(24)),
list(range(30)),
list(range(36)),
list(range(24)),
list(range(18)),
]
mqdm.pool(
process_fn,
batches,
desc="Parallel",
n_workers=5)
if __name__ == "__main__":
main()
Simple parallel threads¶
import mqdm
import time
def process_fn(xs):
for _ in mqdm.mqdm(xs, desc="batch"):
time.sleep(0.1)
return len(xs)
def main():
batches = [
list(range(18)),
list(range(24)),
list(range(30)),
list(range(36)),
list(range(24)),
list(range(18)),
]
mqdm.pool(
process_fn,
batches,
desc="Parallel threads",
pool_mode="thread",
n_workers=3)
if __name__ == "__main__":
main()
To use a thread pool, just pass pool_mode="thread".
Passing additional arguments¶
import mqdm
import time
def process_fn(xs, arg_a_for_process_fn=0, arg_b=0):
desc = f"a={arg_a_for_process_fn} b={arg_b}"
for _ in mqdm.mqdm(xs, desc=desc):
time.sleep(0.1)
return len(xs) + arg_a_for_process_fn + arg_b
def main():
batches = [
list(range(18)),
list(range(24)),
list(range(30)),
list(range(36)),
]
mqdm.pool(
process_fn,
batches,
n_workers=3,
# These will be passed to process_fn
arg_a_for_process_fn=5,
arg_b=6,
)
if __name__ == "__main__":
main()
Extra keyword arguments can be passed directly to mqdm.pool, which will
pass them to each function call.
Parallel with complex arguments¶
Use mqdm.args(...) to bundle multiple arguments for the worker function.
import mqdm
import time
def process_fn(x, arg_a, arg_b, *, some_key=None, arg_c=0):
"""A function with multiple arguments."""
for _ in mqdm.mqdm(range(x), desc=f"{some_key} ยท c={arg_c}"):
time.sleep(0.1)
def main():
# List of items to process
xs = [18, 24, 30, 36]
# Some lookup table for additional information
keys = {18: "oak", 24: "pine", 30: "elm", 36: "cedar"}
mqdm.pool(
process_fn,
# mqdm.args wraps function arguments so they can be
# applied to process_fn.
[
mqdm.args(
# Positional arguments for process_fn: x, arg_a=5, arg_b=6
x, 5, 6,
# Keyword arguments for process_fn: some_key=keys[x]
some_key=keys[x])
for x in xs
],
n_workers=3,
# Additional keyword args passed to all tasks
arg_c=6,
)
if __name__ == "__main__":
main()
Use mqdm.args(...) when each item needs its own positional or keyword values.
This is similar in spirit to a joblib.delayed()(*a, **kw) payload.
Note - when an argument is shared across all items, it is more efficient to use
a regular keyword argument to mqdm.pool.
ipool for streaming results¶
If you want results as they complete instead of collecting a final list:
import mqdm
import time
def process_fn(x):
time.sleep(x)
return x * 2
def main():
delays = [0.9, 0.5, 1.2, 0.7, 0.4, 1.0]
for result in mqdm.ipool(process_fn, delays, n_workers=3):
mqdm.print(f"result -> {result}")
if __name__ == "__main__":
main()
This is the nearest equivalent to concurrent.futures.as_completed(...), but
with a coordinated progress display built in.
Async pools¶
For asyncio workloads, use mqdm.aipool(...) and mqdm.apool(...).
aipool(...)is the async streaming form:async for result in mqdm.aipool(...)apool(...)collects into a list:results = await mqdm.apool(...)
import asyncio
import mqdm
async def fetch_one(i, delay=0.05):
n = i * (1 if i%2 else -1) # variety
for _ in mqdm.mqdm(range(50 + n), desc=f"fetching {i}", leave=False):
await asyncio.sleep(delay)
return i * 10
async def main():
await mqdm.apool(
fetch_one,
range(10),
desc="fetching",
n_workers=3,
)
asyncio.run(main())
These APIs are intentionally separate from the process/thread pool helpers. They
use bounded asyncio task concurrency, support async iterables and async worker
functions, and fall back to asyncio.to_thread(...) for sync callables.
Worker bars¶
mqdm works seamlessly inside worker functions, allowing each task to have its own progress bar running in parallel.
import mqdm
import time
def process_fn(n):
for _ in mqdm.mqdm(range(n), desc=f"worker {n}", leave=False):
time.sleep(0.1)
return n
def main():
mqdm.pool(
process_fn,
[16, 22, 28, 34, 20, 26,
16, 22, 28, 34, 20, 26],
desc="process pool",
n_workers=3,
)
if __name__ == "__main__":
main()
Interactive sessions with python / IPython¶
For interactive sessions, you have to switch multiprocessing to use the fork start method. This is because the default spawn method launches a worker process by calling the current script. But with interactive sessions, there is no script to call.
python <<< '
# Call me!
import multiprocessing as mp
mp.set_start_method("fork", force=True)
# Okay now we're ok
import mqdm
mqdm.pool(
example_fn,
xs,
desc="Very important work",
n_workers=4,
)
'