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

reverting changes to hand off to realDiehl for debugging lockups on recv

Browse Source
This commit is contained in:
fawce
2012-02-16 15:41:51 -05:00
parent 4216dee08d
commit 755ef6f72d
2 changed files with 36 additions and 26 deletions
Download Patch File Download Diff File Expand all files Collapse all files
zipline/messaging.py
+34 -24
View File
@@ -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}
zipline/test/test_messaging.py
+2 -2
View File
@@ -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)