IPython/parallel/client/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
#-----------------------------------------------------------------------------

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']