TaskArray

class PlasmaCalcs.tools.multiprocessing.TaskArray(tasks, *, shape=None, checks=True, **kw_super)

Bases: TaskContainer, ContainerOfArray

a container for multiple tasks; each Task is a function, args, & kwargs.

Calling self will perform all tasks, returning the result (and updating self.result as well).

tasks: array-like of Tasks, iterables, or callables.: the Tasks to be performed.

any non-Task input will be used to create a Task.

E.g. TaskArray([f1, (f2, args2), Task(f3, args3, kwargs3)])

–> [Task(f1), Task(f2, args2), Task(f3, args3, kwargs3)]
shape: None or tuple: shape that the array of tasks should have.

if None, infer tasks shape using self.task_nest_shape(tasks).

if provided, use the specified shape instead of trying to infer it.
checks: bool: whether to perform checks on the inputs, and possibly convert inputs if needed.

if False, assumes that tasks is already a numpy array of Task objects.

assign_task_idx: {assign_task_idx} printable_process_name: {printable_process_name} errors_ok: {errors_ok} result_missing: {result_missing}

Methods

`__call__`(*[, kw, idx, reset, skip_done, ...])	'perform all tasks in self, returning the results.
`assign_task_idx`()	assign task.i for tasks in self, based on their positions in self.
`coarsen`([ncoarse, idx])	return a TaskPartition containing TaskGroups of size ncoarse.
`empty`(shape)	return a TaskArray of shape shape, filled with UNSET_TASK.
`enumerate`([idx])	iterate through i in idx, yielding (i, self[i]) pairs.
`errors_ok_tuple`([value])	returns tuple of okay errors.
`init_result`()	set self.result = container with similar shape as self, filled with RESULT_MISSING.
`new_empty`([fill])	return a new array of the same shape as self, filled with the value fill.
`size`([idx])	return the number of objects in the container, or in idx if provided.
`task_nest_shape`(nested_list)	returns the implied shape for numpy object array of tasks from nested_list.
`_enumerate_all`()	iterate through all objs in self, yielding (i, self[i]) pairs.
`_size_all`()	the number of objects in this container.

Attributes

`dtype`	alias to self.data.dtype
`errors_ok`	tuple of Exception types which are okay for tasks to raise.
`ndim`	alias to self.data.ndim
`printable_process_name`	return the name to be used for progress updates, if any.
`shape`	alias to self.data.shape
`tasks`	alias to data

__call__(*, kw={}, idx=None, reset=False, skip_done=False, ncpu=None, timeout=None, ncoarse=1, print_freq=None, errors_ok=UNSET, result_missing=UNSET)

‘perform all tasks in self, returning the results.

OPTIONS (AFFECTS ALL TASKS)

kw: dict

kwargs for task will be task.kw, but updated with kw.

E.g. if task.kw = {‘x’: 1}, and kw = {‘y’: 2}, –> task called with x=1, y=2.

OPTIONS (AFFECTS WHICH TASKS ARE PERFORMED)

idx: None or iterable of indices

None –> perform all tasks in self.

iterable of indices –> perform only these tasks.

reset: bool

whether to reset self.result to all RESULT_MISSING, before starting this operation.

skip_done: bool

whether to skip tasks that already have a result (i.e. self.result[idx] != RESULT_MISSING).

OPTIONS (AFFECTS MULTIPROCESSING STRATEGY)

ncpu: None or int

max number of cpus to use for multiprocessing.

None –> use multiprocessing.cpu_count()

int –> use this value. if 0 or 1, do not use multiprocessing here.

Note: will actually use min(ncpu, number of calls to be made);

e.g. if ncpu=4 but len(arg_kw_tuples)=2, will only use 2 cpus.

timeout: None or int

max duration, in seconds. Must be None or integer (due to limitations of signal.alarm method)

None –> no time limit.

Note: if time_limit is reached, will raise a TimeoutError and save the result so far.

(in this case, any not-yet-calculated values will each be RESULT_MISSING.)

ncoarse: int

if >1, group tasks into groups of size ncoarse before performing them.

OPTIONS (MISC)

print_freq: None, or number (possibly negative or 0)

>0 –> Minimum number of seconds between progress updates.

=0 –> print every progress update.

<0 –> never print progress updates.

None –> use DEFAULTS.PROGRESS_UPDATES_PRINT_FREQ

errors_ok: UNSET or bool, Exception type, or tuple of Exception types

whether it is okay for some tasks to produce certain errors.

False –> crash if any task crashes. Equivalent to errors_ok=().

True –> except Exception (not BaseException, though). Equivalent to errors_ok=Exception.

Exception type or tuple –> except this type (or these types, if tuple).

UNSET –> use self.errors_ok.

result_missing: UNSET or any object

result to record for tasks which crash (if errors_ok!=False). Default RESULT_MISSING.

UNSET –> use self.result_missing.

_enumerate_all(): iterate through all objs in self, yielding (i, self[i]) pairs.

_size_all(): the number of objects in this container. == self.data.size

assign_task_idx(): assign task.i for tasks in self, based on their positions in self.

coarsen(ncoarse=5, *, idx=None): return a TaskPartition containing TaskGroups of size ncoarse.

Useful for coarsening a TaskContainer for more efficient multiprocessing;

grouping tasks together can reduce the overhead of multiprocessing,

while still allowing for parallel processing as the groups are run in parallel.

if idx is provided, only group the tasks with those indices.

property dtype: alias to self.data.dtype

classmethod empty(shape): return a TaskArray of shape shape, filled with UNSET_TASK.

enumerate(idx=None): iterate through i in idx, yielding (i, self[i]) pairs.

If idx is None, iterate through all objs in self (see self._enumerate_all).

property errors_ok: tuple of Exception types which are okay for tasks to raise.

setting self.errors_ok = False –> use empty tuple, i.e. no errors are okay.

setting self.errors_ok = errtype –> use errors_ok = (errtype,).

setting errors_ok will crash if it includes any parent class of KeyboardInterrupt,

e.g. errors_ok=BaseException will crash, but errors_ok=Exception will be fine.

See also: self.errors_ok_tuple

errors_ok_tuple(value=UNSET): returns tuple of okay errors. UNSET –> self.errors_ok.

False –> (). errtype –> (errtype,).

if result includes any parent class of KeyboardInterrupt, raises InputError.

e.g. errors_ok_tuple(BaseException) will crash, but errors_ok_tuple(Exception) will be fine.

init_result(): set self.result = container with similar shape as self, filled with RESULT_MISSING.

Then, return self.result.

The idea is that self.result[idx] will correspond to the result of self[idx].

property ndim: alias to self.data.ndim

new_empty(fill=UNSET): return a new array of the same shape as self, filled with the value fill.

property printable_process_name: return the name to be used for progress updates, if any.

If None, use the default: “[type(self)].__call__”.

property shape: alias to self.data.shape

size(idx=None): return the number of objects in the container, or in idx if provided.

task_cls: alias of Task

static task_nest_shape(nested_list): returns the implied shape for numpy object array of tasks from nested_list.

This will be the most natural shape to use if each element of nested_list

is an iterable (e.g. tuple) or callable (e.g. function or Task),

and there is no desire to iter(f) for any callable f in the nested_list.

property tasks: alias to data