tools/net/ynl: improve async notification handling

The notification handling in ynl is currently very simple, using sleep()
to wait a period of time and then handling all the buffered messages in
a single batch.

This patch changes the notification handling so that messages are
processed as they are received. This makes it possible to use ynl as a
library that supplies notifications in a timely manner.

- Change check_ntf() to be a generator that yields 1 notification at a
  time and blocks until a notification is available.
- Use the --sleep parameter to set an alarm and exit when it fires.

This means that the CLI has the same interface, but notifications get
printed as they are received:

./tools/net/ynl/cli.py --spec <SPEC> --subscribe <TOPIC> [ --sleep <SECS> ]

Here is an example python snippet that shows how to use ynl as a library
for receiving notifications:

    ynl = YnlFamily(f"{dir}/rt_route.yaml")
    ynl.ntf_subscribe('rtnlgrp-ipv4-route')

    for event in ynl.check_ntf():
        handle(event)

Signed-off-by: Donald Hunter <[email protected]>
Tested-by: Kory Maincent <[email protected]>
Link: https://patch.msgid.link/[email protected]
Signed-off-by: Paolo Abeni <[email protected]>

Author: donaldh

tools/net/ynl/cli.py

@@ -5,6 +5,7 @@
 import json
 import pprint
 import time
+import signal
 
 from lib import YnlFamily, Netlink, NlError
 
@@ -17,6 +18,8 @@ def default(self, obj):
             return list(obj)
         return json.JSONEncoder.default(self, obj)
 
+def handle_timeout(sig, frame):
+    exit(0)
 
 def main():
     description = """
@@ -81,7 +84,8 @@ def output(msg):
         ynl.ntf_subscribe(args.ntf)
 
     if args.sleep:
-        time.sleep(args.sleep)
+        signal.signal(signal.SIGALRM, handle_timeout)
+        signal.alarm(args.sleep)
 
     if args.list_ops:
         for op_name, op in ynl.ops.items():
@@ -106,8 +110,8 @@ def output(msg):
         exit(1)
 
     if args.ntf:
-        ynl.check_ntf()
-        output(ynl.async_msg_queue)
+        for msg in ynl.check_ntf():
+            output(msg)
 
 
 if __name__ == "__main__":

tools/net/ynl/lib/ynl.py

@@ -12,6 +12,8 @@
 import yaml
 import ipaddress
 import uuid
+import queue
+import time
 
 from .nlspec import SpecFamily
 
@@ -489,7 +491,7 @@ def __init__(self, def_path, schema=None, process_unknown=False,
         self.sock.setsockopt(Netlink.SOL_NETLINK, Netlink.NETLINK_GET_STRICT_CHK, 1)
 
         self.async_msg_ids = set()
-        self.async_msg_queue = []
+        self.async_msg_queue = queue.Queue()
 
         for msg in self.msgs.values():
             if msg.is_async:
@@ -903,32 +905,39 @@ def handle_ntf(self, decoded):
 
         msg['name'] = op['name']
         msg['msg'] = attrs
-        self.async_msg_queue.append(msg)
+        self.async_msg_queue.put(msg)
 
-    def check_ntf(self):
+    def check_ntf(self, interval=0.1):
         while True:
             try:
                 reply = self.sock.recv(self._recv_size, socket.MSG_DONTWAIT)
-            except BlockingIOError:
-                return
+                nms = NlMsgs(reply)
+                self._recv_dbg_print(reply, nms)
+                for nl_msg in nms:
+                    if nl_msg.error:
+                        print("Netlink error in ntf!?", os.strerror(-nl_msg.error))
+                        print(nl_msg)
+                        continue
+                    if nl_msg.done:
+                        print("Netlink done while checking for ntf!?")
+                        continue
 
-            nms = NlMsgs(reply)
-            self._recv_dbg_print(reply, nms)
-            for nl_msg in nms:
-                if nl_msg.error:
-                    print("Netlink error in ntf!?", os.strerror(-nl_msg.error))
-                    print(nl_msg)
-                    continue
-                if nl_msg.done:
-                    print("Netlink done while checking for ntf!?")
-                    continue
+                    decoded = self.nlproto.decode(self, nl_msg, None)
+                    if decoded.cmd() not in self.async_msg_ids:
+                        print("Unexpected msg id while checking for ntf", decoded)
+                        continue
 
-                decoded = self.nlproto.decode(self, nl_msg, None)
-                if decoded.cmd() not in self.async_msg_ids:
-                    print("Unexpected msg id done while checking for ntf", decoded)
-                    continue
+                    self.handle_ntf(decoded)
+            except BlockingIOError:
+                pass
 
-                self.handle_ntf(decoded)
+            try:
+                yield self.async_msg_queue.get_nowait()
+            except queue.Empty:
+                try:
+                    time.sleep(interval)
+                except KeyboardInterrupt:
+                    return
 
     def operation_do_attributes(self, name):
       """