[PATCH net-next v1 0/2] tools/net/ynl: rework async notification handling

[PATCH net-next v1 0/2] tools/net/ynl: rework async notification handling

[Donald Hunter]

Revert patch 1bf70e6c3a53 which modified check_ntf() and instead add a
new poll_ntf() with async notification semantics.

See patch 2 for the description.

Donald Hunter (2):
  Revert "tools/net/ynl: improve async notification handling"
  tools/net/ynl: add async notification handling

 tools/net/ynl/cli.py     | 16 +++++----------
 tools/net/ynl/lib/ynl.py | 43 +++++++++++++++++++++++-----------------
 2 files changed, 30 insertions(+), 29 deletions(-)

-- 
2.47.0

[PATCH net-next v1 1/2] Revert "tools/net/ynl: improve async notification handling"

[Donald Hunter]

This reverts commit 1bf70e6c3a5346966c25e0a1ff492945b25d3f80.

This modification to check_ntf() is being reverted so that its behaviour
remains equivalent to ynl_ntf_check() in the C YNL. Instead a new
poll_ntf() will be added in a separate patch.

Signed-off-by: Donald Hunter <donald.hunter@gmail.com>
---
 tools/net/ynl/cli.py     | 10 +++-----
 tools/net/ynl/lib/ynl.py | 49 ++++++++++++++++------------------------
 2 files changed, 23 insertions(+), 36 deletions(-)

diff --git a/tools/net/ynl/cli.py b/tools/net/ynl/cli.py
index 9e95016b85b3..b8481f401376 100755
--- a/tools/net/ynl/cli.py
+++ b/tools/net/ynl/cli.py
@@ -5,7 +5,6 @@ import argparse
 import json
 import pprint
 import time
-import signal
 
 from lib import YnlFamily, Netlink, NlError
 
@@ -18,8 +17,6 @@ class YnlEncoder(json.JSONEncoder):
             return list(obj)
         return json.JSONEncoder.default(self, obj)
 
-def handle_timeout(sig, frame):
-    exit(0)
 
 def main():
     description = """
@@ -84,8 +81,7 @@ def main():
         ynl.ntf_subscribe(args.ntf)
 
     if args.sleep:
-        signal.signal(signal.SIGALRM, handle_timeout)
-        signal.alarm(args.sleep)
+        time.sleep(args.sleep)
 
     if args.list_ops:
         for op_name, op in ynl.ops.items():
@@ -110,8 +106,8 @@ def main():
         exit(1)
 
     if args.ntf:
-        for msg in ynl.check_ntf():
-            output(msg)
+        ynl.check_ntf()
+        output(ynl.async_msg_queue)
 
 
 if __name__ == "__main__":
diff --git a/tools/net/ynl/lib/ynl.py b/tools/net/ynl/lib/ynl.py
index 92f85698c50e..c22c22bf2cb7 100644
--- a/tools/net/ynl/lib/ynl.py
+++ b/tools/net/ynl/lib/ynl.py
@@ -12,8 +12,6 @@ import sys
 import yaml
 import ipaddress
 import uuid
-import queue
-import time
 
 from .nlspec import SpecFamily
 
@@ -491,7 +489,7 @@ class YnlFamily(SpecFamily):
         self.sock.setsockopt(Netlink.SOL_NETLINK, Netlink.NETLINK_GET_STRICT_CHK, 1)
 
         self.async_msg_ids = set()
-        self.async_msg_queue = queue.Queue()
+        self.async_msg_queue = []
 
         for msg in self.msgs.values():
             if msg.is_async:
@@ -905,39 +903,32 @@ class YnlFamily(SpecFamily):
 
         msg['name'] = op['name']
         msg['msg'] = attrs
-        self.async_msg_queue.put(msg)
+        self.async_msg_queue.append(msg)
 
-    def check_ntf(self, interval=0.1):
+    def check_ntf(self):
         while True:
             try:
                 reply = self.sock.recv(self._recv_size, socket.MSG_DONTWAIT)
-                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
+            except BlockingIOError:
+                return
 
-                    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
+            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
 
-                    self.handle_ntf(decoded)
-            except BlockingIOError:
-                pass
+                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
 
-            try:
-                yield self.async_msg_queue.get_nowait()
-            except queue.Empty:
-                try:
-                    time.sleep(interval)
-                except KeyboardInterrupt:
-                    return
+                self.handle_ntf(decoded)
 
     def operation_do_attributes(self, name):
       """
-- 
2.47.0

[PATCH net-next v1 2/2] tools/net/ynl: add async notification handling

[Donald Hunter]

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 adds async notification handling so that messages can be
processed as they are received. This makes it possible to use ynl as a
library that supplies notifications in a timely manner.

- Add poll_ntf() to be a generator that yields 1 notification at a
  time and blocks until a notification is available.
- Add a --duration parameter to the CLI, with --sleep as an alias.

./tools/net/ynl/cli.py \
    --spec <SPEC> --subscribe <TOPIC> [ --duration <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.poll_ntf():
        handle(event)

Signed-off-by: Donald Hunter <donald.hunter@gmail.com>
---
 tools/net/ynl/cli.py     | 14 ++++++--------
 tools/net/ynl/lib/ynl.py | 22 +++++++++++++++++++---
 2 files changed, 25 insertions(+), 11 deletions(-)

diff --git a/tools/net/ynl/cli.py b/tools/net/ynl/cli.py
index b8481f401376..9853fd261ee4 100755
--- a/tools/net/ynl/cli.py
+++ b/tools/net/ynl/cli.py
@@ -4,7 +4,6 @@
 import argparse
 import json
 import pprint
-import time
 
 from lib import YnlFamily, Netlink, NlError
 
@@ -17,7 +16,6 @@ class YnlEncoder(json.JSONEncoder):
             return list(obj)
         return json.JSONEncoder.default(self, obj)
 
-
 def main():
     description = """
     YNL CLI utility - a general purpose netlink utility that uses YAML
@@ -43,7 +41,10 @@ def main():
     group.add_argument('--list-ops', action='store_true')
     group.add_argument('--list-msgs', action='store_true')
 
-    parser.add_argument('--sleep', dest='sleep', type=int)
+    parser.add_argument('--duration', dest='duration', type=int,
+                        help='when subscribed, watch for DURATION seconds')
+    parser.add_argument('--sleep', dest='duration', type=int,
+                        help='alias for duration')
     parser.add_argument('--subscribe', dest='ntf', type=str)
     parser.add_argument('--replace', dest='flags', action='append_const',
                         const=Netlink.NLM_F_REPLACE)
@@ -80,9 +81,6 @@ def main():
     if args.ntf:
         ynl.ntf_subscribe(args.ntf)
 
-    if args.sleep:
-        time.sleep(args.sleep)
-
     if args.list_ops:
         for op_name, op in ynl.ops.items():
             print(op_name, " [", ", ".join(op.modes), "]")
@@ -106,8 +104,8 @@ def main():
         exit(1)
 
     if args.ntf:
-        ynl.check_ntf()
-        output(ynl.async_msg_queue)
+        for msg in ynl.poll_ntf(duration=args.duration):
+            output(msg)
 
 
 if __name__ == "__main__":
diff --git a/tools/net/ynl/lib/ynl.py b/tools/net/ynl/lib/ynl.py
index c22c22bf2cb7..3eca432f5d7b 100644
--- a/tools/net/ynl/lib/ynl.py
+++ b/tools/net/ynl/lib/ynl.py
@@ -12,6 +12,8 @@ import sys
 import yaml
 import ipaddress
 import uuid
+import queue
+import time
 
 from .nlspec import SpecFamily
 
@@ -489,7 +491,7 @@ class YnlFamily(SpecFamily):
         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,7 +905,7 @@ class YnlFamily(SpecFamily):
 
         msg['name'] = op['name']
         msg['msg'] = attrs
-        self.async_msg_queue.append(msg)
+        self.async_msg_queue.put(msg)
 
     def check_ntf(self):
         while True:
@@ -925,11 +927,25 @@ class YnlFamily(SpecFamily):
 
                 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)
+                    print("Unexpected msg id while checking for ntf", decoded)
                     continue
 
                 self.handle_ntf(decoded)
 
+    def poll_ntf(self, interval=0.1, duration=None):
+        endtime = time.time() + duration if duration else None
+        while True:
+            try:
+                self.check_ntf()
+                yield self.async_msg_queue.get_nowait()
+            except queue.Empty:
+                try:
+                    time.sleep(interval)
+                except KeyboardInterrupt:
+                    return
+            if endtime and endtime < time.time():
+                return
+
     def operation_do_attributes(self, name):
       """
       For a given operation name, find and return a supported
-- 
2.47.0

Re: [PATCH net-next v1 2/2] tools/net/ynl: add async notification handling

[Jakub Kicinski]

On Fri,  8 Nov 2024 12:38:16 +0000 Donald Hunter wrote:
> +    def poll_ntf(self, interval=0.1, duration=None):
> +        endtime = time.time() + duration if duration else None

could we default duration to 0 and always check endtime?
I think we can assume that time doesn't go back for simplicity

> +        while True:
> +            try:
> +                self.check_ntf()
> +                yield self.async_msg_queue.get_nowait()
> +            except queue.Empty:
> +                try:
> +                    time.sleep(interval)

Maybe select or epoll would be better that periodic checks?

> +                except KeyboardInterrupt:
> +                    return
> +            if endtime and endtime < time.time():
> +                return

Re: [PATCH net-next v1 2/2] tools/net/ynl: add async notification handling

[Donald Hunter]

Jakub Kicinski <kuba@kernel.org> writes:

> On Fri,  8 Nov 2024 12:38:16 +0000 Donald Hunter wrote:
>> +    def poll_ntf(self, interval=0.1, duration=None):
>> +        endtime = time.time() + duration if duration else None
>
> could we default duration to 0 and always check endtime?
> I think we can assume that time doesn't go back for simplicity

I don't follow; what are you suggesting I initialise endtime to when
duration is 0 ?

>> +        while True:
>> +            try:
>> +                self.check_ntf()
>> +                yield self.async_msg_queue.get_nowait()
>> +            except queue.Empty:
>> +                try:
>> +                    time.sleep(interval)
>
> Maybe select or epoll would be better that periodic checks?

This was the limit of my python knowledge TBH. I can try using python
selectors but I suspect periodic checks will still be needed to reliably
check the endtime.

>> +                except KeyboardInterrupt:
>> +                    return
>> +            if endtime and endtime < time.time():
>> +                return

Re: [PATCH net-next v1 2/2] tools/net/ynl: add async notification handling

[Jakub Kicinski]

On Mon, 11 Nov 2024 11:06:18 +0000 Donald Hunter wrote:
> > On Fri,  8 Nov 2024 12:38:16 +0000 Donald Hunter wrote:  
> >> +    def poll_ntf(self, interval=0.1, duration=None):
> >> +        endtime = time.time() + duration if duration else None  
> >
> > could we default duration to 0 and always check endtime?
> > I think we can assume that time doesn't go back for simplicity  
> 
> I don't follow; what are you suggesting I initialise endtime to when
> duration is 0 ?

I was suggesting:

	def poll_nft([...], duration=0)

	endtime = time.time() + duration

> >> +        while True:
> >> +            try:
> >> +                self.check_ntf()
> >> +                yield self.async_msg_queue.get_nowait()
> >> +            except queue.Empty:
> >> +                try:
> >> +                    time.sleep(interval)  
> >
> > Maybe select or epoll would be better that periodic checks?  
> 
> This was the limit of my python knowledge TBH. I can try using python
> selectors but I suspect periodic checks will still be needed to reliably
> check the endtime.

I thought select is pretty trivial to use in python, basically:

	sock, _, _ = select.select([sock], [], [], timeout=to)
	if sock:
		handle_sock()
	to = endtime - time.time()
	if to <= 0:
		return

Re: [PATCH net-next v1 2/2] tools/net/ynl: add async notification handling

[Donald Hunter]

Jakub Kicinski <kuba@kernel.org> writes:

> On Mon, 11 Nov 2024 11:06:18 +0000 Donald Hunter wrote:
>> > On Fri,  8 Nov 2024 12:38:16 +0000 Donald Hunter wrote:  
>> >> +    def poll_ntf(self, interval=0.1, duration=None):
>> >> +        endtime = time.time() + duration if duration else None  
>> >
>> > could we default duration to 0 and always check endtime?
>> > I think we can assume that time doesn't go back for simplicity  
>> 
>> I don't follow; what are you suggesting I initialise endtime to when
>> duration is 0 ?
>
> I was suggesting:
>
> 	def poll_nft([...], duration=0)
>
> 	endtime = time.time() + duration

I want it to run forever if a duration is not provided, but here
endtime == starttime so it would exit immediately.

I thought the original approach was fairly pythonic - if duration is not
specified (None) then there would be no endtime (None).

>> >> +        while True:
>> >> +            try:
>> >> +                self.check_ntf()
>> >> +                yield self.async_msg_queue.get_nowait()
>> >> +            except queue.Empty:
>> >> +                try:
>> >> +                    time.sleep(interval)  
>> >
>> > Maybe select or epoll would be better that periodic checks?  
>> 
>> This was the limit of my python knowledge TBH. I can try using python
>> selectors but I suspect periodic checks will still be needed to reliably
>> check the endtime.
>
> I thought select is pretty trivial to use in python, basically:
>
> 	sock, _, _ = select.select([sock], [], [], timeout=to)
> 	if sock:
> 		handle_sock()
> 	to = endtime - time.time()
> 	if to <= 0:
> 		return

Yep, thanks. I sketched out roughly this, but using the selectors module
which will use epoll under the covers.

https://docs.python.org/3/library/selectors.html

Re: [PATCH net-next v1 2/2] tools/net/ynl: add async notification handling

[Jakub Kicinski]

On Tue, 12 Nov 2024 09:16:02 +0000 Donald Hunter wrote:
> >> I don't follow; what are you suggesting I initialise endtime to when
> >> duration is 0 ?  
> >
> > I was suggesting:
> >
> > 	def poll_nft([...], duration=0)
> >
> > 	endtime = time.time() + duration  
> 
> I want it to run forever if a duration is not provided, but here
> endtime == starttime so it would exit immediately.
> 
> I thought the original approach was fairly pythonic - if duration is not
> specified (None) then there would be no endtime (None).

Ah, makes perfect sense in hindsight, I misread the code.