The IPython task interface#
The task interface to the cluster presents the engines as a fault tolerant, dynamic load-balanced system of workers. Unlike the direct interface, in the task interface the user has no direct access to individual engines. By allowing the IPython scheduler to assign work, this interface is simultaneously simpler and more powerful.
Best of all, the user can use both of these interfaces running at the same time to take advantage of their respective strengths. When the user can break up the user’s work into segments that do not depend on previous execution, the task interface is ideal. But it also has more power and flexibility, allowing the user to guide the distribution of jobs, without having to assign tasks to engines explicitly.
As always, the first step is to start a cluster and connect a
For more detailed information about starting the controller and engines, see our introduction to using IPython for parallel computing.
In : import ipyparallel as ipp In : cluster = ipp.Cluster() In : cluster.start_cluster_sync() In : rc = cluster.connect_client_sync() In : rc.wait_for_engines(4)
In : lview = rc.load_balanced_view() # default load-balanced view
For more information, see the in-depth explanation of Views.
Quick and easy parallelism#
In many cases, you want to apply a Python function to a sequence of
objects, but in parallel. Like the direct interface, these can be
implemented via the task interface. The exact same tools can perform these
actions in load-balanced ways as well as multiplexed ways: a parallel version
@view.parallel() function decorator. If one specifies the
balanced=True, then they are dynamically load balanced. Thus, if the
execution time per item varies significantly, you should use the versions in
the task interface.
map(), use a LoadBalancedView:
In : lview.block = True In : serial_result = map(lambda x:x**10, range(32)) In : parallel_result = lview.map(lambda x:x**10, range(32)) In : serial_result==parallel_result Out: True
Parallel function decorator#
Parallel functions are just like normal functions, but they can be called on sequences and in parallel. The direct interface provides a decorator that turns any Python function into a parallel function:
In : @lview.parallel() ....: def f(x): ....: return 10.0 * x ** 4 ....: In : f.map(range(32)) # this is done in parallel Out: [0.0, 10.0, 160.0, ...]
Often, pure atomic load-balancing is too primitive for your work. In these cases, you
may want to associate some kind of
Dependency that describes when, where, or whether
a task can be run. In IPython, we provide two types of dependencies:
[Functional Dependencies] and [Graph Dependencies]
It is important to note that the pure ZeroMQ scheduler does not support dependencies, and you will see errors or warnings if you try to use dependencies with the pure scheduler.
Functional dependencies are used to determine whether a given engine is capable of running
a particular task. This is implemented via a special
UnmetDependency, found in
ipyparallel.error. Its use is very simple:
if a task fails with an UnmetDependency exception, then the scheduler, instead of relaying
the error up to the client like any other error, catches the error, and submits the task
to a different engine. This will repeat indefinitely, and a task will never be submitted
to a given engine a second time.
You can manually raise the
UnmetDependency yourself, but IPython has provided
some decorators for facilitating this behavior.
There are two decorators and a class used for functional dependencies:
In : import ipyparallel as ipp
The simplest sort of dependency is requiring that a Python module is available. The
@ipp.require decorator lets you define a function that will only run on engines where names
you specify are importable:
In : @ipp.require('numpy', 'zmq') ....: def myfunc(): ....: return dostuff()
Now, any time you apply
myfunc(), the task will only run on a machine that has
numpy and pyzmq available, and when
myfunc() is called, numpy and zmq will be imported.
You can also require specific objects, not just module names:
def foo(a): return a * a @ipp.require(foo) def bar(b): return foo(b) @ipp.require(bar) def baz(c, d): return bar(c) - bar(d) view.apply_sync(baz, 4, 5)
@ipp.depend decorator lets you decorate any function with any other function to
evaluate the dependency. The dependency function will be called at the start of the task,
and if it returns
False, then the dependency will be considered unmet, and the task
will be assigned to another engine. If the dependency returns anything other than
False, the rest of the task will continue.
In : def platform_specific(plat): ....: import sys ....: return sys.platform == plat In : @ipp.depend(platform_specific, 'darwin') ....: def mactask(): ....: do_mac_stuff() In : @ipp.depend(platform_specific, 'nt') ....: def wintask(): ....: do_windows_stuff()
In this case, any time you apply
mactask, it will only run on an macOS machine.
@ipp.depend is like
apply, in that it has a
@ipp.depend(f, *args, **kwargs)
You don’t have to use the decorators on your tasks, if for instance you may want
to run tasks with a single function but varying dependencies, you can directly construct
dependent object that the decorators use:
In : def mytask(*args): ....: dostuff() In : mactask = dependent(mytask, platform_specific, 'darwin') # this is the same as decorating the declaration of mytask with @ipp.depend # but you can do it again: In : wintask = dependent(mytask, platform_specific, 'nt') # in general: In : t = dependent(f, g, *dargs, **dkwargs) # is equivalent to: In : @ipp.depend(g, *dargs, **dkwargs) ....: def t(a, b, c): ....: # contents of f
Sometimes you want to restrict the time and/or location to run a given task as a function
of the time and/or location of other tasks. This is implemented via a subclass of
set, called a
Dependency. A Dependency is a set of
corresponding to tasks, and a few attributes to guide how to decide when the Dependency
has been met.
The switches we provide for interpreting whether a given dependency set has been met:
Whether the dependency is considered met if any of the dependencies are done, or only after all of them have finished. This is set by a Dependency’s
allboolean attribute, which defaults to
- success [default: True]
Whether to consider tasks that succeeded as fulfilling dependencies.
- failure [default]
Whether to consider tasks that failed as fulfilling dependencies. using
failure=True,success=Falseis useful for setting up cleanup tasks, to be run only when tasks have failed.
Sometimes you want to run a task after another, but only if that task succeeded. In this case,
success should be
failure should be
False. However sometimes you may
not care whether the task succeeds, and always want the second task to run, in which case you
success=failure=True. The default behavior is to only use successes.
There are other switches for interpretation that are made at the task level. These are
specified via keyword arguments to the view’s
You may want to run a task after a given set of dependencies have been run and/or run it where another set of dependencies are met. To support this, every task has an
afterdependency to restrict time, and a
followdependency to restrict destination.
You may also want to set a time-limit for how long the scheduler should wait before a task’s dependencies are met. This is done via a
timeout, which defaults to 0, which indicates that the task should never timeout. If the timeout is reached, and the scheduler still hasn’t been able to assign the task to an engine, the task will fail with a
Dependencies only work within the task scheduler. You cannot instruct a load-balanced task to run after a job submitted via the MUX interface.
The simplest form of Dependencies is with
all=True, success=True, failure=False. In these cases,
you can skip using Dependency objects, and pass msg_ids or AsyncResult objects as the
after keywords to
In : client.block = False In : ar = lview.apply(f, args, kwargs) In : ar2 = lview.apply(f2) In : with lview.temp_flags(after=[ar, ar2]): ....: ar3 = lview.apply(f3) In : with lview.temp_flags(follow=[ar], timeout=2.5) ....: ar4 = lview.apply(f3)
The schedulers do perform some analysis on graph dependencies to determine whether they
are not possible to be met. If the scheduler does discover that a dependency cannot be
met, then the task will fail with an
ImpossibleDependency error. This way, if the
scheduler realized that a task can never be run, it won’t sit indefinitely in the
scheduler clogging the pipeline.
The basic cases that are checked:
depending on nonexistent messages
followdependencies were run on more than one machine and
any dependencies failed and
all=True, success=True, failures=False
all dependencies failed and
all=False, success=True, failure=False
This analysis has not been proven to be rigorous, so it is likely possible for tasks to become impossible to run in obscure situations, so a timeout may be a good choice.
Retries and Resubmit#
Another flag for tasks is
retries. This is an integer, specifying how many times
a task should be resubmitted after failure. This is useful for tasks that should still run
if their engine was shutdown, or may have some statistical chance of failing. The default
is to not retry tasks.
Sometimes you may want to re-run a task. This could be because it failed for some reason, and
you have fixed the error, or because you want to restore the cluster to an interrupted state.
For this, the
Client has a
rc.resubmit() method. This takes one or more
msg_ids, and returns an
AsyncHubResult for the result(s). You cannot resubmit
a task that is pending - only those that have finished, either successful or unsuccessful.
There are a variety of valid ways to determine where jobs should be assigned in a
load-balancing situation. In IPython, we support several standard schemes, and
even make it easy to define your own. The scheme can be selected via the
argument to ipcontroller, or in the
of a controller config object.
The built-in routing schemes:
To select one of these schemes:
$ ipcontroller --scheme=<schemename> for instance: $ ipcontroller --scheme=lru
lru: Least Recently Used
Always assign work to the least-recently-used engine. A close relative of round-robin, it will be fair with respect to the number of tasks, agnostic with respect to runtime of each task.
plainrandom: Plain Random
Randomly picks an engine on which to run.
twobin: Two-Bin Random
Pick two engines at random, and use the LRU of the two. This is known to be better than plain random in many cases, but requires a small amount of computation.
leastload: Least Load
This is the default scheme
Always assign tasks to the engine with the fewest outstanding tasks (LRU breaks tie).
weighted: Weighted Two-Bin Random
Pick two engines at random using the number of outstanding tasks as inverse weights, and use the one with the lower load.
Tasks can be assigned greedily as they are submitted. If their dependencies are
met, they will be assigned to an engine right away, and multiple tasks can be
assigned to an engine at a given time. This limit is set with the
TaskScheduler.hwm (high water mark) configurable in your
ipcontroller_config.py config file, with:
# the most common choices are: c.TaskSheduler.hwm = 0 # (minimal latency, default in IPython < 0.13) # or c.TaskScheduler.hwm = 1 # (most-informed balancing, default in ≥ 0.13)
In IPython < 0.13, the default is 0, or no-limit. That is, there is no limit to the number of tasks that can be outstanding on a given engine. This greatly benefits the latency of execution, because network traffic can be hidden behind computation. However, this means that workload is assigned without knowledge of how long each task might take, and can result in poor load-balancing, particularly for submitting a collection of heterogeneous tasks all at once. You can limit this effect by setting hwm to a positive integer, 1 being maximum load-balancing (a task will never be waiting if there is an idle engine), and any larger number being a compromise between load-balancing and latency-hiding.
In practice, some users have been confused by having this optimization on by default, so the default value has been changed to 1 in IPython 0.13. This can be slower, but has more obvious behavior and won’t result in assigning too many tasks to some engines in heterogeneous cases.
Pure ZMQ Scheduler#
For maximum throughput, the ‘pure’ scheme is not Python at all, but a C-level
MonitoredQueue from PyZMQ, which uses a ZeroMQ
DEALER socket to perform all
load-balancing. This scheduler does not support any of the advanced features of the Python
Disabled features when using the ZMQ Scheduler:
- Engine unregistration
Task farming will be disabled if an engine unregisters. Further, if an engine is unregistered during computation, the scheduler may not recover.
Since there is no Python logic inside the Scheduler, routing decisions cannot be made based on message content.
- Early destination notification
The Python schedulers know which engine gets which task, and notify the Hub. This allows graceful handling of Engines coming and going. There is no way to know where ZeroMQ messages have gone, so there is no way to know what tasks are on which engine until they finish. This makes recovery from engine shutdown very difficult.
TODO: performance comparisons
LoadBalancedView has many more powerful features that allow quite a bit
of flexibility in how tasks are defined and run. The next places to look are
in the following classes:
The following is an overview of how to use these classes together:
Define some functions to be run as tasks
Submit your tasks to using the
apply()method of your
A demo of DAG Dependencies with NetworkX and IPython.