diff --git a/zipline/components/aggregator.py b/zipline/components/aggregator.py index 8a55e3d9..e4a6fba4 100644 --- a/zipline/components/aggregator.py +++ b/zipline/components/aggregator.py @@ -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): diff --git a/zipline/components/tradesimulation.py b/zipline/components/tradesimulation.py index c3c84efe..8452be49 100644 --- a/zipline/components/tradesimulation.py +++ b/zipline/components/tradesimulation.py @@ -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() diff --git a/zipline/core/component.py b/zipline/core/component.py index 608a3ad1..b2353cc2 100644 --- a/zipline/core/component.py +++ b/zipline/core/component.py @@ -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 diff --git a/zipline/core/monitor.py b/zipline/core/monitor.py index d5b9af53..acf012b3 100644 --- a/zipline/core/monitor.py +++ b/zipline/core/monitor.py @@ -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() diff --git a/zipline/finance/risk.py b/zipline/finance/risk.py index 8e687c49..8e52ad3d 100644 --- a/zipline/finance/risk.py +++ b/zipline/finance/risk.py @@ -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) diff --git a/zipline/transforms/base.py b/zipline/transforms/base.py index f2161a76..9a232ad9 100644 --- a/zipline/transforms/base.py +++ b/zipline/transforms/base.py @@ -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)