diff --git a/zipline/component.py b/zipline/component.py index cfdc8781..7f947ee9 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): @@ -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 @@ -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 @@ -310,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): @@ -326,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 # ----------- @@ -347,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']) @@ -427,8 +447,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]) @@ -436,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 e39bde7d..edae9782 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/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..2c4caebc 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): """ @@ -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 @@ -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(): @@ -112,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: @@ -128,7 +133,7 @@ class ComponentHost(Component): return False - def loop(self): + def loop(self, lockstep=True): while not self.is_timed_out(): # wait for synchronization request @@ -144,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: @@ -168,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 @@ -211,8 +216,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() @@ -240,13 +266,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 @@ -343,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() @@ -463,11 +487,32 @@ 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() + if message == str(CONTROL_PROTOCOL.DONE): self.signal_done() return diff --git a/zipline/monitor.py b/zipline/monitor.py index 5db2ea54..0e007d3a 100644 --- a/zipline/monitor.py +++ b/zipline/monitor.py @@ -1,29 +1,133 @@ +import time +import gevent +import itertools +# pyzmq import zmq -from zipline.protocol import CONTROL_PROTOCOL, CONTROL_FRAME +# gevent_zeromq +import gevent_zeromq +# zmq_ctypes +#import zmq_ctypes + +from protocol import CONTROL_PROTOCOL, CONTROL_FRAME, \ + CONTROL_UNFRAME, CONTROL_STATES, INVALID_CONTROL_FRAME + +from gpoll import _Poller as GeventPoller + +# 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,169 +137,349 @@ 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.context = None + self.zmq = None + self.zmq_poller = 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 - self.dologging = True else: - self.logging = False - self.dologging = False + import util as qutil + self.logging = qutil.LOGGER - self.success = 0 - self.failed = 0 + def init_zmq(self, flavor): - def run(self, debug=False, context=None): + 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): """ - Run's the loop for the broker. + Give the controller a set set of components to manage and + a set of state transitions for the entire system. """ + + # 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 - if not context: - self._ctx = zmq.Context.instance() - else: - self._ctx = context + # Start off in RUNNING, state + self.state = self.states[0] - #if not debug: - #return self._poll_fast() # the c loop - #else: - return self._poll() # use a python loop + def run(self): + self.init_zmq(self.zmq_flavor) - def _poll_fast(self): + try: + return self._poll() # use a python loop + except KeyboardInterrupt: + self.logging.info('Shutdown event loop') + + def log_status(self): """ - C version of the polling forwarder. + Snapshot of the tracked components at every period. """ - self.pull = self._ctx.socket(zmq.PULL) - self.pub = self._ctx.socket(zmq.PUB) + #self.logging.info("[Controller] Tracking : %s" % ([c for c in self.tracked],)) + pass - self.pull.bind(self.pull_socket) - self.pub.bind(self.pub_socket) + # ------------- + # Publications + # ------------- - zmq.device(zmq.FORWARDER, self.pull, self.pub) + 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, + '' + ) + self.pub.send(kill_frame) + + def send_softkill(self): + soft_frame = CONTROL_FRAME( + CONTROL_PROTOCOL.KILL, + '' + ) + self.pub.send(soft_frame) + + # ----------- + # Event Loops + # ----------- 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 = self.context.socket(self.zmq.PUB) self.pub.bind(self.pub_socket) - self.associated.extend([self.pull, self.pub]) + self.router = self.context.socket(self.zmq.ROUTER) + self.router.bind(self.route_socket) - while self.polling: - 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)) - self.failed += 1 - continue + self.associated.extend([self.pub, self.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.send_heart() + + while self.polling: + # Reset the responses for this cycle + + socks = dict(poller.poll(self.period)) + tic = time.time() + + if tic - self.ctime > self.period: + 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: + self.logging.error('Invalid frame', rawmessage) + pass + + self.beat() + + if self.zmq_flavor == 'green': + gevent.sleep(0) + + if not self.polling: + break + + # After loop exits self.terminated = True + def beat(self): + + # 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) + + # ------------------ + # Component Handlers + # ------------------ + + # The various "states of being that a component can inform us + # of + def new(self, component): + self.logging.info('[Controller] Alive "%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): + self.logging.info('[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 + self.logging.info('[Controller] Component "%s" done.' % component) + + def exception(self, component, failure): + self.logging.error('Component "%s" in exception state' % component) + + # ----------------- + # 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 + self.logging.error("Weird stuff happened: %s" % msg) + + # 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) + # ------------------- # 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. """ if not context: - context = zmq.Context.instance() + context = self.zmq.Context.instance() + + s = context.socket(zmq.DEALER) + s.setsockopt(zmq.IDENTITY, identity) + s.connect(self.route_socket) - s = context.socket(zmq.PUSH) - s.connect(self.push_socket) 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. """ if not context: - context = zmq.Context.instance() + context = self.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): + + if not self.polling: + return + self.polling = False - if not context: - context = zmq.Context.instance() + assert hard or soft, """ Must specify kill hard or soft """ - #logging.info('Shutdown controller') + if hard: + self.state = CONTROL_STATES.SHUTDOWN - s = self.message_sender(context) - s.send(CONTROL_FRAME( - 'controller', - CONTROL_PROTOCOL.SHUTDOWN, - )) + self.logging.info('[Controller] Hard Shutdown') - #for asoc in self.associated: - #asoc.close() + #for asoc in self.associated: + #asoc.close() - def destroy(self): - """ - Manual cleanup. - """ - self.shutdown() + if soft: + self.state = CONTROL_STATES.TERMINATE - #def __del__(self): - #self.shutdown() + self.logging.info('[Controller] Soft Shutdown') + self.send_softkill() - def qos(self): - if not self.debug: - return - return float(self.success) / (self.success + self.failed) + #for asoc in self.associated: + #asoc.close() if __name__ == '__main__': + print 'Running on ',\ 'tcp://127.0.0.1:5000', \ 'tcp://127.0.0.1:5001', @@ -204,4 +488,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 d373ca9e..f4ee5d68 100644 --- a/zipline/protocol.py +++ b/zipline/protocol.py @@ -134,39 +134,44 @@ from protocol_utils import Enum, FrameExceptionFactory, namedict INVALID_CONTROL_FRAME = FrameExceptionFactory('CONTROL') -CONTROL_PROTOCOL = Enum( - 'INIT' , # 0 - req - 'INFO' , # 1 - req - 'STATUS' , # 2 - req - 'SHUTDOWN' , # 3 - req - 'KILL' , # 4 - req - - 'OK' , # 5 - rep - 'DONE' , # 6 - rep - 'EXCEPTION' , # 7 - rep +CONTROL_STATES = Enum( + 'RUNNING', + 'SHUTDOWN', # a soft kill + 'TERMINATE', # a hard kill ) -def CONTROL_FRAME(id, status): - assert isinstance(id, basestring,) - assert isinstance(status, int) +CONTROL_PROTOCOL = Enum( + 'HEARTBEAT' , # 0 - req + 'SHUTDOWN' , # 1 - req + 'KILL' , # 2 - req - return msgpack.dumps(tuple([id, status])) + 'OK' , # 3 - rep + 'DONE' , # 4 - rep + 'EXCEPTION' , # 5 - rep +) -def CONTORL_UNFRAME(msg): +def CONTROL_FRAME(event, payload): + assert isinstance(event, int,) + assert isinstance(payload, basestring) + + return msgpack.dumps(tuple([event, payload])) + +def CONTROL_UNFRAME(msg): + """ + A status code and a message. + """ 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: 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..c43f21fc 100644 --- a/zipline/simulator.py +++ b/zipline/simulator.py @@ -41,7 +41,9 @@ class Simulator(ComponentHost): return 'Simple Simulator' def launch_controller(self): - thread = threading.Thread(target=self.controller.run) + thread = threading.Thread( + target=self.controller.run, + ) thread.start() self.subthreads.append(thread) @@ -68,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() 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()