From c02d15016a161f3136ce21c771c68ad02a345073 Mon Sep 17 00:00:00 2001 From: fawce Date: Tue, 24 Jul 2012 23:43:40 -0400 Subject: [PATCH] added timeouts for component when waiting to hear from the monitor. proof of concept exception relay for algorithm's initialize method. --- tests/test_exception_handling.py | 20 ++++-- tests/test_finance.py | 3 - zipline/components/tradesimulation.py | 4 +- zipline/core/component.py | 95 +++++++++++++++++++-------- zipline/core/monitor.py | 9 ++- zipline/utils/test_utils.py | 6 ++ 6 files changed, 97 insertions(+), 40 deletions(-) diff --git a/tests/test_exception_handling.py b/tests/test_exception_handling.py index 15ea1871..48472ad4 100644 --- a/tests/test_exception_handling.py +++ b/tests/test_exception_handling.py @@ -1,3 +1,5 @@ +import zmq + from unittest2 import TestCase from collections import defaultdict from logbook.compat import LoggingHandler @@ -7,6 +9,8 @@ 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 + DEFAULT_TIMEOUT = 15 # seconds EXTENDED_TIMEOUT = 90 @@ -21,8 +25,11 @@ class FinanceTestCase(TestCase): self.zipline_test_config = { 'allocator' : allocator, 'sid' : 133, - 'devel' : True + 'devel' : False, + 'results_socket' : allocator.lease(1)[0] } + self.ctx = zmq.Context() + self.log_handler = LoggingHandler() self.log_handler.push_application() @@ -37,12 +44,17 @@ class FinanceTestCase(TestCase): self.zipline_test_config['simulation_style'] = \ SIMULATION_STYLE.FIXED_SLIPPAGE self.zipline_test_config['algorithm'] = ExceptionAlgorithm('initialize') - self.zipline_test_config['devel'] = False zipline = SimulatedTrading.create_test_zipline( **self.zipline_test_config ) - zipline.simulate(blocking=True) + output, _ = drain_zipline(self, zipline) + self.assertEqual(output, ['EXCEPTION']) self.assertTrue(zipline.sim.ready()) - self.assertTrue(zipline.sim.exception) + self.assertFalse(zipline.sim.exception) + + # TODO: + # - exception protocol to use prefix/payload as EXCEPT, + # and the stack trace + # - test exception in handle_data diff --git a/tests/test_finance.py b/tests/test_finance.py index 37686805..db77f3ec 100644 --- a/tests/test_finance.py +++ b/tests/test_finance.py @@ -120,7 +120,6 @@ class FinanceTestCase(TestCase): self.zipline_test_config['order_count'] = 100 self.zipline_test_config['trade_count'] = 200 zipline = SimulatedTrading.create_test_zipline(**self.zipline_test_config) - zipline.simulate(blocking=False) assert_single_position(self, zipline) @@ -145,8 +144,6 @@ class FinanceTestCase(TestCase): **self.zipline_test_config ) - zipline.simulate(blocking=False) - output, transaction_count = drain_zipline(self, zipline) self.assertTrue(zipline.sim.ready()) diff --git a/zipline/components/tradesimulation.py b/zipline/components/tradesimulation.py index 9fcefdd7..11b170f6 100644 --- a/zipline/components/tradesimulation.py +++ b/zipline/components/tradesimulation.py @@ -186,8 +186,8 @@ class TradeSimulationClient(Component): self.do_op(self.algorithm.handle_data, data) def exception_callback(self, trace): - log.info(trace) - pass + if self.results_socket: + self.out_socket.send("EXCEPTION") def do_op(self, callable_op, *args, **kwargs): """ Wrap a callable operation with the zmq logbook diff --git a/zipline/core/component.py b/zipline/core/component.py index 2b8d425b..b0938b1f 100644 --- a/zipline/core/component.py +++ b/zipline/core/component.py @@ -19,6 +19,8 @@ import gevent_zeromq # zmq_ctypes #import zmq_ctypes +from zipline.core.monitor import PARAMETERS + from zipline.utils.gpoll import _Poller as GeventPoller from zipline.protocol import CONTROL_PROTOCOL, COMPONENT_STATE, \ COMPONENT_FAILURE, CONTROL_FRAME, CONTROL_UNFRAME @@ -157,7 +159,6 @@ class Component(object): ZMQ in all flavors. Have it your way. mp - Distinct contexts | pyzmq - thread - Same context | pyzmq green - Same context | gevent_zeromq pypy - Same context | zmq_ctypes @@ -170,11 +171,6 @@ class Component(object): # The the process title so you can watch it in top setproctitle(self.__class__.__name__) 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() @@ -214,6 +210,7 @@ class Component(object): self.signal_ready() self.lock_ready() + self.wait_ready() # ----------------------- # YOU SHALL NOT PASS!!!!! @@ -341,6 +338,11 @@ class Component(object): # doing work self.control_out.send(heartbeat_frame) 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 @@ -371,7 +373,7 @@ class Component(object): Tear down ( fast ) as a mode of failure in the simulation or on service halt. """ - raise NotImplementedError + sys.exit(1) # ---------------------- # Internal Maintenance @@ -399,33 +401,69 @@ class Component(object): # in the locked quasimode. Respond to HEARTBEAT and GO # messages. + start_wait = time.time() + while self.waiting: - #socks = dict(self.poll.poll(self.heartbeat_timeout)) + socks = dict(self.poll.poll(100)) - msg = self.control_in.recv() - event, payload = CONTROL_UNFRAME(msg) + assert self.control_in, \ + 'Component does not have a control_in socket' - # ==== - # Go - # ==== + if socks.get(self.control_in) == zmq.POLLIN: - # A distributed lock from the controller to ensure - # synchronized start. + msg = self.control_in.recv() + event, payload = CONTROL_UNFRAME(msg) - if event == CONTROL_PROTOCOL.HEARTBEAT: - heartbeat_frame = CONTROL_FRAME( - CONTROL_PROTOCOL.OK, - payload - ) - self.control_out.send(heartbeat_frame) - log.info('Prestart Heartbeat ' + self.get_id) + # ==== + # Go + # ==== + + # A distributed lock from the controller to ensure + # synchronized start. + + if event == CONTROL_PROTOCOL.HEARTBEAT: + heartbeat_frame = CONTROL_FRAME( + CONTROL_PROTOCOL.OK, + payload + ) + self.control_out.send(heartbeat_frame) + log.info('Prestart Heartbeat ' + self.get_id) + + elif event == CONTROL_PROTOCOL.GO: + # Side effectful call from the controller to unlock + # and begin doing work only when the entire topology + # of the system beings to come online + log.info('Unlocking ' + self.__class__.__name__) + self.unlock_ready() + + # ========= + # 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() + self.shutdown() + break + + # ========= + # Hard Kill + # ========= + + # Just exit. + elif event == CONTROL_PROTOCOL.KILL: + self.kill() + break + + elif time.time() - start_wait > PARAMETERS.MAX_COMPONENT_WAIT: + log.info('No go signal from monitor, %s exiting' \ + % self.__class__.__name__) + self.kill() + break - elif event == CONTROL_PROTOCOL.GO: - # Side effectful call from the controller to unlock - # and begin doing work only when the entire topology - # of the system beings to come online - log.info('Unlocking ' + self.__class__.__name__) - self.unlock_ready() def signal_ready(self): log.info(self.__class__.__name__ + ' is ready') @@ -459,7 +497,6 @@ class Component(object): Will inform the system that the component has failed and how it has failed. """ - if scope == 'algo': self.error_state = COMPONENT_FAILURE.ALGOEXCEPT else: diff --git a/zipline/core/monitor.py b/zipline/core/monitor.py index c9c66fda..94866f0b 100644 --- a/zipline/core/monitor.py +++ b/zipline/core/monitor.py @@ -42,7 +42,11 @@ log = logbook.Logger('Controller') # the system. PARAMETERS = ndict(dict( - GENERATIONAL_PERIOD = 30, #seconds + # time Monitor will wait for a heartbeat, in seconds + GENERATIONAL_PERIOD = 10, + # time Component will wait for GO and for a heartbeat before + # timing out. + MAX_COMPONENT_WAIT = 20, ALLOWED_SKIPPED_HEARTBEATS = 10, ALLOWED_INVALID_HEARTBEATS = 3, PRESTART_HEARBEATS = 3, @@ -523,7 +527,7 @@ class Controller(object): Shutdown the system on failure. """ log.error('System in exception state, shutting down') - self.shutdown(soft=True) + self.shutdown(hard=True, soft=False) def exception(self, component, failure): universal = self.exception_universal @@ -650,6 +654,7 @@ class Controller(object): if hard and not self.devel: self.state = CONTROL_STATES.TERMINATE log.info('Hard Shutdown') + self.send_hardkill() if soft and not self.devel: self.state = CONTROL_STATES.TERMINATE diff --git a/zipline/utils/test_utils.py b/zipline/utils/test_utils.py index cf9924da..12139382 100644 --- a/zipline/utils/test_utils.py +++ b/zipline/utils/test_utils.py @@ -10,12 +10,18 @@ def drain_zipline(test, zipline): test.receiver = test.ctx.socket(zmq.PULL) test.receiver.bind(test.zipline_test_config['results_socket']) + # start the simulation + zipline.simulate(blocking=False) + output = [] transaction_count = 0 while True: msg = test.receiver.recv() if msg == str(zp.CONTROL_PROTOCOL.DONE): break + elif msg == "EXCEPTION": + output.append(msg) + break else: update = zp.BT_UPDATE_UNFRAME(msg) output.append(update)