reorganize Factory classes to follow relocation of Session object

IPython/parallel/apps/ipclusterapp.py

@@ -308,7 +308,7 b' class IPClusterEngines(BaseParallelApplication):'
         klass = import_item(clsname)
 
         launcher = klass(
-            work_dir=self.profile_dir.location, config=self.config, logname=self.log.name
+            work_dir=self.profile_dir.location, config=self.config, log=self.log
         )
         return launcher
     

IPython/parallel/apps/iploggerapp.py

@@ -69,7 +69,7 b' class IPLoggerApp(BaseParallelApplication):'
     
     def init_watcher(self):
         try:
-            self.watcher = LogWatcher(config=self.config, logname=self.log.name)
+            self.watcher = LogWatcher(config=self.config, log=self.log)
         except:
             self.log.error("Couldn't start the LogWatcher", exc_info=True)
             self.exit(1)

IPython/parallel/apps/launcher.py

@@ -49,14 +49,12 b' except ImportError:'
 
 from zmq.eventloop import ioloop
 
-# from IPython.config.configurable import Configurable
+from IPython.config.configurable import Configurable
 from IPython.utils.text import EvalFormatter
 from IPython.utils.traitlets import Any, Int, List, Unicode, Dict, Instance
 from IPython.utils.path import get_ipython_module_path
 from IPython.utils.process import find_cmd, pycmd2argv, FindCmdError
 
-from IPython.parallel.factory import LoggingFactory
-
 from .win32support import forward_read_events
 
 from .winhpcjob import IPControllerTask, IPEngineTask, IPControllerJob, IPEngineSetJob
@@ -97,7 +95,7 b' class UnknownStatus(LauncherError):'
     pass
 
 
-class BaseLauncher(LoggingFactory):
+class BaseLauncher(Configurable):
     """An asbtraction for starting, stopping and signaling a process."""
 
     # In all of the launchers, the work_dir is where child processes will be
@@ -109,6 +107,7 b' class BaseLauncher(LoggingFactory):'
     # the work_dir option.
     work_dir = Unicode(u'.')
     loop = Instance('zmq.eventloop.ioloop.IOLoop')
+    log = Instance('logging.Logger', ('root',))
     
     start_data = Any()
     stop_data = Any()
@@ -390,7 +389,7 b' class LocalEngineSetLauncher(BaseLauncher):'
         self.profile_dir = unicode(profile_dir)
         dlist = []
         for i in range(n):
-            el = self.launcher_class(work_dir=self.work_dir, config=self.config, logname=self.log.name)
+            el = self.launcher_class(work_dir=self.work_dir, config=self.config, log=self.log)
             # Copy the engine args over to each engine launcher.
             el.engine_args = copy.deepcopy(self.engine_args)
             el.on_stop(self._notice_engine_stopped)
@@ -612,7 +611,7 b' class SSHEngineSetLauncher(LocalEngineSetLauncher):'
             else:
                 user=None
             for i in range(n):
-                el = self.launcher_class(work_dir=self.work_dir, config=self.config, logname=self.log.name)
+                el = self.launcher_class(work_dir=self.work_dir, config=self.config, log=self.log)
                 
                 # Copy the engine args over to each engine launcher.
                 i

IPython/parallel/apps/logwatcher.py

@@ -19,21 +19,23 b' import sys'
 import zmq
 from zmq.eventloop import ioloop, zmqstream
 
+from IPython.config.configurable import Configurable
 from IPython.utils.traitlets import Int, Unicode, Instance, List
 
-from IPython.parallel.factory import LoggingFactory
-
 #-----------------------------------------------------------------------------
 # Classes
 #-----------------------------------------------------------------------------
 
 
-class LogWatcher(LoggingFactory):
+class LogWatcher(Configurable):
     """A simple class that receives messages on a SUB socket, as published
     by subclasses of `zmq.log.handlers.PUBHandler`, and logs them itself.
     
     This can subscribe to multiple topics, but defaults to all topics.
     """
+    
+    log = Instance('logging.Logger', ('root',))
+    
     # configurables
     topics = List([''], config=True,
         help="The ZMQ topics to subscribe to. Default is to subscribe to all messages")

IPython/parallel/controller/heartmonitor.py

@@ -18,8 +18,8 b' import zmq'
 from zmq.devices import ThreadDevice
 from zmq.eventloop import ioloop, zmqstream
 
+from IPython.config.configurable import Configurable
 from IPython.utils.traitlets import Set, Instance, CFloat
-from IPython.parallel.factory import LoggingFactory
 
 class Heart(object):
     """A basic heart object for responding to a HeartMonitor.
@@ -47,7 +47,7 b' class Heart(object):'
     def start(self):
         return self.device.start()
         
-class HeartMonitor(LoggingFactory):
+class HeartMonitor(Configurable):
     """A basic HeartMonitor class
     pingstream: a PUB stream
     pongstream: an XREP stream
@@ -58,6 +58,7 b' class HeartMonitor(LoggingFactory):'
         ' (in ms) [default: 100]',
     )
     
+    log = Instance('logging.Logger', ('root',))
     pingstream=Instance('zmq.eventloop.zmqstream.ZMQStream')
     pongstream=Instance('zmq.eventloop.zmqstream.ZMQStream')
     loop = Instance('zmq.eventloop.ioloop.IOLoop')

IPython/parallel/controller/hub.py

@@ -31,7 +31,9 b' from IPython.utils.traitlets import ('
 from IPython.utils.jsonutil import ISO8601, extract_dates
 
 from IPython.parallel import error, util
-from IPython.parallel.factory import RegistrationFactory, LoggingFactory
+from IPython.parallel.factory import RegistrationFactory
+
+from IPython.zmq.session import SessionFactory
 
 from .heartmonitor import HeartMonitor
 
@@ -231,8 +233,10 b' class HubFactory(RegistrationFactory):'
         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), 
-                                config=self.config)
+        self.heartmonitor = HeartMonitor(loop=loop, config=self.config, log=self.log,
+                                pingstream=ZMQStream(hpub,loop),
+                                pongstream=ZMQStream(hrep,loop)
+                            )
 
         ### Client connections ###
         # Notifier socket
@@ -287,10 +291,10 b' class HubFactory(RegistrationFactory):'
         self.hub = Hub(loop=loop, session=self.session, monitor=sub, heartmonitor=self.heartmonitor,
                 query=q, notifier=n, resubmit=r, db=self.db,
                 engine_info=self.engine_info, client_info=self.client_info,
-                logname=self.log.name)
+                log=self.log)
     
 
-class Hub(LoggingFactory):
+class Hub(SessionFactory):
     """The IPython Controller Hub with 0MQ connections
     
     Parameters
@@ -328,7 +332,6 b' class Hub(LoggingFactory):'
     _idcounter=Int(0)
     
     # objects from constructor:
-    loop=Instance(ioloop.IOLoop)
     query=Instance(ZMQStream)
     monitor=Instance(ZMQStream)
     notifier=Instance(ZMQStream)

IPython/parallel/controller/scheduler.py

@@ -612,7 +612,8 b' class TaskScheduler(SessionFactory):'
         for msg_id in jobs:
             raw_msg, targets, after, follow, timeout = self.depending[msg_id]
             
-            if after.unreachable(self.all_completed, self.all_failed) or follow.unreachable(self.all_completed, self.all_failed):
+            if after.unreachable(self.all_completed, self.all_failed)\
+                    or follow.unreachable(self.all_completed, self.all_failed):
                 self.fail_unreachable(msg_id)
             
             elif after.check(self.all_completed, self.all_failed): # time deps met, maybe run
@@ -643,9 +644,9 b' class TaskScheduler(SessionFactory):'
     
 
 
-def launch_scheduler(in_addr, out_addr, mon_addr, not_addr, config=None,logname='ZMQ', 
-                            log_url=None, loglevel=logging.DEBUG,
-                            identity=b'task'):
+def launch_scheduler(in_addr, out_addr, mon_addr, not_addr, config=None,
+                        logname='root', log_url=None, loglevel=logging.DEBUG,
+                        identity=b'task'):
     from zmq.eventloop import ioloop
     from zmq.eventloop.zmqstream import ZMQStream
     
@@ -671,13 +672,13 b' def launch_scheduler(in_addr, out_addr, mon_addr, not_addr, config=None,logname='
     # setup logging. Note that these will not work in-process, because they clobber
     # existing loggers.
     if log_url:
-        connect_logger(logname, ctx, log_url, root="scheduler", loglevel=loglevel)
+        log = connect_logger(logname, ctx, log_url, root="scheduler", loglevel=loglevel)
     else:
-        local_logger(logname, loglevel)
+        log = local_logger(logname, loglevel)
     
     scheduler = TaskScheduler(client_stream=ins, engine_stream=outs,
                             mon_stream=mons, notifier_stream=nots,
-                            loop=loop, logname=logname,
+                            loop=loop, log=log,
                             config=config)
     scheduler.start()
     try:

IPython/parallel/factory.py

@@ -1,7 +1,7 b''
 """Base config factories."""
 
 #-----------------------------------------------------------------------------
-#  Copyright (C) 2008-2009  The IPython Development Team
+#  Copyright (C) 2010-2011  The IPython Development Team
 #
 #  Distributed under the terms of the BSD License.  The full license is in
 #  the file COPYING, distributed as part of this software.
@@ -22,39 +22,12 b' from IPython.config.configurable import Configurable'
 from IPython.utils.traitlets import Int, Instance, Unicode
 
 from IPython.parallel.util import select_random_ports
-from IPython.zmq.session import Session
+from IPython.zmq.session import Session, SessionFactory
 
 #-----------------------------------------------------------------------------
 # 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 = Unicode('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.parallel inherits"""
-    
-    # not configurable:
-    context = Instance('zmq.Context')
-    def _context_default(self):
-        return zmq.Context.instance()
-    
-    session = Instance('IPython.zmq.session.Session')
-    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)
-        
-        # construct the session
-        self.session = Session(**kwargs)
-    
 
 class RegistrationFactory(SessionFactory):
     """The Base Configurable for objects that involve registration."""

IPython/parallel/util.py

@@ -452,6 +452,7 b' def connect_engine_logger(context, iface, engine, loglevel=logging.DEBUG):'
     handler.setLevel(loglevel)
     logger.addHandler(handler)
     logger.setLevel(loglevel)
+    return logger
 
 def local_logger(logname, loglevel=logging.DEBUG):
     loglevel = integer_loglevel(loglevel)
@@ -463,4 +464,5 b' def local_logger(logname, loglevel=logging.DEBUG):'
     handler.setLevel(loglevel)
     logger.addHandler(handler)
     logger.setLevel(loglevel)
+    return logger
 

IPython/zmq/session.py

@@ -13,6 +13,7 b''
 #-----------------------------------------------------------------------------
 
 import hmac
+import logging
 import os
 import pprint
 import uuid
@@ -27,6 +28,7 b' except:'
 
 import zmq
 from zmq.utils import jsonapi
+from zmq.eventloop.ioloop import IOLoop
 from zmq.eventloop.zmqstream import ZMQStream
 
 from IPython.config.configurable import Configurable
@@ -73,6 +75,36 b' DELIM="<IDS|MSG>"'
 # Classes
 #-----------------------------------------------------------------------------
 
+class SessionFactory(Configurable):
+    """The Base class for configurables that have a Session, Context, logger,
+    and IOLoop.
+    """
+    
+    log = Instance('logging.Logger', ('', logging.WARN))
+    
+    logname = Unicode('')
+    def _logname_changed(self, name, old, new):
+        self.log = logging.getLogger(new)
+    
+    # not configurable:
+    context = Instance('zmq.Context')
+    def _context_default(self):
+        return zmq.Context.instance()
+    
+    session = Instance('IPython.zmq.session.Session')
+    
+    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)
+        
+        if self.session is None:
+            # construct the session
+            self.session = Session(**kwargs)
+    
+
 class Message(object):
     """A simple message object that maps dict keys to attributes.
 
@@ -143,7 +175,7 b' class Session(Configurable):'
         else:
             self.pack = import_item(new)
 
-    unpacker = Unicode('json',config=True,
+    unpacker = Unicode('json', config=True,
         help="""The name of the unpacker for unserializing messages.
         Only used with custom functions for `packer`.""")
     def _unpacker_changed(self, name, old, new):
@@ -156,7 +188,7 b' class Session(Configurable):'
         else:
             self.unpack = import_item(new)
         
-    session = CStr('',config=True,
+    session = CStr('', config=True,
         help="""The UUID identifying this session.""")
     def _session_default(self):
         return bytes(uuid.uuid4())