add DirectView.importer contextmanager, demote targets to mutable flag

* @require now also takes modules, and will import
* IPython.zmq.parallel is the new entrypoint, not client

IPython/zmq/parallel/__init__.py

@@ -1,23 +1,24 b''
 """The IPython ZMQ-based parallel computing interface."""
 #-----------------------------------------------------------------------------
 #  Copyright (C) 2011 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 .asyncresult import *
-# from .client import Client
-# from .dependency import *
-# from .remotefunction import *
-# from .view import *
-
 import zmq
 
 if zmq.__version__ < '2.1.3':
     raise ImportError("IPython.zmq.parallel requires pyzmq/0MQ >= 2.1.3, you appear to have %s"%zmq.__version__)
 
+from .asyncresult import *
+from .client import Client
+from .dependency import *
+from .remotefunction import *
+from .view import *
+
+

IPython/zmq/parallel/client.py

@@ -1,1343 +1,1293 b''
 """A semi-synchronous Client for the ZMQ cluster"""
 #-----------------------------------------------------------------------------
 #  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 os
 import json
 import time
 import warnings
 from datetime import datetime
 from getpass import getpass
 from pprint import pprint
 
 pjoin = os.path.join
 
 import zmq
 # from zmq.eventloop import ioloop, zmqstream
 
 from IPython.utils.path import get_ipython_dir
 from IPython.utils.pickleutil import Reference
 from IPython.utils.traitlets import (HasTraits, Int, Instance, CUnicode, 
                                     Dict, List, Bool, Str, Set)
 from IPython.external.decorator import decorator
 from IPython.external.ssh import tunnel
 
 from . import error
 from . import util
 from . import streamsession as ss
 from .asyncresult import AsyncResult, AsyncMapResult, AsyncHubResult
 from .clusterdir import ClusterDir, ClusterDirError
 from .dependency import Dependency, depend, require, dependent
 from .remotefunction import remote, parallel, ParallelFunction, RemoteFunction
 from .view import DirectView, LoadBalancedView
 
 #--------------------------------------------------------------------------
 # Decorators for Client methods
 #--------------------------------------------------------------------------
 
 @decorator
 def spin_first(f, self, *args, **kwargs):
     """Call spin() to sync state prior to calling the method."""
     self.spin()
     return f(self, *args, **kwargs)
 
 @decorator
 def default_block(f, self, *args, **kwargs):
     """Default to self.block; preserve self.block."""
     block = kwargs.get('block',None)
     block = self.block if block is None else block
     saveblock = self.block
     self.block = block
     try:
         ret = f(self, *args, **kwargs)
     finally:
         self.block = saveblock
     return ret
 
 
 #--------------------------------------------------------------------------
 # Classes
 #--------------------------------------------------------------------------
 
 class Metadata(dict):
     """Subclass of dict for initializing metadata values.
     
     Attribute access works on keys.
     
     These objects have a strict set of keys - errors will raise if you try
     to add new keys.
     """
     def __init__(self, *args, **kwargs):
         dict.__init__(self)
         md = {'msg_id' : None,
               'submitted' : None,
               'started' : None,
               'completed' : None,
               'received' : None,
               'engine_uuid' : None,
               'engine_id' : None,
               'follow' : None,
               'after' : None,
               'status' : None,
 
               'pyin' : None,
               'pyout' : None,
               'pyerr' : None,
               'stdout' : '',
               'stderr' : '',
             }
         self.update(md)
         self.update(dict(*args, **kwargs))
     
     def __getattr__(self, key):
         """getattr aliased to getitem"""
         if key in self.iterkeys():
             return self[key]
         else:
             raise AttributeError(key)
 
     def __setattr__(self, key, value):
         """setattr aliased to setitem, with strict"""
         if key in self.iterkeys():
             self[key] = value
         else:
             raise AttributeError(key)
     
     def __setitem__(self, key, value):
         """strict static key enforcement"""
         if key in self.iterkeys():
             dict.__setitem__(self, key, value)
         else:
             raise KeyError(key)
         
 
 class Client(HasTraits):
     """A semi-synchronous client to the IPython ZMQ cluster
     
     Parameters
     ----------
     
     url_or_file : bytes; zmq url or path to ipcontroller-client.json
         Connection information for the Hub's registration.  If a json connector
         file is given, then likely no further configuration is necessary.
         [Default: use profile]
     profile : bytes
         The name of the Cluster profile to be used to find connector information.
         [Default: 'default']
     context : zmq.Context
         Pass an existing zmq.Context instance, otherwise the client will create its own.
     username : bytes
         set username to be passed to the Session object
     debug : bool
         flag for lots of message printing for debug purposes
  
     #-------------- ssh related args ----------------
     # These are args for configuring the ssh tunnel to be used
     # credentials are used to forward connections over ssh to the Controller
     # Note that the ip given in `addr` needs to be relative to sshserver
     # The most basic case is to leave addr as pointing to localhost (127.0.0.1),
     # and set sshserver as the same machine the Controller is on. However, 
     # the only requirement is that sshserver is able to see the Controller
     # (i.e. is within the same trusted network).
     
     sshserver : str
         A string of the form passed to ssh, i.e. 'server.tld' or '[e-mail]:port'
         If keyfile or password is specified, and this is not, it will default to
         the ip given in addr.
     sshkey : str; path to public ssh key file
         This specifies a key to be used in ssh login, default None.
         Regular default ssh keys will be used without specifying this argument.
     password : str 
         Your ssh password to sshserver. Note that if this is left None,
         you will be prompted for it if passwordless key based login is unavailable.
     paramiko : bool
         flag for whether to use paramiko instead of shell ssh for tunneling.
         [default: True on win32, False else]
     
     ------- exec authentication args -------
     If even localhost is untrusted, you can have some protection against
     unauthorized execution by using a key.  Messages are still sent
     as cleartext, so if someone can snoop your loopback traffic this will
     not help against malicious attacks.
     
     exec_key : str
         an authentication key or file containing a key
         default: None
     
     
     Attributes
     ----------
     
     ids : list of int engine IDs
         requesting the ids attribute always synchronizes
         the registration state. To request ids without synchronization,
         use semi-private _ids attributes.
     
     history : list of msg_ids
         a list of msg_ids, keeping track of all the execution
         messages you have submitted in order.
     
     outstanding : set of msg_ids
         a set of msg_ids that have been submitted, but whose
         results have not yet been received.
     
     results : dict
         a dict of all our results, keyed by msg_id
     
     block : bool
         determines default behavior when block not specified
         in execution methods
     
     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
         queue_status, get_result, purge, result_status
     
     control methods
         abort, shutdown
     
     """
     
     
     block = Bool(False)
     outstanding = Set()
     results = Instance('collections.defaultdict', (dict,))
     metadata = Instance('collections.defaultdict', (Metadata,))
     history = List()
     debug = Bool(False)
     profile=CUnicode('default')
     
     _outstanding_dict = Instance('collections.defaultdict', (set,))
     _ids = List()
     _connected=Bool(False)
     _ssh=Bool(False)
     _context = Instance('zmq.Context')
     _config = Dict()
     _engines=Instance(util.ReverseDict, (), {})
     # _hub_socket=Instance('zmq.Socket')
     _query_socket=Instance('zmq.Socket')
     _control_socket=Instance('zmq.Socket')
     _iopub_socket=Instance('zmq.Socket')
     _notification_socket=Instance('zmq.Socket')
     _mux_socket=Instance('zmq.Socket')
     _task_socket=Instance('zmq.Socket')
     _task_scheme=Str()
-    _balanced_views=Dict()
-    _direct_views=Dict()
     _closed = False
     _ignored_control_replies=Int(0)
     _ignored_hub_replies=Int(0)
     
     def __init__(self, url_or_file=None, profile='default', cluster_dir=None, ipython_dir=None,
             context=None, username=None, debug=False, exec_key=None,
             sshserver=None, sshkey=None, password=None, paramiko=None,
             timeout=10
             ):
         super(Client, self).__init__(debug=debug, profile=profile)
         if context is None:
             context = zmq.Context.instance()
         self._context = context
             
         
         self._setup_cluster_dir(profile, cluster_dir, ipython_dir)
         if self._cd is not None:
             if url_or_file is None:
                 url_or_file = pjoin(self._cd.security_dir, 'ipcontroller-client.json')
         assert url_or_file is not None, "I can't find enough information to connect to a hub!"\
             " Please specify at least one of url_or_file or profile."
         
         try:
             util.validate_url(url_or_file)
         except AssertionError:
             if not os.path.exists(url_or_file):
                 if self._cd:
                     url_or_file = os.path.join(self._cd.security_dir, url_or_file)
                 assert os.path.exists(url_or_file), "Not a valid connection file or url: %r"%url_or_file
             with open(url_or_file) as f:
                 cfg = json.loads(f.read())
         else:
             cfg = {'url':url_or_file}
         
         # sync defaults from args, json:
         if sshserver:
             cfg['ssh'] = sshserver
         if exec_key:
             cfg['exec_key'] = exec_key
         exec_key = cfg['exec_key']
         sshserver=cfg['ssh']
         url = cfg['url']
         location = cfg.setdefault('location', None)
         cfg['url'] = util.disambiguate_url(cfg['url'], location)
         url = cfg['url']
         
         self._config = cfg
         
         self._ssh = bool(sshserver or sshkey or password)
         if self._ssh and sshserver is None:
             # default to ssh via localhost
             sshserver = url.split('://')[1].split(':')[0]
         if self._ssh and password is None:
             if tunnel.try_passwordless_ssh(sshserver, sshkey, paramiko):
                 password=False
             else:
                 password = getpass("SSH Password for %s: "%sshserver)
         ssh_kwargs = dict(keyfile=sshkey, password=password, paramiko=paramiko)
         if exec_key is not None and os.path.isfile(exec_key):
             arg = 'keyfile'
         else:
             arg = 'key'
         key_arg = {arg:exec_key}
         if username is None:
             self.session = ss.StreamSession(**key_arg)
         else:
             self.session = ss.StreamSession(username, **key_arg)
         self._query_socket = self._context.socket(zmq.XREQ)
         self._query_socket.setsockopt(zmq.IDENTITY, self.session.session)
         if self._ssh:
             tunnel.tunnel_connection(self._query_socket, url, sshserver, **ssh_kwargs)
         else:
             self._query_socket.connect(url)
         
         self.session.debug = self.debug
         
         self._notification_handlers = {'registration_notification' : self._register_engine,
                                     'unregistration_notification' : self._unregister_engine,
                                     'shutdown_notification' : lambda msg: self.close(),
                                     }
         self._queue_handlers = {'execute_reply' : self._handle_execute_reply,
                                 'apply_reply' : self._handle_apply_reply}
         self._connect(sshserver, ssh_kwargs, timeout)
         
     def __del__(self):
         """cleanup sockets, but _not_ context."""
         self.close()
     
     def _setup_cluster_dir(self, profile, cluster_dir, ipython_dir):
         if ipython_dir is None:
             ipython_dir = get_ipython_dir()
         if cluster_dir is not None:
             try:
                 self._cd = ClusterDir.find_cluster_dir(cluster_dir)
                 return
             except ClusterDirError:
                 pass
         elif profile is not None:
             try:
                 self._cd = ClusterDir.find_cluster_dir_by_profile(
                     ipython_dir, profile)
                 return
             except ClusterDirError:
                 pass
         self._cd = None
     
     def _update_engines(self, engines):
         """Update our engines dict and _ids from a dict of the form: {id:uuid}."""
         for k,v in engines.iteritems():
             eid = int(k)
             self._engines[eid] = bytes(v) # force not unicode
             self._ids.append(eid)
         self._ids = sorted(self._ids)
         if sorted(self._engines.keys()) != range(len(self._engines)) and \
                         self._task_scheme == 'pure' and self._task_socket:
             self._stop_scheduling_tasks()
     
     def _stop_scheduling_tasks(self):
         """Stop scheduling tasks because an engine has been unregistered
         from a pure ZMQ scheduler.
         """
         self._task_socket.close()
         self._task_socket = None
         msg = "An engine has been unregistered, and we are using pure " +\
               "ZMQ task scheduling.  Task farming will be disabled."
         if self.outstanding:
             msg += " If you were running tasks when this happened, " +\
                    "some `outstanding` msg_ids may never resolve."
         warnings.warn(msg, RuntimeWarning)
     
     def _build_targets(self, targets):
         """Turn valid target IDs or 'all' into two lists:
         (int_ids, uuids).
         """
         if targets is None:
             targets = self._ids
         elif isinstance(targets, str):
             if targets.lower() == 'all':
                 targets = self._ids
             else:
                 raise TypeError("%r not valid str target, must be 'all'"%(targets))
         elif isinstance(targets, int):
+            if targets < 0:
+                targets = self.ids[targets]
+            if targets not in self.ids:
+                raise IndexError("No such engine: %i"%targets)
             targets = [targets]
+        
+        if isinstance(targets, slice):
+            indices = range(len(self._ids))[targets]
+            ids = self.ids
+            targets = [ ids[i] for i in indices ]
+        
+        if not isinstance(targets, (tuple, list, xrange)):
+            raise TypeError("targets by int/slice/collection of ints only, not %s"%(type(targets)))
+            
         return [self._engines[t] for t in targets], list(targets)
     
     def _connect(self, sshserver, ssh_kwargs, timeout):
         """setup all our socket connections to the cluster. This is called from
         __init__."""
         
         # Maybe allow reconnecting?
         if self._connected:
             return
         self._connected=True
 
         def connect_socket(s, url):
             url = util.disambiguate_url(url, self._config['location'])
             if self._ssh:
                 return tunnel.tunnel_connection(s, url, sshserver, **ssh_kwargs)
             else:
                 return s.connect(url)
             
         self.session.send(self._query_socket, 'connection_request')
         r,w,x = zmq.select([self._query_socket],[],[], timeout)
         if not r:
             raise error.TimeoutError("Hub connection request timed out")
         idents,msg = self.session.recv(self._query_socket,mode=0)
         if self.debug:
             pprint(msg)
         msg = ss.Message(msg)
         content = msg.content
         self._config['registration'] = dict(content)
         if content.status == 'ok':
             if content.mux:
                 self._mux_socket = self._context.socket(zmq.XREQ)
                 self._mux_socket.setsockopt(zmq.IDENTITY, self.session.session)
                 connect_socket(self._mux_socket, content.mux)
             if content.task:
                 self._task_scheme, task_addr = content.task
                 self._task_socket = self._context.socket(zmq.XREQ)
                 self._task_socket.setsockopt(zmq.IDENTITY, self.session.session)
                 connect_socket(self._task_socket, task_addr)
             if content.notification:
                 self._notification_socket = self._context.socket(zmq.SUB)
                 connect_socket(self._notification_socket, content.notification)
                 self._notification_socket.setsockopt(zmq.SUBSCRIBE, b'')
             # if content.query:
             #     self._query_socket = self._context.socket(zmq.XREQ)
             #     self._query_socket.setsockopt(zmq.IDENTITY, self.session.session)
             #     connect_socket(self._query_socket, content.query)
             if content.control:
                 self._control_socket = self._context.socket(zmq.XREQ)
                 self._control_socket.setsockopt(zmq.IDENTITY, self.session.session)
                 connect_socket(self._control_socket, content.control)
             if content.iopub:
                 self._iopub_socket = self._context.socket(zmq.SUB)
                 self._iopub_socket.setsockopt(zmq.SUBSCRIBE, b'')
                 self._iopub_socket.setsockopt(zmq.IDENTITY, self.session.session)
                 connect_socket(self._iopub_socket, content.iopub)
             self._update_engines(dict(content.engines))
         else:
             self._connected = False
             raise Exception("Failed to connect!")
     
     #--------------------------------------------------------------------------
     # handlers and callbacks for incoming messages
     #--------------------------------------------------------------------------
     
     def _unwrap_exception(self, content):
         """unwrap exception, and remap engine_id to int."""
         e = error.unwrap_exception(content)
         # print e.traceback
         if e.engine_info:
             e_uuid = e.engine_info['engine_uuid']
             eid = self._engines[e_uuid]
             e.engine_info['engine_id'] = eid
         return e
     
     def _extract_metadata(self, header, parent, content):
         md = {'msg_id' : parent['msg_id'],
               'received' : datetime.now(),
               'engine_uuid' : header.get('engine', None),
               'follow' : parent.get('follow', []),
               'after' : parent.get('after', []),
               'status' : content['status'],
             }
         
         if md['engine_uuid'] is not None:
             md['engine_id'] = self._engines.get(md['engine_uuid'], None)
         
         if 'date' in parent:
             md['submitted'] = datetime.strptime(parent['date'], util.ISO8601)
         if 'started' in header:
             md['started'] = datetime.strptime(header['started'], util.ISO8601)
         if 'date' in header:
             md['completed'] = datetime.strptime(header['date'], util.ISO8601)
         return md
     
     def _register_engine(self, msg):
         """Register a new engine, and update our connection info."""
         content = msg['content']
         eid = content['id']
         d = {eid : content['queue']}
         self._update_engines(d)
 
     def _unregister_engine(self, msg):
         """Unregister an engine that has died."""
         content = msg['content']
         eid = int(content['id'])
         if eid in self._ids:
             self._ids.remove(eid)
             uuid = self._engines.pop(eid)
             
             self._handle_stranded_msgs(eid, uuid)
                 
         if self._task_socket and self._task_scheme == 'pure':
             self._stop_scheduling_tasks()
     
     def _handle_stranded_msgs(self, eid, uuid):
         """Handle messages known to be on an engine when the engine unregisters.
         
         It is possible that this will fire prematurely - that is, an engine will
         go down after completing a result, and the client will be notified
         of the unregistration and later receive the successful result.
         """
         
         outstanding = self._outstanding_dict[uuid]
         
         for msg_id in list(outstanding):
             if msg_id in self.results:
                 # we already 
                 continue
             try:
                 raise error.EngineError("Engine %r died while running task %r"%(eid, msg_id))
             except:
                 content = error.wrap_exception()
             # build a fake message:
             parent = {}
             header = {}
             parent['msg_id'] = msg_id
             header['engine'] = uuid
             header['date'] = datetime.now().strftime(util.ISO8601)
             msg = dict(parent_header=parent, header=header, content=content)
             self._handle_apply_reply(msg)
     
     def _handle_execute_reply(self, msg):
         """Save the reply to an execute_request into our results.
         
         execute messages are never actually used. apply is used instead.
         """
         
         parent = msg['parent_header']
         msg_id = parent['msg_id']
         if msg_id not in self.outstanding:
             if msg_id in self.history:
                 print ("got stale result: %s"%msg_id)
             else:
                 print ("got unknown result: %s"%msg_id)
         else:
             self.outstanding.remove(msg_id)
         self.results[msg_id] = self._unwrap_exception(msg['content'])
     
     def _handle_apply_reply(self, msg):
         """Save the reply to an apply_request into our results."""
         parent = msg['parent_header']
         msg_id = parent['msg_id']
         if msg_id not in self.outstanding:
             if msg_id in self.history:
                 print ("got stale result: %s"%msg_id)
                 print self.results[msg_id]
                 print msg
             else:
                 print ("got unknown result: %s"%msg_id)
         else:
             self.outstanding.remove(msg_id)
         content = msg['content']
         header = msg['header']
         
         # construct metadata:
         md = self.metadata[msg_id]
         md.update(self._extract_metadata(header, parent, content))
         # is this redundant?
         self.metadata[msg_id] = md
         
         e_outstanding = self._outstanding_dict[md['engine_uuid']]
         if msg_id in e_outstanding:
             e_outstanding.remove(msg_id)
         
         # construct result:
         if content['status'] == 'ok':
             self.results[msg_id] = util.unserialize_object(msg['buffers'])[0]
         elif content['status'] == 'aborted':
             self.results[msg_id] = error.TaskAborted(msg_id)
         elif content['status'] == 'resubmitted':
             # TODO: handle resubmission
             pass
         else:
             self.results[msg_id] = self._unwrap_exception(content)
     
     def _flush_notifications(self):
         """Flush notifications of engine registrations waiting
         in ZMQ queue."""
         msg = self.session.recv(self._notification_socket, mode=zmq.NOBLOCK)
         while msg is not None:
             if self.debug:
                 pprint(msg)
             msg = msg[-1]
             msg_type = msg['msg_type']
             handler = self._notification_handlers.get(msg_type, None)
             if handler is None:
                 raise Exception("Unhandled message type: %s"%msg.msg_type)
             else:
                 handler(msg)
             msg = self.session.recv(self._notification_socket, mode=zmq.NOBLOCK)
     
     def _flush_results(self, sock):
         """Flush task or queue results waiting in ZMQ queue."""
         msg = self.session.recv(sock, mode=zmq.NOBLOCK)
         while msg is not None:
             if self.debug:
                 pprint(msg)
             msg = msg[-1]
             msg_type = msg['msg_type']
             handler = self._queue_handlers.get(msg_type, None)
             if handler is None:
                 raise Exception("Unhandled message type: %s"%msg.msg_type)
             else:
                 handler(msg)
             msg = self.session.recv(sock, mode=zmq.NOBLOCK)
     
     def _flush_control(self, sock):
         """Flush replies from the control channel waiting
         in the ZMQ queue.
         
         Currently: ignore them."""
         if self._ignored_control_replies <= 0:
             return
         msg = self.session.recv(sock, mode=zmq.NOBLOCK)
         while msg is not None:
             self._ignored_control_replies -= 1
             if self.debug:
                 pprint(msg)
             msg = self.session.recv(sock, mode=zmq.NOBLOCK)
     
     def _flush_ignored_control(self):
         """flush ignored control replies"""
         while self._ignored_control_replies > 0:
             self.session.recv(self._control_socket)
             self._ignored_control_replies -= 1
     
     def _flush_ignored_hub_replies(self):
         msg = self.session.recv(self._query_socket, mode=zmq.NOBLOCK)
         while msg is not None:
             msg = self.session.recv(self._query_socket, mode=zmq.NOBLOCK)
     
     def _flush_iopub(self, sock):
         """Flush replies from the iopub channel waiting
         in the ZMQ queue.
         """
         msg = self.session.recv(sock, mode=zmq.NOBLOCK)
         while msg is not None:
             if self.debug:
                 pprint(msg)
             msg = msg[-1]
             parent = msg['parent_header']
             msg_id = parent['msg_id']
             content = msg['content']
             header = msg['header']
             msg_type = msg['msg_type']
             
             # init metadata:
             md = self.metadata[msg_id]
             
             if msg_type == 'stream':
                 name = content['name']
                 s = md[name] or ''
                 md[name] = s + content['data']
             elif msg_type == 'pyerr':
                 md.update({'pyerr' : self._unwrap_exception(content)})
             else:
                 md.update({msg_type : content['data']})
             
             # reduntant?
             self.metadata[msg_id] = md
             
             msg = self.session.recv(sock, mode=zmq.NOBLOCK)
     
     #--------------------------------------------------------------------------
     # len, getitem
     #--------------------------------------------------------------------------
     
     def __len__(self):
         """len(client) returns # of engines."""
         return len(self.ids)
     
     def __getitem__(self, key):
         """index access returns DirectView multiplexer objects
         
         Must be int, slice, or list/tuple/xrange of ints"""
         if not isinstance(key, (int, slice, tuple, list, xrange)):
             raise TypeError("key by int/slice/iterable of ints only, not %s"%(type(key)))
         else:
-            return self._get_view(key, balanced=False)
+            return self.direct_view(key)
     
     #--------------------------------------------------------------------------
     # Begin public methods
     #--------------------------------------------------------------------------
     
     @property
     def ids(self):
         """Always up-to-date ids property."""
         self._flush_notifications()
         # always copy:
         return list(self._ids)
         
     def close(self):
         if self._closed:
             return
         snames = filter(lambda n: n.endswith('socket'), dir(self))
         for socket in map(lambda name: getattr(self, name), snames):
             if isinstance(socket, zmq.Socket) and not socket.closed:
                 socket.close()
         self._closed = True
     
     def spin(self):
         """Flush any registration notifications and execution results
         waiting in the ZMQ queue.
         """
         if self._notification_socket:
             self._flush_notifications()
         if self._mux_socket:
             self._flush_results(self._mux_socket)
         if self._task_socket:
             self._flush_results(self._task_socket)
         if self._control_socket:
             self._flush_control(self._control_socket)
         if self._iopub_socket:
             self._flush_iopub(self._iopub_socket)
         if self._query_socket:
             self._flush_ignored_hub_replies()
     
     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
         """
         tic = time.time()
         if jobs is None:
             theids = self.outstanding
         else:
             if isinstance(jobs, (int, str, AsyncResult)):
                 jobs = [jobs]
             theids = set()
             for job in jobs:
                 if isinstance(job, int):
                     # index access
                     job = self.history[job]
                 elif isinstance(job, AsyncResult):
                     map(theids.add, job.msg_ids)
                     continue
                 theids.add(job)
         if not theids.intersection(self.outstanding):
             return True
         self.spin()
         while theids.intersection(self.outstanding):
             if timeout >= 0 and ( time.time()-tic ) > timeout:
                 break
             time.sleep(1e-3)
             self.spin()
         return len(theids.intersection(self.outstanding)) == 0
     
     #--------------------------------------------------------------------------
     # Control methods
     #--------------------------------------------------------------------------
     
     @spin_first
     @default_block
     def clear(self, targets=None, block=None):
         """Clear the namespace in target(s)."""
         targets = self._build_targets(targets)[0]
         for t in targets:
             self.session.send(self._control_socket, 'clear_request', content={}, ident=t)
         error = False
         if self.block:
             self._flush_ignored_control()
             for i in range(len(targets)):
                 idents,msg = self.session.recv(self._control_socket,0)
                 if self.debug:
                     pprint(msg)
                 if msg['content']['status'] != 'ok':
                     error = self._unwrap_exception(msg['content'])
         else:
             self._ignored_control_replies += len(targets)
         if error:
             raise error
         
     
     @spin_first
     @default_block
     def abort(self, jobs=None, targets=None, block=None):
         """Abort specific jobs from the execution queues of target(s).
         
         This is a mechanism to prevent jobs that have already been submitted
         from executing.
         
         Parameters
         ----------
         
         jobs : msg_id, list of msg_ids, or AsyncResult
             The jobs to be aborted
         
         
         """
         targets = self._build_targets(targets)[0]
         msg_ids = []
         if isinstance(jobs, (basestring,AsyncResult)):
             jobs = [jobs]
         bad_ids = filter(lambda obj: not isinstance(obj, (basestring, AsyncResult)), jobs)
         if bad_ids:
             raise TypeError("Invalid msg_id type %r, expected str or AsyncResult"%bad_ids[0])
         for j in jobs:
             if isinstance(j, AsyncResult):
                 msg_ids.extend(j.msg_ids)
             else:
                 msg_ids.append(j)
         content = dict(msg_ids=msg_ids)
         for t in targets:
             self.session.send(self._control_socket, 'abort_request', 
                     content=content, ident=t)
         error = False
         if self.block:
             self._flush_ignored_control()
             for i in range(len(targets)):
                 idents,msg = self.session.recv(self._control_socket,0)
                 if self.debug:
                     pprint(msg)
                 if msg['content']['status'] != 'ok':
                     error = self._unwrap_exception(msg['content'])
         else:
             self._ignored_control_replies += len(targets)
         if error:
             raise error
     
     @spin_first
     @default_block
     def shutdown(self, targets=None, restart=False, hub=False, block=None):
         """Terminates one or more engine processes, optionally including the hub."""
         if hub:
             targets = 'all'
         targets = self._build_targets(targets)[0]
         for t in targets:
             self.session.send(self._control_socket, 'shutdown_request', 
                         content={'restart':restart},ident=t)
         error = False
         if block or hub:
             self._flush_ignored_control()
             for i in range(len(targets)):
                 idents,msg = self.session.recv(self._control_socket, 0)
                 if self.debug:
                     pprint(msg)
                 if msg['content']['status'] != 'ok':
                     error = self._unwrap_exception(msg['content'])
         else:
             self._ignored_control_replies += len(targets)
         
         if hub:
             time.sleep(0.25)
             self.session.send(self._query_socket, 'shutdown_request')
             idents,msg = self.session.recv(self._query_socket, 0)
             if self.debug:
                 pprint(msg)
             if msg['content']['status'] != 'ok':
                 error = self._unwrap_exception(msg['content'])
         
         if error:
             raise error
     
     #--------------------------------------------------------------------------
     # Execution methods
     #--------------------------------------------------------------------------
     
     @default_block
     def _execute(self, code, targets='all', 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
         targets : int/str/list of ints/strs
                 the engines on which to execute
                 default : all
         block : bool
                 whether or not to wait until done to return
                 default: self.block
         """
         return self[targets].execute(code, block=block)
     
     def _maybe_raise(self, result):
         """wrapper for maybe raising an exception if apply failed."""
         if isinstance(result, error.RemoteError):
             raise result
         
         return result
     
     def send_apply_message(self, socket, f, args=None, kwargs=None, subheader=None, track=False,
                             ident=None):
         """construct and send an apply message via a socket.
         
         This is the principal method with which all engine execution is performed by views.
         """
                             
         assert not self._closed, "cannot use me anymore, I'm closed!"
         # defaults:
         args = args if args is not None else []
         kwargs = kwargs if kwargs is not None else {}
         subheader = subheader if subheader is not None else {}
         
         # validate arguments
         if not callable(f):
             raise TypeError("f must be callable, not %s"%type(f))
         if not isinstance(args, (tuple, list)):
             raise TypeError("args must be tuple or list, not %s"%type(args))
         if not isinstance(kwargs, dict):
             raise TypeError("kwargs must be dict, not %s"%type(kwargs))
         if not isinstance(subheader, dict):
             raise TypeError("subheader must be dict, not %s"%type(subheader))
         
         if not self._ids:
             # flush notification socket if no engines yet
             any_ids = self.ids
             if not any_ids:
                 raise error.NoEnginesRegistered("Can't execute without any connected engines.")
                 # enforce types of f,args,kwargs
         
         bufs = util.pack_apply_message(f,args,kwargs)
         
         msg = self.session.send(socket, "apply_request", buffers=bufs, ident=ident,
                             subheader=subheader, track=track)
         
         msg_id = msg['msg_id']
         self.outstanding.add(msg_id)
         if ident:
             # possibly routed to a specific engine
             if isinstance(ident, list):
                 ident = ident[-1]
             if ident in self._engines.values():
                 # save for later, in case of engine death
                 self._outstanding_dict[ident].add(msg_id)
         self.history.append(msg_id)
         self.metadata[msg_id]['submitted'] = datetime.now()
         
         return msg
 
     #--------------------------------------------------------------------------
     # construct a View object
     #--------------------------------------------------------------------------
     
-    def _cache_view(self, targets, balanced):
-        """save views, so subsequent requests don't create new objects."""
-        if balanced:
-            # validate whether we can run
-            if not self._task_socket:
-                msg = "Task farming is disabled"
-                if self._task_scheme == 'pure':
-                    msg += " because the pure ZMQ scheduler cannot handle"
-                    msg += " disappearing engines."
-                raise RuntimeError(msg)
-            socket = self._task_socket
-            view_class = LoadBalancedView
-            view_cache = self._balanced_views
-        else:
-            socket = self._mux_socket
-            view_class = DirectView
-            view_cache = self._direct_views
-        
-        # use str, since often targets will be a list
-        key = str(targets)
-        if key not in view_cache:
-            view_cache[key] = view_class(client=self, socket=socket, targets=targets)
-        
-        return view_cache[key]
-    
     def load_balanced_view(self, targets=None):
         """construct a DirectView object.
         
         If no arguments are specified, create a LoadBalancedView
         using all engines.
         
         Parameters
         ----------
         
         targets: list,slice,int,etc. [default: use all engines]
             The subset of engines across which to load-balance
         """
-        return self._get_view(targets, balanced=True)
+        if targets is None:
+            targets = self._build_targets(targets)[1]
+        return LoadBalancedView(client=self, socket=self._task_socket, targets=targets)
     
     def direct_view(self, targets='all'):
         """construct a DirectView object.
         
         If no targets are specified, create a DirectView
         using all engines.
         
         Parameters
         ----------
         
         targets: list,slice,int,etc. [default: use all engines]
             The engines to use for the View
         """
-        return self._get_view(targets, balanced=False)
-    
-    def _get_view(self, targets, balanced):
-        """Method for constructing View objects.
-        
-        If no arguments are specified, create a LoadBalancedView
-        using all engines.  If only `targets` specified, it will
-        be a DirectView.  This method is the underlying implementation
-        of ``client.__getitem__``.
-        
-        Parameters
-        ----------
-        
-        targets: list,slice,int,etc. [default: use all engines]
-            The engines to use for the View
-        balanced : bool [default: False if targets specified, True else]
-            whether to build a LoadBalancedView or a DirectView
-        
-        """
-        
-        if targets in (None,'all'):
-            if balanced:
-                return self._cache_view(None,True)
-            else:
-                targets = slice(None)
-        
-        if isinstance(targets, int):
-            if targets < 0:
-                targets = self.ids[targets]
-            if targets not in self.ids:
-                raise IndexError("No such engine: %i"%targets)
-            return self._cache_view(targets, balanced)
-        
-        if isinstance(targets, slice):
-            indices = range(len(self.ids))[targets]
-            ids = sorted(self._ids)
-            targets = [ ids[i] for i in indices ]
-        
-        if isinstance(targets, (tuple, list, xrange)):
-            _,targets = self._build_targets(list(targets))
-            return self._cache_view(targets, balanced)
-        else:
-            raise TypeError("targets by int/slice/collection of ints only, not %s"%(type(targets)))
+        single = isinstance(targets, int)
+        targets = self._build_targets(targets)[1]
+        if single:
+            targets = targets[0]
+        return DirectView(client=self, socket=self._mux_socket, targets=targets)
     
     #--------------------------------------------------------------------------
     # Data movement (TO BE REMOVED)
     #--------------------------------------------------------------------------
     
     @default_block
     def _push(self, ns, targets='all', block=None, track=False):
         """Push the contents of `ns` into the namespace on `target`"""
         if not isinstance(ns, dict):
             raise TypeError("Must be a dict, not %s"%type(ns))
         result = self.apply(util._push, kwargs=ns, targets=targets, block=block, bound=True, balanced=False, track=track)
         if not block:
             return result
     
     @default_block
     def _pull(self, keys, targets='all', block=None):
         """Pull objects from `target`'s namespace by `keys`"""
         if isinstance(keys, basestring):
             pass
         elif isinstance(keys, (list,tuple,set)):
             for key in keys:
                 if not isinstance(key, basestring):
                     raise TypeError("keys must be str, not type %r"%type(key))
         else:
             raise TypeError("keys must be strs, not %r"%keys)
         result = self.apply(util._pull, (keys,), targets=targets, block=block, bound=True, balanced=False)
         return result
     
     #--------------------------------------------------------------------------
     # Query methods
     #--------------------------------------------------------------------------
     
     @spin_first
     @default_block
     def get_result(self, indices_or_msg_ids=None, block=None):
         """Retrieve a result by msg_id or history index, wrapped in an AsyncResult object.
         
         If the client already has the results, no request to the Hub will be made.
         
         This is a convenient way to construct AsyncResult objects, which are wrappers
         that include metadata about execution, and allow for awaiting results that
         were not submitted by this Client.
         
         It can also be a convenient way to retrieve the metadata associated with
         blocking execution, since it always retrieves
         
         Examples
         --------
         ::
         
             In [10]: r = client.apply()
         
         Parameters
         ----------
         
         indices_or_msg_ids : integer history index, str msg_id, or list of either
             The indices or msg_ids of indices to be retrieved
         
         block : bool
             Whether to wait for the result to be done
         
         Returns
         -------
         
         AsyncResult
             A single AsyncResult object will always be returned.
         
         AsyncHubResult
             A subclass of AsyncResult that retrieves results from the Hub
         
         """
         if indices_or_msg_ids is None:
             indices_or_msg_ids = -1
         
         if not isinstance(indices_or_msg_ids, (list,tuple)):
             indices_or_msg_ids = [indices_or_msg_ids]
         
         theids = []
         for id in indices_or_msg_ids:
             if isinstance(id, int):
                 id = self.history[id]
             if not isinstance(id, str):
                 raise TypeError("indices must be str or int, not %r"%id)
             theids.append(id)
         
         local_ids = filter(lambda msg_id: msg_id in self.history or msg_id in self.results, theids)
         remote_ids = filter(lambda msg_id: msg_id not in local_ids, theids)
         
         if remote_ids:
             ar = AsyncHubResult(self, msg_ids=theids)
         else:
             ar = AsyncResult(self, msg_ids=theids)
         
         if block:
             ar.wait()
         
         return ar
     
     @spin_first
     def result_status(self, msg_ids, status_only=True):
         """Check on the status of the result(s) of the apply request with `msg_ids`.
         
         If status_only is False, then the actual results will be retrieved, else
         only the status of the results will be checked.
         
         Parameters
         ----------
         
         msg_ids : list of msg_ids
             if int:
                 Passed as index to self.history for convenience.
         status_only : bool (default: True)
             if False:
                 Retrieve the actual results of completed tasks.
         
         Returns
         -------
         
         results : dict
             There will always be the keys 'pending' and 'completed', which will
             be lists of msg_ids that are incomplete or complete. If `status_only`
             is False, then completed results will be keyed by their `msg_id`.
         """
         if not isinstance(msg_ids, (list,tuple)):
             msg_ids = [msg_ids]
             
         theids = []
         for msg_id in msg_ids:
             if isinstance(msg_id, int):
                 msg_id = self.history[msg_id]
             if not isinstance(msg_id, basestring):
                 raise TypeError("msg_ids must be str, not %r"%msg_id)
             theids.append(msg_id)
         
         completed = []
         local_results = {}
         
         # comment this block out to temporarily disable local shortcut:
         for msg_id in theids:
             if msg_id in self.results:
                 completed.append(msg_id)
                 local_results[msg_id] = self.results[msg_id]
                 theids.remove(msg_id)
         
         if theids: # some not locally cached
             content = dict(msg_ids=theids, status_only=status_only)
             msg = self.session.send(self._query_socket, "result_request", content=content)
             zmq.select([self._query_socket], [], [])
             idents,msg = self.session.recv(self._query_socket, zmq.NOBLOCK)
             if self.debug:
                 pprint(msg)
             content = msg['content']
             if content['status'] != 'ok':
                 raise self._unwrap_exception(content)
             buffers = msg['buffers']
         else:
             content = dict(completed=[],pending=[])
         
         content['completed'].extend(completed)
         
         if status_only:
             return content
         
         failures = []
         # load cached results into result:
         content.update(local_results)
         # update cache with results:
         for msg_id in sorted(theids):
             if msg_id in content['completed']:
                 rec = content[msg_id]
                 parent = rec['header']
                 header = rec['result_header']
                 rcontent = rec['result_content']
                 iodict = rec['io']
                 if isinstance(rcontent, str):
                     rcontent = self.session.unpack(rcontent)
                 
                 md = self.metadata[msg_id]
                 md.update(self._extract_metadata(header, parent, rcontent))
                 md.update(iodict)
                 
                 if rcontent['status'] == 'ok':
                     res,buffers = util.unserialize_object(buffers)
                 else:
                     print rcontent
                     res = self._unwrap_exception(rcontent)
                     failures.append(res)
                 
                 self.results[msg_id] = res
                 content[msg_id] = res
         
         if len(theids) == 1 and failures:
                 raise failures[0]
         
         error.collect_exceptions(failures, "result_status")
         return content
 
     @spin_first
     def queue_status(self, targets='all', verbose=False):
         """Fetch the status of engine queues.
         
         Parameters
         ----------
         
         targets : int/str/list of ints/strs
                 the engines whose states are to be queried.
                 default : all
         verbose : bool
                 Whether to return lengths only, or lists of ids for each element
         """
         engine_ids = self._build_targets(targets)[1]
         content = dict(targets=engine_ids, verbose=verbose)
         self.session.send(self._query_socket, "queue_request", content=content)
         idents,msg = self.session.recv(self._query_socket, 0)
         if self.debug:
             pprint(msg)
         content = msg['content']
         status = content.pop('status')
         if status != 'ok':
             raise self._unwrap_exception(content)
         content = util.rekey(content)
         if isinstance(targets, int):
             return content[targets]
         else:
             return content
         
     @spin_first
     def purge_results(self, jobs=[], targets=[]):
         """Tell the Hub to forget results.
         
         Individual results can be purged by msg_id, or the entire
         history of specific targets can be purged.
         
         Parameters
         ----------
         
         jobs : str or list of str or AsyncResult objects
                 the msg_ids whose results should be forgotten.
         targets : int/str/list of ints/strs
                 The targets, by uuid or int_id, whose entire history is to be purged.
                 Use `targets='all'` to scrub everything from the Hub's memory.
                 
                 default : None
         """
         if not targets and not jobs:
             raise ValueError("Must specify at least one of `targets` and `jobs`")
         if targets:
             targets = self._build_targets(targets)[1]
         
         # construct msg_ids from jobs
         msg_ids = []
         if isinstance(jobs, (basestring,AsyncResult)):
             jobs = [jobs]
         bad_ids = filter(lambda obj: not isinstance(obj, (basestring, AsyncResult)), jobs)
         if bad_ids:
             raise TypeError("Invalid msg_id type %r, expected str or AsyncResult"%bad_ids[0])
         for j in jobs:
             if isinstance(j, AsyncResult):
                 msg_ids.extend(j.msg_ids)
             else:
                 msg_ids.append(j)
         
         content = dict(targets=targets, msg_ids=msg_ids)
         self.session.send(self._query_socket, "purge_request", content=content)
         idents, msg = self.session.recv(self._query_socket, 0)
         if self.debug:
             pprint(msg)
         content = msg['content']
         if content['status'] != 'ok':
             raise self._unwrap_exception(content)
 
 
 __all__ = [ 'Client', 
             'depend', 
             'require', 
             'remote',
             'parallel',
             'RemoteFunction',
             'ParallelFunction',
             'DirectView',
             'LoadBalancedView',
             'AsyncResult',
             'AsyncMapResult',
             'Reference'
             ]

IPython/zmq/parallel/dependency.py

@@ -1,184 +1,196 b''
 """Dependency utilities"""
 #-----------------------------------------------------------------------------
 #  Copyright (C) 2010-2011  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.
 #-----------------------------------------------------------------------------
 
+from types import ModuleType
+
 from .asyncresult import AsyncResult
 from .error import UnmetDependency
 from .util import interactive
 
 class depend(object):
     """Dependency decorator, for use with tasks.
     
     `@depend` lets you define a function for engine dependencies
     just like you use `apply` for tasks.
     
     
     Examples
     --------
     ::
     
         @depend(df, a,b, c=5)
         def f(m,n,p)
         
         view.apply(f, 1,2,3)
     
     will call df(a,b,c=5) on the engine, and if it returns False or
     raises an UnmetDependency error, then the task will not be run
     and another engine will be tried.
     """
     def __init__(self, f, *args, **kwargs):
         self.f = f
         self.args = args
         self.kwargs = kwargs
     
     def __call__(self, f):
         return dependent(f, self.f, *self.args, **self.kwargs)
 
 class dependent(object):
     """A function that depends on another function.
     This is an object to prevent the closure used
     in traditional decorators, which are not picklable.
     """
 
     def __init__(self, f, df, *dargs, **dkwargs):
         self.f = f
         self.func_name = getattr(f, '__name__', 'f')
         self.df = df
         self.dargs = dargs
         self.dkwargs = dkwargs
 
     def __call__(self, *args, **kwargs):
         # if hasattr(self.f, 'func_globals') and hasattr(self.df, 'func_globals'):
         #     self.df.func_globals = self.f.func_globals
         if self.df(*self.dargs, **self.dkwargs) is False:
             raise UnmetDependency()
         return self.f(*args, **kwargs)
     
     @property
     def __name__(self):
         return self.func_name
 
 @interactive
 def _require(*names):
     """Helper for @require decorator."""
     from IPython.zmq.parallel.error import UnmetDependency
     user_ns = globals()
     for name in names:
         if name in user_ns:
             continue
         try:
             exec 'import %s'%name in user_ns
         except ImportError:
             raise UnmetDependency(name)
     return True
 
-def require(*names):
+def require(*mods):
     """Simple decorator for requiring names to be importable.
     
     Examples
     --------
     
     In [1]: @require('numpy')
        ...: def norm(a):
        ...:     import numpy
        ...:     return numpy.linalg.norm(a,2)
     """
+    names = []
+    for mod in mods:
+        if isinstance(mod, ModuleType):
+            mod = mod.__name__
+        
+        if isinstance(mod, basestring):
+            names.append(mod)
+        else:
+            raise TypeError("names must be modules or module names, not %s"%type(mod))
+    
     return depend(_require, *names)
 
 class Dependency(set):
     """An object for representing a set of msg_id dependencies.
     
     Subclassed from set().
     
     Parameters
     ----------
     dependencies: list/set of msg_ids or AsyncResult objects or output of Dependency.as_dict()
         The msg_ids to depend on
     all : bool [default True]
         Whether the dependency should be considered met when *all* depending tasks have completed
         or only when *any* have been completed.
     success : bool [default True]
         Whether to consider successes as fulfilling dependencies.
     failure : bool [default False]
         Whether to consider failures as fulfilling dependencies.
     
     If `all=success=True` and `failure=False`, then the task will fail with an ImpossibleDependency
         as soon as the first depended-upon task fails.
     """
     
     all=True
     success=True
     failure=True
     
     def __init__(self, dependencies=[], all=True, success=True, failure=False):
         if isinstance(dependencies, dict):
             # load from dict
             all = dependencies.get('all', True)
             success = dependencies.get('success', success)
             failure = dependencies.get('failure', failure)
             dependencies = dependencies.get('dependencies', [])
         ids = []
         
         # extract ids from various sources:
         if isinstance(dependencies, (basestring, AsyncResult)):
             dependencies = [dependencies]
         for d in dependencies:
             if isinstance(d, basestring):
                 ids.append(d)
             elif isinstance(d, AsyncResult):
                 ids.extend(d.msg_ids)
             else:
                 raise TypeError("invalid dependency type: %r"%type(d))
         
         set.__init__(self, ids)
         self.all = all
         if not (success or failure):
             raise ValueError("Must depend on at least one of successes or failures!")
         self.success=success
         self.failure = failure
     
     def check(self, completed, failed=None):
         """check whether our dependencies have been met."""
         if len(self) == 0:
             return True
         against = set()
         if self.success:
             against = completed
         if failed is not None and self.failure:
             against = against.union(failed)
         if self.all:
             return self.issubset(against)
         else:
             return not self.isdisjoint(against)
     
     def unreachable(self, completed, failed=None):
         """return whether this dependency has become impossible."""
         if len(self) == 0:
             return False
         against = set()
         if not self.success:
             against = completed
         if failed is not None and not self.failure:
             against = against.union(failed)
         if self.all:
             return not self.isdisjoint(against)
         else:
             return self.issubset(against)
         
     
     def as_dict(self):
         """Represent this dependency as a dict. For json compatibility."""
         return dict(
             dependencies=list(self),
             all=self.all,
             success=self.success,
             failure=self.failure
         )
 
 
 __all__ = ['depend', 'require', 'dependent', 'Dependency']
 

IPython/zmq/parallel/launcher.py

@@ -1,971 +1,971 b''
 #!/usr/bin/env python
 # encoding: utf-8
 """
 Facilities for launching IPython processes asynchronously.
 """
 
 #-----------------------------------------------------------------------------
 #  Copyright (C) 2008-2009  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 copy
 import logging
 import os
 import re
 import stat
 
 from signal import SIGINT, SIGTERM
 try:
     from signal import SIGKILL
 except ImportError:
     SIGKILL=SIGTERM
 
 from subprocess import Popen, PIPE, STDOUT
 try:
     from subprocess import check_output
 except ImportError:
     # pre-2.7, define check_output with Popen
     def check_output(*args, **kwargs):
         kwargs.update(dict(stdout=PIPE))
         p = Popen(*args, **kwargs)
         out,err = p.communicate()
         return out
 
 from zmq.eventloop import ioloop
 
 from IPython.external import Itpl
 # from IPython.config.configurable import Configurable
 from IPython.utils.traitlets import Any, Str, Int, List, Unicode, Dict, Instance, CUnicode
 from IPython.utils.path import get_ipython_module_path
 from IPython.utils.process import find_cmd, pycmd2argv, FindCmdError
 
 from .factory import LoggingFactory
 
-# load winhpcjob from IPython.kernel
+# load winhpcjob only on Windows
 try:
-    from IPython.kernel.winhpcjob import (
+    from .winhpcjob import (
         IPControllerTask, IPEngineTask,
         IPControllerJob, IPEngineSetJob
     )
 except ImportError:
     pass
 
 
 #-----------------------------------------------------------------------------
 # Paths to the kernel apps
 #-----------------------------------------------------------------------------
 
 
 ipclusterz_cmd_argv = pycmd2argv(get_ipython_module_path(
     'IPython.zmq.parallel.ipclusterapp'
 ))
 
 ipenginez_cmd_argv = pycmd2argv(get_ipython_module_path(
     'IPython.zmq.parallel.ipengineapp'
 ))
 
 ipcontrollerz_cmd_argv = pycmd2argv(get_ipython_module_path(
     'IPython.zmq.parallel.ipcontrollerapp'
 ))
 
 #-----------------------------------------------------------------------------
 # Base launchers and errors
 #-----------------------------------------------------------------------------
 
 
 class LauncherError(Exception):
     pass
 
 
 class ProcessStateError(LauncherError):
     pass
 
 
 class UnknownStatus(LauncherError):
     pass
 
 
 class BaseLauncher(LoggingFactory):
     """An asbtraction for starting, stopping and signaling a process."""
 
     # In all of the launchers, the work_dir is where child processes will be
     # run. This will usually be the cluster_dir, but may not be. any work_dir
     # passed into the __init__ method will override the config value.
     # This should not be used to set the work_dir for the actual engine
     # and controller. Instead, use their own config files or the
     # controller_args, engine_args attributes of the launchers to add
     # the --work-dir option.
     work_dir = Unicode(u'.')
     loop = Instance('zmq.eventloop.ioloop.IOLoop')
     
     start_data = Any()
     stop_data = Any()
     
     def _loop_default(self):
         return ioloop.IOLoop.instance()
 
     def __init__(self, work_dir=u'.', config=None, **kwargs):
         super(BaseLauncher, self).__init__(work_dir=work_dir, config=config, **kwargs)
         self.state = 'before' # can be before, running, after
         self.stop_callbacks = []
         self.start_data = None
         self.stop_data = None
 
     @property
     def args(self):
         """A list of cmd and args that will be used to start the process.
 
         This is what is passed to :func:`spawnProcess` and the first element
         will be the process name.
         """
         return self.find_args()
 
     def find_args(self):
         """The ``.args`` property calls this to find the args list.
 
         Subcommand should implement this to construct the cmd and args.
         """
         raise NotImplementedError('find_args must be implemented in a subclass')
 
     @property
     def arg_str(self):
         """The string form of the program arguments."""
         return ' '.join(self.args)
 
     @property
     def running(self):
         """Am I running."""
         if self.state == 'running':
             return True
         else:
             return False
 
     def start(self):
         """Start the process.
 
         This must return a deferred that fires with information about the
         process starting (like a pid, job id, etc.).
         """
         raise NotImplementedError('start must be implemented in a subclass')
 
     def stop(self):
         """Stop the process and notify observers of stopping.
 
         This must return a deferred that fires with information about the
         processing stopping, like errors that occur while the process is
         attempting to be shut down. This deferred won't fire when the process
         actually stops. To observe the actual process stopping, see
         :func:`observe_stop`.
         """
         raise NotImplementedError('stop must be implemented in a subclass')
 
     def on_stop(self, f):
         """Get a deferred that will fire when the process stops.
 
         The deferred will fire with data that contains information about
         the exit status of the process.
         """
         if self.state=='after':
             return f(self.stop_data)
         else:
             self.stop_callbacks.append(f)
 
     def notify_start(self, data):
         """Call this to trigger startup actions.
 
         This logs the process startup and sets the state to 'running'.  It is
         a pass-through so it can be used as a callback.
         """
 
         self.log.info('Process %r started: %r' % (self.args[0], data))
         self.start_data = data
         self.state = 'running'
         return data
 
     def notify_stop(self, data):
         """Call this to trigger process stop actions.
 
         This logs the process stopping and sets the state to 'after'. Call
         this to trigger all the deferreds from :func:`observe_stop`."""
 
         self.log.info('Process %r stopped: %r' % (self.args[0], data))
         self.stop_data = data
         self.state = 'after'
         for i in range(len(self.stop_callbacks)):
             d = self.stop_callbacks.pop()
             d(data)
         return data
 
     def signal(self, sig):
         """Signal the process.
 
         Return a semi-meaningless deferred after signaling the process.
 
         Parameters
         ----------
         sig : str or int
             'KILL', 'INT', etc., or any signal number
         """
         raise NotImplementedError('signal must be implemented in a subclass')
 
 
 #-----------------------------------------------------------------------------
 # Local process launchers
 #-----------------------------------------------------------------------------
 
 
 class LocalProcessLauncher(BaseLauncher):
     """Start and stop an external process in an asynchronous manner.
 
     This will launch the external process with a working directory of
     ``self.work_dir``.
     """
 
     # This is used to to construct self.args, which is passed to 
     # spawnProcess.
     cmd_and_args = List([])
     poll_frequency = Int(100) # in ms
 
     def __init__(self, work_dir=u'.', config=None, **kwargs):
         super(LocalProcessLauncher, self).__init__(
             work_dir=work_dir, config=config, **kwargs
         )
         self.process = None
         self.start_deferred = None
         self.poller = None
 
     def find_args(self):
         return self.cmd_and_args
 
     def start(self):
         if self.state == 'before':
             self.process = Popen(self.args,
                 stdout=PIPE,stderr=PIPE,stdin=PIPE,
                 env=os.environ,
                 cwd=self.work_dir
             )
             
             self.loop.add_handler(self.process.stdout.fileno(), self.handle_stdout, self.loop.READ)
             self.loop.add_handler(self.process.stderr.fileno(), self.handle_stderr, self.loop.READ)
             self.poller = ioloop.PeriodicCallback(self.poll, self.poll_frequency, self.loop)
             self.poller.start()
             self.notify_start(self.process.pid)
         else:
             s = 'The process was already started and has state: %r' % self.state
             raise ProcessStateError(s)
 
     def stop(self):
         return self.interrupt_then_kill()
 
     def signal(self, sig):
         if self.state == 'running':
             self.process.send_signal(sig)
 
     def interrupt_then_kill(self, delay=2.0):
         """Send INT, wait a delay and then send KILL."""
         self.signal(SIGINT)
         self.killer  = ioloop.DelayedCallback(lambda : self.signal(SIGKILL), delay*1000, self.loop)
         self.killer.start()
 
     # callbacks, etc:
     
     def handle_stdout(self, fd, events):
         line = self.process.stdout.readline()
         # a stopped process will be readable but return empty strings
         if line:
             self.log.info(line[:-1])
         else:
             self.poll()
     
     def handle_stderr(self, fd, events):
         line = self.process.stderr.readline()
         # a stopped process will be readable but return empty strings
         if line:
             self.log.error(line[:-1])
         else:
             self.poll()
     
     def poll(self):
         status = self.process.poll()
         if status is not None:
             self.poller.stop()
             self.loop.remove_handler(self.process.stdout.fileno())
             self.loop.remove_handler(self.process.stderr.fileno())
             self.notify_stop(dict(exit_code=status, pid=self.process.pid))
         return status
 
 class LocalControllerLauncher(LocalProcessLauncher):
     """Launch a controller as a regular external process."""
 
     controller_cmd = List(ipcontrollerz_cmd_argv, config=True)
     # Command line arguments to ipcontroller.
     controller_args = List(['--log-to-file','--log-level', str(logging.INFO)], config=True)
 
     def find_args(self):
         return self.controller_cmd + self.controller_args
 
     def start(self, cluster_dir):
         """Start the controller by cluster_dir."""
         self.controller_args.extend(['--cluster-dir', cluster_dir])
         self.cluster_dir = unicode(cluster_dir)
         self.log.info("Starting LocalControllerLauncher: %r" % self.args)
         return super(LocalControllerLauncher, self).start()
 
 
 class LocalEngineLauncher(LocalProcessLauncher):
     """Launch a single engine as a regular externall process."""
 
     engine_cmd = List(ipenginez_cmd_argv, config=True)
     # Command line arguments for ipengine.
     engine_args = List(
         ['--log-to-file','--log-level', str(logging.INFO)], config=True
     )
 
     def find_args(self):
         return self.engine_cmd + self.engine_args
 
     def start(self, cluster_dir):
         """Start the engine by cluster_dir."""
         self.engine_args.extend(['--cluster-dir', cluster_dir])
         self.cluster_dir = unicode(cluster_dir)
         return super(LocalEngineLauncher, self).start()
 
 
 class LocalEngineSetLauncher(BaseLauncher):
     """Launch a set of engines as regular external processes."""
 
     # Command line arguments for ipengine.
     engine_args = List(
         ['--log-to-file','--log-level', str(logging.INFO)], config=True
     )
     # launcher class
     launcher_class = LocalEngineLauncher
     
     launchers = Dict()
     stop_data = Dict()
     
     def __init__(self, work_dir=u'.', config=None, **kwargs):
         super(LocalEngineSetLauncher, self).__init__(
             work_dir=work_dir, config=config, **kwargs
         )
         self.stop_data = {}
 
     def start(self, n, cluster_dir):
         """Start n engines by profile or cluster_dir."""
         self.cluster_dir = unicode(cluster_dir)
         dlist = []
         for i in range(n):
             el = self.launcher_class(work_dir=self.work_dir, config=self.config, logname=self.log.name)
             # Copy the engine args over to each engine launcher.
             el.engine_args = copy.deepcopy(self.engine_args)
             el.on_stop(self._notice_engine_stopped)
             d = el.start(cluster_dir)
             if i==0:
                 self.log.info("Starting LocalEngineSetLauncher: %r" % el.args)
             self.launchers[i] = el
             dlist.append(d)
         self.notify_start(dlist)
         # The consumeErrors here could be dangerous
         # dfinal = gatherBoth(dlist, consumeErrors=True)
         # dfinal.addCallback(self.notify_start)
         return dlist
 
     def find_args(self):
         return ['engine set']
 
     def signal(self, sig):
         dlist = []
         for el in self.launchers.itervalues():
             d = el.signal(sig)
             dlist.append(d)
         # dfinal = gatherBoth(dlist, consumeErrors=True)
         return dlist
 
     def interrupt_then_kill(self, delay=1.0):
         dlist = []
         for el in self.launchers.itervalues():
             d = el.interrupt_then_kill(delay)
             dlist.append(d)
         # dfinal = gatherBoth(dlist, consumeErrors=True)
         return dlist
 
     def stop(self):
         return self.interrupt_then_kill()
     
     def _notice_engine_stopped(self, data):
         pid = data['pid']
         for idx,el in self.launchers.iteritems():
             if el.process.pid == pid:
                 break
         self.launchers.pop(idx)
         self.stop_data[idx] = data
         if not self.launchers:
             self.notify_stop(self.stop_data)
 
 
 #-----------------------------------------------------------------------------
 # MPIExec launchers
 #-----------------------------------------------------------------------------
 
 
 class MPIExecLauncher(LocalProcessLauncher):
     """Launch an external process using mpiexec."""
 
     # The mpiexec command to use in starting the process.
     mpi_cmd = List(['mpiexec'], config=True)
     # The command line arguments to pass to mpiexec.
     mpi_args = List([], config=True)
     # The program to start using mpiexec.
     program = List(['date'], config=True)
     # The command line argument to the program.
     program_args = List([], config=True)
     # The number of instances of the program to start.
     n = Int(1, config=True)
 
     def find_args(self):
         """Build self.args using all the fields."""
         return self.mpi_cmd + ['-n', str(self.n)] + self.mpi_args + \
                self.program + self.program_args
 
     def start(self, n):
         """Start n instances of the program using mpiexec."""
         self.n = n
         return super(MPIExecLauncher, self).start()
 
 
 class MPIExecControllerLauncher(MPIExecLauncher):
     """Launch a controller using mpiexec."""
 
     controller_cmd = List(ipcontrollerz_cmd_argv, config=True)
     # Command line arguments to ipcontroller.
     controller_args = List(['--log-to-file','--log-level', str(logging.INFO)], config=True)
     n = Int(1, config=False)
 
     def start(self, cluster_dir):
         """Start the controller by cluster_dir."""
         self.controller_args.extend(['--cluster-dir', cluster_dir])
         self.cluster_dir = unicode(cluster_dir)
         self.log.info("Starting MPIExecControllerLauncher: %r" % self.args)
         return super(MPIExecControllerLauncher, self).start(1)
 
     def find_args(self):
         return self.mpi_cmd + ['-n', self.n] + self.mpi_args + \
                self.controller_cmd + self.controller_args
 
 
 class MPIExecEngineSetLauncher(MPIExecLauncher):
 
     program = List(ipenginez_cmd_argv, config=True)
     # Command line arguments for ipengine.
     program_args = List(
         ['--log-to-file','--log-level', str(logging.INFO)], config=True
     )
     n = Int(1, config=True)
 
     def start(self, n, cluster_dir):
         """Start n engines by profile or cluster_dir."""
         self.program_args.extend(['--cluster-dir', cluster_dir])
         self.cluster_dir = unicode(cluster_dir)
         self.n = n
         self.log.info('Starting MPIExecEngineSetLauncher: %r' % self.args)
         return super(MPIExecEngineSetLauncher, self).start(n)
 
 #-----------------------------------------------------------------------------
 # SSH launchers
 #-----------------------------------------------------------------------------
 
 # TODO: Get SSH Launcher working again.
 
 class SSHLauncher(LocalProcessLauncher):
     """A minimal launcher for ssh.
 
     To be useful this will probably have to be extended to use the ``sshx``
     idea for environment variables.  There could be other things this needs
     as well.
     """
 
     ssh_cmd = List(['ssh'], config=True)
     ssh_args = List(['-tt'], config=True)
     program = List(['date'], config=True)
     program_args = List([], config=True)
     hostname = CUnicode('', config=True)
     user = CUnicode('', config=True)
     location = CUnicode('')
 
     def _hostname_changed(self, name, old, new):
         if self.user:
             self.location = u'%s@%s' % (self.user, new)
         else:
             self.location = new
 
     def _user_changed(self, name, old, new):
         self.location = u'%s@%s' % (new, self.hostname)
 
     def find_args(self):
         return self.ssh_cmd + self.ssh_args + [self.location] + \
                self.program + self.program_args
 
     def start(self, cluster_dir, hostname=None, user=None):
         self.cluster_dir = unicode(cluster_dir)
         if hostname is not None:
             self.hostname = hostname
         if user is not None:
             self.user = user
         
         return super(SSHLauncher, self).start()
     
     def signal(self, sig):
         if self.state == 'running':
             # send escaped ssh connection-closer
             self.process.stdin.write('~.')
             self.process.stdin.flush()
         
 
 
 class SSHControllerLauncher(SSHLauncher):
 
     program = List(ipcontrollerz_cmd_argv, config=True)
     # Command line arguments to ipcontroller.
     program_args = List(['-r', '--log-to-file','--log-level', str(logging.INFO)], config=True)
 
 
 class SSHEngineLauncher(SSHLauncher):
     program = List(ipenginez_cmd_argv, config=True)
     # Command line arguments for ipengine.
     program_args = List(
         ['--log-to-file','--log-level', str(logging.INFO)], config=True
     )
     
 class SSHEngineSetLauncher(LocalEngineSetLauncher):
     launcher_class = SSHEngineLauncher
     engines = Dict(config=True)
     
     def start(self, n, cluster_dir):
         """Start engines by profile or cluster_dir.
         `n` is ignored, and the `engines` config property is used instead.
         """
         
         self.cluster_dir = unicode(cluster_dir)
         dlist = []
         for host, n in self.engines.iteritems():
             if isinstance(n, (tuple, list)):
                 n, args = n
             else:
                 args = copy.deepcopy(self.engine_args)
             
             if '@' in host:
                 user,host = host.split('@',1)
             else:
                 user=None
             for i in range(n):
                 el = self.launcher_class(work_dir=self.work_dir, config=self.config, logname=self.log.name)
                 
                 # Copy the engine args over to each engine launcher.
                 i
                 el.program_args = args
                 el.on_stop(self._notice_engine_stopped)
                 d = el.start(cluster_dir, user=user, hostname=host)
                 if i==0:
                     self.log.info("Starting SSHEngineSetLauncher: %r" % el.args)
                 self.launchers[host+str(i)] = el
                 dlist.append(d)
         self.notify_start(dlist)
         return dlist
     
 
 
 #-----------------------------------------------------------------------------
 # Windows HPC Server 2008 scheduler launchers
 #-----------------------------------------------------------------------------
 
 
 # This is only used on Windows.
 def find_job_cmd():
     if os.name=='nt':
         try:
             return find_cmd('job')
         except FindCmdError:
             return 'job'
     else:
         return 'job'
 
 
 class WindowsHPCLauncher(BaseLauncher):
 
     # A regular expression used to get the job id from the output of the 
     # submit_command.
     job_id_regexp = Str(r'\d+', config=True)
     # The filename of the instantiated job script.
     job_file_name = CUnicode(u'ipython_job.xml', config=True)
     # The full path to the instantiated job script. This gets made dynamically
     # by combining the work_dir with the job_file_name.
     job_file = CUnicode(u'')
     # The hostname of the scheduler to submit the job to
     scheduler = CUnicode('', config=True)
     job_cmd = CUnicode(find_job_cmd(), config=True)
 
     def __init__(self, work_dir=u'.', config=None, **kwargs):
         super(WindowsHPCLauncher, self).__init__(
             work_dir=work_dir, config=config, **kwargs
         )
 
     @property
     def job_file(self):
         return os.path.join(self.work_dir, self.job_file_name)
 
     def write_job_file(self, n):
         raise NotImplementedError("Implement write_job_file in a subclass.")
 
     def find_args(self):
         return [u'job.exe']
         
     def parse_job_id(self, output):
         """Take the output of the submit command and return the job id."""
         m = re.search(self.job_id_regexp, output)
         if m is not None:
             job_id = m.group()
         else:
             raise LauncherError("Job id couldn't be determined: %s" % output)
         self.job_id = job_id
         self.log.info('Job started with job id: %r' % job_id)
         return job_id
 
     def start(self, n):
         """Start n copies of the process using the Win HPC job scheduler."""
         self.write_job_file(n)
         args = [
             'submit',
             '/jobfile:%s' % self.job_file,
             '/scheduler:%s' % self.scheduler
         ]
         self.log.info("Starting Win HPC Job: %s" % (self.job_cmd + ' ' + ' '.join(args),))
         # Twisted will raise DeprecationWarnings if we try to pass unicode to this
         output = check_output([self.job_cmd]+args,
             env=os.environ,
             cwd=self.work_dir,
             stderr=STDOUT
         )
         job_id = self.parse_job_id(output)
         self.notify_start(job_id)
         return job_id
 
     def stop(self):
         args = [
             'cancel',
             self.job_id,
             '/scheduler:%s' % self.scheduler
         ]
         self.log.info("Stopping Win HPC Job: %s" % (self.job_cmd + ' ' + ' '.join(args),))
         try:
             output = check_output([self.job_cmd]+args,
                 env=os.environ,
                 cwd=self.work_dir,
                 stderr=STDOUT
             )
         except:
             output = 'The job already appears to be stoppped: %r' % self.job_id
         self.notify_stop(dict(job_id=self.job_id, output=output))  # Pass the output of the kill cmd
         return output
 
 
 class WindowsHPCControllerLauncher(WindowsHPCLauncher):
 
     job_file_name = CUnicode(u'ipcontroller_job.xml', config=True)
     extra_args = List([], config=False)
 
     def write_job_file(self, n):
         job = IPControllerJob(config=self.config)
 
         t = IPControllerTask(config=self.config)
         # The tasks work directory is *not* the actual work directory of 
         # the controller. It is used as the base path for the stdout/stderr
         # files that the scheduler redirects to.
         t.work_directory = self.cluster_dir
         # Add the --cluster-dir and from self.start().
         t.controller_args.extend(self.extra_args)
         job.add_task(t)
 
         self.log.info("Writing job description file: %s" % self.job_file)
         job.write(self.job_file)
 
     @property
     def job_file(self):
         return os.path.join(self.cluster_dir, self.job_file_name)
 
     def start(self, cluster_dir):
         """Start the controller by cluster_dir."""
         self.extra_args = ['--cluster-dir', cluster_dir]
         self.cluster_dir = unicode(cluster_dir)
         return super(WindowsHPCControllerLauncher, self).start(1)
 
 
 class WindowsHPCEngineSetLauncher(WindowsHPCLauncher):
 
     job_file_name = CUnicode(u'ipengineset_job.xml', config=True)
     extra_args = List([], config=False)
 
     def write_job_file(self, n):
         job = IPEngineSetJob(config=self.config)
 
         for i in range(n):
             t = IPEngineTask(config=self.config)
             # The tasks work directory is *not* the actual work directory of 
             # the engine. It is used as the base path for the stdout/stderr
             # files that the scheduler redirects to.
             t.work_directory = self.cluster_dir
             # Add the --cluster-dir and from self.start().
             t.engine_args.extend(self.extra_args)
             job.add_task(t)
 
         self.log.info("Writing job description file: %s" % self.job_file)
         job.write(self.job_file)
 
     @property
     def job_file(self):
         return os.path.join(self.cluster_dir, self.job_file_name)
 
     def start(self, n, cluster_dir):
         """Start the controller by cluster_dir."""
         self.extra_args = ['--cluster-dir', cluster_dir]
         self.cluster_dir = unicode(cluster_dir)
         return super(WindowsHPCEngineSetLauncher, self).start(n)
 
 
 #-----------------------------------------------------------------------------
 # Batch (PBS) system launchers
 #-----------------------------------------------------------------------------
 
 class BatchSystemLauncher(BaseLauncher):
     """Launch an external process using a batch system.
 
     This class is designed to work with UNIX batch systems like PBS, LSF,
     GridEngine, etc.  The overall model is that there are different commands
     like qsub, qdel, etc. that handle the starting and stopping of the process.
 
     This class also has the notion of a batch script. The ``batch_template``
     attribute can be set to a string that is a template for the batch script.
     This template is instantiated using Itpl. Thus the template can use
     ${n} fot the number of instances. Subclasses can add additional variables
     to the template dict.
     """
 
     # Subclasses must fill these in.  See PBSEngineSet
     # The name of the command line program used to submit jobs.
     submit_command = List([''], config=True)
     # The name of the command line program used to delete jobs.
     delete_command = List([''], config=True)
     # A regular expression used to get the job id from the output of the 
     # submit_command.
     job_id_regexp = CUnicode('', config=True)
     # The string that is the batch script template itself.
     batch_template = CUnicode('', config=True)
     # The file that contains the batch template
     batch_template_file = CUnicode(u'', config=True)
     # The filename of the instantiated batch script.
     batch_file_name = CUnicode(u'batch_script', config=True)
     # The PBS Queue
     queue = CUnicode(u'', config=True)
     
     # not configurable, override in subclasses
     # PBS Job Array regex
     job_array_regexp = CUnicode('')
     job_array_template = CUnicode('')
     # PBS Queue regex
     queue_regexp = CUnicode('')
     queue_template = CUnicode('')
     # The default batch template, override in subclasses
     default_template = CUnicode('')
     # The full path to the instantiated batch script.
     batch_file = CUnicode(u'')
     # the format dict used with batch_template:
     context = Dict()
 
     
     def find_args(self):
         return self.submit_command + [self.batch_file]
     
     def __init__(self, work_dir=u'.', config=None, **kwargs):
         super(BatchSystemLauncher, self).__init__(
             work_dir=work_dir, config=config, **kwargs
         )
         self.batch_file = os.path.join(self.work_dir, self.batch_file_name)
 
     def parse_job_id(self, output):
         """Take the output of the submit command and return the job id."""
         m = re.search(self.job_id_regexp, output)
         if m is not None:
             job_id = m.group()
         else:
             raise LauncherError("Job id couldn't be determined: %s" % output)
         self.job_id = job_id
         self.log.info('Job submitted with job id: %r' % job_id)
         return job_id
 
     def write_batch_script(self, n):
         """Instantiate and write the batch script to the work_dir."""
         self.context['n'] = n
         self.context['queue'] = self.queue
         print self.context
         # first priority is batch_template if set
         if self.batch_template_file and not self.batch_template:
             # second priority is batch_template_file
             with open(self.batch_template_file) as f:
                 self.batch_template = f.read()
         if not self.batch_template:
             # third (last) priority is default_template
             self.batch_template = self.default_template
         
         regex = re.compile(self.job_array_regexp)
         # print regex.search(self.batch_template)
         if not regex.search(self.batch_template):
             self.log.info("adding job array settings to batch script")
             firstline, rest = self.batch_template.split('\n',1)
             self.batch_template = u'\n'.join([firstline, self.job_array_template, rest])
         
         regex = re.compile(self.queue_regexp)
         # print regex.search(self.batch_template)
         if self.queue and not regex.search(self.batch_template):
             self.log.info("adding PBS queue settings to batch script")
             firstline, rest = self.batch_template.split('\n',1)
             self.batch_template = u'\n'.join([firstline, self.queue_template, rest])
             
         script_as_string = Itpl.itplns(self.batch_template, self.context)
         self.log.info('Writing instantiated batch script: %s' % self.batch_file)
     
         with open(self.batch_file, 'w') as f:
             f.write(script_as_string)
         os.chmod(self.batch_file, stat.S_IRUSR | stat.S_IWUSR | stat.S_IXUSR)
 
     def start(self, n, cluster_dir):
         """Start n copies of the process using a batch system."""
         # Here we save profile and cluster_dir in the context so they
         # can be used in the batch script template as ${profile} and
         # ${cluster_dir}
         self.context['cluster_dir'] = cluster_dir
         self.cluster_dir = unicode(cluster_dir)
         self.write_batch_script(n)
         output = check_output(self.args, env=os.environ)
         
         job_id = self.parse_job_id(output)
         self.notify_start(job_id)
         return job_id
 
     def stop(self):
         output = check_output(self.delete_command+[self.job_id], env=os.environ)
         self.notify_stop(dict(job_id=self.job_id, output=output)) # Pass the output of the kill cmd
         return output
 
 
 class PBSLauncher(BatchSystemLauncher):
     """A BatchSystemLauncher subclass for PBS."""
 
     submit_command = List(['qsub'], config=True)
     delete_command = List(['qdel'], config=True)
     job_id_regexp = CUnicode(r'\d+', config=True)
     
     batch_file = CUnicode(u'')
     job_array_regexp = CUnicode('#PBS\W+-t\W+[\w\d\-\$]+')
     job_array_template = CUnicode('#PBS -t 1-$n')
     queue_regexp = CUnicode('#PBS\W+-q\W+\$?\w+')
     queue_template = CUnicode('#PBS -q $queue')
 
 
 class PBSControllerLauncher(PBSLauncher):
     """Launch a controller using PBS."""
 
     batch_file_name = CUnicode(u'pbs_controller', config=True)
     default_template= CUnicode("""#!/bin/sh
 #PBS -V
 #PBS -N ipcontrollerz
 %s --log-to-file --cluster-dir $cluster_dir
 """%(' '.join(ipcontrollerz_cmd_argv)))
 
     def start(self, cluster_dir):
         """Start the controller by profile or cluster_dir."""
         self.log.info("Starting PBSControllerLauncher: %r" % self.args)
         return super(PBSControllerLauncher, self).start(1, cluster_dir)
 
 
 class PBSEngineSetLauncher(PBSLauncher):
     """Launch Engines using PBS"""
     batch_file_name = CUnicode(u'pbs_engines', config=True)
     default_template= CUnicode(u"""#!/bin/sh
 #PBS -V
 #PBS -N ipenginez
 %s --cluster-dir $cluster_dir
 """%(' '.join(ipenginez_cmd_argv)))
 
     def start(self, n, cluster_dir):
         """Start n engines by profile or cluster_dir."""
         self.log.info('Starting %n engines with PBSEngineSetLauncher: %r' % (n, self.args))
         return super(PBSEngineSetLauncher, self).start(n, cluster_dir)
 
 #SGE is very similar to PBS
 
 class SGELauncher(PBSLauncher):
     """Sun GridEngine is a PBS clone with slightly different syntax"""
     job_array_regexp = CUnicode('#$$\W+-t\W+[\w\d\-\$]+')
     job_array_template = CUnicode('#$$ -t 1-$n')
     queue_regexp = CUnicode('#$$\W+-q\W+\$?\w+')
     queue_template = CUnicode('#$$ -q $queue')
 
 class SGEControllerLauncher(SGELauncher):
     """Launch a controller using SGE."""
 
     batch_file_name = CUnicode(u'sge_controller', config=True)
     default_template= CUnicode(u"""#$$ -V
 #$$ -S /bin/sh
 #$$ -N ipcontrollerz
 %s --log-to-file --cluster-dir $cluster_dir
 """%(' '.join(ipcontrollerz_cmd_argv)))
 
     def start(self, cluster_dir):
         """Start the controller by profile or cluster_dir."""
         self.log.info("Starting PBSControllerLauncher: %r" % self.args)
         return super(PBSControllerLauncher, self).start(1, cluster_dir)
 
 class SGEEngineSetLauncher(SGELauncher):
     """Launch Engines with SGE"""
     batch_file_name = CUnicode(u'sge_engines', config=True)
     default_template = CUnicode("""#$$ -V
 #$$ -S /bin/sh
 #$$ -N ipenginez
 %s --cluster-dir $cluster_dir
 """%(' '.join(ipenginez_cmd_argv)))
 
     def start(self, n, cluster_dir):
         """Start n engines by profile or cluster_dir."""
         self.log.info('Starting %n engines with SGEEngineSetLauncher: %r' % (n, self.args))
         return super(SGEEngineSetLauncher, self).start(n, cluster_dir)
 
 
 #-----------------------------------------------------------------------------
 # A launcher for ipcluster itself!
 #-----------------------------------------------------------------------------
 
 
 class IPClusterLauncher(LocalProcessLauncher):
     """Launch the ipcluster program in an external process."""
 
     ipcluster_cmd = List(ipclusterz_cmd_argv, config=True)
     # Command line arguments to pass to ipcluster.
     ipcluster_args = List(
         ['--clean-logs', '--log-to-file', '--log-level', str(logging.INFO)], config=True)
     ipcluster_subcommand = Str('start')
     ipcluster_n = Int(2)
 
     def find_args(self):
         return self.ipcluster_cmd + [self.ipcluster_subcommand] + \
             ['-n', repr(self.ipcluster_n)] + self.ipcluster_args
 
     def start(self):
         self.log.info("Starting ipcluster: %r" % self.args)
         return super(IPClusterLauncher, self).start()
 

IPython/zmq/parallel/tests/test_client.py

@@ -1,147 +1,138 b''
 """Tests for parallel client.py"""
 
 #-------------------------------------------------------------------------------
 #  Copyright (C) 2011  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 time
 from tempfile import mktemp
 
 import zmq
 
 from IPython.zmq.parallel import client as clientmod
 from IPython.zmq.parallel import error
 from IPython.zmq.parallel.asyncresult import AsyncResult, AsyncHubResult
 from IPython.zmq.parallel.view import LoadBalancedView, DirectView
 
 from clienttest import ClusterTestCase, segfault, wait, add_engines
 
 def setup():
     add_engines(4)
 
 class TestClient(ClusterTestCase):
     
     def test_ids(self):
         n = len(self.client.ids)
         self.add_engines(3)
         self.assertEquals(len(self.client.ids), n+3)
     
     def test_view_indexing(self):
         """test index access for views"""
         self.add_engines(2)
         targets = self.client._build_targets('all')[-1]
         v = self.client[:]
         self.assertEquals(v.targets, targets)
         t = self.client.ids[2]
         v = self.client[t]
         self.assert_(isinstance(v, DirectView))
         self.assertEquals(v.targets, t)
         t = self.client.ids[2:4]
         v = self.client[t]
         self.assert_(isinstance(v, DirectView))
         self.assertEquals(v.targets, t)
         v = self.client[::2]
         self.assert_(isinstance(v, DirectView))
         self.assertEquals(v.targets, targets[::2])
         v = self.client[1::3]
         self.assert_(isinstance(v, DirectView))
         self.assertEquals(v.targets, targets[1::3])
         v = self.client[:-3]
         self.assert_(isinstance(v, DirectView))
         self.assertEquals(v.targets, targets[:-3])
         v = self.client[-1]
         self.assert_(isinstance(v, DirectView))
         self.assertEquals(v.targets, targets[-1])
         self.assertRaises(TypeError, lambda : self.client[None])
     
-    def test_view_cache(self):
-        """test that multiple view requests return the same object"""
-        v = self.client[:2]
-        v2 =self.client[:2]
-        self.assertTrue(v is v2)
-        v = self.client.load_balanced_view()
-        v2 = self.client.load_balanced_view(targets=None)
-        self.assertTrue(v is v2)
-    
     def test_targets(self):
         """test various valid targets arguments"""
         build = self.client._build_targets
         ids = self.client.ids
         idents,targets = build(None)
         self.assertEquals(ids, targets)
     
     def test_clear(self):
         """test clear behavior"""
         # self.add_engines(2)
         v = self.client[:]
         v.block=True
         v.push(dict(a=5))
         v.pull('a')
         id0 = self.client.ids[-1]
         self.client.clear(targets=id0)
         self.client[:-1].pull('a')
         self.assertRaisesRemote(NameError, self.client[id0].get, 'a')
         self.client.clear(block=True)
         for i in self.client.ids:
             # print i
             self.assertRaisesRemote(NameError, self.client[i].get, 'a')
     
     def test_get_result(self):
         """test getting results from the Hub."""
         c = clientmod.Client(profile='iptest')
         # self.add_engines(1)
         t = c.ids[-1]
         ar = c[t].apply_async(wait, 1)
         # give the monitor time to notice the message
         time.sleep(.25)
         ahr = self.client.get_result(ar.msg_ids)
         self.assertTrue(isinstance(ahr, AsyncHubResult))
         self.assertEquals(ahr.get(), ar.get())
         ar2 = self.client.get_result(ar.msg_ids)
         self.assertFalse(isinstance(ar2, AsyncHubResult))
         c.close()
     
     def test_ids_list(self):
         """test client.ids"""
         # self.add_engines(2)
         ids = self.client.ids
         self.assertEquals(ids, self.client._ids)
         self.assertFalse(ids is self.client._ids)
         ids.remove(ids[-1])
         self.assertNotEquals(ids, self.client._ids)
     
     def test_queue_status(self):
         # self.addEngine(4)
         ids = self.client.ids
         id0 = ids[0]
         qs = self.client.queue_status(targets=id0)
         self.assertTrue(isinstance(qs, dict))
         self.assertEquals(sorted(qs.keys()), ['completed', 'queue', 'tasks'])
         allqs = self.client.queue_status()
         self.assertTrue(isinstance(allqs, dict))
         self.assertEquals(sorted(allqs.keys()), self.client.ids)
         for eid,qs in allqs.items():
             self.assertTrue(isinstance(qs, dict))
             self.assertEquals(sorted(qs.keys()), ['completed', 'queue', 'tasks'])
 
     def test_shutdown(self):
         # self.addEngine(4)
         ids = self.client.ids
         id0 = ids[0]
         self.client.shutdown(id0, block=True)
         while id0 in self.client.ids:
             time.sleep(0.1)
             self.client.spin()
         
         self.assertRaises(IndexError, lambda : self.client[id0])
         
     def test_result_status(self):
         pass
         # to be written

IPython/zmq/parallel/tests/test_view.py

@@ -1,287 +1,301 b''
 """test View objects"""
 #-------------------------------------------------------------------------------
 #  Copyright (C) 2011  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 time
 from tempfile import mktemp
 
 import zmq
 
 from IPython.zmq.parallel import client as clientmod
 from IPython.zmq.parallel import error
 from IPython.zmq.parallel.asyncresult import AsyncResult, AsyncHubResult, AsyncMapResult
 from IPython.zmq.parallel.view import LoadBalancedView, DirectView
 from IPython.zmq.parallel.util import interactive
 
 from IPython.zmq.parallel.tests import add_engines
 
 from .clienttest import ClusterTestCase, segfault, wait, skip_without
 
 def setup():
     add_engines(3)
 
 class TestView(ClusterTestCase):
     
     def test_segfault_task(self):
         """test graceful handling of engine death (balanced)"""
         # self.add_engines(1)
         ar = self.client[-1].apply_async(segfault)
         self.assertRaisesRemote(error.EngineError, ar.get)
         eid = ar.engine_id
         while eid in self.client.ids:
             time.sleep(.01)
             self.client.spin()
     
     def test_segfault_mux(self):
         """test graceful handling of engine death (direct)"""
         # self.add_engines(1)
         eid = self.client.ids[-1]
         ar = self.client[eid].apply_async(segfault)
         self.assertRaisesRemote(error.EngineError, ar.get)
         eid = ar.engine_id
         while eid in self.client.ids:
             time.sleep(.01)
             self.client.spin()
     
     def test_push_pull(self):
         """test pushing and pulling"""
         data = dict(a=10, b=1.05, c=range(10), d={'e':(1,2),'f':'hi'})
         t = self.client.ids[-1]
         v = self.client[t]
         push = v.push
         pull = v.pull
         v.block=True
         nengines = len(self.client)
         push({'data':data})
         d = pull('data')
         self.assertEquals(d, data)
         self.client[:].push({'data':data})
         d = self.client[:].pull('data', block=True)
         self.assertEquals(d, nengines*[data])
         ar = push({'data':data}, block=False)
         self.assertTrue(isinstance(ar, AsyncResult))
         r = ar.get()
         ar = self.client[:].pull('data', block=False)
         self.assertTrue(isinstance(ar, AsyncResult))
         r = ar.get()
         self.assertEquals(r, nengines*[data])
         self.client[:].push(dict(a=10,b=20))
         r = self.client[:].pull(('a','b'))
         self.assertEquals(r, nengines*[[10,20]])
     
     def test_push_pull_function(self):
         "test pushing and pulling functions"
         def testf(x):
             return 2.0*x
         
         t = self.client.ids[-1]
         self.client[t].block=True
         push = self.client[t].push
         pull = self.client[t].pull
         execute = self.client[t].execute
         push({'testf':testf})
         r = pull('testf')
         self.assertEqual(r(1.0), testf(1.0))
         execute('r = testf(10)')
         r = pull('r')
         self.assertEquals(r, testf(10))
         ar = self.client[:].push({'testf':testf}, block=False)
         ar.get()
         ar = self.client[:].pull('testf', block=False)
         rlist = ar.get()
         for r in rlist:
             self.assertEqual(r(1.0), testf(1.0))
         execute("def g(x): return x*x")
         r = pull(('testf','g'))
         self.assertEquals((r[0](10),r[1](10)), (testf(10), 100))
     
     def test_push_function_globals(self):
         """test that pushed functions have access to globals"""
         @interactive
         def geta():
             return a
         # self.add_engines(1)
         v = self.client[-1]
         v.block=True
         v['f'] = geta
         self.assertRaisesRemote(NameError, v.execute, 'b=f()')
         v.execute('a=5')
         v.execute('b=f()')
         self.assertEquals(v['b'], 5)
     
     def test_push_function_defaults(self):
         """test that pushed functions preserve default args"""
         def echo(a=10):
             return a
         v = self.client[-1]
         v.block=True
         v['f'] = echo
         v.execute('b=f()')
         self.assertEquals(v['b'], 10)
 
     def test_get_result(self):
         """test getting results from the Hub."""
         c = clientmod.Client(profile='iptest')
         # self.add_engines(1)
         t = c.ids[-1]
         v = c[t]
         v2 = self.client[t]
         ar = v.apply_async(wait, 1)
         # give the monitor time to notice the message
         time.sleep(.25)
         ahr = v2.get_result(ar.msg_ids)
         self.assertTrue(isinstance(ahr, AsyncHubResult))
         self.assertEquals(ahr.get(), ar.get())
         ar2 = v2.get_result(ar.msg_ids)
         self.assertFalse(isinstance(ar2, AsyncHubResult))
         c.spin()
         c.close()
     
     def test_run_newline(self):
         """test that run appends newline to files"""
         tmpfile = mktemp()
         with open(tmpfile, 'w') as f:
             f.write("""def g():
                 return 5
                 """)
         v = self.client[-1]
         v.run(tmpfile, block=True)
         self.assertEquals(v.apply_sync(lambda f: f(), clientmod.Reference('g')), 5)
 
     def test_apply_tracked(self):
         """test tracking for apply"""
         # self.add_engines(1)
         t = self.client.ids[-1]
         v = self.client[t]
         v.block=False
         def echo(n=1024*1024, **kwargs):
             with v.temp_flags(**kwargs):
                 return v.apply(lambda x: x, 'x'*n)
         ar = echo(1, track=False)
         self.assertTrue(isinstance(ar._tracker, zmq.MessageTracker))
         self.assertTrue(ar.sent)
         ar = echo(track=True)
         self.assertTrue(isinstance(ar._tracker, zmq.MessageTracker))
         self.assertEquals(ar.sent, ar._tracker.done)
         ar._tracker.wait()
         self.assertTrue(ar.sent)
         
     def test_push_tracked(self):
         t = self.client.ids[-1]
         ns = dict(x='x'*1024*1024)
         v = self.client[t]
         ar = v.push(ns, block=False, track=False)
         self.assertTrue(isinstance(ar._tracker, zmq.MessageTracker))
         self.assertTrue(ar.sent)
         
         ar = v.push(ns, block=False, track=True)
         self.assertTrue(isinstance(ar._tracker, zmq.MessageTracker))
         self.assertEquals(ar.sent, ar._tracker.done)
         ar._tracker.wait()
         self.assertTrue(ar.sent)
         ar.get()
         
     def test_scatter_tracked(self):
         t = self.client.ids
         x='x'*1024*1024
         ar = self.client[t].scatter('x', x, block=False, track=False)
         self.assertTrue(isinstance(ar._tracker, zmq.MessageTracker))
         self.assertTrue(ar.sent)
         
         ar = self.client[t].scatter('x', x, block=False, track=True)
         self.assertTrue(isinstance(ar._tracker, zmq.MessageTracker))
         self.assertEquals(ar.sent, ar._tracker.done)
         ar._tracker.wait()
         self.assertTrue(ar.sent)
         ar.get()
     
     def test_remote_reference(self):
         v = self.client[-1]
         v['a'] = 123
         ra = clientmod.Reference('a')
         b = v.apply_sync(lambda x: x, ra)
         self.assertEquals(b, 123)
 
 
     def test_scatter_gather(self):
         view = self.client[:]
         seq1 = range(16)
         view.scatter('a', seq1)
         seq2 = view.gather('a', block=True)
         self.assertEquals(seq2, seq1)
         self.assertRaisesRemote(NameError, view.gather, 'asdf', block=True)
     
     @skip_without('numpy')
     def test_scatter_gather_numpy(self):
         import numpy
         from numpy.testing.utils import assert_array_equal, assert_array_almost_equal
         view = self.client[:]
         a = numpy.arange(64)
         view.scatter('a', a)
         b = view.gather('a', block=True)
         assert_array_equal(b, a)
     
     def test_map(self):
         view = self.client[:]
         def f(x):
             return x**2
         data = range(16)
         r = view.map_sync(f, data)
         self.assertEquals(r, map(f, data))
     
     def test_scatterGatherNonblocking(self):
         data = range(16)
         view = self.client[:]
         view.scatter('a', data, block=False)
         ar = view.gather('a', block=False)
         self.assertEquals(ar.get(), data)
     
     @skip_without('numpy')
     def test_scatter_gather_numpy_nonblocking(self):
         import numpy
         from numpy.testing.utils import assert_array_equal, assert_array_almost_equal
         a = numpy.arange(64)
         view = self.client[:]
         ar = view.scatter('a', a, block=False)
         self.assertTrue(isinstance(ar, AsyncResult))
         amr = view.gather('a', block=False)
         self.assertTrue(isinstance(amr, AsyncMapResult))
         assert_array_equal(amr.get(), a)
 
     def test_execute(self):
         view = self.client[:]
         # self.client.debug=True
         execute = view.execute
         ar = execute('c=30', block=False)
         self.assertTrue(isinstance(ar, AsyncResult))
         ar = execute('d=[0,1,2]', block=False)
         self.client.wait(ar, 1)
         self.assertEquals(len(ar.get()), len(self.client))
         for c in view['c']:
             self.assertEquals(c, 30)
     
     def test_abort(self):
         view = self.client[-1]
         ar = view.execute('import time; time.sleep(0.25)', block=False)
         ar2 = view.apply_async(lambda : 2)
         ar3 = view.apply_async(lambda : 3)
         view.abort(ar2)
         view.abort(ar3.msg_ids)
         self.assertRaises(error.TaskAborted, ar2.get)
         self.assertRaises(error.TaskAborted, ar3.get)
         
     def test_temp_flags(self):
         view = self.client[-1]
         view.block=True
         with view.temp_flags(block=False):
             self.assertFalse(view.block)
         self.assertTrue(view.block)
     
+    def test_importer(self):
+        view = self.client[-1]
+        view.clear(block=True)
+        with view.importer:
+            import re
+        
+        @interactive
+        def findall(pat, s):
+            # this globals() step isn't necessary in real code
+            # only to prevent a closure in the test
+            return globals()['re'].findall(pat, s)
+        
+        self.assertEquals(view.apply_sync(findall, '\w+', 'hello world'), 'hello world'.split())
+    

IPython/zmq/parallel/view.py

@@ -1,920 +1,1028 b''
 """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
+from IPython.utils.traitlets import HasTraits, Any, Bool, List, Dict, Set, Int, Instance, CFloat
 
 from IPython.external.decorator import decorator
 
 from . import map as Map
 from . import util
 from .asyncresult import AsyncResult, AsyncMapResult
 from .dependency import Dependency, dependent
 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
 #-----------------------------------------------------------------------------
 
 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.zmq.parallel.client.Client')
     
     _socket = Instance('zmq.Socket')
-    _ntargets = Int(1)
-    _flag_names = List(['block', 'track'])
+    _flag_names = List(['targets', 'block', 'track'])
     _targets = Any()
     _idents = Any()
     
-    def __init__(self, client=None, socket=None, targets=None):
+    def __init__(self, client=None, socket=None, **flags):
         super(View, self).__init__(client=client, _socket=socket)
-        self._ntargets = 1 if isinstance(targets, (int,type(None))) else len(targets)
         self.block = client.block
         
-        self._idents, self._targets = self.client._build_targets(targets)
-        if targets is None or isinstance(targets, int):
-            self._targets = targets
-        for name in self._flag_names:
-            # set flags, if they haven't been set yet
-            setattr(self, name, getattr(self, name, None))
+        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)
+        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!")
-    
     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
-        yield
-        # postflight: restore saved flags
-        self.set_flags(**saved_flags)
+        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, block=None):
+    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
-        return self.client.abort(jobs=jobs, targets=self._targets, block=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, verbose=False):
+    def queue_status(self, targets=None, verbose=False):
         """Fetch the Queue status of my engines"""
-        return self.client.queue_status(targets=self._targets, verbose=verbose)
+        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
+            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, 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.
+
+        In [10]: with v.importer:
+           ....:     import numpy
+           ....:     
+        importing numpy on engine(s)
+        
+        """
+        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, block=None, track=None):
+    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 self._idents:
+        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=self._targets, tracker=tracker)
+        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, block=None):
+    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)
+        return self._really_apply(util._execute, args=(code,), block=block, targets=targets)
     
-    def run(self, filename, block=None):
+    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)
+        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, block=None, track=None):
+    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)
+        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, block=True):
+    def pull(self, names, targets=None, block=True):
         """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 applier(util._pull, names)
+        return self._really_apply(util._pull, (names,), block=block, targets=targets)
     
-    def scatter(self, key, seq, dist='b', flatten=False, block=None, track=None):
+    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 = self._targets
+        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):
-            push = self.client[engineid].push
             partition = mapObject.getPartition(seq, index, nparts)
             if flatten and len(partition) == 1:
-                r = push({key: partition[0]}, block=False, track=track)
+                ns = {key: partition[0]}
             else:
-                r = push({key: partition},block=False, track=track)
+                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', block=None):
+    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(self._targets):
-            
-            msg_ids.extend(self.client[engineid].pull(key, block=False).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, block=False):
+    def clear(self, targets=None, 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)
+        targets = targets if targets is not None else self.targets
+        return self.client.clear(targets=targets, block=block)
     
-    def kill(self, block=True):
+    def kill(self, targets=None, block=True):
         """Kill my engines."""
         block = block if block is not None else self.block
-        return self.client.kill(targets=self._targets, block=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 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.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.
     
     """
     
-    _flag_names = ['block', 'track', 'follow', 'after', 'timeout']
+    follow=Any()
+    after=Any()
+    timeout=CFloat()
     
-    def __init__(self, client=None, socket=None, targets=None):
-        super(LoadBalancedView, self).__init__(client=client, socket=socket, targets=targets)
-        self._ntargets = 1
+    _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
-        if targets is None:
-            self._targets = None
-            self._idents=[]
     
     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):
+                                        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=self._idents)
+        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']
  No newline at end of file