IPython/zmq/parallel/view.py

"""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
#-----------------------------------------------------------------------------

from IPython.testing import decorators as testdec
from IPython.utils.traitlets import HasTraits, Any, Bool, List, Dict, Set, Int, Instance

from IPython.external.decorator import decorator

from .asyncresult import AsyncResult
from .dependency import Dependency
from .remotefunction import ParallelFunction, parallel, remote

#-----------------------------------------------------------------------------
# Decorators
#-----------------------------------------------------------------------------

@decorator
def myblock(f, self, *args, **kwargs):
    """override client.block with self.block during a call"""
    block = self.client.block
    self.client.block = self.block
    try:
        ret = f(self, *args, **kwargs)
    finally:
        self.client.block = block
    return ret

@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)
    ret = f(self, *args, **kwargs)
    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
#-----------------------------------------------------------------------------

class View(HasTraits):
    """Base View class for more convenint apply(f,*args,**kwargs) syntax via attributes.
    
    Don't use this class, use subclasses.
    """
    block=Bool(False)
    bound=Bool(False)
    track=Bool(False)
    history=List()
    outstanding = Set()
    results = Dict()
    client = Instance('IPython.zmq.parallel.client.Client')
    
    _ntargets = Int(1)
    _balanced = Bool(False)
    _default_names = List(['block', 'bound', 'track'])
    _targets = Any()
    
    def __init__(self, client=None, targets=None):
        super(View, self).__init__(client=client)
        self._targets = targets
        self._ntargets = 1 if isinstance(targets, (int,type(None))) else len(targets)
        self.block = client.block
        
        for name in self._default_names:
            setattr(self, name, getattr(self, name, None))
        
        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)

    @property
    def targets(self):
        return self._targets

    @targets.setter
    def targets(self, value):
        raise AttributeError("Cannot set View `targets` after construction!")
    
    @property
    def balanced(self):
        return self._balanced

    @balanced.setter
    def balanced(self, value):
        raise AttributeError("Cannot set View `balanced` after construction!")
    
    def _defaults(self, *excludes):
        """return dict of our default attributes, excluding names given."""
        d = dict(balanced=self._balanced, targets=self._targets)
        for name in self._default_names:
            if name not in excludes:
                d[name] = getattr(self, name)
        return d
    
    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
        bound : bool
            whether to pass the client's Namespace as the first argument
            to functions called via `apply`.
        track : bool
            whether to create a MessageTracker to allow the user to 
            safely edit after arrays and buffers during non-copying
            sends.
        """
        for key in kwargs:
            if key not in self._default_names:
                raise KeyError("Invalid name: %r"%key)
        for name in ('block', 'bound'):
            if name in kwargs:
                setattr(self, name, kwargs[name])
    
    #----------------------------------------------------------------
    # wrappers for client methods:
    #----------------------------------------------------------------
    @sync_results
    def spin(self):
        """spin the client, and sync"""
        self.client.spin()
    
    @sync_results
    @save_ids
    def apply(self, f, *args, **kwargs):
        """calls f(*args, **kwargs) on remote engines, returning the result.
        
        This method sets all of `client.apply`'s keyword arguments via this
        View's attributes.
        
        if self.block is False:
            returns AsyncResult
        else:
            returns actual result of f(*args, **kwargs)
        """
        return self.client.apply(f, args, kwargs, **self._defaults())

    @save_ids
    def apply_async(self, f, *args, **kwargs):
        """calls f(*args, **kwargs) on remote engines in a nonblocking manner.
        
        returns AsyncResult
        """
        d = self._defaults('block', 'bound')
        return self.client.apply(f,args,kwargs, block=False, bound=False, **d)

    @spin_after
    @save_ids
    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)
        """
        d = self._defaults('block', 'bound', 'track')
        return self.client.apply(f,args,kwargs, block=True, bound=False, **d)

    # @sync_results
    # @save_ids
    # def apply_bound(self, f, *args, **kwargs):
    #     """calls f(*args, **kwargs) bound to engine namespace(s).
    #     
    #     if self.block is False:
    #         returns msg_id
    #     else:
    #         returns actual result of f(*args, **kwargs)
    #     
    #     This method has access to the targets' namespace via globals()
    #     
    #     """
    #     d = self._defaults('bound')
    #     return self.client.apply(f, args, kwargs, bound=True, **d)
    # 
    @sync_results
    @save_ids
    def apply_async_bound(self, f, *args, **kwargs):
        """calls f(*args, **kwargs) bound to engine namespace(s) 
        in a nonblocking manner.
        
        The first argument to `f` will be the Engine's Namespace
        
        returns: AsyncResult
        
        """
        d = self._defaults('block', 'bound')
        return self.client.apply(f, args, kwargs, block=False, bound=True, **d)

    @spin_after
    @save_ids
    def apply_sync_bound(self, f, *args, **kwargs):
        """calls f(*args, **kwargs) bound to engine namespace(s), waiting for the result.
        
        The first argument to `f` will be the Engine's Namespace
        
        returns: actual result of f(*args, **kwargs)
        
        """
        d = self._defaults('block', 'bound')
        return self.client.apply(f, args, kwargs, block=True, bound=True, **d)
    
    def abort(self, jobs=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
        return self.client.abort(jobs=jobs, targets=self._targets, block=block)

    def queue_status(self, verbose=False):
        """Fetch the Queue status of my engines"""
        return self.client.queue_status(targets=self._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)
    
    @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, bound=False, block=True):
        """Decorator for making a RemoteFunction"""
        return remote(self.client, bound=bound, targets=self._targets, block=block, balanced=self._balanced)
    
    def parallel(self, dist='b', bound=False, block=None):
        """Decorator for making a ParallelFunction"""
        block = self.block if block is None else block
        return parallel(self.client, bound=bound, targets=self._targets, block=block, balanced=self._balanced)

@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, targets=None):
        super(DirectView, self).__init__(client=client, targets=targets)
        self._balanced = False
    
    @spin_after
    @save_ids
    def map(self, f, *sequences, **kwargs):
        """view.map(f, *sequences, block=self.block, bound=self.bound) => 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]
        bound : bool
            whether to pass the client's Namespace as the first argument to `f`
        
        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.get('block', self.block)
        bound = kwargs.get('bound', self.bound)
        for k in kwargs.keys():
            if k not in ['block', 'bound']:
                raise TypeError("invalid keyword arg, %r"%k)
        
        assert len(sequences) > 0, "must have some sequences to map onto!"
        pf = ParallelFunction(self.client, f, block=block, bound=bound,
                        targets=self._targets, balanced=False)
        return pf.map(*sequences)
    
    @sync_results
    @save_ids
    def execute(self, code, block=None):
        """execute some code on my targets."""
        
        block = block if block is not None else self.block
        
        return self.client.execute(code, block=block, targets=self._targets)
    
    @sync_results
    @save_ids
    def run(self, fname, block=None):
        """execute the code in a file on my targets."""
        
        block = block if block is not None else self.block
        
        return self.client.run(fname, block=block, targets=self._targets)
    
    def update(self, ns):
        """update remote namespace with dict `ns`"""
        return self.client.push(ns, targets=self._targets, block=self.block)
    
    def push(self, ns, block=None):
        """update remote namespace with dict `ns`"""
        
        block = block if block is not None else self.block
        
        return self.client.push(ns, targets=self._targets, block=block)

    def get(self, key_s):
        """get object(s) by `key_s` from remote namespace
        will return one object if it is a key.
        It also takes a list of keys, and will return a list of objects."""
        # block = block if block is not None else self.block
        return self.client.pull(key_s, block=True, targets=self._targets)
    
    @sync_results
    @save_ids
    def pull(self, key_s, block=True):
        """get object(s) by `key_s` from remote namespace
        will return one object if it is a key.
        It also takes a list of keys, and will return a list of objects."""
        block = block if block is not None else self.block
        return self.client.pull(key_s, block=block, targets=self._targets)
    
    def scatter(self, key, seq, dist='b', flatten=False, block=None):
        """
        Partition a Python sequence and send the partitions to a set of engines.
        """
        block = block if block is not None else self.block
        
        return self.client.scatter(key, seq, dist=dist, flatten=flatten,
                    targets=self._targets, block=block)
    
    @sync_results
    @save_ids
    def gather(self, key, dist='b', 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
        
        return self.client.gather(key, dist=dist, targets=self._targets, block=block)
    
    def __getitem__(self, key):
        return self.get(key)
    
    def __setitem__(self,key, value):
        self.update({key:value})
    
    def clear(self, block=False):
        """Clear the remote namespaces on my engines."""
        block = block if block is not None else self.block
        return self.client.clear(targets=self._targets, block=block)
    
    def kill(self, block=True):
        """Kill my engines."""
        block = block if block is not None else self.block
        return self.client.kill(targets=self._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 not None:
                pmagic.active_multiengine_client = self
            else:
                print "You must first load the parallelmagic extension " \
                      "by doing '%load_ext parallelmagic'"


@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.view(balanced=True)
    
    or targets can be specified, to restrict the potential destinations:
    
    >>> v = client.view([1,3],balanced=True)
    
    which would restrict loadbalancing to between engines 1 and 3.
    
    """
    
    _default_names = ['block', 'bound', 'follow', 'after', 'timeout']
    
    def __init__(self, client=None, targets=None):
        super(LoadBalancedView, self).__init__(client=client, targets=targets)
        self._ntargets = 1
        self._balanced = True
    
    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
                
    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
        bound : bool
            whether to pass the client's Namespace as the first argument
            to functions called via `apply`.
        track : bool
            whether to create a MessageTracker to allow the user to 
            safely edit after arrays and buffers during non-copying
            sends.
        follow : Dependency, list, msg_id, AsyncResult
            the location dependencies of tasks
        after : Dependency, list, msg_id, AsyncResult
            the time dependencies of tasks
        timeout : int,None
            the timeout to be used for tasks
        """
        
        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, 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
                
    @spin_after
    @save_ids
    def map(self, f, *sequences, **kwargs):
        """view.map(f, *sequences, block=self.block, bound=self.bound, chunk_size=1) => list|AsyncMapResult
        
        Parallel version of builtin `map`, load-balanced by this View.
        
        `block`, `bound`, and `chunk_size` can be specified by keyword only.
        
        Each `chunk_size` 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]
        bound : bool
            whether to pass the client's Namespace as the first argument to `f`
        track : bool
            whether to create a MessageTracker to allow the user to 
            safely edit after arrays and buffers during non-copying
            sends.
        chunk_size : 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)
        bound = kwargs.get('bound', self.bound)
        chunk_size = kwargs.get('chunk_size', 1)
        
        keyset = set(kwargs.keys())
        extra_keys = keyset.difference_update(set(['block', 'bound', 'chunk_size']))
        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.client, f, block=block, bound=bound, 
                                targets=self._targets, balanced=True,
                                chunk_size=chunk_size)
        return pf.map(*sequences)
    
__all__ = ['LoadBalancedView', 'DirectView']