Non-copying sends and numpy arrays

When numpy arrays are passed as arguments to apply or via data-movement methods, they are not copied. This means that you must be careful if you are sending an array that you intend to work on. PyZMQ does allow you to track when a message has been sent so you can know when it is safe to edit the buffer, but IPython only allows for this.

It is also important to note that the non-copying receive of a message is read-only. That means that if you intend to work in-place on an array that you have sent or received, you must copy it. This is true for both numpy arrays sent to engines and numpy arrays retrieved as results.

The following will fail:

In [3]: A = numpy.zeros(2)

In [4]: def setter(a):
   ...:   a[0]=1
   ...:   return a

In [5]: rc[0].apply_sync(setter, A)
---------------------------------------------------------------------------
RuntimeError                              Traceback (most recent call last)<string> in <module>()
<ipython-input-12-c3e7afeb3075> in setter(a)
RuntimeError: array is not writeable

If you do need to edit the array in-place, just remember to copy the array if it's read-only. The :attr:`ndarray.flags.writeable` flag will tell you if you can write to an array.

In [3]: A = numpy.zeros(2)

In [4]: def setter(a):
   ...:     """only copy read-only arrays"""
   ...:     if not a.flags.writeable:
   ...:         a=a.copy()
   ...:     a[0]=1
   ...:     return a

In [5]: rc[0].apply_sync(setter, A)
Out[5]: array([ 1.,  0.])

# note that results will also be read-only:
In [6]: _.flags.writeable
Out[6]: False

If you want to safely edit an array in-place after sending it, you must use the track=True flag. IPython always performs non-copying sends of arrays, which return immediately. You must instruct IPython track those messages at send time in order to know for sure that the send has completed. AsyncResults have a :attr:`sent` property, and :meth:`wait_on_send` method for checking and waiting for 0MQ to finish with a buffer.

In [5]: A = numpy.random.random((1024,1024))

In [6]: view.track=True

In [7]: ar = view.apply_async(lambda x: 2*x, A)

In [8]: ar.sent
Out[8]: False

In [9]: ar.wait_on_send() # blocks until sent is True

What is sendable?

If IPython doesn't know what to do with an object, it will pickle it. There is a short list of objects that are not pickled: buffers, str/bytes objects, and numpy arrays. These are handled specially by IPython in order to prevent the copying of data. Sending bytes or numpy arrays will result in exactly zero in-memory copies of your data (unless the data is very small).

If you have an object that provides a Python buffer interface, then you can always send that buffer without copying - and reconstruct the object on the other side in your own code. It is possible that the object reconstruction will become extensible, so you can add your own non-copying types, but this does not yet exist.

def f(a):
    def inner():
        # inner will have a closure
        return a
    return inner

f1 = f(1)
f2 = f(2)
f1() # returns 1
f2() # returns 2

def g(b):
    # note that `b` is not referenced in inner's scope
    def inner():
        # this inner will *not* have a closure
        return a
    return inner
g1 = g(1)
g2 = g(2)
g1() # raises NameError on 'a'
a=5
g2() # returns 5

In [10]: view.apply(lambda : a)

# is equivalent to
In [11]: view.pull('a')

Apply

The principal method of remote execution is :meth:`apply`, of :class:`~IPython.parallel.client.view.View` objects. The Client provides the full execution and communication API for engines via its low-level :meth:`send_apply_message` method, which is used by all higher level methods of its Views.

f : function

The fuction to be called remotely

args : tuple/list

The positional arguments passed to f

kwargs : dict

The keyword arguments passed to f

flags for all views:

block : bool (default: view.block)

Whether to wait for the result, or return immediately.

False:

returns AsyncResult

True:

returns actual result(s) of f(*args, **kwargs)

if multiple targets:

list of results, matching targets

track : bool [default view.track]

whether to track non-copying sends.

targets : int,list of ints, 'all', None [default view.targets]

Specify the destination of the job.

if 'all' or None:

Run on all active engines

if list:

Run on each specified engine

if int:

Run on single engine

Note that :class:`LoadBalancedView` uses targets to restrict possible destinations. LoadBalanced calls will always execute in just one location.

flags only in LoadBalancedViews:

after : Dependency or collection of msg_ids

Only for load-balanced execution (targets=None) Specify a list of msg_ids as a time-based dependency. This job will only be run after the dependencies have been met.

follow : Dependency or collection of msg_ids

Only for load-balanced execution (targets=None) Specify a list of msg_ids as a location-based dependency. This job will only be run on an engine where this dependency is met.

timeout : float/int or None

Only for load-balanced execution (targets=None) Specify an amount of time (in seconds) for the scheduler to wait for dependencies to be met before failing with a DependencyTimeout.

execute and run

For executing strings of Python code, :class:`DirectView` 's also provide an :meth:`execute` and a :meth:`run` method, which rather than take functions and arguments, take simple strings. execute simply takes a string of Python code to execute, and sends it to the Engine(s). run is the same as execute, but for a file, rather than a string. It is simply a wrapper that does something very similar to execute(open(f).read()).

DirectView

The :class:`.DirectView` is the class for the IPython :ref:`Multiplexing Interface <parallel_multiengine>`.

In [2]: rc[0]
Out[2]: <DirectView 0>

In [2]: rc[0,1,2]
Out[2]: <DirectView [0,1,2]>

# negative index
In [2]: rc[-1]
Out[2]: <DirectView 3>

# or slicing:
In [3]: rc[::2]
Out[3]: <DirectView [0,2]>

In [2]: rc[rc.ids[-1]]
Out[2]: <DirectView 3>

In [3]: rc[rc.ids[::2]]
Out[3]: <DirectView [0,2]>

In [5]: import os

In [6]: dview.apply_sync(os.getpid)
Out[6]: [1354, 1356, 1358, 1360]

In [5]: import socket

In [6]: dview.apply_sync(socket.gethostname)
Out[6]: ['tesla', 'tesla', 'edison', 'edison', 'edison']

In [51]: dview['a']=['foo','bar']

In [52]: dview['a']
Out[52]: [ ['foo', 'bar'], ['foo', 'bar'], ['foo', 'bar'], ['foo', 'bar'] ]

Scatter and gather

Sometimes it is useful to partition a sequence and push the partitions to different engines. In MPI language, this is know as scatter/gather and we follow that terminology. However, it is important to remember that in IPython's :class:`Client` class, :meth:`scatter` is from the interactive IPython session to the engines and :meth:`gather` is from the engines back to the interactive IPython session. For scatter/gather operations between engines, MPI should be used:

In [58]: dview.scatter('a',range(16))
Out[58]: [None,None,None,None]

In [59]: dview['a']
Out[59]: [ [0, 1, 2, 3], [4, 5, 6, 7], [8, 9, 10, 11], [12, 13, 14, 15] ]

In [60]: dview.gather('a')
Out[60]: [0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15]

Push and pull

:meth:`~IPython.parallel.client.view.DirectView.push`

:meth:`~IPython.parallel.client.view.DirectView.pull`

LoadBalancedView

The :class:`~.LoadBalancedView` is the class for load-balanced execution via the task scheduler. These views always run tasks on exactly one engine, but let the scheduler determine where that should be, allowing load-balancing of tasks. The LoadBalancedView does allow you to specify restrictions on where and when tasks can execute, for more complicated load-balanced workflows.

AsyncResults

Our primary representation of the results of remote execution is the :class:`~.AsyncResult` object, based on the object of the same name in the built-in :mod:`multiprocessing.pool` module. Our version provides a superset of that interface.

The basic principle of the AsyncResult is the encapsulation of one or more results not yet completed. Execution methods (including data movement, such as push/pull) will all return AsyncResults when block=False.

The mp.pool.AsyncResult interface

The basic interface of the AsyncResult is exactly that of the AsyncResult in :mod:`multiprocessing.pool`, and consists of four methods:

The stdlib AsyncResult spec

While an AsyncResult is not done, you can check on it with its :meth:`ready` method, which will return whether the AR is done. You can also wait on an AsyncResult with its :meth:`wait` method. This method blocks until the result arrives. If you don't want to wait forever, you can pass a timeout (in seconds) as an argument to :meth:`wait`. :meth:`wait` will always return None, and should never raise an error.

:meth:`ready` and :meth:`wait` are insensitive to the success or failure of the call. After a result is done, :meth:`successful` will tell you whether the call completed without raising an exception.

If you actually want the result of the call, you can use :meth:`get`. Initially, :meth:`get` behaves just like :meth:`wait`, in that it will block until the result is ready, or until a timeout is met. However, unlike :meth:`wait`, :meth:`get` will raise a :exc:`TimeoutError` if the timeout is reached and the result is still not ready. If the result arrives before the timeout is reached, then :meth:`get` will return the result itself if no exception was raised, and will raise an exception if there was.

Here is where we start to expand on the multiprocessing interface. Rather than raising the original exception, a RemoteError will be raised, encapsulating the remote exception with some metadata. If the AsyncResult represents multiple calls (e.g. any time targets is plural), then a CompositeError, a subclass of RemoteError, will be raised.