Skip to content

HTTPS clone URL

Subversion checkout URL

You can clone with HTTPS or Subversion.

Download ZIP
tree: e68cf3acfd
Fetching contributors…

Cannot retrieve contributors at this time

1033 lines (821 sloc) 36.808 kb
"""Views of remote engines."""
#-----------------------------------------------------------------------------
# Copyright (C) 2010 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 imp
import sys
import warnings
from contextlib import contextmanager
from types import ModuleType
import zmq
from IPython.testing import decorators as testdec
from IPython.utils.traitlets import HasTraits, Any, Bool, List, Dict, Set, Int, Instance, CFloat
from IPython.external.decorator import decorator
from IPython.parallel import util
from IPython.parallel.controller.dependency import Dependency, dependent
from . import map as Map
from .asyncresult import AsyncResult, AsyncMapResult
from .remotefunction import ParallelFunction, parallel, remote
#-----------------------------------------------------------------------------
# Decorators
#-----------------------------------------------------------------------------
@decorator
def save_ids(f, self, *args, **kwargs):
"""Keep our history and outstanding attributes up to date after a method call."""
n_previous = len(self.client.history)
try:
ret = f(self, *args, **kwargs)
finally:
nmsgs = len(self.client.history) - n_previous
msg_ids = self.client.history[-nmsgs:]
self.history.extend(msg_ids)
map(self.outstanding.add, msg_ids)
return ret
@decorator
def sync_results(f, self, *args, **kwargs):
"""sync relevant results from self.client to our results attribute."""
ret = f(self, *args, **kwargs)
delta = self.outstanding.difference(self.client.outstanding)
completed = self.outstanding.intersection(delta)
self.outstanding = self.outstanding.difference(completed)
for msg_id in completed:
self.results[msg_id] = self.client.results[msg_id]
return ret
@decorator
def spin_after(f, self, *args, **kwargs):
"""call spin after the method."""
ret = f(self, *args, **kwargs)
self.spin()
return ret
#-----------------------------------------------------------------------------
# Classes
#-----------------------------------------------------------------------------
@testdec.skip_doctest
class View(HasTraits):
"""Base View class for more convenint apply(f,*args,**kwargs) syntax via attributes.
Don't use this class, use subclasses.
Methods
-------
spin
flushes incoming results and registration state changes
control methods spin, and requesting `ids` also ensures up to date
wait
wait on one or more msg_ids
execution methods
apply
legacy: execute, run
data movement
push, pull, scatter, gather
query methods
get_result, queue_status, purge_results, result_status
control methods
abort, shutdown
"""
# flags
block=Bool(False)
track=Bool(True)
targets = Any()
history=List()
outstanding = Set()
results = Dict()
client = Instance('IPython.parallel.Client')
_socket = Instance('zmq.Socket')
_flag_names = List(['targets', 'block', 'track'])
_targets = Any()
_idents = Any()
def __init__(self, client=None, socket=None, **flags):
super(View, self).__init__(client=client, _socket=socket)
self.block = client.block
self.set_flags(**flags)
assert not self.__class__ is View, "Don't use base View objects, use subclasses"
def __repr__(self):
strtargets = str(self.targets)
if len(strtargets) > 16:
strtargets = strtargets[:12]+'...]'
return "<%s %s>"%(self.__class__.__name__, strtargets)
def set_flags(self, **kwargs):
"""set my attribute flags by keyword.
Views determine behavior with a few attributes (`block`, `track`, etc.).
These attributes can be set all at once by name with this method.
Parameters
----------
block : bool
whether to wait for results
track : bool
whether to create a MessageTracker to allow the user to
safely edit after arrays and buffers during non-copying
sends.
"""
for name, value in kwargs.iteritems():
if name not in self._flag_names:
raise KeyError("Invalid name: %r"%name)
else:
setattr(self, name, value)
@contextmanager
def temp_flags(self, **kwargs):
"""temporarily set flags, for use in `with` statements.
See set_flags for permanent setting of flags
Examples
--------
>>> view.track=False
...
>>> with view.temp_flags(track=True):
... ar = view.apply(dostuff, my_big_array)
... ar.tracker.wait() # wait for send to finish
>>> view.track
False
"""
# preflight: save flags, and set temporaries
saved_flags = {}
for f in self._flag_names:
saved_flags[f] = getattr(self, f)
self.set_flags(**kwargs)
# yield to the with-statement block
try:
yield
finally:
# postflight: restore saved flags
self.set_flags(**saved_flags)
#----------------------------------------------------------------
# apply
#----------------------------------------------------------------
@sync_results
@save_ids
def _really_apply(self, f, args, kwargs, block=None, **options):
"""wrapper for client.send_apply_message"""
raise NotImplementedError("Implement in subclasses")
def apply(self, f, *args, **kwargs):
"""calls f(*args, **kwargs) on remote engines, returning the result.
This method sets all apply flags via this View's attributes.
if self.block is False:
returns AsyncResult
else:
returns actual result of f(*args, **kwargs)
"""
return self._really_apply(f, args, kwargs)
def apply_async(self, f, *args, **kwargs):
"""calls f(*args, **kwargs) on remote engines in a nonblocking manner.
returns AsyncResult
"""
return self._really_apply(f, args, kwargs, block=False)
@spin_after
def apply_sync(self, f, *args, **kwargs):
"""calls f(*args, **kwargs) on remote engines in a blocking manner,
returning the result.
returns: actual result of f(*args, **kwargs)
"""
return self._really_apply(f, args, kwargs, block=True)
#----------------------------------------------------------------
# wrappers for client and control methods
#----------------------------------------------------------------
@sync_results
def spin(self):
"""spin the client, and sync"""
self.client.spin()
@sync_results
def wait(self, jobs=None, timeout=-1):
"""waits on one or more `jobs`, for up to `timeout` seconds.
Parameters
----------
jobs : int, str, or list of ints and/or strs, or one or more AsyncResult objects
ints are indices to self.history
strs are msg_ids
default: wait on all outstanding messages
timeout : float
a time in seconds, after which to give up.
default is -1, which means no timeout
Returns
-------
True : when all msg_ids are done
False : timeout reached, some msg_ids still outstanding
"""
if jobs is None:
jobs = self.history
return self.client.wait(jobs, timeout)
def abort(self, jobs=None, targets=None, block=None):
"""Abort jobs on my engines.
Parameters
----------
jobs : None, str, list of strs, optional
if None: abort all jobs.
else: abort specific msg_id(s).
"""
block = block if block is not None else self.block
targets = targets if targets is not None else self.targets
return self.client.abort(jobs=jobs, targets=targets, block=block)
def queue_status(self, targets=None, verbose=False):
"""Fetch the Queue status of my engines"""
targets = targets if targets is not None else self.targets
return self.client.queue_status(targets=targets, verbose=verbose)
def purge_results(self, jobs=[], targets=[]):
"""Instruct the controller to forget specific results."""
if targets is None or targets == 'all':
targets = self.targets
return self.client.purge_results(jobs=jobs, targets=targets)
def shutdown(self, targets=None, restart=False, hub=False, block=None):
"""Terminates one or more engine processes, optionally including the hub.
"""
block = self.block if block is None else block
if targets is None or targets == 'all':
targets = self.targets
return self.client.shutdown(targets=targets, restart=restart, hub=hub, block=block)
@spin_after
def get_result(self, indices_or_msg_ids=None):
"""return one or more results, specified by history index or msg_id.
See client.get_result for details.
"""
if indices_or_msg_ids is None:
indices_or_msg_ids = -1
if isinstance(indices_or_msg_ids, int):
indices_or_msg_ids = self.history[indices_or_msg_ids]
elif isinstance(indices_or_msg_ids, (list,tuple,set)):
indices_or_msg_ids = list(indices_or_msg_ids)
for i,index in enumerate(indices_or_msg_ids):
if isinstance(index, int):
indices_or_msg_ids[i] = self.history[index]
return self.client.get_result(indices_or_msg_ids)
#-------------------------------------------------------------------
# Map
#-------------------------------------------------------------------
def map(self, f, *sequences, **kwargs):
"""override in subclasses"""
raise NotImplementedError
def map_async(self, f, *sequences, **kwargs):
"""Parallel version of builtin `map`, using this view's engines.
This is equivalent to map(...block=False)
See `self.map` for details.
"""
if 'block' in kwargs:
raise TypeError("map_async doesn't take a `block` keyword argument.")
kwargs['block'] = False
return self.map(f,*sequences,**kwargs)
def map_sync(self, f, *sequences, **kwargs):
"""Parallel version of builtin `map`, using this view's engines.
This is equivalent to map(...block=True)
See `self.map` for details.
"""
if 'block' in kwargs:
raise TypeError("map_sync doesn't take a `block` keyword argument.")
kwargs['block'] = True
return self.map(f,*sequences,**kwargs)
def imap(self, f, *sequences, **kwargs):
"""Parallel version of `itertools.imap`.
See `self.map` for details.
"""
return iter(self.map_async(f,*sequences, **kwargs))
#-------------------------------------------------------------------
# Decorators
#-------------------------------------------------------------------
def remote(self, block=True, **flags):
"""Decorator for making a RemoteFunction"""
block = self.block if block is None else block
return remote(self, block=block, **flags)
def parallel(self, dist='b', block=None, **flags):
"""Decorator for making a ParallelFunction"""
block = self.block if block is None else block
return parallel(self, dist=dist, block=block, **flags)
@testdec.skip_doctest
class DirectView(View):
"""Direct Multiplexer View of one or more engines.
These are created via indexed access to a client:
>>> dv_1 = client[1]
>>> dv_all = client[:]
>>> dv_even = client[::2]
>>> dv_some = client[1:3]
This object provides dictionary access to engine namespaces:
# push a=5:
>>> dv['a'] = 5
# pull 'foo':
>>> db['foo']
"""
def __init__(self, client=None, socket=None, targets=None):
super(DirectView, self).__init__(client=client, socket=socket, targets=targets)
@property
def importer(self):
"""sync_imports(local=True) as a property.
See sync_imports for details.
"""
return self.sync_imports(True)
@contextmanager
def sync_imports(self, local=True):
"""Context Manager for performing simultaneous local and remote imports.
'import x as y' will *not* work. The 'as y' part will simply be ignored.
>>> with view.sync_imports():
... from numpy import recarray
importing recarray from numpy on engine(s)
"""
import __builtin__
local_import = __builtin__.__import__
modules = set()
results = []
@util.interactive
def remote_import(name, fromlist, level):
"""the function to be passed to apply, that actually performs the import
on the engine, and loads up the user namespace.
"""
import sys
user_ns = globals()
mod = __import__(name, fromlist=fromlist, level=level)
if fromlist:
for key in fromlist:
user_ns[key] = getattr(mod, key)
else:
user_ns[name] = sys.modules[name]
def view_import(name, globals={}, locals={}, fromlist=[], level=-1):
"""the drop-in replacement for __import__, that optionally imports
locally as well.
"""
# don't override nested imports
save_import = __builtin__.__import__
__builtin__.__import__ = local_import
if imp.lock_held():
# this is a side-effect import, don't do it remotely, or even
# ignore the local effects
return local_import(name, globals, locals, fromlist, level)
imp.acquire_lock()
if local:
mod = local_import(name, globals, locals, fromlist, level)
else:
raise NotImplementedError("remote-only imports not yet implemented")
imp.release_lock()
key = name+':'+','.join(fromlist or [])
if level == -1 and key not in modules:
modules.add(key)
if fromlist:
print "importing %s from %s on engine(s)"%(','.join(fromlist), name)
else:
print "importing %s on engine(s)"%name
results.append(self.apply_async(remote_import, name, fromlist, level))
# restore override
__builtin__.__import__ = save_import
return mod
# override __import__
__builtin__.__import__ = view_import
try:
# enter the block
yield
except ImportError:
if not local:
# ignore import errors if not doing local imports
pass
finally:
# always restore __import__
__builtin__.__import__ = local_import
for r in results:
# raise possible remote ImportErrors here
r.get()
@sync_results
@save_ids
def _really_apply(self, f, args=None, kwargs=None, targets=None, block=None, track=None):
"""calls f(*args, **kwargs) on remote engines, returning the result.
This method sets all of `apply`'s flags via this View's attributes.
Parameters
----------
f : callable
args : list [default: empty]
kwargs : dict [default: empty]
targets : target list [default: self.targets]
where to run
block : bool [default: self.block]
whether to block
track : bool [default: self.track]
whether to ask zmq to track the message, for safe non-copying sends
Returns
-------
if self.block is False:
returns AsyncResult
else:
returns actual result of f(*args, **kwargs) on the engine(s)
This will be a list of self.targets is also a list (even length 1), or
the single result if self.targets is an integer engine id
"""
args = [] if args is None else args
kwargs = {} if kwargs is None else kwargs
block = self.block if block is None else block
track = self.track if track is None else track
targets = self.targets if targets is None else targets
_idents = self.client._build_targets(targets)[0]
msg_ids = []
trackers = []
for ident in _idents:
msg = self.client.send_apply_message(self._socket, f, args, kwargs, track=track,
ident=ident)
if track:
trackers.append(msg['tracker'])
msg_ids.append(msg['msg_id'])
tracker = None if track is False else zmq.MessageTracker(*trackers)
ar = AsyncResult(self.client, msg_ids, fname=f.__name__, targets=targets, tracker=tracker)
if block:
try:
return ar.get()
except KeyboardInterrupt:
pass
return ar
@spin_after
def map(self, f, *sequences, **kwargs):
"""view.map(f, *sequences, block=self.block) => list|AsyncMapResult
Parallel version of builtin `map`, using this View's `targets`.
There will be one task per target, so work will be chunked
if the sequences are longer than `targets`.
Results can be iterated as they are ready, but will become available in chunks.
Parameters
----------
f : callable
function to be mapped
*sequences: one or more sequences of matching length
the sequences to be distributed and passed to `f`
block : bool
whether to wait for the result or not [default self.block]
Returns
-------
if block=False:
AsyncMapResult
An object like AsyncResult, but which reassembles the sequence of results
into a single list. AsyncMapResults can be iterated through before all
results are complete.
else:
list
the result of map(f,*sequences)
"""
block = kwargs.pop('block', self.block)
for k in kwargs.keys():
if k not in ['block', 'track']:
raise TypeError("invalid keyword arg, %r"%k)
assert len(sequences) > 0, "must have some sequences to map onto!"
pf = ParallelFunction(self, f, block=block, **kwargs)
return pf.map(*sequences)
def execute(self, code, targets=None, block=None):
"""Executes `code` on `targets` in blocking or nonblocking manner.
``execute`` is always `bound` (affects engine namespace)
Parameters
----------
code : str
the code string to be executed
block : bool
whether or not to wait until done to return
default: self.block
"""
return self._really_apply(util._execute, args=(code,), block=block, targets=targets)
def run(self, filename, targets=None, block=None):
"""Execute contents of `filename` on my engine(s).
This simply reads the contents of the file and calls `execute`.
Parameters
----------
filename : str
The path to the file
targets : int/str/list of ints/strs
the engines on which to execute
default : all
block : bool
whether or not to wait until done
default: self.block
"""
with open(filename, 'r') as f:
# add newline in case of trailing indented whitespace
# which will cause SyntaxError
code = f.read()+'\n'
return self.execute(code, block=block, targets=targets)
def update(self, ns):
"""update remote namespace with dict `ns`
See `push` for details.
"""
return self.push(ns, block=self.block, track=self.track)
def push(self, ns, targets=None, block=None, track=None):
"""update remote namespace with dict `ns`
Parameters
----------
ns : dict
dict of keys with which to update engine namespace(s)
block : bool [default : self.block]
whether to wait to be notified of engine receipt
"""
block = block if block is not None else self.block
track = track if track is not None else self.track
targets = targets if targets is not None else self.targets
# applier = self.apply_sync if block else self.apply_async
if not isinstance(ns, dict):
raise TypeError("Must be a dict, not %s"%type(ns))
return self._really_apply(util._push, (ns,), block=block, track=track, targets=targets)
def get(self, key_s):
"""get object(s) by `key_s` from remote namespace
see `pull` for details.
"""
# block = block if block is not None else self.block
return self.pull(key_s, block=True)
def pull(self, names, targets=None, block=None):
"""get object(s) by `name` from remote namespace
will return one object if it is a key.
can also take a list of keys, in which case it will return a list of objects.
"""
block = block if block is not None else self.block
targets = targets if targets is not None else self.targets
applier = self.apply_sync if block else self.apply_async
if isinstance(names, basestring):
pass
elif isinstance(names, (list,tuple,set)):
for key in names:
if not isinstance(key, basestring):
raise TypeError("keys must be str, not type %r"%type(key))
else:
raise TypeError("names must be strs, not %r"%names)
return self._really_apply(util._pull, (names,), block=block, targets=targets)
def scatter(self, key, seq, dist='b', flatten=False, targets=None, block=None, track=None):
"""
Partition a Python sequence and send the partitions to a set of engines.
"""
block = block if block is not None else self.block
track = track if track is not None else self.track
targets = targets if targets is not None else self.targets
mapObject = Map.dists[dist]()
nparts = len(targets)
msg_ids = []
trackers = []
for index, engineid in enumerate(targets):
partition = mapObject.getPartition(seq, index, nparts)
if flatten and len(partition) == 1:
ns = {key: partition[0]}
else:
ns = {key: partition}
r = self.push(ns, block=False, track=track, targets=engineid)
msg_ids.extend(r.msg_ids)
if track:
trackers.append(r._tracker)
if track:
tracker = zmq.MessageTracker(*trackers)
else:
tracker = None
r = AsyncResult(self.client, msg_ids, fname='scatter', targets=targets, tracker=tracker)
if block:
r.wait()
else:
return r
@sync_results
@save_ids
def gather(self, key, dist='b', targets=None, block=None):
"""
Gather a partitioned sequence on a set of engines as a single local seq.
"""
block = block if block is not None else self.block
targets = targets if targets is not None else self.targets
mapObject = Map.dists[dist]()
msg_ids = []
for index, engineid in enumerate(targets):
msg_ids.extend(self.pull(key, block=False, targets=engineid).msg_ids)
r = AsyncMapResult(self.client, msg_ids, mapObject, fname='gather')
if block:
try:
return r.get()
except KeyboardInterrupt:
pass
return r
def __getitem__(self, key):
return self.get(key)
def __setitem__(self,key, value):
self.update({key:value})
def clear(self, targets=None, block=False):
"""Clear the remote namespaces on my engines."""
block = block if block is not None else self.block
targets = targets if targets is not None else self.targets
return self.client.clear(targets=targets, block=block)
def kill(self, targets=None, block=True):
"""Kill my engines."""
block = block if block is not None else self.block
targets = targets if targets is not None else self.targets
return self.client.kill(targets=targets, block=block)
#----------------------------------------
# activate for %px,%autopx magics
#----------------------------------------
def activate(self):
"""Make this `View` 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 `Views` created and used by the user,
there is only one active one. The active `View` is used whenever
the magic commands %px and %autopx are used.
The activate() method is called on a given `View` to make it
active. Once this has been done, the magic commands can be used.
"""
try:
# This is injected into __builtins__.
ip = get_ipython()
except NameError:
print "The IPython parallel magics (%result, %px, %autopx) only work within IPython."
else:
pmagic = ip.plugin_manager.get_plugin('parallelmagic')
if pmagic is None:
ip.magic_load_ext('parallelmagic')
pmagic = ip.plugin_manager.get_plugin('parallelmagic')
pmagic.active_view = self
@testdec.skip_doctest
class LoadBalancedView(View):
"""An load-balancing View that only executes via the Task scheduler.
Load-balanced views can be created with the client's `view` method:
>>> v = client.load_balanced_view()
or targets can be specified, to restrict the potential destinations:
>>> v = client.client.load_balanced_view(([1,3])
which would restrict loadbalancing to between engines 1 and 3.
"""
follow=Any()
after=Any()
timeout=CFloat()
_task_scheme = Any()
_flag_names = List(['targets', 'block', 'track', 'follow', 'after', 'timeout'])
def __init__(self, client=None, socket=None, **flags):
super(LoadBalancedView, self).__init__(client=client, socket=socket, **flags)
self._task_scheme=client._task_scheme
def _validate_dependency(self, dep):
"""validate a dependency.
For use in `set_flags`.
"""
if dep is None or isinstance(dep, (str, AsyncResult, Dependency)):
return True
elif isinstance(dep, (list,set, tuple)):
for d in dep:
if not isinstance(d, (str, AsyncResult)):
return False
elif isinstance(dep, dict):
if set(dep.keys()) != set(Dependency().as_dict().keys()):
return False
if not isinstance(dep['msg_ids'], list):
return False
for d in dep['msg_ids']:
if not isinstance(d, str):
return False
else:
return False
return True
def _render_dependency(self, dep):
"""helper for building jsonable dependencies from various input forms."""
if isinstance(dep, Dependency):
return dep.as_dict()
elif isinstance(dep, AsyncResult):
return dep.msg_ids
elif dep is None:
return []
else:
# pass to Dependency constructor
return list(Dependency(dep))
def set_flags(self, **kwargs):
"""set my attribute flags by keyword.
A View is a wrapper for the Client's apply method, but with attributes
that specify keyword arguments, those attributes can be set by keyword
argument with this method.
Parameters
----------
block : bool
whether to wait for results
track : bool
whether to create a MessageTracker to allow the user to
safely edit after arrays and buffers during non-copying
sends.
#
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.
"""
super(LoadBalancedView, self).set_flags(**kwargs)
for name in ('follow', 'after'):
if name in kwargs:
value = kwargs[name]
if self._validate_dependency(value):
setattr(self, name, value)
else:
raise ValueError("Invalid dependency: %r"%value)
if 'timeout' in kwargs:
t = kwargs['timeout']
if not isinstance(t, (int, long, float, type(None))):
raise TypeError("Invalid type for timeout: %r"%type(t))
if t is not None:
if t < 0:
raise ValueError("Invalid timeout: %s"%t)
self.timeout = t
@sync_results
@save_ids
def _really_apply(self, f, args=None, kwargs=None, block=None, track=None,
after=None, follow=None, timeout=None,
targets=None):
"""calls f(*args, **kwargs) on a remote engine, returning the result.
This method temporarily sets all of `apply`'s flags for a single call.
Parameters
----------
f : callable
args : list [default: empty]
kwargs : dict [default: empty]
block : bool [default: self.block]
whether to block
track : bool [default: self.track]
whether to ask zmq to track the message, for safe non-copying sends
!!!!!! TODO: THE REST HERE !!!!
Returns
-------
if self.block is False:
returns AsyncResult
else:
returns actual result of f(*args, **kwargs) on the engine(s)
This will be a list of self.targets is also a list (even length 1), or
the single result if self.targets is an integer engine id
"""
# validate whether we can run
if self._socket.closed:
msg = "Task farming is disabled"
if self._task_scheme == 'pure':
msg += " because the pure ZMQ scheduler cannot handle"
msg += " disappearing engines."
raise RuntimeError(msg)
if self._task_scheme == 'pure':
# pure zmq scheme doesn't support dependencies
msg = "Pure ZMQ scheduler doesn't support dependencies"
if (follow or after):
# hard fail on DAG dependencies
raise RuntimeError(msg)
if isinstance(f, dependent):
# soft warn on functional dependencies
warnings.warn(msg, RuntimeWarning)
# build args
args = [] if args is None else args
kwargs = {} if kwargs is None else kwargs
block = self.block if block is None else block
track = self.track if track is None else track
after = self.after if after is None else after
follow = self.follow if follow is None else follow
timeout = self.timeout if timeout is None else timeout
targets = self.targets if targets is None else targets
if targets is None:
idents = []
else:
idents = self.client._build_targets(targets)[0]
after = self._render_dependency(after)
follow = self._render_dependency(follow)
subheader = dict(after=after, follow=follow, timeout=timeout, targets=idents)
msg = self.client.send_apply_message(self._socket, f, args, kwargs, track=track,
subheader=subheader)
tracker = None if track is False else msg['tracker']
ar = AsyncResult(self.client, msg['msg_id'], fname=f.__name__, targets=None, tracker=tracker)
if block:
try:
return ar.get()
except KeyboardInterrupt:
pass
return ar
@spin_after
@save_ids
def map(self, f, *sequences, **kwargs):
"""view.map(f, *sequences, block=self.block, chunksize=1) => list|AsyncMapResult
Parallel version of builtin `map`, load-balanced by this View.
`block`, and `chunksize` can be specified by keyword only.
Each `chunksize` elements will be a separate task, and will be
load-balanced. This lets individual elements be available for iteration
as soon as they arrive.
Parameters
----------
f : callable
function to be mapped
*sequences: one or more sequences of matching length
the sequences to be distributed and passed to `f`
block : bool
whether to wait for the result or not [default self.block]
track : bool
whether to create a MessageTracker to allow the user to
safely edit after arrays and buffers during non-copying
sends.
chunksize : int
how many elements should be in each task [default 1]
Returns
-------
if block=False:
AsyncMapResult
An object like AsyncResult, but which reassembles the sequence of results
into a single list. AsyncMapResults can be iterated through before all
results are complete.
else:
the result of map(f,*sequences)
"""
# default
block = kwargs.get('block', self.block)
chunksize = kwargs.get('chunksize', 1)
keyset = set(kwargs.keys())
extra_keys = keyset.difference_update(set(['block', 'chunksize']))
if extra_keys:
raise TypeError("Invalid kwargs: %s"%list(extra_keys))
assert len(sequences) > 0, "must have some sequences to map onto!"
pf = ParallelFunction(self, f, block=block, chunksize=chunksize)
return pf.map(*sequences)
__all__ = ['LoadBalancedView', 'DirectView']
Jump to Line
Something went wrong with that request. Please try again.