wassname/catalyst
1
0
Fork 0
mirror of https://github.com/wassname/catalyst.git synced 2026-07-17 11:25:55 +08:00
Code Issues Packages Projects Releases Wiki Activity

eliminated the redundant passing of address information, moved the zmq import to the run method to see if loading in the execution context helps

Browse Source
This commit is contained in:
fawce
2012-02-17 10:57:53 -05:00
parent 4216dee08d
commit 8ad08ee384
4 changed files with 42 additions and 40 deletions
Download Patch File Download Diff File Expand all files Collapse all files
zipline/messaging.py
+24 -23
View File
@@ -10,7 +10,7 @@ import zipline.util as qutil
class Component(object):
def __init__(self, addresses):
def __init__(self):
"""
:addresses: a dict of name_string -> zmq port address strings. Must have the following entries::
@@ -31,11 +31,12 @@ class Component(object):
will also return a Poller.
"""
self.zmq = None
self.context = None
self.addresses = addresses
self.out_socket = None
self.zmq = None
self.context = None
self.addresses = None
self.out_socket = None
self.gevent_needed = False
def get_id(self):
NotImplemented
@@ -51,14 +52,15 @@ class Component(object):
self.done = False
#TODO: figure out why sockets can't be set in the __init__
self.sockets = []
#TODO: figure out why addresses can't be set in the __init__
self.addresses = {'sync_address' : "tcp://127.0.0.1:{port}".format(port=10100),
'data_address' : "tcp://127.0.0.1:{port}".format(port=10101),
'feed_address' : "tcp://127.0.0.1:{port}".format(port=10102),
'merge_address' : "tcp://127.0.0.1:{port}".format(port=10103),
'result_address' : "tcp://127.0.0.1:{port}".format(port=10104)
}
if self.gevent_needed:
qutil.LOGGER.info("Loading gevent specific zmq for {id}".format(id=self.get_id()))
module = __import__('gevent_zmq', 'zmq')
else:
qutil.LOGGER.debug("NOT Loading gevent specific zmq for {id}".format(id=self.get_id()))
module = __import__('zmq')
self.zmq = module
self.context = self.zmq.Context()
self.open()
self.setup_sync()
@@ -66,6 +68,8 @@ class Component(object):
#close all the sockets
for sock in self.sockets:
sock.close()
except:
qutil.LOGGER.exception("Problem running component {id}.".format(id=self.get_id()))
finally:
self.context.destroy()
@@ -168,12 +172,8 @@ class ComponentHost(Component):
"""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, addresses)
if gevent_needed:
module = __import__('gevent_zmq', 'zmq')
else:
module = __import__('zmq')
self.zmq = module
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 = {}
@@ -188,7 +188,8 @@ class ComponentHost(Component):
def register_components(self, component_list):
for component in component_list:
component.zmq = self.zmq
component.gevent_needed = self.gevent_needed
component.addresses = self.addresses
self.components[component.get_id()] = component
self.sync_register[component.get_id()] = datetime.datetime.utcnow()
if(isinstance(component, DataSource)):
@@ -446,8 +447,8 @@ class DataSource(Component):
means looping through all records in a store, converting to a dict, and
calling send(map).
"""
def __init__(self, source_id, addresses):
Component.__init__(self, addresses)
def __init__(self, source_id):
Component.__init__(self)
self.id = source_id
self.cur_event = None
zipline/sources.py
+2 -2
View File
@@ -11,8 +11,8 @@ import zipline.messaging as qmsg
class RandomEquityTrades(qmsg.DataSource):
"""Generates a random stream of trades for testing."""
def __init__(self, sid, source_id, count, addresses):
qmsg.DataSource.__init__(self, source_id, addresses)
def __init__(self, sid, source_id, count):
qmsg.DataSource.__init__(self, source_id)
self.count = count
self.incr = 0
self.sid = sid
zipline/test/client.py
+2 -2
View File
@@ -4,8 +4,8 @@ import zipline.messaging as qmsg
class TestClient(qmsg.Component):
def __init__(self, addresses, utest, expected_msg_count=0):
qmsg.Component.__init__(self, addresses)
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
zipline/test/test_messaging.py
+14 -13
View File
@@ -22,13 +22,14 @@ class ThreadSimulatorTestCase(unittest.TestCase):
def setUp(self):
"""generate some config objects for the datafeed, sources, and transforms."""
self.addresses = {'sync_address' : "tcp://127.0.0.1:{port}".format(port=10100),
'data_address' : "tcp://127.0.0.1:{port}".format(port=10101),
'feed_address' : "tcp://127.0.0.1:{port}".format(port=10102),
'merge_address' : "tcp://127.0.0.1:{port}".format(port=10103),
'result_address' : "tcp://127.0.0.1:{port}".format(port=10104)
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)
@@ -37,9 +38,9 @@ class ThreadSimulatorTestCase(unittest.TestCase):
def test_sources_only(self):
"""streams events from two data sources, no transforms."""
sim = self.get_simulator()
ret1 = RandomEquityTrades(133, "ret1", 400, self.addresses)
ret2 = RandomEquityTrades(134, "ret2", 400, self.addresses)
client = TestClient(self.addresses, self, expected_msg_count=800)
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()
@@ -54,11 +55,11 @@ class ThreadSimulatorTestCase(unittest.TestCase):
verify message count at client.
"""
sim = self.get_simulator()
ret1 = RandomEquityTrades(133, "ret1", 5000, self.addresses)
ret2 = RandomEquityTrades(134, "ret2", 5000, self.addresses)
ret1 = RandomEquityTrades(133, "ret1", 5000)
ret2 = RandomEquityTrades(134, "ret2", 5000)
mavg1 = MovingAverage("mavg1", 30)
mavg2 = MovingAverage("mavg2", 60)
client = TestClient(self.addresses, self, expected_msg_count=10000)
client = TestClient(self, expected_msg_count=10000)
sim.register_components([ret1, ret2, mavg1, mavg2, client])
sim.simulate()
@@ -71,7 +72,7 @@ class ThreadSimulatorTestCase(unittest.TestCase):
mavg1 = MovingAverage("mavg1", 30)
mavg2 = MovingAverage("mavg2", 60)
transforms = {"mavg1":mavg1, "mavg2":mavg2}
client = TestClient(self.addresses, self, expected_msg_count=0)
client = TestClient(self, expected_msg_count=0)
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()