replace dispatcher mechanism

This is purely internal change and no API for applications is
changed. At least, I confirmed that folsom OpenStack plugin works.

With the current dispatcher mechanism, multiple greenlets call
applications' handlers and might be blocked anywhere so we need
various locks to handle that concurrency. This makes things difficult
for application developers.

With this patch, each applications are connected with events. Each
application has the own greenlet(s) to handle events and might send
events to other applications.

If an application registers handlers for some OF events, it subscribes
to OF component and registers the OF events that it's interested. OF
application delivers such OF events to the application and the
application's greenlet executes the handlers.

With this, we can completely remove dispatcher.py and its friends.

Signed-off-by: FUJITA Tomonori <fujita.tomonori@lab.ntt.co.jp>

6 files changed, 129 insertions, 85 deletions

diff --git a/ryu/base/app_manager.py b/ryu/base/app_manager.py
index 4a29c95d..781798ed 100644
--- a/ryu/base/app_manager.py
+++ b/ryu/base/app_manager.py
@@ -17,6 +17,9 @@
 import inspect
 import itertools
 import logging
+import gevent
+
+from gevent.queue import Queue
 
 from ryu import utils
 from ryu.controller.handler import register_instance
@@ -24,20 +27,11 @@ from ryu.controller.controller import Datapath
 
 LOG = logging.getLogger('ryu.base.app_manager')
 
+SERVICE_BRICKS = {}
 
-class RyuAppContext(object):
-    """
-    Base class for Ryu application context
-    """
-    def __init__(self):
-        super(RyuAppContext, self).__init__()
 
-    def close(self):
-        """
-        teardown method
-        The method name, close, is chosen for python context manager
-        """
-        pass
+def lookup_service_brick(name):
+    return SERVICE_BRICKS.get(name)
 
 
 class RyuApp(object):
@@ -55,6 +49,45 @@ class RyuApp(object):
 
     def __init__(self, *_args, **_kwargs):
         super(RyuApp, self).__init__()
+        self.name = self.__class__.__name__
+        self.event_handlers = {}
+        self.observers = {}
+        self.threads = []
+        self.events = Queue()
+        self.threads.append(gevent.spawn(self._event_loop))
+
+    def register_handler(self, ev_cls, handler):
+        assert callable(handler)
+        self.event_handlers.setdefault(ev_cls, [])
+        self.event_handlers[ev_cls].append(handler)
+
+    def register_observer(self, ev_cls, name):
+        self.observers.setdefault(ev_cls, [])
+        self.observers[ev_cls].append(name)
+
+    def get_handlers(self, ev):
+        return self.event_handlers.get(ev.__class__, [])
+
+    def get_observers(self, ev):
+        return self.observers.get(ev.__class__, [])
+
+    def _event_loop(self):
+        while True:
+            ev = self.events.get()
+            handlers = self.get_handlers(ev)
+            for handler in handlers:
+                handler(ev)
+
+    def _send_event(self, ev):
+        self.events.put(ev)
+
+    def send_event(self, name, ev):
+        if name in SERVICE_BRICKS:
+            SERVICE_BRICKS[name]._send_event(ev)
+
+    def send_event_to_observers(self, ev):
+        for observer in self.get_observers(ev):
+            self.send_event(observer, ev)
 
     def close(self):
         """
@@ -102,7 +135,12 @@ class AppManager(object):
 
     def create_contexts(self):
         for key, cls in self.contexts_cls.items():
-            self.contexts[key] = cls()
+            context = cls()
+            self.contexts[key] = context
+            # hack for dpset
+            if context.__class__.__base__ == RyuApp:
+                SERVICE_BRICKS[context.name] = context
+                register_instance(context)
         return self.contexts
 
     def instantiate_apps(self, *args, **kwargs):
@@ -124,6 +162,15 @@ class AppManager(object):
             app = cls(*args, **kwargs)
             register_instance(app)
             self.applications[app_name] = app
+            SERVICE_BRICKS[app.name] = app
+
+        for key, i in SERVICE_BRICKS.items():
+            for _k, m in inspect.getmembers(i, inspect.ismethod):
+                if hasattr(m, 'observer'):
+                    name = m.observer.split('.')[-1]
+                    if name in SERVICE_BRICKS:
+                        brick = SERVICE_BRICKS[name]
+                        brick.register_observer(m.ev_cls, i.name)
 
     def close(self):
         def close_all(close_dict):
diff --git a/ryu/controller/controller.py b/ryu/controller/controller.py
index 1811d8ca..7c33e4a3 100644
--- a/ryu/controller/controller.py
+++ b/ryu/controller/controller.py
@@ -25,6 +25,8 @@ import ssl
 from gevent.server import StreamServer
 from gevent.queue import Queue
 
+import ryu.base.app_manager
+
 from ryu.ofproto import ofproto_common
 from ryu.ofproto import ofproto_parser
 from ryu.ofproto import ofproto_v1_0
@@ -35,7 +37,6 @@ from ryu.ofproto import ofproto_v1_3
 from ryu.ofproto import ofproto_v1_3_parser
 from ryu.ofproto import nx_match
 
-from ryu.controller import dispatcher
 from ryu.controller import handler
 from ryu.controller import ofp_event
 
@@ -123,25 +124,22 @@ class Datapath(object):
         # prevent it from eating memory up
         self.send_q = Queue(16)
 
-        # circular reference self.ev_q.aux == self
-        self.ev_q = dispatcher.EventQueue(handler.QUEUE_NAME_OFP_MSG,
-                                          handler.HANDSHAKE_DISPATCHER,
-                                          self)
-
         self.set_version(max(self.supported_ofp_version))
         self.xid = random.randint(0, self.ofproto.MAX_XID)
         self.id = None  # datapath_id is unknown yet
         self.ports = None
         self.flow_format = ofproto_v1_0.NXFF_OPENFLOW10
+        self.ofp_brick = ryu.base.app_manager.lookup_service_brick('ofp_event')
+        self.set_state(handler.HANDSHAKE_DISPATCHER)
 
     def close(self):
-        """
-        Call this before discarding this datapath object
-        The circular refernce as self.ev_q.aux == self must be broken.
-        """
-        # tell this datapath is dead
-        self.ev_q.set_dispatcher(handler.DEAD_DISPATCHER)
-        self.ev_q.close()
+        self.set_state(handler.DEAD_DISPATCHER)
+
+    def set_state(self, state):
+        self.state = state
+        ev = ofp_event.EventOFPStateChange(self)
+        ev.state = state
+        self.ofp_brick.send_event_to_observers(ev)
 
     def set_version(self, version):
         assert version in self.supported_ofp_version
@@ -169,7 +167,13 @@ class Datapath(object):
                 msg = ofproto_parser.msg(self,
                                          version, msg_type, msg_len, xid, buf)
                 #LOG.debug('queue msg %s cls %s', msg, msg.__class__)
-                self.ev_q.queue(ofp_event.ofp_msg_to_ev(msg))
+                ev = ofp_event.ofp_msg_to_ev(msg)
+                handlers = self.ofp_brick.get_handlers(ev)
+                for handler in handlers:
+                    if self.state in handler.dispatchers:
+                        handler(ev)
+
+                self.ofp_brick.send_event_to_observers(ev)
 
                 buf = buf[required_len:]
                 required_len = ofproto_common.OFP_HEADER_SIZE
@@ -219,10 +223,6 @@ class Datapath(object):
             gevent.kill(send_thr)
             gevent.joinall([send_thr])
 
-    def send_ev(self, ev):
-        #LOG.debug('send_ev %s', ev)
-        self.ev_q.queue(ev)
-
     #
     # Utility methods for convenience
     #
diff --git a/ryu/controller/dpset.py b/ryu/controller/dpset.py
index b6883b54..c960906c 100644
--- a/ryu/controller/dpset.py
+++ b/ryu/controller/dpset.py
@@ -16,8 +16,9 @@
 
 import logging
 
+from ryu.base import app_manager
 from ryu.controller import event
-from ryu.controller import dispatcher
+from ryu.controller import ofp_event
 from ryu.controller import dp_type
 from ryu.controller import handler
 from ryu.controller.handler import set_ev_cls
@@ -25,9 +26,7 @@ from ryu.controller.handler import set_ev_cls
 LOG = logging.getLogger('ryu.controller.dpset')
 
 
-QUEUE_NAME_DPSET = 'dpset'
-DISPATCHER_NAME_DPSET = 'dpset'
-DPSET_EV_DISPATCHER = dispatcher.EventDispatcher(DISPATCHER_NAME_DPSET)
+DPSET_EV_DISPATCHER = 'dpset'
 
 
 class EventDPBase(event.EventBase):
@@ -46,18 +45,16 @@ class EventDP(EventDPBase):
 
 
 # this depends on controller::Datapath and dispatchers in handler
-class DPSet(object):
+class DPSet(app_manager.RyuApp):
     def __init__(self):
         super(DPSet, self).__init__()
+        self.name = 'dpset'
 
         # dp registration and type setting can be occur in any order
         # Sometimes the sw_type is set before dp connection
         self.dp_types = {}
 
         self.dps = {}   # datapath_id => class Datapath
-        self.ev_q = dispatcher.EventQueue(QUEUE_NAME_DPSET,
-                                          DPSET_EV_DISPATCHER)
-        handler.register_instance(self)
 
     def register(self, dp):
         assert dp.id is not None
@@ -68,7 +65,7 @@ class DPSet(object):
             dp.dp_type = dp_type_
 
         self.dps[dp.id] = dp
-        self.ev_q.queue(EventDP(dp, True))
+        self.send_event_to_observers(EventDP(dp, True))
 
     def unregister(self, dp):
         if dp.id in self.dps:
@@ -76,7 +73,7 @@ class DPSet(object):
             assert dp.id not in self.dp_types
             self.dp_types[dp.id] = getattr(dp, 'dp_type', dp_type.UNKNOWN)
 
-            self.ev_q.queue(EventDP(dp, False))
+            self.send_event_to_observers(EventDP(dp, False))
 
     def set_type(self, dp_id, dp_type_=dp_type.UNKNOWN):
         if dp_id in self.dps:
@@ -92,19 +89,14 @@ class DPSet(object):
     def get_all(self):
         return self.dps.items()
 
-    @set_ev_cls(dispatcher.EventDispatcherChange,
-                dispatcher.QUEUE_EV_DISPATCHER)
+    @set_ev_cls(ofp_event.EventOFPStateChange,
+                [handler.MAIN_DISPATCHER, handler.DEAD_DISPATCHER])
     def dispacher_change(self, ev):
-        LOG.debug('dispatcher change q %s dispatcher %s',
-                  ev.ev_q.name, ev.new_dispatcher.name)
-        if ev.ev_q.name != handler.QUEUE_NAME_OFP_MSG:
-            return
-
-        datapath = ev.ev_q.aux
+        datapath = ev.datapath
         assert datapath is not None
-        if ev.new_dispatcher.name == handler.DISPATCHER_NAME_OFP_MAIN:
+        if ev.state == handler.MAIN_DISPATCHER:
             LOG.debug('DPSET: register datapath %s', datapath)
             self.register(datapath)
-        elif ev.new_dispatcher.name == handler.DISPATCHER_NAME_OFP_DEAD:
+        elif ev.state == handler.DEAD_DISPATCHER:
             LOG.debug('DPSET: unregister datapath %s', datapath)
             self.unregister(datapath)
diff --git a/ryu/controller/handler.py b/ryu/controller/handler.py
index ede27df0..9a09e99b 100644
--- a/ryu/controller/handler.py
+++ b/ryu/controller/handler.py
@@ -14,36 +14,39 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-import copy
 import inspect
 import logging
 
-from ryu.controller import dispatcher
 from ryu.controller import ofp_event
 
 LOG = logging.getLogger('ryu.controller.handler')
 
-QUEUE_NAME_OFP_MSG = 'ofp_msg'
-DISPATCHER_NAME_OFP_HANDSHAKE = 'ofp_handshake'
-HANDSHAKE_DISPATCHER = dispatcher.EventDispatcher(
-    DISPATCHER_NAME_OFP_HANDSHAKE)
-DISPATCHER_NAME_OFP_CONFIG = 'ofp_config'
-CONFIG_DISPATCHER = dispatcher.EventDispatcher(DISPATCHER_NAME_OFP_CONFIG)
-DISPATCHER_NAME_OFP_MAIN = 'ofp_main'
-MAIN_DISPATCHER = dispatcher.EventDispatcher(DISPATCHER_NAME_OFP_MAIN)
-DISPATCHER_NAME_OFP_DEAD = 'ofp_dead'
-DEAD_DISPATCHER = dispatcher.EventDispatcher(DISPATCHER_NAME_OFP_DEAD)
+# just represent OF datapath state. datapath specific so should be moved.
+HANDSHAKE_DISPATCHER = "handshake"
+CONFIG_DISPATCHER = "config"
+MAIN_DISPATCHER = "main"
+DEAD_DISPATCHER = "dead"
 
 
+# should be named something like 'observe_event'
 def set_ev_cls(ev_cls, dispatchers):
     def _set_ev_cls_dec(handler):
         handler.ev_cls = ev_cls
-        handler.dispatchers = dispatchers
+        handler.dispatchers = _listify(dispatchers)
+        handler.observer = ev_cls.__module__
         return handler
     return _set_ev_cls_dec
 
 
-def _is_ev_handler(meth):
+def set_ev_handler(ev_cls, dispatchers):
+    def _set_ev_cls_dec(handler):
+        handler.ev_cls = ev_cls
+        handler.dispatchers = _listify(dispatchers)
+        return handler
+    return _set_ev_cls_dec
+
+
+def _is_ev_cls(meth):
     return hasattr(meth, 'ev_cls')
 
 
@@ -58,12 +61,5 @@ def _listify(may_list):
 def register_instance(i):
     for _k, m in inspect.getmembers(i, inspect.ismethod):
         # LOG.debug('instance %s k %s m %s', i, _k, m)
-        if not _is_ev_handler(m):
-            continue
-
-        _dispatchers = _listify(getattr(m, 'dispatchers', None))
-        # LOG.debug("_dispatchers %s", _dispatchers)
-        for d in _dispatchers:
-            # LOG.debug('register dispatcher %s ev %s k %s m %s',
-            #           d.name, m.ev_cls, _k, m)
-            d.register_handler(m.ev_cls, m)
+        if _is_ev_cls(m):
+            i.register_handler(m.ev_cls, m)
diff --git a/ryu/controller/ofp_event.py b/ryu/controller/ofp_event.py
index e5becac0..91146041 100644
--- a/ryu/controller/ofp_event.py
+++ b/ryu/controller/ofp_event.py
@@ -73,3 +73,9 @@ _PARSER_MODULE_LIST = ['ryu.ofproto.ofproto_v1_0_parser',
 for m in _PARSER_MODULE_LIST:
     # print 'loading module %s' % m
     _create_ofp_msg_ev_from_module(m)
+
+
+class EventOFPStateChange(event.EventBase):
+    def __init__(self, dp):
+        super(EventOFPStateChange, self).__init__()
+        self.datapath = dp
diff --git a/ryu/controller/ofp_handler.py b/ryu/controller/ofp_handler.py
index add03013..3ce91989 100644
--- a/ryu/controller/ofp_handler.py
+++ b/ryu/controller/ofp_handler.py
@@ -16,10 +16,12 @@
 
 import logging
 
+import ryu.base.app_manager
+
 from ryu import utils
-from ryu.base import app_manager
+from ryu.controller import handler
 from ryu.controller import ofp_event
-from ryu.controller.handler import set_ev_cls
+from ryu.controller.handler import set_ev_cls, set_ev_handler
 from ryu.controller.handler import HANDSHAKE_DISPATCHER, CONFIG_DISPATCHER,\
     MAIN_DISPATCHER
 
@@ -39,9 +41,10 @@ LOG = logging.getLogger('ryu.controller.ofp_handler')
 # back Echo Reply message.
 
 
-class OFPHandler(app_manager.RyuApp):
+class OFPHandler(ryu.base.app_manager.RyuApp):
     def __init__(self, *args, **kwargs):
         super(OFPHandler, self).__init__(*args, **kwargs)
+        self.name = 'ofp_event'
 
     @staticmethod
     def hello_failed(datapath, error_desc):
@@ -52,7 +55,7 @@ class OFPHandler(app_manager.RyuApp):
         error_msg.data = error_desc
         datapath.send_msg(error_msg)
 
-    @set_ev_cls(ofp_event.EventOFPHello, HANDSHAKE_DISPATCHER)
+    @set_ev_handler(ofp_event.EventOFPHello, HANDSHAKE_DISPATCHER)
     def hello_handler(self, ev):
         LOG.debug('hello ev %s', ev)
         msg = ev.msg
@@ -121,9 +124,9 @@ class OFPHandler(app_manager.RyuApp):
 
         # now move on to config state
         LOG.debug('move onto config mode')
-        datapath.ev_q.set_dispatcher(CONFIG_DISPATCHER)
+        datapath.set_state(CONFIG_DISPATCHER)
 
-    @set_ev_cls(ofp_event.EventOFPSwitchFeatures, CONFIG_DISPATCHER)
+    @set_ev_handler(ofp_event.EventOFPSwitchFeatures, CONFIG_DISPATCHER)
     def switch_features_handler(self, ev):
         msg = ev.msg
         datapath = msg.datapath
@@ -147,10 +150,10 @@ class OFPHandler(app_manager.RyuApp):
         datapath.send_msg(set_config)
 
         LOG.debug('move onto main mode')
-        ev.msg.datapath.ev_q.set_dispatcher(MAIN_DISPATCHER)
+        ev.msg.datapath.set_state(MAIN_DISPATCHER)
 
-    @set_ev_cls(ofp_event.EventOFPEchoRequest,
-                [HANDSHAKE_DISPATCHER, CONFIG_DISPATCHER, MAIN_DISPATCHER])
+    @set_ev_handler(ofp_event.EventOFPEchoRequest,
+                    [HANDSHAKE_DISPATCHER, CONFIG_DISPATCHER, MAIN_DISPATCHER])
     def echo_request_handler(self, ev):
         msg = ev.msg
         datapath = msg.datapath
@@ -159,8 +162,8 @@ class OFPHandler(app_manager.RyuApp):
         echo_reply.data = msg.data
         datapath.send_msg(echo_reply)
 
-    @set_ev_cls(ofp_event.EventOFPErrorMsg,
-                [HANDSHAKE_DISPATCHER, CONFIG_DISPATCHER, MAIN_DISPATCHER])
+    @set_ev_handler(ofp_event.EventOFPErrorMsg,
+                    [HANDSHAKE_DISPATCHER, CONFIG_DISPATCHER, MAIN_DISPATCHER])
     def error_msg_handler(self, ev):
         msg = ev.msg
         LOG.debug('error msg ev %s type 0x%x code 0x%x %s',