wassname/catalyst
1
0
Fork 0
mirror of https://github.com/wassname/catalyst.git synced 2026-07-15 11:22:18 +08:00
Code Issues Packages Projects Releases Wiki Activity

FSM actions for Feed and Merge.

Browse Source
This commit is contained in:
Stephen Diehl
2012-06-11 13:10:05 -04:00
parent fefca77fe2
commit 270e067426
4 changed files with 70 additions and 57 deletions
Download Patch File Download Diff File Expand all files Collapse all files
zipline/components/aggregator.py
+26 -9
View File
@@ -9,16 +9,30 @@ Abstract base class for Feed and Merge.
Feed Merge
"""
import logging
from collections import Counter
import logbook
import zipline.protocol as zp
from zipline.core.component import Component
from zipline.protocol import CONTROL_PROTOCOL, COMPONENT_TYPE, \
CONTROL_FRAME, CONTROL_UNFRAME
from zipline.utils.protocol_utils import Enum
from zipline.transitions import WorkflowMeta
LOGGER = logging.getLogger('ZiplineLogger')
log = logbook.Logger('Aggregate')
# =================
# State Transitions
# =================
INIT, READY, DRAINING = AGGREGATE_STATES = \
Enum( 'INIT', 'READY', 'DRAINING')
AGGREGATE_TRANSITIONS = dict(
do_start = (-1 , INIT) ,
do_run = (INIT , READY) ,
do_drain = (READY , DRAINING) ,
)
class Aggregate(Component):
"""
@@ -32,6 +46,9 @@ class Aggregate(Component):
Feed and Merge define these differently.
"""
abstract = True
__metaclass__ = WorkflowMeta
@property
def get_type(self):
return COMPONENT_TYPE.CONDUIT
@@ -76,22 +93,21 @@ class Aggregate(Component):
if len(self.data_buffer) == self.ds_finished_counter:
#drain any remaining messages in the buffer
LOGGER.debug("draining feed")
log.debug("Draining Feed")
self.drain()
self.signal_done()
else:
try:
event = self.unframe(message)
# deserialization error
except zp.INVALID_DATASOURCE_FRAME as exc:
# Error deserializing
return self.signal_exception(exc)
try:
self.append(event)
self.send_next()
# Invalid message
except zp.INVALID_DATASOURCE_FRAME as exc:
# Invalid message
return self.signal_exception(exc)
# -------------
@@ -102,7 +118,7 @@ class Aggregate(Component):
"""
Send all messages in the buffer.
"""
self.draining = True
self.state = DRAINING
while self.pending_messages() > 0:
self.send_next()
@@ -114,7 +130,8 @@ class Aggregate(Component):
return
event = self.next()
if(event != None):
if event:
self.feed_socket.send(self.frame(event), self.zmq.NOBLOCK)
self.sent_counters[event.source_id] += 1
self.sent_count += 1
zipline/components/feed.py
+14 -26
View File
@@ -1,28 +1,15 @@
import logging
import logbook
from collections import Counter
from zipline.core.component import Component
from zipline.components.aggregator import Aggregate
from zipline.utils.protocol_utils import Enum
from zipline.components.aggregator import Aggregate, \
AGGREGATE_STATES, AGGREGATE_TRANSITIONS
import zipline.protocol as zp
from zipline.transitions import WorkflowMeta
from zipline.protocol import CONTROL_PROTOCOL, COMPONENT_TYPE, \
CONTROL_FRAME, CONTROL_UNFRAME
log = logbook.Logger('Feed')
LOGGER = logging.getLogger('ZiplineLogger')
# FSM
# ---
INIT, READY, DRAINING = FEED_STATES = \
Enum( 'INIT', 'READY', 'DRAINING')
state_transitions = dict(
do_start = (-1 , INIT) ,
do_run = (INIT , READY) ,
do_drain = (READY , DRAINING) ,
)
# =========
# Component
# =========
class Feed(Aggregate):
"""
@@ -32,16 +19,13 @@ class Feed(Aggregate):
one execution context (thread, process, etc) and run in another.
"""
__metaclass__ = WorkflowMeta
states = list(FEED_STATES)
transitions = state_transitions
states = list(AGGREGATE_STATES)
transitions = AGGREGATE_TRANSITIONS
initial_state = -1
def init(self):
self.sent_count = 0
self.received_count = 0
self.draining = False
self.ds_finished_counter = 0
self.data_buffer = {}
@@ -50,12 +34,16 @@ class Feed(Aggregate):
self.sent_counters = Counter()
self.recv_counters = Counter()
self.state = INIT
self.state = AGGREGATE_STATES.INIT
@property
def get_id(self):
return "FEED"
@property
def draining(self):
return self.state == DRAINING
# -------
# Sockets
# -------
zipline/components/merge.py
+7 -2
View File
@@ -1,6 +1,6 @@
import zipline.protocol as zp
from zipline.components.aggregator import Aggregate
from zipline.components.aggregator import Aggregate, \
AGGREGATE_STATES, AGGREGATE_TRANSITIONS
from collections import Counter
@@ -8,6 +8,11 @@ class Merge(Aggregate):
"""
Merges multiple streams of events into single messages.
"""
states = list(AGGREGATE_STATES)
transitions = AGGREGATE_TRANSITIONS
initial_state = -1
def init(self):
self.sent_count = 0
self.received_count = 0
zipline/core/component.py
+23 -20
View File
@@ -488,6 +488,29 @@ class Component(object):
self.sockets.append(self.sync_socket)
# -----------
# FSM Actions
# -----------
#@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))
# ---------------------
# Description and Debug
# ---------------------
@@ -568,23 +591,3 @@ class Component(object):
pid = os.getpid() ,
pointer = hex(id(self)) ,
)
@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))