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

added more frequent heartbeating between requests from monitor. seems to work.

Browse Source
This commit is contained in:
fawce
2012-07-13 12:01:21 -04:00
parent 165d94f784
commit 18cd9a02df
6 changed files with 136 additions and 104 deletions
Download Patch File Download Diff File Expand all files Collapse all files
zipline/components/aggregator.py
+1 -9
View File
@@ -13,7 +13,6 @@ import logbook
import zipline.protocol as zp
from zipline.core.component import Component
from zipline.core.controlled import do_handle_control_events
from zipline.protocol import CONTROL_PROTOCOL, COMPONENT_TYPE
from zipline.transitions import WorkflowMeta
from zipline.utils.protocol_utils import Enum
@@ -65,18 +64,11 @@ class Aggregate(Component):
# -------------
def do_work(self):
# wait for synchronization reply from the host
socks = dict(self.poll.poll(self.heartbeat_timeout))
# ----------------
# Control Dispatch
# ----------------
do_handle_control_events(self, socks)
# -------------
# Work Dispatch
# -------------
if socks.get(self.pull_socket) == self.zmq.POLLIN:
if self.socks.get(self.pull_socket) == self.zmq.POLLIN:
message = self.pull_socket.recv()
if message == str(CONTROL_PROTOCOL.DONE):
zipline/components/tradesimulation.py
+1 -10
View File
@@ -11,7 +11,6 @@ from zipline.utils.protocol_utils import ndict
from zipline.utils.log_utils import ZeroMQLogHandler, stdout_only_pipe
from logbook import Logger, NestedSetup, Processor, queues
from zipline.core.controlled import do_handle_control_events
log = logbook.Logger('TradeSimulation')
@@ -78,17 +77,9 @@ class TradeSimulationClient(Component):
self.stdout_capture = stdout_only_pipe #THIS IS A CLASS!
def do_work(self):
# poll all the sockets
socks = dict(self.poll.poll(self.heartbeat_timeout))
# ----------------
# Control Dispatch
# ----------------
do_handle_control_events(self, socks)
# see if the poller has results for the result_feed
if socks.get(self.result_feed) == self.zmq.POLLIN:
if self.socks.get(self.result_feed) == self.zmq.POLLIN:
self.last_msg_dt = datetime.datetime.utcnow()
zipline/core/component.py
+111 -31
View File
@@ -19,17 +19,13 @@ import gevent_zeromq
# zmq_ctypes
#import zmq_ctypes
from zipline.protocol import CONTROL_UNFRAME
from zipline.utils.gpoll import _Poller as GeventPoller
from zipline.protocol import CONTROL_PROTOCOL, COMPONENT_STATE, \
COMPONENT_FAILURE, CONTROL_FRAME
COMPONENT_FAILURE, CONTROL_FRAME, CONTROL_UNFRAME
log = logbook.Logger('Component')
from zipline.exceptions import ComponentNoInit
from zipline.transitions import WorkflowMeta
log = logbook.Logger('Base')
class Component(object):
@@ -84,8 +80,8 @@ class Component(object):
self.killed = False
self.controller = None
# timeout on heartbeat is very short to avoid burning
# cycles on heartbeating.
self.heartbeat_timeout = 60 * 1000
# 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
@@ -99,6 +95,8 @@ class Component(object):
self.note = None
self.confirmed = False
self.devel = False
self.socks = None
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.
@@ -223,33 +221,25 @@ class Component(object):
# ... until the controller signals GO
self.loop()
self.shutdown()
log.info("Shutdown %r" % self)
self.stop_tic = time.time()
def run(self, catch_exceptions=True):
"""
Run the component.
Optionally takes an argument to catch and log all exceptions
raised during execution. Use this with care since it makes it
very hard to debug since it mucks up your stacktraces.
"""
try:
self._run()
except Exception as exc:
exc_info = sys.exc_info()
self.signal_exception(exc)
if catch_exceptions:
try:
self._run()
except Exception as exc:
exc_info = sys.exc_info()
self.signal_exception(exc)
# Reraise the exception
raise exc_info[0], exc_info[1], exc_info[2]
finally:
self.shutdown()
self.teardown_sockets()
# Reraise the exception
raise exc_info[0], exc_info[1], exc_info[2]
finally:
self.shutdown()
self.teardown_sockets()
log.info("Exiting %r" % self)
def working(self):
"""
@@ -266,12 +256,96 @@ class Component(object):
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
# Only run a single iteration here, just before exit.
# This helps ensure that the Monitor
# Running on every iteration ruins performance.
# ----------------
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:
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()
self.shutdown()
# =========
# Hard Kill
# =========
# Just exit.
elif event == CONTROL_PROTOCOL.KILL:
self.kill()
# =========
# Hard Kill
# =========
# Just exit.
elif self.last_ping and time.time() - self.last_ping > 1:
# send a ping ahead of schedule
heartbeat_frame = CONTROL_FRAME(
CONTROL_PROTOCOL.OK,
str(self.last_ping)
)
# Echo back the heartbeat identifier to tell the
# controller that this component is still alive and
# doing work
self.control_out.send(heartbeat_frame)
# ----------------------------
# Cleanup & Modes of Failure
# ----------------------------
@@ -330,7 +404,7 @@ class Component(object):
# messages.
while self.waiting:
socks = dict(self.poll.poll(self.heartbeat_timeout))
#socks = dict(self.poll.poll(self.heartbeat_timeout))
msg = self.control_in.recv()
event, payload = CONTROL_UNFRAME(msg)
@@ -423,23 +497,29 @@ class Component(object):
self.state_flag = COMPONENT_STATE.DONE
if self.out_socket:
log.info("[%s] sending DONE" % self.get_id)
msg = zmq.Message(str(CONTROL_PROTOCOL.DONE))
self.out_socket.send(msg)
log.info("[%s] sent DONE" % self.get_id)
#notify controller we're done
# notify controller 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)
# there is a narrow race condition where we finish just
# after the Monitor accepts our prior heartbeat, but just
# before the next one is sent. So, we hang around for one
# last heartbeat, and wait an unusually long time.
self.heartbeat(timeout=5000)
# notify internal work look that we're done
self.done = True # TODO: use state flag
#log.info("[%s] DONE" % self.get_id)
# -----------
# Messaging
zipline/core/monitor.py
+21 -43
View File
@@ -43,7 +43,7 @@ log = logbook.Logger('Controller')
# the system.
PARAMETERS = ndict(dict(
GENERATIONAL_PERIOD = 1,
GENERATIONAL_PERIOD = 10, #seconds
ALLOWED_SKIPPED_HEARTBEATS = 10,
ALLOWED_INVALID_HEARTBEATS = 3,
PRESTART_HEARBEATS = 3,
@@ -291,28 +291,38 @@ class Controller(object):
# ==============
# Wait the responses
checktime = self.ctime
while self.alive:
socks = dict(poller.poll(self.period))
socks = dict(poller.poll(0))
tic = time.time()
if tic - self.ctime > self.period:
break
# We break out of this loop if the time between
# sending and receiving the heartbeat is more
# than our poll period.
# if tic - self.ctime > self.period:
# break
if socks.get(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 = []
checktime = time.time()
except INVALID_CONTROL_FRAME:
log.error('Invalid frame', rawmessage)
pass
if tic - checktime > self.period:
log.info("heartbeat loop timedout: %s" % (tic - checktime))
log.info(repr(self.responses))
break
# ================
# Heartbeat Stats
# ================
@@ -323,10 +333,10 @@ class Controller(object):
# Topology Status
# ================
# Is the entire topology told us its DONE
# Has the entire topology told us its DONE
done = len(self.finished) == len(self.topology)
# Is the entire topology shown up to the party
# Has the entire topology shown up to the party
complete = len(self.tracked) == len(self.topology)
if complete:
@@ -365,6 +375,7 @@ class Controller(object):
self.signal_hangup()
if not self.alive:
log.info('Breaking out of Monitor Loop')
break
def signal_hangup(self):
@@ -428,27 +439,13 @@ class Controller(object):
for component in bad:
self.fail(component)
if self.debug:
log.info('Bad component %r' % component)
self.missed_beats.update(missing)
for component in missing:
if self.missed_beats[component] >\
PARAMETERS.ALLOWED_SKIPPED_HEARTBEATS:
# TODO: determine when this propogates to a true
# failure, missing one heartbeat could just mean that
# its CPU overloaded
log.warning('Component missed max heartbeats, failing %s'\
% component)
self.fail_universal()
if self.debug:
log.info('Missing component %r' % component)
if self.debug:
#for component in good:
# log.info('good component %r' % component)
for component in self.tracked:
if component not in self.topology:
@@ -458,11 +455,6 @@ class Controller(object):
# Init Handlers
# --------------
def new_source(self):
#if self.state is CONTROL_STATES.RUNNING:
#self.state = SOURCES_READY
pass
def new_universal(self):
pass
@@ -479,9 +471,7 @@ class Controller(object):
log.info('Now Tracking "%s" ' % component)
universal = self.new_universal
init_handlers = {
'FEED' : self.new_source,
}
init_handlers = {}
if component in (self.topology - self.finished) or self.freeform:
init_handlers.get(component, universal)()
@@ -496,7 +486,6 @@ class Controller(object):
# ------------------
def fail_universal(self):
pass
# TODO: this requires higher order functionality
log.error('System in exception state, shutting down')
self.shutdown(soft=True)
@@ -511,13 +500,7 @@ class Controller(object):
if component in (self.topology - self.finished) or self.freeform:
log.warning('Component "%s" missed heartbeat' % component)
self.tracked.remove(component)
# TODO: determine when this propogates to a true
# failure, missing one heartbeat could just mean that
# its CPU overloaded
#log.warning('Component missed max heartbeats, failing %s'\
# % component)
#fail_handlers.get(component, universal)()
fail_handlers.get(component, universal)()
# -------------------
# Completion Handling
@@ -646,7 +629,7 @@ class Controller(object):
def do_error_replay(self):
for (component, time), error in self.error_replay.iteritems():
log.debug('Component Log for -- %s --:\n%s' % (component, error))
log.info('Component Log for -- %s --:\n%s' % (component, error))
def shutdown(self, hard=False, soft=True):
@@ -665,8 +648,3 @@ class Controller(object):
self.state = CONTROL_STATES.TERMINATE
log.info('Soft Shutdown')
self.send_softkill()
#self.do_error_replay()
#self.pub.close()
#self.router.close()
zipline/finance/risk.py
+1 -1
View File
@@ -255,7 +255,7 @@ class RiskMetrics():
cur_return = math.log(1.0 + r) + cur_return
#this is a guard for a single day returning -100%
except ValueError:
log.warn("{cur} return, zeroing the returns".format(cur=cur_return))
log.debug("{cur} return, zeroing the returns".format(cur=cur_return))
cur_return = 0.0
compounded_returns.append(cur_return)
zipline/transforms/base.py
+1 -10
View File
@@ -4,7 +4,6 @@ import zipline.protocol as zp
from zipline.protocol import CONTROL_PROTOCOL, COMPONENT_TYPE, \
CONTROL_FRAME, CONTROL_UNFRAME
from zipline.core.controlled import do_handle_control_events
import logbook
import time
@@ -54,15 +53,7 @@ class BaseTransform(Component):
"""
socks = dict(self.poll.poll(self.heartbeat_timeout))
# ----------------
# Control Dispatch
# ----------------
do_handle_control_events(self, socks)
if self.feed_socket in socks and socks[self.feed_socket] == self.zmq.POLLIN:
if self.feed_socket in self.socks and self.socks[self.feed_socket] == self.zmq.POLLIN:
message = self.feed_socket.recv()
#import msgpack
#event = msgpack.loads(message)