reworked directory

zipline/__init__.py

@@ -1,3 +1,16 @@
 """
-QSim provides asynchronous simulation of historic data streams, simulated trade execution, and data stream transformations.
-"""
+Zipline
+"""
+
+# This is *not* a place to dump arbitrary classes/modules for convenience,
+# it is a place to expose the public interfaces.
+
+import protocol
+from core.monitor import Controller
+from lines import SimulatedTrading
+
+__all__ = [
+    SimulatedTrading,
+    Controller,
+    protocol,
+]

zipline/core/component.py

@@ -0,0 +1,562 @@
+"""
+Commonly used messaging components.
+
+Contains the base class for all components.
+"""
+
+import os
+import sys
+import uuid
+import time
+import socket
+import gevent
+import traceback
+import humanhash
+
+# pyzmq
+import zmq
+# gevent_zeromq
+import gevent_zeromq
+# zmq_ctypes
+#import zmq_ctypes
+
+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, CONTROL_FRAME
+
+
+class Component(object):
+    """
+    Base class for components. Defines the the base messaging
+    interface for components.
+
+    :param addresses: a dict of name_string -> zmq port address strings.
+                      Must have the following entries
+
+    :param sync_address: socket address used for synchronizing the start of
+                         all workers, heartbeating, and exit notification
+                         will be used in REP/REQ sockets. Bind is always on
+                         the REP side.
+
+    :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 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
+                         will be used in PUB/SUB sockets between Feed and
+                         Transforms. Bind is always on the PUB side.
+
+    :param merge_address: socket address used to publish transformed
+                          values.  will be used in PUSH/PULL from many
+                          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
+                           used in PUB/SUB from one Merge to one or many
+                           clients. Bind is always on the PUB side.
+
+    bind/connect methods will return the correct socket type for each
+    address.
+
+    """
+
+    def __init__(self):
+        self.zmq               = None
+        self.context           = None
+        self.addresses         = None
+
+        self.out_socket        = None
+        self.killed            = False
+        self.controller        = None
+        # timeout after a full minute
+        self.heartbeat_timeout = 60 *1000
+        self.state_flag        = COMPONENT_STATE.OK
+        self.error_state       = COMPONENT_FAILURE.NOFAILURE
+        self.on_done           = None
+
+        self._exception        = None
+        self.fail_time         = None
+        self.start_tic         = None
+        self.stop_tic          = None
+        self.note              = None
+        self.confirmed         = False
+
+        # Humanhashes make this way easier to debug because they
+        # stick in your mind unlike a 32 byte string of random hex.
+        self.guid = uuid.uuid4()
+        self.huid = humanhash.humanize(self.guid.hex)
+
+        self.init()
+
+    def init(self):
+        """
+        Subclasses should override this to extend the setup for
+        the class. Shouldn't have side effects.
+        """
+        pass
+
+    # ------------
+    # Core Methods
+    # ------------
+
+    def open(self):
+        """
+        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 init_zmq(self, flavor):
+        """
+        ZMQ in all flavors. Have it your way.
+
+            mp     - Distinct contexts | pyzmq
+            thread - Same context      | pyzmq
+            green  - Same context      | gevent_zeromq
+            pypy   - Same context      | zmq_ctypes
+
+        """
+
+        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
+
+        raise Exception("Unknown ZeroMQ Flavor")
+
+    def _run(self):
+        self.start_tic = time.time()
+
+        self.done       = False # TODO: use state flag
+        self.sockets    = []
+
+        self.init_zmq(self.zmq_flavor)
+
+        self.setup_poller()
+
+        self.open()
+        self.setup_sync()
+        self.setup_control()
+
+        self.loop()
+        self.shutdown()
+
+        self.stop_tic = time.time()
+
+    def run(self, catch_exceptions=True):
+        """
+        Run the component.
+
+        Optionally takes an argument to catch and log all exceptions raised
+        during execution ues this with care since it makes it very hard to
+        debug since it mucks up your stacktraces.
+        """
+
+        if catch_exceptions:
+            try:
+                self._run()
+            except Exception as exc:
+                exc_info = sys.exc_info()
+                self.signal_exception(exc)
+
+                # Reraise the exception
+                raise exc_info[0], exc_info[1], exc_info[2]
+            finally:
+
+                self.shutdown()
+                self.teardown_sockets()
+
+    def working(self):
+        """
+        Controls when the work loop will start and end
+
+        If we encounter an exception or signal done exit.
+
+        Overload for higher order behavior.
+        """
+        return (not self.done)
+
+    def loop(self, lockstep=True):
+        """
+        Loop to do work while we still have work to do.
+        """
+        while self.working():
+            self.confirm()
+            self.do_work()
+
+    def confirm(self):
+        """
+        Send a synchronization request to the host.
+        """
+        if not self.confirmed:
+            # TODO: proper framing
+            self.sync_socket.send(self.get_id + ":RUN")
+
+            self.receive_sync_ack() # blocking
+            self.confirmed = True
+            
+    def runtime(self):
+        if self.ready() and self.start_tic and self.stop_tic:
+            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.
+        """
+        if self.on_done:
+            self.on_done()
+
+    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, 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
+        else:
+            self.error_state = COMPONENT_FAILURE.HOSTEXCEPT
+
+        self.state_flag = COMPONENT_STATE.EXCEPTION
+        # mark the time of failure so we can track the failure
+        # progogation through the system.
+
+        self.stop_tic = time.time()
+
+        self._exception = exc
+        exc_type, exc_value, exc_traceback = sys.exc_info()
+        trace = '\n>>>'.join(traceback.format_exception(exc_type, exc_value, exc_traceback))
+
+        exception_frame = CONTROL_FRAME(
+            CONTROL_PROTOCOL.EXCEPTION,
+            trace
+        )
+        self.control_out.send(exception_frame)
+
+        qutil.LOGGER.exception("Unexpected error in run for {id}.".format(id=self.get_id))
+
+    def signal_done(self):
+        """
+        Notify down stream components that we're done.
+        """
+
+        self.state_flag = COMPONENT_STATE.DONE
+
+        if self.out_socket:
+            self.out_socket.send(str(CONTROL_PROTOCOL.DONE))
+
+        #notify host we're done
+        # 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
+    # -----------
+
+    def setup_poller(self):
+        """
+        Setup the poller used for multiplexing the incoming data
+        handling sockets.
+        """
+
+        # Initializes the poller class specified by the flavor of
+        # ZeroMQ. Either zmq.Poller or gpoll.Poller .
+        self.poll = self.zmq_poller()
+
+    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))
+
+    def bind_data(self):
+        return self.bind_pull_socket(self.addresses['data_address'])
+
+    def connect_data(self):
+        return self.connect_push_socket(self.addresses['data_address'])
+
+    def bind_feed(self):
+        return self.bind_pub_socket(self.addresses['feed_address'])
+
+    def connect_feed(self):
+        return self.connect_sub_socket(self.addresses['feed_address'])
+
+    def bind_merge(self):
+        return self.bind_pull_socket(self.addresses['merge_address'])
+
+    def connect_merge(self):
+        return self.connect_push_socket(self.addresses['merge_address'])
+
+    def bind_result(self):
+        return self.bind_pub_socket(self.addresses['result_address'])
+
+    def connect_result(self):
+        return self.connect_sub_socket(self.addresses['result_address'])
+
+    def bind_pull_socket(self, addr):
+        pull_socket = self.context.socket(self.zmq.PULL)
+        pull_socket.bind(addr)
+        self.poll.register(pull_socket, self.zmq.POLLIN)
+
+        self.sockets.append(pull_socket)
+
+        return pull_socket
+
+    def connect_push_socket(self, addr):
+        push_socket = self.context.socket(self.zmq.PUSH)
+        push_socket.connect(addr)
+        #push_socket.setsockopt(self.zmq.LINGER,0)
+        self.sockets.append(push_socket)
+        self.out_socket = push_socket
+
+        return push_socket
+
+    def bind_pub_socket(self, addr):
+        pub_socket = self.context.socket(self.zmq.PUB)
+        pub_socket.bind(addr)
+        #pub_socket.setsockopt(self.zmq.LINGER,0)
+        self.out_socket = pub_socket
+
+        return pub_socket
+
+    def connect_sub_socket(self, addr):
+        sub_socket = self.context.socket(self.zmq.SUB)
+        sub_socket.connect(addr)
+        sub_socket.setsockopt(self.zmq.SUBSCRIBE,'')
+        self.sockets.append(sub_socket)
+
+        self.poll.register(sub_socket, self.zmq.POLLIN)
+
+        return sub_socket
+
+    def setup_control(self):
+        """
+        Set up the control socket. Used to monitor the
+        overall status of the simulation and to forcefully tear
+        down the simulation in case of a failure.
+        """
+
+        # Allow for the possibility of not having a controller,
+        # possibly the zipline devsimulator may not want this.
+        if not self.controller:
+            return
+
+        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])
+
+    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))
+
+        self.sync_socket = self.context.socket(self.zmq.REQ)
+        self.sync_socket.connect(self.addresses['sync_address'])
+        #self.sync_socket.setsockopt(self.zmq.LINGER,0)
+
+        self.sync_poller = self.zmq_poller()
+        self.sync_poller.register(self.sync_socket, self.zmq.POLLIN)
+
+        self.sockets.append(self.sync_socket)
+
+    # ---------------------
+    # Description and Debug
+    # ---------------------
+
+    def extern_logger(self):
+        """
+        Pipe logs out to a provided logging interface.
+        """
+        pass
+
+    def setup_extern_logger(self):
+        """
+        Pipe logs out to a provided logging interface.
+        """
+        pass
+
+    @property
+    def get_id(self):
+        """
+        The descriptive name of the component.
+        """
+        # Prevents the bug that Thomas ran into
+        raise NotImplementedError
+
+    @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 note(self):
+        """
+        Information about the component. Mostly used for testing.
+        """
+
+    def get_note(self):
+        return self.note or ''
+
+    def debug(self):
+        """
+        Debug information about the component.
+        """
+        return {
+            'id'         : self.get_id          ,
+            'huid'       : self.huid            ,
+            'host'       : socket.gethostname() ,
+            'pid'        : os.getpid()          ,
+            'memaddress' : hex(id(self))        ,
+            'ready'      : self.successful()    ,
+            'succesfull' : self.ready()         ,
+        }
+
+    def __len__(self):
+        """
+        Some components overload this for debug purposes
+        """
+        raise NotImplementedError
+
+    def __repr__(self):
+        """
+        Return a usefull string representation of the component
+        to indicate its type, unique identifier, and computational
+        context identifier name.
+        """
+
+        return "<{name} {uuid} at {host} {pid} {pointer}>".format(
+            name    = self.get_id          ,
+            uuid    = self.huid            ,
+            host    = socket.gethostname() ,
+            pid     = os.getpid()          ,
+            pointer = hex(id(self))        ,
+        )

zipline/core/messaging.py

@@ -0,0 +1,636 @@
+"""
+Commonly used messaging components.
+"""
+
+import datetime
+
+from collections import Counter
+
+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, CONTROL_FRAME, CONTROL_UNFRAME
+
+class ComponentHost(Component):
+    """
+    Components that can launch multiple sub-components, synchronize their
+    start, and then wait for all components to be finished.
+    """
+
+    def __init__(self, addresses):
+        Component.__init__(self)
+        self.addresses     = addresses
+        self.running       = False
+
+        self.init()
+
+    def init(self):
+        assert hasattr(self, 'zmq_flavor'), \
+        """ You must specify a flavor of ZeroMQ for all
+        ComponentHost subclasses. """
+
+        # Component Registry, keyed by get_id
+        # ----------------------
+        self.components     = {}
+        # ----------------------
+        # Internal Registry, keyed by guid
+        self._components     = {}
+        # ----------------------
+
+        self.sync_register  = {}
+        self.timeout        = datetime.timedelta(seconds=60)
+
+        self.feed           = Feed()
+        self.merge          = Merge()
+        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
+        self.controller.zmq_flavor = self.zmq_flavor
+
+        # Propogate the controller to all the subcomponents
+        for component in self.components.itervalues():
+            component.controller = controller
+
+    def register_components(self, components):
+        """
+        Add the given components to the registry. Establish
+        communication with them.
+        """
+        assert isinstance(components, list)
+        for component in components:
+
+            component.addresses     = self.addresses
+            component.controller    = self.controller
+
+            # Hosts share their zmq flavor with hosted components
+            component.zmq_flavor    = self.zmq_flavor
+
+            self._components[component.guid] = component
+            self.components[component.get_id] = component
+            self.sync_register[component.get_id] = datetime.datetime.utcnow()
+
+            if isinstance(component, DataSource):
+                self.feed.add_source(component.get_id)
+            if isinstance(component, BaseTransform):
+                self.merge.add_source(component.get_id)
+                        
+    def unregister_component(self, component_id):
+        del self.components[component_id]
+        del self.sync_register[component_id]
+
+    def setup_sync(self):
+        """
+        Setup the sync socket and poller. ( Bind )
+        """
+        qutil.LOGGER.debug("Connecting sync server.")
+
+        self.sync_socket = self.context.socket(self.zmq.REP)
+        self.sync_socket.bind(self.addresses['sync_address'])
+
+        self.sync_poller = self.zmq_poller()
+        self.sync_poller.register(self.sync_socket, self.zmq.POLLIN)
+
+        self.sockets.append(self.sync_socket)
+
+    def open(self):
+        for component in self.components.values():
+            self.launch_component(component)
+        self.launch_controller()
+
+    def is_running(self):
+        """
+        DEPRECATED, left in for compatability for now.
+        """
+
+        cur_time = datetime.datetime.utcnow()
+
+        if len(self.components) == 0:
+            qutil.LOGGER.info("Component register is empty.")
+            return False
+
+        return True
+
+    def loop(self, lockstep=True):
+
+        while self.is_running():
+            # wait for synchronization request at start, and DONE at end.
+            # don't timeout.
+            socks = dict(self.sync_poller.poll()) 
+
+            if self.sync_socket in socks and socks[self.sync_socket] == self.zmq.POLLIN:
+                msg = self.sync_socket.recv()
+
+                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.debug("{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()
+
+                #qutil.LOGGER.info("confirmed {id}".format(id=msg))
+                # send synchronization reply
+                self.sync_socket.send('ack', self.zmq.NOBLOCK)
+
+    # ------------------
+    # Simulation Control
+    # ------------------
+
+    def launch_controller(self, controller):
+        raise NotImplementedError
+
+    def launch_component(self, component):
+        raise NotImplementedError
+
+    def teardown_component(self, component):
+        raise NotImplementedError
+
+
+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
+    based on message property dt.  Expects to be instantiated in one execution
+    context (thread, process, etc) and run in another.
+    """
+
+    def __init__(self):
+        Component.__init__(self)
+
+        self.sent_count             = 0
+        self.received_count         = 0
+        self.draining               = False
+        self.ds_finished_counter    = 0
+
+        # Depending on the size of this, might want to use a data
+        # structure with better asymptotics.
+        self.data_buffer            = {}
+        
+        # source_id -> integer count
+        self.sent_counters          = Counter()
+        self.recv_counters          = Counter()
+        
+    def init(self):
+        pass
+
+    @property
+    def get_id(self):
+        return "FEED"
+
+    @property
+    def get_type(self):
+        return COMPONENT_TYPE.CONDUIT
+
+    # -------------
+    # Core Methods
+    # -------------
+
+    def open(self):
+        self.pull_socket = self.bind_data()
+        self.feed_socket = self.bind_feed()
+
+    def do_work(self):
+        # wait for synchronization reply from the host
+        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.signal_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()
+
+            if message == str(CONTROL_PROTOCOL.DONE):
+                self.ds_finished_counter += 1
+
+                if len(self.data_buffer) == self.ds_finished_counter:
+                    #drain any remaining messages in the buffer
+                    qutil.LOGGER.debug("draining feed")
+                    self.drain()
+                    self.signal_done()
+            else:
+                try:
+                    event = self.unframe(message)
+                # deserialization error
+                except zp.INVALID_DATASOURCE_FRAME as exc:
+                    return self.signal_exception(exc)
+
+                try:
+                    self.append(event)
+                    self.send_next()
+
+                # Invalid message
+                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
+    # -------------
+
+    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(self.frame(event), self.zmq.NOBLOCK)
+            self.sent_counters[event.source_id] += 1
+            self.sent_count += 1
+
+    def append(self, event):
+        """
+        Add an event to the buffer for the source specified by
+        source_id.
+        """
+        self.data_buffer[event.source_id].append(event)
+        self.recv_counters[event.source_id] += 1
+        self.received_count += 1
+
+    def next(self):
+        """
+        Get the next message in chronological order.
+        """
+        if not(self.is_full() or self.draining):
+            return
+
+        cur_source = None
+        earliest_source = None
+        earliest_event = None
+        #iterate over the queues of events from all sources 
+        #(1 queue per datasource)
+        for events in self.data_buffer.values():
+            if len(events) == 0:
+                continue
+            cur_source = events
+            first_in_list = events[0]
+            if first_in_list.dt == None:
+                #this is a filler event, discard
+                events.pop(0)
+                continue
+                
+            if (earliest_event == None) or (first_in_list.dt <= earliest_event.dt):
+                earliest_event = first_in_list
+                earliest_source = cur_source
+
+        if earliest_event != None:
+            return earliest_source.pop(0)
+
+    def is_full(self):
+        """
+        Indicates whether the buffer has messages in buffer for
+        all un-DONE, blocking sources.
+        """
+        for source_id, events in self.data_buffer.iteritems():
+            if len(events) == 0:
+                return False
+        return True
+
+    def pending_messages(self):
+        """
+        Returns the count of all events from all sources in the
+        buffer.
+        """
+        total = 0
+        for events in self.data_buffer.values():
+            total += len(events)
+        return total
+
+    def add_source(self, source_id):
+        """
+        Add a data source to the buffer.
+        """
+        self.data_buffer[source_id] = []
+
+    def __len__(self):
+        """
+        Buffer's length is same as internal map holding separate
+        sorted arrays of events keyed by source id.
+        """
+        return len(self.data_buffer)
+
+
+class Merge(Feed):
+    """
+    Merges multiple streams of events into single messages.
+    """
+
+    def __init__(self):
+        Feed.__init__(self)
+
+        self.init()
+
+    def init(self):
+        pass
+
+    @property
+    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()
+
+    def next(self):
+        """Get the next merged message from the feed buffer."""
+        if not (self.is_full() or self.draining):
+            return
+        
+        if self.pending_messages() == 0:
+            return
+
+        #
+        #get the raw event from the passthrough transform.
+        result = self.data_buffer[zp.TRANSFORM_TYPE.PASSTHROUGH].pop(0).PASSTHROUGH
+        for source, events in self.data_buffer.iteritems():
+            if source == zp.TRANSFORM_TYPE.PASSTHROUGH:
+                continue
+            if len(events) > 0:
+                cur = events.pop(0)
+                result.merge(cur)
+        return result
+        
+    def unframe(self, msg):
+        return zp.TRANSFORM_UNFRAME(msg)   
+    
+    def frame(self, event):
+        return zp.MERGE_FRAME(event)
+
+    def append(self, event):
+        """
+        :param event: a namedict with one entry. key is the name of the 
+        transform, value is the transformed value.
+        Add an event to the buffer for the source specified by
+        source_id.
+        """
+
+        self.data_buffer[event.keys()[0]].append(event)
+        self.received_count += 1
+
+
+class BaseTransform(Component):
+    """
+    Top level execution entry point for the transform
+
+    - connects to the feed socket to subscribe to events
+    - connects 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.
+    """
+
+    def __init__(self, name):
+        Component.__init__(self)
+
+        self.state = {
+            'name': name
+        }
+
+        self.init()
+
+    def init(self):
+        pass
+
+    @property
+    def get_id(self):
+        return self.state['name']
+
+    @property
+    def get_type(self):
+        return COMPONENT_TYPE.CONDUIT
+
+    def open(self):
+        """
+        Establishes zmq connections.
+        """
+        #create the feed.
+        self.feed_socket = self.connect_feed()
+        #create the result PUSH
+        self.result_socket = self.connect_merge()
+
+    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
+
+        """
+        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.signal_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
+
+            try:
+                event = self.unframe(message)
+            except zp.INVALID_FEED_FRAME as exc:
+                return self.signal_exception(exc)
+
+            try:
+                cur_state = self.transform(event)
+
+            # 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:
+                transform_frame = self.frame(cur_state)
+            except zp.INVALID_TRANSFORM_FRAME as exc:
+                return self.signal_exception(exc)
+
+            self.result_socket.send(transform_frame, self.zmq.NOBLOCK)
+            
+    def frame(self, cur_state):
+        return zp.TRANSFORM_FRAME(cur_state['name'], cur_state['value'])
+        
+    def unframe(self, msg):
+        return zp.FEED_UNFRAME(msg)
+        
+    def transform(self, event):
+        """
+        Must return the transformed value as a map with::
+
+            {name:"name of new transform", value: "value of new field"}
+
+        Transforms run in parallel and results are merged into a single map, so
+        transform names must be unique.  Best practice is to use the self.state
+        object initialized from the transform configuration, and only set the
+        transformed value::
+
+            self.state['value'] = transformed_value
+        """
+        raise NotImplementedError
+
+
+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
+
+    #TODO, could save some cycles by skipping the _UNFRAME call and just setting value to original msg string.
+    def transform(self, event):
+        return {'name':zp.TRANSFORM_TYPE.PASSTHROUGH, 'value': zp.FEED_FRAME(event) }
+
+
+class DataSource(Component):
+    """
+    Baseclass for data sources. Subclass and implement send_all - usually this
+    means looping through all records in a store, converting to a dict, and
+    calling send(map).
+    
+    Every datasource has a dict property to hold filters::
+        - key -- name of the filter, e.g. SID
+        - value -- a primitive representing the filter. e.g. a list of ints.
+        
+    Modify the datasource's filters via the set_filter(name, value)
+    """
+    def __init__(self, source_id):
+        Component.__init__(self)
+
+        self.id = source_id
+        self.init()
+        self.filter = {}
+
+    def init(self):
+        self.cur_event = None
+
+    def set_filter(self, name, value):
+        self.filter[name] = value
+
+    @property
+    def get_id(self):
+        return self.id
+
+    @property
+    def get_type(self):
+        return COMPONENT_TYPE.SOURCE
+
+    def open(self):
+        self.data_socket = self.connect_data()
+
+    def send(self, event):
+        """
+        Emit data.
+        """
+        assert isinstance(event, zp.namedict)
+
+        event['source_id'] = self.get_id
+        event['type'] = self.get_type
+
+        try:
+            ds_frame = self.frame(event)
+        except zp.INVALID_DATASOURCE_FRAME as exc:
+            return self.signal_exception(exc)
+
+        self.data_socket.send(ds_frame)
+
+    def frame(self, event):
+        return zp.DATASOURCE_FRAME(event)

zipline/core/monitor.py

@@ -0,0 +1,622 @@
+import time
+import gevent
+import itertools
+# pyzmq
+import zmq
+import gevent_zeromq
+
+from collections import OrderedDict
+
+from protocol import CONTROL_PROTOCOL, CONTROL_FRAME, \
+    CONTROL_UNFRAME, CONTROL_STATES, INVALID_CONTROL_FRAME \
+
+states = CONTROL_STATES
+
+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 |
+#  +---+ +---+ +---+ +---+
+
+INIT, SOURCES_READY, RUNNING, TERMINATE = CONTROL_STATES
+
+state_transitions = frozenset([
+    (-1            , INIT),
+    (INIT          , SOURCES_READY),
+    (SOURCES_READY , RUNNING),
+    (INIT          , TERMINATE),
+    (SOURCES_READY , TERMINATE),
+    (RUNNING       , TERMINATE),
+])
+
+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.
+
+    :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(
+            'tcp://127.0.0.1:5000',
+            'tcp://127.0.0.1:5001',
+        )
+
+        # typically you'd want to run this async to your main
+        # program since it blocks indefinetely.
+        controller.manage(
+            [ TOPOLOGY ]
+            [ STATES ]
+        )
+
+    """
+
+    debug = False
+    period = 1
+
+    def __init__(self, pub_socket, route_socket, logging = None):
+
+        self.context    = None
+        self.zmq        = None
+        self.zmq_poller = None
+
+        self.running = False
+        self.polling = False
+        self.tracked = set()
+        self.responses = set()
+
+        self.ctime    = 0
+        self.tic      = time.time()
+        self.freeform = False
+        self._state   = -1
+
+        self.associated = []
+
+        self.pub_socket   = pub_socket
+        self.route_socket = route_socket
+
+        self.error_replay = OrderedDict()
+
+        if logging:
+            self.logging = logging
+        else:
+            import util as qutil
+            self.logging = qutil.LOGGER
+
+    def init_zmq(self, flavor):
+
+        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):
+        """
+        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)
+
+        self.polling = True
+        self.state = CONTROL_STATES.INIT
+
+    @property
+    def state(self):
+        return self._state
+
+    @state.setter
+    def state(self, new):
+        old, self._state = self._state, new
+
+        if (old, new) not in state_transitions:
+            raise RuntimeError("[Controller] Invalid State Transition : %s -> %s" %(old, new))
+        else:
+            self.logging.info("[Controller] State Transition : %s -> %s" %(old, new))
+
+    def run(self):
+        self.running = True
+        self.init_zmq(self.zmq_flavor)
+
+        try:
+            return self._poll() # use a python loop
+        except KeyboardInterrupt:
+            self.logging.info('Shutdown event loop')
+
+    def log_status(self):
+        """
+        Snapshot of the tracked components at every period.
+        """
+        #self.logging.info("[Controller] Tracking : %s" % ([c for c in self.tracked],))
+        pass
+
+    def replay_errors(self):
+        """
+        Replay the errors in the order they were reported to the
+        controller.
+        """
+        return [ a for a in sorted(self.replay_errors.keys())]
+
+    # -------------
+    # Publications
+    # -------------
+
+    def send_heart(self):
+        if not self.running:
+            return
+
+        heartbeat_frame = CONTROL_FRAME(
+            CONTROL_PROTOCOL.HEARTBEAT,
+            str(self.ctime)
+        )
+        self.pub.send(heartbeat_frame)
+
+    def send_hardkill(self):
+        if not self.running:
+            return
+
+        kill_frame = CONTROL_FRAME(
+            CONTROL_PROTOCOL.KILL,
+            ''
+        )
+        self.pub.send(kill_frame)
+
+    def send_softkill(self):
+        if not self.running:
+            return
+
+        soft_frame = CONTROL_FRAME(
+            CONTROL_PROTOCOL.SHUTDOWN,
+            ''
+        )
+        self.pub.send(soft_frame)
+
+    # -----------
+    # Event Loops
+    # -----------
+
+    def _poll(self):
+
+        assert self.route_socket
+        assert self.pub_socket
+        assert self.cancel_socket
+
+        # -- Publish --
+        # =============
+        self.pub = self.context.socket(self.zmq.PUB)
+        self.pub.bind(self.pub_socket)
+
+        # -- Cancel --
+        # =============
+        assert isinstance(self.cancel_socket,basestring), self.cancel_socket
+        self.cancel = self.context.socket(self.zmq.REP)
+        self.cancel.connect(self.cancel_socket)
+
+        # -- Router --
+        # =============
+        self.router = self.context.socket(self.zmq.ROUTER)
+        self.router.bind(self.route_socket)
+
+
+        poller = self.zmq.Poller()
+        poller.register(self.router, self.zmq.POLLIN)
+        poller.register(self.cancel, self.zmq.POLLIN)
+
+        self.associated += [self.pub, self.router, self.cancel]
+
+        # TODO: actually do this
+        self.state = CONTROL_STATES.SOURCES_READY
+
+        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 socks.get(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
+
+                if socks.get(self.cancel) == self.zmq.POLLIN:
+                    self.logging.info('[Controller] Received Cancellation')
+                    rawmessage = self.cancel.recv()
+                    self.cancel.send('')
+                    self.shutdown(soft=True)
+                    break
+
+            self.beat()
+
+            if self.zmq_flavor == 'green':
+                gevent.sleep(0)
+
+            if self.state is CONTROL_STATES.TERMINATE:
+                break
+
+            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)
+
+    # --------------
+    # Init Handlers
+    # --------------
+
+    def new_source(self):
+        if self.state is CONTROL_STATES.RUNNING:
+            self.state = SOURCES_READY
+
+    def new_universal(self):
+        pass
+
+    # The various "states of being that a component can inform us
+    # of
+    def new(self, component):
+        if self.state is CONTROL_STATES.TERMINATE:
+            return
+
+        self.logging.info('[Controller] Now Tracking "%s" ' % component)
+
+        universal = self.new_universal
+        init_handlers = {
+            'FEED' : self.new_source,
+        }
+
+        if component in self.topology or self.freeform:
+            init_handlers.get(component, universal)()
+            self.tracked.add(component)
+        else:
+            # Some sort of socket collision has occured, this is
+            # a very bad failure mode.
+            raise UnknownChatter(component)
+
+    # ------------------
+    # Epic Fail Handling
+    # ------------------
+
+    def fail_universal(self):
+        pass
+        # TODO: this requires higher order functionality
+        #self.logging.error('[Controller] System in exception state, shutting down')
+        #self.shutdown(soft=True)
+
+    def fail(self, component):
+        if self.state is CONTROL_STATES.TERMINATE:
+            return
+
+        universal = self.fail_universal
+        fail_handlers = { }
+
+        if component in self.topology or self.freeform:
+            self.logging.info('[Controller] Component "%s" timed out' % component)
+            self.tracked.remove(component)
+            fail_handlers.get(component, universal)()
+
+    # -------------------
+    # Completion Handling
+    # -------------------
+
+    def done(self, component):
+        self.logging.info('[Controller] Component "%s" done.' % component)
+
+    # --------------
+    # Error Handling
+    # --------------
+
+    def exception_universal(self):
+        """
+        Shutdown the system on failure.
+        """
+        self.logging.error('[Controller] System in exception state, shutting down')
+        self.shutdown(soft=True)
+
+    def exception(self, component, failure):
+        universal = self.exception_universal
+        exception_handlers = { }
+
+        if component in self.topology or self.freeform:
+            self.error_replay[(component, time.time())] = failure
+            self.logging.error('[Controller] Component "%s" in exception state' % component)
+
+            exception_handlers.get(component, universal)()
+        else:
+            raise UnknownChatter(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
+    # -------------------
+
+    # 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 = self.zmq.Context.instance()
+
+        s = context.socket(zmq.DEALER)
+        s.setsockopt(zmq.IDENTITY, identity)
+        s.connect(self.route_socket)
+
+        self.associated.append(s)
+        return s
+
+    def message_listener(self, context = None):
+        """
+        Spin off a socket used for receiving messages from this
+        controller.
+        """
+
+        if not context:
+            context = self.zmq.Context.instance()
+
+        s = context.socket(zmq.SUB)
+        s.connect(self.pub_socket)
+        s.setsockopt(zmq.SUBSCRIBE, '')
+
+        self.associated.append(s)
+        return s
+
+    def do_error_replay(self):
+        for (component, time), error in self.error_replay.iteritems():
+            self.logging.info('[Controller] Error Log for -- %s --:\n%s' %
+                (component, error))
+
+    def shutdown(self, hard=False, soft=True, context=None):
+
+        if not self.polling:
+            return
+
+        self.polling = False
+
+        assert hard or soft, """ Must specify kill hard or soft """
+
+        if hard:
+            self.state = CONTROL_STATES.TERMINATE
+
+            self.logging.info('[Controller] Hard Shutdown')
+
+            #for asoc in self.associated:
+                #asoc.close()
+
+        if soft:
+            self.state = CONTROL_STATES.TERMINATE
+
+            self.logging.info('[Controller] Soft Shutdown')
+            self.send_softkill()
+
+            #for asoc in self.associated:
+                #asoc.close()
+
+        self.do_error_replay()
+
+if __name__ == '__main__':
+
+    print 'Running on '\
+        'tcp://127.0.0.1:5000 '\
+        'tcp://127.0.0.1:5001 '
+
+    controller = Controller(
+        'tcp://127.0.0.1:5000',
+        'tcp://127.0.0.1:5001',
+    )
+    controller.zmq_flavor = 'green'
+
+    controller.manage(
+        'freeform',
+        []
+    )
+    controller.run()

zipline/profile/prof.py

@@ -0,0 +1,104 @@
+"""
+
+Viscosity - Tools for benchmarking ZeroMQ data flow.
+
+"""
+
+import time as timer
+import logging
+import pycounters
+from contextlib import contextmanager, nested
+from pycounters import base
+from pycounters.shortcuts import frequency, time
+from pycounters import shortcuts, reporters, start_auto_reporting, register_reporter
+from pycounters import shortcuts,reporters,report_value, output_report, \
+counters, register_counter, _reporting_decorator_context_manager
+
+JSONFile = "counters.json"
+
+logger = logging.getLogger('simple_example')
+logger.setLevel(logging.DEBUG)
+
+ch = logging.StreamHandler()
+ch.setLevel(logging.DEBUG)
+logger.addHandler(ch)
+
+reporter = reporters.JSONFileReporter(output_file=JSONFile)
+logreport = reporters.LogReporter(logger)
+register_reporter(logreport)
+register_reporter(reporter)
+
+class timecontext:
+
+    def __init__(self, name):
+        self.name = name
+
+    def __enter__(self):
+        cntr = base.GLOBAL_REGISTRY.get_counter(self.name, throw=False)
+        if not cntr:
+            counter = counters.AverageTimeCounter(self.name)
+            register_counter(counter)
+        self.tic = timer.time()
+        return self
+
+    def __exit__(self, exc_type, exc_val, exc_tb):
+        if not exc_type:
+            shortcuts.value(self.name, timer.time() - self.tic)
+
+class ttimecontext:
+
+    def __init__(self, name):
+        self.name = name
+
+    def __enter__(self):
+        counter = base.GLOBAL_REGISTRY.get_counter(self.name, throw=False)
+
+        if not counter:
+            counter = counters.EventCounter(self.name)
+            counter.value = 0
+            register_counter(counter)
+
+        self.counter = counter
+        self.tic = timer.time()
+        return self
+
+    def __exit__(self, exc_type, exc_val, exc_tb):
+        if not exc_type:
+            val = (timer.time() - self.tic)
+            if not self.counter.value:
+                self.counter.value = long(0.0)
+            self.counter.value += val
+
+class occurancecontext:
+
+    def __init__(self, name):
+        self.name = name
+
+    def __enter__(self):
+        cntr = base.GLOBAL_REGISTRY.get_counter(self.name, throw=False)
+        if not cntr:
+            cntr = counters.TotalCounter(self.name)
+            counter = counters.TotalCounter(self.name)
+            register_counter(counter)
+        return self
+
+    def __exit__(self, exc_type, exc_val, exc_tb):
+        shortcuts.value(self.name, 1)
+
+if __name__ == '__main__':
+
+    with timecontext('average time'):
+        for i in xrange(5):
+            x = [2] * 1000
+            timer.sleep(0.01)
+
+    with occurancecontext('totalcount'):
+        for i in xrange(5):
+            x = [2] * 1000
+
+    with ttimecontext('total time'):
+        for i in xrange(5):
+            x = [2] * 1000
+            timer.sleep(1)
+
+    pycounters.output_report()

zipline/topology.py

@@ -1,80 +0,0 @@
-"""
-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     = []

zipline/utils/protocol_utils.py

@@ -0,0 +1,221 @@
+import copy
+import pandas
+from ctypes import Structure, c_ubyte
+from collections import MutableMapping
+from itertools import izip
+
+def Enum(*options):
+    """
+    Fast enums are very important when we want really tight zmq
+    loops. These are probably going to evolve into pure C structs
+    anyways so might as well get going on that.
+    """
+    class cstruct(Structure):
+        _fields_ = [(o, c_ubyte) for o in options]
+        __iter__ = lambda s: iter(range(len(options)))
+    return cstruct(*range(len(options)))
+
+def FrameExceptionFactory(name):
+    """
+    Exception factory with a closure around the frame class name.
+    """
+    class InvalidFrame(Exception):
+        def __init__(self, got):
+            self.got = got
+
+        def __str__(self):
+            return "Invalid {framecls} Frame: {got}".format(
+                framecls = name,
+                got = self.got,
+            )
+
+    return InvalidFrame
+
+class namedict(MutableMapping):
+    """
+
+    Namedicts are dict like objects that have fields accessible by attribute lookup
+    as well as being indexable and iterable::
+
+        HEARTBEAT_PROTOCOL = namedict({
+            'REQ' : b'\x01',
+            'REP' : b'\x02',
+        })
+
+        HEARTBEAT_PROTOCOL.REQ # syntactic sugar
+        HEARTBEAT_PROTOCOL.REP # oh suga suga
+
+    For more complex structs use collections.namedtuple:
+    """
+
+    def __init__(self, dct=None):
+        if(dct):
+            self.__dict__.update(dct)
+
+    def __setitem__(self, key, value):
+        """
+        Required for use by pymongo as_class parameter to find.
+        """
+        if(key == '_id'):
+            self.__dict__['id'] = value
+        else:
+            self.__dict__[key] = value
+
+    def __getitem__(self, key):
+        return self.__dict__[key]
+
+    def __delitem__(self, key):
+        del self.__dict__[key]
+
+    def __iter__(self):
+        return self.__dict__.iterkeys()
+
+    def __len__(self):
+        return len(self.__dict__)
+
+    def keys(self):
+        return self.__dict__.keys()
+
+    def as_dict(self):
+        # shallow copy is O(n)
+        return copy.copy(self.__dict__)
+
+    def delete(self, key):
+        del(self.__dict__[key])
+
+    def merge(self, other_nd):
+        assert isinstance(other_nd, namedict)
+        self.__dict__.update(other_nd.__dict__)
+
+    def __repr__(self):
+        return "namedict: " + str(self.__dict__)
+
+    def __eq__(self, other):
+        # !!!!!!!!!!!!!!!!!!!!
+        # !!!! DANGEROUS !!!!!
+        # !!!!!!!!!!!!!!!!!!!!
+        return other != None and self.__dict__ == other.__dict__
+
+    def has_attr(self, name):
+        return self.__dict__.has_key(name)
+
+    def as_series(self):
+        s = pandas.Series(self.__dict__)
+        s.name = self.sid
+        return s
+
+class ndict(MutableMapping):
+    """
+    Xtreme Namedicts 2.0
+
+    Ndicts are dict like objects that have fields accessible by attribute
+    lookup as well as being indexable and iterable. Done right
+    this time.
+    """
+
+    def __init__(self, dct=None):
+        self.__internal = dict()
+        self.cls = frozenset(dir(self))
+
+        if dct:
+            self.__internal.update(dct)
+
+    # Abstact Overloads
+    # -----------------
+
+    def __setitem__(self, key, value):
+        """
+        Required for use by pymongo as_class parameter to find.
+        """
+        if key == '_id':
+            self.__internal['id'] = value
+        else:
+            self.__internal[key] = value
+
+
+    def __getattr__(self, key):
+        if key in self.cls:
+            return self.__dict__[key]
+        else:
+            return self.__internal[key]
+
+    def __getitem__(self, key):
+        return self.__internal[key]
+
+    def __delitem__(self, key):
+        del self.__internal[key]
+
+    def __iter__(self):
+        return self.__internal.iterkeys()
+
+    def __len__(self):
+        return len(self.__internal)
+
+    # Compatability with namedicts
+    # ----------------------------
+
+    # for compat, not the Python way to do things though...
+    # Deprecated, use builtin ``del`` operator.
+    delete = __delitem__
+
+    def has_attr(self, key):
+        """
+        Deprecated, use builtin ``in`` operator.
+        """
+        return self.__contains__(key)
+
+    def has_key(self, key):
+        return self.__contains__(key)
+
+    # Custom Methods
+    # --------------
+
+    def copy(self):
+        return ndict(copy.copy(self.__internal))
+
+    def as_dataframe(self):
+        """
+        Return the representation as a Pandas dataframe.
+        """
+        d = pandas.DataFrame(self.__internal)
+        return d
+
+    def as_series(self):
+        """
+        Return the representation as a Pandas time series.
+        """
+        s = pandas.Series(self.__internal)
+        s.name = self.sid
+        return s
+
+    def as_dict(self):
+        """
+        Return the representation as a vanilla Python dict.
+        """
+        # shallow copy is O(n)
+        return copy.copy(self.__internal)
+
+    def merge(self, other_nd):
+        """
+        Merge in place with another ndict.
+        """
+        assert isinstance(other_nd, ndict)
+        self.__internal.update(other_nd.__internal)
+
+    def __repr__(self):
+        return "namedict: " + str(self.__internal)
+
+    # Faster dictionary comparison?
+    #def __eq__(self, other):
+        #assert isinstance(other, ndict)
+
+        #keyeq = set(self.keys()) == set(other.keys())
+
+        #if not keyeq:
+            #return False
+
+        #for i, j in izip(self.itervalues(), other.itervalues()):
+            #if i != j:
+                #return False
+
+        #return True

zipline/version.py

@@ -0,0 +1,9 @@
+BANNER = """
+Zipline {version}
+Released under BSD3
+""".strip()
+
+VERSION = ( 0, 0, 1, 'dev' )
+
+def pretty_version():
+    return BANNER.format(version='.'.join(VERSION))