From 0bf425a8dc3899e0982a38a5674f24839804fcd3 Mon Sep 17 00:00:00 2001 From: Stephen Diehl Date: Wed, 25 Apr 2012 19:52:14 -0400 Subject: [PATCH 1/7] initial work on new protocol --- zipline/monitor.py | 17 +++++++++++++---- zipline/protocol.py | 21 ++++----------------- 2 files changed, 17 insertions(+), 21 deletions(-) diff --git a/zipline/monitor.py b/zipline/monitor.py index f6b55037..e901b3ab 100644 --- a/zipline/monitor.py +++ b/zipline/monitor.py @@ -163,6 +163,7 @@ class Controller(object): self.ctime = 0 self.tic = time.time() self.freeform = False + self._state = -1 self.associated = [] @@ -218,17 +219,25 @@ class Controller(object): self.topology = frozenset(topology) default_states = [ + CONTROL_STATES.INIT, + CONTROL_STATES.SOURCES_READY, CONTROL_STATES.RUNNING, - CONTROL_STATES.SHUTDOWN, CONTROL_STATES.TERMINATE, ] self.states = states or default_states self.polling = True - - # Start off in RUNNING, state self.state = self.states[0] + @property + def state(self): + return self._state + + @state.setter + def state(self, new): + old, self._state = self._state, new + self.logging.info("[Controller] State Transition : %s -> %s" %(old, new)) + def run(self): self.init_zmq(self.zmq_flavor) @@ -472,7 +481,7 @@ class Controller(object): assert hard or soft, """ Must specify kill hard or soft """ if hard: - self.state = CONTROL_STATES.SHUTDOWN + self.state = CONTROL_STATES.TERMINATE self.logging.info('[Controller] Hard Shutdown') diff --git a/zipline/protocol.py b/zipline/protocol.py index c3291556..90a3184a 100644 --- a/zipline/protocol.py +++ b/zipline/protocol.py @@ -126,9 +126,6 @@ from collections import namedtuple from protocol_utils import Enum, FrameExceptionFactory, namedict from date_utils import EPOCH, UN_EPOCH -#import ujson -#import ultrajson_numpy - # ----------------------- # Control Protocol # ----------------------- @@ -136,9 +133,10 @@ from date_utils import EPOCH, UN_EPOCH INVALID_CONTROL_FRAME = FrameExceptionFactory('CONTROL') CONTROL_STATES = Enum( + 'INIT', + 'SOURCES_READY', 'RUNNING', - 'SHUTDOWN', # a soft kill - 'TERMINATE', # a hard kill + 'TERMINATE', ) CONTROL_PROTOCOL = Enum( @@ -149,6 +147,7 @@ CONTROL_PROTOCOL = Enum( 'OK' , # 3 - rep 'DONE' , # 4 - rep 'EXCEPTION' , # 5 - rep + 'SIGNAL' , # 6 - rep ) def CONTROL_FRAME(event, payload): @@ -174,18 +173,6 @@ def CONTROL_UNFRAME(msg): except ValueError: raise INVALID_CONTROL_FRAME(msg) -# ----------------------- -# Heartbeat Protocol -# ----------------------- - -# These encode the msgpack equivelant of 1 and 2. The heartbeat -# frame should only be 1 byte on the wire. - -HEARTBEAT_PROTOCOL = namedict({ - 'REQ' : b'\x01', - 'REP' : b'\x02', -}) - # ----------------------- # Component State # ----------------------- From d09aa1ad935504749040b224be63979107b99ba0 Mon Sep 17 00:00:00 2001 From: Stephen Diehl Date: Thu, 26 Apr 2012 13:52:46 -0400 Subject: [PATCH 2/7] Make monitor have proper state transitions. --- zipline/component.py | 3 +- zipline/monitor.py | 63 +++++++++++++++++++++++++++------------ zipline/protocol_utils.py | 1 + 3 files changed, 47 insertions(+), 20 deletions(-) diff --git a/zipline/component.py b/zipline/component.py index 4e1d6b7f..1f8ebc1f 100644 --- a/zipline/component.py +++ b/zipline/component.py @@ -492,7 +492,8 @@ class Component(object): """ The descriptive name of the component. """ - return 'UNKNOWN COMPONENT' + # Prevents the bug that Thomas ran into + raise NotImplementedError @property def get_type(self): diff --git a/zipline/monitor.py b/zipline/monitor.py index e901b3ab..dde4450b 100644 --- a/zipline/monitor.py +++ b/zipline/monitor.py @@ -9,7 +9,9 @@ import gevent_zeromq #import zmq_ctypes from protocol import CONTROL_PROTOCOL, CONTROL_FRAME, \ - CONTROL_UNFRAME, CONTROL_STATES, INVALID_CONTROL_FRAME + CONTROL_UNFRAME, CONTROL_STATES, INVALID_CONTROL_FRAME \ + +states = CONTROL_STATES from gpoll import _Poller as GeventPoller @@ -103,6 +105,17 @@ from gpoll import _Poller as GeventPoller # | 0 | | 0 | | 0 | | 0 | # +---+ +---+ +---+ +---+ +INIT, SOURCES_READY, RUNNING, TERMINATE = CONTROL_STATES + +state_transitions = frozenset([ + (-1 , INIT), + (INIT , SOURCES_READY), + (SOURCES_READY , RUNNING), + (INIT , TERMINATE), + (SOURCES_READY , TERMINATE), + (RUNNING , TERMINATE), +]) + class UnknownChatter(Exception): def __init__(self, name): self.named = name @@ -218,16 +231,8 @@ class Controller(object): self.freeform = False self.topology = frozenset(topology) - default_states = [ - CONTROL_STATES.INIT, - CONTROL_STATES.SOURCES_READY, - CONTROL_STATES.RUNNING, - CONTROL_STATES.TERMINATE, - ] - - self.states = states or default_states self.polling = True - self.state = self.states[0] + self.state = CONTROL_STATES.INIT @property def state(self): @@ -236,7 +241,11 @@ class Controller(object): @state.setter def state(self, new): old, self._state = self._state, new - self.logging.info("[Controller] State Transition : %s -> %s" %(old, new)) + + if (old, new) not in state_transitions: + raise RuntimeError("[Controller] Invalid State Transition : %s -> %s" %(old, new)) + else: + self.logging.info("[Controller] State Transition : %s -> %s" %(old, new)) def run(self): self.init_zmq(self.zmq_flavor) @@ -302,6 +311,8 @@ class Controller(object): poller = self.zmq.Poller() poller.register(self.router, self.zmq.POLLIN) + self.state = CONTROL_STATES.SOURCES_READY + buffer = [] for i in itertools.count(0): @@ -372,26 +383,38 @@ class Controller(object): # Component Handlers # ------------------ + + def new_source(self): + if self.state is CONTROL_STATES.RUNNING: + self.state = SOURCES_READY + + def new_universal(self): + pass + # The various "states of being that a component can inform us # of def new(self, component): - self.logging.info('[Controller] Alive "%s" ' % component) + self.logging.info('[Controller] Now Tracking "%s" ' % component) + + universal = self.new_universal + component_initializers = { + 'FEED' : self.new_source, + } if component in self.topology or self.freeform: + component_initializers.get(component, universal)() self.tracked.add(component) else: # Some sort of socket collision has occured, this is # a very bad failure mode. raise UnknownChatter(component) + def fail(self, component): self.logging.info('[Controller] Component "%s" timed out' % component) self.tracked.remove(component) def done(self, component): - # TODO: This will be what we ship off to vbench at some - # point... - # print component finished at self.ctime self.logging.info('[Controller] Component "%s" done.' % component) def exception(self, component, failure): @@ -499,16 +522,18 @@ class Controller(object): if __name__ == '__main__': - print 'Running on ',\ - 'tcp://127.0.0.1:5000', \ - 'tcp://127.0.0.1:5001', + print 'Running on '\ + 'tcp://127.0.0.1:5000 '\ + 'tcp://127.0.0.1:5001 ' controller = Controller( 'tcp://127.0.0.1:5000', 'tcp://127.0.0.1:5001', ) + controller.zmq_flavor = 'green' + controller.manage( 'freeform', [] ) - controller.run('green') + controller.run() diff --git a/zipline/protocol_utils.py b/zipline/protocol_utils.py index 8d21b1a0..60c90814 100644 --- a/zipline/protocol_utils.py +++ b/zipline/protocol_utils.py @@ -12,6 +12,7 @@ def Enum(*options): """ class cstruct(Structure): _fields_ = [(o, c_ubyte) for o in options] + __iter__ = lambda s: iter(range(len(options))) return cstruct(*range(len(options))) def FrameExceptionFactory(name): From e0d3811aedbfeb6a182b15dfa534cca0391e8955 Mon Sep 17 00:00:00 2001 From: Stephen Diehl Date: Thu, 26 Apr 2012 14:26:06 -0400 Subject: [PATCH 3/7] Fail fast in monitor on errors. --- zipline/monitor.py | 81 ++++++++++++++++++++++++++++++---------------- 1 file changed, 54 insertions(+), 27 deletions(-) diff --git a/zipline/monitor.py b/zipline/monitor.py index dde4450b..664bdb6d 100644 --- a/zipline/monitor.py +++ b/zipline/monitor.py @@ -3,10 +3,9 @@ import gevent import itertools # pyzmq import zmq -# gevent_zeromq import gevent_zeromq -# zmq_ctypes -#import zmq_ctypes + +from collections import OrderedDict from protocol import CONTROL_PROTOCOL, CONTROL_FRAME, \ CONTROL_UNFRAME, CONTROL_STATES, INVALID_CONTROL_FRAME \ @@ -162,28 +161,27 @@ class Controller(object): def __init__(self, pub_socket, route_socket, logging = None): - self.context = None - self.zmq = None + self.context = None + self.zmq = None self.zmq_poller = None polling = False self.polling = polling - self.tracked = set() self.responses = set() - self.ctime = 0 - self.tic = time.time() + self.ctime = 0 + self.tic = time.time() self.freeform = False - self._state = -1 + self._state = -1 self.associated = [] - self.pub_socket = pub_socket + self.pub_socket = pub_socket self.route_socket = route_socket - self.error_replay = {} + self.error_replay = OrderedDict() if logging: self.logging = logging @@ -196,23 +194,23 @@ class Controller(object): assert self.zmq_flavor in ['thread', 'mp', 'green'] if flavor == 'mp': - self.zmq = zmq - self.context = self.zmq.Context() + self.zmq = zmq + self.context = self.zmq.Context() self.zmq_poller = self.zmq.Poller return if flavor == 'thread': - self.zmq = zmq - self.context = self.zmq.Context.instance() + self.zmq = zmq + self.context = self.zmq.Context.instance() self.zmq_poller = self.zmq.Poller return if flavor == 'green': - self.zmq = gevent_zeromq.zmq - self.context = self.zmq.Context.instance() + self.zmq = gevent_zeromq.zmq + self.context = self.zmq.Context.instance() self.zmq_poller = GeventPoller return if flavor == 'pypy': - self.zmq = zmq - self.context = self.zmq.Context.instance() + self.zmq = zmq + self.context = self.zmq.Context.instance() self.zmq_poller = self.zmq.Poller return @@ -350,6 +348,9 @@ class Controller(object): if self.zmq_flavor == 'green': gevent.sleep(0) + if self.state is CONTROL_STATES.TERMINATE: + break + if not self.polling: break @@ -379,10 +380,9 @@ class Controller(object): for component in bad: self.fail(component) - # ------------------ - # Component Handlers - # ------------------ - + # -------------- + # Init Handlers + # -------------- def new_source(self): if self.state is CONTROL_STATES.RUNNING: @@ -397,12 +397,12 @@ class Controller(object): self.logging.info('[Controller] Now Tracking "%s" ' % component) universal = self.new_universal - component_initializers = { + init_handlers = { 'FEED' : self.new_source, } if component in self.topology or self.freeform: - component_initializers.get(component, universal)() + init_handlers.get(component, universal)() self.tracked.add(component) else: # Some sort of socket collision has occured, this is @@ -417,9 +417,29 @@ class Controller(object): def done(self, component): self.logging.info('[Controller] Component "%s" done.' % component) + # -------------- + # Error Handling + # -------------- + + def exception_universal(self): + """ + Shutdown the system on failure. + """ + self.state = CONTROL_STATES.TERMINATE + self.logging.error('[Controller] System in exception state, shutting down') + def exception(self, component, failure): - self.error_replay[time.time()] = failure - self.logging.error('Component "%s" in exception state' % component) + universal = self.exception_universal + exception_handlers = { + } + + if component in self.topology or self.freeform: + self.error_replay[(component, time.time())] = failure + self.logging.error('[Controller] Component "%s" in exception state' % component) + + exception_handlers.get(component, universal)() + else: + raise UnknownChatter(component) # ----------------- # Protocol Handling @@ -494,6 +514,11 @@ class Controller(object): self.associated.append(s) return s + def do_error_replay(self): + for (component, time), error in self.error_replay: + self.logging.info('[Controller] Error Log for -- %s --:\n%s' % + (component, error)) + def shutdown(self, hard=False, soft=True, context=None): if not self.polling: @@ -520,6 +545,8 @@ class Controller(object): #for asoc in self.associated: #asoc.close() + self.do_error_replay() + if __name__ == '__main__': print 'Running on '\ From 08f7b115acbbf268fb56a3f24e5e376d449bb06a Mon Sep 17 00:00:00 2001 From: Stephen Diehl Date: Thu, 26 Apr 2012 14:42:38 -0400 Subject: [PATCH 4/7] Cleaner exception/shutdown procedures. --- zipline/component.py | 5 +++-- zipline/messaging.py | 4 ++-- zipline/monitor.py | 4 ++-- 3 files changed, 7 insertions(+), 6 deletions(-) diff --git a/zipline/component.py b/zipline/component.py index 1f8ebc1f..d82c8fb9 100644 --- a/zipline/component.py +++ b/zipline/component.py @@ -10,6 +10,7 @@ import uuid import time import socket import gevent +import traceback import humanhash # pyzmq @@ -305,11 +306,11 @@ class Component(object): self._exception = exc exc_type, exc_value, exc_traceback = sys.exc_info() - self.stack_trace = exc_traceback + trace = '\n>>>'.join(traceback.format_exception(exc_type, exc_value, exc_traceback)) exception_frame = CONTROL_FRAME( CONTROL_PROTOCOL.EXCEPTION, - str(exc) + trace ) self.control_out.send(exception_frame) diff --git a/zipline/messaging.py b/zipline/messaging.py index 6cd4154d..fd1875c1 100644 --- a/zipline/messaging.py +++ b/zipline/messaging.py @@ -227,7 +227,7 @@ class Feed(Component): # -- Soft Kill -- elif event == CONTROL_PROTOCOL.SHUTDOWN: - self.done() + self.signal_done() self.shutdown() # -- Hard Kill -- @@ -496,7 +496,7 @@ class BaseTransform(Component): # -- Soft Kill -- elif event == CONTROL_PROTOCOL.SHUTDOWN: - self.done() + self.signal_done() self.shutdown() # -- Hard Kill -- diff --git a/zipline/monitor.py b/zipline/monitor.py index 664bdb6d..915043b4 100644 --- a/zipline/monitor.py +++ b/zipline/monitor.py @@ -425,8 +425,8 @@ class Controller(object): """ Shutdown the system on failure. """ - self.state = CONTROL_STATES.TERMINATE self.logging.error('[Controller] System in exception state, shutting down') + self.shutdown(soft=True) def exception(self, component, failure): universal = self.exception_universal @@ -515,7 +515,7 @@ class Controller(object): return s def do_error_replay(self): - for (component, time), error in self.error_replay: + for (component, time), error in self.error_replay.iteritems(): self.logging.info('[Controller] Error Log for -- %s --:\n%s' % (component, error)) From cfe71196c5cad6e0df3e8d3c94789b096ce312a4 Mon Sep 17 00:00:00 2001 From: Stephen Diehl Date: Thu, 26 Apr 2012 14:45:28 -0400 Subject: [PATCH 5/7] Fail on timeouts & timeout handlers. --- zipline/monitor.py | 21 +++++++++++++++++++-- 1 file changed, 19 insertions(+), 2 deletions(-) diff --git a/zipline/monitor.py b/zipline/monitor.py index 915043b4..3f4af9fc 100644 --- a/zipline/monitor.py +++ b/zipline/monitor.py @@ -409,10 +409,27 @@ class Controller(object): # a very bad failure mode. raise UnknownChatter(component) + # ------------------ + # Epic Fail Handling + # ------------------ + + def universal(self): + self.logging.error('[Controller] System in exception state, shutting down') + self.terminate(soft=True) def fail(self, component): - self.logging.info('[Controller] Component "%s" timed out' % component) - self.tracked.remove(component) + universal = self.fail_universal + fail_handlers = { + } + + if component in self.topology or self.freeform: + self.logging.info('[Controller] Component "%s" timed out' % component) + self.tracked.remove(component) + fail_handlers.get(component, universal)() + + # ------------------- + # Completion Handling + # ------------------- def done(self, component): self.logging.info('[Controller] Component "%s" done.' % component) From d9294d2ff9ca5f42cc81bf3ee36fac3286e1897e Mon Sep 17 00:00:00 2001 From: Stephen Diehl Date: Thu, 26 Apr 2012 21:21:57 -0400 Subject: [PATCH 6/7] Base class for transitions. --- zipline/transitions.py | 87 ++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 87 insertions(+) create mode 100644 zipline/transitions.py diff --git a/zipline/transitions.py b/zipline/transitions.py new file mode 100644 index 00000000..9a68926f --- /dev/null +++ b/zipline/transitions.py @@ -0,0 +1,87 @@ +import types +from collections import Container, Hashable, Callable + +class Any(object): pass + +class Workflow(Container, Callable): + + def __init__(self, states, transitions, initial_state): + self.simple = set() + self.complx = [] + + if isinstance(states[0], tuple): + self.groups = {b for _,b in states} + else: + self.groups = set() + + matcher = lambda b: lambda f,t : t == b + + for (a, b) in transitions.itervalues(): + if a is Any: + self.complx.append(matcher(b)) + if isinstance(a, Hashable) and isinstance(b, Hashable): + self.simple.add((a,b)) + + def __call__(self, **kwargs): + if 'group' in kwargs: + return self.groups + + def __contains__(self, state): + if state in self.simple: + return True + for match in self.complx: + if match(*state): + return True + else: + return False + +class Flowable: + + @property + def state(self): + if not hasattr(self, '_state'): + self._state = self.initial_state + else: + return self._state + + @state.setter + def state(self, new): + if not hasattr(self, '_state'): + self._state = self.initial_state + + old = self._state + + if (old, new) in self.workflow: + self._state = new + else: + raise RuntimeError("Invalid State Transition : %s -> %s" %(old, new)) + +class WorkflowMeta(type): + """ + Base metaclass component workflows. + """ + + def __new__(cls, name, mro, attrs): + + state = attrs.get('states', None) + transitions = attrs.get('transitions', None) + initial_state = attrs.get('initial_state', None) + + if attrs.get('workflow'): + raise RuntimeError('`workflow` is a reserved attribute.') + + if not state: + raise RuntimeError('Must specify states') + + if not transitions: + raise RuntimeError('Must specify transitions') + + if not transitions: + raise RuntimeError('Must specify initial_state') + + new_class = super(WorkflowMeta, cls).__new__( + cls, name, mro+(Flowable,), attrs + ) + new_class.workflow = Workflow(state, transitions, initial_state) + + return new_class From 0cf48698b814c18b93b6a3c618071e852a51075e Mon Sep 17 00:00:00 2001 From: Stephen Diehl Date: Fri, 27 Apr 2012 15:56:30 -0400 Subject: [PATCH 7/7] Dispatch cancellation in REQ/REP --- zipline/monitor.py | 60 +++++++++++++++++++++++++++++++++++++--------- 1 file changed, 49 insertions(+), 11 deletions(-) diff --git a/zipline/monitor.py b/zipline/monitor.py index 3f4af9fc..6f72989b 100644 --- a/zipline/monitor.py +++ b/zipline/monitor.py @@ -165,9 +165,8 @@ class Controller(object): self.zmq = None self.zmq_poller = None - polling = False - - self.polling = polling + self.running = False + self.polling = False self.tracked = set() self.responses = set() @@ -246,6 +245,7 @@ class Controller(object): self.logging.info("[Controller] State Transition : %s -> %s" %(old, new)) def run(self): + self.running = True self.init_zmq(self.zmq_flavor) try: @@ -272,6 +272,9 @@ class Controller(object): # ------------- def send_heart(self): + if not self.running: + return + heartbeat_frame = CONTROL_FRAME( CONTROL_PROTOCOL.HEARTBEAT, str(self.ctime) @@ -279,6 +282,9 @@ class Controller(object): self.pub.send(heartbeat_frame) def send_hardkill(self): + if not self.running: + return + kill_frame = CONTROL_FRAME( CONTROL_PROTOCOL.KILL, '' @@ -286,6 +292,9 @@ class Controller(object): self.pub.send(kill_frame) def send_softkill(self): + if not self.running: + return + soft_frame = CONTROL_FRAME( CONTROL_PROTOCOL.SHUTDOWN, '' @@ -298,17 +307,34 @@ class Controller(object): def _poll(self): + assert self.route_socket + assert self.pub_socket + assert self.cancel_socket + + # -- Publish -- + # ============= self.pub = self.context.socket(self.zmq.PUB) self.pub.bind(self.pub_socket) + # -- Cancel -- + # ============= + assert isinstance(self.cancel_socket,basestring), self.cancel_socket + self.cancel = self.context.socket(self.zmq.REP) + self.cancel.connect(self.cancel_socket) + + # -- Router -- + # ============= self.router = self.context.socket(self.zmq.ROUTER) self.router.bind(self.route_socket) - self.associated.extend([self.pub, self.router]) poller = self.zmq.Poller() poller.register(self.router, self.zmq.POLLIN) + poller.register(self.cancel, self.zmq.POLLIN) + self.associated += [self.pub, self.router, self.cancel] + + # TODO: actually do this self.state = CONTROL_STATES.SOURCES_READY buffer = [] @@ -343,6 +369,12 @@ class Controller(object): self.logging.error('Invalid frame', rawmessage) pass + if self.cancel in socks and socks[self.cancel] == self.zmq.POLLIN: + self.logging.info('[Controller] Received Cancellation') + rawmessage = self.cancel.recv() + self.shutdown(soft=True) + break + self.beat() if self.zmq_flavor == 'green': @@ -394,6 +426,9 @@ class Controller(object): # The various "states of being that a component can inform us # of def new(self, component): + if self.state is CONTROL_STATES.TERMINATE: + return + self.logging.info('[Controller] Now Tracking "%s" ' % component) universal = self.new_universal @@ -413,14 +448,18 @@ class Controller(object): # Epic Fail Handling # ------------------ - def universal(self): - self.logging.error('[Controller] System in exception state, shutting down') - self.terminate(soft=True) + def fail_universal(self): + pass + # TODO: this requires higher order functionality + #self.logging.error('[Controller] System in exception state, shutting down') + #self.shutdown(soft=True) def fail(self, component): + if self.state is CONTROL_STATES.TERMINATE: + return + universal = self.fail_universal - fail_handlers = { - } + fail_handlers = { } if component in self.topology or self.freeform: self.logging.info('[Controller] Component "%s" timed out' % component) @@ -447,8 +486,7 @@ class Controller(object): def exception(self, component, failure): universal = self.exception_universal - exception_handlers = { - } + exception_handlers = { } if component in self.topology or self.freeform: self.error_replay[(component, time.time())] = failure