diff --git a/tests/test_components.py b/tests/test_components.py new file mode 100644 index 00000000..02b05a69 --- /dev/null +++ b/tests/test_components.py @@ -0,0 +1,106 @@ +import zmq +import pytz +from datetime import datetime, timedelta + +from unittest2 import TestCase +from collections import defaultdict + +from zipline.finance.trading import SIMULATION_STYLE +from zipline.core.devsimulator import AddressAllocator +from zipline.lines import SimulatedTrading + +from zipline.utils.test_utils import ( + drain_zipline, + check, + setup_logger, + teardown_logger, + launch_component, + create_monitor, + launch_monitor +) + + +from zipline.core import Component +from zipline.core.component import ComponentSocketArgs +from zipline.protocol import ( + DATASOURCE_FRAME, + DATASOURCE_UNFRAME +) + +from zipline.gens.tradegens import SpecificEquityTrades +from zipline.gens.utils import hash_args +from zipline.gens.zmqgen import gen_from_poller + +import logbook +log = logbook.Logger('ComponentTestCase') + +allocator = AddressAllocator(1000) + + +class ComponentTestCase(TestCase): + + leased_sockets = defaultdict(list) + + def setUp(self): + self.zipline_test_config = { + 'allocator' : allocator, + 'sid' : 133, + 'devel' : False, + 'results_socket' : allocator.lease(1)[0], + 'simulation_style' : SIMULATION_STYLE.FIXED_SLIPPAGE + } + self.ctx = zmq.Context() + setup_logger(self) + + def tearDown(self): + self.ctx.term() + teardown_logger(self) + + def test_specific_equity_source(self): + filter = [1,2,3,4] + #Set up source a. One minute between events. + args_a = tuple() + kwargs_a = { + 'sids' : [1,2], + 'start' : datetime(2012,6,6,0,tzinfo=pytz.utc), + 'delta' : timedelta(minutes = 1), + 'filter' : filter, + 'count' : 100 + } + + c_id = SpecificEquityTrades.__name__ + hash_args(args_a, kwargs_a) + mon = create_monitor(allocator) + + out_socket_args = ComponentSocketArgs( + style=zmq.PUSH, + uri=allocator.lease(1)[0], + bind=True + ) + + c = Component( + SpecificEquityTrades, + args_a, + kwargs_a, + c_id, + out_socket_args, + DATASOURCE_FRAME, + mon + ) + + mon.manage(set([c.get_id])) + mon_proc = launch_monitor(mon) + + # launch in a process + proc = launch_component(c) + + pull_socket = self.ctx.socket(zmq.PULL) + pull_socket.connect(out_socket_args.uri) + poller = zmq.Poller() + poller.register(pull_socket, zmq.POLLIN) + unframe = DATASOURCE_UNFRAME + for msg in gen_from_poller(poller, pull_socket, unframe): + # assert things about the messages. + log.info(msg) + + pull_socket.close() + log.info("DONE!") diff --git a/tests/test_monitor.py b/tests/test_monitor.py index 5f55aaee..3d063954 100644 --- a/tests/test_monitor.py +++ b/tests/test_monitor.py @@ -1,7 +1,7 @@ from zipline.utils.test_utils import setup_logger, teardown_logger from unittest2 import TestCase, skip -from zipline.core.monitor import Controller +from zipline.core.monitor import Monitor class TestMonitor(TestCase): def setUp(self): @@ -15,12 +15,12 @@ class TestMonitor(TestCase): pub_socket = 'tcp://127.0.0.1:5000' route_socket = 'tcp://127.0.0.1:5001' - con = Controller(pub_socket, route_socket) - con.manage([]) + mon = Monitor(pub_socket, route_socket) + mon.manage([]) def test_init_topology(self): pub_socket = 'tcp://127.0.0.1:5000' route_socket = 'tcp://127.0.0.1:5001' - con = Controller(pub_socket, route_socket, ) - con.manage([ 'a', 'b', 'c', 'd' ]) + mon = Monitor(pub_socket, route_socket, ) + mon.manage([ 'a', 'b', 'c', 'd' ]) diff --git a/zipline/__init__.py b/zipline/__init__.py index 23bcca40..a84cd345 100644 --- a/zipline/__init__.py +++ b/zipline/__init__.py @@ -6,14 +6,14 @@ Zipline # it is a place to expose the public interfaces. import protocol # namespace -from core.monitor import Controller +from core.monitor import Monitor from lines import SimulatedTrading from core.host import ComponentHost from utils.protocol_utils import ndict __all__ = [ SimulatedTrading, - Controller, + Monitor, ComponentHost, protocol, ndict diff --git a/zipline/core/__init__.py b/zipline/core/__init__.py index d487dd05..a7d6b1f8 100644 --- a/zipline/core/__init__.py +++ b/zipline/core/__init__.py @@ -1,9 +1,9 @@ from host import ComponentHost from component import Component -from monitor import Controller +from monitor import Monitor __all__ = [ Component, - Controller, + Monitor, ComponentHost ] diff --git a/zipline/core/component.py b/zipline/core/component.py index 962966f4..f9dc63c1 100644 --- a/zipline/core/component.py +++ b/zipline/core/component.py @@ -11,10 +11,13 @@ import logbook import traceback import humanhash from setproctitle import setproctitle +from collections import namedtuple # pyzmq import zmq +from zipline.gens.zmqgen import gen_from_poller + from zipline.core.monitor import PARAMETERS from zipline.protocol import ( @@ -26,146 +29,85 @@ from zipline.protocol import ( EXCEPTION_FRAME ) -log = logbook.Logger('Component') -from zipline.exceptions import ComponentNoInit +log = logbook.Logger('Component') class KillSignal(Exception): def __init__(self): pass +ComponentSocketArgs = namedtuple('ComponentSocket',['uri','style','bind']) + class Component(object): - """ - Base class for components. Defines the the base messaging - interface for components. - - :param addresses: a dict of name_string -> zmq port address strings. - Must have the following entries - - :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 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 - will be used in PUB/SUB sockets between Feed and - Transforms. Bind is always on the PUB side. - - :param merge_address: socket address used to publish transformed - values. will be used in PUSH/PULL from many - transforms to one Merge Bind will always be on - the PULL side (we always have N producers and - 1 consumer) - - :param results_address: socket address used to publish merged data - source feed and transforms to clients will be - used in PUB/SUB from one Merge to one or many - clients. Bind is always on the PUB side. - - bind/connect methods will return the correct socket type for each - address. - - """ - # ------------ # Construction # ------------ - abstract = True - #__metaclass__ = WorkflowMeta + def __init__(self, + gen_func, + gen_args, + gen_kwargs, + component_id, + out_socket_args, + frame, + monitor, + in_socket_args=None, + unframe=None + ): - def __init__(self, *args, **kwargs): - self.zmq = None - self.context = None - self.addresses = None - self.waiting = None + assert component_id, \ + "Every component needs a unique and invariant identifier" + assert isinstance(component_id, basestring), \ + "Components must have string IDs" + assert isinstance(out_socket_args, ComponentSocketArgs), \ + "out_socket_args args must be ComponentSocketArgs" + + if in_socket_args: + assert isinstance(in_socket_args, ComponentSocketArgs), \ + "in_socket_args args must be ComponentSocketArgs" + + # ----------------- + # Generator + # ----------------- + self.component_id = component_id + self.gen_args = gen_args + self.gen_kwargs = gen_kwargs + self.gen_func = gen_func + self.generator = None + self.frame = frame + + # lock for waiting on monitor "GO" + self.waiting = None + + # ----------------- + # ZMQ properties + # ----------------- + self.in_socket_args = in_socket_args + self.out_socket_args = out_socket_args + self.zmq = None + self.context = None + self.out_socket = None + self.in_socket = None + self.monitor = monitor + self.unframe = unframe - self.out_socket = None - self.killed = False - self.controller = None - # timeout on heartbeat is very short to avoid burning - # cycles on heartbeating. unit is milliconds - self.heartbeat_timeout = 0 # TODO: state_flag is deprecated, remove - # TODO: error_state is deprecated, remove - self.state_flag = COMPONENT_STATE.OK - self.error_state = COMPONENT_FAILURE.NOFAILURE - self.on_done = None + self.state_flag = COMPONENT_STATE.OK - self._exception = None - self.fail_time = None - self.start_tic = None - self.stop_tic = None - self.note = None - self.confirmed = False - self.devel = False - self.socks = None - self.last_ping = None + # track time of last ping we received from monitor + self.last_ping = None # Humanhashes make this way easier to debug because they stick # in your mind unlike a 32 byte string of random hex. - self.guid = uuid.uuid4() - self.huid = humanhash.humanize(self.guid.hex) - - # This is where component specific constructors should be - # defined. Arguments passed to init are threaded through. - self.init(*args, **kwargs) - - def init(self): - """ - Subclasses should override this to extend the setup for the - class. Shouldn't have side effects. - """ - raise ComponentNoInit(self.__class__) + self.guid = uuid.uuid4() + self.huid = humanhash.humanize(self.guid.hex) # ------------ # Core Methods # ------------ - def open(self): - """ - Open the connections needed to start doing work. - """ - raise NotImplementedError - - def ready(self): - """ - Return ``True`` if and only if the component has finished - execution. - """ - return self.state_flag in [COMPONENT_STATE.DONE, \ - COMPONENT_STATE.EXCEPTION] - - def successful(self): - """ - Return ``True`` if and only if the component has finished - execution successfully, that is, without raising an error. - """ - return self.state_flag == COMPONENT_STATE.DONE and not \ - self.exception - - @property - def exception(self): - """ - Holds the exception that the component failed on, or ``None`` if - the component has not failed. - """ - return self._exception - - def do_work(self): - raise NotImplementedError - - def init_zmq(self): - self.zmq = zmq - self.context = self.zmq.Context() - self.zmq_poller = self.zmq.Poller - # The the process title so you can watch it in top - setproctitle(self.__class__.__name__) - return def _run(self): """ @@ -174,33 +116,32 @@ class Component(object): The core logic of the all components is run here. """ + # The process title so you can watch it in top, ps. + setproctitle(self.gen_func.__name__) + log.info("Start %r" % self) log.info("Pid %s" % os.getpid()) log.info("Group %s" % os.getpgrp()) - self.start_tic = time.time() - - self.done = False # TODO: use state flag self.sockets = [] - self.init_zmq() - - self.setup_poller() - - self.setup_control() self.open() self.signal_ready() self.lock_ready() self.wait_ready() + # ----------------------- # YOU SHALL NOT PASS!!!!! # ----------------------- - # ... until the controller signals GO + # ... until the monitor signals GO - self.loop() + for event in self.generator: + self.heartbeat() + msg = self.frame(event) + self.out_socket.send(msg) - self.stop_tic = time.time() + self.signal_done() def run(self, catch_exceptions=True): """ @@ -214,119 +155,10 @@ class Component(object): else: # if we get a kill signal, forcibly close all the # sockets. - # exc_info = sys.exc_info() - # self.relay_exception(exc_info[0], exc_info[1], exc_info[2]) self.teardown_sockets() - finally: - self.shutdown() log.info("Exiting %r" % self) - def working(self): - """ - Controls when the work loop will start and end - - If we encounter an exception or signal done exit. - - Overload for higher order behavior. - """ - return (not self.done) - - def loop(self, lockstep=True): - """ - Loop to do work while we still have work to do. - """ - while self.working(): - self.heartbeat() - self.do_work() - - def runtime(self): - if self.ready() and self.start_tic and self.stop_tic: - return self.stop_tic - self.start_tic - - def heartbeat(self, timeout=0): - # wait for synchronization reply from the host - self.socks = dict(self.poll.poll(timeout)) - - # ---------------- - # Control Dispatch - # ---------------- - assert self.control_in, 'Component does not have a control_in socket' - - # If we're in devel mode drop out because the controller - # isn't guaranteed to be around anymore - if self.devel: - log.warn("Skipping heartbeat because of devel flag") - return - - if self.socks.get(self.control_in) == zmq.POLLIN: - msg = self.control_in.recv() - event, payload = CONTROL_UNFRAME(msg) - - # =========== - # Heartbeat - # =========== - - # The controller will send out a single number packed in - # a CONTROL_FRAME with ``heartbeat`` event every - # (n)-seconds. The component then has n seconds to - # respond to it. If not then it will be considered as - # malfunctioning or maybe CPU bound. - - if event == CONTROL_PROTOCOL.HEARTBEAT: - # Heart outgoing - heartbeat_frame = CONTROL_FRAME( - CONTROL_PROTOCOL.OK, - payload - ) - - self.last_ping = float(payload) - # Echo back the heartbeat identifier to tell the - # controller that this component is still alive and - # doing work - self.control_out.send(heartbeat_frame) - - - # ========= - # Soft Kill - # ========= - - # Try and clean up properly and send out any reports or - # data that are done during a clean shutdown. Inform the - # controller that we're done. - elif event == CONTROL_PROTOCOL.SHUTDOWN: - self.signal_done() - - # ========= - # Hard Kill - # ========= - - # Just exit. - elif event == CONTROL_PROTOCOL.KILL: - self.kill() - - # In case we didn't receive a ping, send a pre-emptive - # pong to the monitor. - elif hasattr(self, 'control_out') and \ - self.last_ping and \ - time.time() - self.last_ping > 1: - # send a ping ahead of schedule - pre_pong = time.time() - heartbeat_frame = CONTROL_FRAME( - CONTROL_PROTOCOL.OK, - str(pre_pong) - ) - - # Echo back the heartbeat identifier to tell the - # controller that this component is still alive and - # doing work - self.control_out.send(heartbeat_frame, self.zmq.NOBLOCK) - self.last_ping = pre_pong - elif self.last_ping and \ - time.time() - self.last_ping > PARAMETERS.MAX_COMPONENT_WAIT: - # monitor is gone without sending the shutdown - # signal, do a hard exit. - self.kill() # ---------------------------- # Cleanup & Modes of Failure @@ -342,16 +174,6 @@ class Component(object): for sock in self.sockets: sock.close() - def shutdown(self): - """ - Clean shutdown. - - Tear down after normal operation. - """ - if self.on_done: - log.warn("{id} calling done.".format(id=self.get_id)) - self.on_done() - def kill(self): """ Unclean shutdown. @@ -359,9 +181,64 @@ class Component(object): Tear down ( fast ) as a mode of failure in the simulation or on service halt. """ - # sys.exit(1) raise KillSignal() + def signal_exception(self, exc=None, scope=None): + """ + All exceptions inside any component should boil back to + this handler. + + Will inform the system that the component has failed and how it + has failed. + """ + self.state_flag = COMPONENT_STATE.EXCEPTION + exc_type, exc_value, exc_traceback = sys.exc_info() + + # if a downstream component fails, this component may try + # sending when there are zero connections to the socket, + # which will raise ZMQError(EAGAIN). So, it doesn't make + # sense to relay this exception to Monitor and the rest + # of the zipline. + if isinstance(exc, zmq.ZMQError) and exc.errno == zmq.EAGAIN: + log.warn("{id} raised a ZMQError(EAGAIN) not relaying"\ + .format(id=self.get_id)) + return + + # sys.stdout.write(trace) + log.exception("Unexpected error in run for {id}.".format(id=self.get_id)) + + try: + log.info('{id} sending exception to monitor'\ + .format(id=self.get_id)) + msg = EXCEPTION_FRAME( + exc_traceback, + exc_type.__name__, + exc_value.message + ) + + exception_frame = CONTROL_FRAME( + CONTROL_PROTOCOL.EXCEPTION, + msg + ) + self.control_out.send(exception_frame, self.zmq.NOBLOCK) + # The monitor should relay the exception back + # to all zipline components. Wait here until the + # notice arrives, and we can assume other zipline + # components have broken out of their message + # loops. + for i in xrange(PARAMETERS.MAX_COMPONENT_WAIT): + self.heartbeat(timeout=1000) + log.warn("{id} never heard back from monitor."\ + .format(id=self.get_id)) + + except KillSignal: + log.info("{id} received confirmation from monitor"\ + .format(id=self.get_id)) + except: + log.exception("Exception waiting for monitor reply") + + + # ---------------------- # Internal Maintenance # ---------------------- @@ -405,7 +282,7 @@ class Component(object): # Go # ==== - # A distributed lock from the controller to ensure + # A distributed lock from the monitor to ensure # synchronized start. if event == CONTROL_PROTOCOL.HEARTBEAT: @@ -417,7 +294,7 @@ class Component(object): log.info('Prestart Heartbeat ' + self.get_id) elif event == CONTROL_PROTOCOL.GO: - # Side effectful call from the controller to unlock + # Side effectful call from the monitor to unlock # and begin doing work only when the entire topology # of the system beings to come online log.info('Unlocking ' + self.__class__.__name__) @@ -429,7 +306,7 @@ class Component(object): # Try and clean up properly and send out any reports or # data that are done during a clean shutdown. Inform the - # controller that we're done. + # monitor that we're done. elif event == CONTROL_PROTOCOL.SHUTDOWN: self.signal_done() break @@ -449,97 +326,90 @@ class Component(object): self.kill() break + def heartbeat(self, timeout=0): + # wait for synchronization reply from the host + socks = dict(self.poll.poll(timeout)) + + # ---------------- + # Control Dispatch + # ---------------- + assert self.control_in, 'Component does not have a control_in socket' + + if socks.get(self.control_in) == zmq.POLLIN: + msg = self.control_in.recv() + event, payload = CONTROL_UNFRAME(msg) + + # =========== + # Heartbeat + # =========== + + # The monitor will send out a single number packed in + # a CONTROL_FRAME with ``heartbeat`` event every + # (n)-seconds. The component then has n seconds to + # respond to it. If not then it will be considered as + # malfunctioning or maybe CPU bound. + + if event == CONTROL_PROTOCOL.HEARTBEAT: + # Heart outgoing + heartbeat_frame = CONTROL_FRAME( + CONTROL_PROTOCOL.OK, + payload + ) + + self.last_ping = float(payload) + # Echo back the heartbeat identifier to tell the + # monitor that this component is still alive and + # doing work + self.control_out.send(heartbeat_frame) + + + # ========= + # Soft Kill + # ========= + + # Try and clean up properly and send out any reports or + # data that are done during a clean shutdown. Inform the + # monitor that we're done. + elif event == CONTROL_PROTOCOL.SHUTDOWN: + self.signal_done() + + # ========= + # Hard Kill + # ========= + + # Just exit. + elif event == CONTROL_PROTOCOL.KILL: + self.kill() + + # In case we didn't receive a ping, send a pre-emptive + # pong to the monitor. + elif self.last_ping and time.time() - self.last_ping > 1: + # send a ping ahead of schedule + pre_pong = time.time() + heartbeat_frame = CONTROL_FRAME( + CONTROL_PROTOCOL.OK, + str(pre_pong) + ) + + # Echo back the heartbeat identifier to tell the + # monitor that this component is still alive and + # doing work + self.control_out.send(heartbeat_frame, self.zmq.NOBLOCK) + self.last_ping = pre_pong + elif self.last_ping and \ + time.time() - self.last_ping > PARAMETERS.MAX_COMPONENT_WAIT: + # monitor is gone without sending the shutdown + # signal, do a hard exit. + self.kill() + def signal_ready(self): log.info(self.__class__.__name__ + ' is ready') - - if hasattr(self, 'control_out'): - frame = CONTROL_FRAME( - CONTROL_PROTOCOL.READY, - '' - ) - self.control_out.send(frame) - - def signal_cancel(self): - self.done = True - - # TODO: no hasattr hacks - #if not self.controller: - if hasattr(self, 'control_out'): - frame = CONTROL_FRAME( - CONTROL_PROTOCOL.SHUTDOWN, - None - ) - self.control_out.send(frame) - - # then proceeds to do shutdown(), and teardown_sockets() - # to complete the process - - def signal_exception(self, exc=None, scope=None): - """ - All exceptions inside any component should boil back to - this handler. - - Will inform the system that the component has failed and how it - has failed. - """ - - if scope == 'algo': - self.error_state = COMPONENT_FAILURE.ALGOEXCEPT - else: - self.error_state = COMPONENT_FAILURE.HOSTEXCEPT - - self.state_flag = COMPONENT_STATE.EXCEPTION - # mark the time of failure so we can track the failure - # progogation through the system. - - self.stop_tic = time.time() - - self._exception = exc - exc_type, exc_value, exc_traceback = sys.exc_info() - - # if a downstream component fails, this component may try - # sending when there are zero connections to the socket, - # which will raise ZMQError(EAGAIN). So, it doesn't make - # sense to relay this exception to Monitor and the rest - # of the zipline. - if isinstance(exc, zmq.ZMQError) and exc.errno == zmq.EAGAIN: - log.warn("{id} raised a ZMQError(EAGAIN) not relaying"\ - .format(id=self.get_id)) - return - - # sys.stdout.write(trace) - log.exception("Unexpected error in run for {id}.".format(id=self.get_id)) - - if hasattr(self, 'control_out') and self.control_out: - try: - log.info('{id} sending exception to controller'\ - .format(id=self.get_id)) - msg = EXCEPTION_FRAME( - exc_traceback, - exc_type.__name__, - exc_value.message - ) - - exception_frame = CONTROL_FRAME( - CONTROL_PROTOCOL.EXCEPTION, - msg - ) - self.control_out.send(exception_frame, self.zmq.NOBLOCK) - # The controller should relay the exception back - # to all zipline components. Wait here until the - # notice arrives, and we can assume other zipline - # components have broken out of their message - # loops. - for i in xrange(PARAMETERS.MAX_COMPONENT_WAIT): - self.heartbeat(timeout=1000) - log.warn("{id} never heard back from monitor."\ - .format(id=self.get_id)) - except KillSignal: - log.info("{id} received confirmation from controller"\ - .format(id=self.get_id)) - except: - log.exception("Exception waiting for controller reply") + frame = CONTROL_FRAME( + CONTROL_PROTOCOL.READY, + '' + ) + self.control_out.send(frame) def signal_done(self): """ @@ -550,20 +420,18 @@ class Component(object): # notify internal work loop that we're done self.done = True # TODO: use state flag - if hasattr(self, 'out_socket') and self.out_socket: - msg = zmq.Message(str(CONTROL_PROTOCOL.DONE)) - self.out_socket.send(msg) + msg = zmq.Message(str(CONTROL_PROTOCOL.DONE)) + self.out_socket.send(msg) - if hasattr(self, 'control_out'): - # notify controller we're done - done_frame = CONTROL_FRAME( - CONTROL_PROTOCOL.DONE, - '' - ) + # notify monitor we're done + done_frame = CONTROL_FRAME( + CONTROL_PROTOCOL.DONE, + '' + ) - self.control_out.send(done_frame) - log.info("[%s] sent control done" % self.get_id) + self.control_out.send(done_frame) + log.info("[%s] sent control done" % self.get_id) # there is a narrow race condition where we finish just # after the Monitor accepts our prior heartbeat, but just @@ -571,43 +439,64 @@ class Component(object): # last heartbeat, and wait an unusually long time. self.heartbeat(timeout=5000) - - - # ----------- # Messaging # ----------- - def setup_poller(self): + def open(self): """ - Setup the poller used for multiplexing the incoming data - handling sockets. + Open the connections needed to start doing work. + Perform any setup that must be done within process. """ - self.poll = self.zmq_poller() - def bind_data(self): - return self.bind_pull_socket(self.addresses['data_address']) + self.zmq = zmq + self.context = self.zmq.Context() + self.poll = self.zmq.Poller() - def connect_data(self): - return self.connect_push_socket(self.addresses['data_address']) + self.setup_control() - def bind_feed(self): - return self.bind_pub_socket(self.addresses['feed_address']) + if self.in_socket_args: + self.in_socket = self.open_socket(self.in_socket_args) + poller_gen = gen_from_poller( + self.poller, + self.in_socket, + self.unframe + ) + self.generator = self.gen_func( + poller_gen, + *self.gen_args, + **self.gen_kwargs + ) + else: + self.generator = self.gen_func(*self.gen_args, **self.gen_kwargs) - def connect_feed(self): - return self.connect_sub_socket(self.addresses['feed_address']) + self.out_socket = self.open_socket(self.out_socket_args) - def bind_merge(self): - return self.bind_pull_socket(self.addresses['merge_address']) + def open_socket(self, sock_args): + if sock_args.bind: + return self.bind_socket(sock_args) + else: + return self.connect_socket(sock_args) - def connect_merge(self): - return self.connect_push_socket(self.addresses['merge_address']) + def bind_socket(self, sock_args): + if sock_args.style == zmq.PULL: + return self.bind_pull_socket(sock_args.uri) + if sock_args.style == zmq.PUSH: + return self.bind_push_socket(sock_args.uri) + if sock_args.style == zmq.PUB: + return self.bind_pub_socket(sock_args.uri) - def bind_result(self): - return self.bind_push_socket(self.addresses['results_address']) + raise Exception("Invalid socket arguments") - def connect_result(self): - return self.connect_pull_socket(self.addresses['results_address']) + def connect_socket(self, sock_args): + if sock_args.style == zmq.PULL: + return self.connect_pull_socket(sock_args.uri) + if sock_args.style == zmq.PUSH: + return self.connect_push_socket(sock_args.uri) + if sock_args.style == zmq.SUB: + return self.connect_sub_socket(sock_args.uri) + + raise Exception("Invalid socket arguments") def bind_push_socket(self, addr): push_socket = self.context.socket(self.zmq.PUSH) @@ -638,7 +527,6 @@ class Component(object): def connect_push_socket(self, addr): push_socket = self.context.socket(self.zmq.PUSH) push_socket.connect(addr) - #push_socket.setsockopt(self.zmq.LINGER,0) self.sockets.append(push_socket) self.out_socket = push_socket @@ -647,7 +535,6 @@ class Component(object): def bind_pub_socket(self, addr): pub_socket = self.context.socket(self.zmq.PUB) pub_socket.bind(addr) - #pub_socket.setsockopt(self.zmq.LINGER, 0) self.out_socket = pub_socket return pub_socket @@ -668,92 +555,29 @@ class Component(object): of the simulation and to forcefully tear down the simulation in case of a failure. """ - - # Allow for the possibility of not having a controller, - # possibly the zipline devsimulator may not want this. - if not self.controller: - return - - self.control_out = self.controller.message_sender( + self.control_out = self.monitor.message_sender( identity = self.get_id, context = self.context, ) - self.control_in = self.controller.message_listener( + self.control_in = self.monitor.message_listener( context = self.context ) self.poll.register(self.control_in, self.zmq.POLLIN) self.sockets.extend([self.control_in, self.control_out]) - # ----------- - # FSM Actions - # ----------- - - #@property - #def state(self): - #if not hasattr(self, '_state'): - #self._state = self.initial_state - #else: - #return self._state - - #@state.setter - #def state(self, new): - #if not hasattr(self, '_state'): - #self._state = self.initial_state - - #old = self._state - - #if (old, new) in self.workflow: - #self._state = new - #else: - #raise RuntimeError("Invalid State Transition : %s -> %s" %(old, new)) - # --------------------- # Description and Debug # --------------------- - def extern_logger(self): - """ - Pipe logs out to a provided logging interface. - """ - pass - - def setup_extern_logger(self): - """ - Pipe logs out to a provided logging interface. - """ - pass - @property def get_id(self): """ - The descriptive name of the component. + The time invariant name for this component. + Must be unique within this zipline. """ - # Prevents the bug that Thomas ran into - raise NotImplementedError - - @property - def get_type(self): - """ - The data flow type of the component. - - - ``SOURCE`` - - ``CONDUIT`` - - ``SINK`` - - """ - raise NotImplementedError - - @property - def get_pure(self): - """ - Describes whehter this component purely functional, i.e. for a - given set of inputs is it guaranteed to always give the same - output . Components that are side-effectful are, generally, not - pure. - """ - return False + return self.component_id def debug(self): """ @@ -766,18 +590,12 @@ class Component(object): 'pid' : os.getpid() , 'memaddress' : hex(id(self)) , 'ready' : self.successful() , - 'succesfull' : self.ready() , + 'successful' : self.ready() , } - def __len__(self): - """ - Some components overload this for debug purposes - """ - raise NotImplementedError - def __repr__(self): """ - Return a usefull string representation of the component to + Return a useful string representation of the component to indicate its type, unique identifier, and computational context identifier name. """ diff --git a/zipline/core/host.py b/zipline/core/host.py index ea1ca0aa..37de82aa 100644 --- a/zipline/core/host.py +++ b/zipline/core/host.py @@ -103,7 +103,7 @@ class ComponentHost(object): log.info('== Roll Call ==') - log.info('Controller') + log.info('Monitor') self.launch_controller() diff --git a/zipline/core/monitor.py b/zipline/core/monitor.py index ff6adf0c..183de45b 100644 --- a/zipline/core/monitor.py +++ b/zipline/core/monitor.py @@ -38,7 +38,7 @@ class UnknownChatter(Exception): return """Component calling itself "%s" talking on unexpected channel""" % self.named -log = logbook.Logger('Controller') +log = logbook.Logger('Monitor') # The scalars determining the timing of the monitor behavior for # the system. @@ -56,7 +56,7 @@ PARAMETERS = ndict(dict( SYSTEM_TIMEOUT = 50, )) -class Controller(object): +class Monitor(object): """ A N to M messaging system for inter component communication. @@ -103,8 +103,6 @@ class Controller(object): self.route_socket = route_socket self.exception_socket = exception_socket - self.error_replay = OrderedDict() - self.missed_beats = Counter() self.send_sighup = send_sighup @@ -499,7 +497,6 @@ class Controller(object): # Error Handling # -------------- def exception(self, component, exception_data): - self.error_replay[(component, time.time())] = exception_data log.error('Component in exception state: %s. Shutting down system and sending exception data to listeners.'\ % component) # Send the exception message out to listeners. @@ -616,10 +613,6 @@ class Controller(object): self.associated.append(s) return s - def do_error_replay(self): - for (component, time), error in self.error_replay.iteritems(): - log.info('Component Log for -- %s --:\n%s' % (component, error)) - def kill(self): """Aggressively exit the whole zipline. """ diff --git a/zipline/core/process.py b/zipline/core/process.py index b2a01429..dc1fcd3c 100644 --- a/zipline/core/process.py +++ b/zipline/core/process.py @@ -40,13 +40,13 @@ class ProcessSimulator(ComponentHost): # invoked by the host's open() def launch_controller(self): - proc = multiprocessing.Process(target=self.controller.run) + proc = multiprocessing.Process(target=self.monitor.run) proc.start() self.con = proc # Process specific - self.controller_process = proc - self.mapping[proc.pid] = 'Controller' + self.monitor_process = proc + self.mapping[proc.pid] = 'Monitor' def launch_component(self, component): proc = multiprocessing.Process(target=component.run) @@ -81,7 +81,7 @@ class ProcessSimulator(ComponentHost): process.join(timeout=1) process.terminate() - self.controller.shutdown(soft=True) + self.monitor.shutdown(soft=True) self.running = False self.con.terminate() diff --git a/zipline/gens/utils.py b/zipline/gens/utils.py index d51452bf..d76966f5 100644 --- a/zipline/gens/utils.py +++ b/zipline/gens/utils.py @@ -19,8 +19,8 @@ def mock_raw_event(sid, dt): def mock_done(id): return ndict({ - 'dt' : "DONE", - "source_id" : id, + 'dt' : "DONE", + "source_id" : id, 'tnfm_id' : id, 'tnfm_value': None, 'type' : 0 @@ -43,7 +43,7 @@ def roundrobin(sources, namestrings): """ assert len(sources) == len(namestrings) mapping = OrderedDict(zip(namestrings, sources)) - + # While our generators have not been exhausted, pull elements while mapping.keys() != []: for namestring, source in mapping.iteritems(): diff --git a/zipline/gens/zmqgen.py b/zipline/gens/zmqgen.py new file mode 100644 index 00000000..e51e3bab --- /dev/null +++ b/zipline/gens/zmqgen.py @@ -0,0 +1,27 @@ +import zmq +import zipline.protocol as zp + +def gen_from_pull_socket(socket_uri, context, unframe): + """ + A generator that takes a socket_uri, and yields + messages from the poller until it gets a zp.CONTROL_PROTOCOL.DONE. + """ + pull_socket = context.socket(zmq.PULL) + pull_socket.connect(socket_uri) + poller = zmq.Poller() + poller.register(pull_socket, zmq.POLLIN) + + return gen_from_poller(poller, pull_socket, unframe) + +def gen_from_poller(poller, in_socket, unframe): + + while True: + socks = dict(poller.poll(1000)) + + if socks.get(in_socket) == zmq.POLLIN: + message = in_socket.recv() + if message == str(zp.CONTROL_PROTOCOL.DONE): + break + else: + event = unframe(message) + yield event diff --git a/zipline/lines.py b/zipline/lines.py index 6a70bc14..3d3ab2a6 100644 --- a/zipline/lines.py +++ b/zipline/lines.py @@ -70,7 +70,7 @@ from zipline.transforms import BaseTransform from zipline.test_algorithms import TestAlgorithm from zipline.components import TradeSimulationClient from zipline.core.process import ProcessSimulator -from zipline.core.monitor import Controller +from zipline.core.monitor import Monitor from zipline.finance.trading import SIMULATION_STYLE log = logbook.Logger('Lines') @@ -131,7 +131,7 @@ class SimulatedTrading(object): 'results_address' : sockets[4], } - self.con = Controller( + self.monitor = Monitor( # pub socket sockets[5], # route socket @@ -163,7 +163,7 @@ class SimulatedTrading(object): #setup transforms self.transforms = {} - self.sim.register_controller( self.con ) + self.sim.register_monitor( self.monitor ) @staticmethod @@ -348,15 +348,15 @@ class SimulatedTrading(object): return base | transforms | sources - def setup_controller(self): + def setup_monitor(self): """ - Prepare the controller to manage the topology specified + Prepare the monitor to manage the topology specified by this line. """ - self.con.manage(self.topology) + self.monitor.manage(self.topology) def simulate(self, blocking=True): - self.setup_controller() + self.setup_monitor() self.started = True self.sim_context = self.sim.simulate() diff --git a/zipline/utils/test_utils.py b/zipline/utils/test_utils.py index 6655c265..eda5a133 100644 --- a/zipline/utils/test_utils.py +++ b/zipline/utils/test_utils.py @@ -1,3 +1,4 @@ +import multiprocessing import zmq import time import zipline.protocol as zp @@ -6,6 +7,7 @@ import blist from zipline.utils.date_utils import EPOCH from itertools import izip from logbook import FileHandler +from zipline.core.monitor import Monitor def setup_logger(test, path='/var/log/zipline/zipline.log'): test.log_handler = FileHandler(path) @@ -140,3 +142,33 @@ def assert_single_position(test, zipline): sid, "Portfolio should have one position in " + str(sid) ) + + +def launch_component(component): + proc = multiprocessing.Process(target=component.run) + proc.start() + return proc + +def launch_monitor(monitor): + proc = multiprocessing.Process(target=monitor.run) + proc.start() + return proc + + +def create_monitor(allocator): + sockets = allocator.lease(3) + mon = Monitor( + # pub socket + sockets[0], + # route socket + sockets[1], + # exception socket to match tradesimclient's result + # socket, because we want to relay exceptions to the + # same listener + sockets[2], + # this controller is expected to run in a test, so no + # need to signal the parent process on success or error. + send_sighup=False + ) + + return mon