of: add echo request support

Add a new feature: an echo request loop, which implements a liveness
check for connected Datapaths. This feature is off by default, but can
be enabled by setting the config file option
maximum_unreplied_echo_requests to a positive integer value.

Signed-off-by: Victor J. Orlikowski <vjo@duke.edu>
Signed-off-by: FUJITA Tomonori <fujita.tomonori@lab.ntt.co.jp>

2 files changed, 40 insertions, 2 deletions

diff --git a/ryu/controller/controller.py b/ryu/controller/controller.py
index 90a800ee..9a009446 100644
--- a/ryu/controller/controller.py
+++ b/ryu/controller/controller.py
@@ -62,7 +62,14 @@ CONF.register_cli_opts([
 CONF.register_opts([
     cfg.FloatOpt('socket-timeout',
                  default=5.0,
-                 help='Time, in seconds, to await completion of socket operations.')
+                 help='Time, in seconds, to await completion of socket operations.'),
+    cfg.FloatOpt('echo-request-interval',
+                 default=15.0,
+                 help='Time, in seconds, between sending echo requests to a datapath.'),
+    cfg.IntOpt('maximum-unreplied-echo-requests',
+               default=0,
+               min=0,
+               help='Maximum number of unreplied echo requests before datapath is disconnected.')
 ])
 
 
@@ -136,6 +143,10 @@ class Datapath(ofproto_protocol.ProtocolDesc):
         self.send_q = hub.Queue(16)
         self._send_q_sem = hub.BoundedSemaphore(self.send_q.maxsize)
 
+        self.echo_request_interval = CONF.echo_request_interval
+        self.max_unreplied_echo_requests = CONF.maximum_unreplied_echo_requests
+        self.unreplied_echo_requests = []
+
         self.xid = random.randint(0, self.ofproto.MAX_XID)
         self.id = None  # datapath_id is unknown yet
         self._ports = None
@@ -279,6 +290,23 @@ class Datapath(ofproto_protocol.ProtocolDesc):
         # LOG.debug('send_msg %s', msg)
         self.send(msg.buf)
 
+    def _echo_request_loop(self):
+        if not self.max_unreplied_echo_requests:
+            return
+        while (self.send_q and
+               (len(self.unreplied_echo_requests) <= self.max_unreplied_echo_requests)):
+            echo_req = self.ofproto_parser.OFPEchoRequest(self)
+            self.unreplied_echo_requests.append(self.set_xid(echo_req))
+            self.send_msg(echo_req)
+            hub.sleep(self.echo_request_interval)
+        self.close()
+
+    def acknowledge_echo_reply(self, xid):
+        try:
+            self.unreplied_echo_requests.remove(xid)
+        except:
+            pass
+
     def serve(self):
         send_thr = hub.spawn(self._send_loop)
 
@@ -286,11 +314,14 @@ class Datapath(ofproto_protocol.ProtocolDesc):
         hello = self.ofproto_parser.OFPHello(self)
         self.send_msg(hello)
 
+        echo_thr = hub.spawn(self._echo_request_loop)
+
         try:
             self._recv_loop()
         finally:
             hub.kill(send_thr)
-            hub.joinall([send_thr])
+            hub.kill(echo_thr)
+            hub.joinall([send_thr, echo_thr])
             self.is_active = False
 
     #
diff --git a/ryu/controller/ofp_handler.py b/ryu/controller/ofp_handler.py
index 7ec1c955..edff696d 100644
--- a/ryu/controller/ofp_handler.py
+++ b/ryu/controller/ofp_handler.py
@@ -238,6 +238,13 @@ class OFPHandler(ryu.base.app_manager.RyuApp):
         echo_reply.data = msg.data
         datapath.send_msg(echo_reply)
 
+    @set_ev_handler(ofp_event.EventOFPEchoReply,
+                    [HANDSHAKE_DISPATCHER, CONFIG_DISPATCHER, MAIN_DISPATCHER])
+    def echo_reply_handler(self, ev):
+        msg = ev.msg
+        datapath = msg.datapath
+        datapath.acknowledge_echo_reply(msg.xid)
+
     @set_ev_handler(ofp_event.EventOFPErrorMsg,
                     [HANDSHAKE_DISPATCHER, CONFIG_DISPATCHER, MAIN_DISPATCHER])
     def error_msg_handler(self, ev):