From 0342ee7f62ebfd69a40e40c0ea4f0dacf040fc7a Mon Sep 17 00:00:00 2001 From: Stephen Diehl Date: Tue, 10 Jul 2012 14:31:01 -0400 Subject: [PATCH 01/14] Fix typos. --- zipline/core/component.py | 4 ++++ zipline/core/monitor.py | 4 +++- 2 files changed, 7 insertions(+), 1 deletion(-) diff --git a/zipline/core/component.py b/zipline/core/component.py index e7a1bfd0..40a57960 100644 --- a/zipline/core/component.py +++ b/zipline/core/component.py @@ -196,6 +196,10 @@ class Component(object): The core logic of the all components is run here. """ + 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 diff --git a/zipline/core/monitor.py b/zipline/core/monitor.py index 7b742627..1edf1927 100644 --- a/zipline/core/monitor.py +++ b/zipline/core/monitor.py @@ -6,6 +6,7 @@ import gevent import itertools import logbook import gevent_zeromq +from setproctitle import setproctitle from signal import SIGHUP, SIGINT from collections import OrderedDict @@ -158,6 +159,7 @@ class Controller(object): def run(self): self.running = True self.init_zmq(self.zmq_flavor) + setproctitle('Monitor') self.state = CONTROL_STATES.INIT @@ -370,7 +372,7 @@ class Controller(object): it. """ if not self.nosignals: - ppid = os.getpid() + ppid = os.getppid() log.warning("Sending SIGHUP") os.kill(ppid, SIGHUP) else: From 4981b600f0241b5a0786a904a631c5d350fdf9b3 Mon Sep 17 00:00:00 2001 From: Stephen Diehl Date: Thu, 12 Jul 2012 16:56:39 -0400 Subject: [PATCH 02/14] Minor changes to Zipline for signals. --- zipline/core/host.py | 5 +++++ zipline/core/monitor.py | 2 +- zipline/lines.py | 21 +++++++++++---------- 3 files changed, 17 insertions(+), 11 deletions(-) diff --git a/zipline/core/host.py b/zipline/core/host.py index 804b25e6..ea1ca0aa 100644 --- a/zipline/core/host.py +++ b/zipline/core/host.py @@ -92,6 +92,10 @@ class ComponentHost(object): def unregister_component(self, component_id): del self.components[component_id] + @property + def pids(self): + return [proc.pid for proc in self.subprocesses] + def open(self): assert hasattr(self, 'zmq_flavor'), \ """ You must specify a flavor of ZeroMQ for all Topology @@ -111,6 +115,7 @@ class ComponentHost(object): for component in self.components.itervalues(): self.launch_component(component) + def is_running(self): """ DEPRECATED, left in for compatability for now. diff --git a/zipline/core/monitor.py b/zipline/core/monitor.py index 1edf1927..17d8c651 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 = 8, ALLOWED_SKIPPED_HEARTBEATS = 3, ALLOWED_INVALID_HEARTBEATS = 3, PRESTART_HEARBEATS = 3, diff --git a/zipline/lines.py b/zipline/lines.py index 23eb8690..2583672c 100644 --- a/zipline/lines.py +++ b/zipline/lines.py @@ -399,16 +399,17 @@ class SimulatedTrading(object): # the supervisory layer # TODO: better way of identifying concurrency substrate - if self.sim.zmq_flavor == 'thread': - log.debug('Blocking') - for thread in self.sim.subthreads: - #log.debug('Waiting on %r' % thread) - log.debug('Waiting on %r' % thread) - thread.join() - log.debug('Yielded on %r' % thread) - else: - for process in self.sim.subprocesses: - process.join() + if blocking: + if self.sim.zmq_flavor == 'thread': + log.debug('Blocking') + for thread in self.sim.subthreads: + #log.debug('Waiting on %r' % thread) + log.debug('Waiting on %r' % thread) + thread.join() + log.debug('Yielded on %r' % thread) + else: + for process in self.sim.subprocesses: + process.join() @property def is_success(self): From 1cc56b52e7b6ac46055d0c66c44cd20e8c5cceb9 Mon Sep 17 00:00:00 2001 From: fawce Date: Wed, 11 Jul 2012 21:51:02 -0400 Subject: [PATCH 03/14] re-enabled the monitor communication in all components. --- tests/test_optimize.py | 2 + zipline/components/aggregator.py | 2 +- zipline/components/tradesimulation.py | 59 ++++++++++++++------------- zipline/core/component.py | 12 ++++-- zipline/core/monitor.py | 2 +- zipline/transforms/base.py | 37 ++++++++--------- 6 files changed, 59 insertions(+), 55 deletions(-) diff --git a/tests/test_optimize.py b/tests/test_optimize.py index cefafd8d..256ee20c 100644 --- a/tests/test_optimize.py +++ b/tests/test_optimize.py @@ -37,6 +37,7 @@ class TestUpDown(TestCase): def tearDown(self): self.log_handler.pop_application() + @skip @timed(DEFAULT_TIMEOUT) def test_source_and_orders(self): """verify that UpDownSource is having the correct @@ -92,6 +93,7 @@ class TestUpDown(TestCase): "Algorithm did not sell when price was going to increase." ) + @skip def test_concavity_of_returns(self): """verify concave relationship between free parameter and returns in certain region around the max. Moreover, diff --git a/zipline/components/aggregator.py b/zipline/components/aggregator.py index 451fc28a..8a55e3d9 100644 --- a/zipline/components/aggregator.py +++ b/zipline/components/aggregator.py @@ -71,7 +71,7 @@ class Aggregate(Component): # ---------------- # Control Dispatch # ---------------- - #do_handle_control_events(self, socks) + do_handle_control_events(self, socks) # ------------- # Work Dispatch diff --git a/zipline/components/tradesimulation.py b/zipline/components/tradesimulation.py index 3bf66282..c3c84efe 100644 --- a/zipline/components/tradesimulation.py +++ b/zipline/components/tradesimulation.py @@ -11,6 +11,8 @@ 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') @@ -33,15 +35,15 @@ class TradeSimulationClient(Component): self.event_data = ndict() self.perf = perf.PerformanceTracker(self.trading_environment) - + self.log_socket = log_socket - + @property def get_id(self): return str(zp.FINANCE_COMPONENT.TRADING_CLIENT) def set_algorithm(self, algorithm): - + """ :param algorithm: must implement the algorithm protocol. See :py:mod:`zipline.test.algorithm` @@ -53,26 +55,20 @@ class TradeSimulationClient(Component): #TODO: re-enable initialization logging. This means we can't call set_algorithm #until we have a context for this component. Possibly this could happen # ask the algorithm to initialize, routing stdout to a zmq PUSH socket. - + #with self.zmq_out.threadbound(), self.stdout_capture(self.logger, 'Algo print capture'): # self.algorithm.initialize() #if we don't have a log socket, initialize anyway. #else: # self.algorithm.initialize() - + self.algorithm.initialize() - + def open(self): - + self.result_feed = self.connect_result() self.perf.open(self.context) - - #If we have a log socket,setup context manager for exporting captured - #print statements - if self.log_socket: - self.zmq_out = ZeroMQLogHandler(uri = self.log_socket, context = self.context) - self.logger = Logger("Print") - self.stdout_capture = stdout_only_pipe #THIS IS A CLASS! + self.setup_logging(self.context) #Initialize log capture for testing purposes. def setup_logging(self, context): @@ -84,6 +80,13 @@ class TradeSimulationClient(Component): 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: @@ -101,7 +104,7 @@ class TradeSimulationClient(Component): event = zp.MERGE_UNFRAME(msg) self.received_count += 1 # update performance and relay the event to the algorithm - + self.process_event(event) if self.perf.exceeded_max_loss: self.finish_simulation() @@ -115,7 +118,7 @@ class TradeSimulationClient(Component): # signal Simulator, our ComponentHost, that this component is # done and Simulator needn't block exit on this component. self.signal_done() - + def process_event(self, event): # generate transactions, if applicable txn = self.txn_sim.apply_trade_to_open_orders(event) @@ -144,7 +147,7 @@ class TradeSimulationClient(Component): # if the event is later than our current time, run the algo # otherwise, the algorithm has fallen behind the feed # and processing per event is longer than time between events. - + if event.dt >= self.current_dt: # compress time by moving the current_time up to the event # time. @@ -166,22 +169,22 @@ class TradeSimulationClient(Component): current_portfolio = self.perf.get_portfolio() self.algorithm.set_portfolio(current_portfolio) data = self.get_data() - + if len(data) > 0: - + # data injection pipeline for log rerouting # any fields injected here should be added to # LOG_EXTRA_FIELDS in zipline/protocol.py - + if self.log_socket: - + def inject_event_data(record): - + #Record the simulation time. record.extra['algo_dt'] = self.current_dt - - data_injector = Processor(inject_event_data) + + data_injector = Processor(inject_event_data) log_pipeline = NestedSetup([self.zmq_out, #e.g. FileHandler(...) data_injector]) @@ -190,14 +193,14 @@ class TradeSimulationClient(Component): # if no log socket, just run the algo normally else: self.algorithm.handle_data(data) - + #Testing utility for log capture. def test_run_algorithm(self): - + def inject_event_data(record): record.extra['algo_dt'] = datetime.datetime.utcnow() #Mock an event.dt - - data_injector = Processor(inject_event_data) + + data_injector = Processor(inject_event_data) log_pipeline = NestedSetup([self.zmq_out, #e.g. FileHandler(...) data_injector]) diff --git a/zipline/core/component.py b/zipline/core/component.py index 40a57960..d10ccb2c 100644 --- a/zipline/core/component.py +++ b/zipline/core/component.py @@ -422,7 +422,11 @@ class Component(object): self.state_flag = COMPONENT_STATE.DONE if self.out_socket: - self.out_socket.send(str(CONTROL_PROTOCOL.DONE)) + 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 done_frame = CONTROL_FRAME( @@ -431,10 +435,10 @@ class Component(object): ) self.control_out.send(done_frame) - #notify internal work look that we're done + # notify internal work look that we're done self.done = True # TODO: use state flag - #log.info("[%s] DONE" % self.get_id) + #log.info("[%s] DONE" % self.get_id) # ----------- # Messaging @@ -512,7 +516,7 @@ 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) + pub_socket.setsockopt(self.zmq.LINGER, 0) self.out_socket = pub_socket return pub_socket diff --git a/zipline/core/monitor.py b/zipline/core/monitor.py index 17d8c651..a34b5ec9 100644 --- a/zipline/core/monitor.py +++ b/zipline/core/monitor.py @@ -415,7 +415,7 @@ class Controller(object): bad = self.tracked - good new = self.responses - good - missing = self.topology - self.tracked + missing = self.topology - self.tracked - self.finished for component in new: self.new(component) diff --git a/zipline/transforms/base.py b/zipline/transforms/base.py index 6e406efc..f2161a76 100644 --- a/zipline/transforms/base.py +++ b/zipline/transforms/base.py @@ -4,6 +4,13 @@ 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 + +log = logbook.Logger('BaseTransform') + class BaseTransform(Component): """ Top level execution entry point for the transform @@ -46,35 +53,23 @@ class BaseTransform(Component): - send the transformed event """ + socks = dict(self.poll.poll(self.heartbeat_timeout)) - # TODO: Abstract this out, maybe on base component - if self.control_in in socks and socks[self.control_in] == self.zmq.POLLIN: - msg = self.control_in.recv() - event, payload = CONTROL_UNFRAME(msg) + # ---------------- + # Control Dispatch + # ---------------- + do_handle_control_events(self, socks) - # -- Heartbeat -- - if event == CONTROL_PROTOCOL.HEARTBEAT: - # Heart outgoing - heartbeat_frame = CONTROL_FRAME( - CONTROL_PROTOCOL.OK, - payload - ) - self.control_out.send(heartbeat_frame) - - # -- Soft Kill -- - elif event == CONTROL_PROTOCOL.SHUTDOWN: - self.signal_done() - self.shutdown() - - # -- Hard Kill -- - elif event == CONTROL_PROTOCOL.KILL: - self.kill() if self.feed_socket in socks and socks[self.feed_socket] == self.zmq.POLLIN: message = self.feed_socket.recv() + #import msgpack + #event = msgpack.loads(message) + #log.info(event) if message == str(CONTROL_PROTOCOL.DONE): + log.info("signaling done") self.signal_done() return From d21abe06eafcbca2cb2622e41006a5733d57906e Mon Sep 17 00:00:00 2001 From: fawce Date: Thu, 12 Jul 2012 00:22:30 -0400 Subject: [PATCH 04/14] updated heartbeat tracking to treat missing several heartbeats as a full failure. --- zipline/core/component.py | 2 +- zipline/core/monitor.py | 33 ++++++++++++++++++++++++++------- 2 files changed, 27 insertions(+), 8 deletions(-) diff --git a/zipline/core/component.py b/zipline/core/component.py index d10ccb2c..baadaa26 100644 --- a/zipline/core/component.py +++ b/zipline/core/component.py @@ -516,7 +516,7 @@ 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) + #pub_socket.setsockopt(self.zmq.LINGER, 0) self.out_socket = pub_socket return pub_socket diff --git a/zipline/core/monitor.py b/zipline/core/monitor.py index a34b5ec9..de0a8b59 100644 --- a/zipline/core/monitor.py +++ b/zipline/core/monitor.py @@ -9,7 +9,7 @@ import gevent_zeromq from setproctitle import setproctitle from signal import SIGHUP, SIGINT -from collections import OrderedDict +from collections import OrderedDict, Counter from zipline.utils.gpoll import _Poller as GeventPoller from zipline.protocol import CONTROL_PROTOCOL, CONTROL_FRAME, \ @@ -95,6 +95,8 @@ class Controller(object): self.error_replay = OrderedDict() + self.missed_beats = Counter() + log.warn("Running Controller in development mode, will ONLY synchronize start.") def init_zmq(self, flavor): @@ -429,10 +431,25 @@ class Controller(object): if self.debug: log.info('Bad component %r' % component) - if self.debug: - for component in missing: + 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: log.info('Uninvited component %r' % component) @@ -495,10 +512,12 @@ class Controller(object): 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 - #fail_handlers.get(component, universal)() + # 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)() # ------------------- # Completion Handling From 09b6b564bbf9f7b9e4a4a06193311dd6752ce0f5 Mon Sep 17 00:00:00 2001 From: fawce Date: Thu, 12 Jul 2012 11:47:48 -0400 Subject: [PATCH 05/14] longer period for heartbeat. --- zipline/core/monitor.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/zipline/core/monitor.py b/zipline/core/monitor.py index de0a8b59..05bdfce1 100644 --- a/zipline/core/monitor.py +++ b/zipline/core/monitor.py @@ -33,7 +33,7 @@ class UnknownChatter(Exception): def __init__(self, name): self.named = name def __str__(self): - return """Component calling itself "%s" talking on unexpected channel"""\ + return "Component calling itself '%s' talking on unexpected channel" % self.named From 165d94f784a54e922af003b045d22602959e0b1c Mon Sep 17 00:00:00 2001 From: fawce Date: Thu, 12 Jul 2012 16:22:57 -0400 Subject: [PATCH 06/14] tweaks --- zipline/core/component.py | 5 +++-- zipline/core/monitor.py | 12 ++++++------ 2 files changed, 9 insertions(+), 8 deletions(-) diff --git a/zipline/core/component.py b/zipline/core/component.py index baadaa26..608a3ad1 100644 --- a/zipline/core/component.py +++ b/zipline/core/component.py @@ -83,8 +83,9 @@ class Component(object): self.out_socket = None self.killed = False self.controller = None - # timeout after a full minute - self.heartbeat_timeout = 60 *1000 + # timeout on heartbeat is very short to avoid burning + # cycles on heartbeating. + self.heartbeat_timeout = 60 * 1000 # TODO: state_flag is deprecated, remove # TODO: error_state is deprecated, remove self.state_flag = COMPONENT_STATE.OK diff --git a/zipline/core/monitor.py b/zipline/core/monitor.py index 05bdfce1..d5b9af53 100644 --- a/zipline/core/monitor.py +++ b/zipline/core/monitor.py @@ -43,8 +43,8 @@ log = logbook.Logger('Controller') # the system. PARAMETERS = ndict(dict( - GENERATIONAL_PERIOD = 8, - ALLOWED_SKIPPED_HEARTBEATS = 3, + GENERATIONAL_PERIOD = 1, + ALLOWED_SKIPPED_HEARTBEATS = 10, ALLOWED_INVALID_HEARTBEATS = 3, PRESTART_HEARBEATS = 3, SOURCES_START_HEARTBEATS = 3, @@ -414,8 +414,8 @@ class Controller(object): # triggers the end of the topology. good = self.tracked & self.responses - bad = self.tracked - good - new = self.responses - good + bad = self.tracked - good - self.finished + new = self.responses - good - self.finished missing = self.topology - self.tracked - self.finished @@ -483,7 +483,7 @@ class Controller(object): 'FEED' : self.new_source, } - if component in self.topology or self.freeform: + if component in (self.topology - self.finished) or self.freeform: init_handlers.get(component, universal)() self.tracked.add(component) else: @@ -508,7 +508,7 @@ class Controller(object): universal = self.fail_universal fail_handlers = { } - if component in self.topology or self.freeform: + if component in (self.topology - self.finished) or self.freeform: log.warning('Component "%s" missed heartbeat' % component) self.tracked.remove(component) From 18cd9a02dfe1c7c7cd29df7b440dae9ac77c537d Mon Sep 17 00:00:00 2001 From: fawce Date: Fri, 13 Jul 2012 12:01:21 -0400 Subject: [PATCH 07/14] added more frequent heartbeating between requests from monitor. seems to work. --- zipline/components/aggregator.py | 10 +- zipline/components/tradesimulation.py | 11 +- zipline/core/component.py | 142 ++++++++++++++++++++------ zipline/core/monitor.py | 64 ++++-------- zipline/finance/risk.py | 2 +- zipline/transforms/base.py | 11 +- 6 files changed, 136 insertions(+), 104 deletions(-) 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) From ceb90133636b4a74a6d796fcaeacd6c2faccb738 Mon Sep 17 00:00:00 2001 From: fawce Date: Fri, 13 Jul 2012 14:54:35 -0400 Subject: [PATCH 08/14] heartbeats working, however the zipline is now fully sequential. --- zipline/components/aggregator.py | 1 + zipline/core/component.py | 6 +++--- zipline/core/monitor.py | 10 +++++++--- 3 files changed, 11 insertions(+), 6 deletions(-) diff --git a/zipline/components/aggregator.py b/zipline/components/aggregator.py index e4a6fba4..cc589789 100644 --- a/zipline/components/aggregator.py +++ b/zipline/components/aggregator.py @@ -114,6 +114,7 @@ class Aggregate(Component): """ while self.pending_messages() > 0: event = self.next() + self.heartbeat() if event: self.send(event) diff --git a/zipline/core/component.py b/zipline/core/component.py index b2353cc2..99e56184 100644 --- a/zipline/core/component.py +++ b/zipline/core/component.py @@ -334,17 +334,17 @@ class Component(object): # Just exit. 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(self.last_ping) + 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.last_ping = pre_pong # ---------------------------- # Cleanup & Modes of Failure diff --git a/zipline/core/monitor.py b/zipline/core/monitor.py index acf012b3..a9b66b4f 100644 --- a/zipline/core/monitor.py +++ b/zipline/core/monitor.py @@ -312,7 +312,7 @@ class Controller(object): if not self.router.getsockopt(self.zmq.RCVMORE): self.handle_recv(buffer[:]) buffer = [] - checktime = time.time() + #checktime = time.time() except INVALID_CONTROL_FRAME: log.error('Invalid frame', rawmessage) @@ -563,9 +563,13 @@ class Controller(object): # Go to your bosom; knock there, and ask your heart what # it doth know... self.responses.add(identity) - elif status < self.ctime: + elif float(status) < self.ctime: # False face must hide what the false heart doth know. - log.warning('Delayed heartbeat received.') + log.warning('Delayed heartbeat received: %s' % msg) + elif float(status) > self.ctime: + # Pre-emptive heartbeat from the component + # log.info("pre-emptive pong: %s" % msg) + self.responses.add(identity) else: # Otherwise its something weird and we don't know # what to do so just say so, probably line noise From 276725ba4d26496eed4b28045173d2e9481d231e Mon Sep 17 00:00:00 2001 From: fawce Date: Fri, 13 Jul 2012 16:11:15 -0400 Subject: [PATCH 09/14] three letter difference between sequential and parallel. --- zipline/components/aggregator.py | 2 +- zipline/components/tradesimulation.py | 4 +--- 2 files changed, 2 insertions(+), 4 deletions(-) diff --git a/zipline/components/aggregator.py b/zipline/components/aggregator.py index cc589789..6eb99fe9 100644 --- a/zipline/components/aggregator.py +++ b/zipline/components/aggregator.py @@ -92,7 +92,7 @@ class Aggregate(Component): try: self.append(event) - if not (self.is_full() or self.draining): + if self.is_full() or self.draining: event = self.next() if event: diff --git a/zipline/components/tradesimulation.py b/zipline/components/tradesimulation.py index 8452be49..eacae981 100644 --- a/zipline/components/tradesimulation.py +++ b/zipline/components/tradesimulation.py @@ -176,9 +176,7 @@ class TradeSimulationClient(Component): record.extra['algo_dt'] = self.current_dt data_injector = Processor(inject_event_data) - log_pipeline = NestedSetup([self.zmq_out, - #e.g. FileHandler(...) - data_injector]) + log_pipeline = NestedSetup([self.zmq_out,data_injector]) with log_pipeline.threadbound(), self.stdout_capture(self.logger, ''): self.algorithm.handle_data(data) # if no log socket, just run the algo normally From cd998d622ccfc58dde6d30474d08c2206f1fb6cf Mon Sep 17 00:00:00 2001 From: fawce Date: Sat, 14 Jul 2012 15:06:34 -0400 Subject: [PATCH 10/14] fixed logging bugs, added a DONE signal for logging. --- zipline/components/tradesimulation.py | 9 ++++++ zipline/core/monitor.py | 6 ++-- zipline/protocol.py | 13 ++++---- zipline/utils/log_utils.py | 45 ++++++++++++++------------- 4 files changed, 41 insertions(+), 32 deletions(-) diff --git a/zipline/components/tradesimulation.py b/zipline/components/tradesimulation.py index eacae981..764c3b27 100644 --- a/zipline/components/tradesimulation.py +++ b/zipline/components/tradesimulation.py @@ -106,10 +106,18 @@ class TradeSimulationClient(Component): # ended. Perf will internally calculate the full risk report. self.perf.handle_simulation_end() + # signal that the logging stream is done, close the + # logging socket + self.logging_done() + # signal Simulator, our ComponentHost, that this component is # done and Simulator needn't block exit on this component. self.signal_done() + def logging_done(self): + self.zmq_out.socket.send(zp.LOG_DONE) + self.zmq_out.close() + def process_event(self, event): # generate transactions, if applicable txn = self.txn_sim.apply_trade_to_open_orders(event) @@ -184,6 +192,7 @@ class TradeSimulationClient(Component): self.algorithm.handle_data(data) #Testing utility for log capture. + # TODO: remove test code from here. def test_run_algorithm(self): def inject_event_data(record): diff --git a/zipline/core/monitor.py b/zipline/core/monitor.py index a9b66b4f..c5a5149d 100644 --- a/zipline/core/monitor.py +++ b/zipline/core/monitor.py @@ -291,7 +291,6 @@ class Controller(object): # ============== # Wait the responses - checktime = self.ctime while self.alive: socks = dict(poller.poll(0)) @@ -312,14 +311,13 @@ class Controller(object): 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)) + if tic - self.ctime > self.period: + log.info("heartbeat loop timedout: %s" % (tic - self.ctime)) log.info(repr(self.responses)) break diff --git a/zipline/protocol.py b/zipline/protocol.py index 35cef68b..df98061b 100644 --- a/zipline/protocol.py +++ b/zipline/protocol.py @@ -608,6 +608,7 @@ SIMULATION_STYLE = Enum( LOG_FIELDS = set(['func_name', 'lineno', 'time', 'msg',\ 'level', 'channel', ]) LOG_EXTRA_FIELDS = set(['algo_dt',]) +LOG_DONE = "DONE" def LOG_FRAME(payload): """ @@ -621,14 +622,14 @@ def LOG_FRAME(payload): 'level' : 4, #Logbook enum 'channel' : 'MyLogger' } - + Frame checks that we have all expected fields and exports an event/payload dict as JSON. """ assert isinstance(payload, dict), \ "LOG_FRAME expected a dict" - + assert payload.has_key('algo_dt'), \ "LOG_FRAME with no algo_dt" assert payload.has_key('time'), \ @@ -639,11 +640,11 @@ def LOG_FRAME(payload): "LOG_FRAME with no level" assert payload.has_key('msg'),\ "LOG_FRAME with no message" - + data = {} data['e'] = 'LOG' data['p'] = payload - + return msgpack.dumps(data) def LOG_UNFRAME(msg): @@ -653,7 +654,5 @@ def LOG_UNFRAME(msg): record = msgpack.loads(msg) assert record['e'] == 'LOG' assert record.has_key('p') - + return record['p'] - - diff --git a/zipline/utils/log_utils.py b/zipline/utils/log_utils.py index bf08cf41..32a6b286 100644 --- a/zipline/utils/log_utils.py +++ b/zipline/utils/log_utils.py @@ -4,13 +4,16 @@ import pytz import datetime from logbook import NOTSET -from logbook.handlers import Handler +from logbook.handlers import Handler, FileHandler from zipline.protocol import LOG_FRAME, LOG_FIELDS, \ LOG_EXTRA_FIELDS from contextlib import contextmanager + +log = logbook.Logger("LogUtils") + class redirecter(object): def __init__(self, logger, name): self.logger = logger @@ -30,10 +33,10 @@ class redirecter(object): self.logger.error(out_form) self.buffer = bytes() -class log_redirecter(object): +class log_redirecter(object): def __init__(self, logger): self.logger = logger - + def write(self, line): #Absorb blank lines from print statements. if line =='\n': @@ -42,7 +45,7 @@ class log_redirecter(object): else: #TODO: add logic to guarantee we made this self.logger.info(line.strip('\n')) - + def flush(self, final=False): pass @@ -70,7 +73,7 @@ def stdout_only_pipe(logger, pipe_name): import sys orig_fd = sys.stdout sys.stdout = log_redirecter(logger) - + yield sys.stdout.flush() sys.stdout = orig_fd @@ -82,7 +85,7 @@ class ZeroMQLogHandler(Handler): Setup is similar to logbook.queues.ZeroMQHandler, except we connect instead of binding and we extract record fields into a dict. """ - + def __init__(self, uri=None, level=NOTSET, filter=None, bubble=False, context=None, fds = LOG_FIELDS, extra_fds = LOG_EXTRA_FIELDS): Handler.__init__(self, level, filter, bubble) @@ -109,21 +112,21 @@ class ZeroMQLogHandler(Handler): fields to make json happy. """ from zipline.utils.date_utils import EPOCH - + #Needed to extract record info from dictionary. record.pull_information() #Logbook stores record times as datetime objects, which - #can't be serialized by JSON, so we need to convert to + #can't be serialized by JSON, so we need to convert to #unix epoch representation. if record.time: assert isinstance(record.time, datetime.datetime) - + time = record.time.replace(tzinfo = pytz.utc) #logbook measures time in utc already, no need to convert. record.time = EPOCH(time) - + #Do the same if algo_dt is a datetime object. if record.extra.has_key('algo_dt'): algo_dt = record.extra['algo_dt'] @@ -131,31 +134,31 @@ class ZeroMQLogHandler(Handler): if isinstance(algo_dt, datetime.datetime): algo_dt = EPOCH(algo_dt.replace(tzinfo = pytz.utc)) record.extra['algo_dt'] = algo_dt - + data = {} - + #Extract all the fields we care about from LogRecord's internal - #dictionary. + #dictionary. for field in iter(self.fds): if record.__dict__.has_key(field): data[field] = record.__dict__[field] else: data[field] = None - + for field in iter(self.extra_fds): if record.extra.has_key(field): data[field] = record.extra[field] else: data[field] = None return data - + def emit(self, record): """Extract relevant fields and send info as JSON over a zmq socket.""" - payload = self.export_record(record) - self.socket.send(LOG_FRAME(payload)) - + logger = FileHandler('/var/log/zipline/zipline.log') + with logger.applicationbound(): + payload = self.export_record(record) + self.socket.send(LOG_FRAME(payload)) + def close(self): - #self.socket.close() - pass - + self.socket.close() From 1262dcdef1b07d2b55502343e74b1a4738a44591 Mon Sep 17 00:00:00 2001 From: fawce Date: Tue, 17 Jul 2012 01:28:17 -0400 Subject: [PATCH 11/14] logging converted to share socket with performance --- zipline/components/feed.py | 1 + zipline/components/tradesimulation.py | 50 ++++++++++++++++----------- zipline/core/component.py | 6 ++-- zipline/finance/performance.py | 12 +++---- zipline/lines.py | 30 ++++++---------- zipline/protocol.py | 18 ++++++---- zipline/utils/log_utils.py | 19 +++++----- 7 files changed, 69 insertions(+), 67 deletions(-) diff --git a/zipline/components/feed.py b/zipline/components/feed.py index 719a99a6..1af9812c 100644 --- a/zipline/components/feed.py +++ b/zipline/components/feed.py @@ -80,6 +80,7 @@ class Feed(Aggregate): Get the next message in chronological order. """ + # TODO: this is redundant to the guard in aggregator. # is_full and draining defined in aggregator if not(self.is_full() or self.draining): return diff --git a/zipline/components/tradesimulation.py b/zipline/components/tradesimulation.py index 764c3b27..08ed1b07 100644 --- a/zipline/components/tradesimulation.py +++ b/zipline/components/tradesimulation.py @@ -1,6 +1,8 @@ import logbook import datetime +import zmq + import zipline.protocol as zp import zipline.finance.performance as perf @@ -18,7 +20,7 @@ log = logbook.Logger('TradeSimulation') class TradeSimulationClient(Component): - def init(self, trading_environment, sim_style, log_socket): + def init(self, trading_environment, sim_style, results_socket): self.received_count = 0 self.prev_dt = None self.event_queue = None @@ -34,8 +36,8 @@ class TradeSimulationClient(Component): self.event_data = ndict() self.perf = perf.PerformanceTracker(self.trading_environment) - - self.log_socket = log_socket + self.zmq_out = None + self.results_socket = results_socket @property def get_id(self): @@ -64,20 +66,34 @@ class TradeSimulationClient(Component): self.algorithm.initialize() def open(self): - self.result_feed = self.connect_result() - self.perf.open(self.context) - self.setup_logging(self.context) + if not self.results_socket: + log.warn(" No results socket, will not broadcast sim data.") + else: + sock = self.context.socket(zmq.PUSH) + sock.connect(self.results_socket) + self.results_socket = sock + self.sockets.append(sock) + self.out_socket = sock + + + self.setup_logging(sock) + self.perf.publish_to(sock) + #Initialize log capture for testing purposes. - def setup_logging(self, context): - if self.log_socket: - self.zmq_out = ZeroMQLogHandler(uri = self.log_socket, context = context) - self.logger = Logger("Print") - self.stdout_capture = stdout_only_pipe #THIS IS A CLASS! + def setup_logging(self, socket = None): + sock = socket or self.results_socket + + self.zmq_out = ZeroMQLogHandler( + socket = sock, + ) + + self.logger = Logger("Print") + # THIS IS A CLASS! + self.stdout_capture = stdout_only_pipe def do_work(self): - # see if the poller has results for the result_feed if self.socks.get(self.result_feed) == self.zmq.POLLIN: @@ -106,17 +122,10 @@ class TradeSimulationClient(Component): # ended. Perf will internally calculate the full risk report. self.perf.handle_simulation_end() - # signal that the logging stream is done, close the - # logging socket - self.logging_done() - # signal Simulator, our ComponentHost, that this component is # done and Simulator needn't block exit on this component. self.signal_done() - def logging_done(self): - self.zmq_out.socket.send(zp.LOG_DONE) - self.zmq_out.close() def process_event(self, event): # generate transactions, if applicable @@ -174,8 +183,7 @@ class TradeSimulationClient(Component): # data injection pipeline for log rerouting # any fields injected here should be added to # LOG_EXTRA_FIELDS in zipline/protocol.py - - if self.log_socket: + if self.zmq_out: def inject_event_data(record): diff --git a/zipline/core/component.py b/zipline/core/component.py index 99e56184..2d2aa94e 100644 --- a/zipline/core/component.py +++ b/zipline/core/component.py @@ -53,7 +53,7 @@ class Component(object): the PULL side (we always have N producers and 1 consumer) - :param result_address: socket address used to publish merged data + :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. @@ -554,10 +554,10 @@ class Component(object): return self.connect_push_socket(self.addresses['merge_address']) def bind_result(self): - return self.bind_push_socket(self.addresses['result_address']) + return self.bind_push_socket(self.addresses['results_address']) def connect_result(self): - return self.connect_pull_socket(self.addresses['result_address']) + return self.connect_pull_socket(self.addresses['results_address']) def bind_push_socket(self, addr): push_socket = self.context.socket(self.zmq.PUSH) diff --git a/zipline/finance/performance.py b/zipline/finance/performance.py index ab0c17e6..7844b706 100644 --- a/zipline/finance/performance.py +++ b/zipline/finance/performance.py @@ -214,8 +214,9 @@ class PerformanceTracker(object): Publish the performance results asynchronously to a socket. """ - assert isinstance(results_addr, basestring), type(results_addr) - self.results_addr = results_addr + #assert isinstance(results_addr, basestring), type(results_addr) + #self.results_addr = results_addr + self.results_socket = results_addr def to_dict(self): """ @@ -261,7 +262,7 @@ class PerformanceTracker(object): self.todays_performance.calculate_performance() def handle_market_close(self): - + # add the return results from today to the list of DailyReturn objects. todays_date = self.market_close.replace(hour=0, minute=0, second=0) todays_return_obj = risk.DailyReturn( @@ -347,12 +348,9 @@ class PerformanceTracker(object): if self.results_socket: log.info("about to stream the risk report...") risk_dict = self.risk_report.to_dict() - + msg = zp.RISK_FRAME(risk_dict) self.results_socket.send(msg) - # this signals that the simulation is complete. - self.results_socket.send("DONE") - class Position(object): diff --git a/zipline/lines.py b/zipline/lines.py index 2583672c..f56c1eb3 100644 --- a/zipline/lines.py +++ b/zipline/lines.py @@ -122,26 +122,25 @@ class SimulatedTrading(object): self.leased_sockets = [] self.sim_context = None - sockets = self.allocate_sockets(8) + sockets = self.allocate_sockets(7) addresses = { 'sync_address' : sockets[0], 'data_address' : sockets[1], 'feed_address' : sockets[2], 'merge_address' : sockets[3], - 'result_address' : sockets[4], - 'order_address' : sockets[5] + # TODO: this refers to the results of the merge, a + # horribly confusing name for the socket. + 'results_address' : sockets[4], } self.con = Controller( + sockets[5], sockets[6], - sockets[7], devel = self.devel ) # TODO: Not freeform - self.con.manage( - 'freeform' - ) + self.con.manage('freeform') self.started = False @@ -152,19 +151,15 @@ class SimulatedTrading(object): self.trading_client = TradeSimulationClient( self.trading_environment, self.sim_style, - config['log_socket'] + config['results_socket'] ) self.add_client(self.trading_client) # setup all sources self.sources = {} - #self.order_source = OrderDataSource() - #self.add_source(self.order_source) #setup transforms - #self.transaction_sim = TransactionSimulator(self.sim_style) self.transforms = {} - #self.add_transform(self.transaction_sim) self.sim.register_controller( self.con ) @@ -260,10 +255,10 @@ class SimulatedTrading(object): order_count ) - if config.has_key('log_socket'): - log_socket = config['log_socket'] + if config.has_key('results_socket'): + results_socket = config['results_socket'] else: - log_socket = None + results_socket = None #------------------- # Simulation #------------------- @@ -273,7 +268,7 @@ class SimulatedTrading(object): 'allocator' : allocator, 'simulator_class' : simulator_class, 'simulation_style' : simulation_style, - 'log_socket' : log_socket, + 'results_socket' : results_socket, 'devel' : config.get('devel', False) }) #------------------- @@ -322,9 +317,6 @@ class SimulatedTrading(object): def get_cumulative_performance(self): return self.trading_client.perf.cumulative_performance.to_dict() - def publish_to(self, results_socket): - self.trading_client.perf.publish_to(results_socket) - def allocate_sockets(self, n): """ Allocate sockets local to this line, track them so diff --git a/zipline/protocol.py b/zipline/protocol.py index df98061b..3f0de718 100644 --- a/zipline/protocol.py +++ b/zipline/protocol.py @@ -513,9 +513,9 @@ def BT_UPDATE_FRAME(prefix, payload): def BT_UPDATE_UNFRAME(msg): """ - Risk and Perf framing methods prefix the payload with + Risk, Perf, and LOG framing methods prefix the payload with a shorthand for their type. That way, all messages received from the socket - can be PERF_FRAMED(), whether they are risk or perf. + can be PERF_FRAMED(), whether they are risk, perf, or log. """ prefix, payload = msgpack.loads(msg, use_list=True) return dict(prefix=prefix, payload=payload) @@ -642,17 +642,21 @@ def LOG_FRAME(payload): "LOG_FRAME with no message" data = {} - data['e'] = 'LOG' + data['e'] = 'log' data['p'] = payload - return msgpack.dumps(data) + return BT_UPDATE_FRAME('LOG', data) def LOG_UNFRAME(msg): """ Expects a json serialized dictionary in event/payload format. """ record = msgpack.loads(msg) - assert record['e'] == 'LOG' - assert record.has_key('p') + assert isinstance(record, tuple) + assert len(record) == 2 + assert record[0] == 'LOG' + payload = record[1] + assert payload['e'] == 'log' + assert payload.has_key('p') - return record['p'] + return payload['p'] diff --git a/zipline/utils/log_utils.py b/zipline/utils/log_utils.py index 32a6b286..6bcf80f8 100644 --- a/zipline/utils/log_utils.py +++ b/zipline/utils/log_utils.py @@ -86,7 +86,7 @@ class ZeroMQLogHandler(Handler): instead of binding and we extract record fields into a dict. """ - def __init__(self, uri=None, level=NOTSET, filter=None, bubble=False, + def __init__(self, socket=None, level=NOTSET, filter=None, bubble=False, context=None, fds = LOG_FIELDS, extra_fds = LOG_EXTRA_FIELDS): Handler.__init__(self, level, filter, bubble) try: @@ -97,11 +97,11 @@ class ZeroMQLogHandler(Handler): #: the zero mq context self.context = context #: the zero mq socket. - self.socket = self.context.socket(zmq.PUSH) + self.socket = socket #self.context.socket(zmq.PUSH) - self.uri = uri - if uri is not None: - self.socket.connect(uri) + #self.uri = uri + #if uri is not None: + # self.socket.connect(uri) self.fds = fds self.extra_fds = extra_fds @@ -155,10 +155,9 @@ class ZeroMQLogHandler(Handler): def emit(self, record): """Extract relevant fields and send info as JSON over a zmq socket.""" - logger = FileHandler('/var/log/zipline/zipline.log') - with logger.applicationbound(): - payload = self.export_record(record) - self.socket.send(LOG_FRAME(payload)) + payload = self.export_record(record) + self.socket.send(LOG_FRAME(payload)) def close(self): - self.socket.close() + pass + #self.socket.close() From b416f7f4183302f77396890dd5d9c6a4707e1650 Mon Sep 17 00:00:00 2001 From: fawce Date: Tue, 17 Jul 2012 14:09:38 -0400 Subject: [PATCH 12/14] tweak --- zipline/core/monitor.py | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/zipline/core/monitor.py b/zipline/core/monitor.py index c5a5149d..33db3e8d 100644 --- a/zipline/core/monitor.py +++ b/zipline/core/monitor.py @@ -33,8 +33,7 @@ class UnknownChatter(Exception): def __init__(self, name): self.named = name def __str__(self): - return "Component calling itself '%s' talking on unexpected channel" - % self.named + return """Component calling itself "%s" talking on unexpected channel""" % self.named log = logbook.Logger('Controller') From 6415f59398b2687e8171255fac0700b663b6f94a Mon Sep 17 00:00:00 2001 From: fawce Date: Tue, 17 Jul 2012 17:43:02 -0400 Subject: [PATCH 13/14] changing protocol remove e/p dictionary --- zipline/protocol.py | 5 +---- 1 file changed, 1 insertion(+), 4 deletions(-) diff --git a/zipline/protocol.py b/zipline/protocol.py index 3f0de718..3e7b57cf 100644 --- a/zipline/protocol.py +++ b/zipline/protocol.py @@ -641,11 +641,8 @@ def LOG_FRAME(payload): assert payload.has_key('msg'),\ "LOG_FRAME with no message" - data = {} - data['e'] = 'log' - data['p'] = payload - return BT_UPDATE_FRAME('LOG', data) + return BT_UPDATE_FRAME('LOG', payload) def LOG_UNFRAME(msg): """ From fb2b0c3be581c4795719d099db7b913e7ecc53e8 Mon Sep 17 00:00:00 2001 From: fawce Date: Wed, 18 Jul 2012 17:20:31 -0400 Subject: [PATCH 14/14] responses to sdiehl code review --- zipline/components/tradesimulation.py | 3 ++- zipline/core/component.py | 10 ++-------- zipline/core/monitor.py | 10 ++++------ zipline/protocol.py | 6 ++---- 4 files changed, 10 insertions(+), 19 deletions(-) diff --git a/zipline/components/tradesimulation.py b/zipline/components/tradesimulation.py index 08ed1b07..6c761114 100644 --- a/zipline/components/tradesimulation.py +++ b/zipline/components/tradesimulation.py @@ -90,7 +90,8 @@ class TradeSimulationClient(Component): ) self.logger = Logger("Print") - # THIS IS A CLASS! + # N.B. that this is a class, which is instantiated later + # in run_algorithm. The class provides a generator. self.stdout_capture = stdout_only_pipe def do_work(self): diff --git a/zipline/core/component.py b/zipline/core/component.py index 2d2aa94e..841800e1 100644 --- a/zipline/core/component.py +++ b/zipline/core/component.py @@ -269,9 +269,6 @@ class Component(object): # ---------------- # 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' @@ -327,11 +324,8 @@ class Component(object): elif event == CONTROL_PROTOCOL.KILL: self.kill() - # ========= - # Hard Kill - # ========= - - # Just exit. + # 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() diff --git a/zipline/core/monitor.py b/zipline/core/monitor.py index 33db3e8d..11b91fa8 100644 --- a/zipline/core/monitor.py +++ b/zipline/core/monitor.py @@ -295,12 +295,6 @@ class Controller(object): socks = dict(poller.poll(0)) tic = time.time() - # 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() @@ -315,6 +309,10 @@ class Controller(object): log.error('Invalid frame', rawmessage) pass + # 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: log.info("heartbeat loop timedout: %s" % (tic - self.ctime)) log.info(repr(self.responses)) diff --git a/zipline/protocol.py b/zipline/protocol.py index 3e7b57cf..866e5410 100644 --- a/zipline/protocol.py +++ b/zipline/protocol.py @@ -646,14 +646,12 @@ def LOG_FRAME(payload): def LOG_UNFRAME(msg): """ - Expects a json serialized dictionary in event/payload format. + msg should be a tuple of ('LOG',dict) """ record = msgpack.loads(msg) assert isinstance(record, tuple) assert len(record) == 2 assert record[0] == 'LOG' payload = record[1] - assert payload['e'] == 'log' - assert payload.has_key('p') - return payload['p'] + return payload