cleanup pass

IPython/zmq/parallel/__init__.py

@@ -0,0 +1,18 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 *
+

IPython/zmq/parallel/asyncresult.py

@@ -1,294 +1,305 b''
 """AsyncResult objects for the client"""
 #-----------------------------------------------------------------------------
 #  Copyright (C) 2010  The IPython Development Team
 #
 #  Distributed under the terms of the BSD License.  The full license is in
 #  the file COPYING, distributed as part of this software.
 #-----------------------------------------------------------------------------
 
 #-----------------------------------------------------------------------------
 # Imports
 #-----------------------------------------------------------------------------
 
 import time
 
 from IPython.external.decorator import decorator
 from . import error
 
 #-----------------------------------------------------------------------------
 # Classes
 #-----------------------------------------------------------------------------
 
 @decorator
 def check_ready(f, self, *args, **kwargs):
     """Call spin() to sync state prior to calling the method."""
     self.wait(0)
     if not self._ready:
         raise error.TimeoutError("result not ready")
     return f(self, *args, **kwargs)
 
 class AsyncResult(object):
     """Class for representing results of non-blocking calls.
     
-    Provides the same interface as :py:class:`multiprocessing.AsyncResult`.
+    Provides the same interface as :py:class:`multiprocessing.pool.AsyncResult`.
     """
     
     msg_ids = None
     
     def __init__(self, client, msg_ids, fname='unknown'):
         self._client = client
         if isinstance(msg_ids, basestring):
             msg_ids = [msg_ids]
         self.msg_ids = msg_ids
         self._fname=fname
         self._ready = False
         self._success = None
         self._single_result = len(msg_ids) == 1
     
     def __repr__(self):
         if self._ready:
             return "<%s: finished>"%(self.__class__.__name__)
         else:
             return "<%s: %s>"%(self.__class__.__name__,self._fname)
     
     
     def _reconstruct_result(self, res):
-        """
+        """Reconstruct our result from actual result list (always a list)
+        
         Override me in subclasses for turning a list of results
         into the expected form.
         """
         if self._single_result:
             return res[0]
         else:
             return res
         
     def get(self, timeout=-1):
         """Return the result when it arrives. 
         
         If `timeout` is not ``None`` and the result does not arrive within
         `timeout` seconds then ``TimeoutError`` is raised. If the
         remote call raised an exception then that exception will be reraised
-        by get().
+        by get() inside a `RemoteError`.
         """
         if not self.ready():
             self.wait(timeout)
         
         if self._ready:
             if self._success:
                 return self._result
             else:
                 raise self._exception
         else:
             raise error.TimeoutError("Result not ready.")
     
     def ready(self):
         """Return whether the call has completed."""
         if not self._ready:
             self.wait(0)
         return self._ready
     
     def wait(self, timeout=-1):
         """Wait until the result is available or until `timeout` seconds pass.
+        
+        This method always returns None.
         """
         if self._ready:
             return
         self._ready = self._client.barrier(self.msg_ids, timeout)
         if self._ready:
             try:
                 results = map(self._client.results.get, self.msg_ids)
                 self._result = results
                 if self._single_result:
                     r = results[0]
                     if isinstance(r, Exception):
                         raise r
                 else:
                     results = error.collect_exceptions(results, self._fname)
                 self._result = self._reconstruct_result(results)
             except Exception, e:
                 self._exception = e
                 self._success = False
             else:
                 self._success = True
             finally:
                 self._metadata = map(self._client.metadata.get, self.msg_ids)
             
     
     def successful(self):
         """Return whether the call completed without raising an exception. 
         
         Will raise ``AssertionError`` if the result is not ready.
         """
-        assert self._ready
+        assert self.ready()
         return self._success
     
     #----------------------------------------------------------------
     # Extra methods not in mp.pool.AsyncResult
     #----------------------------------------------------------------
     
     def get_dict(self, timeout=-1):
-        """Get the results as a dict, keyed by engine_id."""
+        """Get the results as a dict, keyed by engine_id.
+        
+        timeout behavior is described in `get()`.
+        """
+        
         results = self.get(timeout)
         engine_ids = [ md['engine_id'] for md in self._metadata ]
         bycount = sorted(engine_ids, key=lambda k: engine_ids.count(k))
         maxcount = bycount.count(bycount[-1])
         if maxcount > 1:
             raise ValueError("Cannot build dict, %i jobs ran on engine #%i"%(
                     maxcount, bycount[-1]))
         
         return dict(zip(engine_ids,results))
     
     @property
     @check_ready
     def result(self):
-        """result property."""
+        """result property wrapper for `get(timeout=0)`."""
         return self._result
     
     # abbreviated alias:
     r = result
     
     @property
     @check_ready
     def metadata(self):
-        """metadata property."""
+        """property for accessing execution metadata."""
         if self._single_result:
             return self._metadata[0]
         else:
             return self._metadata
     
     @property
     def result_dict(self):
         """result property as a dict."""
         return self.get_dict(0)
     
     def __dict__(self):
         return self.get_dict(0)
 
     #-------------------------------------
     # dict-access
     #-------------------------------------
     
     @check_ready
     def __getitem__(self, key):
         """getitem returns result value(s) if keyed by int/slice, or metadata if key is str.
         """
         if isinstance(key, int):
             return error.collect_exceptions([self._result[key]], self._fname)[0]
         elif isinstance(key, slice):
             return error.collect_exceptions(self._result[key], self._fname)
         elif isinstance(key, basestring):
             values = [ md[key] for md in self._metadata ]
             if self._single_result:
                 return values[0]
             else:
                 return values
         else:
             raise TypeError("Invalid key type %r, must be 'int','slice', or 'str'"%type(key))
     
     @check_ready
     def __getattr__(self, key):
-        """getattr maps to getitem for convenient access to metadata."""
+        """getattr maps to getitem for convenient attr access to metadata."""
         if key not in self._metadata[0].keys():
             raise AttributeError("%r object has no attribute %r"%(
                     self.__class__.__name__, key))
         return self.__getitem__(key)
     
     # asynchronous iterator:
     def __iter__(self):
         if self._single_result:
             raise TypeError("AsyncResults with a single result are not iterable.")
         try:
             rlist = self.get(0)
         except error.TimeoutError:
             # wait for each result individually
             for msg_id in self.msg_ids:
                 ar = AsyncResult(self._client, msg_id, self._fname)
                 yield ar.get()
         else:
             # already done
             for r in rlist:
                 yield r
 
 
     
 class AsyncMapResult(AsyncResult):
     """Class for representing results of non-blocking gathers.
     
     This will properly reconstruct the gather.
     """
     
     def __init__(self, client, msg_ids, mapObject, fname=''):
         AsyncResult.__init__(self, client, msg_ids, fname=fname)
         self._mapObject = mapObject
         self._single_result = False
     
     def _reconstruct_result(self, res):
         """Perform the gather on the actual results."""
         return self._mapObject.joinPartitions(res)
     
     # asynchronous iterator:
     def __iter__(self):
         try:
             rlist = self.get(0)
         except error.TimeoutError:
             # wait for each result individually
             for msg_id in self.msg_ids:
                 ar = AsyncResult(self._client, msg_id, self._fname)
                 rlist = ar.get()
                 try:
                     for r in rlist:
                         yield r
                 except TypeError:
                     # flattened, not a list
                     # this could get broken by flattened data that returns iterables
                     # but most calls to map do not expose the `flatten` argument
                     yield rlist
         else:
             # already done
             for r in rlist:
                 yield r
 
 
 class AsyncHubResult(AsyncResult):
-    """Class to wrap pending results that must be requested from the Hub"""
+    """Class to wrap pending results that must be requested from the Hub.
+    
+    Note that waiting/polling on these objects requires polling the Hubover the network,
+    so use `AsyncHubResult.wait()` sparingly.
+    """
     
     def wait(self, timeout=-1):
         """wait for result to complete."""
         start = time.time()
         if self._ready:
             return
         local_ids = filter(lambda msg_id: msg_id in self._client.outstanding, self.msg_ids)
         local_ready = self._client.barrier(local_ids, timeout)
         if local_ready:
             remote_ids = filter(lambda msg_id: msg_id not in self._client.results, self.msg_ids)
             if not remote_ids:
                 self._ready = True
             else:
                 rdict = self._client.result_status(remote_ids, status_only=False)
                 pending = rdict['pending']
                 while pending and (timeout < 0 or time.time() < start+timeout):
                     rdict = self._client.result_status(remote_ids, status_only=False)
                     pending = rdict['pending']
                     if pending:
                         time.sleep(0.1)
                 if not pending:
                     self._ready = True
         if self._ready:
             try:
                 results = map(self._client.results.get, self.msg_ids)
                 self._result = results
                 if self._single_result:
                     r = results[0]
                     if isinstance(r, Exception):
                         raise r
                 else:
                     results = error.collect_exceptions(results, self._fname)
                 self._result = self._reconstruct_result(results)
             except Exception, e:
                 self._exception = e
                 self._success = False
             else:
                 self._success = True
             finally:
                 self._metadata = map(self._client.metadata.get, self.msg_ids)
         
 __all__ = ['AsyncResult', 'AsyncMapResult', 'AsyncHubResult']
  No newline at end of file

IPython/zmq/parallel/client.py

@@ -1,1499 +1,1501 b''
 """A semi-synchronous Client for the ZMQ controller"""
 #-----------------------------------------------------------------------------
 #  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 map as Map
+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 .util import ReverseDict, disambiguate_url, validate_url
+from .remotefunction import remote, parallel, ParallelFunction, RemoteFunction
+from .util import ReverseDict, validate_url, disambiguate_url
 from .view import DirectView, LoadBalancedView
 
 #--------------------------------------------------------------------------
 # helpers for implementing old MEC API via client.apply
 #--------------------------------------------------------------------------
 
 def _push(ns):
     """helper method for implementing `client.push` via `client.apply`"""
     globals().update(ns)
 
 def _pull(keys):
     """helper method for implementing `client.pull` via `client.apply`"""
     g = globals()
     if isinstance(keys, (list,tuple, set)):
         for key in keys:
             if not g.has_key(key):
                 raise NameError("name '%s' is not defined"%key)
         return map(g.get, keys)
     else:
         if not g.has_key(keys):
             raise NameError("name '%s' is not defined"%keys)
         return g.get(keys)
 
 def _clear():
     """helper method for implementing `client.clear` via `client.apply`"""
     globals().clear()
 
 def _execute(code):
     """helper method for implementing `client.execute` via `client.apply`"""
     exec code in globals()
     
 
 #--------------------------------------------------------------------------
 # Decorators for Client methods
 #--------------------------------------------------------------------------
 
 @decorator
 def spinfirst(f, self, *args, **kwargs):
     """Call spin() to sync state prior to calling the method."""
     self.spin()
     return f(self, *args, **kwargs)
 
 @decorator
 def defaultblock(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 controller
     
     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 : set 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
     
     barrier
         wait on one or more msg_ids
     
     execution methods
         apply
         legacy: execute, run
     
     query methods
         queue_status, get_result, purge
     
     control methods
         abort, shutdown
     
     """
     
     
     block = Bool(False)
     outstanding=Set()
     results = Dict()
     metadata = Dict()
     history = List()
     debug = Bool(False)
     profile=CUnicode('default')
     
     _ids = List()
     _connected=Bool(False)
     _ssh=Bool(False)
     _context = Instance('zmq.Context')
     _config = Dict()
     _engines=Instance(ReverseDict, (), {})
     _registration_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
     
     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,
             ):
         super(Client, self).__init__(debug=debug, profile=profile)
         if context is None:
             context = zmq.Context()
         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 controller!"\
             " Please specify at least one of url_or_file or profile."
         
         try:
             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'] = 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._registration_socket = self._context.socket(zmq.XREQ)
         self._registration_socket.setsockopt(zmq.IDENTITY, self.session.session)
         if self._ssh:
             tunnel.tunnel_connection(self._registration_socket, url, sshserver, **ssh_kwargs)
         else:
             self._registration_socket.connect(url)
         
         self.session.debug = self.debug
         
         self._notification_handlers = {'registration_notification' : self._register_engine,
                                     'unregistration_notification' : self._unregister_engine,
                                     }
         self._queue_handlers = {'execute_reply' : self._handle_execute_reply,
                                 'apply_reply' : self._handle_apply_reply}
         self._connect(sshserver, ssh_kwargs)
         
     
     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
     
     @property
     def ids(self):
         """Always up-to-date ids property."""
         self._flush_notifications()
         return 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):
             socket.close()
         self._closed = True
     
     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):
             targets = [targets]
         return [self._engines[t] for t in targets], list(targets)
     
     def _connect(self, sshserver, ssh_kwargs):
         """setup all our socket connections to the controller. This is called from
         __init__."""
         
         # Maybe allow reconnecting?
         if self._connected:
             return
         self._connected=True
 
         def connect_socket(s, url):
             url = 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._registration_socket, 'connection_request')
         idents,msg = self.session.recv(self._registration_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.PAIR)
                 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.PAIR)
                 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, "")
             if content.query:
                 self._query_socket = self._context.socket(zmq.PAIR)
                 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.PAIR)
                 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, '')
                 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 engineid to int."""
-        e = ss.unwrap_exception(content)
+        e = error.unwrap_exception(content)
         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 _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)
             self._engines.pop(eid)
         if self._task_socket and self._task_scheme == 'pure':
             self._stop_scheduling_tasks()
     
     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'], ss.ISO8601)
+            md['submitted'] = datetime.strptime(parent['date'], util.ISO8601)
         if 'started' in header:
-            md['started'] = datetime.strptime(header['started'], ss.ISO8601)
+            md['started'] = datetime.strptime(header['started'], util.ISO8601)
         if 'date' in header:
-            md['completed'] = datetime.strptime(header['date'], ss.ISO8601)
+            md['completed'] = datetime.strptime(header['date'], util.ISO8601)
         return md
     
     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.setdefault(msg_id, Metadata())
         md.update(self._extract_metadata(header, parent, content))
         self.metadata[msg_id] = md
         
         # construct result:
         if content['status'] == 'ok':
-            self.results[msg_id] = ss.unserialize_object(msg['buffers'])[0]
+            self.results[msg_id] = util.unserialize_object(msg['buffers'])[0]
         elif content['status'] == 'aborted':
             self.results[msg_id] = error.AbortedTask(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."""
         msg = self.session.recv(sock, mode=zmq.NOBLOCK)
         while msg is not None:
             if self.debug:
                 pprint(msg)
             msg = self.session.recv(sock, 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.setdefault(msg_id, Metadata())
             
             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']})
             
             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.view(key, balanced=False)
     
     #--------------------------------------------------------------------------
     # Begin public methods
     #--------------------------------------------------------------------------
     
     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)
     
     def barrier(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
     #--------------------------------------------------------------------------
     
     @spinfirst
     @defaultblock
     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:
             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'])
         if error:
             return error
         
     
     @spinfirst
     @defaultblock
     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:
             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'])
         if error:
             return error
     
     @spinfirst
     @defaultblock
     def shutdown(self, targets=None, restart=False, controller=False, block=None):
         """Terminates one or more engine processes, optionally including the controller."""
         if controller:
             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 controller:
             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'])
         
         if controller:
             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
     #--------------------------------------------------------------------------
     
     @defaultblock
     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
         """
         result = self.apply(_execute, (code,), targets=targets, block=block, bound=True, balanced=False)
         if not block:
             return result
     
     def run(self, filename, targets='all', block=None):
         """Execute contents of `filename` on 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, 'rb') as f:
             code = f.read()
         return self.execute(code, targets=targets, 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 _build_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))
         
     @defaultblock
     def apply(self, f, args=None, kwargs=None, bound=True, block=None,
                         targets=None, balanced=None,
                         after=None, follow=None, timeout=None):
         """Call `f(*args, **kwargs)` on a remote engine(s), returning the result.
         
         This is the central execution command for the client.
         
         Parameters
         ----------
         
         f : function
             The fuction to be called remotely
         args : tuple/list
             The positional arguments passed to `f`
         kwargs : dict
             The keyword arguments passed to `f`
         bound : bool (default: True)
             Whether to execute in the Engine(s) namespace, or in a clean
             namespace not affecting the engine.
         block : bool (default: self.block)
             Whether to wait for the result, or return immediately.
             False:
                 returns AsyncResult
             True:
                 returns actual result(s) of f(*args, **kwargs)
                 if multiple targets:
                     list of results, matching `targets`
         targets : int,list of ints, 'all', None
             Specify the destination of the job.
             if None:
                 Submit via Task queue for load-balancing.
             if 'all':
                 Run on all active engines
             if list:
                 Run on each specified engine
             if int:
                 Run on single engine
         
         balanced : bool, default None
             whether to load-balance.  This will default to True
             if targets is unspecified, or False if targets is specified.
             
             The following arguments are only used when balanced is True:
         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.
         
         after,follow,timeout only used if `balanced=True`.
         
         Returns
         -------
         
         if block is False:
             return AsyncResult wrapping msg_ids
             output of AsyncResult.get() is identical to that of `apply(...block=True)`
         else:
             if single target:
                 return result of `f(*args, **kwargs)`
             else:
                 return list of results, matching `targets`
         """
         assert not self._closed, "cannot use me anymore, I'm closed!"
         # defaults:
         block = block if block is not None else self.block
         args = args if args is not None else []
         kwargs = kwargs if kwargs is not None else {}
         
         if balanced is None:
             if targets is None:
                 # default to balanced if targets unspecified
                 balanced = True
             else:
                 # otherwise default to multiplexing
                 balanced = False
         
         if targets is None and balanced is False:
             # default to all if *not* balanced, and targets is unspecified
             targets = 'all'
         
         # enforce types of f,args,kwrags
         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))
         
         options  = dict(bound=bound, block=block, targets=targets)
         
         if balanced:
             return self._apply_balanced(f, args, kwargs, timeout=timeout, 
                                         after=after, follow=follow, **options)
         elif follow or after or timeout:
                 msg = "follow, after, and timeout args are only used for"
                 msg += " load-balanced execution."
                 raise ValueError(msg)
         else:
             return self._apply_direct(f, args, kwargs, **options)
     
     def _apply_balanced(self, f, args, kwargs, bound=None, block=None, targets=None,
                             after=None, follow=None, timeout=None):
         """call f(*args, **kwargs) remotely in a load-balanced manner.
         
         This is a private method, see `apply` for details.
         Not to be called directly!
         """
         
         loc = locals()
         for name in ('bound', 'block'):
             assert loc[name] is not None, "kwarg %r must be specified!"%name
         
         if self._task_socket is None:
             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)
         
         # defaults:
         args = args if args is not None else []
         kwargs = kwargs if kwargs is not None else {}
         
         if targets:
             idents,_ = self._build_targets(targets)
         else:
             idents = []
         
         after = self._build_dependency(after)
         follow = self._build_dependency(follow)
         subheader = dict(after=after, follow=follow, timeout=timeout, targets=idents)
-        bufs = ss.pack_apply_message(f,args,kwargs)
+        bufs = util.pack_apply_message(f,args,kwargs)
         content = dict(bound=bound)
         
         msg = self.session.send(self._task_socket, "apply_request", 
                 content=content, buffers=bufs, subheader=subheader)
         msg_id = msg['msg_id']
         self.outstanding.add(msg_id)
         self.history.append(msg_id)
         ar = AsyncResult(self, [msg_id], fname=f.__name__)
         if block:
             try:
                 return ar.get()
             except KeyboardInterrupt:
                 return ar
         else:
             return ar
     
     def _apply_direct(self, f, args, kwargs, bound=None, block=None, targets=None):
         """Then underlying method for applying functions to specific engines
         via the MUX queue.
         
         This is a private method, see `apply` for details.
         Not to be called directly!
         """
         loc = locals()
         for name in ('bound', 'block', 'targets'):
             assert loc[name] is not None, "kwarg %r must be specified!"%name
         
         idents,targets = self._build_targets(targets)
         
         subheader = {}
         content = dict(bound=bound)
-        bufs = ss.pack_apply_message(f,args,kwargs)
+        bufs = util.pack_apply_message(f,args,kwargs)
         
         msg_ids = []
         for ident in idents:
             msg = self.session.send(self._mux_socket, "apply_request", 
                     content=content, buffers=bufs, ident=ident, subheader=subheader)
             msg_id = msg['msg_id']
             self.outstanding.add(msg_id)
             self.history.append(msg_id)
             msg_ids.append(msg_id)
         ar = AsyncResult(self, msg_ids, fname=f.__name__)
         if block:
             try:
                 return ar.get()
             except KeyboardInterrupt:
                 return ar
         else:
             return ar
     
     #--------------------------------------------------------------------------
     # construct a View object
     #--------------------------------------------------------------------------
     
     @defaultblock
     def remote(self, bound=True, block=None, targets=None, balanced=None):
         """Decorator for making a RemoteFunction"""
         return remote(self, bound=bound, targets=targets, block=block, balanced=balanced)
     
     @defaultblock
     def parallel(self, dist='b', bound=True, block=None, targets=None, balanced=None):
         """Decorator for making a ParallelFunction"""
         return parallel(self, bound=bound, targets=targets, block=block, balanced=balanced)
     
     def _cache_view(self, targets, balanced):
         """save views, so subsequent requests don't create new objects."""
         if balanced:
             view_class = LoadBalancedView
             view_cache = self._balanced_views
         else:
             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, targets=targets)
         
         return view_cache[key]
     
     def view(self, targets=None, balanced=None):
         """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
         
         """
         
         balanced = (targets is None) if balanced is None else balanced
         
         if targets is None:
             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)))
     
     #--------------------------------------------------------------------------
     # Data movement
     #--------------------------------------------------------------------------
     
     @defaultblock
     def push(self, ns, targets='all', block=None):
         """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(_push, (ns,), targets=targets, block=block, bound=True, balanced=False)
         if not block:
             return result
     
     @defaultblock
     def pull(self, keys, targets='all', block=None):
         """Pull objects from `target`'s namespace by `keys`"""
         if isinstance(keys, str):
             pass
         elif isinstance(keys, (list,tuple,set)):
             for key in keys:
                 if not isinstance(key, str):
                     raise TypeError
         result = self.apply(_pull, (keys,), targets=targets, block=block, bound=True, balanced=False)
         return result
     
     @defaultblock
     def scatter(self, key, seq, dist='b', flatten=False, targets='all', block=None):
         """
         Partition a Python sequence and send the partitions to a set of engines.
         """
         targets = self._build_targets(targets)[-1]
         mapObject = Map.dists[dist]()
         nparts = len(targets)
         msg_ids = []
         for index, engineid in enumerate(targets):
             partition = mapObject.getPartition(seq, index, nparts)
             if flatten and len(partition) == 1:
                 r = self.push({key: partition[0]}, targets=engineid, block=False)
             else:
                 r = self.push({key: partition}, targets=engineid, block=False)
             msg_ids.extend(r.msg_ids)
         r = AsyncResult(self, msg_ids, fname='scatter')
         if block:
             r.get()
         else:
             return r
     
     @defaultblock
     def gather(self, key, dist='b', targets='all', block=None):
         """
         Gather a partitioned sequence on a set of engines as a single local seq.
         """
         
         targets = self._build_targets(targets)[-1]
         mapObject = Map.dists[dist]()
         msg_ids = []
         for index, engineid in enumerate(targets):
             msg_ids.extend(self.pull(key, targets=engineid,block=False).msg_ids)
         
         r = AsyncMapResult(self, msg_ids, mapObject, fname='gather')
         if block:
             return r.get()
         else:
             return r
     
     #--------------------------------------------------------------------------
     # Query methods
     #--------------------------------------------------------------------------
     
     @spinfirst
     @defaultblock
     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
     
     @spinfirst
     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)):
             indices_or_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.setdefault(msg_id, Metadata())
                 md.update(self._extract_metadata(header, parent, rcontent))
                 md.update(iodict)
                 
                 if rcontent['status'] == 'ok':
-                    res,buffers = ss.unserialize_object(buffers)
+                    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
 
     @spinfirst
     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
         """
         targets = self._build_targets(targets)[1]
         content = dict(targets=targets, 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)
-        return ss.rekey(content)
+        return util.rekey(content)
         
     @spinfirst
     def purge_results(self, jobs=[], targets=[]):
         """Tell the controller 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 strs 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 controller'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'
+            'AsyncMapResult',
+            'Reference'
             ]

IPython/zmq/parallel/clusterdir.py

@@ -1,538 +1,537 b''
 #!/usr/bin/env python
 # encoding: utf-8
 """
 The IPython cluster directory
 """
 
 #-----------------------------------------------------------------------------
 #  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
 #-----------------------------------------------------------------------------
 
 from __future__ import with_statement
 
 import os
 import logging
 import re
 import shutil
 import sys
-import warnings
 
 from IPython.config.loader import PyFileConfigLoader
 from IPython.config.configurable import Configurable
 from IPython.core.application import Application, BaseAppConfigLoader
 from IPython.core.crashhandler import CrashHandler
 from IPython.core import release
 from IPython.utils.path import (
     get_ipython_package_dir,
     expand_path
 )
 from IPython.utils.traitlets import Unicode
 
 #-----------------------------------------------------------------------------
 # Module errors
 #-----------------------------------------------------------------------------
 
 class ClusterDirError(Exception):
     pass
 
 
 class PIDFileError(Exception):
     pass
 
 
 #-----------------------------------------------------------------------------
 # Class for managing cluster directories
 #-----------------------------------------------------------------------------
 
 class ClusterDir(Configurable):
     """An object to manage the cluster directory and its resources.
 
     The cluster directory is used by :command:`ipengine`, 
     :command:`ipcontroller` and :command:`ipclsuter` to manage the
     configuration, logging and security of these applications.
 
     This object knows how to find, create and manage these directories. This
     should be used by any code that want's to handle cluster directories.
     """
 
     security_dir_name = Unicode('security')
     log_dir_name = Unicode('log')
     pid_dir_name = Unicode('pid')
     security_dir = Unicode(u'')
     log_dir = Unicode(u'')
     pid_dir = Unicode(u'')
     location = Unicode(u'')
 
     def __init__(self, location=u''):
         super(ClusterDir, self).__init__(location=location)
 
     def _location_changed(self, name, old, new):
         if not os.path.isdir(new):
             os.makedirs(new)
         self.security_dir = os.path.join(new, self.security_dir_name)
         self.log_dir = os.path.join(new, self.log_dir_name)
         self.pid_dir = os.path.join(new, self.pid_dir_name)
         self.check_dirs()
 
     def _log_dir_changed(self, name, old, new):
         self.check_log_dir()
 
     def check_log_dir(self):
         if not os.path.isdir(self.log_dir):
             os.mkdir(self.log_dir)
 
     def _security_dir_changed(self, name, old, new):
         self.check_security_dir()
 
     def check_security_dir(self):
         if not os.path.isdir(self.security_dir):
             os.mkdir(self.security_dir, 0700)
         os.chmod(self.security_dir, 0700)
 
     def _pid_dir_changed(self, name, old, new):
         self.check_pid_dir()
 
     def check_pid_dir(self):
         if not os.path.isdir(self.pid_dir):
             os.mkdir(self.pid_dir, 0700)
         os.chmod(self.pid_dir, 0700)
 
     def check_dirs(self):
         self.check_security_dir()
         self.check_log_dir()
         self.check_pid_dir()
 
     def load_config_file(self, filename):
         """Load a config file from the top level of the cluster dir.
 
         Parameters
         ----------
         filename : unicode or str
             The filename only of the config file that must be located in
             the top-level of the cluster directory.
         """
         loader = PyFileConfigLoader(filename, self.location)
         return loader.load_config()
 
     def copy_config_file(self, config_file, path=None, overwrite=False):
         """Copy a default config file into the active cluster directory.
 
         Default configuration files are kept in :mod:`IPython.config.default`.
         This function moves these from that location to the working cluster
         directory.
         """
         if path is None:
             import IPython.config.default
             path = IPython.config.default.__file__.split(os.path.sep)[:-1]
             path = os.path.sep.join(path)
         src = os.path.join(path, config_file)
         dst = os.path.join(self.location, config_file)
         if not os.path.isfile(dst) or overwrite:
             shutil.copy(src, dst)
 
     def copy_all_config_files(self, path=None, overwrite=False):
         """Copy all config files into the active cluster directory."""
         for f in [u'ipcontrollerz_config.py', u'ipenginez_config.py',
                   u'ipclusterz_config.py']:
             self.copy_config_file(f, path=path, overwrite=overwrite)
 
     @classmethod
     def create_cluster_dir(csl, cluster_dir):
         """Create a new cluster directory given a full path.
 
         Parameters
         ----------
         cluster_dir : str
             The full path to the cluster directory.  If it does exist, it will
             be used.  If not, it will be created.
         """
         return ClusterDir(location=cluster_dir)
 
     @classmethod
     def create_cluster_dir_by_profile(cls, path, profile=u'default'):
         """Create a cluster dir by profile name and path.
 
         Parameters
         ----------
         path : str
             The path (directory) to put the cluster directory in.
         profile : str
             The name of the profile.  The name of the cluster directory will
             be "clusterz_<profile>".
         """
         if not os.path.isdir(path):
             raise ClusterDirError('Directory not found: %s' % path)
         cluster_dir = os.path.join(path, u'clusterz_' + profile)
         return ClusterDir(location=cluster_dir)
 
     @classmethod
     def find_cluster_dir_by_profile(cls, ipython_dir, profile=u'default'):
         """Find an existing cluster dir by profile name, return its ClusterDir.
 
         This searches through a sequence of paths for a cluster dir.  If it
         is not found, a :class:`ClusterDirError` exception will be raised.
 
         The search path algorithm is:
         1. ``os.getcwd()``
         2. ``ipython_dir``
         3. The directories found in the ":" separated 
            :env:`IPCLUSTER_DIR_PATH` environment variable.
 
         Parameters
         ----------
         ipython_dir : unicode or str
             The IPython directory to use.
         profile : unicode or str
             The name of the profile.  The name of the cluster directory
             will be "clusterz_<profile>".
         """
         dirname = u'clusterz_' + profile
         cluster_dir_paths = os.environ.get('IPCLUSTER_DIR_PATH','')
         if cluster_dir_paths:
             cluster_dir_paths = cluster_dir_paths.split(':')
         else:
             cluster_dir_paths = []
         paths = [os.getcwd(), ipython_dir] + cluster_dir_paths
         for p in paths:
             cluster_dir = os.path.join(p, dirname)
             if os.path.isdir(cluster_dir):
                 return ClusterDir(location=cluster_dir)
         else:
             raise ClusterDirError('Cluster directory not found in paths: %s' % dirname)
 
     @classmethod
     def find_cluster_dir(cls, cluster_dir):
         """Find/create a cluster dir and return its ClusterDir.
 
         This will create the cluster directory if it doesn't exist.
 
         Parameters
         ----------
         cluster_dir : unicode or str
             The path of the cluster directory.  This is expanded using
             :func:`IPython.utils.genutils.expand_path`.
         """
         cluster_dir = expand_path(cluster_dir)
         if not os.path.isdir(cluster_dir):
             raise ClusterDirError('Cluster directory not found: %s' % cluster_dir)
         return ClusterDir(location=cluster_dir)
 
 
 #-----------------------------------------------------------------------------
 # Command line options
 #-----------------------------------------------------------------------------
 
 class ClusterDirConfigLoader(BaseAppConfigLoader):
 
     def _add_cluster_profile(self, parser):
         paa = parser.add_argument
         paa('-p', '--profile',
             dest='Global.profile',type=unicode,
             help=
             """The string name of the profile to be used. This determines the name
             of the cluster dir as: cluster_<profile>. The default profile is named
             'default'.  The cluster directory is resolve this way if the
             --cluster-dir option is not used.""",
             metavar='Global.profile')
 
     def _add_cluster_dir(self, parser):
         paa = parser.add_argument
         paa('--cluster-dir',
             dest='Global.cluster_dir',type=unicode,
             help="""Set the cluster dir. This overrides the logic used by the
             --profile option.""",
             metavar='Global.cluster_dir')
 
     def _add_work_dir(self, parser):
         paa = parser.add_argument
         paa('--work-dir',
             dest='Global.work_dir',type=unicode,
             help='Set the working dir for the process.',
             metavar='Global.work_dir')
 
     def _add_clean_logs(self, parser):
         paa = parser.add_argument
         paa('--clean-logs',
             dest='Global.clean_logs', action='store_true',
             help='Delete old log flies before starting.')
 
     def _add_no_clean_logs(self, parser):
         paa = parser.add_argument
         paa('--no-clean-logs',
             dest='Global.clean_logs', action='store_false',
             help="Don't Delete old log flies before starting.")
 
     def _add_arguments(self):
         super(ClusterDirConfigLoader, self)._add_arguments()
         self._add_cluster_profile(self.parser)
         self._add_cluster_dir(self.parser)
         self._add_work_dir(self.parser)
         self._add_clean_logs(self.parser)
         self._add_no_clean_logs(self.parser)
 
 
 #-----------------------------------------------------------------------------
 # Crash handler for this application
 #-----------------------------------------------------------------------------
 
 
 _message_template = """\
 Oops, $self.app_name crashed. We do our best to make it stable, but...
 
 A crash report was automatically generated with the following information:
   - A verbatim copy of the crash traceback.
   - Data on your current $self.app_name configuration.
 
 It was left in the file named:
 \t'$self.crash_report_fname'
 If you can email this file to the developers, the information in it will help
 them in understanding and correcting the problem.
 
 You can mail it to: $self.contact_name at $self.contact_email
 with the subject '$self.app_name Crash Report'.
 
 If you want to do it now, the following command will work (under Unix):
 mail -s '$self.app_name Crash Report' $self.contact_email < $self.crash_report_fname
 
 To ensure accurate tracking of this issue, please file a report about it at:
 $self.bug_tracker
 """
 
 class ClusterDirCrashHandler(CrashHandler):
     """sys.excepthook for IPython itself, leaves a detailed report on disk."""
 
     message_template = _message_template
 
     def __init__(self, app):
         contact_name = release.authors['Brian'][0]
         contact_email = release.authors['Brian'][1]
         bug_tracker = 'http://github.com/ipython/ipython/issues'
         super(ClusterDirCrashHandler,self).__init__(
             app, contact_name, contact_email, bug_tracker
         )
 
 
 #-----------------------------------------------------------------------------
 # Main application
 #-----------------------------------------------------------------------------
 
 class ApplicationWithClusterDir(Application):
     """An application that puts everything into a cluster directory.
 
     Instead of looking for things in the ipython_dir, this type of application
     will use its own private directory called the "cluster directory"
     for things like config files, log files, etc.
 
     The cluster directory is resolved as follows:
 
     * If the ``--cluster-dir`` option is given, it is used.
     * If ``--cluster-dir`` is not given, the application directory is 
       resolve using the profile name as ``cluster_<profile>``. The search 
       path for this directory is then i) cwd if it is found there
       and ii) in ipython_dir otherwise.
 
     The config file for the application is to be put in the cluster
     dir and named the value of the ``config_file_name`` class attribute.
     """
 
     command_line_loader = ClusterDirConfigLoader
     crash_handler_class = ClusterDirCrashHandler
     auto_create_cluster_dir = True
     # temporarily override default_log_level to INFO
     default_log_level = logging.INFO
 
     def create_default_config(self):
         super(ApplicationWithClusterDir, self).create_default_config()
         self.default_config.Global.profile = u'default'
         self.default_config.Global.cluster_dir = u''
         self.default_config.Global.work_dir = os.getcwd()
         self.default_config.Global.log_to_file = False
         self.default_config.Global.log_url = None
         self.default_config.Global.clean_logs = False
 
     def find_resources(self):
         """This resolves the cluster directory.
 
         This tries to find the cluster directory and if successful, it will
         have done:
         * Sets ``self.cluster_dir_obj`` to the :class:`ClusterDir` object for 
           the application.
         * Sets ``self.cluster_dir`` attribute of the application and config
           objects.
 
         The algorithm used for this is as follows:
         1. Try ``Global.cluster_dir``.
         2. Try using ``Global.profile``.
         3. If both of these fail and ``self.auto_create_cluster_dir`` is
            ``True``, then create the new cluster dir in the IPython directory.
         4. If all fails, then raise :class:`ClusterDirError`.
         """
 
         try:
             cluster_dir = self.command_line_config.Global.cluster_dir
         except AttributeError:
             cluster_dir = self.default_config.Global.cluster_dir
         cluster_dir = expand_path(cluster_dir)
         try:
             self.cluster_dir_obj = ClusterDir.find_cluster_dir(cluster_dir)
         except ClusterDirError:
             pass
         else:
             self.log.info('Using existing cluster dir: %s' % \
                 self.cluster_dir_obj.location
             )
             self.finish_cluster_dir()
             return
 
         try:
             self.profile = self.command_line_config.Global.profile
         except AttributeError:
             self.profile = self.default_config.Global.profile
         try:
             self.cluster_dir_obj = ClusterDir.find_cluster_dir_by_profile(
                 self.ipython_dir, self.profile)
         except ClusterDirError:
             pass
         else:
             self.log.info('Using existing cluster dir: %s' % \
                 self.cluster_dir_obj.location
             )
             self.finish_cluster_dir()
             return
 
         if self.auto_create_cluster_dir:
             self.cluster_dir_obj = ClusterDir.create_cluster_dir_by_profile(
                 self.ipython_dir, self.profile
             )
             self.log.info('Creating new cluster dir: %s' % \
                 self.cluster_dir_obj.location
             )
             self.finish_cluster_dir()
         else:
             raise ClusterDirError('Could not find a valid cluster directory.')
 
     def finish_cluster_dir(self):
         # Set the cluster directory
         self.cluster_dir = self.cluster_dir_obj.location
         
         # These have to be set because they could be different from the one
         # that we just computed.  Because command line has the highest
         # priority, this will always end up in the master_config.
         self.default_config.Global.cluster_dir = self.cluster_dir
         self.command_line_config.Global.cluster_dir = self.cluster_dir
 
     def find_config_file_name(self):
         """Find the config file name for this application."""
         # For this type of Application it should be set as a class attribute.
         if not hasattr(self, 'default_config_file_name'):
             self.log.critical("No config filename found")
         else:
             self.config_file_name = self.default_config_file_name
 
     def find_config_file_paths(self):
         # Set the search path to to the cluster directory. We should NOT
         # include IPython.config.default here as the default config files
         # are ALWAYS automatically moved to the cluster directory.
         conf_dir = os.path.join(get_ipython_package_dir(), 'config', 'default')
         self.config_file_paths = (self.cluster_dir,)
 
     def pre_construct(self):
         # The log and security dirs were set earlier, but here we put them
         # into the config and log them.
         config = self.master_config
         sdir = self.cluster_dir_obj.security_dir
         self.security_dir = config.Global.security_dir = sdir
         ldir = self.cluster_dir_obj.log_dir
         self.log_dir = config.Global.log_dir = ldir
         pdir = self.cluster_dir_obj.pid_dir
         self.pid_dir = config.Global.pid_dir = pdir
         self.log.info("Cluster directory set to: %s" % self.cluster_dir)
         config.Global.work_dir = unicode(expand_path(config.Global.work_dir))
         # Change to the working directory. We do this just before construct
         # is called so all the components there have the right working dir.
         self.to_work_dir()
 
     def to_work_dir(self):
         wd = self.master_config.Global.work_dir
         if unicode(wd) != unicode(os.getcwd()):
             os.chdir(wd)
             self.log.info("Changing to working dir: %s" % wd)
 
     def start_logging(self):
         # Remove old log files
         if self.master_config.Global.clean_logs:
             log_dir = self.master_config.Global.log_dir
             for f in os.listdir(log_dir):
                 if re.match(r'%s-\d+\.(log|err|out)'%self.name,f):
                 # if f.startswith(self.name + u'-') and f.endswith('.log'):
                     os.remove(os.path.join(log_dir, f))
         # Start logging to the new log file
         if self.master_config.Global.log_to_file:
             log_filename = self.name + u'-' + str(os.getpid()) + u'.log'
             logfile = os.path.join(self.log_dir, log_filename)
             open_log_file = open(logfile, 'w')
         elif self.master_config.Global.log_url:
             open_log_file = None
         else:
             open_log_file = sys.stdout
         if open_log_file is not None:
             self.log.removeHandler(self._log_handler)
             self._log_handler = logging.StreamHandler(open_log_file)
             self._log_formatter = logging.Formatter("[%(name)s] %(message)s")
             self._log_handler.setFormatter(self._log_formatter)
             self.log.addHandler(self._log_handler)
         # log.startLogging(open_log_file)
 
     def write_pid_file(self, overwrite=False):
         """Create a .pid file in the pid_dir with my pid.
 
         This must be called after pre_construct, which sets `self.pid_dir`.
         This raises :exc:`PIDFileError` if the pid file exists already.
         """
         pid_file = os.path.join(self.pid_dir, self.name + u'.pid')
         if os.path.isfile(pid_file):
             pid = self.get_pid_from_file()
             if not overwrite:
                 raise PIDFileError(
                     'The pid file [%s] already exists. \nThis could mean that this '
                     'server is already running with [pid=%s].' % (pid_file, pid)
                 )
         with open(pid_file, 'w') as f:
             self.log.info("Creating pid file: %s" % pid_file)
             f.write(repr(os.getpid())+'\n')
 
     def remove_pid_file(self):
         """Remove the pid file.
 
         This should be called at shutdown by registering a callback with
         :func:`reactor.addSystemEventTrigger`. This needs to return
         ``None``.
         """
         pid_file = os.path.join(self.pid_dir, self.name + u'.pid')
         if os.path.isfile(pid_file):
             try:
                 self.log.info("Removing pid file: %s" % pid_file)
                 os.remove(pid_file)
             except:
                 self.log.warn("Error removing the pid file: %s" % pid_file)
 
     def get_pid_from_file(self):
         """Get the pid from the pid file.
 
         If the  pid file doesn't exist a :exc:`PIDFileError` is raised.
         """
         pid_file = os.path.join(self.pid_dir, self.name + u'.pid')
         if os.path.isfile(pid_file):
             with open(pid_file, 'r') as f:
                 pid = int(f.read().strip())
                 return pid
         else:
             raise PIDFileError('pid file not found: %s' % pid_file)
 

IPython/zmq/parallel/controller.py

@@ -1,115 +1,115 b''
 #!/usr/bin/env python
 """The IPython Controller with 0MQ
 This is a collection of one Hub and several Schedulers.
 """
 #-----------------------------------------------------------------------------
 #  Copyright (C) 2010  The IPython Development Team
 #
 #  Distributed under the terms of the BSD License.  The full license is in
 #  the file COPYING, distributed as part of this software.
 #-----------------------------------------------------------------------------
 
 #-----------------------------------------------------------------------------
 # Imports
 #-----------------------------------------------------------------------------
 from __future__ import print_function
 
 import logging
 from multiprocessing import Process
 
 import zmq
 from zmq.devices import ProcessMonitoredQueue
 # internal:
 from IPython.utils.importstring import import_item
-from IPython.utils.traitlets import Int, Str, Instance, List, Bool
+from IPython.utils.traitlets import Int, CStr, Instance, List, Bool
 
 from .entry_point import signal_children
 from .hub import Hub, HubFactory
 from .scheduler import launch_scheduler
 
 #-----------------------------------------------------------------------------
 # Configurable
 #-----------------------------------------------------------------------------
 
 
 class ControllerFactory(HubFactory):
     """Configurable for setting up a Hub and Schedulers."""
     
     usethreads = Bool(False, config=True)
     # pure-zmq downstream HWM
     hwm = Int(0, config=True)
     
     # internal
     children = List()
-    mq_class = Str('zmq.devices.ProcessMonitoredQueue')
+    mq_class = CStr('zmq.devices.ProcessMonitoredQueue')
     
     def _usethreads_changed(self, name, old, new):
         self.mq_class = 'zmq.devices.%sMonitoredQueue'%('Thread' if new else 'Process')
         
     def __init__(self, **kwargs):
         super(ControllerFactory, self).__init__(**kwargs)
         self.subconstructors.append(self.construct_schedulers)
     
     def start(self):
         super(ControllerFactory, self).start()
         child_procs = []
         for child in self.children:
             child.start()
             if isinstance(child, ProcessMonitoredQueue):
                 child_procs.append(child.launcher)
             elif isinstance(child, Process):
                 child_procs.append(child)
         if child_procs:
             signal_children(child_procs)
         
     
     def construct_schedulers(self):
         children = self.children
         mq = import_item(self.mq_class)
         
         maybe_inproc = 'inproc://monitor' if self.usethreads else self.monitor_url
         # IOPub relay (in a Process)
         q = mq(zmq.PUB, zmq.SUB, zmq.PUB, 'N/A','iopub')
         q.bind_in(self.client_info['iopub'])
         q.bind_out(self.engine_info['iopub'])
         q.setsockopt_out(zmq.SUBSCRIBE, '')
         q.connect_mon(maybe_inproc)
         q.daemon=True
         children.append(q)
 
         # Multiplexer Queue (in a Process)
         q = mq(zmq.XREP, zmq.XREP, zmq.PUB, 'in', 'out')
         q.bind_in(self.client_info['mux'])
         q.bind_out(self.engine_info['mux'])
         q.connect_mon(maybe_inproc)
         q.daemon=True
         children.append(q)
 
         # Control Queue (in a Process)
         q = mq(zmq.XREP, zmq.XREP, zmq.PUB, 'incontrol', 'outcontrol')
         q.bind_in(self.client_info['control'])
         q.bind_out(self.engine_info['control'])
         q.connect_mon(maybe_inproc)
         q.daemon=True
         children.append(q)
         # Task Queue (in a Process)
         if self.scheme == 'pure':
             self.log.warn("task::using pure XREQ Task scheduler")
             q = mq(zmq.XREP, zmq.XREQ, zmq.PUB, 'intask', 'outtask')
             q.setsockopt_out(zmq.HWM, self.hwm)
             q.bind_in(self.client_info['task'][1])
             q.bind_out(self.engine_info['task'])
             q.connect_mon(maybe_inproc)
             q.daemon=True
             children.append(q)
         elif self.scheme == 'none':
             self.log.warn("task::using no Task scheduler")
             
         else:
             self.log.info("task::using Python %s Task scheduler"%self.scheme)
             sargs = (self.client_info['task'][1], self.engine_info['task'], self.monitor_url, self.client_info['notification'])
             kwargs = dict(scheme=self.scheme,logname=self.log.name, loglevel=self.log.level, config=self.config)
             q = Process(target=launch_scheduler, args=sargs, kwargs=kwargs)
             q.daemon=True
             children.append(q)
 

IPython/zmq/parallel/dependency.py

@@ -1,111 +1,153 b''
 """Dependency utilities"""
 
 from IPython.external.decorator import decorator
 
 from .asyncresult import AsyncResult
 from .error import UnmetDependency
 
 
 class depend(object):
-    """Dependency decorator, for use with tasks."""
+    """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 self.df(*self.dargs, **self.dkwargs) is False:
             raise UnmetDependency()
         return self.f(*args, **kwargs)
     
     @property
     def __name__(self):
         return self.func_name
 
 def _require(*names):
+    """Helper for @require decorator."""
     for name in names:
         try:
             __import__(name)
         except ImportError:
             return False
     return True
 
 def require(*names):
+    """Simple decorator for requiring names to be importable.
+    
+    Examples
+    --------
+    
+    In [1]: @require('numpy')
+       ...: def norm(a):
+       ...:     import numpy
+       ...:     return numpy.linalg.norm(a,2)
+    """
     return depend(_require, *names)
 
 class Dependency(set):
     """An object for representing a set of msg_id dependencies.
     
-    Subclassed from set()."""
+    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_only : bool [default True]
+        Whether to consider only successes for Dependencies, or consider failures as well.
+        If `all=success_only=True`, then this task will fail with an ImpossibleDependency
+        as soon as the first depended-upon task fails.
+    """
     
     all=True
     success_only=True
     
     def __init__(self, dependencies=[], all=True, success_only=True):
         if isinstance(dependencies, dict):
             # load from dict
             all = dependencies.get('all', True)
             success_only = dependencies.get('success_only', success_only)
             dependencies = dependencies.get('dependencies', [])
         ids = []
         if isinstance(dependencies, AsyncResult):
             ids.extend(AsyncResult.msg_ids)
         else:
             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
         self.success_only=success_only
     
     def check(self, completed, failed=None):
         if failed is not None and not self.success_only:
             completed = completed.union(failed)
         if len(self) == 0:
             return True
         if self.all:
             return self.issubset(completed)
         else:
             return not self.isdisjoint(completed)
     
     def unreachable(self, failed):
         if len(self) == 0 or len(failed) == 0 or not self.success_only:
             return False
         # print self, self.success_only, self.all, failed
         if self.all:
             return not self.isdisjoint(failed)
         else:
             return self.issubset(failed)
         
     
     def as_dict(self):
         """Represent this dependency as a dict. For json compatibility."""
         return dict(
             dependencies=list(self),
             all=self.all,
             success_only=self.success_only,
         )
     
 
 __all__ = ['depend', 'require', 'dependent', 'Dependency']
 

IPython/zmq/parallel/dictdb.py

@@ -1,152 +1,152 b''
 """A Task logger that presents our DB interface, 
 but exists entirely in memory and implemented with dicts.
 
 TaskRecords are dicts of the form:
 {
     'msg_id' : str(uuid),
     'client_uuid' : str(uuid),
     'engine_uuid' : str(uuid) or None,
     'header' : dict(header),
     'content': dict(content),
     'buffers': list(buffers),
     'submitted': datetime,
     'started': datetime or None,
     'completed': datetime or None,
     'resubmitted': datetime or None,
     'result_header' : dict(header) or None,
     'result_content' : dict(content) or None,
     'result_buffers' : list(buffers) or None,
 }
 With this info, many of the special categories of tasks can be defined by query:
 
 pending:  completed is None
 client's outstanding: client_uuid = uuid && completed is None
 MIA: arrived is None (and completed is None)
 etc.
 
 EngineRecords are dicts of the form:
 {
     'eid' : int(id),
     'uuid': str(uuid)
 }
 This may be extended, but is currently.
 
 We support a subset of mongodb operators:
     $lt,$gt,$lte,$gte,$ne,$in,$nin,$all,$mod,$exists
 """
 #-----------------------------------------------------------------------------
 #  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.
 #-----------------------------------------------------------------------------
 
 
 from datetime import datetime
 
 filters = {
- '$eq' : lambda a,b: a==b,
  '$lt' : lambda a,b: a < b,
  '$gt' : lambda a,b: b > a,
+ '$eq' : lambda a,b: a == b,
+ '$ne' : lambda a,b: a != b,
  '$lte': lambda a,b: a <= b,
  '$gte': lambda a,b: a >= b,
- '$ne' : lambda a,b: not a==b,
  '$in' : lambda a,b: a in b,
  '$nin': lambda a,b: a not in b,
- '$all' : lambda a,b: all([ a in bb for bb in b ]),
+ '$all': lambda a,b: all([ a in bb for bb in b ]),
  '$mod': lambda a,b: a%b[0] == b[1],
  '$exists' : lambda a,b: (b and a is not None) or (a is None and not b)
 }
 
 
 class CompositeFilter(object):
     """Composite filter for matching multiple properties."""
     
     def __init__(self, dikt):
         self.tests = []
         self.values = []
         for key, value in dikt.iteritems():
             self.tests.append(filters[key])
             self.values.append(value)
 
     def __call__(self, value):
         for test,check in zip(self.tests, self.values):
             if not test(value, check):
                 return False
         return True
 
 class BaseDB(object):
     """Empty Parent class so traitlets work on DB."""
     pass
 
 class DictDB(BaseDB):
     """Basic in-memory dict-based object for saving Task Records.
     
     This is the first object to present the DB interface
     for logging tasks out of memory.
     
     The interface is based on MongoDB, so adding a MongoDB
     backend should be straightforward.
     """
     _records = None
     
     def __init__(self, *args, **kwargs):
         self._records = dict()
     
     def _match_one(self, rec, tests):
         """Check if a specific record matches tests."""
         for key,test in tests.iteritems():
             if not test(rec.get(key, None)):
                 return False
         return True
         
     def _match(self, check, id_only=True):
         """Find all the matches for a check dict."""
         matches = {}
         tests = {}
         for k,v in check.iteritems():
             if isinstance(v, dict):
                 tests[k] = CompositeFilter(v)
             else:
                 tests[k] = lambda o: o==v
         
         for msg_id, rec in self._records.iteritems():
             if self._match_one(rec, tests):
                 matches[msg_id] = rec
         if id_only:
             return matches.keys()
         else:
             return matches
             
     
     def add_record(self, msg_id, rec):
         """Add a new Task Record, by msg_id."""
         if self._records.has_key(msg_id):
             raise KeyError("Already have msg_id %r"%(msg_id))
         self._records[msg_id] = rec
     
     def get_record(self, msg_id):
         """Get a specific Task Record, by msg_id."""
         if not self._records.has_key(msg_id):
             raise KeyError("No such msg_id %r"%(msg_id))
         return self._records[msg_id]
     
     def update_record(self, msg_id, rec):
         """Update the data in an existing record."""
         self._records[msg_id].update(rec)
     
     def drop_matching_records(self, check):
         """Remove a record from the DB."""
         matches = self._match(check, id_only=True)
         for m in matches:
             del self._records[m]
         
     def drop_record(self, msg_id):
         """Remove a record from the DB."""
         del self._records[msg_id]
         
     
     def find_records(self, check, id_only=False):
         """Find records matching a query dict."""
         matches = self._match(check, id_only)
         return matches
  No newline at end of file

IPython/zmq/parallel/engine.py

@@ -1,147 +1,139 b''
 #!/usr/bin/env python
 """A simple engine that talks to a controller over 0MQ.
 it handles registration, etc. and launches a kernel
-connected to the Controller's queue(s).
+connected to the Controller's Schedulers.
 """
 from __future__ import print_function
 
-import logging
 import sys
 import time
-import uuid
-from pprint import pprint
 
 import zmq
 from zmq.eventloop import ioloop, zmqstream
 
 # internal
-from IPython.config.configurable import Configurable
 from IPython.utils.traitlets import Instance, Str, Dict, Int, Type, CFloat
 # from IPython.utils.localinterfaces import LOCALHOST 
 
 from . import heartmonitor
 from .factory import RegistrationFactory
 from .streamkernel import Kernel
 from .streamsession import Message
 from .util import disambiguate_url
 
-def printer(*msg):
-    # print (self.log.handlers, file=sys.__stdout__)
-    self.log.info(str(msg))
-
 class EngineFactory(RegistrationFactory):
     """IPython engine"""
     
     # configurables:
     user_ns=Dict(config=True)
     out_stream_factory=Type('IPython.zmq.iostream.OutStream', config=True)
     display_hook_factory=Type('IPython.zmq.displayhook.DisplayHook', config=True)
     location=Str(config=True)
     timeout=CFloat(2,config=True)
     
     # not configurable:
     id=Int(allow_none=True)
     registrar=Instance('zmq.eventloop.zmqstream.ZMQStream')
     kernel=Instance(Kernel)
     
     
     def __init__(self, **kwargs):
         super(EngineFactory, self).__init__(**kwargs)
         ctx = self.context
         
         reg = ctx.socket(zmq.PAIR)
         reg.setsockopt(zmq.IDENTITY, self.ident)
         reg.connect(self.url)
         self.registrar = zmqstream.ZMQStream(reg, self.loop)
         
     def register(self):
         """send the registration_request"""
         
         self.log.info("registering")
         content = dict(queue=self.ident, heartbeat=self.ident, control=self.ident)
         self.registrar.on_recv(self.complete_registration)
         # print (self.session.key)
         self.session.send(self.registrar, "registration_request",content=content)
     
     def complete_registration(self, msg):
         # print msg
         self._abort_dc.stop()
         ctx = self.context
         loop = self.loop
         identity = self.ident
         print (identity)
         
         idents,msg = self.session.feed_identities(msg)
         msg = Message(self.session.unpack_message(msg))
         
         if msg.content.status == 'ok':
             self.id = int(msg.content.id)
             
             # create Shell Streams (MUX, Task, etc.):
             queue_addr = msg.content.mux
             shell_addrs = [ str(queue_addr) ]
             task_addr = msg.content.task
             if task_addr:
                 shell_addrs.append(str(task_addr))
             shell_streams = []
             for addr in shell_addrs:
                 stream = zmqstream.ZMQStream(ctx.socket(zmq.PAIR), loop)
                 stream.setsockopt(zmq.IDENTITY, identity)
                 stream.connect(disambiguate_url(addr, self.location))
                 shell_streams.append(stream)
             
             # control stream:
             control_addr = str(msg.content.control)
             control_stream = zmqstream.ZMQStream(ctx.socket(zmq.PAIR), loop)
             control_stream.setsockopt(zmq.IDENTITY, identity)
             control_stream.connect(disambiguate_url(control_addr, self.location))
             
             # create iopub stream:
             iopub_addr = msg.content.iopub
             iopub_stream = zmqstream.ZMQStream(ctx.socket(zmq.PUB), loop)
             iopub_stream.setsockopt(zmq.IDENTITY, identity)
             iopub_stream.connect(disambiguate_url(iopub_addr, self.location))
             
             # launch heartbeat
             hb_addrs = msg.content.heartbeat
             # print (hb_addrs)
             
             # # Redirect input streams and set a display hook.
             if self.out_stream_factory:
                 sys.stdout = self.out_stream_factory(self.session, iopub_stream, u'stdout')
                 sys.stdout.topic = 'engine.%i.stdout'%self.id
                 sys.stderr = self.out_stream_factory(self.session, iopub_stream, u'stderr')
                 sys.stderr.topic = 'engine.%i.stderr'%self.id
             if self.display_hook_factory:
                 sys.displayhook = self.display_hook_factory(self.session, iopub_stream)
                 sys.displayhook.topic = 'engine.%i.pyout'%self.id
             
             self.kernel = Kernel(config=self.config, int_id=self.id, ident=self.ident, session=self.session, 
                     control_stream=control_stream, shell_streams=shell_streams, iopub_stream=iopub_stream, 
                     loop=loop, user_ns = self.user_ns, logname=self.log.name)
             self.kernel.start()
             hb_addrs = [ disambiguate_url(addr, self.location) for addr in hb_addrs ]
             heart = heartmonitor.Heart(*map(str, hb_addrs), heart_id=identity)
             # ioloop.DelayedCallback(heart.start, 1000, self.loop).start()
             heart.start()
             
             
         else:
             self.log.fatal("Registration Failed: %s"%msg)
             raise Exception("Registration Failed: %s"%msg)
         
         self.log.info("Completed registration with id %i"%self.id)
     
     
     def abort(self):
         self.log.fatal("Registration timed out")
         self.session.send(self.registrar, "unregistration_request", content=dict(id=self.id))
         time.sleep(1)
         sys.exit(255)
     
     def start(self):
         dc = ioloop.DelayedCallback(self.register, 0, self.loop)
         dc.start()
         self._abort_dc = ioloop.DelayedCallback(self.abort, self.timeout*1000, self.loop)
         self._abort_dc.start()
 

IPython/zmq/parallel/error.py

@@ -1,292 +1,313 b''
 # encoding: utf-8
 
 """Classes and functions for kernel related errors and exceptions."""
 from __future__ import print_function
 
+import sys
+import traceback
+
 __docformat__ = "restructuredtext en"
 
 # Tell nose to skip this module
 __test__ = {}
 
 #-------------------------------------------------------------------------------
 #  Copyright (C) 2008  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.
 #-------------------------------------------------------------------------------
 
 #-------------------------------------------------------------------------------
 # Error classes
 #-------------------------------------------------------------------------------
 class IPythonError(Exception):
     """Base exception that all of our exceptions inherit from.
 
     This can be raised by code that doesn't have any more specific
     information."""
 
     pass
 
 # Exceptions associated with the controller objects
 class ControllerError(IPythonError): pass
 
 class ControllerCreationError(ControllerError): pass
 
 
 # Exceptions associated with the Engines
 class EngineError(IPythonError): pass
 
 class EngineCreationError(EngineError): pass
 
 class KernelError(IPythonError):
     pass
 
 class NotDefined(KernelError):
     def __init__(self, name):
         self.name = name
         self.args = (name,)
 
     def __repr__(self):
         return '<NotDefined: %s>' % self.name
     
     __str__ = __repr__
 
 
 class QueueCleared(KernelError):
     pass
 
 
 class IdInUse(KernelError):
     pass
 
 
 class ProtocolError(KernelError):
     pass
 
 
 class ConnectionError(KernelError):
     pass
 
 
 class InvalidEngineID(KernelError):
     pass
 
 
 class NoEnginesRegistered(KernelError):
     pass
 
 
 class InvalidClientID(KernelError):
     pass
 
 
 class InvalidDeferredID(KernelError):
     pass
 
 
 class SerializationError(KernelError):
     pass
 
 
 class MessageSizeError(KernelError):
     pass
 
 
 class PBMessageSizeError(MessageSizeError):
     pass
 
 
 class ResultNotCompleted(KernelError):
     pass
 
 
 class ResultAlreadyRetrieved(KernelError):
     pass
 
 class ClientError(KernelError):
     pass
 
 
 class TaskAborted(KernelError):
     pass
 
 
 class TaskTimeout(KernelError):
     pass
 
 
 class NotAPendingResult(KernelError):
     pass
 
 
 class UnpickleableException(KernelError):
     pass
 
 
 class AbortedPendingDeferredError(KernelError):
     pass
 
 
 class InvalidProperty(KernelError):
     pass
 
 
 class MissingBlockArgument(KernelError):
     pass
 
 
 class StopLocalExecution(KernelError):
     pass
 
 
 class SecurityError(KernelError):
     pass
 
 
 class FileTimeoutError(KernelError):
     pass
 
 class TimeoutError(KernelError):
     pass
 
 class UnmetDependency(KernelError):
     pass
 
 class ImpossibleDependency(UnmetDependency):
     pass
 
 class DependencyTimeout(ImpossibleDependency):
     pass
 
 class InvalidDependency(ImpossibleDependency):
     pass
 
 class RemoteError(KernelError):
     """Error raised elsewhere"""
     ename=None
     evalue=None
     traceback=None
     engine_info=None
     
     def __init__(self, ename, evalue, traceback, engine_info=None):
         self.ename=ename
         self.evalue=evalue
         self.traceback=traceback
         self.engine_info=engine_info or {}
         self.args=(ename, evalue)
     
     def __repr__(self):
         engineid = self.engine_info.get('engine_id', ' ')
         return "<Remote[%s]:%s(%s)>"%(engineid, self.ename, self.evalue)
         
     def __str__(self):
         sig = "%s(%s)"%(self.ename, self.evalue)
         if self.traceback:
             return sig + '\n' + self.traceback
         else:
             return sig
     
 
 class TaskRejectError(KernelError):
     """Exception to raise when a task should be rejected by an engine.
     
     This exception can be used to allow a task running on an engine to test
     if the engine (or the user's namespace on the engine) has the needed
     task dependencies.  If not, the task should raise this exception.  For
     the task to be retried on another engine, the task should be created
     with the `retries` argument > 1.
     
     The advantage of this approach over our older properties system is that
     tasks have full access to the user's namespace on the engines and the
     properties don't have to be managed or tested by the controller.
     """
 
 
 class CompositeError(RemoteError):
     """Error for representing possibly multiple errors on engines"""
     def __init__(self, message, elist):
         Exception.__init__(self, *(message, elist))
         # Don't use pack_exception because it will conflict with the .message
         # attribute that is being deprecated in 2.6 and beyond.
         self.msg = message
         self.elist = elist
         self.args = [ e[0] for e in elist ]
 
     def _get_engine_str(self, ei):
         if not ei:
             return '[Engine Exception]'
         else:
             return '[%s:%s]: ' % (ei['engine_id'], ei['method'])
 
     def _get_traceback(self, ev):
         try:
             tb = ev._ipython_traceback_text
         except AttributeError:
             return 'No traceback available'
         else:
             return tb
 
     def __str__(self):
         s = str(self.msg)
         for en, ev, etb, ei in self.elist:
             engine_str = self._get_engine_str(ei)
             s = s + '\n' + engine_str + en + ': ' + str(ev)
         return s
     
     def __repr__(self):
         return "CompositeError(%i)"%len(self.elist)
 
     def print_tracebacks(self, excid=None):
         if excid is None:
             for (en,ev,etb,ei) in self.elist:
                 print (self._get_engine_str(ei))
                 print (etb or 'No traceback available')
                 print ()
         else:
             try:
                 en,ev,etb,ei = self.elist[excid]
             except:
                 raise IndexError("an exception with index %i does not exist"%excid)
             else:
                 print (self._get_engine_str(ei))
                 print (etb or 'No traceback available')
     
     def raise_exception(self, excid=0):
         try:
             en,ev,etb,ei = self.elist[excid]
         except:
             raise IndexError("an exception with index %i does not exist"%excid)
         else:
             raise RemoteError(en, ev, etb, ei)
 
 
 def collect_exceptions(rdict_or_list, method='unspecified'):
     """check a result dict for errors, and raise CompositeError if any exist.
     Passthrough otherwise."""
     elist = []
     if isinstance(rdict_or_list, dict):
         rlist = rdict_or_list.values()
     else:
         rlist = rdict_or_list
     for r in rlist:
         if isinstance(r, RemoteError):
             en, ev, etb, ei = r.ename, r.evalue, r.traceback, r.engine_info
             # Sometimes we could have CompositeError in our list.  Just take
             # the errors out of them and put them in our new list.  This 
             # has the effect of flattening lists of CompositeErrors into one
             # CompositeError
             if en=='CompositeError':
                 for e in ev.elist:
                     elist.append(e)
             else:
                 elist.append((en, ev, etb, ei))
     if len(elist)==0:
         return rdict_or_list
     else:
         msg = "one or more exceptions from call to method: %s" % (method)
         # This silliness is needed so the debugger has access to the exception
         # instance (e in this case)
         try:
             raise CompositeError(msg, elist)
         except CompositeError as e:
             raise e
 
+def wrap_exception(engine_info={}):
+    etype, evalue, tb = sys.exc_info()
+    stb = traceback.format_exception(etype, evalue, tb)
+    exc_content = {
+        'status' : 'error',
+        'traceback' : stb,
+        'ename' : unicode(etype.__name__),
+        'evalue' : unicode(evalue),
+        'engine_info' : engine_info
+    }
+    return exc_content
+
+def unwrap_exception(content):
+    err = RemoteError(content['ename'], content['evalue'], 
+                ''.join(content['traceback']),
+                content.get('engine_info', {}))
+    return err
+

IPython/zmq/parallel/factory.py

@@ -1,152 +1,152 b''
 """Base config factories."""
 
 #-----------------------------------------------------------------------------
 #  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 logging
 import os
 import uuid
 
 from zmq.eventloop.ioloop import IOLoop
 
 from IPython.config.configurable import Configurable
 from IPython.utils.importstring import import_item
 from IPython.utils.traitlets import Str,Int,Instance, CUnicode, CStr
 
 import IPython.zmq.parallel.streamsession as ss
 from IPython.zmq.parallel.entry_point import select_random_ports
 
 #-----------------------------------------------------------------------------
 # Classes
 #-----------------------------------------------------------------------------
 class LoggingFactory(Configurable):
     """A most basic class, that has a `log` (type:`Logger`) attribute, set via a `logname` Trait."""
     log = Instance('logging.Logger', ('ZMQ', logging.WARN))
-    logname = CStr('ZMQ')
+    logname = CUnicode('ZMQ')
     def _logname_changed(self, name, old, new):
         self.log = logging.getLogger(new)
     
 
 class SessionFactory(LoggingFactory):
     """The Base factory from which every factory in IPython.zmq.parallel inherits"""
     
     packer = Str('',config=True)
     unpacker = Str('',config=True)
     ident = CStr('',config=True)
     def _ident_default(self):
         return str(uuid.uuid4())
-    username = Str(os.environ.get('USER','username'),config=True)
-    exec_key = CStr('',config=True)
+    username = CUnicode(os.environ.get('USER','username'),config=True)
+    exec_key = CUnicode('',config=True)
     # not configurable:
     context = Instance('zmq.Context', (), {})
     session = Instance('IPython.zmq.parallel.streamsession.StreamSession')
     loop = Instance('zmq.eventloop.ioloop.IOLoop', allow_none=False)
     def _loop_default(self):
         return IOLoop.instance()
     
     
     def __init__(self, **kwargs):
         super(SessionFactory, self).__init__(**kwargs)
         exec_key = self.exec_key or None
         # set the packers:
         if not self.packer:
             packer_f = unpacker_f = None
         elif self.packer.lower() == 'json':
             packer_f = ss.json_packer
             unpacker_f = ss.json_unpacker
         elif self.packer.lower() == 'pickle':
             packer_f = ss.pickle_packer
             unpacker_f = ss.pickle_unpacker
         else:
             packer_f = import_item(self.packer)
             unpacker_f = import_item(self.unpacker)
         
         # construct the session
         self.session = ss.StreamSession(self.username, self.ident, packer=packer_f, unpacker=unpacker_f, key=exec_key)
     
 
 class RegistrationFactory(SessionFactory):
     """The Base Configurable for objects that involve registration."""
     
     url = Str('', config=True) # url takes precedence over ip,regport,transport
     transport = Str('tcp', config=True)
     ip = Str('127.0.0.1', config=True)
     regport = Instance(int, config=True)
     def _regport_default(self):
         # return 10101
         return select_random_ports(1)[0]
     
     def __init__(self, **kwargs):
         super(RegistrationFactory, self).__init__(**kwargs)
         self._propagate_url()
         self._rebuild_url()
         self.on_trait_change(self._propagate_url, 'url')
         self.on_trait_change(self._rebuild_url, 'ip')
         self.on_trait_change(self._rebuild_url, 'transport')
         self.on_trait_change(self._rebuild_url, 'regport')
     
     def _rebuild_url(self):
         self.url = "%s://%s:%i"%(self.transport, self.ip, self.regport)
         
     def _propagate_url(self):
         """Ensure self.url contains full transport://interface:port"""
         if self.url:
             iface = self.url.split('://',1)
             if len(iface) == 2:
                 self.transport,iface = iface
             iface = iface.split(':')
             self.ip = iface[0]
             if iface[1]:
                 self.regport = int(iface[1])
 
 #-----------------------------------------------------------------------------
 # argparse argument extenders
 #-----------------------------------------------------------------------------
 
 
 def add_session_arguments(parser):
     paa = parser.add_argument
     paa('--ident',
         type=str, dest='SessionFactory.ident', 
         help='set the ZMQ and session identity [default: random uuid]',
         metavar='identity')
     # paa('--execkey',
     #     type=str, dest='SessionFactory.exec_key', 
     #     help='path to a file containing an execution key.',
     #     metavar='execkey')
     paa('--packer',
         type=str, dest='SessionFactory.packer', 
         help='method to serialize messages: {json,pickle} [default: json]',
         metavar='packer')
     paa('--unpacker',
         type=str, dest='SessionFactory.unpacker', 
         help='inverse function of `packer`.  Only necessary when using something other than json|pickle',
         metavar='packer')
 
 def add_registration_arguments(parser):
     paa = parser.add_argument
     paa('--ip',
         type=str, dest='RegistrationFactory.ip',
         help="The IP used for registration [default: localhost]",
         metavar='ip')
     paa('--transport',
         type=str, dest='RegistrationFactory.transport',
         help="The ZeroMQ transport used for registration [default: tcp]",
         metavar='transport')
     paa('--url',
         type=str, dest='RegistrationFactory.url', 
         help='set transport,ip,regport in one go, e.g. tcp://127.0.0.1:10101',
         metavar='url')
     paa('--regport',
         type=int, dest='RegistrationFactory.regport',
         help="The port used for registration [default: 10101]",
         metavar='ip')

IPython/zmq/parallel/hub.py

@@ -1,1054 +1,1052 b''
 #!/usr/bin/env python
 """The IPython Controller Hub with 0MQ
 This is the master object that handles connections from engines and clients,
 and monitors traffic through the various queues.
 """
 #-----------------------------------------------------------------------------
 #  Copyright (C) 2010  The IPython Development Team
 #
 #  Distributed under the terms of the BSD License.  The full license is in
 #  the file COPYING, distributed as part of this software.
 #-----------------------------------------------------------------------------
 
 #-----------------------------------------------------------------------------
 # Imports
 #-----------------------------------------------------------------------------
 from __future__ import print_function
 
-import logging
 import sys
 import time
 from datetime import datetime
 
 import zmq
 from zmq.eventloop import ioloop
 from zmq.eventloop.zmqstream import ZMQStream
 
 # internal:
-from IPython.config.configurable import Configurable
 from IPython.utils.importstring import import_item
 from IPython.utils.traitlets import HasTraits, Instance, Int, CStr, Str, Dict, Set, List, Bool
 
 from .entry_point import select_random_ports
 from .factory import RegistrationFactory, LoggingFactory
 
+from . import error
 from .heartmonitor import HeartMonitor
-from .streamsession import Message, wrap_exception, ISO8601
-from .util import validate_url_container
+from .util import validate_url_container, ISO8601
 
 try:
     from pymongo.binary import Binary
 except ImportError:
     MongoDB=None
 else:
     from mongodb import MongoDB
     
 #-----------------------------------------------------------------------------
 # Code
 #-----------------------------------------------------------------------------
 
 def _passer(*args, **kwargs):
     return
 
 def _printer(*args, **kwargs):
     print (args)
     print (kwargs)
 
 def init_record(msg):
     """Initialize a TaskRecord based on a request."""
     header = msg['header']
     return {
         'msg_id' : header['msg_id'],
         'header' : header,
         'content': msg['content'],
         'buffers': msg['buffers'],
         'submitted': datetime.strptime(header['date'], ISO8601),
         'client_uuid' : None,
         'engine_uuid' : None,
         'started': None,
         'completed': None,
         'resubmitted': None,
         'result_header' : None,
         'result_content' : None,
         'result_buffers' : None,
         'queue' : None,
         'pyin' : None,
         'pyout': None,
         'pyerr': None,
         'stdout': '',
         'stderr': '',
     }
 
 
 class EngineConnector(HasTraits):
     """A simple object for accessing the various zmq connections of an object.
     Attributes are:
     id (int): engine ID
     uuid (str): uuid (unused?)
     queue (str): identity of queue's XREQ socket
     registration (str): identity of registration XREQ socket
     heartbeat (str): identity of heartbeat XREQ socket
     """
     id=Int(0)
     queue=Str()
     control=Str()
     registration=Str()
     heartbeat=Str()
     pending=Set()
 
 class HubFactory(RegistrationFactory):
     """The Configurable for setting up a Hub."""
     
     # name of a scheduler scheme
     scheme = Str('leastload', config=True)
     
     # port-pairs for monitoredqueues:
     hb = Instance(list, config=True)
     def _hb_default(self):
         return select_random_ports(2)
     
     mux = Instance(list, config=True)
     def _mux_default(self):
         return select_random_ports(2)
     
     task = Instance(list, config=True)
     def _task_default(self):
         return select_random_ports(2)
     
     control = Instance(list, config=True)
     def _control_default(self):
         return select_random_ports(2)
     
     iopub = Instance(list, config=True)
     def _iopub_default(self):
         return select_random_ports(2)
     
     # single ports:
     mon_port = Instance(int, config=True)
     def _mon_port_default(self):
         return select_random_ports(1)[0]
     
     query_port = Instance(int, config=True)
     def _query_port_default(self):
         return select_random_ports(1)[0]
     
     notifier_port = Instance(int, config=True)
     def _notifier_port_default(self):
         return select_random_ports(1)[0]
     
     ping = Int(1000, config=True) # ping frequency
     
     engine_ip = CStr('127.0.0.1', config=True)
     engine_transport = CStr('tcp', config=True)
     
     client_ip = CStr('127.0.0.1', config=True)
     client_transport = CStr('tcp', config=True)
     
     monitor_ip = CStr('127.0.0.1', config=True)
     monitor_transport = CStr('tcp', config=True)
     
     monitor_url = CStr('')
     
     db_class = CStr('IPython.zmq.parallel.dictdb.DictDB', config=True)
     
     # not configurable
     db = Instance('IPython.zmq.parallel.dictdb.BaseDB')
     heartmonitor = Instance('IPython.zmq.parallel.heartmonitor.HeartMonitor')
     subconstructors = List()
     _constructed = Bool(False)
     
     def _ip_changed(self, name, old, new):
         self.engine_ip = new
         self.client_ip = new
         self.monitor_ip = new
         self._update_monitor_url()
     
     def _update_monitor_url(self):
         self.monitor_url = "%s://%s:%i"%(self.monitor_transport, self.monitor_ip, self.mon_port)
     
     def _transport_changed(self, name, old, new):
         self.engine_transport = new
         self.client_transport = new
         self.monitor_transport = new
         self._update_monitor_url()
         
     def __init__(self, **kwargs):
         super(HubFactory, self).__init__(**kwargs)
         self._update_monitor_url()
         # self.on_trait_change(self._sync_ips, 'ip')
         # self.on_trait_change(self._sync_transports, 'transport')
         self.subconstructors.append(self.construct_hub)
     
     
     def construct(self):
         assert not self._constructed, "already constructed!"
         
         for subc in self.subconstructors:
             subc()
         
         self._constructed = True
         
     
     def start(self):
         assert self._constructed, "must be constructed by self.construct() first!"
         self.heartmonitor.start()
         self.log.info("Heartmonitor started")
     
     def construct_hub(self):
         """construct"""
         client_iface = "%s://%s:"%(self.client_transport, self.client_ip) + "%i"
         engine_iface = "%s://%s:"%(self.engine_transport, self.engine_ip) + "%i"
         
         ctx = self.context
         loop = self.loop
         
         # Registrar socket
         reg = ZMQStream(ctx.socket(zmq.XREP), loop)
         reg.bind(client_iface % self.regport)
         self.log.info("Hub listening on %s for registration."%(client_iface%self.regport))
         if self.client_ip != self.engine_ip:
             reg.bind(engine_iface % self.regport)
             self.log.info("Hub listening on %s for registration."%(engine_iface%self.regport))
         
         ### Engine connections ###
 
         # heartbeat
         hpub = ctx.socket(zmq.PUB)
         hpub.bind(engine_iface % self.hb[0])
         hrep = ctx.socket(zmq.XREP)
         hrep.bind(engine_iface % self.hb[1])
         self.heartmonitor = HeartMonitor(loop=loop, pingstream=ZMQStream(hpub,loop), pongstream=ZMQStream(hrep,loop), 
                                 period=self.ping, logname=self.log.name)
 
         ### Client connections ###
         # Clientele socket
         c = ZMQStream(ctx.socket(zmq.XREP), loop)
         c.bind(client_iface%self.query_port)
         # Notifier socket
         n = ZMQStream(ctx.socket(zmq.PUB), loop)
         n.bind(client_iface%self.notifier_port)
 
         ### build and launch the queues ###
 
         # monitor socket
         sub = ctx.socket(zmq.SUB)
         sub.setsockopt(zmq.SUBSCRIBE, "")
         sub.bind(self.monitor_url)
         sub.bind('inproc://monitor')
         sub = ZMQStream(sub, loop)
         
         # connect the db
         self.db = import_item(self.db_class)(self.session.session)
         time.sleep(.25)
 
         # build connection dicts
         self.engine_info = {
             'control' : engine_iface%self.control[1],
             'mux': engine_iface%self.mux[1],
             'heartbeat': (engine_iface%self.hb[0], engine_iface%self.hb[1]),
             'task' : engine_iface%self.task[1],
             'iopub' : engine_iface%self.iopub[1],
             # 'monitor' : engine_iface%self.mon_port,
             }
 
         self.client_info = {
             'control' : client_iface%self.control[0],
             'query': client_iface%self.query_port,
             'mux': client_iface%self.mux[0],
             'task' : (self.scheme, client_iface%self.task[0]),
             'iopub' : client_iface%self.iopub[0],
             'notification': client_iface%self.notifier_port
             }
         self.log.debug("hub::Hub engine addrs: %s"%self.engine_info)
         self.log.debug("hub::Hub client addrs: %s"%self.client_info)
         self.hub = Hub(loop=loop, session=self.session, monitor=sub, heartmonitor=self.heartmonitor,
                 registrar=reg, clientele=c, notifier=n, db=self.db,
                 engine_info=self.engine_info, client_info=self.client_info,
                 logname=self.log.name)
     
 
 class Hub(LoggingFactory):
     """The IPython Controller Hub with 0MQ connections
     
     Parameters
     ==========
     loop: zmq IOLoop instance
     session: StreamSession object
     <removed> context: zmq context for creating new connections (?)
     queue: ZMQStream for monitoring the command queue (SUB)
     registrar: ZMQStream for engine registration requests (XREP)
     heartbeat: HeartMonitor object checking the pulse of the engines
     clientele: ZMQStream for client connections (XREP)
                 not used for jobs, only query/control commands
     notifier: ZMQStream for broadcasting engine registration changes (PUB)
     db: connection to db for out of memory logging of commands
                 NotImplemented
     engine_info: dict of zmq connection information for engines to connect
                 to the queues.
     client_info: dict of zmq connection information for engines to connect
                 to the queues.
     """
     # internal data structures:
     ids=Set() # engine IDs
     keytable=Dict()
     by_ident=Dict()
     engines=Dict()
     clients=Dict()
     hearts=Dict()
     pending=Set()
     queues=Dict()  # pending msg_ids keyed by engine_id
     tasks=Dict() # pending msg_ids submitted as tasks, keyed by client_id
     completed=Dict() # completed msg_ids keyed by engine_id
     all_completed=Set() # completed msg_ids keyed by engine_id
     # mia=None
     incoming_registrations=Dict()
     registration_timeout=Int()
     _idcounter=Int(0)
     
     # objects from constructor:
     loop=Instance(ioloop.IOLoop)
     registrar=Instance(ZMQStream)
     clientele=Instance(ZMQStream)
     monitor=Instance(ZMQStream)
     heartmonitor=Instance(HeartMonitor)
     notifier=Instance(ZMQStream)
     db=Instance(object)
     client_info=Dict()
     engine_info=Dict()
     
     
     def __init__(self, **kwargs):
         """
         # universal:
         loop: IOLoop for creating future connections
         session: streamsession for sending serialized data
         # engine:
         queue: ZMQStream for monitoring queue messages
         registrar: ZMQStream for engine registration
         heartbeat: HeartMonitor object for tracking engines
         # client:
         clientele: ZMQStream for client connections
         # extra:
         db: ZMQStream for db connection (NotImplemented)
         engine_info: zmq address/protocol dict for engine connections
         client_info: zmq address/protocol dict for client connections
         """
         
         super(Hub, self).__init__(**kwargs)
         self.registration_timeout = max(5000, 2*self.heartmonitor.period)
         
         # validate connection dicts:
         for k,v in self.client_info.iteritems():
             if k == 'task':
                 validate_url_container(v[1])
             else:
                 validate_url_container(v)
         # validate_url_container(self.client_info)
         validate_url_container(self.engine_info)
         
         # register our callbacks
         self.registrar.on_recv(self.dispatch_register_request)
         self.clientele.on_recv(self.dispatch_client_msg)
         self.monitor.on_recv(self.dispatch_monitor_traffic)
         
         self.heartmonitor.add_heart_failure_handler(self.handle_heart_failure)
         self.heartmonitor.add_new_heart_handler(self.handle_new_heart)
         
         self.monitor_handlers = { 'in' : self.save_queue_request,
                                 'out': self.save_queue_result,
                                 'intask': self.save_task_request,
                                 'outtask': self.save_task_result,
                                 'tracktask': self.save_task_destination,
                                 'incontrol': _passer,
                                 'outcontrol': _passer,
                                 'iopub': self.save_iopub_message,
         }
         
         self.client_handlers = {'queue_request': self.queue_status,
                                 'result_request': self.get_results,
                                 'purge_request': self.purge_results,
                                 'load_request': self.check_load,
                                 'resubmit_request': self.resubmit_task,
                                 'shutdown_request': self.shutdown_request,
                                 }
         
         self.registrar_handlers = {'registration_request' : self.register_engine,
                                 'unregistration_request' : self.unregister_engine,
                                 'connection_request': self.connection_request,
         }
         
         self.log.info("hub::created hub")
     
     @property
     def _next_id(self):
         """gemerate a new ID.
         
         No longer reuse old ids, just count from 0."""
         newid = self._idcounter
         self._idcounter += 1
         return newid
         # newid = 0
         # incoming = [id[0] for id in self.incoming_registrations.itervalues()]
         # # print newid, self.ids, self.incoming_registrations
         # while newid in self.ids or newid in incoming:
         #     newid += 1
         # return newid
     
     #-----------------------------------------------------------------------------
     # message validation
     #-----------------------------------------------------------------------------
     
     def _validate_targets(self, targets):
         """turn any valid targets argument into a list of integer ids"""
         if targets is None:
             # default to all
             targets = self.ids
             
         if isinstance(targets, (int,str,unicode)):
             # only one target specified
             targets = [targets]
         _targets = []
         for t in targets:
             # map raw identities to ids
             if isinstance(t, (str,unicode)):
                 t = self.by_ident.get(t, t)
             _targets.append(t)
         targets = _targets
         bad_targets = [ t for t in targets if t not in self.ids ]
         if bad_targets:
             raise IndexError("No Such Engine: %r"%bad_targets)
         if not targets:
             raise IndexError("No Engines Registered")
         return targets
     
     def _validate_client_msg(self, msg):
         """validates and unpacks headers of a message. Returns False if invalid,
         (ident, header, parent, content)"""
         client_id = msg[0]
         try:
             msg = self.session.unpack_message(msg[1:], content=True)
         except:
             self.log.error("client::Invalid Message %s"%msg, exc_info=True)
             return False
         
         msg_type = msg.get('msg_type', None)
         if msg_type is None:
             return False
         header = msg.get('header')
         # session doesn't handle split content for now:
         return client_id, msg
         
     
     #-----------------------------------------------------------------------------
     # dispatch methods (1 per stream)
     #-----------------------------------------------------------------------------
     
     def dispatch_register_request(self, msg):
         """"""
         self.log.debug("registration::dispatch_register_request(%s)"%msg)
         idents,msg = self.session.feed_identities(msg)
         if not idents:
             self.log.error("Bad Queue Message: %s"%msg, exc_info=True)
             return
         try:
             msg = self.session.unpack_message(msg,content=True)
         except:
             self.log.error("registration::got bad registration message: %s"%msg, exc_info=True)
             return
         
         msg_type = msg['msg_type']
         content = msg['content']
         
         handler = self.registrar_handlers.get(msg_type, None)
         if handler is None:
             self.log.error("registration::got bad registration message: %s"%msg)
         else:
             handler(idents, msg)
     
     def dispatch_monitor_traffic(self, msg):
         """all ME and Task queue messages come through here, as well as
         IOPub traffic."""
         self.log.debug("monitor traffic: %s"%msg[:2])
         switch = msg[0]
         idents, msg = self.session.feed_identities(msg[1:])
         if not idents:
             self.log.error("Bad Monitor Message: %s"%msg)
             return
         handler = self.monitor_handlers.get(switch, None)
         if handler is not None:
             handler(idents, msg)
         else:
             self.log.error("Invalid monitor topic: %s"%switch)
         
     
     def dispatch_client_msg(self, msg):
         """Route messages from clients"""
         idents, msg = self.session.feed_identities(msg)
         if not idents:
             self.log.error("Bad Client Message: %s"%msg)
             return
         client_id = idents[0]
         try:
             msg = self.session.unpack_message(msg, content=True)
         except:
-            content = wrap_exception()
+            content = error.wrap_exception()
             self.log.error("Bad Client Message: %s"%msg, exc_info=True)
             self.session.send(self.clientele, "hub_error", ident=client_id, 
                     content=content)
             return
         
         # print client_id, header, parent, content
         #switch on message type:
         msg_type = msg['msg_type']
         self.log.info("client:: client %s requested %s"%(client_id, msg_type))
         handler = self.client_handlers.get(msg_type, None)
         try:
             assert handler is not None, "Bad Message Type: %s"%msg_type
         except:
-            content = wrap_exception()
+            content = error.wrap_exception()
             self.log.error("Bad Message Type: %s"%msg_type, exc_info=True)
             self.session.send(self.clientele, "hub_error", ident=client_id, 
                     content=content)
             return
         else:
             handler(client_id, msg)
             
     def dispatch_db(self, msg):
         """"""
         raise NotImplementedError
     
     #---------------------------------------------------------------------------
     # handler methods (1 per event)
     #---------------------------------------------------------------------------
     
     #----------------------- Heartbeat --------------------------------------
     
     def handle_new_heart(self, heart):
         """handler to attach to heartbeater.
         Called when a new heart starts to beat.
         Triggers completion of registration."""
         self.log.debug("heartbeat::handle_new_heart(%r)"%heart)
         if heart not in self.incoming_registrations:
             self.log.info("heartbeat::ignoring new heart: %r"%heart)
         else:
             self.finish_registration(heart)
         
     
     def handle_heart_failure(self, heart):
         """handler to attach to heartbeater.
         called when a previously registered heart fails to respond to beat request.
         triggers unregistration"""
         self.log.debug("heartbeat::handle_heart_failure(%r)"%heart)
         eid = self.hearts.get(heart, None)
         queue = self.engines[eid].queue
         if eid is None:
             self.log.info("heartbeat::ignoring heart failure %r"%heart)
         else:
             self.unregister_engine(heart, dict(content=dict(id=eid, queue=queue)))
     
     #----------------------- MUX Queue Traffic ------------------------------
     
     def save_queue_request(self, idents, msg):
         if len(idents) < 2:
             self.log.error("invalid identity prefix: %s"%idents)
             return
         queue_id, client_id = idents[:2]
         try:
             msg = self.session.unpack_message(msg, content=False)
         except:
             self.log.error("queue::client %r sent invalid message to %r: %s"%(client_id, queue_id, msg), exc_info=True)
             return
         
         eid = self.by_ident.get(queue_id, None)
         if eid is None:
             self.log.error("queue::target %r not registered"%queue_id)
             self.log.debug("queue::    valid are: %s"%(self.by_ident.keys()))
             return
             
         header = msg['header']
         msg_id = header['msg_id']
         record = init_record(msg)
         record['engine_uuid'] = queue_id
         record['client_uuid'] = client_id
         record['queue'] = 'mux'
         if MongoDB is not None and isinstance(self.db, MongoDB):
             record['buffers'] = map(Binary, record['buffers'])
         self.pending.add(msg_id)
         self.queues[eid].append(msg_id)
         self.db.add_record(msg_id, record)
     
     def save_queue_result(self, idents, msg):
         if len(idents) < 2:
             self.log.error("invalid identity prefix: %s"%idents)
             return
             
         client_id, queue_id = idents[:2]
         try:
             msg = self.session.unpack_message(msg, content=False)
         except:
             self.log.error("queue::engine %r sent invalid message to %r: %s"%(
                     queue_id,client_id, msg), exc_info=True)
             return
         
         eid = self.by_ident.get(queue_id, None)
         if eid is None:
             self.log.error("queue::unknown engine %r is sending a reply: "%queue_id)
             self.log.debug("queue::       %s"%msg[2:])
             return
         
         parent = msg['parent_header']
         if not parent:
             return
         msg_id = parent['msg_id']
         if msg_id in self.pending:
             self.pending.remove(msg_id)
             self.all_completed.add(msg_id)
             self.queues[eid].remove(msg_id)
             self.completed[eid].append(msg_id)
             rheader = msg['header']
             completed = datetime.strptime(rheader['date'], ISO8601)
             started = rheader.get('started', None)
             if started is not None:
                 started = datetime.strptime(started, ISO8601)
             result = {
                 'result_header' : rheader,
                 'result_content': msg['content'],
                 'started' : started,
                 'completed' : completed
             }
             if MongoDB is not None and isinstance(self.db, MongoDB):
                 result['result_buffers'] = map(Binary, msg['buffers'])
             else:
                 result['result_buffers'] = msg['buffers']
             self.db.update_record(msg_id, result)
         else:
             self.log.debug("queue:: unknown msg finished %s"%msg_id)
             
     #--------------------- Task Queue Traffic ------------------------------
     
     def save_task_request(self, idents, msg):
         """Save the submission of a task."""
         client_id = idents[0]
         
         try:
             msg = self.session.unpack_message(msg, content=False)
         except:
             self.log.error("task::client %r sent invalid task message: %s"%(
                     client_id, msg), exc_info=True)
             return
         record = init_record(msg)
         if MongoDB is not None and isinstance(self.db, MongoDB):
             record['buffers'] = map(Binary, record['buffers'])
         record['client_uuid'] = client_id
         record['queue'] = 'task'
         header = msg['header']
         msg_id = header['msg_id']
         self.pending.add(msg_id)
         self.db.add_record(msg_id, record)
     
     def save_task_result(self, idents, msg):
         """save the result of a completed task."""
         client_id = idents[0]
         try:
             msg = self.session.unpack_message(msg, content=False)
         except:
             self.log.error("task::invalid task result message send to %r: %s"%(
                     client_id, msg), exc_info=True)
             raise
             return
         
         parent = msg['parent_header']
         if not parent:
             # print msg
             self.log.warn("Task %r had no parent!"%msg)
             return
         msg_id = parent['msg_id']
         
         header = msg['header']
         engine_uuid = header.get('engine', None)
         eid = self.by_ident.get(engine_uuid, None)
         
         if msg_id in self.pending:
             self.pending.remove(msg_id)
             self.all_completed.add(msg_id)
             if eid is not None:
                 self.completed[eid].append(msg_id)
                 if msg_id in self.tasks[eid]:
                     self.tasks[eid].remove(msg_id)
             completed = datetime.strptime(header['date'], ISO8601)
             started = header.get('started', None)
             if started is not None:
                 started = datetime.strptime(started, ISO8601)
             result = {
                 'result_header' : header,
                 'result_content': msg['content'],
                 'started' : started,
                 'completed' : completed,
                 'engine_uuid': engine_uuid
             }
             if MongoDB is not None and isinstance(self.db, MongoDB):
                 result['result_buffers'] = map(Binary, msg['buffers'])
             else:
                 result['result_buffers'] = msg['buffers']
             self.db.update_record(msg_id, result)
             
         else:
             self.log.debug("task::unknown task %s finished"%msg_id)
     
     def save_task_destination(self, idents, msg):
         try:
             msg = self.session.unpack_message(msg, content=True)
         except:
             self.log.error("task::invalid task tracking message", exc_info=True)
             return
         content = msg['content']
         # print (content)
         msg_id = content['msg_id']
         engine_uuid = content['engine_id']
         eid = self.by_ident[engine_uuid]
         
         self.log.info("task::task %s arrived on %s"%(msg_id, eid))
         # if msg_id in self.mia:
         #     self.mia.remove(msg_id)
         # else:
         #     self.log.debug("task::task %s not listed as MIA?!"%(msg_id))
         
         self.tasks[eid].append(msg_id)
         # self.pending[msg_id][1].update(received=datetime.now(),engine=(eid,engine_uuid))
         self.db.update_record(msg_id, dict(engine_uuid=engine_uuid))
     
     def mia_task_request(self, idents, msg):
         raise NotImplementedError
         client_id = idents[0]
         # content = dict(mia=self.mia,status='ok')
         # self.session.send('mia_reply', content=content, idents=client_id)
         
         
     #--------------------- IOPub Traffic ------------------------------
     
     def save_iopub_message(self, topics, msg):
         """save an iopub message into the db"""
         # print (topics)
         try:
             msg = self.session.unpack_message(msg, content=True)
         except:
             self.log.error("iopub::invalid IOPub message", exc_info=True)
             return
         
         parent = msg['parent_header']
         if not parent:
             self.log.error("iopub::invalid IOPub message: %s"%msg)
             return
         msg_id = parent['msg_id']
         msg_type = msg['msg_type']
         content = msg['content']
         
         # ensure msg_id is in db
         try:
             rec = self.db.get_record(msg_id)
         except:
             self.log.error("iopub::IOPub message has invalid parent", exc_info=True)
             return
         # stream
         d = {}
         if msg_type == 'stream':
             name = content['name']
             s = rec[name] or ''
             d[name] = s + content['data']
             
         elif msg_type == 'pyerr':
             d['pyerr'] = content
         else:
             d[msg_type] = content['data']
         
         self.db.update_record(msg_id, d)
         
     
             
     #-------------------------------------------------------------------------
     # Registration requests
     #-------------------------------------------------------------------------
         
     def connection_request(self, client_id, msg):
         """Reply with connection addresses for clients."""
         self.log.info("client::client %s connected"%client_id)
         content = dict(status='ok')
         content.update(self.client_info)
         jsonable = {}
         for k,v in self.keytable.iteritems():
             jsonable[str(k)] = v
         content['engines'] = jsonable
         self.session.send(self.registrar, 'connection_reply', content, parent=msg, ident=client_id)
     
     def register_engine(self, reg, msg):
         """Register a new engine."""
         content = msg['content']
         try:
             queue = content['queue']
         except KeyError:
             self.log.error("registration::queue not specified", exc_info=True)
             return
         heart = content.get('heartbeat', None)
         """register a new engine, and create the socket(s) necessary"""
         eid = self._next_id
         # print (eid, queue, reg, heart)
         
         self.log.debug("registration::register_engine(%i, %r, %r, %r)"%(eid, queue, reg, heart))
         
         content = dict(id=eid,status='ok')
         content.update(self.engine_info)
         # check if requesting available IDs:
         if queue in self.by_ident:
             try:
                 raise KeyError("queue_id %r in use"%queue)
             except:
-                content = wrap_exception()
+                content = error.wrap_exception()
                 self.log.error("queue_id %r in use"%queue, exc_info=True)
         elif heart in self.hearts: # need to check unique hearts?
             try:
                 raise KeyError("heart_id %r in use"%heart)
             except:
                 self.log.error("heart_id %r in use"%heart, exc_info=True)
-                content = wrap_exception()
+                content = error.wrap_exception()
         else:
             for h, pack in self.incoming_registrations.iteritems():
                 if heart == h:
                     try:
                         raise KeyError("heart_id %r in use"%heart)
                     except:
                         self.log.error("heart_id %r in use"%heart, exc_info=True)
-                        content = wrap_exception()
+                        content = error.wrap_exception()
                     break
                 elif queue == pack[1]:
                     try:
                         raise KeyError("queue_id %r in use"%queue)
                     except:
                         self.log.error("queue_id %r in use"%queue, exc_info=True)
-                        content = wrap_exception()
+                        content = error.wrap_exception()
                     break
         
         msg = self.session.send(self.registrar, "registration_reply", 
                 content=content, 
                 ident=reg)
         
         if content['status'] == 'ok':
             if heart in self.heartmonitor.hearts:
                 # already beating
                 self.incoming_registrations[heart] = (eid,queue,reg[0],None)
                 self.finish_registration(heart)
             else:
                 purge = lambda : self._purge_stalled_registration(heart)
                 dc = ioloop.DelayedCallback(purge, self.registration_timeout, self.loop)
                 dc.start()
                 self.incoming_registrations[heart] = (eid,queue,reg[0],dc)
         else:
             self.log.error("registration::registration %i failed: %s"%(eid, content['evalue']))
         return eid
     
     def unregister_engine(self, ident, msg):
         """Unregister an engine that explicitly requested to leave."""
         try:
             eid = msg['content']['id']
         except:
             self.log.error("registration::bad engine id for unregistration: %s"%ident, exc_info=True)
             return
         self.log.info("registration::unregister_engine(%s)"%eid)
         content=dict(id=eid, queue=self.engines[eid].queue)
         self.ids.remove(eid)
         self.keytable.pop(eid)
         ec = self.engines.pop(eid)
         self.hearts.pop(ec.heartbeat)
         self.by_ident.pop(ec.queue)
         self.completed.pop(eid)
         for msg_id in self.queues.pop(eid):
             msg = self.pending.remove(msg_id)
             ############## TODO: HANDLE IT ################
         
         if self.notifier:
             self.session.send(self.notifier, "unregistration_notification", content=content)
     
     def finish_registration(self, heart):
         """Second half of engine registration, called after our HeartMonitor
         has received a beat from the Engine's Heart."""
         try: 
             (eid,queue,reg,purge) = self.incoming_registrations.pop(heart)
         except KeyError:
             self.log.error("registration::tried to finish nonexistant registration", exc_info=True)
             return
         self.log.info("registration::finished registering engine %i:%r"%(eid,queue))
         if purge is not None:
             purge.stop()
         control = queue
         self.ids.add(eid)
         self.keytable[eid] = queue
         self.engines[eid] = EngineConnector(id=eid, queue=queue, registration=reg, 
                                     control=control, heartbeat=heart)
         self.by_ident[queue] = eid
         self.queues[eid] = list()
         self.tasks[eid] = list()
         self.completed[eid] = list()
         self.hearts[heart] = eid
         content = dict(id=eid, queue=self.engines[eid].queue)
         if self.notifier:
             self.session.send(self.notifier, "registration_notification", content=content)
         self.log.info("engine::Engine Connected: %i"%eid)
     
     def _purge_stalled_registration(self, heart):
         if heart in self.incoming_registrations:
             eid = self.incoming_registrations.pop(heart)[0]
             self.log.info("registration::purging stalled registration: %i"%eid)
         else:
             pass
             
     #-------------------------------------------------------------------------
     # Client Requests
     #-------------------------------------------------------------------------
     
     def shutdown_request(self, client_id, msg):
         """handle shutdown request."""
         # s = self.context.socket(zmq.XREQ)
         # s.connect(self.client_connections['mux'])
         # time.sleep(0.1)
         # for eid,ec in self.engines.iteritems():
         #     self.session.send(s, 'shutdown_request', content=dict(restart=False), ident=ec.queue)
         # time.sleep(1)
         self.session.send(self.clientele, 'shutdown_reply', content={'status': 'ok'}, ident=client_id)
         dc = ioloop.DelayedCallback(lambda : self._shutdown(), 1000, self.loop)
         dc.start()
     
     def _shutdown(self):
         self.log.info("hub::hub shutting down.")
         time.sleep(0.1)
         sys.exit(0)
         
     
     def check_load(self, client_id, msg):
         content = msg['content']
         try:
             targets = content['targets']
             targets = self._validate_targets(targets)
         except:
-            content = wrap_exception()
+            content = error.wrap_exception()
             self.session.send(self.clientele, "hub_error", 
                     content=content, ident=client_id)
             return
         
         content = dict(status='ok')
         # loads = {}
         for t in targets:
             content[bytes(t)] = len(self.queues[t])+len(self.tasks[t])
         self.session.send(self.clientele, "load_reply", content=content, ident=client_id)
             
     
     def queue_status(self, client_id, msg):
         """Return the Queue status of one or more targets.
         if verbose: return the msg_ids
         else: return len of each type.
         keys: queue (pending MUX jobs)
             tasks (pending Task jobs)
             completed (finished jobs from both queues)"""
         content = msg['content']
         targets = content['targets']
         try:
             targets = self._validate_targets(targets)
         except:
-            content = wrap_exception()
+            content = error.wrap_exception()
             self.session.send(self.clientele, "hub_error", 
                     content=content, ident=client_id)
             return
         verbose = content.get('verbose', False)
         content = dict(status='ok')
         for t in targets:
             queue = self.queues[t]
             completed = self.completed[t]
             tasks = self.tasks[t]
             if not verbose:
                 queue = len(queue)
                 completed = len(completed)
                 tasks = len(tasks)
             content[bytes(t)] = {'queue': queue, 'completed': completed , 'tasks': tasks}
             # pending
         self.session.send(self.clientele, "queue_reply", content=content, ident=client_id)
     
     def purge_results(self, client_id, msg):
         """Purge results from memory. This method is more valuable before we move
         to a DB based message storage mechanism."""
         content = msg['content']
         msg_ids = content.get('msg_ids', [])
         reply = dict(status='ok')
         if msg_ids == 'all':
             self.db.drop_matching_records(dict(completed={'$ne':None}))
         else:
             for msg_id in msg_ids:
                 if msg_id in self.all_completed:
                     self.db.drop_record(msg_id)
                 else:
                     if msg_id in self.pending:
                         try:
                             raise IndexError("msg pending: %r"%msg_id)
                         except:
-                            reply = wrap_exception()
+                            reply = error.wrap_exception()
                     else:
                         try:
                             raise IndexError("No such msg: %r"%msg_id)
                         except:
-                            reply = wrap_exception()
+                            reply = error.wrap_exception()
                     break
             eids = content.get('engine_ids', [])
             for eid in eids:
                 if eid not in self.engines:
                     try:
                         raise IndexError("No such engine: %i"%eid)
                     except:
-                        reply = wrap_exception()
+                        reply = error.wrap_exception()
                     break
                 msg_ids = self.completed.pop(eid)
                 uid = self.engines[eid].queue
                 self.db.drop_matching_records(dict(engine_uuid=uid, completed={'$ne':None}))
         
         self.session.send(self.clientele, 'purge_reply', content=reply, ident=client_id)
     
     def resubmit_task(self, client_id, msg, buffers):
         """Resubmit a task."""
         raise NotImplementedError
     
     def get_results(self, client_id, msg):
         """Get the result of 1 or more messages."""
         content = msg['content']
         msg_ids = sorted(set(content['msg_ids']))
         statusonly = content.get('status_only', False)
         pending = []
         completed = []
         content = dict(status='ok')
         content['pending'] = pending
         content['completed'] = completed
         buffers = []
         if not statusonly:
             content['results'] = {}
             records = self.db.find_records(dict(msg_id={'$in':msg_ids}))
         for msg_id in msg_ids:
             if msg_id in self.pending:
                 pending.append(msg_id)
             elif msg_id in self.all_completed:
                 completed.append(msg_id)
                 if not statusonly:
                     rec = records[msg_id]
                     io_dict = {}
                     for key in 'pyin pyout pyerr stdout stderr'.split():
                             io_dict[key] = rec[key]
                     content[msg_id] = { 'result_content': rec['result_content'],
                                         'header': rec['header'],
                                         'result_header' : rec['result_header'],
                                         'io' : io_dict,
                                       }
                     buffers.extend(map(str, rec['result_buffers']))
             else:
                 try:
                     raise KeyError('No such message: '+msg_id)
                 except:
-                    content = wrap_exception()
+                    content = error.wrap_exception()
                 break
         self.session.send(self.clientele, "result_reply", content=content, 
                                             parent=msg, ident=client_id,
                                             buffers=buffers)
 

IPython/zmq/parallel/remotefunction.py

@@ -1,174 +1,203 b''
 """Remote Functions and decorators for the client."""
 #-----------------------------------------------------------------------------
 #  Copyright (C) 2010  The IPython Development Team
 #
 #  Distributed under the terms of the BSD License.  The full license is in
 #  the file COPYING, distributed as part of this software.
 #-----------------------------------------------------------------------------
 
 #-----------------------------------------------------------------------------
 # Imports
 #-----------------------------------------------------------------------------
 
 import warnings
 
 from IPython.testing import decorators as testdec
 
 from . import map as Map
 from .asyncresult import AsyncMapResult
 
 #-----------------------------------------------------------------------------
 # Decorators
 #-----------------------------------------------------------------------------
 
 @testdec.skip_doctest
 def remote(client, bound=True, block=None, targets=None, balanced=None):
     """Turn a function into a remote function.
     
     This method can be used for map:
     
     In [1]: @remote(client,block=True)
        ...: def func(a):
        ...:    pass
     """
     
     def remote_function(f):
         return RemoteFunction(client, f, bound, block, targets, balanced)
     return remote_function
 
 @testdec.skip_doctest
 def parallel(client, dist='b', bound=True, block=None, targets='all', balanced=None):
     """Turn a function into a parallel remote function.
     
     This method can be used for map:
     
     In [1]: @parallel(client,block=True)
        ...: def func(a):
        ...:    pass
     """
     
     def parallel_function(f):
         return ParallelFunction(client, f, dist, bound, block, targets, balanced)
     return parallel_function
 
 #--------------------------------------------------------------------------
 # Classes
 #--------------------------------------------------------------------------
 
 class RemoteFunction(object):
     """Turn an existing function into a remote function.
     
     Parameters
     ----------
     
     client : Client instance
         The client to be used to connect to engines
     f : callable
         The function to be wrapped into a remote function
     bound : bool [default: False]
         Whether the affect the remote namespace when called
     block : bool [default: None]
         Whether to wait for results or not.  The default behavior is
         to use the current `block` attribute of `client`
     targets : valid target list [default: all]
         The targets on which to execute.
     balanced : bool
         Whether to load-balance with the Task scheduler or not
     """
     
     client = None # the remote connection
     func = None # the wrapped function
     block = None # whether to block
     bound = None # whether to affect the namespace
     targets = None # where to execute
     balanced = None # whether to load-balance
     
     def __init__(self, client, f, bound=False, block=None, targets=None, balanced=None):
         self.client = client
         self.func = f
         self.block=block
         self.bound=bound
         self.targets=targets
         if balanced is None:
             if targets is None:
                 balanced = True
             else:
                 balanced = False
         self.balanced = balanced
     
     def __call__(self, *args, **kwargs):
         return self.client.apply(self.func, args=args, kwargs=kwargs,
                 block=self.block, targets=self.targets, bound=self.bound, balanced=self.balanced)
     
 
 class ParallelFunction(RemoteFunction):
-    """Class for mapping a function to sequences."""
+    """Class for mapping a function to sequences.
+    
+    This will distribute the sequences according the a mapper, and call
+    the function on each sub-sequence.  If called via map, then the function
+    will be called once on each element, rather that each sub-sequence.
+    
+    Parameters
+    ----------
+    
+    client : Client instance
+        The client to be used to connect to engines
+    f : callable
+        The function to be wrapped into a remote function
+    bound : bool [default: False]
+        Whether the affect the remote namespace when called
+    block : bool [default: None]
+        Whether to wait for results or not.  The default behavior is
+        to use the current `block` attribute of `client`
+    targets : valid target list [default: all]
+        The targets on which to execute.
+    balanced : bool
+        Whether to load-balance with the Task scheduler or not
+    chunk_size : int or None
+        The size of chunk to use when breaking up sequences in a load-balanced manner
+    """
     def __init__(self, client, f, dist='b', bound=False, block=None, targets='all', balanced=None, chunk_size=None):
         super(ParallelFunction, self).__init__(client,f,bound,block,targets,balanced)
         self.chunk_size = chunk_size
         
         mapClass = Map.dists[dist]
         self.mapObject = mapClass()
     
     def __call__(self, *sequences):
         len_0 = len(sequences[0])
         for s in sequences:
             if len(s)!=len_0:
                 msg = 'all sequences must have equal length, but %i!=%i'%(len_0,len(s))
                 raise ValueError(msg)
         
         if self.balanced:
             if self.chunk_size:
                 nparts = len_0/self.chunk_size + int(len_0%self.chunk_size > 0)
             else:
                 nparts = len_0
             targets = [self.targets]*nparts
         else:
             if self.chunk_size:
                 warnings.warn("`chunk_size` is ignored when `balanced=False", UserWarning)
             # multiplexed:
             targets = self.client._build_targets(self.targets)[-1]
             nparts = len(targets)
         
         msg_ids = []
         # my_f = lambda *a: map(self.func, *a)
         for index, t in enumerate(targets):
             args = []
             for seq in sequences:
                 part = self.mapObject.getPartition(seq, index, nparts)
                 if len(part) == 0:
                     continue
                 else:
                     args.append(part)
             if not args:
                 continue
             
             # print (args)
             if hasattr(self, '_map'):
                 f = map
                 args = [self.func]+args
             else:
                 f=self.func
             ar = self.client.apply(f, args=args, block=False, bound=self.bound, 
                         targets=t, balanced=self.balanced)
             
             msg_ids.append(ar.msg_ids[0])
         
         r = AsyncMapResult(self.client, msg_ids, self.mapObject, fname=self.func.__name__)
         if self.block:
             try:
                 return r.get()
             except KeyboardInterrupt:
                 return r
         else:
             return r
     
     def map(self, *sequences):
-        """call a function on each element of a sequence remotely."""
+        """call a function on each element of a sequence remotely. 
+        This should behave very much like the builtin map, but return an AsyncMapResult
+        if self.block is False.
+        """
+        # set _map as a flag for use inside self.__call__
         self._map = True
         try:
             ret = self.__call__(*sequences)
         finally:
             del self._map
         return ret
 
+__all__ = ['remote', 'parallel', 'RemoteFunction', 'ParallelFunction']
  No newline at end of file

IPython/zmq/parallel/scheduler.py

@@ -1,581 +1,580 b''
 """The Python scheduler for rich scheduling.
 
 The Pure ZMQ scheduler does not allow routing schemes other than LRU,
 nor does it check msg_id DAG dependencies. For those, a slightly slower
 Python Scheduler exists.
 """
 
 #----------------------------------------------------------------------
 # Imports
 #----------------------------------------------------------------------
 
 from __future__ import print_function
 
 import logging
 import sys
 
 from datetime import datetime, timedelta
 from random import randint, random
 from types import FunctionType
 
 try:
     import numpy
 except ImportError:
     numpy = None
 
 import zmq
 from zmq.eventloop import ioloop, zmqstream
 
 # local imports
 from IPython.external.decorator import decorator
 from IPython.utils.traitlets import Instance, Dict, List, Set
 
 from . import error
-from . import streamsession as ss
 from .dependency import Dependency
 from .entry_point import connect_logger, local_logger
 from .factory import SessionFactory
 
 
 @decorator
 def logged(f,self,*args,**kwargs):
     # print ("#--------------------")
     self.log.debug("scheduler::%s(*%s,**%s)"%(f.func_name, args, kwargs))
     # print ("#--")
     return f(self,*args, **kwargs)
 
 #----------------------------------------------------------------------
 # Chooser functions
 #----------------------------------------------------------------------
 
 def plainrandom(loads):
     """Plain random pick."""
     n = len(loads)
     return randint(0,n-1)
 
 def lru(loads):
     """Always pick the front of the line.
     
     The content of `loads` is ignored.
     
     Assumes LRU ordering of loads, with oldest first.
     """
     return 0
 
 def twobin(loads):
     """Pick two at random, use the LRU of the two.
     
     The content of loads is ignored.
     
     Assumes LRU ordering of loads, with oldest first.
     """
     n = len(loads)
     a = randint(0,n-1)
     b = randint(0,n-1)
     return min(a,b)
 
 def weighted(loads):
     """Pick two at random using inverse load as weight.
     
     Return the less loaded of the two.
     """
     # weight 0 a million times more than 1:
     weights = 1./(1e-6+numpy.array(loads))
     sums = weights.cumsum()
     t = sums[-1]
     x = random()*t
     y = random()*t
     idx = 0
     idy = 0
     while sums[idx] < x:
         idx += 1
     while sums[idy] < y:
         idy += 1
     if weights[idy] > weights[idx]:
         return idy
     else:
         return idx
 
 def leastload(loads):
     """Always choose the lowest load.
     
     If the lowest load occurs more than once, the first
     occurance will be used.  If loads has LRU ordering, this means
     the LRU of those with the lowest load is chosen.
     """
     return loads.index(min(loads))
 
 #---------------------------------------------------------------------
 # Classes
 #---------------------------------------------------------------------
 # store empty default dependency:
 MET = Dependency([])
 
 class TaskScheduler(SessionFactory):
     """Python TaskScheduler object.
     
     This is the simplest object that supports msg_id based
     DAG dependencies. *Only* task msg_ids are checked, not
     msg_ids of jobs submitted via the MUX queue.
     
     """
     
     # input arguments:
     scheme = Instance(FunctionType, default=leastload) # function for determining the destination
     client_stream = Instance(zmqstream.ZMQStream) # client-facing stream
     engine_stream = Instance(zmqstream.ZMQStream) # engine-facing stream
     notifier_stream = Instance(zmqstream.ZMQStream) # hub-facing sub stream
     mon_stream = Instance(zmqstream.ZMQStream) # hub-facing pub stream
     
     # internals:
     graph = Dict() # dict by msg_id of [ msg_ids that depend on key ]
     depending = Dict() # dict by msg_id of (msg_id, raw_msg, after, follow)
     pending = Dict() # dict by engine_uuid of submitted tasks
     completed = Dict() # dict by engine_uuid of completed tasks
     failed = Dict() # dict by engine_uuid of failed tasks
     destinations = Dict() # dict by msg_id of engine_uuids where jobs ran (reverse of completed+failed)
     clients = Dict() # dict by msg_id for who submitted the task
     targets = List() # list of target IDENTs
     loads = List() # list of engine loads
     all_completed = Set() # set of all completed tasks
     all_failed = Set() # set of all failed tasks
     all_done = Set() # set of all finished tasks=union(completed,failed)
     all_ids = Set() # set of all submitted task IDs
     blacklist = Dict() # dict by msg_id of locations where a job has encountered UnmetDependency
     auditor = Instance('zmq.eventloop.ioloop.PeriodicCallback')
     
     
     def start(self):
         self.engine_stream.on_recv(self.dispatch_result, copy=False)
         self._notification_handlers = dict(
             registration_notification = self._register_engine,
             unregistration_notification = self._unregister_engine
         )
         self.notifier_stream.on_recv(self.dispatch_notification)
         self.auditor = ioloop.PeriodicCallback(self.audit_timeouts, 2e3, self.loop) # 1 Hz
         self.auditor.start()
         self.log.info("Scheduler started...%r"%self)
     
     def resume_receiving(self):
         """Resume accepting jobs."""
         self.client_stream.on_recv(self.dispatch_submission, copy=False)
     
     def stop_receiving(self):
         """Stop accepting jobs while there are no engines.
         Leave them in the ZMQ queue."""
         self.client_stream.on_recv(None)
     
     #-----------------------------------------------------------------------
     # [Un]Registration Handling
     #-----------------------------------------------------------------------
     
     def dispatch_notification(self, msg):
         """dispatch register/unregister events."""
         idents,msg = self.session.feed_identities(msg)
         msg = self.session.unpack_message(msg)
         msg_type = msg['msg_type']
         handler = self._notification_handlers.get(msg_type, None)
         if handler is None:
             raise Exception("Unhandled message type: %s"%msg_type)
         else:
             try:
                 handler(str(msg['content']['queue']))
             except KeyError:
                 self.log.error("task::Invalid notification msg: %s"%msg)
     
     @logged
     def _register_engine(self, uid):
         """New engine with ident `uid` became available."""
         # head of the line:
         self.targets.insert(0,uid)
         self.loads.insert(0,0)
         # initialize sets
         self.completed[uid] = set()
         self.failed[uid] = set()
         self.pending[uid] = {}
         if len(self.targets) == 1:
             self.resume_receiving()
 
     def _unregister_engine(self, uid):
         """Existing engine with ident `uid` became unavailable."""
         if len(self.targets) == 1:
             # this was our only engine
             self.stop_receiving()
         
         # handle any potentially finished tasks:
         self.engine_stream.flush()
         
         self.completed.pop(uid)
         self.failed.pop(uid)
         # don't pop destinations, because it might be used later
         # map(self.destinations.pop, self.completed.pop(uid))
         # map(self.destinations.pop, self.failed.pop(uid))
         
         idx = self.targets.index(uid)
         self.targets.pop(idx)
         self.loads.pop(idx)
         
         # wait 5 seconds before cleaning up pending jobs, since the results might
         # still be incoming
         if self.pending[uid]:
             dc = ioloop.DelayedCallback(lambda : self.handle_stranded_tasks(uid), 5000, self.loop)
             dc.start()
     
     @logged
     def handle_stranded_tasks(self, engine):
         """Deal with jobs resident in an engine that died."""
         lost = self.pending.pop(engine)
         
         for msg_id, (raw_msg, targets, MET, follow, timeout) in lost.iteritems():
             self.all_failed.add(msg_id)
             self.all_done.add(msg_id)
             idents,msg = self.session.feed_identities(raw_msg, copy=False)
             msg = self.session.unpack_message(msg, copy=False, content=False)
             parent = msg['header']
             idents = [idents[0],engine]+idents[1:]
             print (idents)
             try:
                 raise error.EngineError("Engine %r died while running task %r"%(engine, msg_id))
             except:
-                content = ss.wrap_exception()
+                content = error.wrap_exception()
             msg = self.session.send(self.client_stream, 'apply_reply', content, 
                                                     parent=parent, ident=idents)
             self.session.send(self.mon_stream, msg, ident=['outtask']+idents)
             self.update_graph(msg_id)
     
     
     #-----------------------------------------------------------------------
     # Job Submission
     #-----------------------------------------------------------------------
     @logged
     def dispatch_submission(self, raw_msg):
         """Dispatch job submission to appropriate handlers."""
         # ensure targets up to date:
         self.notifier_stream.flush()
         try:
             idents, msg = self.session.feed_identities(raw_msg, copy=False)
             msg = self.session.unpack_message(msg, content=False, copy=False)
         except:
             self.log.error("task::Invaid task: %s"%raw_msg, exc_info=True)
             return
         
         # send to monitor
         self.mon_stream.send_multipart(['intask']+raw_msg, copy=False)
         
         header = msg['header']
         msg_id = header['msg_id']
         self.all_ids.add(msg_id)
         
         # targets
         targets = set(header.get('targets', []))
         
         # time dependencies
         after = Dependency(header.get('after', []))
         if after.all:
             after.difference_update(self.all_completed)
             if not after.success_only:
                 after.difference_update(self.all_failed)
         if after.check(self.all_completed, self.all_failed):
             # recast as empty set, if `after` already met,
             # to prevent unnecessary set comparisons
             after = MET
         
         # location dependencies
         follow = Dependency(header.get('follow', []))
         
         # turn timeouts into datetime objects:
         timeout = header.get('timeout', None)
         if timeout:
             timeout = datetime.now() + timedelta(0,timeout,0)
         
         args = [raw_msg, targets, after, follow, timeout]
         
         # validate and reduce dependencies:
         for dep in after,follow:
             # check valid:
             if msg_id in dep or dep.difference(self.all_ids):
                 self.depending[msg_id] = args
                 return self.fail_unreachable(msg_id, error.InvalidDependency)
             # check if unreachable:
             if dep.unreachable(self.all_failed):
                 self.depending[msg_id] = args
                 return self.fail_unreachable(msg_id)
         
         if after.check(self.all_completed, self.all_failed):
             # time deps already met, try to run
             if not self.maybe_run(msg_id, *args):
                 # can't run yet
                 self.save_unmet(msg_id, *args)
         else:
             self.save_unmet(msg_id, *args)
     
     # @logged
     def audit_timeouts(self):
         """Audit all waiting tasks for expired timeouts."""
         now = datetime.now()
         for msg_id in self.depending.keys():
             # must recheck, in case one failure cascaded to another:
             if msg_id in self.depending:
                 raw,after,targets,follow,timeout = self.depending[msg_id]
                 if timeout and timeout < now:
                     self.fail_unreachable(msg_id, timeout=True)
                 
     @logged
     def fail_unreachable(self, msg_id, why=error.ImpossibleDependency):
         """a task has become unreachable, send a reply with an ImpossibleDependency
         error."""
         if msg_id not in self.depending:
             self.log.error("msg %r already failed!"%msg_id)
             return
         raw_msg,targets,after,follow,timeout = self.depending.pop(msg_id)
         for mid in follow.union(after):
             if mid in self.graph:
                 self.graph[mid].remove(msg_id)
         
         # FIXME: unpacking a message I've already unpacked, but didn't save:
         idents,msg = self.session.feed_identities(raw_msg, copy=False)
         msg = self.session.unpack_message(msg, copy=False, content=False)
         header = msg['header']
         
         try:
             raise why()
         except:
-            content = ss.wrap_exception()
+            content = error.wrap_exception()
         
         self.all_done.add(msg_id)
         self.all_failed.add(msg_id)
         
         msg = self.session.send(self.client_stream, 'apply_reply', content, 
                                                 parent=header, ident=idents)
         self.session.send(self.mon_stream, msg, ident=['outtask']+idents)
         
         self.update_graph(msg_id, success=False)
     
     @logged
     def maybe_run(self, msg_id, raw_msg, targets, after, follow, timeout):
         """check location dependencies, and run if they are met."""
         blacklist = self.blacklist.setdefault(msg_id, set())
         if follow or targets or blacklist:
             # we need a can_run filter
             def can_run(idx):
                 target = self.targets[idx]
                 # check targets
                 if targets and target not in targets:
                     return False
                 # check blacklist
                 if target in blacklist:
                     return False
                 # check follow
                 return follow.check(self.completed[target], self.failed[target])
             
             indices = filter(can_run, range(len(self.targets)))
             if not indices:
                 # couldn't run
                 if follow.all:
                     # check follow for impossibility
                     dests = set()
                     relevant = self.all_completed if follow.success_only else self.all_done
                     for m in follow.intersection(relevant):
                         dests.add(self.destinations[m])
                     if len(dests) > 1:
                         self.fail_unreachable(msg_id)
                         return False
                 if targets:
                     # check blacklist+targets for impossibility
                     targets.difference_update(blacklist)
                     if not targets or not targets.intersection(self.targets):
                         self.fail_unreachable(msg_id)
                         return False
                 return False
         else:
             indices = None
             
         self.submit_task(msg_id, raw_msg, targets, follow, timeout, indices)
         return True
             
     @logged
     def save_unmet(self, msg_id, raw_msg, targets, after, follow, timeout):
         """Save a message for later submission when its dependencies are met."""
         self.depending[msg_id] = [raw_msg,targets,after,follow,timeout]
         # track the ids in follow or after, but not those already finished
         for dep_id in after.union(follow).difference(self.all_done):
             if dep_id not in self.graph:
                 self.graph[dep_id] = set()
             self.graph[dep_id].add(msg_id)
     
     @logged
     def submit_task(self, msg_id, raw_msg, targets, follow, timeout, indices=None):
         """Submit a task to any of a subset of our targets."""
         if indices:
             loads = [self.loads[i] for i in indices]
         else:
             loads = self.loads
         idx = self.scheme(loads)
         if indices:
             idx = indices[idx]
         target = self.targets[idx]
         # print (target, map(str, msg[:3]))
         self.engine_stream.send(target, flags=zmq.SNDMORE, copy=False)
         self.engine_stream.send_multipart(raw_msg, copy=False)
         self.add_job(idx)
         self.pending[target][msg_id] = (raw_msg, targets, MET, follow, timeout)
         content = dict(msg_id=msg_id, engine_id=target)
         self.session.send(self.mon_stream, 'task_destination', content=content, 
                         ident=['tracktask',self.session.session])
     
     #-----------------------------------------------------------------------
     # Result Handling
     #-----------------------------------------------------------------------
     @logged
     def dispatch_result(self, raw_msg):
         """dispatch method for result replies"""
         try:
             idents,msg = self.session.feed_identities(raw_msg, copy=False)
             msg = self.session.unpack_message(msg, content=False, copy=False)
         except:
             self.log.error("task::Invaid result: %s"%raw_msg, exc_info=True)
             return
         
         header = msg['header']
         if header.get('dependencies_met', True):
             success = (header['status'] == 'ok')
             self.handle_result(idents, msg['parent_header'], raw_msg, success)
             # send to Hub monitor
             self.mon_stream.send_multipart(['outtask']+raw_msg, copy=False)
         else:
             self.handle_unmet_dependency(idents, msg['parent_header'])
         
     @logged
     def handle_result(self, idents, parent, raw_msg, success=True):
         """handle a real task result, either success or failure"""
         # first, relay result to client
         engine = idents[0]
         client = idents[1]
         # swap_ids for XREP-XREP mirror
         raw_msg[:2] = [client,engine]
         # print (map(str, raw_msg[:4]))
         self.client_stream.send_multipart(raw_msg, copy=False)
         # now, update our data structures
         msg_id = parent['msg_id']
         self.blacklist.pop(msg_id, None)
         self.pending[engine].pop(msg_id)
         if success:
             self.completed[engine].add(msg_id)
             self.all_completed.add(msg_id)
         else:
             self.failed[engine].add(msg_id)
             self.all_failed.add(msg_id)
         self.all_done.add(msg_id)
         self.destinations[msg_id] = engine
         
         self.update_graph(msg_id, success)
         
     @logged
     def handle_unmet_dependency(self, idents, parent):
         """handle an unmet dependency"""
         engine = idents[0]
         msg_id = parent['msg_id']
         
         if msg_id not in self.blacklist:
             self.blacklist[msg_id] = set()
         self.blacklist[msg_id].add(engine)
         
         args = self.pending[engine].pop(msg_id)
         raw,targets,after,follow,timeout = args
         
         if self.blacklist[msg_id] == targets:
             self.depending[msg_id] = args
             return self.fail_unreachable(msg_id)
         
         elif not self.maybe_run(msg_id, *args):
             # resubmit failed, put it back in our dependency tree
             self.save_unmet(msg_id, *args)
         
     
     @logged
     def update_graph(self, dep_id, success=True):
         """dep_id just finished. Update our dependency
         graph and submit any jobs that just became runable."""
         # print ("\n\n***********")
         # pprint (dep_id)
         # pprint (self.graph)
         # pprint (self.depending)
         # pprint (self.all_completed)
         # pprint (self.all_failed)
         # print ("\n\n***********\n\n")
         if dep_id not in self.graph:
             return
         jobs = self.graph.pop(dep_id)
         
         for msg_id in jobs:
             raw_msg, targets, after, follow, timeout = self.depending[msg_id]
             # if dep_id in after:
             #     if after.all and (success or not after.success_only):
             #         after.remove(dep_id)
             
             if after.unreachable(self.all_failed) or follow.unreachable(self.all_failed):
                 self.fail_unreachable(msg_id)
             
             elif after.check(self.all_completed, self.all_failed): # time deps met, maybe run
                 if self.maybe_run(msg_id, raw_msg, targets, MET, follow, timeout):
                     
                     self.depending.pop(msg_id)
                     for mid in follow.union(after):
                         if mid in self.graph:
                             self.graph[mid].remove(msg_id)
     
     #----------------------------------------------------------------------
     # methods to be overridden by subclasses
     #----------------------------------------------------------------------
     
     def add_job(self, idx):
         """Called after self.targets[idx] just got the job with header.
         Override with subclasses.  The default ordering is simple LRU.
         The default loads are the number of outstanding jobs."""
         self.loads[idx] += 1
         for lis in (self.targets, self.loads):
             lis.append(lis.pop(idx))
             
     
     def finish_job(self, idx):
         """Called after self.targets[idx] just finished a job.
         Override with subclasses."""
         self.loads[idx] -= 1
     
 
 
 def launch_scheduler(in_addr, out_addr, mon_addr, not_addr, config=None,logname='ZMQ', 
                             log_addr=None, loglevel=logging.DEBUG, scheme='lru'):
     from zmq.eventloop import ioloop
     from zmq.eventloop.zmqstream import ZMQStream
     
     ctx = zmq.Context()
     loop = ioloop.IOLoop()
     print (in_addr, out_addr, mon_addr, not_addr)
     ins = ZMQStream(ctx.socket(zmq.XREP),loop)
     ins.bind(in_addr)
     outs = ZMQStream(ctx.socket(zmq.XREP),loop)
     outs.bind(out_addr)
     mons = ZMQStream(ctx.socket(zmq.PUB),loop)
     mons.connect(mon_addr)
     nots = ZMQStream(ctx.socket(zmq.SUB),loop)
     nots.setsockopt(zmq.SUBSCRIBE, '')
     nots.connect(not_addr)
     
     scheme = globals().get(scheme, None)
     # setup logging
     if log_addr:
         connect_logger(logname, ctx, log_addr, root="scheduler", loglevel=loglevel)
     else:
         local_logger(logname, loglevel)
     
     scheduler = TaskScheduler(client_stream=ins, engine_stream=outs,
                             mon_stream=mons, notifier_stream=nots,
                             scheme=scheme, loop=loop, logname=logname,
                             config=config)
     scheduler.start()
     try:
         loop.start()
     except KeyboardInterrupt:
         print ("interrupted, exiting...", file=sys.__stderr__)
 

IPython/zmq/parallel/streamkernel.py

@@ -1,487 +1,484 b''
 #!/usr/bin/env python
 """
 Kernel adapted from kernel.py to use ZMQ Streams
 """
 
 #-----------------------------------------------------------------------------
 # Imports
 #-----------------------------------------------------------------------------
 
 # Standard library imports.
 from __future__ import print_function
-import __builtin__
 
-import logging
-import os
 import sys
 import time
-import traceback
 
 from code import CommandCompiler
 from datetime import datetime
 from pprint import pprint
 from signal import SIGTERM, SIGKILL
 
 # System library imports.
 import zmq
 from zmq.eventloop import ioloop, zmqstream
 
 # Local imports.
 from IPython.core import ultratb
-from IPython.utils.traitlets import HasTraits, Instance, List, Int, Dict, Set, Str
+from IPython.utils.traitlets import Instance, List, Int, Dict, Set, Str
 from IPython.zmq.completer import KernelCompleter
 from IPython.zmq.iostream import OutStream
 from IPython.zmq.displayhook import DisplayHook
 
 from . import heartmonitor
 from .client import Client
+from .error import wrap_exception
 from .factory import SessionFactory
-from .streamsession import StreamSession, Message, extract_header, serialize_object,\
-                unpack_apply_message, ISO8601, wrap_exception
+from .streamsession import StreamSession
+from .util import serialize_object, unpack_apply_message, ISO8601
 
 def printer(*args):
     pprint(args, stream=sys.__stdout__)
 
 
 class _Passer:
     """Empty class that implements `send()` that does nothing."""
     def send(self, *args, **kwargs):
         pass
     send_multipart = send
     
 
 #-----------------------------------------------------------------------------
 # Main kernel class
 #-----------------------------------------------------------------------------
 
 class Kernel(SessionFactory):
 
     #---------------------------------------------------------------------------
     # Kernel interface
     #---------------------------------------------------------------------------
     
     # kwargs:
     int_id = Int(-1, config=True)
     user_ns = Dict(config=True)
     exec_lines = List(config=True)
     
     control_stream = Instance(zmqstream.ZMQStream)
     task_stream = Instance(zmqstream.ZMQStream)
     iopub_stream = Instance(zmqstream.ZMQStream)
     client = Instance('IPython.zmq.parallel.client.Client')
     
     # internals
     shell_streams = List()
     compiler = Instance(CommandCompiler, (), {})
     completer = Instance(KernelCompleter)
     
     aborted = Set()
     shell_handlers = Dict()
     control_handlers = Dict()
     
     def _set_prefix(self):
         self.prefix = "engine.%s"%self.int_id
     
     def _connect_completer(self):
         self.completer = KernelCompleter(self.user_ns)
     
     def __init__(self, **kwargs):
         super(Kernel, self).__init__(**kwargs)
         self._set_prefix()
         self._connect_completer()
         
         self.on_trait_change(self._set_prefix, 'id')
         self.on_trait_change(self._connect_completer, 'user_ns')
         
         # Build dict of handlers for message types
         for msg_type in ['execute_request', 'complete_request', 'apply_request', 
                 'clear_request']:
             self.shell_handlers[msg_type] = getattr(self, msg_type)
         
         for msg_type in ['shutdown_request', 'abort_request']+self.shell_handlers.keys():
             self.control_handlers[msg_type] = getattr(self, msg_type)
         
         self._initial_exec_lines()
     
     def _wrap_exception(self, method=None):
         e_info = dict(engine_uuid=self.ident, engine_id=self.int_id, method=method)
         content=wrap_exception(e_info)
         return content
     
     def _initial_exec_lines(self):
         s = _Passer()
         content = dict(silent=True, user_variable=[],user_expressions=[])
         for line in self.exec_lines:
             self.log.debug("executing initialization: %s"%line)
             content.update({'code':line})
             msg = self.session.msg('execute_request', content)
             self.execute_request(s, [], msg)
         
         
     #-------------------- control handlers -----------------------------
     def abort_queues(self):
         for stream in self.shell_streams:
             if stream:
                 self.abort_queue(stream)
     
     def abort_queue(self, stream):
         while True:
             try:
                 msg = self.session.recv(stream, zmq.NOBLOCK,content=True)
             except zmq.ZMQError as e:
                 if e.errno == zmq.EAGAIN:
                     break
                 else:
                     return
             else:
                 if msg is None:
                     return
                 else:
                     idents,msg = msg
                 
                 # assert self.reply_socketly_socket.rcvmore(), "Unexpected missing message part."
                 # msg = self.reply_socket.recv_json()
             self.log.info("Aborting:")
             self.log.info(str(msg))
             msg_type = msg['msg_type']
             reply_type = msg_type.split('_')[0] + '_reply'
             # reply_msg = self.session.msg(reply_type, {'status' : 'aborted'}, msg)
             # self.reply_socket.send(ident,zmq.SNDMORE)
             # self.reply_socket.send_json(reply_msg)
             reply_msg = self.session.send(stream, reply_type, 
                         content={'status' : 'aborted'}, parent=msg, ident=idents)[0]
             self.log.debug(str(reply_msg))
             # We need to wait a bit for requests to come in. This can probably
             # be set shorter for true asynchronous clients.
             time.sleep(0.05)
     
     def abort_request(self, stream, ident, parent):
         """abort a specifig msg by id"""
         msg_ids = parent['content'].get('msg_ids', None)
         if isinstance(msg_ids, basestring):
             msg_ids = [msg_ids]
         if not msg_ids:
             self.abort_queues()
         for mid in msg_ids:
             self.aborted.add(str(mid))
         
         content = dict(status='ok')
         reply_msg = self.session.send(stream, 'abort_reply', content=content, 
                 parent=parent, ident=ident)[0]
         self.log.debug(str(reply_msg))
     
     def shutdown_request(self, stream, ident, parent):
         """kill ourself.  This should really be handled in an external process"""
         try:
             self.abort_queues()
         except:
             content = self._wrap_exception('shutdown')
         else:
             content = dict(parent['content'])
             content['status'] = 'ok'
         msg = self.session.send(stream, 'shutdown_reply',
                                 content=content, parent=parent, ident=ident)
         # msg = self.session.send(self.pub_socket, 'shutdown_reply',
         #                         content, parent, ident)
         # print >> sys.__stdout__, msg
         # time.sleep(0.2)
         dc = ioloop.DelayedCallback(lambda : sys.exit(0), 1000, self.loop)
         dc.start()
     
     def dispatch_control(self, msg):
         idents,msg = self.session.feed_identities(msg, copy=False)
         try:
             msg = self.session.unpack_message(msg, content=True, copy=False)
         except:
             self.log.error("Invalid Message", exc_info=True)
             return
         
         header = msg['header']
         msg_id = header['msg_id']
         
         handler = self.control_handlers.get(msg['msg_type'], None)
         if handler is None:
             self.log.error("UNKNOWN CONTROL MESSAGE TYPE: %r"%msg['msg_type'])
         else:
             handler(self.control_stream, idents, msg)
     
 
     #-------------------- queue helpers ------------------------------
     
     def check_dependencies(self, dependencies):
         if not dependencies:
             return True
         if len(dependencies) == 2 and dependencies[0] in 'any all'.split():
             anyorall = dependencies[0]
             dependencies = dependencies[1]
         else:
             anyorall = 'all'
         results = self.client.get_results(dependencies,status_only=True)
         if results['status'] != 'ok':
             return False
         
         if anyorall == 'any':
             if not results['completed']:
                 return False
         else:
             if results['pending']:
                 return False
         
         return True
     
     def check_aborted(self, msg_id):
         return msg_id in self.aborted
     
     #-------------------- queue handlers -----------------------------
     
     def clear_request(self, stream, idents, parent):
         """Clear our namespace."""
         self.user_ns = {}
         msg = self.session.send(stream, 'clear_reply', ident=idents, parent=parent, 
                 content = dict(status='ok'))
         self._initial_exec_lines()
     
     def execute_request(self, stream, ident, parent):
         self.log.debug('execute request %s'%parent)
         try:
             code = parent[u'content'][u'code']
         except:
             self.log.error("Got bad msg: %s"%parent, exc_info=True)
             return
         self.session.send(self.iopub_stream, u'pyin', {u'code':code},parent=parent,
                             ident='%s.pyin'%self.prefix)
         started = datetime.now().strftime(ISO8601)
         try:
             comp_code = self.compiler(code, '<zmq-kernel>')
             # allow for not overriding displayhook
             if hasattr(sys.displayhook, 'set_parent'):
                 sys.displayhook.set_parent(parent)
                 sys.stdout.set_parent(parent)
                 sys.stderr.set_parent(parent)
             exec comp_code in self.user_ns, self.user_ns
         except:
             exc_content = self._wrap_exception('execute')
             # exc_msg = self.session.msg(u'pyerr', exc_content, parent)
             self.session.send(self.iopub_stream, u'pyerr', exc_content, parent=parent,
                             ident='%s.pyerr'%self.prefix)
             reply_content = exc_content
         else:
             reply_content = {'status' : 'ok'}
         
         reply_msg = self.session.send(stream, u'execute_reply', reply_content, parent=parent, 
                     ident=ident, subheader = dict(started=started))
         self.log.debug(str(reply_msg))
         if reply_msg['content']['status'] == u'error':
             self.abort_queues()
 
     def complete_request(self, stream, ident, parent):
         matches = {'matches' : self.complete(parent),
                    'status' : 'ok'}
         completion_msg = self.session.send(stream, 'complete_reply',
                                            matches, parent, ident)
         # print >> sys.__stdout__, completion_msg
 
     def complete(self, msg):
         return self.completer.complete(msg.content.line, msg.content.text)
     
     def apply_request(self, stream, ident, parent):
         # flush previous reply, so this request won't block it
         stream.flush(zmq.POLLOUT)
         
         try:
             content = parent[u'content']
             bufs = parent[u'buffers']
             msg_id = parent['header']['msg_id']
             bound = content.get('bound', False)
         except:
             self.log.error("Got bad msg: %s"%parent, exc_info=True)
             return
         # pyin_msg = self.session.msg(u'pyin',{u'code':code}, parent=parent)
         # self.iopub_stream.send(pyin_msg)
         # self.session.send(self.iopub_stream, u'pyin', {u'code':code},parent=parent)
         sub = {'dependencies_met' : True, 'engine' : self.ident,
                 'started': datetime.now().strftime(ISO8601)}
         try:
             # allow for not overriding displayhook
             if hasattr(sys.displayhook, 'set_parent'):
                 sys.displayhook.set_parent(parent)
                 sys.stdout.set_parent(parent)
                 sys.stderr.set_parent(parent)
             # exec "f(*args,**kwargs)" in self.user_ns, self.user_ns
             if bound:
                 working = self.user_ns
                 suffix = str(msg_id).replace("-","")
                 prefix = "_"
                 
             else:
                 working = dict()
                 suffix = prefix = "_" # prevent keyword collisions with lambda
             f,args,kwargs = unpack_apply_message(bufs, working, copy=False)
             # if f.fun
             fname = getattr(f, '__name__', 'f')
             
             fname = prefix+fname.strip('<>')+suffix
             argname = prefix+"args"+suffix
             kwargname = prefix+"kwargs"+suffix
             resultname = prefix+"result"+suffix
             
             ns = { fname : f, argname : args, kwargname : kwargs }
             # print ns
             working.update(ns)
             code = "%s=%s(*%s,**%s)"%(resultname, fname, argname, kwargname)
             exec code in working, working
             result = working.get(resultname)
             # clear the namespace
             if bound:
                 for key in ns.iterkeys():
                     self.user_ns.pop(key)
             else:
                 del working
             
             packed_result,buf = serialize_object(result)
             result_buf = [packed_result]+buf
         except:
             exc_content = self._wrap_exception('apply')
             # exc_msg = self.session.msg(u'pyerr', exc_content, parent)
             self.session.send(self.iopub_stream, u'pyerr', exc_content, parent=parent,
                                 ident='%s.pyerr'%self.prefix)
             reply_content = exc_content
             result_buf = []
             
             if exc_content['ename'] == 'UnmetDependency':
                 sub['dependencies_met'] = False
         else:
             reply_content = {'status' : 'ok'}
         
         # put 'ok'/'error' status in header, for scheduler introspection:
         sub['status'] = reply_content['status']
         
         reply_msg = self.session.send(stream, u'apply_reply', reply_content, 
                     parent=parent, ident=ident,buffers=result_buf, subheader=sub)
 
         # if reply_msg['content']['status'] == u'error':
         #     self.abort_queues()
     
     def dispatch_queue(self, stream, msg):
         self.control_stream.flush()
         idents,msg = self.session.feed_identities(msg, copy=False)
         try:
             msg = self.session.unpack_message(msg, content=True, copy=False)
         except:
             self.log.error("Invalid Message", exc_info=True)
             return
             
         
         header = msg['header']
         msg_id = header['msg_id']
         if self.check_aborted(msg_id):
             self.aborted.remove(msg_id)
             # is it safe to assume a msg_id will not be resubmitted?
             reply_type = msg['msg_type'].split('_')[0] + '_reply'
             reply_msg = self.session.send(stream, reply_type, 
                         content={'status' : 'aborted'}, parent=msg, ident=idents)
             return
         handler = self.shell_handlers.get(msg['msg_type'], None)
         if handler is None:
             self.log.error("UNKNOWN MESSAGE TYPE: %r"%msg['msg_type'])
         else:
             handler(stream, idents, msg)
     
     def start(self):
         #### stream mode:
         if self.control_stream:
             self.control_stream.on_recv(self.dispatch_control, copy=False)
             self.control_stream.on_err(printer)
         
         def make_dispatcher(stream):
             def dispatcher(msg):
                 return self.dispatch_queue(stream, msg)
             return dispatcher
         
         for s in self.shell_streams:
             s.on_recv(make_dispatcher(s), copy=False)
             s.on_err(printer)
         
         if self.iopub_stream:
             self.iopub_stream.on_err(printer)
         
         #### while True mode:
         # while True:
         #     idle = True
         #     try:
         #         msg = self.shell_stream.socket.recv_multipart(
         #                     zmq.NOBLOCK, copy=False)
         #     except zmq.ZMQError, e:
         #         if e.errno != zmq.EAGAIN:
         #             raise e
         #     else:
         #         idle=False
         #         self.dispatch_queue(self.shell_stream, msg)
         #             
         #     if not self.task_stream.empty():
         #         idle=False
         #         msg = self.task_stream.recv_multipart()
         #         self.dispatch_queue(self.task_stream, msg)
         #     if idle:
         #         # don't busywait
         #         time.sleep(1e-3)
 
 def make_kernel(int_id, identity, control_addr, shell_addrs, iopub_addr, hb_addrs, 
                 client_addr=None, loop=None, context=None, key=None,
                 out_stream_factory=OutStream, display_hook_factory=DisplayHook):
     """NO LONGER IN USE"""
     # create loop, context, and session:
     if loop is None:
         loop = ioloop.IOLoop.instance()
     if context is None:
         context = zmq.Context()
     c = context
     session = StreamSession(key=key)
     # print (session.key)
     # print (control_addr, shell_addrs, iopub_addr, hb_addrs)
     
     # create Control Stream
     control_stream = zmqstream.ZMQStream(c.socket(zmq.PAIR), loop)
     control_stream.setsockopt(zmq.IDENTITY, identity)
     control_stream.connect(control_addr)
     
     # create Shell Streams (MUX, Task, etc.):
     shell_streams = []
     for addr in shell_addrs:
         stream = zmqstream.ZMQStream(c.socket(zmq.PAIR), loop)
         stream.setsockopt(zmq.IDENTITY, identity)
         stream.connect(addr)
         shell_streams.append(stream)
     
     # create iopub stream:
     iopub_stream = zmqstream.ZMQStream(c.socket(zmq.PUB), loop)
     iopub_stream.setsockopt(zmq.IDENTITY, identity)
     iopub_stream.connect(iopub_addr)
     
     # Redirect input streams and set a display hook.
     if out_stream_factory:
         sys.stdout = out_stream_factory(session, iopub_stream, u'stdout')
         sys.stdout.topic = 'engine.%i.stdout'%int_id
         sys.stderr = out_stream_factory(session, iopub_stream, u'stderr')
         sys.stderr.topic = 'engine.%i.stderr'%int_id
     if display_hook_factory:
         sys.displayhook = display_hook_factory(session, iopub_stream)
         sys.displayhook.topic = 'engine.%i.pyout'%int_id
     
     
     # launch heartbeat
     heart = heartmonitor.Heart(*map(str, hb_addrs), heart_id=identity)
     heart.start()
     
     # create (optional) Client
     if client_addr:
         client = Client(client_addr, username=identity)
     else:
         client = None
     
     kernel = Kernel(id=int_id, session=session, control_stream=control_stream,
             shell_streams=shell_streams, iopub_stream=iopub_stream, 
             client=client, loop=loop)
     kernel.start()
     return loop, c, kernel
 

IPython/zmq/parallel/streamsession.py

@@ -1,542 +1,377 b''
 #!/usr/bin/env python
 """edited session.py to work with streams, and move msg_type to the header
 """
 
 
 import os
 import pprint
-import sys
-import traceback
 import uuid
 from datetime import datetime
 
 try:
     import cPickle
     pickle = cPickle
 except:
     cPickle = None
     import pickle
 
 import zmq
 from zmq.utils import jsonapi
 from zmq.eventloop.zmqstream import ZMQStream
 
-from IPython.utils.pickleutil import can, uncan, canSequence, uncanSequence
-from IPython.utils.newserialized import serialize, unserialize
-
-from .error import RemoteError
+from .util import ISO8601
 
 # packer priority: jsonlib[2], cPickle, simplejson/json, pickle
 json_name = '' if not jsonapi.jsonmod else jsonapi.jsonmod.__name__
 if json_name in ('jsonlib', 'jsonlib2'):
     use_json = True
 elif json_name:
     if cPickle is None:
         use_json = True
     else:
         use_json = False
 else:
     use_json = False
 
 def squash_unicode(obj):
     if isinstance(obj,dict):
         for key in obj.keys():
             obj[key] = squash_unicode(obj[key])
             if isinstance(key, unicode):
                 obj[squash_unicode(key)] = obj.pop(key)
     elif isinstance(obj, list):
         for i,v in enumerate(obj):
             obj[i] = squash_unicode(v)
     elif isinstance(obj, unicode):
         obj = obj.encode('utf8')
     return obj
 
 json_packer = jsonapi.dumps
 json_unpacker = lambda s: squash_unicode(jsonapi.loads(s))
 
 pickle_packer = lambda o: pickle.dumps(o,-1)
 pickle_unpacker = pickle.loads
 
 if use_json:
     default_packer = json_packer
     default_unpacker = json_unpacker
 else:
     default_packer = pickle_packer
     default_unpacker = pickle_unpacker
 
 
 DELIM="<IDS|MSG>"
-ISO8601="%Y-%m-%dT%H:%M:%S.%f"
-
-def wrap_exception(engine_info={}):
-    etype, evalue, tb = sys.exc_info()
-    stb = traceback.format_exception(etype, evalue, tb)
-    exc_content = {
-        'status' : 'error',
-        'traceback' : stb,
-        'ename' : unicode(etype.__name__),
-        'evalue' : unicode(evalue),
-        'engine_info' : engine_info
-    }
-    return exc_content
-
-def unwrap_exception(content):
-    err = RemoteError(content['ename'], content['evalue'], 
-                ''.join(content['traceback']),
-                content.get('engine_info', {}))
-    return err
-    
 
 class Message(object):
     """A simple message object that maps dict keys to attributes.
 
     A Message can be created from a dict and a dict from a Message instance
     simply by calling dict(msg_obj)."""
     
     def __init__(self, msg_dict):
         dct = self.__dict__
         for k, v in dict(msg_dict).iteritems():
             if isinstance(v, dict):
                 v = Message(v)
             dct[k] = v
 
     # Having this iterator lets dict(msg_obj) work out of the box.
     def __iter__(self):
         return iter(self.__dict__.iteritems())
     
     def __repr__(self):
         return repr(self.__dict__)
 
     def __str__(self):
         return pprint.pformat(self.__dict__)
 
     def __contains__(self, k):
         return k in self.__dict__
 
     def __getitem__(self, k):
         return self.__dict__[k]
 
 
 def msg_header(msg_id, msg_type, username, session):
     date=datetime.now().strftime(ISO8601)
     return locals()
 
 def extract_header(msg_or_header):
     """Given a message or header, return the header."""
     if not msg_or_header:
         return {}
     try:
         # See if msg_or_header is the entire message.
         h = msg_or_header['header']
     except KeyError:
         try:
             # See if msg_or_header is just the header
             h = msg_or_header['msg_id']
         except KeyError:
             raise
         else:
             h = msg_or_header
     if not isinstance(h, dict):
         h = dict(h)
     return h
 
-def rekey(dikt):
-    """Rekey a dict that has been forced to use str keys where there should be
-    ints by json.  This belongs in the jsonutil added by fperez."""
-    for k in dikt.iterkeys():
-        if isinstance(k, str):
-            ik=fk=None
-            try:
-                ik = int(k)
-            except ValueError:
-                try:
-                    fk = float(k)
-                except ValueError:
-                    continue
-            if ik is not None:
-                nk = ik
-            else:
-                nk = fk
-            if nk in dikt:
-                raise KeyError("already have key %r"%nk)
-            dikt[nk] = dikt.pop(k)
-    return dikt
-
-def serialize_object(obj, threshold=64e-6):
-    """Serialize an object into a list of sendable buffers.
-    
-    Parameters
-    ----------
-    
-    obj : object
-        The object to be serialized
-    threshold : float
-        The threshold for not double-pickling the content.
-        
-    
-    Returns
-    -------
-    ('pmd', [bufs]) :
-        where pmd is the pickled metadata wrapper,
-        bufs is a list of data buffers
-    """
-    databuffers = []
-    if isinstance(obj, (list, tuple)):
-        clist = canSequence(obj)
-        slist = map(serialize, clist)
-        for s in slist:
-            if s.typeDescriptor in ('buffer', 'ndarray') or s.getDataSize() > threshold:
-                databuffers.append(s.getData())
-                s.data = None
-        return pickle.dumps(slist,-1), databuffers
-    elif isinstance(obj, dict):
-        sobj = {}
-        for k in sorted(obj.iterkeys()):
-            s = serialize(can(obj[k]))
-            if s.typeDescriptor in ('buffer', 'ndarray') or s.getDataSize() > threshold:
-                databuffers.append(s.getData())
-                s.data = None
-            sobj[k] = s
-        return pickle.dumps(sobj,-1),databuffers
-    else:
-        s = serialize(can(obj))
-        if s.typeDescriptor in ('buffer', 'ndarray') or s.getDataSize() > threshold:
-            databuffers.append(s.getData())
-            s.data = None
-        return pickle.dumps(s,-1),databuffers
-            
-        
-def unserialize_object(bufs):
-    """reconstruct an object serialized by serialize_object from data buffers."""
-    bufs = list(bufs)
-    sobj = pickle.loads(bufs.pop(0))
-    if isinstance(sobj, (list, tuple)):
-        for s in sobj:
-            if s.data is None:
-                s.data = bufs.pop(0)
-        return uncanSequence(map(unserialize, sobj)), bufs
-    elif isinstance(sobj, dict):
-        newobj = {}
-        for k in sorted(sobj.iterkeys()):
-            s = sobj[k]
-            if s.data is None:
-                s.data = bufs.pop(0)
-            newobj[k] = uncan(unserialize(s))
-        return newobj, bufs
-    else:
-        if sobj.data is None:
-            sobj.data = bufs.pop(0)
-        return uncan(unserialize(sobj)), bufs
-
-def pack_apply_message(f, args, kwargs, threshold=64e-6):
-    """pack up a function, args, and kwargs to be sent over the wire
-    as a series of buffers. Any object whose data is larger than `threshold`
-    will not have their data copied (currently only numpy arrays support zero-copy)"""
-    msg = [pickle.dumps(can(f),-1)]
-    databuffers = [] # for large objects
-    sargs, bufs = serialize_object(args,threshold)
-    msg.append(sargs)
-    databuffers.extend(bufs)
-    skwargs, bufs = serialize_object(kwargs,threshold)
-    msg.append(skwargs)
-    databuffers.extend(bufs)
-    msg.extend(databuffers)
-    return msg
-
-def unpack_apply_message(bufs, g=None, copy=True):
-    """unpack f,args,kwargs from buffers packed by pack_apply_message()
-    Returns: original f,args,kwargs"""
-    bufs = list(bufs) # allow us to pop
-    assert len(bufs) >= 3, "not enough buffers!"
-    if not copy:
-        for i in range(3):
-            bufs[i] = bufs[i].bytes
-    cf = pickle.loads(bufs.pop(0))
-    sargs = list(pickle.loads(bufs.pop(0)))
-    skwargs = dict(pickle.loads(bufs.pop(0)))
-    # print sargs, skwargs
-    f = uncan(cf, g)
-    for sa in sargs:
-        if sa.data is None:
-            m = bufs.pop(0)
-            if sa.getTypeDescriptor() in ('buffer', 'ndarray'):
-                if copy:
-                    sa.data = buffer(m)
-                else:
-                    sa.data = m.buffer
-            else:
-                if copy:
-                    sa.data = m
-                else:
-                    sa.data = m.bytes
-    
-    args = uncanSequence(map(unserialize, sargs), g)
-    kwargs = {}
-    for k in sorted(skwargs.iterkeys()):
-        sa = skwargs[k]
-        if sa.data is None:
-            sa.data = bufs.pop(0)
-        kwargs[k] = uncan(unserialize(sa), g)
-    
-    return f,args,kwargs
-
 class StreamSession(object):
     """tweaked version of IPython.zmq.session.Session, for development in Parallel"""
     debug=False
     key=None
     
     def __init__(self, username=None, session=None, packer=None, unpacker=None, key=None, keyfile=None):
         if username is None:
             username = os.environ.get('USER','username')
         self.username = username
         if session is None:
             self.session = str(uuid.uuid4())
         else:
             self.session = session
         self.msg_id = str(uuid.uuid4())
         if packer is None:
             self.pack = default_packer
         else:
             if not callable(packer):
                 raise TypeError("packer must be callable, not %s"%type(packer))
             self.pack = packer
         
         if unpacker is None:
             self.unpack = default_unpacker
         else:
             if not callable(unpacker):
                 raise TypeError("unpacker must be callable, not %s"%type(unpacker))
             self.unpack = unpacker
         
         if key is not None and keyfile is not None:
             raise TypeError("Must specify key OR keyfile, not both")
         if keyfile is not None:
             with open(keyfile) as f:
                 self.key = f.read().strip()
         else:
             self.key = key
         if isinstance(self.key, unicode):
             self.key = self.key.encode('utf8')
         # print key, keyfile, self.key
         self.none = self.pack({})
             
     def msg_header(self, msg_type):
         h = msg_header(self.msg_id, msg_type, self.username, self.session)
         self.msg_id = str(uuid.uuid4())
         return h
 
     def msg(self, msg_type, content=None, parent=None, subheader=None):
         msg = {}
         msg['header'] = self.msg_header(msg_type)
         msg['msg_id'] = msg['header']['msg_id']
         msg['parent_header'] = {} if parent is None else extract_header(parent)
         msg['msg_type'] = msg_type
         msg['content'] = {} if content is None else content
         sub = {} if subheader is None else subheader
         msg['header'].update(sub)
         return msg
 
     def check_key(self, msg_or_header):
         """Check that a message's header has the right key"""
         if self.key is None:
             return True
         header = extract_header(msg_or_header)
         return header.get('key', None) == self.key
             
         
     def send(self, stream, msg_or_type, content=None, buffers=None, parent=None, subheader=None, ident=None):
         """Build and send a message via stream or socket.
         
         Parameters
         ----------
         
         stream : zmq.Socket or ZMQStream
             the socket-like object used to send the data
         msg_or_type : str or Message/dict
             Normally, msg_or_type will be a msg_type unless a message is being sent more
             than once.
         
         Returns
         -------
         (msg,sent) : tuple
             msg : Message
                 the nice wrapped dict-like object containing the headers
             
         """
         if isinstance(msg_or_type, (Message, dict)):
             # we got a Message, not a msg_type
             # don't build a new Message
             msg = msg_or_type
             content = msg['content']
         else:
             msg = self.msg(msg_or_type, content, parent, subheader)
         buffers = [] if buffers is None else buffers
         to_send = []
         if isinstance(ident, list):
             # accept list of idents
             to_send.extend(ident)
         elif ident is not None:
             to_send.append(ident)
         to_send.append(DELIM)
         if self.key is not None:
             to_send.append(self.key)
         to_send.append(self.pack(msg['header']))
         to_send.append(self.pack(msg['parent_header']))
         
         if content is None:
             content = self.none
         elif isinstance(content, dict):
             content = self.pack(content)
         elif isinstance(content, str):
             # content is already packed, as in a relayed message
             pass
         else:
             raise TypeError("Content incorrect type: %s"%type(content))
         to_send.append(content)
         flag = 0
         if buffers:
             flag = zmq.SNDMORE
         stream.send_multipart(to_send, flag, copy=False)
         for b in buffers[:-1]:
             stream.send(b, flag, copy=False)
         if buffers:
             stream.send(buffers[-1], copy=False)
         # omsg = Message(msg)
         if self.debug:
             pprint.pprint(msg)
             pprint.pprint(to_send)
             pprint.pprint(buffers)
         return msg
     
     def send_raw(self, stream, msg, flags=0, copy=True, ident=None):
         """Send a raw message via ident path.
         
         Parameters
         ----------
         msg : list of sendable buffers"""
         to_send = []
         if isinstance(ident, str):
             ident = [ident]
         if ident is not None:
             to_send.extend(ident)
         to_send.append(DELIM)
         if self.key is not None:
             to_send.append(self.key)
         to_send.extend(msg)
         stream.send_multipart(msg, flags, copy=copy)
     
     def recv(self, socket, mode=zmq.NOBLOCK, content=True, copy=True):
         """receives and unpacks a message
         returns [idents], msg"""
         if isinstance(socket, ZMQStream):
             socket = socket.socket
         try:
             msg = socket.recv_multipart(mode)
         except zmq.ZMQError as e:
             if e.errno == zmq.EAGAIN:
                 # We can convert EAGAIN to None as we know in this case
                 # recv_multipart won't return None.
                 return None
             else:
                 raise
         # return an actual Message object
         # determine the number of idents by trying to unpack them.
         # this is terrible:
         idents, msg = self.feed_identities(msg, copy)
         try:
             return idents, self.unpack_message(msg, content=content, copy=copy)
         except Exception as e:
             print (idents, msg)
             # TODO: handle it
             raise e
     
     def feed_identities(self, msg, copy=True):
         """feed until DELIM is reached, then return the prefix as idents and remainder as
         msg. This is easily broken by setting an IDENT to DELIM, but that would be silly.
         
         Parameters
         ----------
         msg : a list of Message or bytes objects
             the message to be split
         copy : bool
             flag determining whether the arguments are bytes or Messages
         
         Returns
         -------
         (idents,msg) : two lists
             idents will always be a list of bytes - the indentity prefix
             msg will be a list of bytes or Messages, unchanged from input
             msg should be unpackable via self.unpack_message at this point.
         """
         ikey = int(self.key is not None)
         minlen = 3 + ikey
         msg = list(msg)
         idents = []
         while len(msg) > minlen:
             if copy:
                 s = msg[0]
             else:
                 s = msg[0].bytes
             if s == DELIM:
                 msg.pop(0)
                 break
             else:
                 idents.append(s)
                 msg.pop(0)
                 
         return idents, msg
     
     def unpack_message(self, msg, content=True, copy=True):
         """Return a message object from the format
         sent by self.send.
         
         Parameters:
         -----------
         
         content : bool (True)
             whether to unpack the content dict (True), 
             or leave it serialized (False)
         
         copy : bool (True)
             whether to return the bytes (True), 
             or the non-copying Message object in each place (False)
         
         """
         ikey = int(self.key is not None)
         minlen = 3 + ikey
         message = {}
         if not copy:
             for i in range(minlen):
                 msg[i] = msg[i].bytes
         if ikey:
             if not self.key == msg[0]:
                 raise KeyError("Invalid Session Key: %s"%msg[0])
         if not len(msg) >= minlen:
             raise TypeError("malformed message, must have at least %i elements"%minlen)
         message['header'] = self.unpack(msg[ikey+0])
         message['msg_type'] = message['header']['msg_type']
         message['parent_header'] = self.unpack(msg[ikey+1])
         if content:
             message['content'] = self.unpack(msg[ikey+2])
         else:
             message['content'] = msg[ikey+2]
         
         message['buffers'] = msg[ikey+3:]# [ m.buffer for m in msg[3:] ]
         return message
         
 
 def test_msg2obj():
     am = dict(x=1)
     ao = Message(am)
     assert ao.x == am['x']
 
     am['y'] = dict(z=1)
     ao = Message(am)
     assert ao.y.z == am['y']['z']
     
     k1, k2 = 'y', 'z'
     assert ao[k1][k2] == am[k1][k2]
     
     am2 = dict(ao)
     assert am['x'] == am2['x']
     assert am['y']['z'] == am2['y']['z']

IPython/zmq/parallel/tests/test_streamsession.py

@@ -1,82 +1,82 b''
 
 import os
 import uuid
 import zmq
 
 from zmq.tests import BaseZMQTestCase
 
 # from IPython.zmq.tests import SessionTestCase
 from IPython.zmq.parallel import streamsession as ss
 
 class SessionTestCase(BaseZMQTestCase):
     
     def setUp(self):
         BaseZMQTestCase.setUp(self)
         self.session = ss.StreamSession()
 
 class TestSession(SessionTestCase):
     
     def test_msg(self):
         """message format"""
         msg = self.session.msg('execute')
         thekeys = set('header msg_id parent_header msg_type content'.split())
         s = set(msg.keys())
         self.assertEquals(s, thekeys)
         self.assertTrue(isinstance(msg['content'],dict))
         self.assertTrue(isinstance(msg['header'],dict))
         self.assertTrue(isinstance(msg['parent_header'],dict))
         self.assertEquals(msg['msg_type'], 'execute')
         
             
     
     def test_args(self):
         """initialization arguments for StreamSession"""
         s = ss.StreamSession()
         self.assertTrue(s.pack is ss.default_packer)
         self.assertTrue(s.unpack is ss.default_unpacker)
         self.assertEquals(s.username, os.environ.get('USER', 'username'))
         
         s = ss.StreamSession(username=None)
         self.assertEquals(s.username, os.environ.get('USER', 'username'))
         
         self.assertRaises(TypeError, ss.StreamSession, packer='hi')
         self.assertRaises(TypeError, ss.StreamSession, unpacker='hi')
         u = str(uuid.uuid4())
         s = ss.StreamSession(username='carrot', session=u)
         self.assertEquals(s.session, u)
         self.assertEquals(s.username, 'carrot')
         
     
-    def test_rekey(self):
-        """rekeying dict around json str keys"""
-        d = {'0': uuid.uuid4(), 0:uuid.uuid4()}
-        self.assertRaises(KeyError, ss.rekey, d)
-        
-        d = {'0': uuid.uuid4(), 1:uuid.uuid4(), 'asdf':uuid.uuid4()}
-        d2 = {0:d['0'],1:d[1],'asdf':d['asdf']}
-        rd = ss.rekey(d)
-        self.assertEquals(d2,rd)
-        
-        d = {'1.5':uuid.uuid4(),'1':uuid.uuid4()}
-        d2 = {1.5:d['1.5'],1:d['1']}
-        rd = ss.rekey(d)
-        self.assertEquals(d2,rd)
-        
-        d = {'1.0':uuid.uuid4(),'1':uuid.uuid4()}
-        self.assertRaises(KeyError, ss.rekey, d)
-    
+    # def test_rekey(self):
+    #     """rekeying dict around json str keys"""
+    #     d = {'0': uuid.uuid4(), 0:uuid.uuid4()}
+    #     self.assertRaises(KeyError, ss.rekey, d)
+    #     
+    #     d = {'0': uuid.uuid4(), 1:uuid.uuid4(), 'asdf':uuid.uuid4()}
+    #     d2 = {0:d['0'],1:d[1],'asdf':d['asdf']}
+    #     rd = ss.rekey(d)
+    #     self.assertEquals(d2,rd)
+    #     
+    #     d = {'1.5':uuid.uuid4(),'1':uuid.uuid4()}
+    #     d2 = {1.5:d['1.5'],1:d['1']}
+    #     rd = ss.rekey(d)
+    #     self.assertEquals(d2,rd)
+    #     
+    #     d = {'1.0':uuid.uuid4(),'1':uuid.uuid4()}
+    #     self.assertRaises(KeyError, ss.rekey, d)
+    # 
     def test_unique_msg_ids(self):
         """test that messages receive unique ids"""
         ids = set()
         for i in range(2**12):
             h = self.session.msg_header('test')
             msg_id = h['msg_id']
             self.assertTrue(msg_id not in ids)
             ids.add(msg_id)
     
     def test_feed_identities(self):
         """scrub the front for zmq IDENTITIES"""
         theids = "engine client other".split()
         content = dict(code='whoda',stuff=object())
         themsg = self.session.msg('execute',content=content)
         pmsg = theids

IPython/zmq/parallel/util.py

@@ -1,119 +1,271 b''
-"""some generic utilities"""
+"""some generic utilities for dealing with classes, urls, and serialization"""
 import re
 import socket
 
+try:
+    import cPickle
+    pickle = cPickle
+except:
+    cPickle = None
+    import pickle
+
+
+from IPython.utils.pickleutil import can, uncan, canSequence, uncanSequence
+from IPython.utils.newserialized import serialize, unserialize
+
+ISO8601="%Y-%m-%dT%H:%M:%S.%f"
+
 class ReverseDict(dict):
     """simple double-keyed subset of dict methods."""
     
     def __init__(self, *args, **kwargs):
         dict.__init__(self, *args, **kwargs)
         self._reverse = dict()
         for key, value in self.iteritems():
             self._reverse[value] = key
     
     def __getitem__(self, key):
         try:
             return dict.__getitem__(self, key)
         except KeyError:
             return self._reverse[key]
     
     def __setitem__(self, key, value):
         if key in self._reverse:
             raise KeyError("Can't have key %r on both sides!"%key)
         dict.__setitem__(self, key, value)
         self._reverse[value] = key
     
     def pop(self, key):
         value = dict.pop(self, key)
         self._reverse.pop(value)
         return value
     
     def get(self, key, default=None):
         try:
             return self[key]
         except KeyError:
             return default
-    
 
 def validate_url(url):
     """validate a url for zeromq"""
     if not isinstance(url, basestring):
         raise TypeError("url must be a string, not %r"%type(url))
     url = url.lower()
     
     proto_addr = url.split('://')
     assert len(proto_addr) == 2, 'Invalid url: %r'%url
     proto, addr = proto_addr
     assert proto in ['tcp','pgm','epgm','ipc','inproc'], "Invalid protocol: %r"%proto
     
     # domain pattern adapted from http://www.regexlib.com/REDetails.aspx?regexp_id=391
     # author: Remi Sabourin
     pat = re.compile(r'^([\w\d]([\w\d\-]{0,61}[\w\d])?\.)*[\w\d]([\w\d\-]{0,61}[\w\d])?$')
     
     if proto == 'tcp':
         lis = addr.split(':')
         assert len(lis) == 2, 'Invalid url: %r'%url
         addr,s_port = lis
         try:
             port = int(s_port)
         except ValueError:
             raise AssertionError("Invalid port %r in url: %r"%(port, url))
         
         assert addr == '*' or pat.match(addr) is not None, 'Invalid url: %r'%url
         
     else:
         # only validate tcp urls currently
         pass
     
     return True
 
 
 def validate_url_container(container):
     """validate a potentially nested collection of urls."""
     if isinstance(container, basestring):
         url = container
         return validate_url(url)
     elif isinstance(container, dict):
         container = container.itervalues()
     
     for element in container:
         validate_url_container(element)
 
 
 def split_url(url):
     """split a zmq url (tcp://ip:port) into ('tcp','ip','port')."""
     proto_addr = url.split('://')
     assert len(proto_addr) == 2, 'Invalid url: %r'%url
     proto, addr = proto_addr
     lis = addr.split(':')
     assert len(lis) == 2, 'Invalid url: %r'%url
     addr,s_port = lis
     return proto,addr,s_port
     
 def disambiguate_ip_address(ip, location=None):
     """turn multi-ip interfaces '0.0.0.0' and '*' into connectable
     ones, based on the location (default interpretation of location is localhost)."""
     if ip in ('0.0.0.0', '*'):
         external_ips = socket.gethostbyname_ex(socket.gethostname())[2]
         if location is None or location in external_ips:
             ip='127.0.0.1'
         elif location:
             return location
     return ip
 
 def disambiguate_url(url, location=None):
     """turn multi-ip interfaces '0.0.0.0' and '*' into connectable
     ones, based on the location (default interpretation is localhost).
     
     This is for zeromq urls, such as tcp://*:10101."""
     try:
         proto,ip,port = split_url(url)
     except AssertionError:
         # probably not tcp url; could be ipc, etc.
         return url
     
     ip = disambiguate_ip_address(ip,location)
     
     return "%s://%s:%s"%(proto,ip,port)
 
 
+def rekey(dikt):
+    """Rekey a dict that has been forced to use str keys where there should be
+    ints by json.  This belongs in the jsonutil added by fperez."""
+    for k in dikt.iterkeys():
+        if isinstance(k, str):
+            ik=fk=None
+            try:
+                ik = int(k)
+            except ValueError:
+                try:
+                    fk = float(k)
+                except ValueError:
+                    continue
+            if ik is not None:
+                nk = ik
+            else:
+                nk = fk
+            if nk in dikt:
+                raise KeyError("already have key %r"%nk)
+            dikt[nk] = dikt.pop(k)
+    return dikt
+
+def serialize_object(obj, threshold=64e-6):
+    """Serialize an object into a list of sendable buffers.
+    
+    Parameters
+    ----------
+    
+    obj : object
+        The object to be serialized
+    threshold : float
+        The threshold for not double-pickling the content.
+        
+    
+    Returns
+    -------
+    ('pmd', [bufs]) :
+        where pmd is the pickled metadata wrapper,
+        bufs is a list of data buffers
+    """
+    databuffers = []
+    if isinstance(obj, (list, tuple)):
+        clist = canSequence(obj)
+        slist = map(serialize, clist)
+        for s in slist:
+            if s.typeDescriptor in ('buffer', 'ndarray') or s.getDataSize() > threshold:
+                databuffers.append(s.getData())
+                s.data = None
+        return pickle.dumps(slist,-1), databuffers
+    elif isinstance(obj, dict):
+        sobj = {}
+        for k in sorted(obj.iterkeys()):
+            s = serialize(can(obj[k]))
+            if s.typeDescriptor in ('buffer', 'ndarray') or s.getDataSize() > threshold:
+                databuffers.append(s.getData())
+                s.data = None
+            sobj[k] = s
+        return pickle.dumps(sobj,-1),databuffers
+    else:
+        s = serialize(can(obj))
+        if s.typeDescriptor in ('buffer', 'ndarray') or s.getDataSize() > threshold:
+            databuffers.append(s.getData())
+            s.data = None
+        return pickle.dumps(s,-1),databuffers
+            
+        
+def unserialize_object(bufs):
+    """reconstruct an object serialized by serialize_object from data buffers."""
+    bufs = list(bufs)
+    sobj = pickle.loads(bufs.pop(0))
+    if isinstance(sobj, (list, tuple)):
+        for s in sobj:
+            if s.data is None:
+                s.data = bufs.pop(0)
+        return uncanSequence(map(unserialize, sobj)), bufs
+    elif isinstance(sobj, dict):
+        newobj = {}
+        for k in sorted(sobj.iterkeys()):
+            s = sobj[k]
+            if s.data is None:
+                s.data = bufs.pop(0)
+            newobj[k] = uncan(unserialize(s))
+        return newobj, bufs
+    else:
+        if sobj.data is None:
+            sobj.data = bufs.pop(0)
+        return uncan(unserialize(sobj)), bufs
+
+def pack_apply_message(f, args, kwargs, threshold=64e-6):
+    """pack up a function, args, and kwargs to be sent over the wire
+    as a series of buffers. Any object whose data is larger than `threshold`
+    will not have their data copied (currently only numpy arrays support zero-copy)"""
+    msg = [pickle.dumps(can(f),-1)]
+    databuffers = [] # for large objects
+    sargs, bufs = serialize_object(args,threshold)
+    msg.append(sargs)
+    databuffers.extend(bufs)
+    skwargs, bufs = serialize_object(kwargs,threshold)
+    msg.append(skwargs)
+    databuffers.extend(bufs)
+    msg.extend(databuffers)
+    return msg
+
+def unpack_apply_message(bufs, g=None, copy=True):
+    """unpack f,args,kwargs from buffers packed by pack_apply_message()
+    Returns: original f,args,kwargs"""
+    bufs = list(bufs) # allow us to pop
+    assert len(bufs) >= 3, "not enough buffers!"
+    if not copy:
+        for i in range(3):
+            bufs[i] = bufs[i].bytes
+    cf = pickle.loads(bufs.pop(0))
+    sargs = list(pickle.loads(bufs.pop(0)))
+    skwargs = dict(pickle.loads(bufs.pop(0)))
+    # print sargs, skwargs
+    f = uncan(cf, g)
+    for sa in sargs:
+        if sa.data is None:
+            m = bufs.pop(0)
+            if sa.getTypeDescriptor() in ('buffer', 'ndarray'):
+                if copy:
+                    sa.data = buffer(m)
+                else:
+                    sa.data = m.buffer
+            else:
+                if copy:
+                    sa.data = m
+                else:
+                    sa.data = m.bytes
+    
+    args = uncanSequence(map(unserialize, sargs), g)
+    kwargs = {}
+    for k in sorted(skwargs.iterkeys()):
+        sa = skwargs[k]
+        if sa.data is None:
+            sa.data = bufs.pop(0)
+        kwargs[k] = uncan(unserialize(sa), g)
+    
+    return f,args,kwargs
+

IPython/zmq/parallel/view.py

@@ -1,657 +1,658 b''
-"""Views of remote engines"""
+"""Views of remote engines."""
 #-----------------------------------------------------------------------------
 #  Copyright (C) 2010  The IPython Development Team
 #
 #  Distributed under the terms of the BSD License.  The full license is in
 #  the file COPYING, distributed as part of this software.
 #-----------------------------------------------------------------------------
 
 #-----------------------------------------------------------------------------
 # Imports
 #-----------------------------------------------------------------------------
 
 from IPython.testing import decorators as testdec
-from IPython.utils.traitlets import HasTraits, Bool, List, Dict, Set, Int, Instance
+from IPython.utils.traitlets import HasTraits, Any, Bool, List, Dict, Set, Int, Instance
 
 from IPython.external.decorator import decorator
 
 from .asyncresult import AsyncResult
 from .dependency import Dependency
 from .remotefunction import ParallelFunction, parallel, remote
 
 #-----------------------------------------------------------------------------
 # Decorators
 #-----------------------------------------------------------------------------
 
 @decorator
 def myblock(f, self, *args, **kwargs):
     """override client.block with self.block during a call"""
     block = self.client.block
     self.client.block = self.block
     try:
         ret = f(self, *args, **kwargs)
     finally:
         self.client.block = block
     return ret
 
 @decorator
 def save_ids(f, self, *args, **kwargs):
     """Keep our history and outstanding attributes up to date after a method call."""
     n_previous = len(self.client.history)
     ret = f(self, *args, **kwargs)
     nmsgs = len(self.client.history) - n_previous
     msg_ids = self.client.history[-nmsgs:]
     self.history.extend(msg_ids)
     map(self.outstanding.add, msg_ids)
     return ret
 
 @decorator
 def sync_results(f, self, *args, **kwargs):
     """sync relevant results from self.client to our results attribute."""
     ret = f(self, *args, **kwargs)
     delta = self.outstanding.difference(self.client.outstanding)
     completed = self.outstanding.intersection(delta)
     self.outstanding = self.outstanding.difference(completed)
     for msg_id in completed:
         self.results[msg_id] = self.client.results[msg_id]
     return ret
 
 @decorator
 def spin_after(f, self, *args, **kwargs):
     """call spin after the method."""
     ret = f(self, *args, **kwargs)
     self.spin()
     return ret
 
 #-----------------------------------------------------------------------------
 # Classes
 #-----------------------------------------------------------------------------
 
 class View(HasTraits):
     """Base View class for more convenint apply(f,*args,**kwargs) syntax via attributes.
     
     Don't use this class, use subclasses.
     """
     block=Bool(False)
     bound=Bool(False)
     history=List()
     outstanding = Set()
     results = Dict()
     client = Instance('IPython.zmq.parallel.client.Client')
     
     _ntargets = Int(1)
     _balanced = Bool(False)
     _default_names = List(['block', 'bound'])
-    _targets = None
+    _targets = Any()
     
     def __init__(self, client=None, targets=None):
         super(View, self).__init__(client=client)
         self._targets = targets
         self._ntargets = 1 if isinstance(targets, (int,type(None))) else len(targets)
         self.block = client.block
         
         for name in self._default_names:
             setattr(self, name, getattr(self, name, None))
         
         assert not self.__class__ is View, "Don't use base View objects, use subclasses"
         
 
     def __repr__(self):
         strtargets = str(self._targets)
         if len(strtargets) > 16:
             strtargets = strtargets[:12]+'...]'
         return "<%s %s>"%(self.__class__.__name__, strtargets)
 
     @property
     def targets(self):
         return self._targets
 
     @targets.setter
     def targets(self, value):
         raise AttributeError("Cannot set View `targets` after construction!")
     
     @property
     def balanced(self):
         return self._balanced
 
     @balanced.setter
     def balanced(self, value):
         raise AttributeError("Cannot set View `balanced` after construction!")
     
     def _defaults(self, *excludes):
         """return dict of our default attributes, excluding names given."""
         d = dict(balanced=self._balanced, targets=self._targets)
         for name in self._default_names:
             if name not in excludes:
                 d[name] = getattr(self, name)
         return d
     
     def set_flags(self, **kwargs):
         """set my attribute flags by keyword.
         
         A View is a wrapper for the Client's apply method, but
         with attributes that specify keyword arguments, those attributes
         can be set by keyword argument with this method.
         
         Parameters
         ----------
         
         block : bool
             whether to wait for results
         bound : bool
             whether to use the client's namespace
         """
         for key in kwargs:
             if key not in self._default_names:
                 raise KeyError("Invalid name: %r"%key)
         for name in ('block', 'bound'):
             if name in kwargs:
                 setattr(self, name, kwargs[name])
     
     #----------------------------------------------------------------
     # wrappers for client methods:
     #----------------------------------------------------------------
     @sync_results
     def spin(self):
         """spin the client, and sync"""
         self.client.spin()
     
     @sync_results
     @save_ids
     def apply(self, f, *args, **kwargs):
         """calls f(*args, **kwargs) on remote engines, returning the result.
         
         This method does not involve the engine's namespace.
         
         if self.block is False:
             returns msg_id
         else:
             returns actual result of f(*args, **kwargs)
         """
         return self.client.apply(f, args, kwargs, **self._defaults())
 
     @save_ids
     def apply_async(self, f, *args, **kwargs):
         """calls f(*args, **kwargs) on remote engines in a nonblocking manner.
         
         This method does not involve the engine's namespace.
         
         returns msg_id
         """
         d = self._defaults('block', 'bound')
         return self.client.apply(f,args,kwargs, block=False, bound=False, **d)
 
     @spin_after
     @save_ids
     def apply_sync(self, f, *args, **kwargs):
         """calls f(*args, **kwargs) on remote engines in a blocking manner,
          returning the result.
         
         This method does not involve the engine's namespace.
         
         returns: actual result of f(*args, **kwargs)
         """
         d = self._defaults('block', 'bound')
         return self.client.apply(f,args,kwargs, block=True, bound=False, **d)
 
     # @sync_results
     # @save_ids
     # def apply_bound(self, f, *args, **kwargs):
     #     """calls f(*args, **kwargs) bound to engine namespace(s).
     #     
     #     if self.block is False:
     #         returns msg_id
     #     else:
     #         returns actual result of f(*args, **kwargs)
     #     
     #     This method has access to the targets' namespace via globals()
     #     
     #     """
     #     d = self._defaults('bound')
     #     return self.client.apply(f, args, kwargs, bound=True, **d)
     # 
     @sync_results
     @save_ids
     def apply_async_bound(self, f, *args, **kwargs):
         """calls f(*args, **kwargs) bound to engine namespace(s) 
         in a nonblocking manner.
         
         returns: msg_id
         
         This method has access to the targets' namespace via globals()
         
         """
         d = self._defaults('block', 'bound')
         return self.client.apply(f, args, kwargs, block=False, bound=True, **d)
 
     @spin_after
     @save_ids
     def apply_sync_bound(self, f, *args, **kwargs):
         """calls f(*args, **kwargs) bound to engine namespace(s), waiting for the result.
         
         returns: actual result of f(*args, **kwargs)
         
         This method has access to the targets' namespace via globals()
         
         """
         d = self._defaults('block', 'bound')
         return self.client.apply(f, args, kwargs, block=True, bound=True, **d)
     
     def abort(self, jobs=None, block=None):
         """Abort jobs on my engines.
         
         Parameters
         ----------
         
         jobs : None, str, list of strs, optional
             if None: abort all jobs.
             else: abort specific msg_id(s).
         """
         block = block if block is not None else self.block
         return self.client.abort(jobs=jobs, targets=self._targets, block=block)
 
     def queue_status(self, verbose=False):
         """Fetch the Queue status of my engines"""
         return self.client.queue_status(targets=self._targets, verbose=verbose)
     
     def purge_results(self, jobs=[], targets=[]):
         """Instruct the controller to forget specific results."""
         if targets is None or targets == 'all':
             targets = self._targets
         return self.client.purge_results(jobs=jobs, targets=targets)
     
     @spin_after
     def get_result(self, indices_or_msg_ids=None):
         """return one or more results, specified by history index or msg_id.
         
         See client.get_result for details.
         
         """
         
         if indices_or_msg_ids is None:
             indices_or_msg_ids = -1
         if isinstance(indices_or_msg_ids, int):
             indices_or_msg_ids = self.history[indices_or_msg_ids]
         elif isinstance(indices_or_msg_ids, (list,tuple,set)):
             indices_or_msg_ids = list(indices_or_msg_ids)
             for i,index in enumerate(indices_or_msg_ids):
                 if isinstance(index, int):
                     indices_or_msg_ids[i] = self.history[index]
         return self.client.get_result(indices_or_msg_ids)
     
     #-------------------------------------------------------------------
     # Map
     #-------------------------------------------------------------------
     
     def map(self, f, *sequences, **kwargs):
         """override in subclasses"""
         raise NotImplementedError
     
     def map_async(self, f, *sequences, **kwargs):
         """Parallel version of builtin `map`, using this view's engines.
         
         This is equivalent to map(...block=False)
         
         See `self.map` for details.
         """
         if 'block' in kwargs:
             raise TypeError("map_async doesn't take a `block` keyword argument.")
         kwargs['block'] = False
         return self.map(f,*sequences,**kwargs)
     
     def map_sync(self, f, *sequences, **kwargs):
         """Parallel version of builtin `map`, using this view's engines.
         
         This is equivalent to map(...block=True)
         
         See `self.map` for details.
         """
         if 'block' in kwargs:
             raise TypeError("map_sync doesn't take a `block` keyword argument.")
         kwargs['block'] = True
         return self.map(f,*sequences,**kwargs)
     
     def imap(self, f, *sequences, **kwargs):
         """Parallel version of `itertools.imap`.
         
         See `self.map` for details.
         """
         
         return iter(self.map_async(f,*sequences, **kwargs))
     
     #-------------------------------------------------------------------
     # Decorators
     #-------------------------------------------------------------------
     
     def remote(self, bound=True, block=True):
         """Decorator for making a RemoteFunction"""
         return remote(self.client, bound=bound, targets=self._targets, block=block, balanced=self._balanced)
     
     def parallel(self, dist='b', bound=True, block=None):
         """Decorator for making a ParallelFunction"""
         block = self.block if block is None else block
         return parallel(self.client, bound=bound, targets=self._targets, block=block, balanced=self._balanced)
 
 @testdec.skip_doctest
 class DirectView(View):
     """Direct Multiplexer View of one or more engines.
     
     These are created via indexed access to a client:
     
     >>> dv_1 = client[1]
     >>> dv_all = client[:]
     >>> dv_even = client[::2]
     >>> dv_some = client[1:3]
     
     This object provides dictionary access to engine namespaces:
     
     # push a=5:
     >>> dv['a'] = 5 
     # pull 'foo':
     >>> db['foo']
     
     """
     
     def __init__(self, client=None, targets=None):
         super(DirectView, self).__init__(client=client, targets=targets)
         self._balanced = False
     
     @spin_after
     @save_ids
     def map(self, f, *sequences, **kwargs):
         """view.map(f, *sequences, block=self.block, bound=self.bound) => list|AsyncMapResult
         
         Parallel version of builtin `map`, using this View's `targets`.
         
         There will be one task per target, so work will be chunked
         if the sequences are longer than `targets`.  
         
         Results can be iterated as they are ready, but will become available in chunks.
         
         Parameters
         ----------
         
         f : callable
             function to be mapped
         *sequences: one or more sequences of matching length
             the sequences to be distributed and passed to `f`
         block : bool
             whether to wait for the result or not [default self.block]
         bound : bool
             whether to have access to the engines' namespaces [default self.bound]
         
         Returns
         -------
         
         if block=False:
             AsyncMapResult
                 An object like AsyncResult, but which reassembles the sequence of results
                 into a single list. AsyncMapResults can be iterated through before all
                 results are complete.
         else:
             list
                 the result of map(f,*sequences)
         """
         
         block = kwargs.get('block', self.block)
         bound = kwargs.get('bound', self.bound)
         for k in kwargs.keys():
             if k not in ['block', 'bound']:
                 raise TypeError("invalid keyword arg, %r"%k)
         
         assert len(sequences) > 0, "must have some sequences to map onto!"
         pf = ParallelFunction(self.client, f, block=block, bound=bound,
                         targets=self._targets, balanced=False)
         return pf.map(*sequences)
     
     @sync_results
     @save_ids
     def execute(self, code, block=None):
         """execute some code on my targets."""
         
         block = block if block is not None else self.block
         
         return self.client.execute(code, block=block, targets=self._targets)
     
     @sync_results
     @save_ids
     def run(self, fname, block=None):
         """execute the code in a file on my targets."""
         
         block = block if block is not None else self.block
         
         return self.client.run(fname, block=block, targets=self._targets)
     
     def update(self, ns):
         """update remote namespace with dict `ns`"""
         return self.client.push(ns, targets=self._targets, block=self.block)
     
     def push(self, ns, block=None):
         """update remote namespace with dict `ns`"""
         
         block = block if block is not None else self.block
         
         return self.client.push(ns, targets=self._targets, block=block)
 
     def get(self, key_s):
         """get object(s) by `key_s` from remote namespace
         will return one object if it is a key.
         It also takes a list of keys, and will return a list of objects."""
         # block = block if block is not None else self.block
         return self.client.pull(key_s, block=True, targets=self._targets)
     
     @sync_results
     @save_ids
     def pull(self, key_s, block=True):
         """get object(s) by `key_s` from remote namespace
         will return one object if it is a key.
         It also takes a list of keys, and will return a list of objects."""
         block = block if block is not None else self.block
         return self.client.pull(key_s, block=block, targets=self._targets)
     
     def scatter(self, key, seq, dist='b', flatten=False, block=None):
         """
         Partition a Python sequence and send the partitions to a set of engines.
         """
         block = block if block is not None else self.block
         
         return self.client.scatter(key, seq, dist=dist, flatten=flatten,
                     targets=self._targets, block=block)
     
     @sync_results
     @save_ids
     def gather(self, key, dist='b', block=None):
         """
         Gather a partitioned sequence on a set of engines as a single local seq.
         """
         block = block if block is not None else self.block
         
         return self.client.gather(key, dist=dist, targets=self._targets, block=block)
     
     def __getitem__(self, key):
         return self.get(key)
     
     def __setitem__(self,key, value):
         self.update({key:value})
     
     def clear(self, block=False):
         """Clear the remote namespaces on my engines."""
         block = block if block is not None else self.block
         return self.client.clear(targets=self._targets, block=block)
     
     def kill(self, block=True):
         """Kill my engines."""
         block = block if block is not None else self.block
         return self.client.kill(targets=self._targets, block=block)
     
     #----------------------------------------
     # activate for %px,%autopx magics
     #----------------------------------------
     def activate(self):
         """Make this `View` active for parallel magic commands.
         
         IPython has a magic command syntax to work with `MultiEngineClient` objects.
         In a given IPython session there is a single active one.  While
         there can be many `Views` created and used by the user, 
         there is only one active one.  The active `View` is used whenever 
         the magic commands %px and %autopx are used.
         
         The activate() method is called on a given `View` to make it 
         active.  Once this has been done, the magic commands can be used.
         """
         
         try:
             # This is injected into __builtins__.
             ip = get_ipython()
         except NameError:
             print "The IPython parallel magics (%result, %px, %autopx) only work within IPython."
         else:
             pmagic = ip.plugin_manager.get_plugin('parallelmagic')
             if pmagic is not None:
                 pmagic.active_multiengine_client = self
             else:
                 print "You must first load the parallelmagic extension " \
                       "by doing '%load_ext parallelmagic'"
 
 
 @testdec.skip_doctest
 class LoadBalancedView(View):
     """An load-balancing View that only executes via the Task scheduler.
     
     Load-balanced views can be created with the client's `view` method:
     
     >>> v = client.view(balanced=True)
     
     or targets can be specified, to restrict the potential destinations:
     
     >>> v = client.view([1,3],balanced=True)
     
     which would restrict loadbalancing to between engines 1 and 3.
     
     """
     
     _default_names = ['block', 'bound', 'follow', 'after', 'timeout']
     
     def __init__(self, client=None, targets=None):
         super(LoadBalancedView, self).__init__(client=client, targets=targets)
         self._ntargets = 1
         self._balanced = True
     
     def _validate_dependency(self, dep):
         """validate a dependency.
         
         For use in `set_flags`.
         """
         if dep is None or isinstance(dep, (str, AsyncResult, Dependency)):
             return True
         elif isinstance(dep, (list,set, tuple)):
             for d in dep:
                 if not isinstance(d, str, AsyncResult):
                     return False
         elif isinstance(dep, dict):
             if set(dep.keys()) != set(Dependency().as_dict().keys()):
                 return False
             if not isinstance(dep['msg_ids'], list):
                 return False
             for d in dep['msg_ids']:
                 if not isinstance(d, str):
                     return False
         else:
             return False
                 
     def set_flags(self, **kwargs):
         """set my attribute flags by keyword.
         
         A View is a wrapper for the Client's apply method, but with attributes
         that specify keyword arguments, those attributes can be set by keyword
         argument with this method.
         
         Parameters
         ----------
         
         block : bool
             whether to wait for results
         bound : bool
             whether to use the engine's namespace
         follow : Dependency, list, msg_id, AsyncResult
             the location dependencies of tasks
         after : Dependency, list, msg_id, AsyncResult
             the time dependencies of tasks
         timeout : int,None
             the timeout to be used for tasks
         """
         
         super(LoadBalancedView, self).set_flags(**kwargs)
         for name in ('follow', 'after'):
             if name in kwargs:
                 value = kwargs[name]
                 if self._validate_dependency(value):
                     setattr(self, name, value)
                 else:
                     raise ValueError("Invalid dependency: %r"%value)
         if 'timeout' in kwargs:
             t = kwargs['timeout']
             if not isinstance(t, (int, long, float, None)):
                 raise TypeError("Invalid type for timeout: %r"%type(t))
             if t is not None:
                 if t < 0:
                     raise ValueError("Invalid timeout: %s"%t)
             self.timeout = t
                 
     @spin_after
     @save_ids
     def map(self, f, *sequences, **kwargs):
         """view.map(f, *sequences, block=self.block, bound=self.bound, chunk_size=1) => list|AsyncMapResult
         
         Parallel version of builtin `map`, load-balanced by this View.
         
         `block`, `bound`, and `chunk_size` can be specified by keyword only.
         
         Each `chunk_size` elements will be a separate task, and will be
         load-balanced. This lets individual elements be available for iteration
         as soon as they arrive.
         
         Parameters
         ----------
         
         f : callable
             function to be mapped
         *sequences: one or more sequences of matching length
             the sequences to be distributed and passed to `f`
         block : bool
             whether to wait for the result or not [default self.block]
         bound : bool
             whether to use the engine's namespace [default self.bound]
         chunk_size : int
             how many elements should be in each task [default 1]
         
         Returns
         -------
         
         if block=False:
             AsyncMapResult
                 An object like AsyncResult, but which reassembles the sequence of results
                 into a single list. AsyncMapResults can be iterated through before all
                 results are complete.
             else:
                 the result of map(f,*sequences)
         
         """
         
         # default
         block = kwargs.get('block', self.block)
         bound = kwargs.get('bound', self.bound)
         chunk_size = kwargs.get('chunk_size', 1)
         
         keyset = set(kwargs.keys())
         extra_keys = keyset.difference_update(set(['block', 'bound', 'chunk_size']))
         if extra_keys:
             raise TypeError("Invalid kwargs: %s"%list(extra_keys))
         
         assert len(sequences) > 0, "must have some sequences to map onto!"
         
         pf = ParallelFunction(self.client, f, block=block, bound=bound, 
                                 targets=self._targets, balanced=True,
                                 chunk_size=chunk_size)
         return pf.map(*sequences)
     
+__all__ = ['LoadBalancedView', 'DirectView']
  No newline at end of file