Merge pull request #24 from quantopian/exception_reporting

Exception reporting

zipline/component.py

@@ -24,7 +24,7 @@ from datetime import datetime
 import zipline.util as qutil
 from zipline.gpoll import _Poller as GeventPoller
 from zipline.protocol import CONTROL_PROTOCOL, COMPONENT_STATE, \
-    COMPONENT_FAILURE, BACKTEST_STATE
+    COMPONENT_FAILURE, BACKTEST_STATE, CONTROL_FRAME
 
 
 class Component(object):
@@ -42,9 +42,9 @@ class Component(object):
 
     :param data_address: socket address used for data sources to stream
                          their records. Will be used in PUSH/PULL sockets
-                         between data sources and a ParallelBuffer (aka
-                         the Feed). Bind will always be on the PULL side
-                         (we always have N producers and 1 consumer)
+                         between data sources and a Feed. Bind will always
+                         be on the PULL side (we always have N producers and
+                         1 consumer)
 
     :param feed_address: socket address used to publish consolidated feed
                          from serialization of data sources
@@ -53,9 +53,9 @@ class Component(object):
 
     :param merge_address: socket address used to publish transformed
                           values.  will be used in PUSH/PULL from many
-                          transforms to one MergedParallelBuffer (aka the
-                          Merge). Bind will always be on the PULL side (we
-                          always have N producers and 1 consumer)
+                          transforms to one Merge Bind will always be on
+                          the PULL side (we always have N producers and
+                          1 consumer)
 
     :param result_address: socket address used to publish merged data
                            source feed and transforms to clients will be
@@ -201,28 +201,25 @@ class Component(object):
         debug since it mucks up your stacktraces.
         """
 
-        fail = None
-
         if catch_exceptions:
             try:
                 self._run()
             except Exception as exc:
-                # TODO, we want to do this grab the stack
-                # frame so we can preserve stacktraces when we
-                # reraise the exception.
+                exc_info = sys.exc_info()
                 self.signal_exception(exc)
-                fail = exc
+
+                # Reraise the exception
+                raise exc_info[0], exc_info[1], exc_info[2]
             finally:
 
                 self.shutdown()
                 self.teardown_sockets()
-        else:
-            try:
-                self._run()
-            finally:
-                self.shutdown()
-                self.teardown_sockets()
-
+        #else:
+            #try:
+                #self._run()
+            #finally:
+                #self.shutdown()
+                #self.teardown_sockets()
 
     def working(self):
         """
@@ -234,7 +231,7 @@ class Component(object):
         """
         return (not self.done)
 
-    def loop(self):
+    def loop(self, lockstep=True):
         """
         Loop to do work while we still have work to do.
         """
@@ -294,6 +291,12 @@ class Component(object):
     # ----------------------
 
     def signal_exception(self, exc=None, scope=None):
+        """
+        This is *very* important error tracking handler.
+
+        Will inform the system that the component has failed and
+        how it has failed.
+        """
 
         if scope == 'algo':
             self.error_state = COMPONENT_FAILURE.ALGOEXCEPT
@@ -310,6 +313,12 @@ class Component(object):
         exc_type, exc_value, exc_traceback = sys.exc_info()
         self.stack_trace = exc_traceback
 
+        exception_frame = CONTROL_FRAME(
+            CONTROL_PROTOCOL.EXCEPTION,
+            str(exc)
+        )
+        self.control_out.send(exception_frame)
+
         qutil.LOGGER.exception("Unexpected error in run for {id}.".format(id=self.get_id))
 
     def signal_done(self):
@@ -326,10 +335,19 @@ class Component(object):
         # TODO: proper framing
         self.sync_socket.send(self.get_id + ":" + str(CONTROL_PROTOCOL.DONE))
 
+        #notify controller we're done
+        done_frame = CONTROL_FRAME(
+            CONTROL_PROTOCOL.DONE,
+            ''
+        )
+        self.control_out.send(done_frame)
+
         self.receive_sync_ack()
         #notify internal work look that we're done
         self.done = True # TODO: use state flag
 
+        qutil.LOGGER.info("[%s] DONE" % self.get_id)
+
     # -----------
     #  Messaging
     # -----------
@@ -347,13 +365,15 @@ class Component(object):
     def receive_sync_ack(self):
         """
         Wait for synchronization reply from the host.
+
+        DEPRECATED, left in for compatability for now.
         """
 
         socks = dict(self.sync_poller.poll(self.heartbeat_timeout))
         if self.sync_socket in socks and socks[self.sync_socket] == self.zmq.POLLIN:
             message = self.sync_socket.recv()
-        else:
-            raise Exception("Sync ack timed out on response for {id}".format(id=self.get_id))
+        #else:
+            #raise Exception("Sync ack timed out on response for {id}".format(id=self.get_id))
 
     def bind_data(self):
         return self.bind_pull_socket(self.addresses['data_address'])
@@ -427,8 +447,14 @@ class Component(object):
         if not self.controller:
             return
 
-        self.control_out = self.controller.message_sender(context=self.context)
-        self.control_in = self.controller.message_listener(context=self.context)
+        self.control_out = self.controller.message_sender(
+            identity = self.get_id,
+            context  = self.context,
+        )
+
+        self.control_in = self.controller.message_listener(
+            context = self.context
+        )
 
         self.poll.register(self.control_in, self.zmq.POLLIN)
         self.sockets.extend([self.control_in, self.control_out])
@@ -436,6 +462,8 @@ class Component(object):
     def setup_sync(self):
         """
         Setup the sync socket and poller. ( Connect )
+
+        DEPRECATED, left in for compatability for now.
         """
 
         qutil.LOGGER.debug("Connecting sync client for {id}".format(id=self.get_id))

zipline/finance/trading.py

@@ -3,6 +3,7 @@ import pytz
 import math
 import pandas
 
+# from gevent.select import select
 from zmq.core.poll import select
 
 import zipline.messaging as qmsg
@@ -200,6 +201,9 @@ class OrderDataSource(qmsg.DataSource):
         #pull all orders from client.
         orders = []
         count = 0
+
+        # TODO : this can be written in a concurrency agnostic
+        # way... have a chat with Fawce about this ~Steve
         while True:
                         
             (rlist, wlist, xlist) = select(

zipline/lines.py

@@ -149,6 +149,11 @@ class SimulatedTrading(object):
             sockets[7],
             logging = qutil.LOGGER
         )
+
+        # TODO: Not freeform
+        self.con.manage(
+            'freeform'
+        )
         
         self.started = False
         

zipline/messaging.py

@@ -8,7 +8,7 @@ import zipline.util as qutil
 from zipline.component import Component
 import zipline.protocol as zp
 from zipline.protocol import CONTROL_PROTOCOL, COMPONENT_TYPE, \
-    COMPONENT_STATE
+    COMPONENT_STATE, CONTROL_FRAME, CONTROL_UNFRAME
 
 class ComponentHost(Component):
     """
@@ -39,8 +39,8 @@ class ComponentHost(Component):
         self.sync_register  = {}
         self.timeout        = datetime.timedelta(seconds=5)
 
-        self.feed           = ParallelBuffer()
-        self.merge          = MergedParallelBuffer()
+        self.feed           = Feed()
+        self.merge          = Merge()
         self.passthrough    = PassthroughTransform()
         self.controller     = None
 
@@ -56,6 +56,7 @@ class ComponentHost(Component):
             raise Exception("There can be only one!")
 
         self.controller = controller
+        self.controller.zmq_flavor = self.zmq_flavor
 
         # Propogate the controller to all the subcomponents
         for component in self.components.itervalues():
@@ -112,6 +113,10 @@ class ComponentHost(Component):
         self.launch_controller()
 
     def is_timed_out(self):
+        """
+        DEPRECATED, left in for compatability for now.
+        """
+
         cur_time = datetime.datetime.utcnow()
 
         if len(self.components) == 0:
@@ -128,7 +133,7 @@ class ComponentHost(Component):
 
         return False
 
-    def loop(self):
+    def loop(self, lockstep=True):
 
         while not self.is_timed_out():
             # wait for synchronization request
@@ -144,7 +149,7 @@ class ComponentHost(Component):
                     self.signal_exception(exc)
 
                 if status == str(CONTROL_PROTOCOL.DONE): # TODO: other way around
-                    qutil.LOGGER.info("{id} is DONE".format(id=sync_id))
+                    #qutil.LOGGER.debug("{id} is DONE".format(id=sync_id))
                     self.unregister_component(sync_id)
                     self.state_flag = COMPONENT_STATE.DONE
                 else:
@@ -168,7 +173,7 @@ class ComponentHost(Component):
         raise NotImplementedError
 
 
-class ParallelBuffer(Component):
+class Feed(Component):
     """
     Connects to N PULL sockets, publishing all messages received to a PUB
     socket.  Published messages are guaranteed to be in chronological order
@@ -211,8 +216,29 @@ class ParallelBuffer(Component):
         # wait for synchronization reply from the host
         socks = dict(self.poll.poll(self.heartbeat_timeout)) #timeout after 2 seconds.
 
+        # TODO: Abstract this out, maybe on base component
         if self.control_in in socks and socks[self.control_in] == self.zmq.POLLIN:
             msg = self.control_in.recv()
+            event, payload = CONTROL_UNFRAME(msg)
+
+            # -- Heartbeat --
+            if event == CONTROL_PROTOCOL.HEARTBEAT:
+                # Heart outgoing
+                heartbeat_frame = CONTROL_FRAME(
+                    CONTROL_PROTOCOL.OK,
+                    payload
+                )
+                self.control_out.send(heartbeat_frame)
+
+            # -- Soft Kill --
+            elif event == CONTROL_PROTOCOL.SHUTDOWN:
+                self.done()
+                self.shutdown()
+
+            # -- Hard Kill --
+            elif event == CONTROL_PROTOCOL.KILL:
+                self.kill()
+
 
         if self.pull_socket in socks and socks[self.pull_socket] == self.zmq.POLLIN:
             message = self.pull_socket.recv()
@@ -240,13 +266,11 @@ class ParallelBuffer(Component):
                 except zp.INVALID_DATASOURCE_FRAME as exc:
                     return self.signal_exception(exc)
 
-    #
     def unframe(self, msg):
         return zp.DATASOURCE_UNFRAME(msg)
 
     def frame(self, event):
         return zp.FEED_FRAME(event)
-    
 
     # -------------
     # Flow Control
@@ -343,13 +367,13 @@ class ParallelBuffer(Component):
         return len(self.data_buffer)
 
 
-class MergedParallelBuffer(ParallelBuffer):
+class Merge(Feed):
     """
     Merges multiple streams of events into single messages.
     """
 
     def __init__(self):
-        ParallelBuffer.__init__(self)
+        Feed.__init__(self)
 
         self.init()
 
@@ -463,11 +487,32 @@ class BaseTransform(Component):
         """
         socks = dict(self.poll.poll(self.heartbeat_timeout))
 
+        # TODO: Abstract this out, maybe on base component
         if self.control_in in socks and socks[self.control_in] == self.zmq.POLLIN:
             msg = self.control_in.recv()
+            event, payload = CONTROL_UNFRAME(msg)
+
+            # -- Heartbeat --
+            if event == CONTROL_PROTOCOL.HEARTBEAT:
+                # Heart outgoing
+                heartbeat_frame = CONTROL_FRAME(
+                    CONTROL_PROTOCOL.OK,
+                    payload
+                )
+                self.control_out.send(heartbeat_frame)
+
+            # -- Soft Kill --
+            elif event == CONTROL_PROTOCOL.SHUTDOWN:
+                self.done()
+                self.shutdown()
+
+            # -- Hard Kill --
+            elif event == CONTROL_PROTOCOL.KILL:
+                self.kill()
 
         if self.feed_socket in socks and socks[self.feed_socket] == self.zmq.POLLIN:
             message = self.feed_socket.recv()
+
             if message == str(CONTROL_PROTOCOL.DONE):
                 self.signal_done()
                 return

zipline/monitor.py

@@ -1,29 +1,133 @@
+import time
+import gevent
+import itertools
+# pyzmq
 import zmq
-from zipline.protocol import CONTROL_PROTOCOL, CONTROL_FRAME
+# gevent_zeromq
+import gevent_zeromq
+# zmq_ctypes
+#import zmq_ctypes
+
+from protocol import CONTROL_PROTOCOL, CONTROL_FRAME, \
+    CONTROL_UNFRAME, CONTROL_STATES, INVALID_CONTROL_FRAME
+
+from gpoll import _Poller as GeventPoller
+
+# Roll Call ( Discovery )
+# -----------------------
+#
+#       Controller ( 'foo', 'bar', 'fizz', 'pop' )
+#    ------------------
+#    |     |     |     |
+#  +---+
+#  | 0 |   ?     ?     ?
+#  +---+
+#    |
+#   IDENTITY: foo
+#   get message: PROTOCOL.HEARTBEAT
+#   reply with PROTOCOL.OK
+#
+#       Controller topology = ( 'foo', 'bar', 'fizz', 'pop' )
+#       'foo' in topology = YES ->
+#           track 'foo'
+#    ------------------
+#    |     |     |     |
+#  +---+
+#  | 1 |   ?     ?     ?
+#  +---+
+
+# Heartbeating
+# ------------
+#
+#       Controller ( time = 2.717828 )
+#    ------------------
+#    |     |     |     |
+#  +---+ +---+ +---+ +---+
+#  | 0 | | 0 | | 0 | | 0 |
+#  +---+ +---+ +---+ +---+
+#    |
+#   IDENTITY: foo
+#   get message: time = 2.717828
+#   reply with [ foo, 2.71828 ]
+#
+#       Controller ( foo.status = OK )
+#    ------------------
+#    |     |     |     |
+#  +---+ +---+ +---+ +---+
+#  | 1 | | 0 | | 0 | | 0 |
+#  +---+ +---+ +---+ +---+
+#    |
+#  Controller tracks this node as good
+#  for this heartbeat
+
+# Shutdown
+# --------
+#
+#       Controller ( state = RUNNING )
+#    ------------------
+#    |     |     |     |
+#  +---+ +---+ +---+ +---+
+#  | 1 | | 1 | | 1 | | 1 |
+#  +---+ +---+ +---+ +---+
+#    |
+#   IDENTITY: foo
+#   send [ DONE ]
+
+#       Controller ( state = SHUTDOWN )
+#       Controller topology.remove('foo')
+#    ------------------
+#          |     |     |
+#  +---+ +---+ +---+ +---+
+#  |   | | 1 | | 1 | | 1 |
+#  +---+ +---+ +---+ +---+
+#    |
+#   IDENTITY: foo
+#   yield, stop sending messages
+
+# Termination
+# ------------
+#
+#       Controller ( state = TERMINATE )
+#    ------------------
+#    |     |     |     |
+#  +---+ +---+ +---+ +---+
+#  | 1 | | 1 | | 1 | | 1 |
+#  +---+ +---+ +---+ +---+
+#    |
+#   get message PROTOCOL.KILL
+
+#       Controller ( state = TERMINATE )
+#    ------------------
+#    |     |     |     |
+#  +---+ +---+ +---+ +---+
+#  | 0 | | 0 | | 0 | | 0 |
+#  +---+ +---+ +---+ +---+
+
+class UnknownChatter(Exception):
+    def __init__(self, name):
+        self.named = name
+    def __str__(self):
+        return """Component calling itself "%s" talking on unexpected channel"""\
+            % self.named
 
 class Controller(object):
     """
     A N to M messaging system for inter component communication.
-    Ostensibly a broker of sorts. Putting messages to the broker
-    is durable, if the broker goes down messages will queue up
-    until the HWM and then go out when the new broker comes up.
 
-    The other end is not durable, it is simply PUB/SUB which has
-    the benefit of of allowing more fluid time evolution of the
-    whole system since the messaging passing topology will not
-    alter itself as a result of more nodes listening.
-
-    The actual brokerin' is either a Python loop ( slow ) or a
-    zmq.FORWARDER device ( fast ).
-
-    :param pull_socket: Socket to subscribe to for republication of
-                        published messages. The endpoint for
-                        :func message_sender: .
     :param pub_socket: Socket to publish messages, the starting
                        point of :func message_listener: .
+
+    :param route_socket: Socket to listen for status updates for
+                         the individual components.
+                         :func message_sender: .
+
     :param logging: Logging interface for tracking broker state
         Defaults to None
 
+    Topology is the set of components we expect to show up.
+    States are the transitions the sytems go through. The
+    simplest is from RUNNING -> NOT RUNNING .
+
     Usage::
 
         controller = Controller(
@@ -33,169 +137,349 @@ class Controller(object):
 
         # typically you'd want to run this async to your main
         # program since it blocks indefinetely.
-        controller.run()
-
-
-        sub = self.controller.message_listener()
-        push = self.controller.message_sender()
-
-        push.send('DIE')
-        sub.recv()
+        controller.manage(
+            [ TOPOLOGY ]
+            [ STATES ]
+        )
 
     """
 
     debug = False
+    period = 1
 
-    def __init__(self, pull_socket, pub_socket, logging = None):
+    def __init__(self, pub_socket, route_socket, logging = None):
+
+        self.context = None
+        self.zmq = None
+        self.zmq_poller = None
 
-        self._ctx = None
         polling = False
 
+        self.polling = polling
+
+        self.tracked = set()
+        self.responses = set()
+
+        self.ctime = 0
+        self.tic = time.time()
+        self.freeform = False
+
         self.associated = []
 
-        self.pull_socket = pull_socket
-        self.push_socket = pull_socket # same port
         self.pub_socket = pub_socket
-        self.sub_socket = pub_socket # same port
-        self.terminated = False
+        self.route_socket = route_socket
 
         if logging:
             self.logging = logging
-            self.dologging = True
         else:
-            self.logging = False
-            self.dologging = False
+            import util as qutil
+            self.logging = qutil.LOGGER
 
-        self.success = 0
-        self.failed = 0
+    def init_zmq(self, flavor):
 
-    def run(self, debug=False, context=None):
+        assert self.zmq_flavor in ['thread', 'mp', 'green']
+
+        if flavor == 'mp':
+            self.zmq = zmq
+            self.context = self.zmq.Context()
+            self.zmq_poller = self.zmq.Poller
+            return
+        if flavor == 'thread':
+            self.zmq = zmq
+            self.context = self.zmq.Context.instance()
+            self.zmq_poller = self.zmq.Poller
+            return
+        if flavor == 'green':
+            self.zmq = gevent_zeromq.zmq
+            self.context = self.zmq.Context.instance()
+            self.zmq_poller = GeventPoller
+            return
+        if flavor == 'pypy':
+            self.zmq = zmq
+            self.context = self.zmq.Context.instance()
+            self.zmq_poller = self.zmq.Poller
+            return
+
+    def manage(self, topology, states=None, context=None):
         """
-        Run's the loop for the broker.
+        Give the controller a set set of components to manage and
+        a set of state transitions for the entire system.
         """
+
+        # A freeform topology is where we heartbeat with anything
+        # that shows up.
+        if topology == 'freeform':
+            self.freeform = True
+            self.topology = frozenset([])
+        else:
+            self.freeform = False
+            self.topology = frozenset(topology)
+
+        default_states = [
+            CONTROL_STATES.RUNNING,
+            CONTROL_STATES.SHUTDOWN,
+            CONTROL_STATES.TERMINATE,
+        ]
+
+        self.states = states or default_states
         self.polling = True
 
-        if not context:
-            self._ctx = zmq.Context.instance()
-        else:
-            self._ctx = context
+        # Start off in RUNNING, state
+        self.state = self.states[0]
 
-        #if not debug:
-            #return self._poll_fast() # the c loop
-        #else:
-        return self._poll() # use a python loop
+    def run(self):
+        self.init_zmq(self.zmq_flavor)
 
-    def _poll_fast(self):
+        try:
+            return self._poll() # use a python loop
+        except KeyboardInterrupt:
+            self.logging.info('Shutdown event loop')
+
+    def log_status(self):
         """
-        C version of the polling forwarder.
+        Snapshot of the tracked components at every period.
         """
-        self.pull = self._ctx.socket(zmq.PULL)
-        self.pub = self._ctx.socket(zmq.PUB)
+        #self.logging.info("[Controller] Tracking : %s" % ([c for c in self.tracked],))
+        pass
 
-        self.pull.bind(self.pull_socket)
-        self.pub.bind(self.pub_socket)
+    # -------------
+    # Publications
+    # -------------
 
-        zmq.device(zmq.FORWARDER, self.pull, self.pub)
+    def send_heart(self):
+        heartbeat_frame = CONTROL_FRAME(
+            CONTROL_PROTOCOL.HEARTBEAT,
+            str(self.ctime)
+        )
+        self.pub.send(heartbeat_frame)
+
+    def send_hardkill(self):
+        kill_frame = CONTROL_FRAME(
+            CONTROL_PROTOCOL.KILL,
+            ''
+        )
+        self.pub.send(kill_frame)
+
+    def send_softkill(self):
+        soft_frame = CONTROL_FRAME(
+            CONTROL_PROTOCOL.KILL,
+            ''
+        )
+        self.pub.send(soft_frame)
+
+    # -----------
+    # Event Loops
+    # -----------
 
     def _poll(self):
-        """
-        Python version of the polling forwarder. With logging,
-        mostly used for debugging.
-        """
 
-        self.pull = self._ctx.socket(zmq.PULL)
-        self.pub = self._ctx.socket(zmq.PUB)
-
-        self.pull.bind(self.pull_socket)
+        self.pub = self.context.socket(self.zmq.PUB)
         self.pub.bind(self.pub_socket)
 
-        self.associated.extend([self.pull, self.pub])
+        self.router = self.context.socket(self.zmq.ROUTER)
+        self.router.bind(self.route_socket)
 
-        while self.polling:
-            try:
-                msg = self.pull.recv()
-                self.pub.send(msg)
-            except KeyboardInterrupt:
-                self.polling = False
-                break
-            except zmq.ZMQError:
-                self.polling = False
-                break
-            except Exception as e:
-                # Its common to wrap these in wildcard exceptions so
-                # that we don't loose messages, ever
-                if self.logging:
-                    self.logging.error(str(e))
-                self.failed += 1
-                continue
+        self.associated.extend([self.pub, self.router])
 
+        poller = self.zmq.Poller()
+        poller.register(self.router, self.zmq.POLLIN)
+
+        buffer = []
+
+        for i in itertools.count(0):
+            self.log_status()
+            self.responses = set()
+
+            self.ctime = time.time()
+            self.send_heart()
+
+            while self.polling:
+                # Reset the responses for this cycle
+
+                socks = dict(poller.poll(self.period))
+                tic = time.time()
+
+                if tic - self.ctime > self.period:
+                    break
+
+                if self.router in socks and socks[self.router] == self.zmq.POLLIN:
+                    rawmessage = self.router.recv()
+
+                    if rawmessage:
+                        buffer.append(rawmessage)
+
+                    try:
+                        if not self.router.getsockopt(self.zmq.RCVMORE):
+                            self.handle_recv(buffer[:])
+                            buffer = []
+                    except INVALID_CONTROL_FRAME:
+                        self.logging.error('Invalid frame', rawmessage)
+                        pass
+
+            self.beat()
+
+            if self.zmq_flavor == 'green':
+                gevent.sleep(0)
+
+            if not self.polling:
+                break
+
+        # After loop exits
         self.terminated = True
 
+    def beat(self):
+
+        # These the set overloaded operations
+        # A & B ~ set.intersection
+        # A - B ~ set.difference
+
+        # * good - Components we are currently tracking and who just sent
+        #          us back the right response.
+        # * bad - Components we are currently tracking but who did not
+        #         send us back a response.
+        # * new - Components we haven't heard from yet, but sent back the
+        #         right response.
+
+        good = self.tracked   & self.responses
+        bad  = self.tracked   - good
+        new  = self.responses - good
+
+        for component in new:
+            self.new(component)
+
+        for component in bad:
+            self.fail(component)
+
+    # ------------------
+    # Component Handlers
+    # ------------------
+
+    # The various "states of being that a component can inform us
+    # of
+    def new(self, component):
+        self.logging.info('[Controller] Alive "%s" ' % component)
+
+        if component in self.topology or self.freeform:
+            self.tracked.add(component)
+        else:
+            # Some sort of socket collision has occured, this is
+            # a very bad failure mode.
+            raise UnknownChatter(component)
+
+    def fail(self, component):
+        self.logging.info('[Controller] Component "%s" timed out' % component)
+        self.tracked.remove(component)
+
+    def done(self, component):
+        # TODO: This will be what we ship off to vbench at some
+        # point...
+        # print component finished at self.ctime
+        self.logging.info('[Controller] Component "%s" done.' % component)
+
+    def exception(self, component, failure):
+        self.logging.error('Component "%s" in exception state' % component)
+
+    # -----------------
+    # Protocol Handling
+    # -----------------
+
+    def handle_recv(self, msg):
+        """
+        Check for proper framing at the transport layer.
+        Seperates the proper frames from anything else that might
+        be coming over the wire. Which shouldn't happen ...  right?
+        """
+        identity = msg[0]
+        id, status = CONTROL_UNFRAME(msg[1])
+
+        # A component is telling us its alive:
+        if id is CONTROL_PROTOCOL.OK:
+
+            if status == str(self.ctime):
+                self.responses.add(identity)
+            else:
+                # Otherwise its something weird and we don't know
+                # what to do so just say so
+                self.logging.error("Weird stuff happened: %s" % msg)
+
+        # A component is telling us it failed, and how
+        if id is CONTROL_PROTOCOL.EXCEPTION:
+            self.exception(identity, status)
+
+        # A component is telling us its done with work and won't
+        # be talking to us anymore
+        if id is CONTROL_PROTOCOL.DONE:
+            self.done(identity)
+
     # -------------------
     # Hooks for Endpoints
     # -------------------
 
-    def message_sender(self, context=None):
+    # These are all connects so no complex allocation logic is
+    # needed. Dealers and Subscribers can all come and go as a
+    # function of time without impacting flow of the whole
+    # system.
+
+    def message_sender(self, identity, context = None):
         """
         Spin off a socket used for sending messages to this
         controller.
         """
 
         if not context:
-            context = zmq.Context.instance()
+            context = self.zmq.Context.instance()
+
+        s = context.socket(zmq.DEALER)
+        s.setsockopt(zmq.IDENTITY, identity)
+        s.connect(self.route_socket)
 
-        s = context.socket(zmq.PUSH)
-        s.connect(self.push_socket)
         self.associated.append(s)
         return s
 
-    def message_listener(self, context = None, filters=None):
+    def message_listener(self, context = None):
         """
         Spin off a socket used for receiving messages from this
         controller.
         """
 
         if not context:
-            context = zmq.Context.instance()
+            context = self.zmq.Context.instance()
 
         s = context.socket(zmq.SUB)
-        s.connect(self.sub_socket)
-        s.setsockopt(zmq.SUBSCRIBE, filters or '')
+        s.connect(self.pub_socket)
+        s.setsockopt(zmq.SUBSCRIBE, '')
+
         self.associated.append(s)
         return s
 
-    def shutdown(self, context=None):
+    def shutdown(self, hard=False, soft=True, context=None):
+
+        if not self.polling:
+            return
+
         self.polling = False
 
-        if not context:
-            context = zmq.Context.instance()
+        assert hard or soft, """ Must specify kill hard or soft """
 
-        #logging.info('Shutdown controller')
+        if hard:
+            self.state = CONTROL_STATES.SHUTDOWN
 
-        s = self.message_sender(context)
-        s.send(CONTROL_FRAME(
-            'controller',
-            CONTROL_PROTOCOL.SHUTDOWN,
-        ))
+            self.logging.info('[Controller] Hard Shutdown')
 
-        #for asoc in self.associated:
-            #asoc.close()
+            #for asoc in self.associated:
+                #asoc.close()
 
-    def destroy(self):
-        """
-        Manual cleanup.
-        """
-        self.shutdown()
+        if soft:
+            self.state = CONTROL_STATES.TERMINATE
 
-    #def __del__(self):
-        #self.shutdown()
+            self.logging.info('[Controller] Soft Shutdown')
+            self.send_softkill()
 
-    def qos(self):
-        if not self.debug:
-            return
-        return float(self.success) / (self.success + self.failed)
+            #for asoc in self.associated:
+                #asoc.close()
 
 if __name__ == '__main__':
+
     print 'Running on ',\
         'tcp://127.0.0.1:5000', \
         'tcp://127.0.0.1:5001',
@@ -204,4 +488,8 @@ if __name__ == '__main__':
         'tcp://127.0.0.1:5000',
         'tcp://127.0.0.1:5001',
     )
-    controller.run()
+    controller.manage(
+        'freeform',
+        []
+    )
+    controller.run('green')

zipline/protocol.py

@@ -134,39 +134,44 @@ from protocol_utils import Enum, FrameExceptionFactory, namedict
 
 INVALID_CONTROL_FRAME = FrameExceptionFactory('CONTROL')
 
-CONTROL_PROTOCOL = Enum(
-    'INIT'      , # 0 - req
-    'INFO'      , # 1 - req
-    'STATUS'    , # 2 - req
-    'SHUTDOWN'  , # 3 - req
-    'KILL'      , # 4 - req
-
-    'OK'        , # 5 - rep
-    'DONE'      , # 6 - rep
-    'EXCEPTION' , # 7 - rep
+CONTROL_STATES = Enum(
+    'RUNNING',
+    'SHUTDOWN',  # a soft kill
+    'TERMINATE', # a hard kill
 )
 
-def CONTROL_FRAME(id, status):
-    assert isinstance(id, basestring,)
-    assert isinstance(status, int)
+CONTROL_PROTOCOL = Enum(
+    'HEARTBEAT' , # 0 - req
+    'SHUTDOWN'  , # 1 - req
+    'KILL'      , # 2 - req
 
-    return msgpack.dumps(tuple([id, status]))
+    'OK'        , # 3 - rep
+    'DONE'      , # 4 - rep
+    'EXCEPTION' , # 5 - rep
+)
 
-def CONTORL_UNFRAME(msg):
+def CONTROL_FRAME(event, payload):
+    assert isinstance(event, int,)
+    assert isinstance(payload, basestring)
+
+    return msgpack.dumps(tuple([event, payload]))
+
+def CONTROL_UNFRAME(msg):
+    """
+    A status code and a message.
+    """
     assert isinstance(msg, basestring)
 
     try:
-        id, status = msgpack.loads(msg)
-        assert isinstance(id, basestring)
-        assert isinstance(status, int)
+        event, payload = msgpack.loads(msg)
+        assert isinstance(event, int)
+        assert isinstance(payload, basestring)
 
-        return id, status
+        return event, payload
     except TypeError:
         raise INVALID_CONTROL_FRAME(msg)
     except ValueError:
         raise INVALID_CONTROL_FRAME(msg)
-    #except AssertionError:
-        #raise INVALID_CONTROL_FRAME(msg)
 
 # -----------------------
 # Heartbeat Protocol

zipline/simulator.py

@@ -41,7 +41,9 @@ class Simulator(ComponentHost):
         return 'Simple Simulator'
 
     def launch_controller(self):
-        thread = threading.Thread(target=self.controller.run)
+        thread = threading.Thread(
+            target=self.controller.run,
+        )
         thread.start()
 
         self.subthreads.append(thread)
@@ -68,8 +70,8 @@ class Simulator(ComponentHost):
         if not self.running:
             return
 
-        #if self.controller:
-            #self.controller.shutdown()
+        if self.controller:
+            self.controller.shutdown()
 
         for component in self.components.itervalues():
             component.shutdown()

zipline/test/test_monitor.py

@@ -1,78 +0,0 @@
-"""
-Dummy simulator backported from Qexec for development on Zipline.
-"""
-
-import logging
-from multiprocessing import Process
-from unittest2 import TestCase
-
-from zipline.monitor import Controller
-
-import gevent
-from gevent_zeromq import zmq
-
-ctx = zmq.Context()
-
-#TODO: disabled by prefixing the test methods with a d
-class TestControlProtocol(TestCase):
-
-    def setUpController(self):
-        self.controller.run()
-
-    def setUp(self):
-        self.controller = Controller(
-            'tcp://127.0.0.1:5000',
-            'tcp://127.0.0.1:5001',
-        )
-        self.control_proc = Process(target=self.setUpController)
-        self.control_proc.start()
-
-    def tearDown(self):
-        self.control_proc.terminate()
-        ctx.destroy()
-
-    def asyncMessage(self, socket):
-        return socket.recv()
-
-    def send_and_receive(self, push, sub, message='\x01'):
-        msg = gevent.spawn(sub.recv)
-        push.send(message)
-        gevent.sleep(0) # explicit gevent yield
-        msg.join()
-        self.assertEqual(msg.value, message)
-
-    def dtest_control_message(self):
-
-        sub = self.controller.message_listener(context=ctx)
-        message = gevent.spawn(self.asyncMessage, sub)
-
-        push = self.controller.message_sender(context=ctx)
-
-        # Don't like introducing time constants but because of
-        # the way gevent scheduler works zmq will often send all
-        # the messages off before the other thread even gets to
-        # listening.
-        self.send_and_receive(push, sub)
-        sub.close()
-        push.close()
-
-    def dtest_control_delivery(self):
-        # Assert that the number of messages sent on the wire is
-        # the number of messages received, ie we don't drop any.
-        # This is of course depenendent on the topology of the
-        # listeners being fixed. Which normally it isn't.
-
-        sub = self.controller.message_listener(context=ctx)
-        message = gevent.spawn(self.asyncMessage, sub)
-
-        push = self.controller.message_sender(context=ctx)
-
-        # Don't like introducing time constants but because of
-        # the way gevent scheduler works zmq will often send all
-        # the messages off before the other thread even gets to
-        # listening.
-        for i in xrange(25):
-            self.send_and_receive(push, sub)
-
-        sub.close()
-        push.close()