Better tracking of failure modes.

docs/conf.py

@@ -25,7 +25,14 @@ sys.path.append(os.path.abspath('..'))
 
 # Add any Sphinx extension module names here, as strings. They can be extensions
 # coming with Sphinx (named 'sphinx.ext.*') or your custom ones.
-extensions = ['sphinx.ext.autodoc', 'sphinx.ext.intersphinx', 'sphinx.ext.todo', 'sphinx.ext.coverage', 'sphinx.ext.mathjax', 'sphinx.ext.viewcode']
+extensions = [
+    'sphinx.ext.autodoc',
+    'sphinx.ext.intersphinx',
+    'sphinx.ext.todo',
+    'sphinx.ext.coverage',
+    'sphinx.ext.mathjax',
+    'sphinx.ext.viewcode'
+]
 
 # Add any paths that contain templates here, relative to this directory.
 templates_path = ['_templates']

zipline/component.py

@@ -2,10 +2,12 @@
 Commonly used messaging components.
 
 Contains the base class for all components.
+
 """
 
 import os
 import uuid
+import time
 import socket
 import humanhash
 
@@ -15,7 +17,7 @@ from zipline.protocol import CONTROL_PROTOCOL, COMPONENT_STATE
 class Component(object):
     """
     Base class for components. Defines the the base messaging
-    interface between components.
+    interface for components.
 
     :param addresses: a dict of name_string -> zmq port address strings.
                       Must have the following entries
@@ -62,6 +64,10 @@ class Component(object):
         self.controller        = None
         self.heartbeat_timeout = 2000
         self.state_flag        = COMPONENT_STATE.OK # OK | DONE | EXCEPTION
+        self._exception        = None
+
+        self.start_tic         = None
+        self.stop_tic          = None
 
         # Humanhashes make this way easier to debug because they
         # stick in your mind unlike a 32 byte string of random hex.
@@ -82,37 +88,40 @@ class Component(object):
     # ------------
 
     def open(self):
-        raise NotImplementedError
-
-    def teardown_sockets(self):
         """
-        Close all zmq sockets safely.
-        """
-        #close all the sockets
-        for sock in self.sockets:
-            sock.close()
-
-    def destroy(self):
-        """
-        Clean shutdown.
-
-        Tear down after normal operation.
-        """
-        pass
-
-    def kill(self):
-        """
-        Unclean shutdown.
-
-        Tear down ( fast ) as a mode of failure in the
-        simulation or on service halt.
+        Open the connections needed to start doing work.
         """
         raise NotImplementedError
 
+    def ready(self):
+        """
+        Return ``True`` if and only if the component has finished execution.
+        """
+        return self.state_flag in [COMPONENT_STATE.DONE, \
+            COMPONENT_STATE.EXCEPTION]
+
+    def successful(self):
+        """
+        Return ``True`` if and only if the component has finished execution
+        successfully, that is, without raising an error.
+        """
+        return self.state_flag == COMPONENT_STATE.DONE and not \
+            self.exception
+
+    @property
+    def exception(self):
+        """
+        Holds the exception that the component failed on, or
+        ``None`` if the component has not failed.
+        """
+        return self._exception
+
     def do_work(self):
         raise NotImplementedError
 
     def _run(self):
+        self.start_tick = time.clock()
+
         self.done       = False # TODO: use state flag
         self.sockets    = []
 
@@ -132,9 +141,11 @@ class Component(object):
         self.setup_control()
         self.loop()
 
-        self.destroy()
+        self.shutdown()
         self.teardown_sockets()
 
+        self.end_tick = time.clock()
+
         # shouldn't block if we've done our job correctly
         # self.context.term()
 
@@ -159,6 +170,7 @@ class Component(object):
                 self.signal_exception(exc)
                 fail = exc
             finally:
+                # TODO: cleaner
                 if self.context:
                     self.context.destroy()
                 if fail:
@@ -186,12 +198,50 @@ class Component(object):
 
         self.receive_sync_ack() # blocking
 
+    def runtime(self):
+        if self.ready():
+            return self.stop_tic - self.start_tic
+
+    # ----------------------------
+    #  Cleanup & Modes of Failure
+    # ----------------------------
+
+    def teardown_sockets(self):
+        """
+        Close all zmq sockets safely. This is universal, no matter
+        where this is running it will need the sockets closed.
+        """
+        #close all the sockets
+        for sock in self.sockets:
+            sock.close()
+
+    def shutdown(self):
+        """
+        Clean shutdown.
+
+        Tear down after normal operation.
+        """
+        pass
+
+    def kill(self):
+        """
+        Unclean shutdown.
+
+        Tear down ( fast ) as a mode of failure in the
+        simulation or on service halt.
+
+        Context specific.
+        """
+        raise NotImplementedError
+
     # ----------------------
     #  Internal Maintenance
     # ----------------------
 
     def signal_exception(self, exc=None):
         self.state_flag = COMPONENT_STATE.EXCEPTION
+        self._exception = exc
+
         qutil.LOGGER.exception("Unexpected error in run for {id}.".format(id=self.get_id))
 
     def signal_done(self):
@@ -311,7 +361,7 @@ class Component(object):
 
     def setup_sync(self):
         """
-        Setup the sync socket and poller.
+        Setup the sync socket and poller. ( Connect )
         """
 
         qutil.LOGGER.debug("Connecting sync client for {id}".format(id=self.get_id))
@@ -335,8 +385,33 @@ class Component(object):
 
     @property
     def get_id(self):
+        """
+        The descriptive name of the component.
+        """
         return 'UNKNOWN COMPONENT'
 
+    @property
+    def get_type(self):
+        """
+        The data flow type of the component.
+
+        - ``SOURCE``
+        - ``CONDUIT``
+        - ``SINK``
+
+        """
+        raise NotImplementedError
+
+    @property
+    def get_pure(self):
+        """
+        Describes whehter this component purely functional,
+        i.e.  for a given set of inputs is it guaranteed to
+        always give the same output . Components that are
+        side-effectful are, generally, not pure.
+        """
+        return False
+
     def debug(self):
         """
         Debug information about the component.

zipline/messaging.py

@@ -1,12 +1,14 @@
 """
 Commonly used messaging components.
 """
+
 import datetime
 import ujson as json
 
 import zipline.util as qutil
 from zipline.component import Component
-from zipline.protocol import CONTROL_PROTOCOL
+from zipline.protocol import CONTROL_PROTOCOL, COMPONENT_TYPE, \
+    COMPONENT_STATE
 
 class ComponentHost(Component):
     """
@@ -16,48 +18,67 @@ class ComponentHost(Component):
 
     def __init__(self, addresses, gevent_needed=False):
         Component.__init__(self)
-        self.addresses = addresses
-        self.gevent_needed  = gevent_needed
+        self.addresses     = addresses
+        self.gevent_needed = gevent_needed
+        self.running       = False
 
         self.init()
 
     def init(self):
 
+        # Component Registry
+        # ----------------------
+        self.components     = {}
+        # ----------------------
+
         # workaround for defect in threaded use of strptime:
         # http://bugs.python.org/issue11108
         qutil.parse_date("2012/02/13-10:04:28.114")
 
-        self.components     = {}
         self.sync_register  = {}
         self.timeout        = datetime.timedelta(seconds=5)
 
         self.feed           = ParallelBuffer()
         self.merge          = MergedParallelBuffer()
         self.passthrough    = PassthroughTransform()
+        self.controller     = None
 
         #register the feed and the merge
         self.register_components([self.feed, self.merge, self.passthrough])
 
     def register_controller(self, controller):
+        """
+        Add the given components to the registry. Establish
+        communication with them.
+        """
+        if self.controller != None:
+            raise Exception("There can be only one!")
+
         self.controller = controller
 
+        # Propogate the controller to all the subcomponents
         for component in self.components.itervalues():
             component.controller = controller
 
     def register_components(self, components):
-        for component in components:
-            component.gevent_needed = self.gevent_needed
-            component.addresses = self.addresses
+        """
+        Add the given components to the registry. Establish
+        communication with them.
+        """
+        assert isinstance(components, list)
 
-            if self.controller:
-                component.controller = self.controller
+        for component in components:
+
+            component.gevent_needed = self.gevent_needed
+            component.addresses     = self.addresses
+            component.controller    = self.controller
 
             self.components[component.get_id] = component
             self.sync_register[component.get_id] = datetime.datetime.utcnow()
 
-            if(isinstance(component, DataSource)):
+            if isinstance(component, DataSource):
                 self.feed.add_source(component.get_id)
-            if(isinstance(component, BaseTransform)):
+            if isinstance(component, BaseTransform):
                 self.merge.add_source(component.get_id)
 
     def unregister_component(self, component_id):
@@ -66,6 +87,7 @@ class ComponentHost(Component):
 
     def setup_sync(self):
         """
+        Setup the sync socket and poller. ( Bind )
         """
         qutil.LOGGER.debug("Connecting sync server.")
 
@@ -112,17 +134,17 @@ class ComponentHost(Component):
 
             if self.sync_socket in socks and socks[self.sync_socket] == self.zmq.POLLIN:
                 msg = self.sync_socket.recv()
-                parts = msg.split(':')
-                sync_id, status = parts
 
-                # TODO: move into frame protocol
-                if len(parts) != 2:
-                    qutil.LOGGER.info("got bad confirm: {msg}".format(msg=msg))
-                    continue
+                try:
+                    parts = msg.split(':')
+                    sync_id, status = parts
+                except ValueError as exc:
+                    self.signal_exception(exc)
 
                 if status == str(CONTROL_PROTOCOL.DONE): # TODO: other way around
                     qutil.LOGGER.info("{id} is DONE".format(id=sync_id))
                     self.unregister_component(sync_id)
+                    self.state_flag = COMPONENT_STATE.DONE
                 else:
                     self.sync_register[sync_id] = datetime.datetime.utcnow()
 
@@ -143,6 +165,7 @@ class ComponentHost(Component):
     def teardown_component(self, component):
         raise NotImplementedError
 
+
 class ParallelBuffer(Component):
     """
     Connects to N PULL sockets, publishing all messages received to a PUB
@@ -153,12 +176,16 @@ class ParallelBuffer(Component):
 
     def __init__(self):
         Component.__init__(self)
+
         self.sent_count             = 0
         self.received_count         = 0
         self.draining               = False
-        self.data_buffer            = {}
         self.ds_finished_counter    = 0
 
+        # Depending on the size of this, might want to use a data
+        # structure with better asymptotics.
+        self.data_buffer            = {}
+
     def init(self):
         pass
 
@@ -166,8 +193,13 @@ class ParallelBuffer(Component):
     def get_id(self):
         return "FEED"
 
-    def add_source(self, source_id):
-        self.data_buffer[source_id] = []
+    @property
+    def get_type(self):
+        return COMPONENT_TYPE.CONDUIT
+
+    # -------------
+    # Core Methods
+    # -------------
 
     def open(self):
         self.pull_socket = self.bind_data()
@@ -191,9 +223,45 @@ class ParallelBuffer(Component):
                     self.drain()
                     self.signal_done()
             else:
-                event = json.loads(message)
-                self.append(event[u'id'], event)
-                self.send_next()
+                try:
+                    event = json.loads(message)
+
+                # JSON deserialization error
+                except ValueError as exc:
+                    return self.signal_exception(exc)
+
+                try:
+                    self.append(event[u'id'], event)
+                    self.send_next()
+
+                # Invalid message
+                except KeyError as exc:
+                    return self.signal_exception(exc)
+
+    # -------------
+    # Flow Control
+    # -------------
+
+    def drain(self):
+        """
+        Send all messages in the buffer.
+        """
+        self.draining = True
+        while self.pending_messages() > 0:
+            self.send_next()
+
+    def send_next(self):
+        """
+        Send the (chronologically) next message in the buffer.
+        """
+        if(not(self.is_full() or self.draining)):
+            return
+
+        event = self.next()
+
+        if event != None:
+            self.feed_socket.send(json.dumps(event), self.zmq.NOBLOCK)
+            self.sent_count += 1
 
     def append(self, source_id, value):
         """
@@ -242,25 +310,11 @@ class ParallelBuffer(Component):
             total += len(events)
         return total
 
-    def drain(self):
+    def add_source(self, source_id):
         """
-        Send all messages in the buffer
+        Add a data source to the buffer.
         """
-        self.draining = True
-        while(self.pending_messages() > 0):
-            self.send_next()
-
-    def send_next(self):
-        """
-        Send the (chronologically) next message in the buffer.
-        """
-        if(not(self.is_full() or self.draining)):
-            return
-
-        event = self.next()
-        if(event != None):
-            self.feed_socket.send(json.dumps(event), self.zmq.NOBLOCK)
-            self.sent_count += 1
+        self.data_buffer[source_id] = []
 
     def __len__(self):
         """
@@ -287,6 +341,10 @@ class MergedParallelBuffer(ParallelBuffer):
     def get_id(self):
         return "MERGE"
 
+    @property
+    def get_type(self):
+        return COMPONENT_TYPE.CONDUIT
+
     def open(self):
         self.pull_socket = self.bind_merge()
         self.feed_socket = self.bind_result()
@@ -309,13 +367,13 @@ class MergedParallelBuffer(ParallelBuffer):
 
 class BaseTransform(Component):
     """
-    Top level execution entry point for the transform::
+    Top level execution entry point for the transform
 
-        - connects to the feed socket to subscribe to events
-        - connets to the result socket (most oftened bound by a TransformsMerge) to PUSH transforms
-        - processes all messages received from feed, until DONE message received
-        - pushes all transforms
-        - sends DONE to result socket, closes all sockets and context
+    - connects to the feed socket to subscribe to events
+    - connets to the result socket (most oftened bound by a TransformsMerge) to PUSH transforms
+    - processes all messages received from feed, until DONE message received
+    - pushes all transforms
+    - sends DONE to result socket, closes all sockets and context
 
     Parent class for feed transforms. Subclass and override transform
     method to create a new derived value from the combined feed.
@@ -323,8 +381,10 @@ class BaseTransform(Component):
 
     def __init__(self, name):
         Component.__init__(self)
-        self.state = {}
-        self.state['name'] = name
+
+        self.state = {
+            'name': name
+        }
 
         self.init()
 
@@ -335,6 +395,10 @@ class BaseTransform(Component):
     def get_id(self):
         return self.state['name']
 
+    @property
+    def get_type(self):
+        return COMPONENT_TYPE.CONDUIT
+
     def open(self):
         """
         Establishes zmq connections.
@@ -347,9 +411,11 @@ class BaseTransform(Component):
     def do_work(self):
         """
         Loops until feed's DONE message is received:
-            - receive an event from the data feed
-            - call transform (subclass' method) on event
-            - send the transformed event
+
+        - receive an event from the data feed
+        - call transform (subclass' method) on event
+        - send the transformed event
+
         """
         socks = dict(self.poll.poll(self.heartbeat_timeout))
 
@@ -362,12 +428,28 @@ class BaseTransform(Component):
                 self.signal_done()
                 return
 
-            event = json.loads(message)
-            cur_state = self.transform(event)
-            cur_state['dt'] = event['dt']
-            cur_state['id'] = self.state['name']
+            try:
+                event = json.loads(message)
+            except ValueError as exc:
+                return self.signal_exception(exc)
 
-            self.result_socket.send(json.dumps(cur_state), self.zmq.NOBLOCK)
+            try:
+                cur_state = self.transform(event)
+                cur_state['dt'] = event['dt']
+                cur_state['id'] = self.state['name']
+
+            # This is overloaded, so it can fail in all sorts of
+            # unknown ways. Its best to catch it in the
+            # Transformer itself.
+            except Exception as exc:
+                return self.signal_exception(exc)
+
+            try:
+                json_frame = json.dumps(cur_state)
+            except ValueError as exc:
+                return self.signal_exception(exc)
+
+            self.result_socket.send(json_frame, self.zmq.NOBLOCK)
 
     def transform(self, event):
         """
@@ -386,15 +468,30 @@ class BaseTransform(Component):
 
 
 class PassthroughTransform(BaseTransform):
+    """
+    A bypass transform which is also an identity transform::
+
+            +-------+
+        +---|   f   |--->
+            +-------+
+        +------id------->
+
+    """
 
     def __init__(self):
         BaseTransform.__init__(self, "PASSTHROUGH")
 
+        self.init()
+
     def init(self):
         pass
 
+    @property
+    def get_type(self):
+        return COMPONENT_TYPE.CONDUIT
+
     def transform(self, event):
-        return {'value':event}
+        return { 'value': event }
 
 
 class DataSource(Component):
@@ -405,7 +502,8 @@ class DataSource(Component):
     """
     def __init__(self, source_id):
         Component.__init__(self)
-        self.id        = source_id
+
+        self.id = source_id
         self.init()
 
     def init(self):
@@ -415,19 +513,25 @@ class DataSource(Component):
     def get_id(self):
         return self.id
 
+    @property
+    def get_type(self):
+        return COMPONENT_TYPE.SOURCE
+
     def open(self):
-        #create the data sink. Based on http://zguide.zeromq.org/py:tasksink2
         self.data_socket = self.connect_data()
 
     def send(self, event):
         """
-        event is expected to be a dict
-        sets id and type properties in the dict
-        sends to the data_socket.
+        Emit data.
         """
+        assert isinstance(event, dict)
+
         event['id'] = self.id
         event['type'] = self.get_type()
-        self.data_socket.send(json.dumps(event))
 
-    def get_type(self):
-        raise NotImplementedError
+        try:
+            json_frame = json.dumps(event)
+        except ValueError as exc:
+            return self.signal_exception(exc)
+
+        self.data_socket.send(json_frame)

zipline/monitor.py

@@ -1,4 +1,5 @@
 import zmq
+from zipline.protocol import CONTROL_PROTOCOL, CONTROL_FRAME
 
 class Controller(object):
     """
@@ -56,6 +57,7 @@ class Controller(object):
         self.push_socket = pull_socket # same port
         self.pub_socket = pub_socket
         self.sub_socket = pub_socket # same port
+        self.terminated = False
 
         if logging:
             self.logging = logging
@@ -128,6 +130,8 @@ class Controller(object):
                 self.failed += 1
                 continue
 
+        self.terminated = True
+
     # -------------------
     # Hooks for Endpoints
     # -------------------
@@ -161,20 +165,29 @@ class Controller(object):
         self.associated.append(s)
         return s
 
+    def shutdown(self, context=None):
+        self.polling = False
+
+        if not context:
+            context = zmq.Context()
+
+        s = self.message_sender(context)
+        s.send(CONTROL_FRAME(
+            'controller',
+            CONTROL_PROTOCOL.SHUTDOWN,
+        ))
+
+        #for asoc in self.associated:
+            #asoc.close()
+
     def destroy(self):
         """
         Manual cleanup.
         """
-        self.polling = False
-
-        for asoc in self.associated:
-            asoc.close()
-
-        #if self._ctx:
-            #self._ctx.destroy()
+        self.shutdown()
 
     def __del__(self):
-        self.destroy()
+        self.shutdown()
 
     def qos(self):
         if not self.debug:

zipline/protocol.py

@@ -103,18 +103,18 @@ CONTROL_PROTOCOL = Enum(
 )
 
 def CONTROL_FRAME(id, status):
-    assert isinstance(basestring, id)
-    assert isinstance(int, status)
+    assert isinstance(id, basestring,)
+    assert isinstance(status, int)
 
     return msgpack.dumps(tuple([id, status]))
 
 def CONTORL_UNFRAME(msg):
-    assert isinstance(basestring, msg)
+    assert isinstance(msg, basestring)
 
     try:
         id, status = msgpack.loads(msg)
-        assert isinstance(basestring, id)
-        assert isinstance(int, status)
+        assert isinstance(id, basestring)
+        assert isinstance(status, int)
 
         return id, status
     except TypeError:
@@ -140,6 +140,12 @@ HEARTBEAT_PROTOCOL = namedict({
 # Component State
 # ==================
 
+COMPONENT_TYPE = Enum(
+    'SOURCE'  , # 0
+    'CONDUIT' , # 1
+    'SINK'    , # 2
+)
+
 COMPONENT_STATE = Enum(
     'OK'        , # 0
     'DONE'      , # 1

zipline/test/client.py

@@ -2,7 +2,7 @@ import ujson as json
 
 import zipline.util as qutil
 import zipline.messaging as qmsg
-from zipline.protocol import CONTROL_PROTOCOL
+from zipline.protocol import CONTROL_PROTOCOL, COMPONENT_TYPE
 
 class TestClient(qmsg.Component):
 
@@ -15,13 +15,17 @@ class TestClient(qmsg.Component):
         self.init()
 
     def init(self):
-        self.received_count     = 0
-        self.prev_dt            = None
+        self.received_count = 0
+        self.prev_dt        = None
 
     @property
     def get_id(self):
         return "TEST_CLIENT"
 
+    @property
+    def get_type(self):
+        return COMPONENT_TYPE.SINK
+
     def open(self):
         self.data_feed = self.connect_result()
 
@@ -31,25 +35,37 @@ class TestClient(qmsg.Component):
         if self.control_in in socks and socks[self.control_in] == self.zmq.POLLIN:
             msg = self.control_in.recv()
 
-        if self.data_feed in socks and socks[self.data_feed] == self.zmq.POLLIN:   
+        if self.data_feed in socks and socks[self.data_feed] == self.zmq.POLLIN:
             msg = self.data_feed.recv()
 
             if msg == str(CONTROL_PROTOCOL.DONE):
                 qutil.LOGGER.info("Client is DONE!")
                 self.signal_done()
-                self.utest.assertEqual(self.expected_msg_count, self.received_count, 
-                                "The client should have received ({n}) the same number of messages as the feed sent ({m})."
-                                    .format(n=self.received_count, m=self.expected_msg_count))
+                self.utest.assertEqual(
+                    self.expected_msg_count, self.received_count,
+                    "The client should have received ({n}) the same number of \
+                    messages as the feed sent ({m})."
+                    .format(n=self.received_count, m=self.expected_msg_count))
                 return
 
             self.received_count += 1
-            event = json.loads(msg)
 
-            if(self.prev_dt != None):
+            try:
+                event = json.loads(msg)
+
+            # JSON deserialization error
+            except ValueError as exc:
+                return self.signal_exception(exc)
+
+            if self.prev_dt != None:
                 if not event['dt'] >= self.prev_dt:
-                    raise Exception("Message out of order: {date} after {prev}".format(date=event['dt'], prev=prev_dt))
+                    raise Exception(
+                        "Message out of order: {date} after {prev}".format(
+                            date = event['dt'], prev = self.prev_dt
+                        )
+                    )
+            else:
+                self.prev_dt = event['dt']
 
-            self.prev_dt = event['dt']
-
-            if(self.received_count % 100 == 0):
+            if self.received_count % 100 == 0:
                 qutil.LOGGER.info("received {n} messages".format(n=self.received_count))

zipline/test/test_messaging.py

@@ -4,12 +4,13 @@ Test suite for the messaging infrastructure of QSim.
 #don't worry about excessive public methods pylint: disable=R0904
 
 from collections import defaultdict
-import zipline.messaging as qmsg
 
 from zipline.transforms.technical import MovingAverage
 from zipline.sources import RandomEquityTrades
 
 from zipline.test.client import TestClient
+from zipline.test.transform import DivideByZeroTransform
+
 
 # Should not inherit form TestCase since test runners will pick
 # it up as a test. Its a Mixin of sorts at this point.
@@ -111,15 +112,61 @@ class SimulatorTestCase(object):
         # Stop Running
         # ------------
 
-        # TODO: less abrupt later, just shove a StopIteration
-        # down the pipe to make it stop spinning
-        sim.cuc._Thread__stop()
+        self.assertTrue(sim.ready())
+        self.assertFalse(sim.exception)
 
         self.assertEqual(sim.feed.pending_messages(), 0,
             "The feed should be drained of all messages, found {n} remaining."
             .format(n=sim.feed.pending_messages())
         )
 
+    def test_simplefail(self):
+
+        # Simple test just to make sure that the archiecture is
+        # responding.
+
+        # Base Simuation
+        # --------------
+
+        # Allocate sockets for the simulator components
+        sockets = self.allocate_sockets(5)
+
+        addresses = {
+            'sync_address'   : sockets[0],
+            'data_address'   : sockets[1],
+            'feed_address'   : sockets[2],
+            'merge_address'  : sockets[3],
+            'result_address' : sockets[4]
+        }
+
+        sim = self.get_simulator(addresses)
+        con = self.get_controller()
+
+        # Simulation Components
+        # ---------------------
+
+        ret1 = RandomEquityTrades(133, "ret1", 1)
+        ret2 = RandomEquityTrades(134, "ret2", 1)
+        fail_transform = DivideByZeroTransform("fail")
+        client = TestClient(self, expected_msg_count=ret1.count + ret2.count)
+
+        sim.register_controller( con )
+        sim.register_components([ret1, ret2, fail_transform, client])
+
+        # Simulation
+        # ----------
+        sim.simulate()
+
+        # Stop Running
+        # ------------
+
+        self.assertTrue(fail_transform.exception)
+        self.assertFalse(fail_transform.successful())
+
+        self.assertEqual(sim.feed.pending_messages(), 0,
+            "The feed should be drained of all messages, found {n} remaining."
+            .format(n=sim.feed.pending_messages())
+        )
 
     def test_sources_only(self):
 
@@ -156,10 +203,8 @@ class SimulatorTestCase(object):
 
         # Stop Running
         # ------------
-
-        # TODO: less abrupt later, just shove a StopIteration
-        # down the pipe to make it stop spinning
-        sim.cuc._Thread__stop()
+        self.assertTrue(sim.ready())
+        self.assertFalse(sim.exception)
 
         self.assertEqual(sim.feed.pending_messages(), 0,
             "The feed should be drained of all messages, found {n} remaining."
@@ -203,30 +248,10 @@ class SimulatorTestCase(object):
 
         # Stop Running
         # ------------
-
-        # TODO: less abrupt later, just shove a StopIteration
-        # down the pipe to make it stop spinning
-        sim.cuc._Thread__stop()
+        self.assertTrue(sim.ready())
+        self.assertFalse(sim.exception)
 
         self.assertEqual(sim.feed.pending_messages(), 0,
             "The feed should be drained of all messages, found {n} remaining."
             .format(n=sim.feed.pending_messages())
         )
-
-    # TODO used?
-    def dtest_error_in_feed(self):
-
-        ret1 = RandomEquityTrades(133, "ret1", 400)
-        ret2 = RandomEquityTrades(134, "ret2", 400)
-        sources = {"ret1":ret1, "ret2":ret2}
-        mavg1 = MovingAverage("mavg1", 30)
-        mavg2 = MovingAverage("mavg2", 60)
-        transforms = {"mavg1":mavg1, "mavg2":mavg2}
-        client = TestClient(self, expected_msg_count=0)
-        sim = self.get_simulator(sources, transforms, client)
-
-        # TODO: way too long
-        sim.feed = DataFeedErr(sources.keys(), sim.data_address, sim.feed_address, sim.performance_address, qmsg.Sync(sim, "DataFeedErrorGenerator"))
-        sim.simulate()
-
-

zipline/test/transform.py

@@ -0,0 +1,22 @@
+from zipline.messaging import BaseTransform
+from zipline.protocol import COMPONENT_TYPE
+
+class DivideByZeroTransform(BaseTransform):
+    """
+    A transform that fails.
+    """
+
+    def __init__(self, name):
+        BaseTransform.__init__(self, "PASSTHROUGH")
+        self.state['name'] = name
+        self.init()
+
+    def init(self):
+        pass
+
+    @property
+    def get_type(self):
+        return COMPONENT_TYPE.CONDUIT
+
+    def transform(self, event):
+        return { 'value': 0/0 }

zipline/topology.py

@@ -0,0 +1,80 @@
+"""
+Contains the various deployable topologies of ziplines.
+
+This is mostly hardcoded at the moment but as the topologies
+becomes more sophisiticated this logic will be the primary
+router of sockets.
+
+Ontology of Stream Processing
+=============================
+
+Source
+******
+
+A producer of data. The data could be in a datastore, coming from a
+socket, etc. To access this data, we pull from the source. Sources increase the
+total amount of data flowing through the system.  Sources are generally not
+pure since they involve IO.
+
+Sink
+****
+
+A consumer of data. Basic examples would be a sum function (adding up a
+stream of numbers fed in), a datastore sink, a socket etc. We push data
+into a sink. When / If a sink completes processing, it may return some
+value that exists outside of the system.
+
+Sinks decrease the total amount of information flowing through the system.
+
+Conduit
+*******
+
+A transformer of data. We push data into a conduit.  Similar to a sink,
+but instead of returning a single value at the end, a conduit can
+return multiple outputs every time it is pushed to. The returned values
+remain in the system.
+
+Conduits may or may not be pure, it is usefull to distinguish between the
+two since pure conduits have a variety of nice properties under composition
+
+"""
+
+from zipline.protocol import  COMPONENT_TYPE
+
+class Topology(object):
+    pass
+
+class DiamondTopology(Topology):
+    """
+    Exposes a feed, merge, and passthrough bypass::
+
+                                 +--------+
+                     +---------->|        |---------------+
+                     |           +--------+               |
+                     |                                    v
+    +---+----+   +---+----+      +--------+           +--------+    +---+----+
+    |        +-->|        +----->|        |---------->|        |--->|        |
+    +---+----+   +---+----+      +--------+           +--------+    +---+----+
+                     |                                    ^
+                     |           +--------+               |
+                     +---------->|        |---------------+
+                     |           +--------+               |
+                     |                                    |
+                     +------------passthru----------------+
+
+    """
+
+    flow = {
+        'flow'        : COMPONENT_TYPE.SOURCE  ,
+        'serializers' : COMPONENT_TYPE.CONDUIT ,
+        'transforms'  : COMPONENT_TYPE.CONDUIT ,
+        'merges'      : COMPONENT_TYPE.CONDUIT ,
+        'clients'     : COMPONENT_TYPE.SINK    ,
+    }
+
+    def __init__(self):
+        self.sources     = []
+        self.serializers = []
+        self.transforms  = []
+        self.merges      = []
+        self.clients     = []