wassname/catalyst
1
0
Fork 0
mirror of https://github.com/wassname/catalyst.git synced 2026-08-11 11:16:15 +08:00
Code Issues Packages Projects Releases Wiki Activity

added timeouts for component when waiting to hear from the monitor. proof of concept exception relay for algorithm's initialize method.

Browse Source
This commit is contained in:
fawce
2012-07-24 23:43:40 -04:00
parent 6520046aea
commit c02d15016a
6 changed files with 97 additions and 40 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_exception_handling.py
+16 -4
View File
@@ -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
tests/test_finance.py
-3
View File
@@ -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())
zipline/components/tradesimulation.py
+2 -2
View File
@@ -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
zipline/core/component.py
+66 -29
View File
@@ -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:
zipline/core/monitor.py
+7 -2
View File
@@ -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
zipline/utils/test_utils.py
+6
View File
@@ -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)