added zmq controller/engine entry points

IPython/zmq/parallel/entry_point.py

@@ -0,0 +1,74 b''
+""" Defines helper functions for creating kernel entry points and process
+launchers.
+"""
+
+# Standard library imports.
+import logging
+import atexit
+import os
+import socket
+from subprocess import Popen, PIPE
+import sys
+
+# System library imports.
+import zmq
+from zmq.log import handlers
+# Local imports.
+from IPython.core.ultratb import FormattedTB
+from IPython.external.argparse import ArgumentParser
+from IPython.zmq.log import logger
+
+def split_ports(s, n):
+    """Parser helper for multiport strings"""
+    if not s:
+        return tuple([0]*n)
+    ports = map(int, s.split(','))
+    if len(ports) != n:
+        raise ValueError
+    return ports
+
+def select_random_ports(n):
+    """Selects and return n random ports that are open."""
+    ports = []
+    for i in xrange(n):
+        sock = socket.socket()
+        sock.bind(('', 0))
+        ports.append(sock)
+    for i, sock in enumerate(ports):
+        port = sock.getsockname()[1]
+        sock.close()
+        ports[i] = port
+    return ports
+        
+
+def make_argument_parser():
+    """ Creates an ArgumentParser for the generic arguments supported by all 
+    ipcluster entry points.
+    """
+    parser = ArgumentParser()
+    parser.add_argument('--ip', type=str, default='127.0.0.1',
+                        help='set the controller\'s IP address [default: local]')
+    parser.add_argument('--transport', type=str, default='tcp',
+                        help='set the transport to use [default: tcp]')
+    parser.add_argument('--regport', type=int, metavar='PORT', default=10101,
+                        help='set the XREP port for registration [default: 10101]')
+    parser.add_argument('--logport', type=int, metavar='PORT', default=20202,
+                        help='set the PUB port for logging [default: 10201]')
+    parser.add_argument('--loglevel', type=int, metavar='LEVEL', default=logging.DEBUG,
+                        help='set the log level [default: DEBUG]')
+    parser.add_argument('--ident', type=str,
+                        help='set the ZMQ identity [default: random]')
+    parser.add_argument('--url', type=str,
+                        help='set transport,ip,regport in one arg, e.g. tcp://127.0.0.1:10101')
+
+    return parser
+
+
+def connect_logger(context, iface, root="ip", loglevel=logging.DEBUG):
+    lsock = context.socket(zmq.PUB)
+    lsock.connect(iface)
+    handler = handlers.PUBHandler(lsock)
+    handler.setLevel(loglevel)
+    handler.root_topic = root
+    logger.addHandler(handler)
+    
  No newline at end of file

IPython/zmq/parallel/controller.py

@@ -15,15 +15,19 b' This is the master object that handles connections from engines, clients, and'
 # Imports
 #-----------------------------------------------------------------------------
 from datetime import datetime
+import logging
 
 import zmq
 from zmq.eventloop import zmqstream, ioloop
 import uuid
 
 # internal:
-from streamsession import Message, wrap_exception # default_unpacker as unpack, default_packer as pack
 from IPython.zmq.log import logger # a Logger object
+from IPython.zmq.entry_point import bind_port
 
+from streamsession import Message, wrap_exception
+from entry_point import (make_argument_parser, select_random_ports, split_ports,
+                        connect_logger)
 # from messages import json # use the same import switches
 
 #-----------------------------------------------------------------------------
@@ -359,10 +363,11 b' class Controller(object):'
         triggers unregistration"""
         logger.debug("heartbeat::handle_heart_failure(%r)"%heart)
         eid = self.hearts.get(heart, None)
+        queue = self.engines[eid].queue
         if eid is None:
             logger.info("heartbeat::ignoring heart failure %r"%heart)
         else:
-            self.unregister_engine(heart, dict(content=dict(id=eid)))
+            self.unregister_engine(heart, dict(content=dict(id=eid, queue=queue)))
     
     #----------------------- MUX Queue Traffic ------------------------------
     
@@ -642,29 +647,6 b' class Controller(object):'
             # pending
         self.session.send(self.clientele, "queue_reply", content=content, ident=client_id)
     
-    def job_status(self, client_id, msg):
-        """handle queue_status request"""
-        content = msg['content']
-        msg_ids = content['msg_ids']
-        try:
-            targets = self._validate_targets(targets)
-        except:
-            content = wrap_exception()
-            self.session.send(self.clientele, "controller_error", 
-                    content=content, ident=client_id)
-            return
-        verbose = msg.get('verbose', False)
-        content = dict()
-        for t in targets:
-            queue = self.queues[t]
-            completed = self.completed[t]
-            if not verbose:
-                queue = len(queue)
-                completed = len(completed)
-            content[str(t)] = {'queue': queue, 'completed': completed }
-            # pending
-        self.session.send(self.clientele, "queue_reply", content=content, ident=client_id)
-    
     def purge_results(self, client_id, msg):
         content = msg['content']
         msg_ids = content.get('msg_ids', [])
@@ -769,4 +751,171 b' class Controller(object):'
         return eid, msg
     
 
-        
  No newline at end of file
+#--------------------
+# Entry Point
+#--------------------
+
+def main():
+    import time
+    from multiprocessing import Process
+    
+    from zmq.eventloop.zmqstream import ZMQStream
+    from zmq.devices import ProcessMonitoredQueue
+    from zmq.log import handlers
+    
+    import streamsession as session
+    import heartmonitor
+    from scheduler import launch_scheduler
+    
+    parser = make_argument_parser()
+    
+    parser.add_argument('--client', type=int, metavar='PORT', default=0,
+                        help='set the XREP port for clients [default: random]')
+    parser.add_argument('--notice', type=int, metavar='PORT', default=0,
+                        help='set the PUB socket for registration notification [default: random]')
+    parser.add_argument('--hb', type=str, metavar='PORTS',
+                        help='set the 2 ports for heartbeats [default: random]')
+    parser.add_argument('--ping', type=int, default=3000,
+                        help='set the heartbeat period in ms [default: 3000]')
+    parser.add_argument('--monitor', type=int, metavar='PORT', default=0,
+                        help='set the SUB port for queue monitoring [default: random]')
+    parser.add_argument('--mux', type=str, metavar='PORTS',
+                        help='set the XREP ports for the MUX queue [default: random]')
+    parser.add_argument('--task', type=str, metavar='PORTS',
+                        help='set the XREP/XREQ ports for the task queue [default: random]')
+    parser.add_argument('--control', type=str, metavar='PORTS',
+                        help='set the XREP ports for the control queue [default: random]')
+    parser.add_argument('--scheduler', type=str, default='pure',
+                        choices = ['pure', 'lru', 'plainrandom', 'weighted', 'twobin','leastload'],
+                        help='select the task scheduler  [default: pure ZMQ]')
+    
+    args = parser.parse_args()
+    
+    if args.url:
+        args.transport,iface = args.url.split('://')
+        iface = iface.split(':')
+        args.ip = iface[0]
+        if iface[1]:
+            args.regport = iface[1]
+    
+    iface="%s://%s"%(args.transport,args.ip)+':%i'
+    
+    random_ports = 0
+    if args.hb:
+        hb = split_ports(args.hb, 2)
+    else:
+        hb = select_random_ports(2)
+    if args.mux:
+        mux = split_ports(args.mux, 2)
+    else:
+        mux = None
+        random_ports += 2
+    if args.task:
+        task = split_ports(args.task, 2)
+    else:
+        task = None
+        random_ports += 2
+    if args.control:
+        control = split_ports(args.control, 2)
+    else:
+        control = None
+        random_ports += 2
+    
+    ctx = zmq.Context()
+    loop = ioloop.IOLoop.instance()
+    
+    # setup logging
+    connect_logger(ctx, iface%args.logport, root="controller", loglevel=args.loglevel)
+    
+    # Registrar socket
+    reg = ZMQStream(ctx.socket(zmq.XREP), loop)
+    regport = bind_port(reg, args.ip, args.regport)
+    
+    ### Engine connections ###
+    
+    # heartbeat
+    hpub = ctx.socket(zmq.PUB)
+    bind_port(hpub, args.ip, hb[0])
+    hrep = ctx.socket(zmq.XREP)
+    bind_port(hrep, args.ip, hb[1])
+    
+    hmon = heartmonitor.HeartMonitor(loop, ZMQStream(hpub,loop), ZMQStream(hrep,loop),args.ping)
+    hmon.start()
+    
+    ### Client connections ###
+    # Clientele socket
+    c = ZMQStream(ctx.socket(zmq.XREP), loop)
+    cport = bind_port(c, args.ip, args.client)
+    # Notifier socket
+    n = ZMQStream(ctx.socket(zmq.PUB), loop)
+    nport = bind_port(n, args.ip, args.notice)
+    
+    thesession = session.StreamSession(username=args.ident or "controller")
+    
+    ### build and launch the queues ###
+    
+    # monitor socket
+    sub = ctx.socket(zmq.SUB)
+    sub.setsockopt(zmq.SUBSCRIBE, "")
+    monport = bind_port(sub, args.ip, args.monitor)
+    sub = ZMQStream(sub, loop)
+    
+    ports = select_random_ports(random_ports)
+    # Multiplexer Queue (in a Process)
+    if not mux:
+        mux = (ports.pop(),ports.pop())
+    q = ProcessMonitoredQueue(zmq.XREP, zmq.XREP, zmq.PUB, 'in', 'out')
+    q.bind_in(iface%mux[0])
+    q.bind_out(iface%mux[1])
+    q.connect_mon(iface%monport)
+    q.daemon=True
+    q.start()
+    
+    # Control Queue (in a Process)
+    if not control:
+        control = (ports.pop(),ports.pop())
+    q = ProcessMonitoredQueue(zmq.XREP, zmq.XREP, zmq.PUB, 'incontrol', 'outcontrol')
+    q.bind_in(iface%control[0])
+    q.bind_out(iface%control[1])
+    q.connect_mon(iface%monport)
+    q.daemon=True
+    q.start()
+    
+    # Task Queue (in a Process)
+    if not task:
+        task = (ports.pop(),ports.pop())
+    if args.scheduler == 'pure':
+        q = ProcessMonitoredQueue(zmq.XREP, zmq.XREQ, zmq.PUB, 'intask', 'outtask')
+        q.bind_in(iface%task[0])
+        q.bind_out(iface%task[1])
+        q.connect_mon(iface%monport)
+        q.daemon=True
+        q.start()
+    else:
+        sargs = (iface%task[0],iface%task[1],iface%monport,iface%nport,args.scheduler)
+        print sargs
+        p = Process(target=launch_scheduler, args=sargs)
+        p.daemon=True
+        p.start()
+    
+    time.sleep(.25)
+    
+    # build connection dicts
+    engine_addrs = {
+        'control' : iface%control[1],
+        'queue': iface%mux[1],
+        'heartbeat': (iface%hb[0], iface%hb[1]),
+        'task' : iface%task[1],
+        'monitor' : iface%monport,
+        }
+    
+    client_addrs = {
+        'control' : iface%control[0],
+        'query': iface%cport,
+        'queue': iface%mux[0],
+        'task' : iface%task[0],
+        'notification': iface%nport
+        }
+    con = Controller(loop, thesession, sub, reg, hmon, c, n, None, engine_addrs, client_addrs)
+    loop.start()
+    

IPython/zmq/parallel/engine.py

@@ -16,6 +16,7 b' from streamsession import Message, StreamSession'
 from client import Client
 import streamkernel as kernel
 import heartmonitor
+from entry_point import make_argument_parser, connect_logger
 # import taskthread
 # from log import logger
 
@@ -75,6 +76,7 b' class Engine(object):'
             if task_addr:
                 # task as stream:
                 task = self.context.socket(zmq.PAIR)
+                task.setsockopt(zmq.IDENTITY, self.ident)
                 task.connect(str(task_addr))
                 self.task_stream = zmqstream.ZMQStream(task, self.loop)
                 # TaskThread:
@@ -113,16 +115,28 b' class Engine(object):'
         self.register()
         
 
-if __name__ == '__main__':
+def main():
     
+    parser = make_argument_parser()
+    
+    args = parser.parse_args()
+    
+    if args.url:
+        args.transport,iface = args.url.split('://')
+        iface = iface.split(':')
+        args.ip = iface[0]
+        if iface[1]:
+            args.regport = iface[1]
+    
+    iface="%s://%s"%(args.transport,args.ip)+':%i'
     loop = ioloop.IOLoop.instance()
     session = StreamSession()
     ctx = zmq.Context()
 
-    ip = '127.0.0.1'
-    reg_port = 10101
-    connection = ('tcp://%s' % ip) + ':%i'
-    reg_conn = connection % reg_port
+    # setup logging
+    connect_logger(ctx, iface%args.logport, root="engine", loglevel=args.loglevel)
+    
+    reg_conn = iface % args.regport
     print reg_conn
     print >>sys.__stdout__, "Starting the engine..."
     
@@ -130,12 +144,8 b" if __name__ == '__main__':"
     reg.connect(reg_conn)
     reg = zmqstream.ZMQStream(reg, loop)
     client = Client(reg_conn)
-    if len(sys.argv) > 1:
-        queue_id=sys.argv[1]
-    else:
-        queue_id = None
     
-    e = Engine(ctx, loop, session, reg, client, queue_id)
-    dc = ioloop.DelayedCallback(e.start, 500, loop)
+    e = Engine(ctx, loop, session, reg, client, args.ident)
+    dc = ioloop.DelayedCallback(e.start, 100, loop)
     dc.start()
     loop.start()
  No newline at end of file

IPython/zmq/parallel/streamkernel.py

@@ -191,7 +191,7 b' class Kernel(object):'
             # self.reply_socket.send(ident,zmq.SNDMORE)
             # self.reply_socket.send_json(reply_msg)
             reply_msg = self.session.send(stream, reply_type, 
-                        content={'status' : 'aborted'}, parent=msg, ident=idents)
+                        content={'status' : 'aborted'}, parent=msg, ident=idents)[0]
             print>>sys.__stdout__, Message(reply_msg)
             # We need to wait a bit for requests to come in. This can probably
             # be set shorter for true asynchronous clients.
@@ -208,8 +208,8 b' class Kernel(object):'
             self.aborted.add(str(mid))
         
         content = dict(status='ok')
-        reply_msg = self.session.send(stream, 'abort_reply', content=content, parent=parent,
-                                                                    ident=ident)
+        reply_msg = self.session.send(stream, 'abort_reply', content=content, 
+                parent=parent, ident=ident)[0]
         print>>sys.__stdout__, Message(reply_msg)
     
     def kill_request(self, stream, idents, parent):
@@ -312,7 +312,7 b' class Kernel(object):'
         # self.reply_socket.send(ident, zmq.SNDMORE)
         # self.reply_socket.send_json(reply_msg)
         reply_msg = self.session.send(stream, u'execute_reply', reply_content, parent=parent, ident=ident)
-        # print>>sys.__stdout__, Message(reply_msg)
+        print>>sys.__stdout__, Message(reply_msg)
         if reply_msg['content']['status'] == u'error':
             self.abort_queues()
 
@@ -327,6 +327,7 b' class Kernel(object):'
         return self.completer.complete(msg.content.line, msg.content.text)
     
     def apply_request(self, stream, ident, parent):
+        print parent
         try:
             content = parent[u'content']
             bufs = parent[u'buffers']
@@ -399,7 +400,7 b' class Kernel(object):'
         # self.reply_socket.send_json(reply_msg)
         reply_msg = self.session.send(stream, u'apply_reply', reply_content, 
                     parent=parent, ident=ident,buffers=result_buf, subheader=sub)
-        # print>>sys.__stdout__, Message(reply_msg)
+        print>>sys.__stdout__, Message(reply_msg)
         # if reply_msg['content']['status'] == u'error':
         #     self.abort_queues()
     

examples/zmqontroller/config.py

@@ -16,7 +16,7 b' config = {'
     'clientport': 10201,
     'notifierport': 10202,
     
-    'logport': 20201
+    'logport': 20202
 }
 
 

examples/zmqontroller/controller.py

@@ -18,7 +18,7 b' from IPython.zmq.parallel import controller, heartmonitor, streamsession as sess'
 
 def setup():
     """setup a basic controller and open client,registrar, and logging ports. Start the Queue and the heartbeat"""
-    ctx = zmq.Context(1)
+    ctx = zmq.Context()
     loop = ioloop.IOLoop.instance()
     
     # port config

examples/zmqontroller/logwatcher.py

@@ -20,7 +20,7 b''
 #    along with this program.  If not, see <http://www.gnu.org/licenses/>.
 
 import zmq
-logport = 20201
+logport = 20202
 def main(topics, addrs):
     
     context = zmq.Context()

setup.py

@@ -215,9 +215,8 b" if 'setuptools' in sys.modules:"
             'ipython = IPython.frontend.terminal.ipapp:launch_new_instance',
             'ipython-qtconsole = IPython.frontend.qt.console.ipythonqt:main',
             'pycolor = IPython.utils.PyColorize:main',
-            # 'ipcontroller = IPython.kernel.ipcontrollerapp:launch_new_instance',
-            # 'ipengine = IPython.kernel.ipengineapp:launch_new_instance',
-            # 'ipcluster = IPython.kernel.ipclusterapp:launch_new_instance',
+            'ipcontrollerz = IPython.zmq.parallel.controller:main',
+            'ipenginez = IPython.zmq.parallel.engine:main',
             'iptest = IPython.testing.iptest:main',
             'irunner = IPython.lib.irunner:main'
         ]