From 64383f865d5b5eb999a711a8ab03f7b24d2050af Mon Sep 17 00:00:00 2001 From: Stephen Diehl Date: Mon, 20 Feb 2012 16:07:04 -0500 Subject: [PATCH 1/8] Tweak tests. --- zipline/test/client.py | 16 +++----- zipline/test/test_messaging.py | 73 +++++++++++++++++----------------- 2 files changed, 42 insertions(+), 47 deletions(-) diff --git a/zipline/test/client.py b/zipline/test/client.py index 0fad7187..3b975ec2 100644 --- a/zipline/test/client.py +++ b/zipline/test/client.py @@ -3,20 +3,20 @@ import zipline.util as qutil import zipline.messaging as qmsg class TestClient(qmsg.Component): - + def __init__(self, utest, expected_msg_count=0): qmsg.Component.__init__(self) self.received_count = 0 self.expected_msg_count = expected_msg_count self.utest = utest self.prev_dt = None - + def get_id(self): - return "TEST_CLIENT" - + return "TEST_CLIENT" + def open(self): self.data_feed, self.poller = self.connect_result() - + def do_work(self): socks = dict(self.poller.poll(2000)) #timeout after 2 seconds. if self.data_feed in socks and socks[self.data_feed] == self.zmq.POLLIN: @@ -28,7 +28,7 @@ class TestClient(qmsg.Component): "The client should have received ({n}) the same number of messages as the feed sent ({m})." .format(n=self.received_count, m=self.expected_msg_count)) return - + self.received_count += 1 event = json.loads(msg) if(self.prev_dt != None): @@ -38,7 +38,3 @@ class TestClient(qmsg.Component): self.prev_dt = event['dt'] if(self.received_count % 100 == 0): qutil.LOGGER.info("received {n} messages".format(n=self.received_count)) - - - - \ No newline at end of file diff --git a/zipline/test/test_messaging.py b/zipline/test/test_messaging.py index 15e91396..1a42faa1 100644 --- a/zipline/test/test_messaging.py +++ b/zipline/test/test_messaging.py @@ -3,57 +3,56 @@ Test suite for the messaging infrastructure of QSim. """ #don't worry about excessive public methods pylint: disable=R0904 -import unittest2 as unittest -import multiprocessing -import time +import zipline.util as qutil +import zipline.messaging as qmsg from zipline.simulator import ThreadSimulator, ProcessSimulator from zipline.transforms.technical import MovingAverage from zipline.sources import RandomEquityTrades -import zipline.util as qutil -import zipline.messaging as qmsg from zipline.test.client import TestClient -qutil.configure_logging() +# Should not inherit form TestCase since test runners will pick +# it up as a test. -class SimulatorTestCase(unittest.TestCase): - """Tests the message passing: datasources -> feed -> transforms -> merge -> client""" +class SimulatorTestCase(object): def setUp(self): - """generate some config objects for the datafeed, sources, and transforms.""" - self.addresses = {'sync_address' : "tcp://127.0.0.1:10100", - 'data_address' : "tcp://127.0.0.1:10101", - 'feed_address' : "tcp://127.0.0.1:10102", - 'merge_address' : "tcp://127.0.0.1:10103", - 'result_address' : "tcp://127.0.0.1:10104" - } - - self.addressesblarg = "test" - - def get_simulator(self): - return ThreadSimulator(self.addresses) + qutil.configure_logging() + """ + Generate some config objects for the datafeed, sources, and transforms. + """ + + self.addresses = { + 'sync_address' : "tcp://127.0.0.1:10100", + 'data_address' : "tcp://127.0.0.1:10101", + 'feed_address' : "tcp://127.0.0.1:10102", + 'merge_address' : "tcp://127.0.0.1:10103", + 'result_address' : "tcp://127.0.0.1:10104" + } + + self.addressesblarg = "test" + + def get_simulator(self): + raise NotImplementedError + #return ThreadSimulator(self.addresses) def test_sources_only(self): - """streams events from two data sources, no transforms.""" + sim = self.get_simulator() ret1 = RandomEquityTrades(133, "ret1", 400) ret2 = RandomEquityTrades(134, "ret2", 400) client = TestClient(self, expected_msg_count=800) sim.register_components([ret1, ret2, client]) sim.simulate() - - self.assertEqual(sim.feed.pending_messages(), 0, + + 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_transforms(self): - """ - 2 datasources -> feed -> 2 moving average transforms -> transform merge -> testclient - verify message count at client. - """ sim = self.get_simulator() ret1 = RandomEquityTrades(133, "ret1", 5000) ret2 = RandomEquityTrades(134, "ret2", 5000) @@ -62,9 +61,9 @@ class SimulatorTestCase(unittest.TestCase): client = TestClient(self, expected_msg_count=10000) sim.register_components([ret1, ret2, mavg1, mavg2, client]) sim.simulate() - + self.assertEqual(sim.feed.pending_messages(), 0, "The feed should be drained of all messages.") - + def dtest_error_in_feed(self): ret1 = RandomEquityTrades(133, "ret1", 400) ret2 = RandomEquityTrades(134, "ret2", 400) @@ -76,10 +75,10 @@ class SimulatorTestCase(unittest.TestCase): sim = self.get_simulator(sources, transforms, client) sim.feed = DataFeedErr(sources.keys(), sim.data_address, sim.feed_address, sim.performance_address, qmsg.Sync(sim, "DataFeedErrorGenerator")) sim.simulate() - - -class ProcessSimulatorTestCase(SimulatorTestCase): - - def get_simulator(self): - return ProcessSimulator(self.addresses) + + +#class ProcessSimulatorTestCase(SimulatorTestCase): + + #def get_simulator(self): + #return ProcessSimulator(self.addresses) From c3b51b8de8d17176d51e2672ebffc4eeff74847f Mon Sep 17 00:00:00 2001 From: Stephen Diehl Date: Mon, 20 Feb 2012 17:15:06 -0500 Subject: [PATCH 2/8] Migrated the simulator classes to qexec. --- zipline/messaging.py | 5 ++-- zipline/simulator.py | 49 ---------------------------------- zipline/sources.py | 28 +++++++++---------- zipline/test/test_messaging.py | 31 ++++++++++++--------- 4 files changed, 34 insertions(+), 79 deletions(-) delete mode 100644 zipline/simulator.py diff --git a/zipline/messaging.py b/zipline/messaging.py index 671eeea7..23d96b9c 100644 --- a/zipline/messaging.py +++ b/zipline/messaging.py @@ -65,7 +65,7 @@ class Component(object): sock.close() except Exception as e: qutil.LOGGER.exception("Unexpected error in run for {id}.".format(id=self.get_id())) - raise e + raise finally: if(self.context != None): self.context.destroy() @@ -100,8 +100,7 @@ class Component(object): 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']) diff --git a/zipline/simulator.py b/zipline/simulator.py deleted file mode 100644 index c64c9a5f..00000000 --- a/zipline/simulator.py +++ /dev/null @@ -1,49 +0,0 @@ -""" -Provides simulated data feed services... -""" -import multiprocessing -import json -import copy -import threading - -import zipline.util as qutil -import zipline.messaging as qmsg - -class SimulatorBase(qmsg.ComponentHost): - """ - Simulator coordinates the launch and communication of source, feed, transform, and merge components. - """ - - def __init__(self, addresses, gevent_needed=False): - """ - """ - qmsg.ComponentHost.__init__(self, addresses, gevent_needed) - - def simulate(self): - self.run() - - def get_id(self): - return "Simulator" - -class ThreadSimulator(SimulatorBase): - - def __init__(self, addresses): - SimulatorBase.__init__(self, addresses) - - def launch_component(self, component): - qutil.LOGGER.info("starting {name}".format(name=component.get_id())) - thread = threading.Thread(target=component.run) - thread.start() - return thread - -class ProcessSimulator(SimulatorBase): - - def __init__(self, addresses): - SimulatorBase.__init__(self, addresses) - - def launch_component(self, component): - qutil.LOGGER.info("starting {name}".format(name=component.get_id())) - proc = multiprocessing.Process(target=component.run) - proc.start() - return proc - diff --git a/zipline/sources.py b/zipline/sources.py index 385ffe0d..4b1a2a18 100644 --- a/zipline/sources.py +++ b/zipline/sources.py @@ -2,15 +2,14 @@ Provides data handlers that can push messages to a zipline.core.DataFeed """ import datetime -import json import random import zipline.util as qutil -import zipline.messaging as qmsg - +import zipline.messaging as qmsg + class RandomEquityTrades(qmsg.DataSource): """Generates a random stream of trades for testing.""" - + def __init__(self, sid, source_id, count): qmsg.DataSource.__init__(self, source_id) self.count = count @@ -19,22 +18,21 @@ class RandomEquityTrades(qmsg.DataSource): self.trade_start = datetime.datetime.now() self.minute = datetime.timedelta(minutes=1) self.price = random.uniform(5.0, 50.0) - + def get_type(self): - return 'equity_trade' - + return 'equity_trade' + def do_work(self): if(self.incr == self.count): self.signal_done() return self.price = self.price + random.uniform(-0.05, 0.05) - event = {'sid':self.sid, - 'dt':qutil.format_date(self.trade_start + (self.minute * self.incr)), - 'price':self.price, - 'volume':random.randrange(100,10000,100)} + event = { + 'sid' : self.sid, + 'dt' : qutil.format_date(self.trade_start + (self.minute * self.incr)), + 'price' : self.price, + 'volume' : random.randrange(100,10000,100) + } + self.send(event) self.incr += 1 - - - - diff --git a/zipline/test/test_messaging.py b/zipline/test/test_messaging.py index 1a42faa1..a8344283 100644 --- a/zipline/test/test_messaging.py +++ b/zipline/test/test_messaging.py @@ -1,12 +1,17 @@ """ Test suite for the messaging infrastructure of QSim. """ -#don't worry about excessive public methods pylint: disable=R0904 +#don't worry about excessive public methods pylint: disable=R0904 + +# TODO: make sure this can run in parallel... right now this is +# forbiddeen because we hardcode the ports but it should totally +# be possible and then we can the suite much faster. +# +# nosetests --processes=5 import zipline.util as qutil import zipline.messaging as qmsg -from zipline.simulator import ThreadSimulator, ProcessSimulator from zipline.transforms.technical import MovingAverage from zipline.sources import RandomEquityTrades @@ -14,7 +19,6 @@ from zipline.test.client import TestClient # Should not inherit form TestCase since test runners will pick # it up as a test. - class SimulatorTestCase(object): def setUp(self): @@ -24,6 +28,7 @@ class SimulatorTestCase(object): Generate some config objects for the datafeed, sources, and transforms. """ + # TODO: new logic so we don't have to hardcode these self.addresses = { 'sync_address' : "tcp://127.0.0.1:10100", 'data_address' : "tcp://127.0.0.1:10101", @@ -32,11 +37,14 @@ class SimulatorTestCase(object): 'result_address' : "tcp://127.0.0.1:10104" } + # TODO: remove? self.addressesblarg = "test" def get_simulator(self): + """ + Return the simulator instance to be tested. + """ raise NotImplementedError - #return ThreadSimulator(self.addresses) def test_sources_only(self): @@ -48,8 +56,8 @@ class SimulatorTestCase(object): sim.simulate() self.assertEqual(sim.feed.pending_messages(), 0, - "The feed should be drained of all messages, found {n} remaining." - .format(n=sim.feed.pending_messages())) + "The feed should be drained of all messages, found {n} remaining." + .format(n=sim.feed.pending_messages())) def test_transforms(self): @@ -62,9 +70,11 @@ class SimulatorTestCase(object): sim.register_components([ret1, ret2, mavg1, mavg2, client]) sim.simulate() - self.assertEqual(sim.feed.pending_messages(), 0, "The feed should be drained of all messages.") + self.assertEqual(sim.feed.pending_messages(), 0, \ + "The feed should be drained of all messages.") def dtest_error_in_feed(self): + ret1 = RandomEquityTrades(133, "ret1", 400) ret2 = RandomEquityTrades(134, "ret2", 400) sources = {"ret1":ret1, "ret2":ret2} @@ -73,12 +83,9 @@ class SimulatorTestCase(object): transforms = {"mavg1":mavg1, "mavg2":mavg2} client = TestClient(self, expected_msg_count=0) sim = self.get_simulator(sources, transforms, client) + + # TODO: way too long sim.feed = DataFeedErr(sources.keys(), sim.data_address, sim.feed_address, sim.performance_address, qmsg.Sync(sim, "DataFeedErrorGenerator")) sim.simulate() -#class ProcessSimulatorTestCase(SimulatorTestCase): - - #def get_simulator(self): - #return ProcessSimulator(self.addresses) - From c0d53dc86038b12ba6f5f19c1bce26f6077aa73a Mon Sep 17 00:00:00 2001 From: Stephen Diehl Date: Mon, 20 Feb 2012 17:31:33 -0500 Subject: [PATCH 3/8] Update apidocs, flush old on rebuild. --- docs/zipline.rst | 43 +++++++++++++++++++++++++++++++++++++ docs/zipline.test.rst | 19 ++++++++++++++++ docs/zipline.transforms.rst | 19 ++++++++++++++++ pavement.py | 1 + 4 files changed, 82 insertions(+) create mode 100644 docs/zipline.rst create mode 100644 docs/zipline.test.rst create mode 100644 docs/zipline.transforms.rst diff --git a/docs/zipline.rst b/docs/zipline.rst new file mode 100644 index 00000000..cd7def87 --- /dev/null +++ b/docs/zipline.rst @@ -0,0 +1,43 @@ +zipline Package +=============== + +:mod:`zipline` Package +---------------------- + +.. automodule:: zipline.__init__ + :members: + :undoc-members: + :show-inheritance: + +:mod:`messaging` Module +----------------------- + +.. automodule:: zipline.messaging + :members: + :undoc-members: + :show-inheritance: + +:mod:`sources` Module +--------------------- + +.. automodule:: zipline.sources + :members: + :undoc-members: + :show-inheritance: + +:mod:`util` Module +------------------ + +.. automodule:: zipline.util + :members: + :undoc-members: + :show-inheritance: + +Subpackages +----------- + +.. toctree:: + + zipline.test + zipline.transforms + diff --git a/docs/zipline.test.rst b/docs/zipline.test.rst new file mode 100644 index 00000000..5521f57e --- /dev/null +++ b/docs/zipline.test.rst @@ -0,0 +1,19 @@ +test Package +============ + +:mod:`client` Module +-------------------- + +.. automodule:: zipline.test.client + :members: + :undoc-members: + :show-inheritance: + +:mod:`test_messaging` Module +---------------------------- + +.. automodule:: zipline.test.test_messaging + :members: + :undoc-members: + :show-inheritance: + diff --git a/docs/zipline.transforms.rst b/docs/zipline.transforms.rst new file mode 100644 index 00000000..aaf57f49 --- /dev/null +++ b/docs/zipline.transforms.rst @@ -0,0 +1,19 @@ +transforms Package +================== + +:mod:`transforms` Package +------------------------- + +.. automodule:: zipline.transforms + :members: + :undoc-members: + :show-inheritance: + +:mod:`technical` Module +----------------------- + +.. automodule:: zipline.transforms.technical + :members: + :undoc-members: + :show-inheritance: + diff --git a/pavement.py b/pavement.py index 0fd32fd3..5bb20e66 100644 --- a/pavement.py +++ b/pavement.py @@ -185,4 +185,5 @@ def apidocs(): Recursively autogenerate the Sphinx autodoc for the module and its submodules. """ + call('rm docs/zipline.*.rst', shell=True) call('sphinx-apidoc -o docs/ zipline', shell=True) From 351f2779549add63963d4103fbe1b058dde59d85 Mon Sep 17 00:00:00 2001 From: Stephen Diehl Date: Mon, 20 Feb 2012 17:36:14 -0500 Subject: [PATCH 4/8] Add stupid test to make Jenkins happy. --- zipline/test/test_sanity.py | 7 +++++++ 1 file changed, 7 insertions(+) create mode 100644 zipline/test/test_sanity.py diff --git a/zipline/test/test_sanity.py b/zipline/test/test_sanity.py new file mode 100644 index 00000000..416bba57 --- /dev/null +++ b/zipline/test/test_sanity.py @@ -0,0 +1,7 @@ +from unittest2 import TestCase + +class TestEnviroment(TestCase): + + def test_universe(self): + # first order logic is working today. Yay! + self.assertTrue(True != False) From c5824a168f548bc27ea1e7b6b2361e5274d85288 Mon Sep 17 00:00:00 2001 From: Stephen Diehl Date: Tue, 21 Feb 2012 07:26:35 -0500 Subject: [PATCH 5/8] Whitespace cleanup + misc pep8 + notes --- zipline/messaging.py | 180 +++++++++++++++++++++++-------------------- 1 file changed, 98 insertions(+), 82 deletions(-) diff --git a/zipline/messaging.py b/zipline/messaging.py index 23d96b9c..b6482ad0 100644 --- a/zipline/messaging.py +++ b/zipline/messaging.py @@ -7,11 +7,11 @@ import datetime import zipline.util as qutil class Component(object): - + 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. @@ -24,31 +24,37 @@ class Component(object): 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 - + + # TODO: could probably mkae this into a property instead of a + # method def get_id(self): - NotImplemented - + raise NotImplementedError + def open(self): - NotImplemented - + raise NotImplementedError + def do_work(self): - NotImplemented - + raise NotImplementedError + def run(self): + + fail = None + try: #TODO: can't initialize these values in the __init__? self.done = False self.sockets = [] + if self.gevent_needed: qutil.LOGGER.info("Loading gevent specific zmq for {id}".format(id=self.get_id())) import gevent_zeromq @@ -56,25 +62,33 @@ class Component(object): else: import zmq self.zmq = zmq + self.context = self.zmq.Context() self.open() self.setup_sync() self.loop() + #close all the sockets for sock in self.sockets: sock.close() + except Exception as e: qutil.LOGGER.exception("Unexpected error in run for {id}.".format(id=self.get_id())) - raise + fail = e + finally: + if(self.context != None): self.context.destroy() - + + if fail: + raise fail + def loop(self): while not self.done: self.confirm() self.do_work() - + def signal_done(self): #notify down stream components that we're done if(self.out_socket != None): @@ -84,15 +98,17 @@ class Component(object): self.receive_sync_ack() #notify internal work look that we're done self.done = True - + + # TODO: probably don't need a method here ... or move into + # higher level framing protocol def is_done_message(self, message): return message == "DONE" - - def confirm(self): + + def confirm(self): # send a synchronization request to the host self.sync_socket.send(self.get_id() + ":RUN") self.receive_sync_ack() - + def receive_sync_ack(self): # wait for synchronization reply from the host socks = dict(self.sync_poller.poll(2000)) #timeout after 2 seconds. @@ -103,28 +119,28 @@ class Component(object): 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) @@ -132,7 +148,7 @@ class Component(object): poller.register(pull_socket, self.zmq.POLLIN) self.sockets.append(pull_socket) return pull_socket, poller - + def connect_push_socket(self, addr): push_socket = self.context.socket(self.zmq.PUSH) push_socket.connect(addr) @@ -140,14 +156,14 @@ class Component(object): 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) @@ -156,7 +172,7 @@ class Component(object): poller.register(sub_socket, self.zmq.POLLIN) self.sockets.append(sub_socket) return sub_socket, poller - + def setup_sync(self): qutil.LOGGER.debug("Connecting sync client for {id}".format(id=self.get_id())) self.sync_socket = self.context.socket(self.zmq.REQ) @@ -165,7 +181,7 @@ class Component(object): self.sync_poller = self.zmq.Poller() self.sync_poller.register(self.sync_socket, self.zmq.POLLIN) self.sockets.append(self.sync_socket) - + class ComponentHost(Component): """Component that can launch multiple sub-components, synchronize their start, and then wait for all components to be finished.""" @@ -181,10 +197,10 @@ class ComponentHost(Component): self.merge = MergedParallelBuffer() self.passthrough = PassthroughTransform() self.gevent_needed = gevent_needed - + #register the feed and the merge self.register_components([self.feed, self.merge, self.passthrough]) - + def register_components(self, component_list): for component in component_list: component.gevent_needed = self.gevent_needed @@ -195,11 +211,11 @@ class ComponentHost(Component): 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): """Start the sync server.""" qutil.LOGGER.debug("Connecting sync server.") @@ -208,11 +224,11 @@ class ComponentHost(Component): self.poller = self.zmq.Poller() self.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) - + def is_timed_out(self): cur_time = datetime.datetime.utcnow() if(len(self.components) == 0): @@ -223,7 +239,7 @@ class ComponentHost(Component): qutil.LOGGER.info("Time out for {source}. Current component registery: {reg}".format(source=source, reg=self.components)) return True return False - + def loop(self): while not self.is_timed_out(): # wait for synchronization request @@ -239,20 +255,20 @@ class ComponentHost(Component): if(self.is_done_message(status)): qutil.LOGGER.info("{id} is DONE".format(id=sync_id)) self.unregister_component(sync_id) - else: + 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) - + def launch_component(self, component): NotImplemented - + class ParallelBuffer(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 @@ -260,19 +276,19 @@ class ParallelBuffer(Component): self.draining = False self.data_buffer = {} self.ds_finished_counter = 0 - - + + def get_id(self): return "FEED" - + def add_source(self, source_id): self.data_buffer[source_id] = [] - + def open(self): self.pull_socket, self.poller = self.bind_data() - self.feed_socket = self.bind_feed() + self.feed_socket = self.bind_feed() - def do_work(self): + def do_work(self): # wait for synchronization reply from the host socks = dict(self.poller.poll(2000)) #timeout after 2 seconds. @@ -291,18 +307,18 @@ class ParallelBuffer(Component): 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) - + return len(self.data_buffer) + def append(self, source_id, value): """add an event to the buffer for the source specified by source_id""" self.data_buffer[source_id].append(value) self.received_count += 1 - + def next(self): """Get the next message in chronological order""" if(not(self.is_full() or self.draining)): return - + cur = None earliest = None for events in self.data_buffer.values(): @@ -311,58 +327,58 @@ class ParallelBuffer(Component): cur = events if(earliest == None) or (cur[0]['dt'] <= earliest[0]['dt']): earliest = cur - + if(earliest != None): return earliest.pop(0) - + def is_full(self): """indicates whether the buffer has messages in buffer for all un-DONE sources""" for events in self.data_buffer.values(): 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 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 - - + self.sent_count += 1 + + class MergedParallelBuffer(ParallelBuffer): """ Merges multiple streams of events into single messages. """ - + def __init__(self): ParallelBuffer.__init__(self) - + def open(self): self.pull_socket, self.poller = 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 - + #get the raw event from the passthrough transform. result = self.data_buffer["PASSTHROUGH"].pop(0)['value'] for source, events in self.data_buffer.iteritems(): @@ -372,10 +388,10 @@ class MergedParallelBuffer(ParallelBuffer): cur = events.pop(0) result[source] = cur['value'] return result - + def get_id(self): return "MERGE" - + class BaseTransform(Component): """Top level execution entry point for the transform:: @@ -385,8 +401,8 @@ class BaseTransform(Component): - 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 + + Parent class for feed transforms. Subclass and override transform method to create a new derived value from the combined feed.""" def __init__(self, name): @@ -397,19 +413,19 @@ class BaseTransform(Component): def get_id(self): return self.state['name'] - def open(self): + def open(self): """ Establishes zmq connections. - """ - #create the feed. + """ + #create the feed. self.feed_socket, self.poller = 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 + - receive an event from the data feed - call transform (subclass' method) on event - send the transformed event """ @@ -433,18 +449,18 @@ class BaseTransform(Component): self.state['value'] = transformed_value """ NotImplemented - + class PassthroughTransform(BaseTransform): - + def __init__(self): BaseTransform.__init__(self, "PASSTHROUGH") - def transform(self, event): + def transform(self, event): return {'value':event} - + class DataSource(Component): """ - Baseclass for data sources. Subclass and implement send_all - usually this + 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). """ @@ -456,8 +472,8 @@ class DataSource(Component): def get_id(self): return self.id - def open(self): - #create the data sink. Based on http://zguide.zeromq.org/py:tasksink2 + def open(self): + #create the data sink. Based on http://zguide.zeromq.org/py:tasksink2 self.data_socket = self.connect_data() def send(self, event): @@ -466,7 +482,7 @@ class DataSource(Component): sets id and type properties in the dict sends to the data_socket. """ - event['id'] = self.id + event['id'] = self.id event['type'] = self.get_type() self.data_socket.send(json.dumps(event)) From d39d7683a3802707fe1a9c2b06ae6b1548b7a091 Mon Sep 17 00:00:00 2001 From: Stephen Diehl Date: Tue, 21 Feb 2012 12:18:25 -0500 Subject: [PATCH 6/8] Initial zipline work on control sockets. --- zipline/messaging.py | 26 ++++++++++++++++++++++++++ 1 file changed, 26 insertions(+) diff --git a/zipline/messaging.py b/zipline/messaging.py index b6482ad0..aec80d59 100644 --- a/zipline/messaging.py +++ b/zipline/messaging.py @@ -14,6 +14,8 @@ class Component(object): - 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. + - control_address: socket address used for controlling and + monitoring the status of the simulation - 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) @@ -34,6 +36,7 @@ class Component(object): self.addresses = None self.out_socket = None self.gevent_needed = False + self.killed = False # TODO: could probably mkae this into a property instead of a # method @@ -43,6 +46,19 @@ class Component(object): def open(self): raise NotImplementedError + def destroy(self): + """ + Tear down after normal operation. + """ + raise NotImplementedError + + def kill(self): + """ + Tear down ( fast ) as a mode of failure in the + simulation. + """ + raise NotImplementedError + def do_work(self): raise NotImplementedError @@ -66,6 +82,7 @@ class Component(object): self.context = self.zmq.Context() self.open() self.setup_sync() + self.setup_control() self.loop() #close all the sockets @@ -173,6 +190,15 @@ class Component(object): self.sockets.append(sub_socket) return sub_socket, poller + def setup_control(self): + """ + Set up the control socket. Used to monitor the the + overall status of the simulation and to forcefully tear + down the simulation in case of a failure. + """ + qutil.LOGGER.debug("Connecting control socket for {id}".format(id=self.get_id())) + #self.control_socket = self.context.socket(self.zmq.REQ) + def setup_sync(self): qutil.LOGGER.debug("Connecting sync client for {id}".format(id=self.get_id())) self.sync_socket = self.context.socket(self.zmq.REQ) From c8553f27d9085904bf7474a0da76bc807fb14747 Mon Sep 17 00:00:00 2001 From: Stephen Diehl Date: Wed, 22 Feb 2012 12:48:37 -0500 Subject: [PATCH 7/8] Tweak tests to use allocator. --- zipline/test/test_messaging.py | 20 -------------------- 1 file changed, 20 deletions(-) diff --git a/zipline/test/test_messaging.py b/zipline/test/test_messaging.py index a8344283..f8c3a68d 100644 --- a/zipline/test/test_messaging.py +++ b/zipline/test/test_messaging.py @@ -9,7 +9,6 @@ Test suite for the messaging infrastructure of QSim. # # nosetests --processes=5 -import zipline.util as qutil import zipline.messaging as qmsg from zipline.transforms.technical import MovingAverage @@ -21,25 +20,6 @@ from zipline.test.client import TestClient # it up as a test. class SimulatorTestCase(object): - def setUp(self): - qutil.configure_logging() - - """ - Generate some config objects for the datafeed, sources, and transforms. - """ - - # TODO: new logic so we don't have to hardcode these - self.addresses = { - 'sync_address' : "tcp://127.0.0.1:10100", - 'data_address' : "tcp://127.0.0.1:10101", - 'feed_address' : "tcp://127.0.0.1:10102", - 'merge_address' : "tcp://127.0.0.1:10103", - 'result_address' : "tcp://127.0.0.1:10104" - } - - # TODO: remove? - self.addressesblarg = "test" - def get_simulator(self): """ Return the simulator instance to be tested. From 6e1cc72900048e53e93a69f3a315137a817b372b Mon Sep 17 00:00:00 2001 From: Stephen Diehl Date: Wed, 22 Feb 2012 16:02:49 -0500 Subject: [PATCH 8/8] Controller is done. --- zipline/messaging.py | 158 +++++++++++++++++++++++++++++++++++++++++-- 1 file changed, 152 insertions(+), 6 deletions(-) diff --git a/zipline/messaging.py b/zipline/messaging.py index aec80d59..203d31b2 100644 --- a/zipline/messaging.py +++ b/zipline/messaging.py @@ -5,6 +5,133 @@ import json import uuid import datetime import zipline.util as qutil +import zmq + +class Controller(object): + """ + A broker of sorts. + """ + + polling = False + debug = False + + def __init__(self, pull_socket, pub_socket, context=None, logging = None): + + + if not context: + self._ctx = zmq.Context() + else: + self._ctx = context + + self.pull_socket = pull_socket + self.pub_socket = pub_socket + + self.pull = self._ctx.socket(zmq.PULL) + self.pub = self._ctx.socket(zmq.PUB) + + self.associated = [self.pull, self.pub] + + if logging: + self.logging = logging + self.dologging = True + else: + self.logging = False + self.dologging = False + + self.success = 0 + self.failed = 0 + + try: + self.pull.bind(pull_socket) + except zmq.ZMQError: + raise Exception('Cannot not bind on %s' % pull_socket) + + try: + self.pub.bind(pub_socket) + except zmq.ZMQError: + raise Exception('Cannot not bind on %s' % pub_socket) + + def run(self, debug_step=False, stats=True): + self.polling = True + + if self.debug or debug_step: + return self._poll_verbose(True, stats) + else: + return self._poll(False, stats) + + def _poll(self, debug_step, stats): + while self.polling: + try: + self.logging.info('msg') + self.pub.send(self.pull.recv()) + except KeyboardInterrupt: + self.polling = False + break + except Exception as e: + # Its common to wrap these in wildcard exceptions so + # that we don't loose messages, ever + self.logging.error(str(e)) + self.failed += 1 + continue + + def _poll_verbose(self, debug_step, stats): + while self.polling: + try: + if debug_step: + msg = self.pull.recv() + if self.dologging: + self.logging.info(msg) + self.pub.send(msg) + self.success += 1 + except KeyboardInterrupt: + self.polling = False + break + except Exception as e: + # Its common to wrap these in wildcard exceptions so + # that we don't loose messages, ever + self.logging.error(str(e)) + self.failed += 1 + continue + + def qos(self): + return float(self.success) / (self.success + self.failed) + + def destroy(self): + """ + Manual cleanup. + """ + self.polling = False + + for asoc in self.associated: + asoc.close() + + #if self._ctx: + #self._ctx.destroy() + + def __del__(self): + self.destroy() + + def message_sender(self): + """ + Spin off a socket used for sending messages to this + controller. + """ + s = self._ctx.socket(zmq.PUSH) + s.connect(self.pull_socket) + s.setsockopt(zmq.LINGER, -1) + self.associated.append(s) + return s + + def message_listener(self): + """ + Spin off a socket used for receiving messages from this + controller. + """ + s = self._ctx.socket(zmq.SUB) + s.connect(self.pub_socket) + s.setsockopt(zmq.SUBSCRIBE, '') + self.associated.append(s) + return s class Component(object): @@ -196,8 +323,7 @@ class Component(object): overall status of the simulation and to forcefully tear down the simulation in case of a failure. """ - qutil.LOGGER.debug("Connecting control socket for {id}".format(id=self.get_id())) - #self.control_socket = self.context.socket(self.zmq.REQ) + pass def setup_sync(self): qutil.LOGGER.debug("Connecting sync client for {id}".format(id=self.get_id())) @@ -209,11 +335,15 @@ class Component(object): self.sockets.append(self.sync_socket) class ComponentHost(Component): - """Component that can launch multiple sub-components, synchronize their start, and then wait for all - components to be finished.""" + """ + Component that can launch multiple sub-components, synchronize their start, and then wait for all + components to be finished. + """ + def __init__(self, addresses, gevent_needed=False): Component.__init__(self) self.addresses = addresses + #workaround for defect in threaded use of strptime: http://bugs.python.org/issue11108 qutil.parse_date("2012/02/13-10:04:28.114") self.components = {} @@ -223,16 +353,28 @@ class ComponentHost(Component): self.merge = MergedParallelBuffer() self.passthrough = PassthroughTransform() self.gevent_needed = gevent_needed + self.controller = None #register the feed and the merge self.register_components([self.feed, self.merge, self.passthrough]) - def register_components(self, component_list): - for component in component_list: + def register_controller(self, controller): + self.controller = controller + + 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 + + if self.controller: + component.controller = self.controller + 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)): @@ -254,6 +396,7 @@ class ComponentHost(Component): def open(self): for component in self.components.values(): self.launch_component(component) + self.launch_controller() def is_timed_out(self): cur_time = datetime.datetime.utcnow() @@ -287,6 +430,9 @@ class ComponentHost(Component): # send synchronization reply self.sync_socket.send('ack', self.zmq.NOBLOCK) + def launch_controller(self, controller): + NotImplemented + def launch_component(self, component): NotImplemented