mirror of
https://github.com/wassname/catalyst.git
synced 2026-07-10 08:18:24 +08:00
Selectively backport portions of new component.py
This commit is contained in:
Stephen Diehl
+94
-31
@@ -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
|
||||
# -----------
|
||||
|
||||
Reference in New Issue
Block a user