From e8d98084df9f7635e919b79f4c514c7877c7b0f9 Mon Sep 17 00:00:00 2001 From: Stephen Diehl Date: Tue, 3 Apr 2012 14:04:11 -0400 Subject: [PATCH 1/7] Exception tracking at the controller level. --- zipline/component.py | 41 ++-- zipline/lines.py | 5 + zipline/messaging.py | 2 +- zipline/monitor.py | 418 ++++++++++++++++++++++++++++------- zipline/protocol.py | 17 +- zipline/simulator.py | 5 +- zipline/test/test_monitor.py | 78 ------- 7 files changed, 385 insertions(+), 181 deletions(-) diff --git a/zipline/component.py b/zipline/component.py index cfdc8781..9acf2cc2 100644 --- a/zipline/component.py +++ b/zipline/component.py @@ -201,28 +201,25 @@ class Component(object): debug since it mucks up your stacktraces. """ - fail = None - if catch_exceptions: try: self._run() except Exception as exc: - # TODO, we want to do this grab the stack - # frame so we can preserve stacktraces when we - # reraise the exception. + exc_info = sys.exc_info() self.signal_exception(exc) - fail = exc + + # Reraise the exception + raise exc_info[0], exc_info[1], exc_info[2] finally: self.shutdown() self.teardown_sockets() - else: - try: - self._run() - finally: - self.shutdown() - self.teardown_sockets() - + #else: + #try: + #self._run() + #finally: + #self.shutdown() + #self.teardown_sockets() def working(self): """ @@ -234,7 +231,7 @@ class Component(object): """ return (not self.done) - def loop(self): + def loop(self, lockstep=True): """ Loop to do work while we still have work to do. """ @@ -294,6 +291,12 @@ class Component(object): # ---------------------- def signal_exception(self, exc=None, scope=None): + """ + This is *very* important error tracking handler. + + Will inform the system that the component has failed and + how it has failed. + """ if scope == 'algo': self.error_state = COMPONENT_FAILURE.ALGOEXCEPT @@ -427,8 +430,14 @@ class Component(object): if not self.controller: return - self.control_out = self.controller.message_sender(context=self.context) - self.control_in = self.controller.message_listener(context=self.context) + self.control_out = self.controller.message_sender( + identity = self.get_id, + 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]) diff --git a/zipline/lines.py b/zipline/lines.py index 2bc32aa2..2123cfdf 100644 --- a/zipline/lines.py +++ b/zipline/lines.py @@ -149,6 +149,11 @@ class SimulatedTrading(object): sockets[7], logging = qutil.LOGGER ) + + # TODO: Not freeform + self.con.manage( + 'freeform' + ) self.started = False diff --git a/zipline/messaging.py b/zipline/messaging.py index 5e6ff570..552c1115 100644 --- a/zipline/messaging.py +++ b/zipline/messaging.py @@ -128,7 +128,7 @@ class ComponentHost(Component): return False - def loop(self): + def loop(self, lockstep=True): while not self.is_timed_out(): # wait for synchronization request diff --git a/zipline/monitor.py b/zipline/monitor.py index 5db2ea54..2e0f2066 100644 --- a/zipline/monitor.py +++ b/zipline/monitor.py @@ -1,29 +1,131 @@ -import zmq -from zipline.protocol import CONTROL_PROTOCOL, CONTROL_FRAME +import time +import gevent + +from gevent_zeromq import zmq +import util as qutil + +from protocol import CONTROL_PROTOCOL, CONTROL_FRAME, \ + CONTROL_UNFRAME, CONTROL_STATES + +# TODO: print -> qutil.LOGGER.info + +# When you're cold and waiting for the train... draw ascii art! +# +# Roll Call ( Discovery ) +# ----------------------- +# +# Controller ( 'foo', 'bar', 'fizz', 'pop' ) +# ------------------ +# | | | | +# +---+ +# | 0 | ? ? ? +# +---+ +# | +# IDENTITY: foo +# get message: PROTOCOL.HEARTBEAT +# reply with PROTOCOL.OK +# +# Controller topology = ( 'foo', 'bar', 'fizz', 'pop' ) +# 'foo' in topology = YES -> +# track 'foo' +# ------------------ +# | | | | +# +---+ +# | 1 | ? ? ? +# +---+ + +# Heartbeating +# ------------ +# +# Controller ( time = 2.717828 ) +# ------------------ +# | | | | +# +---+ +---+ +---+ +---+ +# | 0 | | 0 | | 0 | | 0 | +# +---+ +---+ +---+ +---+ +# | +# IDENTITY: foo +# get message: time = 2.717828 +# reply with [ foo, 2.71828 ] +# +# Controller ( foo.status = OK ) +# ------------------ +# | | | | +# +---+ +---+ +---+ +---+ +# | 1 | | 0 | | 0 | | 0 | +# +---+ +---+ +---+ +---+ +# | +# Controller tracks this node as good +# for this heartbeat + +# Shutdown +# -------- +# +# Controller ( state = RUNNING ) +# ------------------ +# | | | | +# +---+ +---+ +---+ +---+ +# | 1 | | 1 | | 1 | | 1 | +# +---+ +---+ +---+ +---+ +# | +# IDENTITY: foo +# send [ DONE ] + +# Controller ( state = SHUTDOWN ) +# Controller topology.remove('foo') +# ------------------ +# | | | +# +---+ +---+ +---+ +---+ +# | | | 1 | | 1 | | 1 | +# +---+ +---+ +---+ +---+ +# | +# IDENTITY: foo +# yield, stop sending messages + +# Termination +# ------------ +# +# Controller ( state = TERMINATE ) +# ------------------ +# | | | | +# +---+ +---+ +---+ +---+ +# | 1 | | 1 | | 1 | | 1 | +# +---+ +---+ +---+ +---+ +# | +# get message PROTOCOL.KILL + +# Controller ( state = TERMINATE ) +# ------------------ +# | | | | +# +---+ +---+ +---+ +---+ +# | 0 | | 0 | | 0 | | 0 | +# +---+ +---+ +---+ +---+ + +class UnknownChatter(Exception): + def __init__(self, name): + self.named = name + def __str__(self): + return """Component calling itself "%s" talking on unexpected channel"""\ + % self.named class Controller(object): """ 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 route_socket: Socket to listen for status updates for + the individual components. + :func message_sender: . + :param logging: Logging interface for tracking broker state Defaults to None + Topology is the set of components we expect to show up. + States are the transitions the sytems go through. The + simplest is from RUNNING -> NOT RUNNING . + Usage:: controller = Controller( @@ -33,31 +135,34 @@ class Controller(object): # 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() + controller.manage( + [ TOPOLOGY ] + [ STATES ] + ) """ debug = False + period = 1 - def __init__(self, pull_socket, pub_socket, logging = None): + def __init__(self, pub_socket, route_socket, logging = None): self._ctx = None polling = False + self.polling = polling + + self.tracked = set() + self.responses = set() + + self.ctime = 0 + self.tic = time.time() + self.freeform = False + self.associated = [] - self.pull_socket = pull_socket - self.push_socket = pull_socket # same port self.pub_socket = pub_socket - self.sub_socket = pub_socket # same port - self.terminated = False + self.route_socket = route_socket if logging: self.logging = logging @@ -66,52 +171,67 @@ class Controller(object): self.logging = False self.dologging = False - self.success = 0 - self.failed = 0 + def manage(self, topology, states=None, context=None): + """ + Give the controller a set set of components to manage and + a set of state transitions for the entire system. + """ - def run(self, debug=False, context=None): - """ - Run's the loop for the broker. - """ + # A freeform topology is where we heartbeat with anything + # that shows up. + if topology == 'freeform': + self.freeform = True + self.topology = frozenset([]) + else: + self.freeform = False + self.topology = frozenset(topology) + + default_states = [ + CONTROL_STATES.RUNNING, + CONTROL_STATES.SHUTDOWN, + CONTROL_STATES.TERMINATE, + ] + + self.states = states or default_states self.polling = True + # Start off in RUNNING, state + self.state = self.states[0] + if not context: self._ctx = zmq.Context.instance() else: self._ctx = context - #if not debug: - #return self._poll_fast() # the c loop - #else: + def run(self, flavor): + """ + Abstracts + """ + assert flavor in ['thread', 'mp', 'green'] return self._poll() # use a python loop - def _poll_fast(self): - """ - C version of the polling forwarder. - """ - 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 log_status(self): + print "[Controller] Tracking : %s" % ([c for c in self.tracked],) + gevent.spawn_later(5, self.log_status) def _poll(self): - """ - Python version of the polling forwarder. With logging, - mostly used for debugging. - """ - 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) - self.associated.extend([self.pull, self.pub]) + self.router = self._ctx.socket(zmq.ROUTER) + self.router.bind(self.route_socket) + + self.associated.extend([self.pub, self.router]) + + # Spin the coroutines + gevent.spawn(self.log_status) + gevent.spawn_raw(self.recv_hearts_router) while self.polling: + self.beat() + gevent.sleep(self.period) + try: msg = self.pull.recv() self.pub.send(msg) @@ -126,16 +246,146 @@ class Controller(object): # that we don't loose messages, ever if self.logging: self.logging.error(str(e)) - self.failed += 1 continue + # After loop exits self.terminated = True + def beat(self): + toc = time.time() + + self.ctime += toc - self.tic + self.tic = toc + self.message = str(self.ctime) + + # These the set overloaded operations + # A & B ~ set.intersection + # A - B ~ set.difference + + # * good - Components we are currently tracking and who just sent + # us back the right response. + # * bad - Components we are currently tracking but who did not + # send us back a response. + # * new - Components we haven't heard from yet, but sent back the + # right response. + + good = self.tracked & self.responses + bad = self.tracked - good + new = self.responses - good + + for component in new: + self.new(component) + + for component in bad: + self.fail(component) + + # Reset the responses for this cycle + self.responses = set() + + gevent.spawn_raw(self.pub.send, str(self.ctime)) + + # ------------------ + # Component Handlers + # ------------------ + + # The various "states of being that a component can inform us + # of + def new(self, component): + print '[Controller] Tracking "%s" ' % component + + if component in self.topology or self.freeform: + self.tracked.add(component) + else: + # Some sort of socket collision has occured, this is + # a very bad failure mode. + raise UnknownChatter(component) + + def fail(self, component): + print '[Controller] Component "%s" timed out' % component + self.tracked.remove(component) + + def done(self, component): + # TODO: This will be what we ship off to vbench at some + # point... + # print component finished at self.ctime + pass + + def exception(self, component, failure): + pass + + # -------- + # IO Loops + # -------- + + def recv_hearts(self): + buffer = [] + + while True: + message = self.router.recv(zmq.NOBLOCK, copy=False) + + if message: + buffer.append(message) + if not self.router.rcvmore(): + break + + gevent.spawn_later(0.001, self.handle_pong, buffer) + + def recv_hearts_router(self): + buffer = [] + + while True: + message = self.router.recv() + + if message: + buffer.append(message) + + if not self.router.getsockopt(zmq.RCVMORE): + gevent.spawn_raw(self.handle_recv, buffer[:]) + buffer = [] + + # ----------------- + # Protocol Handling + # ----------------- + + def handle_recv(self, msg): + """ + Check for proper framing at the transport layer. + Seperates the proper frames from anything else that might + be coming over the wire. Which shouldn't happen ... right? + """ + identity = msg[0] + + id, status = CONTROL_UNFRAME(msg[1]) + + # A component is telling us its alive: + if id is CONTROL_PROTOCOL.OK: + + if status == str(self.ctime): + self.responses.add(identity) + else: + # Otherwise its something weird and we don't know + # what to do so just say so + print "Weird stuff happened: %s" % msg[1] + + # A component is telling us it failed, and how + if id is CONTROL_PROTOCOL.EXCEPTION: + self.exception(identity, status) + + # A component is telling us its done with work and won't + # be talking to us anymore + if id is CONTROL_PROTOCOL.DONE: + self.done(identity, status) + # ------------------- # Hooks for Endpoints # ------------------- - def message_sender(self, context=None): + # These are all connects so no complex allocation logic is + # needed. Dealers and Subscribers can all come and go as a + # function of time without impacting flow of the whole + # system. + + def message_sender(self, identity, context = None): """ Spin off a socket used for sending messages to this controller. @@ -144,12 +394,14 @@ class Controller(object): if not context: context = zmq.Context.instance() - s = context.socket(zmq.PUSH) - s.connect(self.push_socket) + s = context.socket(zmq.DEALER) + s.connect(self.route_socket) + s.setsockopt(zmq.IDENTITY, identity) + self.associated.append(s) return s - def message_listener(self, context = None, filters=None): + def message_listener(self, context = None): """ Spin off a socket used for receiving messages from this controller. @@ -159,27 +411,35 @@ class Controller(object): context = zmq.Context.instance() s = context.socket(zmq.SUB) - s.connect(self.sub_socket) - s.setsockopt(zmq.SUBSCRIBE, filters or '') + s.connect(self.pub_socket) + s.setsockopt(zmq.SUBSCRIBE, '') + self.associated.append(s) return s - def shutdown(self, context=None): + def shutdown(self, hard=False, soft=True, context=None): self.polling = False + assert hard or soft, """ Must specify kill hard or soft """ + if not context: context = zmq.Context.instance() - #logging.info('Shutdown controller') + if hard: + self.state = CONTROL_STATES.SHUTDOWN - s = self.message_sender(context) - s.send(CONTROL_FRAME( - 'controller', - CONTROL_PROTOCOL.SHUTDOWN, - )) + print '[Controller] Hard Shutdown' - #for asoc in self.associated: - #asoc.close() + #for asoc in self.associated: + #asoc.close() + + if soft: + self.state = CONTROL_STATES.TERMINATE + + print '[Controller] Soft Shutdown' + + #for asoc in self.associated: + #asoc.close() def destroy(self): """ @@ -190,12 +450,8 @@ class Controller(object): #def __del__(self): #self.shutdown() - def qos(self): - if not self.debug: - return - return float(self.success) / (self.success + self.failed) - if __name__ == '__main__': + print 'Running on ',\ 'tcp://127.0.0.1:5000', \ 'tcp://127.0.0.1:5001', @@ -204,4 +460,8 @@ if __name__ == '__main__': 'tcp://127.0.0.1:5000', 'tcp://127.0.0.1:5001', ) - controller.run() + controller.manage( + 'freeform', + [] + ) + controller.run('green') diff --git a/zipline/protocol.py b/zipline/protocol.py index f8cd1435..a1341179 100644 --- a/zipline/protocol.py +++ b/zipline/protocol.py @@ -134,10 +134,14 @@ from protocol_utils import Enum, FrameExceptionFactory, namedict INVALID_CONTROL_FRAME = FrameExceptionFactory('CONTROL') +CONTROL_STATES = Enum( + 'RUNNING', + 'SHUTDOWN', # a soft kill + 'TERMINATE', # a hard kill +) + CONTROL_PROTOCOL = Enum( - 'INIT' , # 0 - req - 'INFO' , # 1 - req - 'STATUS' , # 2 - req + 'HEARTBEAT' , # 0 - req 'SHUTDOWN' , # 3 - req 'KILL' , # 4 - req @@ -152,7 +156,10 @@ def CONTROL_FRAME(id, status): return msgpack.dumps(tuple([id, status])) -def CONTORL_UNFRAME(msg): +def CONTROL_UNFRAME(msg): + """ + A status code and a message. + """ assert isinstance(msg, basestring) try: @@ -165,8 +172,6 @@ def CONTORL_UNFRAME(msg): raise INVALID_CONTROL_FRAME(msg) except ValueError: raise INVALID_CONTROL_FRAME(msg) - #except AssertionError: - #raise INVALID_CONTROL_FRAME(msg) # ----------------------- # Heartbeat Protocol diff --git a/zipline/simulator.py b/zipline/simulator.py index 8d6480b5..8a3abb27 100644 --- a/zipline/simulator.py +++ b/zipline/simulator.py @@ -41,7 +41,10 @@ class Simulator(ComponentHost): return 'Simple Simulator' def launch_controller(self): - thread = threading.Thread(target=self.controller.run) + thread = threading.Thread( + target=self.controller.run, + args=('thread',) + ) thread.start() self.subthreads.append(thread) diff --git a/zipline/test/test_monitor.py b/zipline/test/test_monitor.py index 0fea158d..e69de29b 100644 --- a/zipline/test/test_monitor.py +++ b/zipline/test/test_monitor.py @@ -1,78 +0,0 @@ -""" -Dummy simulator backported from Qexec for development on Zipline. -""" - -import logging -from multiprocessing import Process -from unittest2 import TestCase - -from zipline.monitor import Controller - -import gevent -from gevent_zeromq import zmq - -ctx = zmq.Context() - -#TODO: disabled by prefixing the test methods with a d -class TestControlProtocol(TestCase): - - def setUpController(self): - self.controller.run() - - def setUp(self): - self.controller = Controller( - 'tcp://127.0.0.1:5000', - 'tcp://127.0.0.1:5001', - ) - self.control_proc = Process(target=self.setUpController) - self.control_proc.start() - - def tearDown(self): - self.control_proc.terminate() - ctx.destroy() - - def asyncMessage(self, socket): - return socket.recv() - - def send_and_receive(self, push, sub, message='\x01'): - msg = gevent.spawn(sub.recv) - push.send(message) - gevent.sleep(0) # explicit gevent yield - msg.join() - self.assertEqual(msg.value, message) - - def dtest_control_message(self): - - sub = self.controller.message_listener(context=ctx) - message = gevent.spawn(self.asyncMessage, sub) - - push = self.controller.message_sender(context=ctx) - - # Don't like introducing time constants but because of - # the way gevent scheduler works zmq will often send all - # the messages off before the other thread even gets to - # listening. - self.send_and_receive(push, sub) - sub.close() - push.close() - - def dtest_control_delivery(self): - # Assert that the number of messages sent on the wire is - # the number of messages received, ie we don't drop any. - # This is of course depenendent on the topology of the - # listeners being fixed. Which normally it isn't. - - sub = self.controller.message_listener(context=ctx) - message = gevent.spawn(self.asyncMessage, sub) - - push = self.controller.message_sender(context=ctx) - - # Don't like introducing time constants but because of - # the way gevent scheduler works zmq will often send all - # the messages off before the other thread even gets to - # listening. - for i in xrange(25): - self.send_and_receive(push, sub) - - sub.close() - push.close() From c2bd2872c1e91ed4598b6a93f02b328ec3b61b9a Mon Sep 17 00:00:00 2001 From: Stephen Diehl Date: Tue, 3 Apr 2012 18:12:16 -0400 Subject: [PATCH 2/7] Concurrency agnostic controller. --- zipline/messaging.py | 1 + zipline/monitor.py | 185 ++++++++++++++++++++++--------------------- zipline/protocol.py | 24 +++--- zipline/simulator.py | 5 +- 4 files changed, 108 insertions(+), 107 deletions(-) diff --git a/zipline/messaging.py b/zipline/messaging.py index 552c1115..1bb80dcf 100644 --- a/zipline/messaging.py +++ b/zipline/messaging.py @@ -56,6 +56,7 @@ class ComponentHost(Component): raise Exception("There can be only one!") self.controller = controller + self.controller.zmq_flavor = self.zmq_flavor # Propogate the controller to all the subcomponents for component in self.components.itervalues(): diff --git a/zipline/monitor.py b/zipline/monitor.py index 2e0f2066..378e09e4 100644 --- a/zipline/monitor.py +++ b/zipline/monitor.py @@ -1,11 +1,19 @@ import time import gevent - -from gevent_zeromq import zmq +import itertools import util as qutil +# pyzmq +import zmq +# gevent_zeromq +import gevent_zeromq +# zmq_ctypes +#import zmq_ctypes + from protocol import CONTROL_PROTOCOL, CONTROL_FRAME, \ - CONTROL_UNFRAME, CONTROL_STATES + CONTROL_UNFRAME, CONTROL_STATES, INVALID_CONTROL_FRAME + +from gpoll import _Poller as GeventPoller # TODO: print -> qutil.LOGGER.info @@ -147,7 +155,10 @@ class Controller(object): def __init__(self, pub_socket, route_socket, logging = None): - self._ctx = None + self.context = None + self.zmq = None + self.zmq_poller = None + polling = False self.polling = polling @@ -171,6 +182,31 @@ class Controller(object): self.logging = False self.dologging = False + def init_zmq(self, flavor): + + assert self.zmq_flavor in ['thread', 'mp', 'green'] + + if flavor == 'mp': + self.zmq = zmq + self.context = self.zmq.Context() + self.zmq_poller = self.zmq.Poller + return + if flavor == 'thread': + self.zmq = zmq + self.context = self.zmq.Context.instance() + self.zmq_poller = self.zmq.Poller + return + if flavor == 'green': + self.zmq = gevent_zeromq.zmq + self.context = self.zmq.Context.instance() + self.zmq_poller = GeventPoller + return + if flavor == 'pypy': + self.zmq = zmq + self.context = self.zmq.Context.instance() + self.zmq_poller = self.zmq.Poller + return + def manage(self, topology, states=None, context=None): """ Give the controller a set set of components to manage and @@ -198,65 +234,75 @@ class Controller(object): # Start off in RUNNING, state self.state = self.states[0] - if not context: - self._ctx = zmq.Context.instance() - else: - self._ctx = context + def run(self): + self.init_zmq(self.zmq_flavor) - def run(self, flavor): - """ - Abstracts - """ - assert flavor in ['thread', 'mp', 'green'] - return self._poll() # use a python loop + try: + return self._poll() # use a python loop + except KeyboardInterrupt: + print 'Shutdown event loop' def log_status(self): print "[Controller] Tracking : %s" % ([c for c in self.tracked],) - gevent.spawn_later(5, self.log_status) + + # ----------- + # Event Loops + # ----------- def _poll(self): - self.pub = self._ctx.socket(zmq.PUB) + self.pub = self.context.socket(self.zmq.PUB) self.pub.bind(self.pub_socket) - self.router = self._ctx.socket(zmq.ROUTER) + self.router = self.context.socket(self.zmq.ROUTER) self.router.bind(self.route_socket) self.associated.extend([self.pub, self.router]) - # Spin the coroutines - gevent.spawn(self.log_status) - gevent.spawn_raw(self.recv_hearts_router) + poller = self.zmq.Poller() + poller.register(self.router, self.zmq.POLLIN) + + buffer = [] + + for i in itertools.count(0): + self.log_status() + self.responses = set() + + self.ctime = time.time() + self.pub.send(str(self.ctime)) + + while self.polling: + # Reset the responses for this cycle + + socks = dict(poller.poll(5)) + tic = time.time() + + if tic - self.ctime > 1: + break + + if self.router in socks and socks[self.router] == self.zmq.POLLIN: + rawmessage = self.router.recv() + + if rawmessage: + buffer.append(rawmessage) + + try: + if not self.router.getsockopt(self.zmq.RCVMORE): + self.handle_recv(buffer[:]) + buffer = [] + except INVALID_CONTROL_FRAME: + print 'Invalid frame', rawmessage + pass - while self.polling: self.beat() - gevent.sleep(self.period) - try: - msg = self.pull.recv() - self.pub.send(msg) - except KeyboardInterrupt: - self.polling = False - break - except zmq.ZMQError: - self.polling = False - break - except Exception as e: - # Its common to wrap these in wildcard exceptions so - # that we don't loose messages, ever - if self.logging: - self.logging.error(str(e)) - continue + if self.zmq_flavor == 'green': + gevent.sleep(0) # After loop exits self.terminated = True def beat(self): - toc = time.time() - - self.ctime += toc - self.tic - self.tic = toc - self.message = str(self.ctime) # These the set overloaded operations # A & B ~ set.intersection @@ -279,11 +325,6 @@ class Controller(object): for component in bad: self.fail(component) - # Reset the responses for this cycle - self.responses = set() - - gevent.spawn_raw(self.pub.send, str(self.ctime)) - # ------------------ # Component Handlers # ------------------ @@ -291,7 +332,7 @@ class Controller(object): # The various "states of being that a component can inform us # of def new(self, component): - print '[Controller] Tracking "%s" ' % component + print '[Controller] New Tracked "%s" ' % component if component in self.topology or self.freeform: self.tracked.add(component) @@ -313,36 +354,6 @@ class Controller(object): def exception(self, component, failure): pass - # -------- - # IO Loops - # -------- - - def recv_hearts(self): - buffer = [] - - while True: - message = self.router.recv(zmq.NOBLOCK, copy=False) - - if message: - buffer.append(message) - if not self.router.rcvmore(): - break - - gevent.spawn_later(0.001, self.handle_pong, buffer) - - def recv_hearts_router(self): - buffer = [] - - while True: - message = self.router.recv() - - if message: - buffer.append(message) - - if not self.router.getsockopt(zmq.RCVMORE): - gevent.spawn_raw(self.handle_recv, buffer[:]) - buffer = [] - # ----------------- # Protocol Handling # ----------------- @@ -354,7 +365,6 @@ class Controller(object): be coming over the wire. Which shouldn't happen ... right? """ identity = msg[0] - id, status = CONTROL_UNFRAME(msg[1]) # A component is telling us its alive: @@ -365,7 +375,7 @@ class Controller(object): else: # Otherwise its something weird and we don't know # what to do so just say so - print "Weird stuff happened: %s" % msg[1] + print "Weird stuff happened: %s" % status, self.ctime # A component is telling us it failed, and how if id is CONTROL_PROTOCOL.EXCEPTION: @@ -392,7 +402,7 @@ class Controller(object): """ if not context: - context = zmq.Context.instance() + context = self.zmq.Context.instance() s = context.socket(zmq.DEALER) s.connect(self.route_socket) @@ -408,7 +418,7 @@ class Controller(object): """ if not context: - context = zmq.Context.instance() + context = self.zmq.Context.instance() s = context.socket(zmq.SUB) s.connect(self.pub_socket) @@ -423,7 +433,7 @@ class Controller(object): assert hard or soft, """ Must specify kill hard or soft """ if not context: - context = zmq.Context.instance() + context = self.zmq.Context.instance() if hard: self.state = CONTROL_STATES.SHUTDOWN @@ -441,15 +451,6 @@ class Controller(object): #for asoc in self.associated: #asoc.close() - def destroy(self): - """ - Manual cleanup. - """ - self.shutdown() - - #def __del__(self): - #self.shutdown() - if __name__ == '__main__': print 'Running on ',\ diff --git a/zipline/protocol.py b/zipline/protocol.py index a1341179..4730e8ff 100644 --- a/zipline/protocol.py +++ b/zipline/protocol.py @@ -142,17 +142,17 @@ CONTROL_STATES = Enum( CONTROL_PROTOCOL = Enum( 'HEARTBEAT' , # 0 - req - 'SHUTDOWN' , # 3 - req - 'KILL' , # 4 - req + 'SHUTDOWN' , # 1 - req + 'KILL' , # 2 - req - 'OK' , # 5 - rep - 'DONE' , # 6 - rep - 'EXCEPTION' , # 7 - rep + 'OK' , # 3 - rep + 'DONE' , # 4 - rep + 'EXCEPTION' , # 5 - rep ) -def CONTROL_FRAME(id, status): - assert isinstance(id, basestring,) - assert isinstance(status, int) +def CONTROL_FRAME(event, payload): + assert isinstance(event, int,) + assert isinstance(payload, basestring) return msgpack.dumps(tuple([id, status])) @@ -163,11 +163,11 @@ def CONTROL_UNFRAME(msg): assert isinstance(msg, basestring) try: - id, status = msgpack.loads(msg) - assert isinstance(id, basestring) - assert isinstance(status, int) + event, payload = msgpack.loads(msg) + assert isinstance(event, int) + assert isinstance(payload, basestring) - return id, status + return event, payload except TypeError: raise INVALID_CONTROL_FRAME(msg) except ValueError: diff --git a/zipline/simulator.py b/zipline/simulator.py index 8a3abb27..c43f21fc 100644 --- a/zipline/simulator.py +++ b/zipline/simulator.py @@ -43,7 +43,6 @@ class Simulator(ComponentHost): def launch_controller(self): thread = threading.Thread( target=self.controller.run, - args=('thread',) ) thread.start() @@ -71,8 +70,8 @@ class Simulator(ComponentHost): if not self.running: return - #if self.controller: - #self.controller.shutdown() + if self.controller: + self.controller.shutdown() for component in self.components.itervalues(): component.shutdown() From e18dab58e5919819d9fc107ca15683b0b715d42f Mon Sep 17 00:00:00 2001 From: Stephen Diehl Date: Wed, 4 Apr 2012 10:47:51 -0400 Subject: [PATCH 3/7] Use proper logging. --- zipline/monitor.py | 42 +++++++++++++++++++++--------------------- 1 file changed, 21 insertions(+), 21 deletions(-) diff --git a/zipline/monitor.py b/zipline/monitor.py index 378e09e4..dc258ca2 100644 --- a/zipline/monitor.py +++ b/zipline/monitor.py @@ -1,8 +1,6 @@ import time import gevent import itertools -import util as qutil - # pyzmq import zmq # gevent_zeromq @@ -15,10 +13,6 @@ from protocol import CONTROL_PROTOCOL, CONTROL_FRAME, \ from gpoll import _Poller as GeventPoller -# TODO: print -> qutil.LOGGER.info - -# When you're cold and waiting for the train... draw ascii art! -# # Roll Call ( Discovery ) # ----------------------- # @@ -151,7 +145,7 @@ class Controller(object): """ debug = False - period = 1 + period = 5 def __init__(self, pub_socket, route_socket, logging = None): @@ -177,10 +171,9 @@ class Controller(object): if logging: self.logging = logging - self.dologging = True else: - self.logging = False - self.dologging = False + import util as qutil + self.logging = qutil.LOGGER def init_zmq(self, flavor): @@ -240,10 +233,10 @@ class Controller(object): try: return self._poll() # use a python loop except KeyboardInterrupt: - print 'Shutdown event loop' + self.logging.info('Shutdown event loop') def log_status(self): - print "[Controller] Tracking : %s" % ([c for c in self.tracked],) + self.logging.info("[Controller] Tracking : %s" % ([c for c in self.tracked],)) # ----------- # Event Loops @@ -274,10 +267,10 @@ class Controller(object): while self.polling: # Reset the responses for this cycle - socks = dict(poller.poll(5)) + socks = dict(poller.poll(self.period)) tic = time.time() - if tic - self.ctime > 1: + if tic - self.ctime > self.period: break if self.router in socks and socks[self.router] == self.zmq.POLLIN: @@ -291,7 +284,7 @@ class Controller(object): self.handle_recv(buffer[:]) buffer = [] except INVALID_CONTROL_FRAME: - print 'Invalid frame', rawmessage + self.logging.error('Invalid frame', rawmessage) pass self.beat() @@ -299,6 +292,9 @@ class Controller(object): if self.zmq_flavor == 'green': gevent.sleep(0) + if not self.polling: + break + # After loop exits self.terminated = True @@ -332,7 +328,7 @@ class Controller(object): # The various "states of being that a component can inform us # of def new(self, component): - print '[Controller] New Tracked "%s" ' % component + self.logging.info('[Controller] New Tracked "%s" ' % component) if component in self.topology or self.freeform: self.tracked.add(component) @@ -342,7 +338,7 @@ class Controller(object): raise UnknownChatter(component) def fail(self, component): - print '[Controller] Component "%s" timed out' % component + self.logging.info('[Controller] Component "%s" timed out' % component) self.tracked.remove(component) def done(self, component): @@ -375,7 +371,7 @@ class Controller(object): else: # Otherwise its something weird and we don't know # what to do so just say so - print "Weird stuff happened: %s" % status, self.ctime + self.logging.error("Weird stuff happened: %s" % status, self.ctime) # A component is telling us it failed, and how if id is CONTROL_PROTOCOL.EXCEPTION: @@ -428,17 +424,21 @@ class Controller(object): return s def shutdown(self, hard=False, soft=True, context=None): + + if not self.polling: + return + self.polling = False assert hard or soft, """ Must specify kill hard or soft """ if not context: - context = self.zmq.Context.instance() + context = zmq.Context.instance() if hard: self.state = CONTROL_STATES.SHUTDOWN - print '[Controller] Hard Shutdown' + self.logging.info('[Controller] Hard Shutdown') #for asoc in self.associated: #asoc.close() @@ -446,7 +446,7 @@ class Controller(object): if soft: self.state = CONTROL_STATES.TERMINATE - print '[Controller] Soft Shutdown' + self.logging.info('[Controller] Soft Shutdown') #for asoc in self.associated: #asoc.close() From 3d8e355cfaf60857ff6d6e8ce670a4ccc70ab89f Mon Sep 17 00:00:00 2001 From: Stephen Diehl Date: Wed, 4 Apr 2012 11:42:58 -0400 Subject: [PATCH 4/7] Integrate controller into do_work. --- zipline/messaging.py | 45 +++++++++++++++++++++++++++++++++++++++++--- zipline/monitor.py | 34 ++++++++++++++++++++++++++++----- zipline/protocol.py | 2 +- 3 files changed, 72 insertions(+), 9 deletions(-) diff --git a/zipline/messaging.py b/zipline/messaging.py index 1bb80dcf..10b45d90 100644 --- a/zipline/messaging.py +++ b/zipline/messaging.py @@ -8,7 +8,7 @@ import zipline.util as qutil from zipline.component import Component import zipline.protocol as zp from zipline.protocol import CONTROL_PROTOCOL, COMPONENT_TYPE, \ - COMPONENT_STATE + COMPONENT_STATE, CONTROL_FRAME, CONTROL_UNFRAME class ComponentHost(Component): """ @@ -212,8 +212,29 @@ class ParallelBuffer(Component): # wait for synchronization reply from the host socks = dict(self.poll.poll(self.heartbeat_timeout)) #timeout after 2 seconds. + # TODO: Abstract this out, maybe on base component if self.control_in in socks and socks[self.control_in] == self.zmq.POLLIN: msg = self.control_in.recv() + event, payload = CONTROL_UNFRAME(msg) + + # -- Heartbeat -- + if event == CONTROL_PROTOCOL.HEARTBEAT: + # Heart outgoing + heartbeat_frame = CONTROL_FRAME( + CONTROL_PROTOCOL.OK, + payload + ) + self.control_out.send(heartbeat_frame) + + # -- Soft Kill -- + elif event == CONTROL_PROTOCOL.SHUTDOWN: + self.done() + self.shutdown() + + # -- Hard Kill -- + elif event == CONTROL_PROTOCOL.KILL: + self.kill() + if self.pull_socket in socks and socks[self.pull_socket] == self.zmq.POLLIN: message = self.pull_socket.recv() @@ -241,13 +262,11 @@ class ParallelBuffer(Component): except zp.INVALID_DATASOURCE_FRAME as exc: return self.signal_exception(exc) - # def unframe(self, msg): return zp.DATASOURCE_UNFRAME(msg) def frame(self, event): return zp.FEED_FRAME(event) - # ------------- # Flow Control @@ -464,8 +483,28 @@ class BaseTransform(Component): """ socks = dict(self.poll.poll(self.heartbeat_timeout)) + # TODO: Abstract this out, maybe on base component if self.control_in in socks and socks[self.control_in] == self.zmq.POLLIN: msg = self.control_in.recv() + event, payload = CONTROL_UNFRAME(msg) + + # -- Heartbeat -- + if event == CONTROL_PROTOCOL.HEARTBEAT: + # Heart outgoing + heartbeat_frame = CONTROL_FRAME( + CONTROL_PROTOCOL.OK, + payload + ) + self.control_out.send(heartbeat_frame) + + # -- Soft Kill -- + elif event == CONTROL_PROTOCOL.SHUTDOWN: + self.done() + self.shutdown() + + # -- Hard Kill -- + elif event == CONTROL_PROTOCOL.KILL: + self.kill() if self.feed_socket in socks and socks[self.feed_socket] == self.zmq.POLLIN: message = self.feed_socket.recv() diff --git a/zipline/monitor.py b/zipline/monitor.py index dc258ca2..6edd0286 100644 --- a/zipline/monitor.py +++ b/zipline/monitor.py @@ -145,7 +145,7 @@ class Controller(object): """ debug = False - period = 5 + period = 1 def __init__(self, pub_socket, route_socket, logging = None): @@ -237,6 +237,30 @@ class Controller(object): def log_status(self): self.logging.info("[Controller] Tracking : %s" % ([c for c in self.tracked],)) + # ------------- + # Publications + # ------------- + + def send_heart(self): + heartbeat_frame = CONTROL_FRAME( + CONTROL_PROTOCOL.HEARTBEAT, + str(self.ctime) + ) + self.pub.send(heartbeat_frame) + + def send_hardkill(self): + kill_frame = CONTROL_FRAME( + CONTROL_PROTOCOL.KILL, + None + ) + self.pub.send(kill_frame) + + def send_softkill(self): + soft_frame = CONTROL_FRAME( + CONTROL_PROTOCOL.KILL, + None + ) + self.pub.send(soft_frame) # ----------- # Event Loops @@ -262,7 +286,7 @@ class Controller(object): self.responses = set() self.ctime = time.time() - self.pub.send(str(self.ctime)) + self.send_heart() while self.polling: # Reset the responses for this cycle @@ -345,10 +369,10 @@ class Controller(object): # TODO: This will be what we ship off to vbench at some # point... # print component finished at self.ctime - pass + self.logging.info('[Controller] Component "%s" done.' % component) def exception(self, component, failure): - pass + self.logging.error('Component "%s"in exception state' % component) # ----------------- # Protocol Handling @@ -401,8 +425,8 @@ class Controller(object): context = self.zmq.Context.instance() s = context.socket(zmq.DEALER) - s.connect(self.route_socket) s.setsockopt(zmq.IDENTITY, identity) + s.connect(self.route_socket) self.associated.append(s) return s diff --git a/zipline/protocol.py b/zipline/protocol.py index 4730e8ff..133e7260 100644 --- a/zipline/protocol.py +++ b/zipline/protocol.py @@ -154,7 +154,7 @@ def CONTROL_FRAME(event, payload): assert isinstance(event, int,) assert isinstance(payload, basestring) - return msgpack.dumps(tuple([id, status])) + return msgpack.dumps(tuple([event, payload])) def CONTROL_UNFRAME(msg): """ From 42f16ccdfbfdae8eac17443b47ff459c8f2a116c Mon Sep 17 00:00:00 2001 From: Stephen Diehl Date: Wed, 4 Apr 2012 12:22:12 -0400 Subject: [PATCH 5/7] Done & Exception tracking through sockets -> Controller --- zipline/component.py | 25 ++++++++++++++++++++++--- zipline/finance/trading.py | 4 ++++ zipline/messaging.py | 7 ++++++- zipline/monitor.py | 18 ++++++++++-------- 4 files changed, 42 insertions(+), 12 deletions(-) diff --git a/zipline/component.py b/zipline/component.py index 9acf2cc2..13b852b9 100644 --- a/zipline/component.py +++ b/zipline/component.py @@ -24,7 +24,7 @@ from datetime import datetime import zipline.util as qutil from zipline.gpoll import _Poller as GeventPoller from zipline.protocol import CONTROL_PROTOCOL, COMPONENT_STATE, \ - COMPONENT_FAILURE, BACKTEST_STATE + COMPONENT_FAILURE, BACKTEST_STATE, CONTROL_FRAME class Component(object): @@ -313,6 +313,12 @@ class Component(object): exc_type, exc_value, exc_traceback = sys.exc_info() self.stack_trace = exc_traceback + exception_frame = CONTROL_FRAME( + CONTROL_PROTOCOL.EXCEPTION, + str(exc) + ) + self.control_out.send(exception_frame) + qutil.LOGGER.exception("Unexpected error in run for {id}.".format(id=self.get_id)) def signal_done(self): @@ -329,10 +335,19 @@ class Component(object): # TODO: proper framing self.sync_socket.send(self.get_id + ":" + str(CONTROL_PROTOCOL.DONE)) + #notify controller we're done + done_frame = CONTROL_FRAME( + CONTROL_PROTOCOL.DONE, + '' + ) + self.control_out.send(done_frame) + self.receive_sync_ack() #notify internal work look that we're done self.done = True # TODO: use state flag + qutil.LOGGER.info("[%s] DONE" % self.get_id) + # ----------- # Messaging # ----------- @@ -350,13 +365,15 @@ class Component(object): def receive_sync_ack(self): """ Wait for synchronization reply from the host. + + DEPRECATED, left in for compatability for now. """ socks = dict(self.sync_poller.poll(self.heartbeat_timeout)) if self.sync_socket in socks and socks[self.sync_socket] == self.zmq.POLLIN: message = self.sync_socket.recv() - else: - raise Exception("Sync ack timed out on response for {id}".format(id=self.get_id)) + #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']) @@ -445,6 +462,8 @@ class Component(object): def setup_sync(self): """ Setup the sync socket and poller. ( Connect ) + + DEPRECATED, left in for compatability for now. """ qutil.LOGGER.debug("Connecting sync client for {id}".format(id=self.get_id)) diff --git a/zipline/finance/trading.py b/zipline/finance/trading.py index 200a50d5..a32418b3 100644 --- a/zipline/finance/trading.py +++ b/zipline/finance/trading.py @@ -3,6 +3,7 @@ import pytz import math import pandas +# from gevent.select import select from zmq.core.poll import select import zipline.messaging as qmsg @@ -200,6 +201,9 @@ class OrderDataSource(qmsg.DataSource): #pull all orders from client. orders = [] count = 0 + + # TODO : this can be written in a concurrency agnostic + # way... have a chat with Fawce about this ~Steve while True: (rlist, wlist, xlist) = select( diff --git a/zipline/messaging.py b/zipline/messaging.py index 10b45d90..749006d0 100644 --- a/zipline/messaging.py +++ b/zipline/messaging.py @@ -113,6 +113,10 @@ class ComponentHost(Component): self.launch_controller() def is_timed_out(self): + """ + DEPRECATED, left in for compatability for now. + """ + cur_time = datetime.datetime.utcnow() if len(self.components) == 0: @@ -145,7 +149,7 @@ class ComponentHost(Component): self.signal_exception(exc) if status == str(CONTROL_PROTOCOL.DONE): # TODO: other way around - qutil.LOGGER.info("{id} is DONE".format(id=sync_id)) + #qutil.LOGGER.debug("{id} is DONE".format(id=sync_id)) self.unregister_component(sync_id) self.state_flag = COMPONENT_STATE.DONE else: @@ -508,6 +512,7 @@ class BaseTransform(Component): 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 diff --git a/zipline/monitor.py b/zipline/monitor.py index 6edd0286..88a73269 100644 --- a/zipline/monitor.py +++ b/zipline/monitor.py @@ -236,7 +236,12 @@ class Controller(object): self.logging.info('Shutdown event loop') def log_status(self): - self.logging.info("[Controller] Tracking : %s" % ([c for c in self.tracked],)) + """ + Snapshot of the tracked components at every period. + """ + #self.logging.info("[Controller] Tracking : %s" % ([c for c in self.tracked],)) + pass + # ------------- # Publications # ------------- @@ -352,7 +357,7 @@ class Controller(object): # The various "states of being that a component can inform us # of def new(self, component): - self.logging.info('[Controller] New Tracked "%s" ' % component) + self.logging.info('[Controller] Alive "%s" ' % component) if component in self.topology or self.freeform: self.tracked.add(component) @@ -372,7 +377,7 @@ class Controller(object): self.logging.info('[Controller] Component "%s" done.' % component) def exception(self, component, failure): - self.logging.error('Component "%s"in exception state' % component) + self.logging.error('Component "%s" in exception state' % component) # ----------------- # Protocol Handling @@ -395,7 +400,7 @@ class Controller(object): else: # Otherwise its something weird and we don't know # what to do so just say so - self.logging.error("Weird stuff happened: %s" % status, self.ctime) + self.logging.error("Weird stuff happened: %s" % msg) # A component is telling us it failed, and how if id is CONTROL_PROTOCOL.EXCEPTION: @@ -404,7 +409,7 @@ class Controller(object): # A component is telling us its done with work and won't # be talking to us anymore if id is CONTROL_PROTOCOL.DONE: - self.done(identity, status) + self.done(identity) # ------------------- # Hooks for Endpoints @@ -456,9 +461,6 @@ class Controller(object): assert hard or soft, """ Must specify kill hard or soft """ - if not context: - context = zmq.Context.instance() - if hard: self.state = CONTROL_STATES.SHUTDOWN From f3035c57394b845268bf2f0b25f283662735271a Mon Sep 17 00:00:00 2001 From: Stephen Diehl Date: Wed, 4 Apr 2012 14:32:31 -0400 Subject: [PATCH 6/7] monitor.py killing softly --- zipline/monitor.py | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/zipline/monitor.py b/zipline/monitor.py index 88a73269..0e007d3a 100644 --- a/zipline/monitor.py +++ b/zipline/monitor.py @@ -256,14 +256,14 @@ class Controller(object): def send_hardkill(self): kill_frame = CONTROL_FRAME( CONTROL_PROTOCOL.KILL, - None + '' ) self.pub.send(kill_frame) def send_softkill(self): soft_frame = CONTROL_FRAME( CONTROL_PROTOCOL.KILL, - None + '' ) self.pub.send(soft_frame) @@ -473,6 +473,7 @@ class Controller(object): self.state = CONTROL_STATES.TERMINATE self.logging.info('[Controller] Soft Shutdown') + self.send_softkill() #for asoc in self.associated: #asoc.close() From c4dbec1e30cc8508c256ad1f5f1df211c3d7bf1d Mon Sep 17 00:00:00 2001 From: Stephen Diehl Date: Wed, 4 Apr 2012 18:03:13 -0400 Subject: [PATCH 7/7] MergedParallelBuffer -> Merge, ParallelBuffer -> Feed --- zipline/component.py | 12 ++++++------ zipline/messaging.py | 10 +++++----- 2 files changed, 11 insertions(+), 11 deletions(-) diff --git a/zipline/component.py b/zipline/component.py index 13b852b9..7f947ee9 100644 --- a/zipline/component.py +++ b/zipline/component.py @@ -42,9 +42,9 @@ class Component(object): :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) + between data sources and a 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 @@ -53,9 +53,9 @@ class Component(object): :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) + transforms to one 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 diff --git a/zipline/messaging.py b/zipline/messaging.py index 749006d0..2c4caebc 100644 --- a/zipline/messaging.py +++ b/zipline/messaging.py @@ -39,8 +39,8 @@ class ComponentHost(Component): self.sync_register = {} self.timeout = datetime.timedelta(seconds=5) - self.feed = ParallelBuffer() - self.merge = MergedParallelBuffer() + self.feed = Feed() + self.merge = Merge() self.passthrough = PassthroughTransform() self.controller = None @@ -173,7 +173,7 @@ class ComponentHost(Component): raise NotImplementedError -class ParallelBuffer(Component): +class Feed(Component): """ Connects to N PULL sockets, publishing all messages received to a PUB socket. Published messages are guaranteed to be in chronological order @@ -367,13 +367,13 @@ class ParallelBuffer(Component): return len(self.data_buffer) -class MergedParallelBuffer(ParallelBuffer): +class Merge(Feed): """ Merges multiple streams of events into single messages. """ def __init__(self): - ParallelBuffer.__init__(self) + Feed.__init__(self) self.init()