Parallel map

To load-balance :meth:`map`,simply use a LoadBalancedView:

In [62]: lview.block = True

In [63]: serial_result = map(lambda x:x**10, range(32))

In [64]: parallel_result = lview.map(lambda x:x**10, range(32))

In [65]: serial_result==parallel_result
Out[65]: True

Parallel function decorator

Parallel functions are just like normal function, but they can be called on sequences and in parallel. The multiengine interface provides a decorator that turns any Python function into a parallel function:

In [10]: @lview.parallel()
   ....: def f(x):
   ....:     return 10.0*x**4
   ....:

In [11]: f.map(range(32))    # this is done in parallel
Out[11]: [0.0,10.0,160.0,...]

Functional Dependencies

Functional dependencies are used to determine whether a given engine is capable of running a particular task. This is implemented via a special :class:`Exception` class, :class:`UnmetDependency`, found in IPython.parallel.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 :class:`UnmetDependency` yourself, but IPython has provided some decorators for facilitating this behavior.

There are two decorators and a class used for functional dependencies:

In [9]: from IPython.parallel import depend, require, dependent

In [10]: @require('numpy', 'zmq')
   ....: def myfunc():
   ....:     return dostuff()

In [10]: def platform_specific(plat):
   ....:    import sys
   ....:    return sys.platform == plat

In [11]: @depend(platform_specific, 'darwin')
   ....: def mactask():
   ....:    do_mac_stuff()

In [12]: @depend(platform_specific, 'nt')
   ....: def wintask():
   ....:    do_windows_stuff()

Graph Dependencies

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 :class:`set`, called a :class:`Dependency`. A Dependency is just a set of msg_ids 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:

any|all

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 :attr:`all` boolean attribute, which defaults to True.

success [default: True]

Whether to consider tasks that succeeded as fulfilling dependencies.

failure [default : False]

Whether to consider tasks that failed as fulfilling dependencies. using failure=True,success=False is 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 True and 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 should use 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 client's :meth:`apply` method.

after,follow

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 after dependency to restrict time, and a follow dependency to restrict destination.

timeout

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 :class:`DependencyTimeout`.

The simplest form of Dependencies is with all=True,success=True,failure=False. In these cases, you can skip using Dependency objects, and just pass msg_ids or AsyncResult objects as the follow and after keywords to :meth:`client.apply`:

In [14]: client.block=False

In [15]: ar = lview.apply(f, args, kwargs)

In [16]: ar2 = lview.apply(f2)

In [17]: with lview.temp_flags(after=[ar,ar2]):
   ....:    ar3 = lview.apply(f3)

In [18]: with lview.temp_flags(follow=[ar], timeout=2.5)
   ....:    ar4 = lview.apply(f3)

Retries

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.

Resubmit

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 :class:`Client` has a :meth:`rc.resubmit` method. This simply takes one or more msg_ids, and returns an :class:`AsyncHubResult` for the result(s). You cannot resubmit a task that is pending - only those that have finished, either successful or unsuccessful.

Greedy Assignment

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 :file:`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 :class:`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 :class:`.Scheduler`.

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.

• Dependencies

  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.