TaskPartition

class PlasmaCalcs.tools.multiprocessing.TaskPartition(original, groups, **kw_super)

a container for a list of TaskGroups, each with Tasks from the same TaskContainer.

Knows to call each TaskGroup in parallel,

but call in series the individual tasks within each TaskGroup.

Helps with TaskContainer.coarsen.

assign_task_idx: bool: whether to assign task.i for each task, based on its position in self.
printable_process_name: None or str: include in progress update printouts. If None, use a reasonable default.

Methods

`__call__`(*[, kw, ncpu, timeout, print_freq])	perform all tasks in all groups of self.
`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.
`enumerate`([idx])	iterate through i in idx, yielding (i, self[i]) pairs.
`errors_ok_tuple`([value])	returns tuple of okay errors.
`from_task_container`(original[, ncoarse, ...])	return a TaskPartition containing TaskGroups of size ncoarse.
`init_result`()	set self.result = container with similar shape as self, filled with RESULT_MISSING.
`new_empty`([fill])	return a new list 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.
`_assign_result_i`(result_i, i)	assign result_i to self.result[i], and also self.original.result[j] for all relevant j.
`_enumerate_all`()	iterate through all objs in self, yielding (i, self[i]) pairs.
`_size_all`()	the number of objects in this container.

Attributes

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

__call__(*, kw={}, ncpu=None, timeout=None, print_freq=None)

perform all tasks in all groups of self. return results formatted to original shape.

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.
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.)

The following kwargs are acceptable for TaskContainer but NOT here:

idx, reset, skip_done, ncoarse, errors_ok, result_missing.

That is because idx, reset, skip_done, and ncoarse are not relevant,

while errors_ok and result_missing are handled by the TaskGroups.

_assign_result_i(result_i, i): assign result_i to self.result[i], and also self.original.result[j] for all relevant j.

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

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.

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.

classmethod from_task_container(original, ncoarse=5, *, idx=None, errors_ok=UNSET, result_missing=UNSET)

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.

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 original.errors_ok.
result_missing: any object: result to record for tasks which crash (if errors_ok!=False). Default RESULT_MISSING.

UNSET –> use original.result_missing.

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].

new_empty(fill=UNSET): return a new list 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__”.

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