allow true single-threaded Controller

Previously --usethreads only put MonitoredQueue devices in threads, and still launched a subprocess for the Python scheduler.  This adds support for the Python Scheduler in a thread.

IPython/parallel/apps/ipcontrollerapp.py

@@ -327,9 +327,15 b' class IPControllerApp(BaseParallelApplication):'
                                 hub.monitor_url, hub.client_info['notification'])
             kwargs = dict(logname='scheduler', loglevel=self.log_level,
                             log_url = self.log_url, config=dict(self.config))
+            if 'Process' in self.mq_class:
+                # run the Python scheduler in a Process
             q = Process(target=launch_scheduler, args=sargs, kwargs=kwargs)
             q.daemon=True
             children.append(q)
+            else:
+                # single-threaded Controller
+                kwargs['in_thread'] = True
+                launch_scheduler(*sargs, **kwargs)
 
     
     def save_urls(self):

IPython/parallel/controller/scheduler.py

@@ -38,6 +38,7 b' from zmq.eventloop import ioloop, zmqstream'
 
 # local imports
 from IPython.external.decorator import decorator
+from IPython.config.application import Application
 from IPython.config.loader import Config
 from IPython.utils.traitlets import Instance, Dict, List, Set, Int, Enum
 
@@ -650,14 +651,21 b' class TaskScheduler(SessionFactory):'
 
 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
+                        identity=b'task', in_thread=False):
+    
+    ZMQStream = zmqstream.ZMQStream
     
     if config:
         # unwrap dict back into Config
         config = Config(config)
 
+    if in_thread:
+        # use instance() to get the same Context/Loop as our parent
+        ctx = zmq.Context.instance()
+        loop = ioloop.IOLoop.instance()
+    else:
+        # in a process, don't use instance() 
+        # for safety with multiprocessing
     ctx = zmq.Context()
     loop = ioloop.IOLoop()
     ins = ZMQStream(ctx.socket(zmq.XREP),loop)
@@ -667,14 +675,16 b' def launch_scheduler(in_addr, out_addr, mon_addr, not_addr, config=None,'
     outs = ZMQStream(ctx.socket(zmq.XREP),loop)
     outs.setsockopt(zmq.IDENTITY, identity)
     outs.bind(out_addr)
-    mons = ZMQStream(ctx.socket(zmq.PUB),loop)
+    mons = zmqstream.ZMQStream(ctx.socket(zmq.PUB),loop)
     mons.connect(mon_addr)
-    nots = ZMQStream(ctx.socket(zmq.SUB),loop)
+    nots = zmqstream.ZMQStream(ctx.socket(zmq.SUB),loop)
     nots.setsockopt(zmq.SUBSCRIBE, '')
     nots.connect(not_addr)
     
-    # setup logging. Note that these will not work in-process, because they clobber
-    # existing loggers.
+    # setup logging.
+    if in_thread:
+        log = Application.instance().log
+    else:
     if log_url:
         log = connect_logger(logname, ctx, log_url, root="scheduler", loglevel=loglevel)
     else:
@@ -685,6 +695,7 b' def launch_scheduler(in_addr, out_addr, mon_addr, not_addr, config=None,'
                             loop=loop, log=log,
                             config=config)
     scheduler.start()
+    if not in_thread:
     try:
         loop.start()
     except KeyboardInterrupt: