diff --git a/docs/conf.py b/docs/conf.py index 743659ff..8ee7280d 100644 --- a/docs/conf.py +++ b/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'] diff --git a/docs/index.rst b/docs/index.rst index 43529a94..44172299 100644 --- a/docs/index.rst +++ b/docs/index.rst @@ -6,7 +6,7 @@ Contents: .. toctree:: - :maxdepth: 2 + :maxdepth: 4 notes.rst modules.rst diff --git a/docs/zipline.rst b/docs/zipline.rst index 3d8a31f0..386b03de 100644 --- a/docs/zipline.rst +++ b/docs/zipline.rst @@ -9,6 +9,14 @@ zipline Package :undoc-members: :show-inheritance: +:mod:`cli` Module +----------------- + +.. automodule:: zipline.cli + :members: + :undoc-members: + :show-inheritance: + :mod:`component` Module ----------------------- @@ -17,30 +25,6 @@ zipline Package :undoc-members: :show-inheritance: -:mod:`daemon` Module --------------------- - -.. automodule:: zipline.daemon - :members: - :undoc-members: - :show-inheritance: - -:mod:`db` Module ----------------- - -.. automodule:: zipline.db - :members: - :undoc-members: - :show-inheritance: - -:mod:`host_settings` Module ---------------------------- - -.. automodule:: zipline.host_settings - :members: - :undoc-members: - :show-inheritance: - :mod:`messaging` Module ----------------------- @@ -65,6 +49,14 @@ zipline Package :undoc-members: :show-inheritance: +:mod:`serial` Module +-------------------- + +.. automodule:: zipline.serial + :members: + :undoc-members: + :show-inheritance: + :mod:`sources` Module --------------------- @@ -73,6 +65,14 @@ zipline Package :undoc-members: :show-inheritance: +:mod:`topos` Module +------------------- + +.. automodule:: zipline.topos + :members: + :undoc-members: + :show-inheritance: + :mod:`util` Module ------------------ @@ -86,7 +86,6 @@ Subpackages .. toctree:: - zipline.finance zipline.test zipline.transforms diff --git a/docs/zipline.test.rst b/docs/zipline.test.rst index 743495b8..443c1e28 100644 --- a/docs/zipline.test.rst +++ b/docs/zipline.test.rst @@ -9,22 +9,6 @@ test Package :undoc-members: :show-inheritance: -:mod:`dummy` Module -------------------- - -.. automodule:: zipline.test.dummy - :members: - :undoc-members: - :show-inheritance: - -:mod:`factory` Module ---------------------- - -.. automodule:: zipline.test.factory - :members: - :undoc-members: - :show-inheritance: - :mod:`test_devsimulator` Module ------------------------------- @@ -33,14 +17,6 @@ test Package :undoc-members: :show-inheritance: -:mod:`test_finance` Module --------------------------- - -.. automodule:: zipline.test.test_finance - :members: - :undoc-members: - :show-inheritance: - :mod:`test_messaging` Module ---------------------------- @@ -49,6 +25,14 @@ test Package :undoc-members: :show-inheritance: +:mod:`test_monitor` Module +-------------------------- + +.. automodule:: zipline.test.test_monitor + :members: + :undoc-members: + :show-inheritance: + :mod:`test_sanity` Module ------------------------- diff --git a/etc/requirements.txt b/etc/requirements.txt index 31957e09..8c2eea08 100644 --- a/etc/requirements.txt +++ b/etc/requirements.txt @@ -3,4 +3,4 @@ pyzmq==2.1.11 gevent-zeromq==0.2.2 msgpack-python==0.1.12 humanhash==0.0.1 -pymongo==2.1.1 +ujson=1.18 diff --git a/zipline/component.py b/zipline/component.py index 7a8b2cc4..8c6621d9 100644 --- a/zipline/component.py +++ b/zipline/component.py @@ -1,8 +1,13 @@ """ Commonly used messaging components. + +Contains the base class for all components. + """ + import os import uuid +import time import socket import humanhash @@ -10,66 +15,113 @@ import zipline.util as qutil from zipline.protocol import CONTROL_PROTOCOL, COMPONENT_STATE 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 ParallelBuffer (aka + the 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 MergedParallelBuffer (aka the + 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): - """ - :addresses: a dict of name_string -> zmq port address strings. Must have the following entries:: - - - 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. - - data_address: socket address used for data sources to stream their records. - will be used in PUSH/PULL sockets between data sources and a ParallelBuffer (aka the Feed). Bind - will always be on the PULL side (we always have N producers and 1 consumer) - - 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. - - merge_address: socket address used to publish transformed values. - will be used in PUSH/PULL from many transforms to one MergedParallelBuffer (aka the Merge). Bind - will always be on the PULL side (we always have N producers and 1 consumer) - - 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. Any sockets on which recv is expected to be called - will also return a Poller. - - """ self.zmq = None self.context = None self.addresses = None self.out_socket = None self.gevent_needed = False self.killed = False + 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. 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): - raise NotImplementedError - - def destroy(self): """ - Tear down after normal operation. + Open the connections needed to start doing work. """ raise NotImplementedError - def kill(self): + def ready(self): """ - Tear down ( fast ) as a mode of failure in the - simulation. + Return ``True`` if and only if the component has finished execution. """ - raise NotImplementedError + 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 = [] @@ -81,7 +133,14 @@ class Component(object): import zmq self.zmq = zmq + # TODO: this can cause max fd errors on BSD machines with + # low ulimits, its perfectly fine to use one Context in + # multithreaded enviroments, its only in multiprocess + # systems where this becomes needed. Add this option. + # + # http://zeromq.github.com/pyzmq/morethanbindings.html#thread-safety self.context = self.zmq.Context() + self.setup_poller() self.open() @@ -89,11 +148,13 @@ class Component(object): self.setup_control() self.loop() - #close all the sockets - for sock in self.sockets: - sock.close() - def run(self, catch_exceptions=False): + self.end_tick = time.clock() + + # shouldn't block if we've done our job correctly + # self.context.term() + + def run(self, catch_exceptions=True): """ Run the component. @@ -114,14 +175,21 @@ class Component(object): self.signal_exception(exc) fail = exc finally: + + self.shutdown() + self.teardown_sockets() + if self.context: self.context.destroy() if fail: raise fail else: - self._run() - if(self.context != None): - self.context.destroy() + try: + self._run() + finally: + self.shutdown() + self.teardown_sockets() + def loop(self): """ @@ -141,12 +209,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): @@ -258,15 +364,15 @@ class Component(object): """ assert self.controller - self.control_out = self.controller.message_sender() - self.control_in = self.controller.message_listener() + self.control_out = self.controller.message_sender(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. + Setup the sync socket and poller. ( Connect ) """ qutil.LOGGER.debug("Connecting sync client for {id}".format(id=self.get_id)) @@ -275,7 +381,7 @@ class Component(object): self.sync_socket.connect(self.addresses['sync_address']) #self.sync_socket.setsockopt(self.zmq.LINGER,0) - # Explictly, a different poller for obvious reasons. + # Explictly a different poller for obvious reasons. # I'm not fond of having this poller init'd as a side # effect of a method call. Still thinking about where to # put it at the moment though... @@ -288,21 +394,66 @@ class Component(object): # 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. + """ 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. """ - return ( - self.get_id , - self.huid , - socket.gethostname() , - os.getpid() , - hex(id(self)) , - ) + 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): """ diff --git a/zipline/messaging.py b/zipline/messaging.py index 70268774..5d367899 100644 --- a/zipline/messaging.py +++ b/zipline/messaging.py @@ -1,14 +1,14 @@ """ Commonly used messaging components. """ -import json -import uuid + import datetime -import zipline.protocol as zp +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): """ @@ -18,13 +18,21 @@ class ComponentHost(Component): def __init__(self, addresses, gevent_needed=False): Component.__init__(self) - self.addresses = addresses + self.addresses = addresses + self.gevent_needed = gevent_needed + self.running = False + self.init() + + def init(self): + + # Component Registry + # ---------------------- self.components = {} + # ---------------------- + self.sync_register = {} self.timeout = datetime.timedelta(seconds=5) - self.gevent_needed = gevent_needed - self.heartbeat_timeout = 2000 self.feed = ParallelBuffer() self.merge = MergedParallelBuffer() @@ -35,25 +43,38 @@ class ComponentHost(Component): 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): @@ -62,6 +83,7 @@ class ComponentHost(Component): def setup_sync(self): """ + Setup the sync socket and poller. ( Bind ) """ qutil.LOGGER.debug("Connecting sync server.") @@ -108,17 +130,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(':') - if len(parts) != 2: - qutil.LOGGER.info("got bad confirm: {msg}".format(msg=msg)) - continue - - sync_id, status = parts + 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() @@ -126,28 +148,18 @@ class ComponentHost(Component): # 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 - -class SimulatorBase(ComponentHost): - """ - Simulator coordinates the launch and communication of source, feed, transform, and merge components. - """ - - def __init__(self, addresses, gevent_needed=False): - """ - """ - ComponentHost.__init__(self, addresses, gevent_needed) - - def simulate(self): - self.run() - - def get_id(self): - return "Simulator" + def teardown_component(self, component): + raise NotImplementedError class ParallelBuffer(Component): @@ -160,20 +172,30 @@ class ParallelBuffer(Component): def __init__(self): Component.__init__(self) + self.sent_count = 0 self.received_count = 0 self.draining = False - #data source component ID -> List of messages - 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 @property 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() @@ -183,32 +205,66 @@ class ParallelBuffer(Component): # wait for synchronization reply from the host socks = dict(self.poll.poll(self.heartbeat_timeout)) #timeout after 2 seconds. + if self.control_in in socks and socks[self.control_in] == self.zmq.POLLIN: + msg = self.control_in.recv() + 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 + #drain any remaining messages in the buffer self.drain() self.signal_done() else: - event = self.unframe(message) - self.append(event) - self.send_next() - - 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) + try: + event = json.loads(message) - def append(self, event): + # 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): """ Add an event to the buffer for the source specified by source_id. """ - self.data_buffer[event.source_id].append(event) + self.data_buffer[source_id].append(value) self.received_count += 1 def next(self): @@ -224,7 +280,7 @@ class ParallelBuffer(Component): if len(events) == 0: continue cur = events - if (earliest == None) or (cur[0].dt <= earliest[0].dt): + if (earliest == None) or (cur[0]['dt'] <= earliest[0]['dt']): earliest = cur if earliest != None: @@ -250,31 +306,18 @@ 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() + self.data_buffer[source_id] = [] - def send_next(self): + def __len__(self): """ - Send the (chronologically) next message in the buffer. + Buffer's length is same as internal map holding separate + sorted arrays of events keyed by source id. """ - 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_count += 1 - - def unframe(self, msg): - return zp.DATASOURCE_UNFRAME(msg) - - def frame(self, event): - return zp.FEED_FRAME(event) + return len(self.data_buffer) class MergedParallelBuffer(ParallelBuffer): @@ -285,6 +328,19 @@ class MergedParallelBuffer(ParallelBuffer): def __init__(self): ParallelBuffer.__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() @@ -295,58 +351,50 @@ class MergedParallelBuffer(ParallelBuffer): return #get the raw event from the passthrough transform. - result = self.data_buffer["PASSTHROUGH"].pop(0).PASSTHROUGH + result = self.data_buffer["PASSTHROUGH"].pop(0)['value'] for source, events in self.data_buffer.iteritems(): if source == "PASSTHROUGH": continue if len(events) > 0: cur = events.pop(0) - result.merge(cur) + result[source] = cur['value'] return result - @property - def get_id(self): - return "MERGE" - - 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.__dict__.keys()[0]].append(event) - self.received_count += 1 - 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.""" + method to create a new derived value from the combined feed. + """ def __init__(self, name): Component.__init__(self) - self.state = {} - self.state['name'] = name + + 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. @@ -359,21 +407,45 @@ 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)) #timeout after 2 seconds. + socks = dict(self.poll.poll(self.heartbeat_timeout)) + + if self.control_in in socks and socks[self.control_in] == self.zmq.POLLIN: + msg = self.control_in.recv() + 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 - event = zp.FEED_UNFRAME(message) - cur_state = self.transform(event) - qutil.LOGGER.info("state of transform is: {state}".format(state=cur_state)) - self.result_socket.send(zp.TRANSFORM_FRAME(cur_state['name'], cur_state['value']), self.zmq.NOBLOCK) + try: + event = json.loads(message) + except ValueError as exc: + return self.signal_exception(exc) + + 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): """ @@ -392,25 +464,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") - 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)) #timeout after 2 seconds. - 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 - #message is already FEED_FRAMEd, send it as the value. - self.result_socket.send(zp.TRANSFORM_FRAME("PASSTHROUGH", message), self.zmq.NOBLOCK) + self.init() + + def init(self): + pass + + @property + def get_type(self): + return COMPONENT_TYPE.CONDUIT + + def transform(self, event): + return { 'value': event } class DataSource(Component): @@ -419,31 +496,38 @@ class DataSource(Component): means looping through all records in a store, converting to a dict, and calling send(map). """ - def __init__(self, source_id, *args): + def __init__(self, source_id): Component.__init__(self) - self.source_id = source_id + + self.id = source_id + self.init() + + def init(self): self.cur_event = None - self.init_ds(args) - - def init_ds(*args): - pass @property def get_id(self): - return self.source_id + 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 get_type(self): - raise NotImplementedError - - - - - - + def send(self, event): + """ + Emit data. + """ + assert isinstance(event, dict) + event['id'] = self.id + event['type'] = self.get_type() + try: + json_frame = json.dumps(event) + except ValueError as exc: + return self.signal_exception(exc) + self.data_socket.send(json_frame) diff --git a/zipline/monitor.py b/zipline/monitor.py index d5fea061..7f6cab2d 100644 --- a/zipline/monitor.py +++ b/zipline/monitor.py @@ -1,16 +1,55 @@ import zmq +from zipline.protocol import CONTROL_PROTOCOL, CONTROL_FRAME class Controller(object): """ - A broker of sorts. + A N to M messaging system for inter component communication. + Ostensibly a broker of sorts. Putting messages to the broker + is durable, if the broker goes down messages will queue up + until the HWM and then go out when the new broker comes up. + + The other end is not durable, it is simply PUB/SUB which has + the benefit of of allowing more fluid time evolution of the + whole system since the messaging passing topology will not + alter itself as a result of more nodes listening. + + The actual brokerin' is either a Python loop ( slow ) or a + zmq.FORWARDER device ( fast ). + + :param pull_socket: Socket to subscribe to for republication of + published messages. The endpoint for + :func message_sender: . + :param pub_socket: Socket to publish messages, the starting + point of :func message_listener: . + :param logging: Logging interface for tracking broker state + Defaults to None + + 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.run() + + + sub = self.controller.message_listener() + push = self.controller.message_sender() + + push.send('DIE') + sub.recv() + """ - polling = False debug = False def __init__(self, pull_socket, pub_socket, logging = None): self._ctx = None + polling = False self.associated = [] @@ -18,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 @@ -30,6 +70,9 @@ class Controller(object): self.failed = 0 def run(self, debug=False, context=None): + """ + Run's the loop for the broker. + """ self.polling = True if not context: @@ -37,10 +80,10 @@ class Controller(object): else: self._ctx = context - if debug: - return self._poll_fast() # the c loop - else: - return self._poll() # use a python loop + #if not debug: + #return self._poll_fast() # the c loop + #else: + return self._poll() # use a python loop def _poll_fast(self): """ @@ -49,6 +92,9 @@ class Controller(object): self.pull = self._ctx.socket(zmq.PULL) self.pub = self._ctx.socket(zmq.PUB) + self.pull.bind(self.pull_socket) + self.pub.bind(self.pub_socket) + zmq.device(zmq.FORWARDER, self.pull, self.pub) def _poll(self): @@ -60,15 +106,14 @@ class Controller(object): self.pull = self._ctx.socket(zmq.PULL) self.pub = self._ctx.socket(zmq.PUB) - self.associated.extend([self.pull, self.pub]) - self.pull.bind(self.pull_socket) self.pub.bind(self.pub_socket) + self.associated.extend([self.pull, self.pub]) + while self.polling: try: msg = self.pull.recv() - print msg self.pub.send(msg) except KeyboardInterrupt: self.polling = False @@ -84,6 +129,8 @@ class Controller(object): self.failed += 1 continue + self.terminated = True + # ------------------- # Hooks for Endpoints # ------------------- @@ -117,20 +164,31 @@ class Controller(object): self.associated.append(s) return s + def shutdown(self, context=None): + self.polling = False + + if not context: + context = zmq.Context() + + #logging.info('Shutdown controller') + + 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 + self.shutdown() - for asoc in self.associated: - asoc.close() - - #if self._ctx: - #self._ctx.destroy() - - def __del__(self): - self.destroy() + #def __del__(self): + #self.shutdown() def qos(self): if not self.debug: diff --git a/zipline/protocol.py b/zipline/protocol.py index dc3fb408..c709019b 100644 --- a/zipline/protocol.py +++ b/zipline/protocol.py @@ -78,13 +78,13 @@ def FrameExceptionFactory(name): class namedict(object): """ - So that you can use: + So that you can use:: foo.BAR -- or -- foo['BAR'] - For more complex strcuts use collections.namedtuple: + For more complex structs use collections.namedtuple: """ def __init__(self, dct=None): @@ -130,18 +130,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: @@ -167,6 +167,12 @@ HEARTBEAT_PROTOCOL = namedict({ # Component State # ================== +COMPONENT_TYPE = Enum( + 'SOURCE' , # 0 + 'CONDUIT' , # 1 + 'SINK' , # 2 +) + COMPONENT_STATE = Enum( 'OK' , # 0 'DONE' , # 1 @@ -426,7 +432,6 @@ def UNPACK_DATE(payload): del(payload.__dict__['micros']) payload['dt'] = dt return payload - FINANCE_PROTOCOL = Enum( 'ORDER' , # 0 - req diff --git a/zipline/serial.py b/zipline/serial.py new file mode 100644 index 00000000..b0eba920 --- /dev/null +++ b/zipline/serial.py @@ -0,0 +1,91 @@ +""" +Format serializer for Zipline. + +Because I'm opinionated about how you should send things over +ZeroMQ. :) +""" + +import zlib +#import blosc +import hmac +import base64 +import numpy +import pandas + +import cPickle as pickle + +# Pickle does the equivelant of builtin ``eval``. Be afraid, be +# very afraid. + +def send_zipped_pickle(socket, obj, flags=0, protocol=-1): + """ + Pickle an object, and zip the pickle before sending it. + """ + p = pickle.dumps(obj, protocol) + z = zlib.compress(p) + return socket.send(z, flags=flags) + +def recv_zipped_pickle(socket, flags=0, protocol=-1): + """ + Unpickle and uncompress a received object. + """ + z = socket.recv(flags) + p = zlib.uncompress(z) + return pickle.loads(p, protocol=protocol) + +# HDF5, Numpy Byte Strings, Pandas arrays should use +# blosc and reconstruct the Python container from the byte string +# on the other side. + +def send_numpy(socket, obj, flags=0): + packed = blosc.pack_array(obj) + return socket.send(packed, flags=flags) + +def recv_numpy(socket, flags=0): + packed = blosc.unpack_array(socket.recv(flags)) + return socket.send(packed, flags=flags) + +def send_pandas(socket, obj, flags=0): + ndarray = obj._data.blocks[0].values + socket.send_multipart(ndarray, flags=flags) + spec = ( + obj._data.index, + obj._data.columns, + obj._data.blocks[0].dtype + ) + return socket.send_multipart(spec, flags) + +def recv_pandas(socket, flags=0): + ndarray = socket.recv_multipart(flags) + spec = socket.recv_multipart(flags) + return pandas.DataFrame._init_ndarray(ndarray, *spec) + +def send_hdf5(self): + pass + +def recv_hdf5(self): + pass + +# Cryptographically secure wire protocol for ZeroMQ Using HMAC. + +# Compare byte strings, backported from Python 3. +def byte_eq(a, b): + return not sum(0 if x==y else 1 for x, y in zip(a, b)) and len(a) == len(b) + +def send_secure(socket, data, key, flags=0): + msg = base64.b64encode(data) + sig = base64.b64encode(hmac.new(key, msg).digest()) + return socket.send(bytes('!') + sig + bytes('?') + msg, flags=flags) + +def recv_secure(socket, data, key, flags): + data = socket.recv(flags=flags) + + try: + sig, msg = data.split(bytes('?'), 1) + except ValueError: + raise Exception('Invalid signature/message pair.') + + if byte_eq(sig[1:], base64.b64encode(hmac.new(key, msg).digest())): + return base64.b64decode(msg) + else: + raise Exception('Cryptographically invalid message received') diff --git a/zipline/test/client.py b/zipline/test/client.py index e9649977..48413114 100644 --- a/zipline/test/client.py +++ b/zipline/test/client.py @@ -1,61 +1,70 @@ -import json +#import logging +import ujson as json + import zipline.util as qutil import zipline.messaging as qmsg +from zipline.protocol import CONTROL_PROTOCOL, COMPONENT_TYPE -from zipline.finance.trading import TradeSimulationClient -import zipline.protocol as zp +#qlogger = logging.getLogger('qexec') class TestClient(qmsg.Component): - """no-op client - Just connects to the merge and counts messages. compares received message count to the expected count.""" def __init__(self, expected_msg_count=0): qmsg.Component.__init__(self) - self.received_count = 0 + + # Generally shouldn't reference unit tests here. + self.utest = utest self.expected_msg_count = expected_msg_count - self.prev_dt = None - self.heartbeat_timeout = 2000 + self.init() + + def init(self): + 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() def do_work(self): socks = dict(self.poll.poll(self.heartbeat_timeout)) - if self.data_feed in socks and socks[self.data_feed] == self.zmq.POLLIN: - msg = self.data_feed.recv() + if self.control_in in socks and socks[self.control_in] == self.zmq.POLLIN: + msg = self.control_in.recv() - if msg == str(zp.CONTROL_PROTOCOL.DONE): + if self.data_feed in socks and socks[self.data_feed] == self.zmq.POLLIN: + msg = self.data_feed.recv() + #logger.info('msg:' + str(msg)) + + if msg == str(CONTROL_PROTOCOL.DONE): qutil.LOGGER.info("Client is DONE!") self.signal_done() - if(self.expected_msg_count > 0): - assert self.received_count == self.expected_msg_count - return self.received_count += 1 - event = zp.MERGE_UNFRAME(msg) - 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)) - self.prev_dt = event.dt - if(self.received_count % 100 == 0): + 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 = self.prev_dt + ) + ) + else: + self.prev_dt = event['dt'] + + if self.received_count % 100 == 0: qutil.LOGGER.info("received {n} messages".format(n=self.received_count)) - -class TestTradingClient(TradeSimulationClient): - - def __init__(self, count): - TradeSimulationClient.__init__(self) - self.count = count - self.incr = 0 - - def handle_events(self, event_queue): - #place an order for 100 shares of sid:133 - if(self.incr < self.count): - self.order(133, 100) - self.incr += 1 - diff --git a/zipline/test/test_devsimulator.py b/zipline/test/test_devsimulator.py index b7a556d8..e69de29b 100644 --- a/zipline/test/test_devsimulator.py +++ b/zipline/test/test_devsimulator.py @@ -1,10 +0,0 @@ -from unittest2 import TestCase -from zipline.test.test_messaging import SimulatorTestCase -from zipline.test.dummy import ThreadPoolExecutorMixin - - -class ThreadPoolExecutor(SimulatorTestCase, TestCase): - - def test_universe(self): - # first order logic is working today. Yay! - self.assertTrue(True != False) diff --git a/zipline/test/test_messaging.py b/zipline/test/test_messaging.py index 32384c97..afa9a5a3 100644 --- a/zipline/test/test_messaging.py +++ b/zipline/test/test_messaging.py @@ -1,19 +1,74 @@ """ -Test suite for the messaging infrastructure of Zipline. +Test suite for the messaging infrastructure of QSim. """ #don't worry about excessive public methods pylint: disable=R0904 -import zipline.messaging as qmsg +from collections import defaultdict from zipline.transforms.technical import MovingAverage from zipline.sources import RandomEquityTrades -from zipline.test.dummy import ThreadPoolExecutorMixin + 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. -class SimulatorTestCase(ThreadPoolExecutorMixin): - +class SimulatorTestCase(object): + + leased_sockets = defaultdict(list) + + def setUp(self): + self.setup_logging() + + # TODO: how to make Nose use this cross-process???? + self.setup_allocator() + + def tearDown(self): + pass + #self.unallocate_sockets() + + # Assert the sockets were properly cleaned up + #self.assertEmpty(self.leased_sockets[self.id()].values()) + + # Assert they were returned to the heap + #self.allocator.socketheap.assert + + def get_simulator(self): + """ + Return a new simulator instance to be tested. + """ + raise NotImplementedError + + def get_controller(self): + """ + Return a new controler for simulator instance to be tested. + """ + raise NotImplementedError + + def setup_allocator(self): + """ + Setup the socket allocator for this test case. + """ + raise NotImplementedError + + def allocate_sockets(self, n): + """ + Allocate sockets local to this test case, track them so + we can gc after test run. + """ + + assert isinstance(n, int) + assert n > 0 + + leased = self.allocator.lease(n) + + self.leased_sockets[self.id()].extend(leased) + return leased + + def unallocate_sockets(self): + self.allocator.reaquire(*self.leased_sockets[self.id()]) + # ------- # Cases # ------- @@ -52,20 +107,68 @@ class SimulatorTestCase(ThreadPoolExecutorMixin): # Simulation # ---------- - sim.simulate() + sim_context = sim.simulate() + sim_context.join() # 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_context = sim.simulate() + sim_context.join() + + # 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): @@ -98,14 +201,13 @@ class SimulatorTestCase(ThreadPoolExecutorMixin): # Simulation # ---------- - sim.simulate() + sim_context = sim.simulate() + sim_context.join() # 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." @@ -145,34 +247,15 @@ class SimulatorTestCase(ThreadPoolExecutorMixin): # Simulation # ---------- - sim.simulate() + sim_context = sim.simulate() + sim_context.join() # 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(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() - - diff --git a/zipline/test/transform.py b/zipline/test/transform.py new file mode 100644 index 00000000..0f81a3f5 --- /dev/null +++ b/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 } diff --git a/zipline/topology.py b/zipline/topology.py new file mode 100644 index 00000000..b7e842c9 --- /dev/null +++ b/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 = [] diff --git a/zipline/transforms/technical.py b/zipline/transforms/technical.py index d235a22a..13b5b1e5 100644 --- a/zipline/transforms/technical.py +++ b/zipline/transforms/technical.py @@ -5,6 +5,7 @@ TODO: add trailing stop """ import datetime + from zipline.messaging import BaseTransform import zipline.util as qutil @@ -16,18 +17,25 @@ class MovingAverage(BaseTransform): def __init__(self, name, days): BaseTransform.__init__(self, name) + self.window = datetime.timedelta(days = days) + + self.init() + + def init(self): self.events = [] self.current_total = 0 - self.window = datetime.timedelta(days = days) def transform(self, event): - """Update the moving average with the latest data point.""" + """ + Update the moving average with the latest data point. + """ self.events.append(event) self.current_total += event.price event_date = event.dt index = 0 + for cur_event in self.events: cur_date = cur_event.dt if(cur_date - event_date) >= self.window: @@ -37,11 +45,10 @@ class MovingAverage(BaseTransform): else: break - if(len(self.events) == 0): + if len(self.events) == 0: return 0.0 self.average = self.current_total/len(self.events) - self.state['value'] = self.average - return self.state + return self.state