|
|
"""Remote Functions and decorators for the client."""
|
|
|
#-----------------------------------------------------------------------------
|
|
|
# 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 map as Map
|
|
|
from asyncresult import AsyncMapResult
|
|
|
|
|
|
#-----------------------------------------------------------------------------
|
|
|
# Decorators
|
|
|
#-----------------------------------------------------------------------------
|
|
|
|
|
|
def remote(client, bound=False, block=None, targets=None):
|
|
|
"""Turn a function into a remote function.
|
|
|
|
|
|
This method can be used for map:
|
|
|
|
|
|
>>> @remote(client,block=True)
|
|
|
def func(a)
|
|
|
"""
|
|
|
def remote_function(f):
|
|
|
return RemoteFunction(client, f, bound, block, targets)
|
|
|
return remote_function
|
|
|
|
|
|
def parallel(client, dist='b', bound=False, block=None, targets='all'):
|
|
|
"""Turn a function into a parallel remote function.
|
|
|
|
|
|
This method can be used for map:
|
|
|
|
|
|
>>> @parallel(client,block=True)
|
|
|
def func(a)
|
|
|
"""
|
|
|
def parallel_function(f):
|
|
|
return ParallelFunction(client, f, dist, bound, block, targets)
|
|
|
return parallel_function
|
|
|
|
|
|
#--------------------------------------------------------------------------
|
|
|
# Classes
|
|
|
#--------------------------------------------------------------------------
|
|
|
|
|
|
class RemoteFunction(object):
|
|
|
"""Turn an existing function into a remote function.
|
|
|
|
|
|
Parameters
|
|
|
----------
|
|
|
|
|
|
client : Client instance
|
|
|
The client to be used to connect to engines
|
|
|
f : callable
|
|
|
The function to be wrapped into a remote function
|
|
|
bound : bool [default: False]
|
|
|
Whether the affect the remote namespace when called
|
|
|
block : bool [default: None]
|
|
|
Whether to wait for results or not. The default behavior is
|
|
|
to use the current `block` attribute of `client`
|
|
|
targets : valid target list [default: all]
|
|
|
The targets on which to execute.
|
|
|
"""
|
|
|
|
|
|
client = None # the remote connection
|
|
|
func = None # the wrapped function
|
|
|
block = None # whether to block
|
|
|
bound = None # whether to affect the namespace
|
|
|
targets = None # where to execute
|
|
|
|
|
|
def __init__(self, client, f, bound=False, block=None, targets=None):
|
|
|
self.client = client
|
|
|
self.func = f
|
|
|
self.block=block
|
|
|
self.bound=bound
|
|
|
self.targets=targets
|
|
|
|
|
|
def __call__(self, *args, **kwargs):
|
|
|
return self.client.apply(self.func, args=args, kwargs=kwargs,
|
|
|
block=self.block, targets=self.targets, bound=self.bound)
|
|
|
|
|
|
|
|
|
class ParallelFunction(RemoteFunction):
|
|
|
"""Class for mapping a function to sequences."""
|
|
|
def __init__(self, client, f, dist='b', bound=False, block=None, targets='all'):
|
|
|
super(ParallelFunction, self).__init__(client,f,bound,block,targets)
|
|
|
mapClass = Map.dists[dist]
|
|
|
self.mapObject = mapClass()
|
|
|
|
|
|
def __call__(self, *sequences):
|
|
|
len_0 = len(sequences[0])
|
|
|
for s in sequences:
|
|
|
if len(s)!=len_0:
|
|
|
raise ValueError('all sequences must have equal length')
|
|
|
|
|
|
if self.targets is None:
|
|
|
# load-balanced:
|
|
|
engines = [None]*len_0
|
|
|
elif isinstance(self.targets, int):
|
|
|
engines = [None]*self.targets
|
|
|
else:
|
|
|
# multiplexed:
|
|
|
engines = self.client._build_targets(self.targets)[-1]
|
|
|
|
|
|
nparts = len(engines)
|
|
|
msg_ids = []
|
|
|
# my_f = lambda *a: map(self.func, *a)
|
|
|
for index, engineid in enumerate(engines):
|
|
|
args = []
|
|
|
for seq in sequences:
|
|
|
part = self.mapObject.getPartition(seq, index, nparts)
|
|
|
if not part:
|
|
|
continue
|
|
|
else:
|
|
|
args.append(part)
|
|
|
if not args:
|
|
|
continue
|
|
|
|
|
|
# print (args)
|
|
|
if hasattr(self, '_map'):
|
|
|
f = map
|
|
|
args = [self.func]+args
|
|
|
else:
|
|
|
f=self.func
|
|
|
mid = self.client.apply(f, args=args, block=False,
|
|
|
bound=self.bound,
|
|
|
targets=engineid).msg_ids[0]
|
|
|
msg_ids.append(mid)
|
|
|
|
|
|
r = AsyncMapResult(self.client, msg_ids, self.mapObject, fname=self.func.__name__)
|
|
|
if self.block:
|
|
|
r.wait()
|
|
|
return r.result
|
|
|
else:
|
|
|
return r
|
|
|
|
|
|
def map(self, *sequences):
|
|
|
"""call a function on each element of a sequence remotely."""
|
|
|
self._map = True
|
|
|
ret = self.__call__(*sequences)
|
|
|
del self._map
|
|
|
return ret
|
|
|
|
|
|
|
