From b19d2d856f1ed77cd34351a35f9ae7864d39048e Mon Sep 17 00:00:00 2001 From: Stephen Diehl Date: Tue, 3 Jul 2012 11:05:01 -0400 Subject: [PATCH] Selectively backport portions of new component.py --- zipline/core/component.py | 125 ++++++++++++++++++++++++++++---------- 1 file changed, 94 insertions(+), 31 deletions(-) diff --git a/zipline/core/component.py b/zipline/core/component.py index 2d0619ce..3778ac5c 100644 --- a/zipline/core/component.py +++ b/zipline/core/component.py @@ -19,6 +19,7 @@ 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 @@ -82,12 +83,15 @@ class Component(object): self.zmq = None self.context = None self.addresses = None + self.waiting = None self.out_socket = None self.killed = False self.controller = None # timeout after a full minute self.heartbeat_timeout = 60 *1000 + # TODO: state_flag is deprecated, remove + # TODO: error_state is deprecated, remove self.state_flag = COMPONENT_STATE.OK self.error_state = COMPONENT_FAILURE.NOFAILURE self.on_done = None @@ -190,6 +194,12 @@ class Component(object): raise Exception("Unknown ZeroMQ Flavor") def _run(self): + """ + The main component loop. This is wrapped inside a + exception reporting context inside of run. + + The core logic of the all components is run here. + """ self.start_tic = time.time() self.done = False # TODO: use state flag @@ -200,9 +210,17 @@ class Component(object): self.setup_poller() self.open() - self.setup_sync() self.setup_control() + self.signal_ready() + self.lock_ready() + + self.wait_ready() + # ----------------------- + # YOU SHALL NOT PASS!!!!! + # ----------------------- + # ... until the controller signals GO + self.loop() self.shutdown() @@ -246,20 +264,8 @@ class Component(object): Loop to do work while we still have work to do. """ while self.working(): - self.confirm() self.do_work() - def confirm(self): - """ - Send a synchronization request to the host. - """ - if not self.confirmed: - # TODO: proper framing - self.sync_socket.send(self.get_id + ":RUN") - - self.receive_sync_ack() # blocking - self.confirmed = True - def runtime(self): if self.ready() and self.start_tic and self.stop_tic: return self.stop_tic - self.start_tic @@ -299,6 +305,81 @@ class Component(object): # Internal Maintenance # ---------------------- + def lock_ready(self): + """ + Unlock the component, topology is now ready to run. + """ + self.waiting = True + + def unlock_ready(self): + """ + Unlock the component, topology is still pending. + """ + self.waiting = False + + def wait_ready(self): + # Implicit side-effect of unlocking the component iff + # the GO message is received from the monitor level. + # This then unlocks the barrier and proceeds to the + # do_work state. + + # Poll on a subset of the control protocol while we exist + # in the locked quasimode. Respond to HEARTBEAT and GO + # messages. + + while self.waiting: + socks = dict(self.poll.poll(self.heartbeat_timeout)) + + msg = self.control_in.recv() + event, payload = CONTROL_UNFRAME(msg) + + # ==== + # 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() + + def signal_ready(self): + log.info(self.__class__.__name__ + ' is ready') + + if hasattr(self, 'control_out'): + frame = CONTROL_FRAME( + CONTROL_PROTOCOL.READY, + '' + ) + self.control_out.send(frame) + + def signal_cancel(self): + self.done = True + + # TODO: no hasattr hacks + #if not self.controller: + if hasattr(self, 'control_out'): + frame = CONTROL_FRAME( + CONTROL_PROTOCOL.SHUTDOWN, + None + ) + self.control_out.send(frame) + + # then proceeds to do shutdown(), and teardown_sockets() + # to complete the process + def signal_exception(self, exc=None, scope=None): """ This is *very* important error tracking handler. @@ -470,24 +551,6 @@ class Component(object): self.poll.register(self.control_in, self.zmq.POLLIN) self.sockets.extend([self.control_in, self.control_out]) - def setup_sync(self): - """ - Setup the sync socket and poller. ( Connect ) - - DEPRECATED, left in for compatability for now. - """ - - #LOGGER.debug("Connecting sync client for {id}".format(id=self.get_id)) - - self.sync_socket = self.context.socket(self.zmq.REQ) - self.sync_socket.connect(self.addresses['sync_address']) - #self.sync_socket.setsockopt(self.zmq.LINGER,0) - - self.sync_poller = self.zmq_poller() - self.sync_poller.register(self.sync_socket, self.zmq.POLLIN) - - self.sockets.append(self.sync_socket) - # ----------- # FSM Actions # -----------