Add SQLite backend, DB backends are Configurable

also fix small numpy typo in newserialized

IPython/config/default/ipcontrollerz_config.py

@@ -1,136 +1,180 b''
 from IPython.config.loader import Config
 
 c = get_config()
 
 #-----------------------------------------------------------------------------
 # Global configuration
 #-----------------------------------------------------------------------------
 
 # Basic Global config attributes
 
 # Start up messages are logged to stdout using the logging module.
 # These all happen before the twisted reactor is started and are
 # useful for debugging purposes. Can be (10=DEBUG,20=INFO,30=WARN,40=CRITICAL) 
 # and smaller is more verbose.
 # c.Global.log_level = 20
 
 # Log to a file in cluster_dir/log, otherwise just log to sys.stdout.
 # c.Global.log_to_file = False
 
 # Remove old logs from cluster_dir/log before starting.
 # c.Global.clean_logs = True
 
 # A list of Python statements that will be run before starting the 
 # controller. This is provided because occasionally certain things need to 
 # be imported in the controller for pickling to work.
 # c.Global.import_statements = ['import math']
 
 # Reuse the controller's JSON files. If False, JSON files are regenerated
 # each time the controller is run. If True, they will be reused, *but*, you
 # also must set the network ports by hand. If set, this will override the
 # values set for the client and engine connections below.
 # c.Global.reuse_files = True
 
 # Enable exec_key authentication on all messages. Default is True
 # c.Global.secure = True
 
 # The working directory for the process. The application will use os.chdir
 # to change to this directory before starting.
 # c.Global.work_dir = os.getcwd()
 
 # The log url for logging to an `iploggerz` application.  This will override
 # log-to-file.
 # c.Global.log_url = 'tcp://127.0.0.1:20202'
 
 # The specific external IP that is used to disambiguate multi-interface URLs.
 # The default behavior is to guess from external IPs gleaned from `socket`.
 # c.Global.location = '192.168.1.123'
 
 # The ssh server remote clients should use to connect to this controller.
 # It must be a machine that can see the interface specified in client_ip.
 # The default for client_ip is localhost, in which case the sshserver must
 # be an external IP of the controller machine.
 # c.Global.sshserver = 'controller.example.com'
 
 # the url to use for registration.  If set, this overrides engine-ip,
 # engine-transport client-ip,client-transport, and regport.
 # c.RegistrationFactory.url = 'tcp://*:12345'
 
 # the port to use for registration.  Clients and Engines both use this
 # port for registration.
 # c.RegistrationFactory.regport = 10101
 
 #-----------------------------------------------------------------------------
 # Configure the Task Scheduler
 #-----------------------------------------------------------------------------
 
 # The routing scheme. 'pure' will use the pure-ZMQ scheduler. Any other 
 # value will use a Python scheduler with various routing schemes.
 # python schemes are: lru, weighted, random, twobin. Default is 'weighted'.
 # Note that the pure ZMQ scheduler does not support many features, such as
 # dying engines, dependencies, or engine-subset load-balancing.
 # c.ControllerFactory.scheme = 'pure'
 
 # The pure ZMQ scheduler can limit the number of outstanding tasks per engine
 # by using the ZMQ HWM option.  This allows engines with long-running tasks
 # to not steal too many tasks from other engines. The default is 0, which
 # means agressively distribute messages, never waiting for them to finish.
 # c.ControllerFactory.hwm = 1
 
 # Whether to use Threads or Processes to start the Schedulers.  Threads will
 # use less resources, but potentially reduce throughput. Default is to 
 # use processes.  Note that the a Python scheduler will always be in a Process.
 # c.ControllerFactory.usethreads
 
 #-----------------------------------------------------------------------------
 # Configure the Hub
 #-----------------------------------------------------------------------------
 
 # Which class to use for the db backend.  Currently supported are DictDB (the
 # default), and MongoDB. Uncomment this line to enable MongoDB, which will
 # slow-down the Hub's responsiveness, but also reduce its memory footprint.
 # c.HubFactory.db_class = 'IPython.zmq.parallel.mongodb.MongoDB'
 
 # The heartbeat ping frequency.  This is the frequency (in ms) at which the
 # Hub pings engines for heartbeats.  This determines how quickly the Hub
 # will react to engines coming and going.  A lower number means faster response
 # time, but more network activity.  The default is 100ms
 # c.HubFactory.ping = 100
 
 # HubFactory queue port pairs, to set by name: mux, iopub, control, task.  Set
 # each as a tuple of length 2 of ints.  The default is to find random
 # available ports
 # c.HubFactory.mux = (10102,10112)
 
 #-----------------------------------------------------------------------------
 # Configure the client connections
 #-----------------------------------------------------------------------------
 
 # Basic client connection config attributes
 
 # The network interface the controller will listen on for client connections.
 # This should be an IP address or interface on the controller. An asterisk
 # means listen on all interfaces. The transport can be any transport
 # supported by zeromq (tcp,epgm,pgm,ib,ipc):
 # c.HubFactory.client_ip = '*'
 # c.HubFactory.client_transport = 'tcp'
 
 # individual client ports to configure by name: query_port, notifier_port
 # c.HubFactory.query_port = 12345
 
 #-----------------------------------------------------------------------------
 # Configure the engine connections
 #-----------------------------------------------------------------------------
 
 # Basic config attributes for the engine connections.
 
 # The network interface the controller will listen on for engine connections.
 # This should be an IP address or interface on the controller. An asterisk
 # means listen on all interfaces. The transport can be any transport
 # supported by zeromq (tcp,epgm,pgm,ib,ipc):
 # c.HubFactory.engine_ip = '*'
 # c.HubFactory.engine_transport = 'tcp'
 
 # set the engine heartbeat ports to use:
 # c.HubFactory.hb = (10303,10313)
 
+#-----------------------------------------------------------------------------
+# Configure the TaskRecord database backend
+#-----------------------------------------------------------------------------
+
+# For memory/persistance reasons, tasks can be stored out-of-memory in a database.
+# Currently, only sqlite and mongodb are supported as backends, but the interface
+# is fairly simple, so advanced developers could write their own backend.
+
+# ----- in-memory configuration --------
+# this line restores the default behavior: in-memory storage of all results.
+# c.HubFactory.db_class = 'IPython.zmq.parallel.dictdb.DictDB'
+
+# ----- sqlite configuration --------
+# use this line to activate sqlite:
+# c.HubFactory.db_class = 'IPython.zmq.parallel.sqlitedb.SQLiteDB'
+
+# You can specify the name of the db-file.  By default, this will be located
+# in the active cluster_dir, e.g. ~/.ipython/clusterz_default/tasks.db
+# c.SQLiteDB.filename = 'tasks.db'
+
+# You can also specify the location of the db-file, if you want it to be somewhere
+# other than the cluster_dir.
+# c.SQLiteDB.location = '/scratch/'
+
+# This will specify the name of the table for the controller to use.  The default
+# behavior is to use the session ID of the SessionFactory object (a uuid). Overriding
+# this will result in results persisting for multiple sessions.
+# c.SQLiteDB.table = 'results'
+
+# ----- mongodb configuration --------
+# use this line to activate mongodb:
+# c.HubFactory.db_class = 'IPython.zmq.parallel.mongodb.MongoDB'
+
+# You can specify the args and kwargs pymongo will use when creating the Connection.
+# For more information on what these options might be, see pymongo documentation.
+# c.MongoDB.connection_kwargs = {}
+# c.MongoDB.connection_args = []
+
+# This will specify the name of the mongo database for the controller to use.  The default
+# behavior is to use the session ID of the SessionFactory object (a uuid). Overriding
+# this will result in task results persisting through multiple sessions.
+# c.MongoDB.database = 'ipythondb'
+
+

IPython/utils/newserialized.py

@@ -1,167 +1,167 b''
 # encoding: utf-8
 # -*- test-case-name: IPython.kernel.test.test_newserialized -*-
 
 """Refactored serialization classes and interfaces."""
 
 __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.
 #-------------------------------------------------------------------------------
 
 #-------------------------------------------------------------------------------
 # Imports
 #-------------------------------------------------------------------------------
 
 import cPickle as pickle
 
 try:
     import numpy
 except ImportError:
     pass
 
 class SerializationError(Exception):
     pass
 
 #-----------------------------------------------------------------------------
 # Classes and functions
 #-----------------------------------------------------------------------------
 
 class ISerialized:
     
     def getData():
         """"""
 
     def getDataSize(units=10.0**6):
         """"""
 
     def getTypeDescriptor():
         """"""
 
     def getMetadata():
         """"""
         
         
 class IUnSerialized:
     
     def getObject():
         """"""
         
 class Serialized(object):
     
     # implements(ISerialized)
     
     def __init__(self, data, typeDescriptor, metadata={}):
         self.data = data
         self.typeDescriptor = typeDescriptor
         self.metadata = metadata
         
     def getData(self):
         return self.data
         
     def getDataSize(self, units=10.0**6):
         return len(self.data)/units
         
     def getTypeDescriptor(self):
         return self.typeDescriptor
         
     def getMetadata(self):
         return self.metadata
 
         
 class UnSerialized(object):
     
     # implements(IUnSerialized)
         
     def __init__(self, obj):
         self.obj = obj
         
     def getObject(self):
         return self.obj
 
         
 class SerializeIt(object):
     
     # implements(ISerialized)
     
     def __init__(self, unSerialized):
         self.data = None
         self.obj = unSerialized.getObject()
         if globals().has_key('numpy') and isinstance(self.obj, numpy.ndarray):
                 if len(self.obj) == 0:         # length 0 arrays can't be reconstructed
                     raise SerializationError("You cannot send a length 0 array")
                 self.obj = numpy.ascontiguousarray(self.obj, dtype=None)
                 self.typeDescriptor = 'ndarray'
                 self.metadata = {'shape':self.obj.shape,
                                  'dtype':self.obj.dtype.str}
         elif isinstance(self.obj, bytes):
             self.typeDescriptor = 'bytes'
             self.metadata = {}
         elif isinstance(self.obj, buffer):
             self.typeDescriptor = 'buffer'
             self.metadata = {}
         else:
             self.typeDescriptor = 'pickle'
             self.metadata = {}
         self._generateData()
     
     def _generateData(self):
         if self.typeDescriptor == 'ndarray':
             self.data = numpy.getbuffer(self.obj)
         elif self.typeDescriptor in ('bytes', 'buffer'):
             self.data = self.obj
         elif self.typeDescriptor == 'pickle':
             self.data = pickle.dumps(self.obj, -1)
         else:
             raise SerializationError("Really wierd serialization error.")
         del self.obj
         
     def getData(self):
         return self.data
         
     def getDataSize(self, units=10.0**6):
         return 1.0*len(self.data)/units
         
     def getTypeDescriptor(self):
         return self.typeDescriptor
         
     def getMetadata(self):
         return self.metadata
 
 
 class UnSerializeIt(UnSerialized):
     
     # implements(IUnSerialized)
     
     def __init__(self, serialized):
         self.serialized = serialized
         
     def getObject(self):
         typeDescriptor = self.serialized.getTypeDescriptor()
         if globals().has_key('numpy') and typeDescriptor == 'ndarray':
                 buf = self.serialized.getData()
-                if isinstance(buf, buffer):
+                if isinstance(buf, (buffer,bytes)):
                     result = numpy.frombuffer(buf, dtype = self.serialized.metadata['dtype'])
                 else:
                     # memoryview
                     result = numpy.array(buf, dtype = self.serialized.metadata['dtype'])
                 result.shape = self.serialized.metadata['shape']
         elif typeDescriptor == 'pickle':
             result = pickle.loads(self.serialized.getData())
         elif typeDescriptor in ('bytes', 'buffer'):
             result = self.serialized.getData()
         else:
             raise SerializationError("Really wierd serialization error.")
         return result
 
 def serialize(obj):
     return SerializeIt(UnSerialized(obj))
     
 def unserialize(serialized):
     return UnSerializeIt(serialized).getObject()

IPython/zmq/parallel/dictdb.py

@@ -1,152 +1,155 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
 
+from IPython.config.configurable import Configurable
+
+from IPython.utils.traitlets import Dict, CUnicode
+
 filters = {
  '$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,
  '$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 ]),
  '$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):
+class BaseDB(Configurable):
     """Empty Parent class so traitlets work on DB."""
-    pass
+    # base configurable traits:
+    session = CUnicode("")
 
 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()
+    _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/hub.py

@@ -1,1052 +1,1042 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 sys
 import time
 from datetime import datetime
 
 import zmq
 from zmq.eventloop import ioloop
 from zmq.eventloop.zmqstream import ZMQStream
 
 # internal:
 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 .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)
+        self.log.info('Hub using DB backend: %r'%(self.db_class.split()[-1]))
+        cdir = self.config.Global.cluster_dir
+        print (cdir)
+        self.db = import_item(self.db_class)(session=self.session.session, config=self.config)
         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.log.debug("Hub engine addrs: %s"%self.engine_info)
+        self.log.debug("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 = 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 = 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']
+
+            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']
+
+            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 = 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 = 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 = 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 = 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 = 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 = 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 = error.wrap_exception()
                     else:
                         try:
                             raise IndexError("No such msg: %r"%msg_id)
                         except:
                             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 = 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 = error.wrap_exception()
                 break
         self.session.send(self.clientele, "result_reply", content=content, 
                                             parent=msg, ident=client_id,
                                             buffers=buffers)
 

IPython/zmq/parallel/ipcontrollerapp.py

@@ -1,427 +1,431 b''
 #!/usr/bin/env python
 # encoding: utf-8
 """
 The IPython controller application.
 """
 
 #-----------------------------------------------------------------------------
 #  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 copy
 import os
 import logging
 import socket
 import stat
 import sys
 import uuid
 
 import zmq
 from zmq.log.handlers import PUBHandler
 from zmq.utils import jsonapi as json
 
 from IPython.config.loader import Config
 from IPython.zmq.parallel import factory
 from IPython.zmq.parallel.controller import ControllerFactory
 from IPython.zmq.parallel.clusterdir import (
     ApplicationWithClusterDir,
     ClusterDirConfigLoader
 )
 from IPython.zmq.parallel.util import disambiguate_ip_address, split_url
 # from IPython.kernel.fcutil import FCServiceFactory, FURLError
 from IPython.utils.traitlets import Instance, Unicode
 
 
 
 #-----------------------------------------------------------------------------
 # Module level variables
 #-----------------------------------------------------------------------------
 
 
 #: The default config file name for this application
 default_config_file_name = u'ipcontrollerz_config.py'
 
 
 _description = """Start the IPython controller for parallel computing.
 
 The IPython controller provides a gateway between the IPython engines and
 clients. The controller needs to be started before the engines and can be
 configured using command line options or using a cluster directory. Cluster
 directories contain config, log and security files and are usually located in
 your ipython directory and named as "clusterz_<profile>". See the --profile
 and --cluster-dir options for details.
 """
 
 #-----------------------------------------------------------------------------
 # Default interfaces
 #-----------------------------------------------------------------------------
 
 # The default client interfaces for FCClientServiceFactory.interfaces
 default_client_interfaces = Config()
 default_client_interfaces.Default.url_file = 'ipcontroller-client.url'
 
 # Make this a dict we can pass to Config.__init__ for the default
 default_client_interfaces = dict(copy.deepcopy(default_client_interfaces.items()))
 
 
 
 # The default engine interfaces for FCEngineServiceFactory.interfaces
 default_engine_interfaces = Config()
 default_engine_interfaces.Default.url_file = u'ipcontroller-engine.url'
 
 # Make this a dict we can pass to Config.__init__ for the default
 default_engine_interfaces = dict(copy.deepcopy(default_engine_interfaces.items()))
 
 
 #-----------------------------------------------------------------------------
 # Service factories
 #-----------------------------------------------------------------------------
 
 # 
 # class FCClientServiceFactory(FCServiceFactory):
 #     """A Foolscap implementation of the client services."""
 # 
 #     cert_file = Unicode(u'ipcontroller-client.pem', config=True)
 #     interfaces = Instance(klass=Config, kw=default_client_interfaces,
 #                           allow_none=False, config=True)
 # 
 # 
 # class FCEngineServiceFactory(FCServiceFactory):
 #     """A Foolscap implementation of the engine services."""
 # 
 #     cert_file = Unicode(u'ipcontroller-engine.pem', config=True)
 #     interfaces = Instance(klass=dict, kw=default_engine_interfaces,
 #                           allow_none=False, config=True)
 # 
 
 #-----------------------------------------------------------------------------
 # Command line options
 #-----------------------------------------------------------------------------
 
 
 class IPControllerAppConfigLoader(ClusterDirConfigLoader):
 
     def _add_arguments(self):
         super(IPControllerAppConfigLoader, self)._add_arguments()
         paa = self.parser.add_argument
         
         ## Hub Config:
         paa('--mongodb', 
             dest='HubFactory.db_class', action='store_const',
             const='IPython.zmq.parallel.mongodb.MongoDB', 
-            help='Use MongoDB task storage [default: in-memory]')
+            help='Use MongoDB for task storage [default: in-memory]')
+        paa('--sqlite', 
+            dest='HubFactory.db_class', action='store_const',
+            const='IPython.zmq.parallel.sqlitedb.SQLiteDB', 
+            help='Use SQLite3 for DB task storage [default: in-memory]')
         paa('--hb',
             type=int, dest='HubFactory.hb', nargs=2,
             help='The (2) ports the Hub\'s Heartmonitor will use for the heartbeat '
             'connections [default: random]',
             metavar='Hub.hb_ports')
         paa('--ping',
             type=int, dest='HubFactory.ping',
             help='The frequency at which the Hub pings the engines for heartbeats '
             ' (in ms) [default: 100]',
             metavar='Hub.ping')
         
         # Client config
         paa('--client-ip',
             type=str, dest='HubFactory.client_ip', 
             help='The IP address or hostname the Hub will listen on for '
             'client connections. Both engine-ip and client-ip can be set simultaneously '
             'via --ip [default: loopback]',
             metavar='Hub.client_ip')
         paa('--client-transport',
             type=str, dest='HubFactory.client_transport', 
             help='The ZeroMQ transport the Hub will use for '
             'client connections. Both engine-transport and client-transport can be set simultaneously '
             'via --transport [default: tcp]',
             metavar='Hub.client_transport')
         paa('--query',
             type=int, dest='HubFactory.query_port', 
             help='The port on which the Hub XREP socket will listen for result queries from clients [default: random]',
             metavar='Hub.query_port')
         paa('--notifier',
             type=int, dest='HubFactory.notifier_port', 
             help='The port on which the Hub PUB socket will listen for notification connections [default: random]',
             metavar='Hub.notifier_port')
         
         # Engine config
         paa('--engine-ip',
             type=str, dest='HubFactory.engine_ip', 
             help='The IP address or hostname the Hub will listen on for '
             'engine connections. This applies to the Hub and its schedulers'
             'engine-ip and client-ip can be set simultaneously '
             'via --ip [default: loopback]',
             metavar='Hub.engine_ip')
         paa('--engine-transport',
             type=str, dest='HubFactory.engine_transport', 
             help='The ZeroMQ transport the Hub will use for '
             'client connections. Both engine-transport and client-transport can be set simultaneously '
             'via --transport [default: tcp]',
             metavar='Hub.engine_transport')
         
         # Scheduler config
         paa('--mux',
             type=int, dest='ControllerFactory.mux', nargs=2,
             help='The (2) ports the MUX scheduler will listen on for client,engine '
             'connections, respectively [default: random]',
             metavar='Scheduler.mux_ports')
         paa('--task',
             type=int, dest='ControllerFactory.task', nargs=2,
             help='The (2) ports the Task scheduler will listen on for client,engine '
             'connections, respectively [default: random]',
             metavar='Scheduler.task_ports')
         paa('--control',
             type=int, dest='ControllerFactory.control', nargs=2,
             help='The (2) ports the Control scheduler will listen on for client,engine '
             'connections, respectively [default: random]',
             metavar='Scheduler.control_ports')
         paa('--iopub',
             type=int, dest='ControllerFactory.iopub', nargs=2,
             help='The (2) ports the IOPub scheduler will listen on for client,engine '
             'connections, respectively [default: random]',
             metavar='Scheduler.iopub_ports')
             
         paa('--scheme',
             type=str, dest='HubFactory.scheme',
             choices = ['pure', 'lru', 'plainrandom', 'weighted', 'twobin','leastload'],
             help='select the task scheduler scheme  [default: Python LRU]',
             metavar='Scheduler.scheme')
         paa('--usethreads',
             dest='ControllerFactory.usethreads', action="store_true",
             help='Use threads instead of processes for the schedulers',
             )
         paa('--hwm',
             dest='ControllerFactory.hwm', type=int,
             help='specify the High Water Mark (HWM) for the downstream '
             'socket in the pure ZMQ scheduler. This is the maximum number '
             'of allowed outstanding tasks on each engine.',
             )
         
         ## Global config
         paa('--log-to-file',
             action='store_true', dest='Global.log_to_file',
             help='Log to a file in the log directory (default is stdout)')
         paa('--log-url',
             type=str, dest='Global.log_url',
             help='Broadcast logs to an iploggerz process [default: disabled]')
         paa('-r','--reuse-files', 
             action='store_true', dest='Global.reuse_files',
             help='Try to reuse existing json connection files.')
         paa('--no-secure',
             action='store_false', dest='Global.secure',
             help='Turn off execution keys (default).')
         paa('--secure',
             action='store_true', dest='Global.secure',
             help='Turn on execution keys.')
         paa('--execkey',
             type=str, dest='Global.exec_key',
             help='path to a file containing an execution key.',
             metavar='keyfile')
         paa('--ssh',
             type=str, dest='Global.sshserver',
             help='ssh url for clients to use when connecting to the Controller '
             'processes. It should be of the form: [user@]server[:port]. The '
             'Controller\'s listening addresses must be accessible from the ssh server',
             metavar='Global.sshserver')
         paa('--location',
             type=str, dest='Global.location',
             help="The external IP or domain name of this machine, used for disambiguating "
             "engine and client connections.",
             metavar='Global.location')
         factory.add_session_arguments(self.parser)
         factory.add_registration_arguments(self.parser)
 
 
 #-----------------------------------------------------------------------------
 # The main application
 #-----------------------------------------------------------------------------
 
 
 class IPControllerApp(ApplicationWithClusterDir):
 
     name = u'ipcontrollerz'
     description = _description
     command_line_loader = IPControllerAppConfigLoader
     default_config_file_name = default_config_file_name
     auto_create_cluster_dir = True
     
 
     def create_default_config(self):
         super(IPControllerApp, self).create_default_config()
         # Don't set defaults for Global.secure or Global.reuse_furls
         # as those are set in a component.
         self.default_config.Global.import_statements = []
         self.default_config.Global.clean_logs = True
         self.default_config.Global.secure = True
         self.default_config.Global.reuse_files = False
         self.default_config.Global.exec_key = "exec_key.key"
         self.default_config.Global.sshserver = None
         self.default_config.Global.location = None
 
     def pre_construct(self):
         super(IPControllerApp, self).pre_construct()
         c = self.master_config
         # The defaults for these are set in FCClientServiceFactory and
         # FCEngineServiceFactory, so we only set them here if the global
         # options have be set to override the class level defaults.
         
         # if hasattr(c.Global, 'reuse_furls'):
         #     c.FCClientServiceFactory.reuse_furls = c.Global.reuse_furls
         #     c.FCEngineServiceFactory.reuse_furls = c.Global.reuse_furls
         #     del c.Global.reuse_furls
         # if hasattr(c.Global, 'secure'):
         #     c.FCClientServiceFactory.secure = c.Global.secure
         #     c.FCEngineServiceFactory.secure = c.Global.secure
         #     del c.Global.secure
     
     def save_connection_dict(self, fname, cdict):
         """save a connection dict to json file."""
         c = self.master_config
         url = cdict['url']
         location = cdict['location']
         if not location:
             try:
                 proto,ip,port = split_url(url)
             except AssertionError:
                 pass
             else:
                 location = socket.gethostbyname_ex(socket.gethostname())[2][-1]
             cdict['location'] = location
         fname = os.path.join(c.Global.security_dir, fname)
         with open(fname, 'w') as f:
             f.write(json.dumps(cdict, indent=2))
         os.chmod(fname, stat.S_IRUSR|stat.S_IWUSR)
     
     def load_config_from_json(self):
         """load config from existing json connector files."""
         c = self.master_config
         # load from engine config
         with open(os.path.join(c.Global.security_dir, 'ipcontroller-engine.json')) as f:
             cfg = json.loads(f.read())
         key = c.SessionFactory.exec_key = cfg['exec_key']
         xport,addr = cfg['url'].split('://')
         c.HubFactory.engine_transport = xport
         ip,ports = addr.split(':')
         c.HubFactory.engine_ip = ip
         c.HubFactory.regport = int(ports)
         c.Global.location = cfg['location']
         
         # load client config
         with open(os.path.join(c.Global.security_dir, 'ipcontroller-client.json')) as f:
             cfg = json.loads(f.read())
         assert key == cfg['exec_key'], "exec_key mismatch between engine and client keys"
         xport,addr = cfg['url'].split('://')
         c.HubFactory.client_transport = xport
         ip,ports = addr.split(':')
         c.HubFactory.client_ip = ip
         c.Global.sshserver = cfg['ssh']
         assert int(ports) == c.HubFactory.regport, "regport mismatch"
     
     def construct(self):
         # This is the working dir by now.
         sys.path.insert(0, '')
         c = self.master_config
         
         self.import_statements()
         reusing = c.Global.reuse_files
         if reusing:
             try:
                 self.load_config_from_json()
             except (AssertionError,IOError):
                 reusing=False
         # check again, because reusing may have failed:
         if reusing:
             pass
         elif c.Global.secure:
             keyfile = os.path.join(c.Global.security_dir, c.Global.exec_key)
             key = str(uuid.uuid4())
             with open(keyfile, 'w') as f:
                 f.write(key)
             os.chmod(keyfile, stat.S_IRUSR|stat.S_IWUSR)
             c.SessionFactory.exec_key = key
         else:
             c.SessionFactory.exec_key = ''
             key = None
         
         try:
             self.factory = ControllerFactory(config=c, logname=self.log.name)
             self.start_logging()
             self.factory.construct()
         except:
             self.log.error("Couldn't construct the Controller", exc_info=True)
             self.exit(1)
         
         if not reusing:
             # save to new json config files
             f = self.factory
             cdict = {'exec_key' : key,
                     'ssh' : c.Global.sshserver,
                     'url' : "%s://%s:%s"%(f.client_transport, f.client_ip, f.regport),
                     'location' : c.Global.location
                     }
             self.save_connection_dict('ipcontroller-client.json', cdict)
             edict = cdict
             edict['url']="%s://%s:%s"%((f.client_transport, f.client_ip, f.regport))
             self.save_connection_dict('ipcontroller-engine.json', edict)
         
     
     def save_urls(self):
         """save the registration urls to files."""
         c = self.master_config
         
         sec_dir = c.Global.security_dir
         cf = self.factory
         
         with open(os.path.join(sec_dir, 'ipcontroller-engine.url'), 'w') as f:
             f.write("%s://%s:%s"%(cf.engine_transport, cf.engine_ip, cf.regport))
         
         with open(os.path.join(sec_dir, 'ipcontroller-client.url'), 'w') as f:
             f.write("%s://%s:%s"%(cf.client_transport, cf.client_ip, cf.regport))
         
     
     def import_statements(self):
         statements = self.master_config.Global.import_statements
         for s in statements:
             try:
                 self.log.msg("Executing statement: '%s'" % s)
                 exec s in globals(), locals()
             except:
                 self.log.msg("Error running statement: %s" % s)
 
     def start_logging(self):
         super(IPControllerApp, self).start_logging()
         if self.master_config.Global.log_url:
             context = self.factory.context
             lsock = context.socket(zmq.PUB)
             lsock.connect(self.master_config.Global.log_url)
             handler = PUBHandler(lsock)
             handler.root_topic = 'controller'
             handler.setLevel(self.log_level)
             self.log.addHandler(handler)
     # 
     def start_app(self):
         # Start the subprocesses:
         self.factory.start()
         self.write_pid_file(overwrite=True)
         try:
             self.factory.loop.start()
         except KeyboardInterrupt:
             self.log.critical("Interrupted, Exiting...\n")
 
 
 def launch_new_instance():
     """Create and run the IPython controller"""
     app = IPControllerApp()
     app.start()
 
 
 if __name__ == '__main__':
     launch_new_instance()

IPython/zmq/parallel/mongodb.py

@@ -1,62 +1,80 b''
 """A TaskRecord backend using mongodb"""
 #-----------------------------------------------------------------------------
 #  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
 
 from pymongo import Connection
+from pymongo.binary import Binary
+
+from IPython.utils.traitlets import Dict, List, CUnicode
 
 from .dictdb import BaseDB
 
 #-----------------------------------------------------------------------------
 # MongoDB class
 #-----------------------------------------------------------------------------
 
 class MongoDB(BaseDB):
     """MongoDB TaskRecord backend."""
-    def __init__(self, session_uuid, *args, **kwargs):
-        self._connection = Connection(*args, **kwargs)
-        self._db = self._connection[session_uuid]
+    
+    connection_args = List(config=True)
+    connection_kwargs = Dict(config=True)
+    database = CUnicode(config=True)
+    _table = Dict()
+    
+    def __init__(self, **kwargs):
+        super(MongoDB, self).__init__(**kwargs)
+        self._connection = Connection(*self.connection_args, **self.connection_kwargs)
+        if not self.database:
+            self.database = self.session
+        self._db = self._connection[self.database]
         self._records = self._db['task_records']
-        self._table = {}
+    
+    def _binary_buffers(self, rec):
+        for key in ('buffers', 'result_buffers'):
+            if key in rec:
+                rec[key] = map(Binary, rec[key])
     
     def add_record(self, msg_id, rec):
         """Add a new Task Record, by msg_id."""
         # print rec
+        rec = _binary_buffers(rec)
         obj_id = self._records.insert(rec)
         self._table[msg_id] = obj_id
     
     def get_record(self, msg_id):
         """Get a specific Task Record, by msg_id."""
         return self._records.find_one(self._table[msg_id])
     
     def update_record(self, msg_id, rec):
         """Update the data in an existing record."""
+        rec = _binary_buffers(rec)
         obj_id = self._table[msg_id]
         self._records.update({'_id':obj_id}, {'$set': rec})
     
     def drop_matching_records(self, check):
         """Remove a record from the DB."""
         self._records.remove(check)
         
     def drop_record(self, msg_id):
         """Remove a record from the DB."""
         obj_id = self._table.pop(msg_id)
         self._records.remove(obj_id)
     
     def find_records(self, check, id_only=False):
         """Find records matching a query dict."""
         matches = list(self._records.find(check))
         if id_only:
             return [ rec['msg_id'] for rec in matches ]
         else:
             data = {}
             for rec in matches:
                 data[rec['msg_id']] = rec
             return data