protocols/ovsdb: Add bulk read support

At times is it convient to read from all clients at the same time. In
particular when searching for what system_id a port or bridge (datapath)
belongs to.

Signed-off-by: Jason Kölker <jason@koelker.net>
Signed-off-by: FUJITA Tomonori <fujita.tomonori@lab.ntt.co.jp>

2 files changed, 38 insertions, 1 deletions

diff --git a/ryu/services/protocols/ovsdb/client.py b/ryu/services/protocols/ovsdb/client.py
index 17806383..7c20eb2a 100644
--- a/ryu/services/protocols/ovsdb/client.py
+++ b/ryu/services/protocols/ovsdb/client.py
@@ -373,8 +373,15 @@ class RemoteOvsdb(app_manager.RyuApp):
     def modify_request_handler(self, ev):
         self._txn_q.append(ev)
 
-    def read_request_handler(self, ev):
+    def read_request_handler(self, ev, bulk=False):
         result = ev.func(self._idl.tables)
+
+        # NOTE(jkoelker) If this was a bulk request, the parent OVSDB app is
+        #                responsible for the reply
+
+        if bulk:
+            return (self.system_id, result)
+
         rep = event.EventReadReply(self.system_id, result)
         self.reply_to_request(ev, rep)
 
diff --git a/ryu/services/protocols/ovsdb/manager.py b/ryu/services/protocols/ovsdb/manager.py
index 0c631973..a2b8a7e2 100644
--- a/ryu/services/protocols/ovsdb/manager.py
+++ b/ryu/services/protocols/ovsdb/manager.py
@@ -88,6 +88,26 @@ class OVSDB(app_manager.RyuApp):
             t = hub.spawn(self._start_remote, sock, client_address)
             self.threads.append(t)
 
+    def _bulk_read_handler(self, ev):
+        results = []
+
+        def done(gt, *args, **kwargs):
+            if gt in self.threads:
+                self.threads.remove(gt)
+
+            results.append(gt.wait())
+
+        threads = []
+        for c in self._clients.values():
+            gt = hub.spawn(c.read_request_handler, ev, bulk=True)
+            threads.append(gt)
+            self.threads.append(gt)
+            gt.link(done)
+
+        hub.joinall(threads)
+        rep = event.EventReadReply(None, results)
+        self.reply_to_request(ev, rep)
+
     def _proxy_event(self, ev):
         system_id = ev.system_id
         client_name = client.RemoteOvsdb.instance_name(system_id)
@@ -174,6 +194,16 @@ class OVSDB(app_manager.RyuApp):
     @handler.set_ev_cls(event.EventReadRequest)
     def read_request_handler(self, ev):
         system_id = ev.system_id
+
+        if system_id is None:
+            def done(gt, *args, **kwargs):
+                if gt in self.threads:
+                    self.threads.remove(gt)
+
+            thread = hub.spawn(self._bulk_read_handler, ev)
+            self.threads.append(thread)
+            return thread.link(done)
+
         client_name = client.RemoteOvsdb.instance_name(system_id)
         remote = self._clients.get(client_name)