Commit 1bf70e6c authored by Donald Hunter's avatar Donald Hunter Committed by Paolo Abeni
Browse files

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: default avatarDonald Hunter <donald.hunter@gmail.com>
Tested-by: default avatarKory Maincent <kory.maincent@bootlin.com>
Link: https://patch.msgid.link/20241018093228.25477-1-donald.hunter@gmail.com


Signed-off-by: default avatarPaolo Abeni <pabeni@redhat.com>
parent d05596f2

tools/net/ynl/cli.py

+7 −3
Original line number Diff line number Diff line
@@ -5,6 +5,7 @@ import argparse 
import json

import pprint

import time

import signal



from lib import YnlFamily, Netlink, NlError
@@ -17,6 +18,8 @@ class YnlEncoder(json.JSONEncoder): 
            return list(obj)

        return json.JSONEncoder.default(self, obj)



def handle_timeout(sig, frame):

    exit(0)



def main():

    description = """
@@ -81,7 +84,8 @@ def main(): 
        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 main(): 
        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

+29 −20
Original line number Diff line number Diff line
@@ -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,15 +905,12 @@ 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):

    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:
@@ -925,10 +924,20 @@ 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)

            except BlockingIOError:

                pass



            try:

                yield self.async_msg_queue.get_nowait()

            except queue.Empty:

                try:

                    time.sleep(interval)

                except KeyboardInterrupt:

                    return



    def operation_do_attributes(self, name):

      """