sc_parallel
Functions to allow parallelization to be performed easily.
NB: Uses multiprocess instead of multiprocessing under the hood for broadest support across platforms (e.g. Jupyter notebooks).
Highlights
sc.parallelize(): as-easy-as-possible parallelizationsc.loadbalancer(): very simple load balancer
Classes
| Name | Description |
|---|---|
| Parallel | Parallelization manager |
Parallel
sc_parallel.Parallel(
func,
iterarg=None,
iterkwargs=None,
args=None,
kwargs=None,
ncpus=None,
maxcpu=None,
maxmem=None,
interval=None,
parallelizer=None,
serial=False,
progress=False,
callback=None,
globaldict=None,
label=None,
capture=False,
die=True,
lbkwargs=None,
**func_kwargs,
)Parallelization manager
For arguments and usage documentation, see sc.parallelize(). Briefly, this class validates input arguments, sets the number of CPUs, creates a process (or thread) pool, starts the jobs running, retrieves the results from each job, and processes them into outputs.
Useful methods
reset(): reset the Parallel object to its initial pre-run state run_async(): the method that actually executes the parallelization (NB, used with every method, not only async ones) monitor(): monitor the progress of an asynchronous run finalize(): get the results from each job and process it run(): shortcut to calling run_async() followed by finalize()
Useful attributes and properties
running (bool): whether or not the jobs are running ready (bool): whether or not the jobs are ready status (str): a string description of the current state (not run, running, or done) jobs (list): a list of jobs to run or being run (empty prior to run) results (list): list of all results (the output from the jobs; empty prior to run) success (list): whether each job completed successfully (true/false) exceptions (list): if not, store the exceptions that were raised times (dict): timing information on when the jobs were started, when they finished, and how long each job took
Example:
import sciris as sc
def slowfunc(i):
sc.randsleep(seed=i)
return i**2
# Standard usage
P = sc.Parallel(slowfunc, iterarg=range(10), parallelizer='multiprocess-async')
P.run_async()
P.monitor()
P.finalize()
print(P.times)- New in version 3.0.0.
- New in version 3.1.0: “globaldict” argument
Methods
| Name | Description |
|---|---|
| disp | Display the full representation of the object |
| finalize | Get results from the jobs and close the pool |
| init | Perform all remaining initialization steps; this can safely be called after object creation |
| make_argslist | Construct argument list |
| make_pool | Make the pool and map function |
| monitor | Monitor progress – only usable with async |
| process_results | Parse the returned results dict into separate lists |
| reset | Reset to the pre-run state |
| run | Actually run the parallelization |
| run_async | Choose how to run in parallel, and do it |
| set_defaults | Define defaults for parallelization |
| set_method | Choose which method to use for parallelization |
| set_ncpus | Configure number of CPUs |
| validate_args | Validate iterarg and iterkwargs |
disp
sc_parallel.Parallel.disp()Display the full representation of the object
finalize
sc_parallel.Parallel.finalize(
get_results=True,
close_pool=True,
process_results=True,
)Get results from the jobs and close the pool
init
sc_parallel.Parallel.init()Perform all remaining initialization steps; this can safely be called after object creation
make_argslist
sc_parallel.Parallel.make_argslist()Construct argument list
make_pool
sc_parallel.Parallel.make_pool()Make the pool and map function
monitor
sc_parallel.Parallel.monitor(interval=0.1, **kwargs)Monitor progress – only usable with async
process_results
sc_parallel.Parallel.process_results()Parse the returned results dict into separate lists
reset
sc_parallel.Parallel.reset()Reset to the pre-run state
run
sc_parallel.Parallel.run()Actually run the parallelization
run_async
sc_parallel.Parallel.run_async()Choose how to run in parallel, and do it
set_defaults
sc_parallel.Parallel.set_defaults()Define defaults for parallelization
set_method
sc_parallel.Parallel.set_method()Choose which method to use for parallelization
set_ncpus
sc_parallel.Parallel.set_ncpus()Configure number of CPUs
validate_args
sc_parallel.Parallel.validate_args()Validate iterarg and iterkwargs
Functions
| Name | Description |
|---|---|
| cpu_count | Alias to multiprocessing.cpu_count() |
| cpuload | Takes a snapshot of current CPU usage via psutil |
| loadbalancer | Delay execution while CPU load is too high – a very simple load balancer. |
| memload | Takes a snapshot of current fraction of memory usage via psutil |
| parallelize | Execute a function in parallel. |
cpu_count
sc_parallel.cpu_count()Alias to multiprocessing.cpu_count()
cpuload
sc_parallel.cpuload(interval=0.1)Takes a snapshot of current CPU usage via psutil
Parameters
| Name | Type | Description | Default |
|---|---|---|---|
| interval | float | number of seconds over which to estimate CPU load | 0.1 |
Returns
| Name | Type | Description |
|---|---|---|
a float between 0-1 representing the fraction of psutil.cpu_percent() currently used. |
loadbalancer
sc_parallel.loadbalancer(
maxcpu=0.9,
maxmem=0.9,
index=None,
interval=None,
cpu_interval=0.1,
maxtime=36000,
label=None,
verbose=None,
**kwargs,
)Delay execution while CPU load is too high – a very simple load balancer.
Parameters
| Name | Type | Description | Default |
|---|---|---|---|
| maxcpu (float) | the maximum CPU load to allow for the task to still start | required | |
| maxmem (float) | the maximum memory usage to allow for the task to still start | required | |
| index (int) | the index of the task – used to start processes asynchronously (default None) | required | |
| interval (float) | the time delay to poll to see if CPU load is OK (default 0.5 seconds) | required | |
| cpu_interval (float) | number of seconds over which to estimate CPU load (default 0.1; too small gives inaccurate readings) | required | |
| maxtime (float) | maximum amount of time to wait to start the task (default 36000 seconds (10 hours)) | required | |
| label (str) | the label to print out when outputting information about task delay or start (default None) | required | |
| verbose (bool) | whether or not to print information about task delay or start (default None, which shows if a task is delayed) | required |
Examples:
# Simplest usage -- delay if CPU or memory load is >80%
sc.loadbalancer()
# Use a maximum CPU load of 50%, maximum memory of 90%, and stagger the start by process number
for nproc in processlist:
sc.loadbalancer(maxload=0.5, maxmem=0.8, index=nproc)- New in version 2.0.0:
maxmemargument;maxloadrenamedmaxcpu - New in version 3.0.0:
maxcpuandmaxmemset to 0.9 by default - New in version 3.2.5: moved from sc_profiling to sc_parallel
memload
sc_parallel.memload()Takes a snapshot of current fraction of memory usage via psutil
Note on the different functions:
- `sc.memload()` checks current total system memory consumption
- `sc.checkram()` checks RAM (virtual memory) used by the current Python process
- `sc.checkmem()` checks memory consumption by a given objectReturns
| Name | Type | Description |
|---|---|---|
a float between 0-1 representing the fraction of psutil.virtual_memory() currently used. |
parallelize
sc_parallel.parallelize(
func,
iterarg=None,
iterkwargs=None,
args=None,
kwargs=None,
ncpus=None,
maxcpu=None,
maxmem=None,
interval=None,
parallelizer=None,
serial=False,
progress=False,
callback=None,
globaldict=None,
capture=False,
die=True,
lbkwargs=None,
**func_kwargs,
)Execute a function in parallel.
Most simply, sc.parallelize() acts as a shortcut for using pool.map. However, it also provides flexibility in how arguments are passed to the function, load balancing, etc.
Either or both of iterarg or iterkwargs can be used. iterarg can be an iterable or an integer; if the latter, it will run the function that number of times and not pass the argument to the function (which may be useful for running “embarrassingly parallel” simulations). iterkwargs is a dict of iterables; each iterable must be the same length (and the same length of iterarg, if it exists), and each dict key will be used as a kwarg to the called function. Any other kwargs passed to sc.parallelize() will also be passed to the function.
This function can either use a fixed number of CPUs or allocate dynamically based on load. If ncpus is None, then it will allocate the number of CPUs dynamically. Memory (maxmem) and CPU load (maxcpu) limits can also be specified.
Parameters
| Name | Type | Description | Default |
|---|---|---|---|
| func (func) | the function to parallelize | required | |
| iterarg (list) | the variable(s) to provide to each process (see examples below) | required | |
| iterkwargs (dict) | another way of providing variables to each process (see examples below) | required | |
| args (list) | positional arguments for each process, the same for all processes | required | |
| kwargs (dict) | keyword arguments for each process, the same for all processes | required | |
| ncpus (int/float) | number of CPUs to use (if <1, treat as a fraction of the total available; if None, use loadbalancer) | required | |
| maxcpu (float) | maximum CPU load; otherwise, delay the start of the next process (not used if ncpus is specified) |
required | |
| maxmem (float) | maximum fraction of virtual memory (RAM); otherwise, delay the start of the next process | required | |
| interval (float) | number of seconds to pause between starting processes for checking load | required | |
| parallelizer (str/func) | parallelization function; default ‘multiprocess’ (see below for details) | required | |
| serial (bool) | whether to skip parallelization and run in serial (useful for debugging; equivalent to parallelizer='serial') |
required | |
| progress (bool) | whether to show a progress bar | required | |
| callback (func) | an optional function to call from each worker | required | |
| globaldict (dict) | an optional global dictionary to pass to each worker via the kwarg “globaldict” (note: may not update properly with low task latency) | required | |
| capture (bool) | if True, capture the output of the task rather than printing it | required | |
| die (bool) | whether to stop immediately if an exception is encountered (otherwise, store the exception as the result) | required | |
| lbkwargs (dict) | if provided, passed to sc.loadbalancer() |
required | |
| func_kwargs (dict) | merged with kwargs (see above) | required |
Returns
| Name | Type | Description |
|---|---|---|
| List of outputs from each process |
Example 1 – simple usage as a shortcut to multiprocess.map():
def f(x):
return x*x
results = sc.parallelize(f, [1,2,3])Example 2 – simple usage for “embarrassingly parallel” processing:
import numpy as np
def rnd():
np.random.seed()
return np.random.random()
results = sc.parallelize(rnd, 10, ncpus=4)Example 3 – three different equivalent ways to use multiple arguments:
def f(x,y):
return x*y
results1 = sc.parallelize(func=f, iterarg=[(1,2),(2,3),(3,4)])
results2 = sc.parallelize(func=f, iterkwargs={'x':[1,2,3], 'y':[2,3,4]})
results3 = sc.parallelize(func=f, iterkwargs=[{'x':1, 'y':2}, {'x':2, 'y':3}, {'x':3, 'y':4}])
assert results1 == results2 == results3Example 4 – using non-iterated arguments and dynamic load balancing:
def myfunc(i, x, y):
np.random.seed()
xy = [x+i*np.random.randn(100), y+i*np.random.randn(100)]
return xy
xylist1 = sc.parallelize(myfunc, iterarg=range(5), kwargs={'x':3, 'y':8}, maxcpu=0.8, interval=0.2) # Use kwargs dict
xylist2 = sc.parallelize(myfunc, x=5, y=10, iterarg=[0,1,2], parallelizer='multiprocessing') # Supply kwargs directly and use a different parallelizer
for p,xylist in enumerate([xylist1, xylist2]):
plt.subplot(2,1,p+1)
for i,xy in enumerate(reversed(xylist)):
plt.scatter(xy[0], xy[1], label='Run %i'%i)
plt.legend()Example 5 – using a custom parallelization function:
def f(x,y):
return [x]*y
import multiprocessing as mp
pool = mp.Pool(processes=2)
results = sc.parallelize(f, iterkwargs=dict(x=[1,2,3], y=[4,5,6]), parallelizer=pool.map) # Note: parallelizer is pool.map, not poolExample 6 – using Sciris as an interface to Dask:
def f(x,y):
return [x]*y
def dask_map(task, argslist):
import dask
queued = [dask.delayed(task)(args) for args in argslist]
return list(dask.compute(*queued))
results = sc.parallelize(f, iterkwargs=dict(x=[1,2,3], y=[4,5,6]), parallelizer=dask_map)Note 1: the default parallelizer "multiprocess" uses dill for pickling, so is the most versatile (e.g., it can pickle non-top-level functions). However, it is also the slowest for passing large amounts of data. You can switch between these with parallelizer='fast' (concurrent.futures) and parallelizer='robust' (multiprocess).
The parallelizer argument allows a wide range of different parallelizers (including different aliases for each), and also supports user-supplied ones. Note that in most cases, the default parallelizer will suffice. However, the full list of options is:
- `None`, `'default'`, `'robust'`, `'multiprocess'`: the slow but robust dill-based parallelizer `multiprocess`
- `'fast'`, `'concurrent'`, `'concurrent.futures'`: the faster but more fragile pickle-based Python-default parallelizer `concurrent.futures`
- `'multiprocessing'`: the previous pickle-based Python default parallelizer, `multiprocessing`
- `'serial'`, `'serial-copy'`: no parallelization (single-threaded); with "-copy", force pickling
- `'thread'`', `'threadpool'`', `'thread-copy'`': thread- rather than process-based parallelization ("-copy" as above)
- User supplied: any `map`-like function that takes in a function and an argument listNote 2: If parallelizing figure generation, use a non-interactive backend, or make sure (a) figure is closed inside the function call, and (b) the figure object is not returned. Otherwise, parallelization won’t increase speed (and might even be slower than serial!).
Note 3: to use on Windows, parallel calls must contained with an if __name__ == '__main__' block.
For example:
import sciris as sc
def f(x,y):
return x*y
if __name__ == '__main__':
results = sc.parallelize(func=f, iterarg=[(1,2),(2,3),(3,4)])
print(results)Note 4: In Python 3.14, the default process start method on Linux was changed from “fork” to “forkserver”. This does not use copy-on-write to share memory with worker processes but rather behaves more like “spawn” on Mac/Windows. It can also result in an EOFError when using WSL on Windows. To restore the previous behaviour, after importing sciris, set the start method to “fork” as follows:
import multiprocessing
multiprocessing.set_start_method("fork", force=True)- New in version 1.1.1: “serial” argument
- New in version 2.0.0: changed default parallelizer from
multiprocess.Pooltoconcurrent.futures.ProcessPoolExecutor; replacedmaxloadwithmaxcpu/maxmem; addedreturnpoolargument - New in version 2.0.4: added “die” argument; changed exception handling
- New in version 3.0.0: new Parallel class; propagated “die” to jobs
- New in version 3.1.0: new “globaldict” argument
- New in version 3.2.5: “capture” and “lbkwargs” arguments