|
|
# encoding: utf-8
|
|
|
# -*- test-case-name: IPython.kernel.test.test_multiengineclient -*-
|
|
|
|
|
|
"""General Classes for IMultiEngine clients."""
|
|
|
|
|
|
__docformat__ = "restructuredtext en"
|
|
|
|
|
|
#-------------------------------------------------------------------------------
|
|
|
# Copyright (C) 2008 The IPython Development Team
|
|
|
#
|
|
|
# Distributed under the terms of the BSD License. The full license is in
|
|
|
# the file COPYING, distributed as part of this software.
|
|
|
#-------------------------------------------------------------------------------
|
|
|
|
|
|
#-------------------------------------------------------------------------------
|
|
|
# Imports
|
|
|
#-------------------------------------------------------------------------------
|
|
|
|
|
|
import sys
|
|
|
import linecache
|
|
|
import warnings
|
|
|
|
|
|
from twisted.python import components
|
|
|
from twisted.python.failure import Failure
|
|
|
from zope.interface import Interface, implements, Attribute
|
|
|
|
|
|
from IPython.utils.coloransi import TermColors
|
|
|
|
|
|
from IPython.kernel.twistedutil import blockingCallFromThread
|
|
|
from IPython.kernel import error
|
|
|
from IPython.kernel.parallelfunction import ParallelFunction
|
|
|
from IPython.kernel.mapper import (
|
|
|
MultiEngineMapper,
|
|
|
IMultiEngineMapperFactory,
|
|
|
IMapper
|
|
|
)
|
|
|
|
|
|
from IPython.kernel.multiengine import IFullSynchronousMultiEngine
|
|
|
|
|
|
|
|
|
#-------------------------------------------------------------------------------
|
|
|
# Pending Result things
|
|
|
#-------------------------------------------------------------------------------
|
|
|
|
|
|
class IPendingResult(Interface):
|
|
|
"""A representation of a result that is pending.
|
|
|
|
|
|
This class is similar to Twisted's `Deferred` object, but is designed to be
|
|
|
used in a synchronous context.
|
|
|
"""
|
|
|
|
|
|
result_id=Attribute("ID of the deferred on the other side")
|
|
|
client=Attribute("A client that I came from")
|
|
|
r=Attribute("An attribute that is a property that calls and returns get_result")
|
|
|
|
|
|
def get_result(default=None, block=True):
|
|
|
"""
|
|
|
Get a result that is pending.
|
|
|
|
|
|
:Parameters:
|
|
|
default
|
|
|
The value to return if the result is not ready.
|
|
|
block : boolean
|
|
|
Should I block for the result.
|
|
|
|
|
|
:Returns: The actual result or the default value.
|
|
|
"""
|
|
|
|
|
|
def add_callback(f, *args, **kwargs):
|
|
|
"""
|
|
|
Add a callback that is called with the result.
|
|
|
|
|
|
If the original result is foo, adding a callback will cause
|
|
|
f(foo, *args, **kwargs) to be returned instead. If multiple
|
|
|
callbacks are registered, they are chained together: the result of
|
|
|
one is passed to the next and so on.
|
|
|
|
|
|
Unlike Twisted's Deferred object, there is no errback chain. Thus
|
|
|
any exception raised will not be caught and handled. User must
|
|
|
catch these by hand when calling `get_result`.
|
|
|
"""
|
|
|
|
|
|
|
|
|
class PendingResult(object):
|
|
|
"""A representation of a result that is not yet ready.
|
|
|
|
|
|
A user should not create a `PendingResult` instance by hand.
|
|
|
|
|
|
Methods
|
|
|
=======
|
|
|
|
|
|
* `get_result`
|
|
|
* `add_callback`
|
|
|
|
|
|
Properties
|
|
|
==========
|
|
|
* `r`
|
|
|
"""
|
|
|
|
|
|
def __init__(self, client, result_id):
|
|
|
"""Create a PendingResult with a result_id and a client instance.
|
|
|
|
|
|
The client should implement `_getPendingResult(result_id, block)`.
|
|
|
"""
|
|
|
self.client = client
|
|
|
self.result_id = result_id
|
|
|
self.called = False
|
|
|
self.raised = False
|
|
|
self.callbacks = []
|
|
|
|
|
|
def get_result(self, default=None, block=True):
|
|
|
"""Get a result that is pending.
|
|
|
|
|
|
This method will connect to an IMultiEngine adapted controller
|
|
|
and see if the result is ready. If the action triggers an exception
|
|
|
raise it and record it. This method records the result/exception once it is
|
|
|
retrieved. Calling `get_result` again will get this cached result or will
|
|
|
re-raise the exception. The .r attribute is a property that calls
|
|
|
`get_result` with block=True.
|
|
|
|
|
|
:Parameters:
|
|
|
default
|
|
|
The value to return if the result is not ready.
|
|
|
block : boolean
|
|
|
Should I block for the result.
|
|
|
|
|
|
:Returns: The actual result or the default value.
|
|
|
"""
|
|
|
|
|
|
if self.called:
|
|
|
if self.raised:
|
|
|
raise self.result[0], self.result[1], self.result[2]
|
|
|
else:
|
|
|
return self.result
|
|
|
try:
|
|
|
result = self.client.get_pending_deferred(self.result_id, block)
|
|
|
except error.ResultNotCompleted:
|
|
|
return default
|
|
|
except:
|
|
|
# Reraise other error, but first record them so they can be reraised
|
|
|
# later if .r or get_result is called again.
|
|
|
self.result = sys.exc_info()
|
|
|
self.called = True
|
|
|
self.raised = True
|
|
|
raise
|
|
|
else:
|
|
|
for cb in self.callbacks:
|
|
|
result = cb[0](result, *cb[1], **cb[2])
|
|
|
self.result = result
|
|
|
self.called = True
|
|
|
return result
|
|
|
|
|
|
def add_callback(self, f, *args, **kwargs):
|
|
|
"""Add a callback that is called with the result.
|
|
|
|
|
|
If the original result is result, adding a callback will cause
|
|
|
f(result, *args, **kwargs) to be returned instead. If multiple
|
|
|
callbacks are registered, they are chained together: the result of
|
|
|
one is passed to the next and so on.
|
|
|
|
|
|
Unlike Twisted's Deferred object, there is no errback chain. Thus
|
|
|
any exception raised will not be caught and handled. User must
|
|
|
catch these by hand when calling `get_result`.
|
|
|
"""
|
|
|
assert callable(f)
|
|
|
self.callbacks.append((f, args, kwargs))
|
|
|
|
|
|
def __cmp__(self, other):
|
|
|
if self.result_id < other.result_id:
|
|
|
return -1
|
|
|
else:
|
|
|
return 1
|
|
|
|
|
|
def _get_r(self):
|
|
|
return self.get_result(block=True)
|
|
|
|
|
|
r = property(_get_r)
|
|
|
"""This property is a shortcut to a `get_result(block=True)`."""
|
|
|
|
|
|
|
|
|
#-------------------------------------------------------------------------------
|
|
|
# Pretty printing wrappers for certain lists
|
|
|
#-------------------------------------------------------------------------------
|
|
|
|
|
|
class ResultList(list):
|
|
|
"""A subclass of list that pretty prints the output of `execute`/`get_result`."""
|
|
|
|
|
|
def __repr__(self):
|
|
|
output = []
|
|
|
# These colored prompts were not working on Windows
|
|
|
if sys.platform == 'win32':
|
|
|
blue = normal = red = green = ''
|
|
|
else:
|
|
|
blue = TermColors.Blue
|
|
|
normal = TermColors.Normal
|
|
|
red = TermColors.Red
|
|
|
green = TermColors.Green
|
|
|
output.append("<Results List>\n")
|
|
|
for cmd in self:
|
|
|
if isinstance(cmd, Failure):
|
|
|
output.append(cmd)
|
|
|
else:
|
|
|
target = cmd.get('id',None)
|
|
|
cmd_num = cmd.get('number',None)
|
|
|
cmd_stdin = cmd.get('input',{}).get('translated','No Input')
|
|
|
cmd_stdout = cmd.get('stdout', None)
|
|
|
cmd_stderr = cmd.get('stderr', None)
|
|
|
output.append("%s[%i]%s In [%i]:%s %s\n" % \
|
|
|
(green, target,
|
|
|
blue, cmd_num, normal, cmd_stdin))
|
|
|
if cmd_stdout:
|
|
|
output.append("%s[%i]%s Out[%i]:%s %s\n" % \
|
|
|
(green, target,
|
|
|
red, cmd_num, normal, cmd_stdout))
|
|
|
if cmd_stderr:
|
|
|
output.append("%s[%i]%s Err[%i]:\n%s %s" % \
|
|
|
(green, target,
|
|
|
red, cmd_num, normal, cmd_stderr))
|
|
|
return ''.join(output)
|
|
|
|
|
|
|
|
|
def wrapResultList(result):
|
|
|
"""A function that wraps the output of `execute`/`get_result` -> `ResultList`."""
|
|
|
if len(result) == 0:
|
|
|
result = [result]
|
|
|
return ResultList(result)
|
|
|
|
|
|
|
|
|
class QueueStatusList(list):
|
|
|
"""A subclass of list that pretty prints the output of `queue_status`."""
|
|
|
|
|
|
def __repr__(self):
|
|
|
output = []
|
|
|
output.append("<Queue Status List>\n")
|
|
|
for e in self:
|
|
|
output.append("Engine: %s\n" % repr(e[0]))
|
|
|
output.append(" Pending: %s\n" % repr(e[1]['pending']))
|
|
|
for q in e[1]['queue']:
|
|
|
output.append(" Command: %s\n" % repr(q))
|
|
|
return ''.join(output)
|
|
|
|
|
|
|
|
|
#-------------------------------------------------------------------------------
|
|
|
# InteractiveMultiEngineClient
|
|
|
#-------------------------------------------------------------------------------
|
|
|
|
|
|
class InteractiveMultiEngineClient(object):
|
|
|
"""A mixin class that add a few methods to a multiengine client.
|
|
|
|
|
|
The methods in this mixin class are designed for interactive usage.
|
|
|
"""
|
|
|
|
|
|
def activate(self):
|
|
|
"""Make this `MultiEngineClient` active for parallel magic commands.
|
|
|
|
|
|
IPython has a magic command syntax to work with `MultiEngineClient` objects.
|
|
|
In a given IPython session there is a single active one. While
|
|
|
there can be many `MultiEngineClient` created and used by the user,
|
|
|
there is only one active one. The active `MultiEngineClient` is used whenever
|
|
|
the magic commands %px and %autopx are used.
|
|
|
|
|
|
The activate() method is called on a given `MultiEngineClient` to make it
|
|
|
active. Once this has been done, the magic commands can be used.
|
|
|
"""
|
|
|
|
|
|
try:
|
|
|
__IPYTHON__.activeController = self
|
|
|
except NameError:
|
|
|
print "The IPython Controller magics only work within IPython."
|
|
|
|
|
|
def __setitem__(self, key, value):
|
|
|
"""Add a dictionary interface for pushing/pulling.
|
|
|
|
|
|
This functions as a shorthand for `push`.
|
|
|
|
|
|
:Parameters:
|
|
|
key : str
|
|
|
What to call the remote object.
|
|
|
value : object
|
|
|
The local Python object to push.
|
|
|
"""
|
|
|
targets, block = self._findTargetsAndBlock()
|
|
|
return self.push({key:value}, targets=targets, block=block)
|
|
|
|
|
|
def __getitem__(self, key):
|
|
|
"""Add a dictionary interface for pushing/pulling.
|
|
|
|
|
|
This functions as a shorthand to `pull`.
|
|
|
|
|
|
:Parameters:
|
|
|
- `key`: A string representing the key.
|
|
|
"""
|
|
|
if isinstance(key, str):
|
|
|
targets, block = self._findTargetsAndBlock()
|
|
|
return self.pull(key, targets=targets, block=block)
|
|
|
else:
|
|
|
raise TypeError("__getitem__ only takes strs")
|
|
|
|
|
|
def __len__(self):
|
|
|
"""Return the number of available engines."""
|
|
|
return len(self.get_ids())
|
|
|
|
|
|
#---------------------------------------------------------------------------
|
|
|
# Make this a context manager for with
|
|
|
#---------------------------------------------------------------------------
|
|
|
|
|
|
def findsource_file(self,f):
|
|
|
linecache.checkcache()
|
|
|
s = findsource(f.f_code) # findsource is not defined!
|
|
|
lnum = f.f_lineno
|
|
|
wsource = s[0][f.f_lineno:]
|
|
|
return strip_whitespace(wsource)
|
|
|
|
|
|
def findsource_ipython(self,f):
|
|
|
from IPython.core import ipapi
|
|
|
self.ip = ipapi.get()
|
|
|
wsource = [l+'\n' for l in
|
|
|
self.ip.IP.input_hist_raw[-1].splitlines()[1:]]
|
|
|
return strip_whitespace(wsource)
|
|
|
|
|
|
def __enter__(self):
|
|
|
f = sys._getframe(1)
|
|
|
local_ns = f.f_locals
|
|
|
global_ns = f.f_globals
|
|
|
if f.f_code.co_filename == '<ipython console>':
|
|
|
s = self.findsource_ipython(f)
|
|
|
else:
|
|
|
s = self.findsource_file(f)
|
|
|
|
|
|
self._with_context_result = self.execute(s)
|
|
|
|
|
|
def __exit__ (self, etype, value, tb):
|
|
|
if issubclass(etype,error.StopLocalExecution):
|
|
|
return True
|
|
|
|
|
|
|
|
|
def remote():
|
|
|
m = 'Special exception to stop local execution of parallel code.'
|
|
|
raise error.StopLocalExecution(m)
|
|
|
|
|
|
def strip_whitespace(source):
|
|
|
# Expand tabs to avoid any confusion.
|
|
|
wsource = [l.expandtabs(4) for l in source]
|
|
|
# Detect the indentation level
|
|
|
done = False
|
|
|
for line in wsource:
|
|
|
if line.isspace():
|
|
|
continue
|
|
|
for col,char in enumerate(line):
|
|
|
if char != ' ':
|
|
|
done = True
|
|
|
break
|
|
|
if done:
|
|
|
break
|
|
|
# Now we know how much leading space there is in the code. Next, we
|
|
|
# extract up to the first line that has less indentation.
|
|
|
# WARNINGS: we skip comments that may be misindented, but we do NOT yet
|
|
|
# detect triple quoted strings that may have flush left text.
|
|
|
for lno,line in enumerate(wsource):
|
|
|
lead = line[:col]
|
|
|
if lead.isspace():
|
|
|
continue
|
|
|
else:
|
|
|
if not lead.lstrip().startswith('#'):
|
|
|
break
|
|
|
# The real 'with' source is up to lno
|
|
|
src_lines = [l[col:] for l in wsource[:lno+1]]
|
|
|
|
|
|
# Finally, check that the source's first non-comment line begins with the
|
|
|
# special call 'remote()'
|
|
|
for nline,line in enumerate(src_lines):
|
|
|
if line.isspace() or line.startswith('#'):
|
|
|
continue
|
|
|
if 'remote()' in line:
|
|
|
break
|
|
|
else:
|
|
|
raise ValueError('remote() call missing at the start of code')
|
|
|
src = ''.join(src_lines[nline+1:])
|
|
|
#print 'SRC:\n<<<<<<<>>>>>>>\n%s<<<<<>>>>>>' % src # dbg
|
|
|
return src
|
|
|
|
|
|
|
|
|
#-------------------------------------------------------------------------------
|
|
|
# The top-level MultiEngine client adaptor
|
|
|
#-------------------------------------------------------------------------------
|
|
|
|
|
|
|
|
|
_prop_warn = """\
|
|
|
|
|
|
We are currently refactoring the task dependency system. This might
|
|
|
involve the removal of this method and other methods related to engine
|
|
|
properties. Please see the docstrings for IPython.kernel.TaskRejectError
|
|
|
for more information."""
|
|
|
|
|
|
|
|
|
class IFullBlockingMultiEngineClient(Interface):
|
|
|
pass
|
|
|
|
|
|
|
|
|
class FullBlockingMultiEngineClient(InteractiveMultiEngineClient):
|
|
|
"""
|
|
|
A blocking client to the `IMultiEngine` controller interface.
|
|
|
|
|
|
This class allows users to use a set of engines for a parallel
|
|
|
computation through the `IMultiEngine` interface. In this interface,
|
|
|
each engine has a specific id (an int) that is used to refer to the
|
|
|
engine, run code on it, etc.
|
|
|
"""
|
|
|
|
|
|
implements(
|
|
|
IFullBlockingMultiEngineClient,
|
|
|
IMultiEngineMapperFactory,
|
|
|
IMapper
|
|
|
)
|
|
|
|
|
|
def __init__(self, smultiengine):
|
|
|
self.smultiengine = smultiengine
|
|
|
self.block = True
|
|
|
self.targets = 'all'
|
|
|
|
|
|
def _findBlock(self, block=None):
|
|
|
if block is None:
|
|
|
return self.block
|
|
|
else:
|
|
|
if block in (True, False):
|
|
|
return block
|
|
|
else:
|
|
|
raise ValueError("block must be True or False")
|
|
|
|
|
|
def _findTargets(self, targets=None):
|
|
|
if targets is None:
|
|
|
return self.targets
|
|
|
else:
|
|
|
if not isinstance(targets, (str,list,tuple,int)):
|
|
|
raise ValueError("targets must be a str, list, tuple or int")
|
|
|
return targets
|
|
|
|
|
|
def _findTargetsAndBlock(self, targets=None, block=None):
|
|
|
return self._findTargets(targets), self._findBlock(block)
|
|
|
|
|
|
def _blockFromThread(self, function, *args, **kwargs):
|
|
|
block = kwargs.get('block', None)
|
|
|
if block is None:
|
|
|
raise error.MissingBlockArgument("'block' keyword argument is missing")
|
|
|
result = blockingCallFromThread(function, *args, **kwargs)
|
|
|
if not block:
|
|
|
result = PendingResult(self, result)
|
|
|
return result
|
|
|
|
|
|
def get_pending_deferred(self, deferredID, block):
|
|
|
return blockingCallFromThread(self.smultiengine.get_pending_deferred, deferredID, block)
|
|
|
|
|
|
def barrier(self, pendingResults):
|
|
|
"""Synchronize a set of `PendingResults`.
|
|
|
|
|
|
This method is a synchronization primitive that waits for a set of
|
|
|
`PendingResult` objects to complete. More specifically, barier does
|
|
|
the following.
|
|
|
|
|
|
* The `PendingResult`s are sorted by result_id.
|
|
|
* The `get_result` method is called for each `PendingResult` sequentially
|
|
|
with block=True.
|
|
|
* If a `PendingResult` gets a result that is an exception, it is
|
|
|
trapped and can be re-raised later by calling `get_result` again.
|
|
|
* The `PendingResult`s are flushed from the controller.
|
|
|
|
|
|
After barrier has been called on a `PendingResult`, its results can
|
|
|
be retrieved by calling `get_result` again or accesing the `r` attribute
|
|
|
of the instance.
|
|
|
"""
|
|
|
|
|
|
# Convert to list for sorting and check class type
|
|
|
prList = list(pendingResults)
|
|
|
for pr in prList:
|
|
|
if not isinstance(pr, PendingResult):
|
|
|
raise error.NotAPendingResult("Objects passed to barrier must be PendingResult instances")
|
|
|
|
|
|
# Sort the PendingResults so they are in order
|
|
|
prList.sort()
|
|
|
# Block on each PendingResult object
|
|
|
for pr in prList:
|
|
|
try:
|
|
|
result = pr.get_result(block=True)
|
|
|
except Exception:
|
|
|
pass
|
|
|
|
|
|
def flush(self):
|
|
|
"""
|
|
|
Clear all pending deferreds/results from the controller.
|
|
|
|
|
|
For each `PendingResult` that is created by this client, the controller
|
|
|
holds on to the result for that `PendingResult`. This can be a problem
|
|
|
if there are a large number of `PendingResult` objects that are created.
|
|
|
|
|
|
Once the result of the `PendingResult` has been retrieved, the result
|
|
|
is removed from the controller, but if a user doesn't get a result (
|
|
|
they just ignore the `PendingResult`) the result is kept forever on the
|
|
|
controller. This method allows the user to clear out all un-retrieved
|
|
|
results on the controller.
|
|
|
"""
|
|
|
r = blockingCallFromThread(self.smultiengine.clear_pending_deferreds)
|
|
|
return r
|
|
|
|
|
|
clear_pending_results = flush
|
|
|
|
|
|
#---------------------------------------------------------------------------
|
|
|
# IEngineMultiplexer related methods
|
|
|
#---------------------------------------------------------------------------
|
|
|
|
|
|
def execute(self, lines, targets=None, block=None):
|
|
|
"""
|
|
|
Execute code on a set of engines.
|
|
|
|
|
|
:Parameters:
|
|
|
lines : str
|
|
|
The Python code to execute as a string
|
|
|
targets : id or list of ids
|
|
|
The engine to use for the execution
|
|
|
block : boolean
|
|
|
If False, this method will return the actual result. If False,
|
|
|
a `PendingResult` is returned which can be used to get the result
|
|
|
at a later time.
|
|
|
"""
|
|
|
targets, block = self._findTargetsAndBlock(targets, block)
|
|
|
result = blockingCallFromThread(self.smultiengine.execute, lines,
|
|
|
targets=targets, block=block)
|
|
|
if block:
|
|
|
result = ResultList(result)
|
|
|
else:
|
|
|
result = PendingResult(self, result)
|
|
|
result.add_callback(wrapResultList)
|
|
|
return result
|
|
|
|
|
|
def push(self, namespace, targets=None, block=None):
|
|
|
"""
|
|
|
Push a dictionary of keys and values to engines namespace.
|
|
|
|
|
|
Each engine has a persistent namespace. This method is used to push
|
|
|
Python objects into that namespace.
|
|
|
|
|
|
The objects in the namespace must be pickleable.
|
|
|
|
|
|
:Parameters:
|
|
|
namespace : dict
|
|
|
A dict that contains Python objects to be injected into
|
|
|
the engine persistent namespace.
|
|
|
targets : id or list of ids
|
|
|
The engine to use for the execution
|
|
|
block : boolean
|
|
|
If False, this method will return the actual result. If False,
|
|
|
a `PendingResult` is returned which can be used to get the result
|
|
|
at a later time.
|
|
|
"""
|
|
|
targets, block = self._findTargetsAndBlock(targets, block)
|
|
|
return self._blockFromThread(self.smultiengine.push, namespace,
|
|
|
targets=targets, block=block)
|
|
|
|
|
|
def pull(self, keys, targets=None, block=None):
|
|
|
"""
|
|
|
Pull Python objects by key out of engines namespaces.
|
|
|
|
|
|
:Parameters:
|
|
|
keys : str or list of str
|
|
|
The names of the variables to be pulled
|
|
|
targets : id or list of ids
|
|
|
The engine to use for the execution
|
|
|
block : boolean
|
|
|
If False, this method will return the actual result. If False,
|
|
|
a `PendingResult` is returned which can be used to get the result
|
|
|
at a later time.
|
|
|
"""
|
|
|
targets, block = self._findTargetsAndBlock(targets, block)
|
|
|
return self._blockFromThread(self.smultiengine.pull, keys, targets=targets, block=block)
|
|
|
|
|
|
def push_function(self, namespace, targets=None, block=None):
|
|
|
"""
|
|
|
Push a Python function to an engine.
|
|
|
|
|
|
This method is used to push a Python function to an engine. This
|
|
|
method can then be used in code on the engines. Closures are not supported.
|
|
|
|
|
|
:Parameters:
|
|
|
namespace : dict
|
|
|
A dict whose values are the functions to be pushed. The keys give
|
|
|
that names that the function will appear as in the engines
|
|
|
namespace.
|
|
|
targets : id or list of ids
|
|
|
The engine to use for the execution
|
|
|
block : boolean
|
|
|
If False, this method will return the actual result. If False,
|
|
|
a `PendingResult` is returned which can be used to get the result
|
|
|
at a later time.
|
|
|
"""
|
|
|
targets, block = self._findTargetsAndBlock(targets, block)
|
|
|
return self._blockFromThread(self.smultiengine.push_function, namespace, targets=targets, block=block)
|
|
|
|
|
|
def pull_function(self, keys, targets=None, block=None):
|
|
|
"""
|
|
|
Pull a Python function from an engine.
|
|
|
|
|
|
This method is used to pull a Python function from an engine.
|
|
|
Closures are not supported.
|
|
|
|
|
|
:Parameters:
|
|
|
keys : str or list of str
|
|
|
The names of the functions to be pulled
|
|
|
targets : id or list of ids
|
|
|
The engine to use for the execution
|
|
|
block : boolean
|
|
|
If False, this method will return the actual result. If False,
|
|
|
a `PendingResult` is returned which can be used to get the result
|
|
|
at a later time.
|
|
|
"""
|
|
|
targets, block = self._findTargetsAndBlock(targets, block)
|
|
|
return self._blockFromThread(self.smultiengine.pull_function, keys, targets=targets, block=block)
|
|
|
|
|
|
def push_serialized(self, namespace, targets=None, block=None):
|
|
|
targets, block = self._findTargetsAndBlock(targets, block)
|
|
|
return self._blockFromThread(self.smultiengine.push_serialized, namespace, targets=targets, block=block)
|
|
|
|
|
|
def pull_serialized(self, keys, targets=None, block=None):
|
|
|
targets, block = self._findTargetsAndBlock(targets, block)
|
|
|
return self._blockFromThread(self.smultiengine.pull_serialized, keys, targets=targets, block=block)
|
|
|
|
|
|
def get_result(self, i=None, targets=None, block=None):
|
|
|
"""
|
|
|
Get a previous result.
|
|
|
|
|
|
When code is executed in an engine, a dict is created and returned. This
|
|
|
method retrieves that dict for previous commands.
|
|
|
|
|
|
:Parameters:
|
|
|
i : int
|
|
|
The number of the result to get
|
|
|
targets : id or list of ids
|
|
|
The engine to use for the execution
|
|
|
block : boolean
|
|
|
If False, this method will return the actual result. If False,
|
|
|
a `PendingResult` is returned which can be used to get the result
|
|
|
at a later time.
|
|
|
"""
|
|
|
targets, block = self._findTargetsAndBlock(targets, block)
|
|
|
result = blockingCallFromThread(self.smultiengine.get_result, i, targets=targets, block=block)
|
|
|
if block:
|
|
|
result = ResultList(result)
|
|
|
else:
|
|
|
result = PendingResult(self, result)
|
|
|
result.add_callback(wrapResultList)
|
|
|
return result
|
|
|
|
|
|
def reset(self, targets=None, block=None):
|
|
|
"""
|
|
|
Reset an engine.
|
|
|
|
|
|
This method clears out the namespace of an engine.
|
|
|
|
|
|
:Parameters:
|
|
|
targets : id or list of ids
|
|
|
The engine to use for the execution
|
|
|
block : boolean
|
|
|
If False, this method will return the actual result. If False,
|
|
|
a `PendingResult` is returned which can be used to get the result
|
|
|
at a later time.
|
|
|
"""
|
|
|
targets, block = self._findTargetsAndBlock(targets, block)
|
|
|
return self._blockFromThread(self.smultiengine.reset, targets=targets, block=block)
|
|
|
|
|
|
def keys(self, targets=None, block=None):
|
|
|
"""
|
|
|
Get a list of all the variables in an engine's namespace.
|
|
|
|
|
|
:Parameters:
|
|
|
targets : id or list of ids
|
|
|
The engine to use for the execution
|
|
|
block : boolean
|
|
|
If False, this method will return the actual result. If False,
|
|
|
a `PendingResult` is returned which can be used to get the result
|
|
|
at a later time.
|
|
|
"""
|
|
|
targets, block = self._findTargetsAndBlock(targets, block)
|
|
|
return self._blockFromThread(self.smultiengine.keys, targets=targets, block=block)
|
|
|
|
|
|
def kill(self, controller=False, targets=None, block=None):
|
|
|
"""
|
|
|
Kill the engines and controller.
|
|
|
|
|
|
This method is used to stop the engine and controller by calling
|
|
|
`reactor.stop`.
|
|
|
|
|
|
:Parameters:
|
|
|
controller : boolean
|
|
|
If True, kill the engines and controller. If False, just the
|
|
|
engines
|
|
|
targets : id or list of ids
|
|
|
The engine to use for the execution
|
|
|
block : boolean
|
|
|
If False, this method will return the actual result. If False,
|
|
|
a `PendingResult` is returned which can be used to get the result
|
|
|
at a later time.
|
|
|
"""
|
|
|
targets, block = self._findTargetsAndBlock(targets, block)
|
|
|
return self._blockFromThread(self.smultiengine.kill, controller, targets=targets, block=block)
|
|
|
|
|
|
def clear_queue(self, targets=None, block=None):
|
|
|
"""
|
|
|
Clear out the controller's queue for an engine.
|
|
|
|
|
|
The controller maintains a queue for each engine. This clear it out.
|
|
|
|
|
|
:Parameters:
|
|
|
targets : id or list of ids
|
|
|
The engine to use for the execution
|
|
|
block : boolean
|
|
|
If False, this method will return the actual result. If False,
|
|
|
a `PendingResult` is returned which can be used to get the result
|
|
|
at a later time.
|
|
|
"""
|
|
|
targets, block = self._findTargetsAndBlock(targets, block)
|
|
|
return self._blockFromThread(self.smultiengine.clear_queue, targets=targets, block=block)
|
|
|
|
|
|
def queue_status(self, targets=None, block=None):
|
|
|
"""
|
|
|
Get the status of an engines queue.
|
|
|
|
|
|
:Parameters:
|
|
|
targets : id or list of ids
|
|
|
The engine to use for the execution
|
|
|
block : boolean
|
|
|
If False, this method will return the actual result. If False,
|
|
|
a `PendingResult` is returned which can be used to get the result
|
|
|
at a later time.
|
|
|
"""
|
|
|
targets, block = self._findTargetsAndBlock(targets, block)
|
|
|
return self._blockFromThread(self.smultiengine.queue_status, targets=targets, block=block)
|
|
|
|
|
|
def set_properties(self, properties, targets=None, block=None):
|
|
|
warnings.warn(_prop_warn)
|
|
|
targets, block = self._findTargetsAndBlock(targets, block)
|
|
|
return self._blockFromThread(self.smultiengine.set_properties, properties, targets=targets, block=block)
|
|
|
|
|
|
def get_properties(self, keys=None, targets=None, block=None):
|
|
|
warnings.warn(_prop_warn)
|
|
|
targets, block = self._findTargetsAndBlock(targets, block)
|
|
|
return self._blockFromThread(self.smultiengine.get_properties, keys, targets=targets, block=block)
|
|
|
|
|
|
def has_properties(self, keys, targets=None, block=None):
|
|
|
warnings.warn(_prop_warn)
|
|
|
targets, block = self._findTargetsAndBlock(targets, block)
|
|
|
return self._blockFromThread(self.smultiengine.has_properties, keys, targets=targets, block=block)
|
|
|
|
|
|
def del_properties(self, keys, targets=None, block=None):
|
|
|
warnings.warn(_prop_warn)
|
|
|
targets, block = self._findTargetsAndBlock(targets, block)
|
|
|
return self._blockFromThread(self.smultiengine.del_properties, keys, targets=targets, block=block)
|
|
|
|
|
|
def clear_properties(self, targets=None, block=None):
|
|
|
warnings.warn(_prop_warn)
|
|
|
targets, block = self._findTargetsAndBlock(targets, block)
|
|
|
return self._blockFromThread(self.smultiengine.clear_properties, targets=targets, block=block)
|
|
|
|
|
|
#---------------------------------------------------------------------------
|
|
|
# IMultiEngine related methods
|
|
|
#---------------------------------------------------------------------------
|
|
|
|
|
|
def get_ids(self):
|
|
|
"""
|
|
|
Returns the ids of currently registered engines.
|
|
|
"""
|
|
|
result = blockingCallFromThread(self.smultiengine.get_ids)
|
|
|
return result
|
|
|
|
|
|
#---------------------------------------------------------------------------
|
|
|
# IMultiEngineCoordinator
|
|
|
#---------------------------------------------------------------------------
|
|
|
|
|
|
def scatter(self, key, seq, dist='b', flatten=False, targets=None, block=None):
|
|
|
"""
|
|
|
Partition a Python sequence and send the partitions to a set of engines.
|
|
|
"""
|
|
|
targets, block = self._findTargetsAndBlock(targets, block)
|
|
|
return self._blockFromThread(self.smultiengine.scatter, key, seq,
|
|
|
dist, flatten, targets=targets, block=block)
|
|
|
|
|
|
def gather(self, key, dist='b', targets=None, block=None):
|
|
|
"""
|
|
|
Gather a partitioned sequence on a set of engines as a single local seq.
|
|
|
"""
|
|
|
targets, block = self._findTargetsAndBlock(targets, block)
|
|
|
return self._blockFromThread(self.smultiengine.gather, key, dist,
|
|
|
targets=targets, block=block)
|
|
|
|
|
|
def raw_map(self, func, seq, dist='b', targets=None, block=None):
|
|
|
"""
|
|
|
A parallelized version of Python's builtin map.
|
|
|
|
|
|
This has a slightly different syntax than the builtin `map`.
|
|
|
This is needed because we need to have keyword arguments and thus
|
|
|
can't use *args to capture all the sequences. Instead, they must
|
|
|
be passed in a list or tuple.
|
|
|
|
|
|
raw_map(func, seqs) -> map(func, seqs[0], seqs[1], ...)
|
|
|
|
|
|
Most users will want to use parallel functions or the `mapper`
|
|
|
and `map` methods for an API that follows that of the builtin
|
|
|
`map`.
|
|
|
"""
|
|
|
targets, block = self._findTargetsAndBlock(targets, block)
|
|
|
return self._blockFromThread(self.smultiengine.raw_map, func, seq,
|
|
|
dist, targets=targets, block=block)
|
|
|
|
|
|
def map(self, func, *sequences):
|
|
|
"""
|
|
|
A parallel version of Python's builtin `map` function.
|
|
|
|
|
|
This method applies a function to sequences of arguments. It
|
|
|
follows the same syntax as the builtin `map`.
|
|
|
|
|
|
This method creates a mapper objects by calling `self.mapper` with
|
|
|
no arguments and then uses that mapper to do the mapping. See
|
|
|
the documentation of `mapper` for more details.
|
|
|
"""
|
|
|
return self.mapper().map(func, *sequences)
|
|
|
|
|
|
def mapper(self, dist='b', targets='all', block=None):
|
|
|
"""
|
|
|
Create a mapper object that has a `map` method.
|
|
|
|
|
|
This method returns an object that implements the `IMapper`
|
|
|
interface. This method is a factory that is used to control how
|
|
|
the map happens.
|
|
|
|
|
|
:Parameters:
|
|
|
dist : str
|
|
|
What decomposition to use, 'b' is the only one supported
|
|
|
currently
|
|
|
targets : str, int, sequence of ints
|
|
|
Which engines to use for the map
|
|
|
block : boolean
|
|
|
Should calls to `map` block or not
|
|
|
"""
|
|
|
return MultiEngineMapper(self, dist, targets, block)
|
|
|
|
|
|
def parallel(self, dist='b', targets=None, block=None):
|
|
|
"""
|
|
|
A decorator that turns a function into a parallel function.
|
|
|
|
|
|
This can be used as:
|
|
|
|
|
|
@parallel()
|
|
|
def f(x, y)
|
|
|
...
|
|
|
|
|
|
f(range(10), range(10))
|
|
|
|
|
|
This causes f(0,0), f(1,1), ... to be called in parallel.
|
|
|
|
|
|
:Parameters:
|
|
|
dist : str
|
|
|
What decomposition to use, 'b' is the only one supported
|
|
|
currently
|
|
|
targets : str, int, sequence of ints
|
|
|
Which engines to use for the map
|
|
|
block : boolean
|
|
|
Should calls to `map` block or not
|
|
|
"""
|
|
|
targets, block = self._findTargetsAndBlock(targets, block)
|
|
|
mapper = self.mapper(dist, targets, block)
|
|
|
pf = ParallelFunction(mapper)
|
|
|
return pf
|
|
|
|
|
|
#---------------------------------------------------------------------------
|
|
|
# IMultiEngineExtras
|
|
|
#---------------------------------------------------------------------------
|
|
|
|
|
|
def zip_pull(self, keys, targets=None, block=None):
|
|
|
targets, block = self._findTargetsAndBlock(targets, block)
|
|
|
return self._blockFromThread(self.smultiengine.zip_pull, keys,
|
|
|
targets=targets, block=block)
|
|
|
|
|
|
def run(self, filename, targets=None, block=None):
|
|
|
"""
|
|
|
Run a Python code in a file on the engines.
|
|
|
|
|
|
:Parameters:
|
|
|
filename : str
|
|
|
The name of the local file to run
|
|
|
targets : id or list of ids
|
|
|
The engine to use for the execution
|
|
|
block : boolean
|
|
|
If False, this method will return the actual result. If False,
|
|
|
a `PendingResult` is returned which can be used to get the result
|
|
|
at a later time.
|
|
|
"""
|
|
|
targets, block = self._findTargetsAndBlock(targets, block)
|
|
|
return self._blockFromThread(self.smultiengine.run, filename,
|
|
|
targets=targets, block=block)
|
|
|
|
|
|
def benchmark(self, push_size=10000):
|
|
|
"""
|
|
|
Run performance benchmarks for the current IPython cluster.
|
|
|
|
|
|
This method tests both the latency of sending command and data to the
|
|
|
engines as well as the throughput of sending large objects to the
|
|
|
engines using push. The latency is measured by having one or more
|
|
|
engines execute the command 'pass'. The throughput is measure by
|
|
|
sending an NumPy array of size `push_size` to one or more engines.
|
|
|
|
|
|
These benchmarks will vary widely on different hardware and networks
|
|
|
and thus can be used to get an idea of the performance characteristics
|
|
|
of a particular configuration of an IPython controller and engines.
|
|
|
|
|
|
This function is not testable within our current testing framework.
|
|
|
"""
|
|
|
import timeit, __builtin__
|
|
|
__builtin__._mec_self = self
|
|
|
benchmarks = {}
|
|
|
repeat = 3
|
|
|
count = 10
|
|
|
|
|
|
timer = timeit.Timer('_mec_self.execute("pass",0)')
|
|
|
result = 1000*min(timer.repeat(repeat,count))/count
|
|
|
benchmarks['single_engine_latency'] = (result,'msec')
|
|
|
|
|
|
timer = timeit.Timer('_mec_self.execute("pass")')
|
|
|
result = 1000*min(timer.repeat(repeat,count))/count
|
|
|
benchmarks['all_engine_latency'] = (result,'msec')
|
|
|
|
|
|
try:
|
|
|
import numpy as np
|
|
|
except:
|
|
|
pass
|
|
|
else:
|
|
|
timer = timeit.Timer(
|
|
|
"_mec_self.push(d)",
|
|
|
"import numpy as np; d = dict(a=np.zeros(%r,dtype='float64'))" % push_size
|
|
|
)
|
|
|
result = min(timer.repeat(repeat,count))/count
|
|
|
benchmarks['all_engine_push'] = (1e-6*push_size*8/result, 'MB/sec')
|
|
|
|
|
|
try:
|
|
|
import numpy as np
|
|
|
except:
|
|
|
pass
|
|
|
else:
|
|
|
timer = timeit.Timer(
|
|
|
"_mec_self.push(d,0)",
|
|
|
"import numpy as np; d = dict(a=np.zeros(%r,dtype='float64'))" % push_size
|
|
|
)
|
|
|
result = min(timer.repeat(repeat,count))/count
|
|
|
benchmarks['single_engine_push'] = (1e-6*push_size*8/result, 'MB/sec')
|
|
|
|
|
|
return benchmarks
|
|
|
|
|
|
|
|
|
components.registerAdapter(FullBlockingMultiEngineClient,
|
|
|
IFullSynchronousMultiEngine, IFullBlockingMultiEngineClient)
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
