From 755ef6f72d03de0082a18f014a9fb30b9e9fd60c Mon Sep 17 00:00:00 2001 From: fawce Date: Thu, 16 Feb 2012 15:41:51 -0500 Subject: [PATCH] reverting changes to hand off to realDiehl for debugging lockups on recv --- zipline/messaging.py | 58 ++++++++++++++++++++-------------- zipline/test/test_messaging.py | 4 +-- 2 files changed, 36 insertions(+), 26 deletions(-) diff --git a/zipline/messaging.py b/zipline/messaging.py index 8c5076d1..a2beb78a 100644 --- a/zipline/messaging.py +++ b/zipline/messaging.py @@ -4,9 +4,8 @@ Commonly used messaging components. import json import uuid import datetime -#from gevent_zeromq import zmq -import zmq import zipline.util as qutil +from gevent_zeromq import zmq as gzmq class Component(object): @@ -47,18 +46,16 @@ class Component(object): def run(self): try: - #TODO: can't initialize a boolean in the __init__? + #TODO: can't initialize these values in the __init__? 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) + 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.context = self.zmq.Context() self.open() self.setup_sync() @@ -66,6 +63,8 @@ class Component(object): #close all the sockets for sock in self.sockets: sock.close() + except: + qutil.LOGGER.exception("Unexpected error in run for {id}.".format(id=self.get_id())) finally: self.context.destroy() @@ -89,15 +88,22 @@ class Component(object): def confirm(self): # send a synchronization request to the host + qutil.LOGGER.debug("sending confirmation...") self.sync_socket.send(self.get_id() + ":RUN") self.receive_sync_ack() def receive_sync_ack(self): # wait for synchronization reply from the host + qutil.LOGGER.debug("polling sync socket for response") socks = dict(self.sync_poller.poll(2000)) #timeout after 2 seconds. - + qutil.LOGGER.debug("done polling") if self.sync_socket in socks and socks[self.sync_socket] == self.zmq.POLLIN: + qutil.LOGGER.debug("attempting to receive...") message = self.sync_socket.recv() + qutil.LOGGER.debug("confirm ack'd") + 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']) @@ -170,18 +176,20 @@ class ComponentHost(Component): def __init__(self, addresses, gevent_needed=False): Component.__init__(self, addresses) if gevent_needed: - module = __import__('gevent_zmq', 'zmq') + qutil.LOGGER.debug("importing the gevent friendly zmq") + module = __import__('zmq','gevent_zeromq') else: module = __import__('zmq') - self.zmq = module + self.zmq = gzmq + qutil.LOGGER.debug("zmq file: {file}".format(file=self.zmq.__file__)) #workaround for defect in threaded use of strptime: http://bugs.python.org/issue11108 qutil.parse_date("2012/02/13-10:04:28.114") self.components = {} self.sync_register = {} self.timeout = datetime.timedelta(seconds=5) - self.feed = ParallelBuffer() - self.merge = MergedParallelBuffer() - self.passthrough = PassthroughTransform() + self.feed = ParallelBuffer(addresses) + self.merge = MergedParallelBuffer(addresses) + self.passthrough = PassthroughTransform(addresses) #register the feed and the merge self.register_components([self.feed, self.merge, self.passthrough]) @@ -253,7 +261,8 @@ class ParallelBuffer(Component): 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): + def __init__(self, addresses): + Component.__init__(self, addresses) self.sent_count = 0 self.received_count = 0 self.draining = False @@ -269,7 +278,7 @@ class ParallelBuffer(Component): 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): # wait for synchronization reply from the host @@ -350,8 +359,8 @@ class MergedParallelBuffer(ParallelBuffer): Merges multiple streams of events into single messages. """ - def __init__(self): - ParallelBuffer.__init__(self) + def __init__(self, addresses): + ParallelBuffer.__init__(self, addresses) def open(self): self.pull_socket, self.poller = self.bind_merge() @@ -388,7 +397,8 @@ class BaseTransform(Component): Parent class for feed transforms. Subclass and override transform method to create a new derived value from the combined feed.""" - def __init__(self, name): + def __init__(self, name, addresses): + Component.__init__(self, addresses) self.state = {} self.state['name'] = name @@ -434,8 +444,8 @@ class BaseTransform(Component): class PassthroughTransform(BaseTransform): - def __init__(self): - BaseTransform.__init__(self, "PASSTHROUGH") + def __init__(self, addresses): + BaseTransform.__init__(self, "PASSTHROUGH", addresses) def transform(self, event): return {'value':event} diff --git a/zipline/test/test_messaging.py b/zipline/test/test_messaging.py index 9e1d5747..5b1c47ce 100644 --- a/zipline/test/test_messaging.py +++ b/zipline/test/test_messaging.py @@ -17,7 +17,7 @@ from zipline.test.client import TestClient qutil.configure_logging() -class ThreadSimulatorTestCase(unittest.TestCase): +class SimulatorTestCase(unittest.TestCase): """Tests the message passing: datasources -> feed -> transforms -> merge -> client""" def setUp(self): @@ -77,7 +77,7 @@ class ThreadSimulatorTestCase(unittest.TestCase): sim.simulate() -class ProcessSimulatorTestCase(ThreadSimulatorTestCase): +class ProcessSimulatorTestCase(SimulatorTestCase): def get_simulator(self): return ProcessSimulator(self.addresses)