From b1e77b24e486cfebf7d2eb359c96842b0142a43e Mon Sep 17 00:00:00 2001 From: Abhishek Verma Date: Tue, 15 Sep 2026 14:20:34 +0200 Subject: [PATCH] mpm: Add rtnetlink-driven xport_info cache invalidation -Add gevent rtnetlink listener for interface and IP changes. -Refresh cached transport info immediately after network changes. -Extend x4xx cache TTL to 300 seconds with epoch-based invalidation. --- mpm/python/usrp_mpm/periph_manager/x4xx.py | 42 ++++++++--- mpm/python/usrp_mpm/sys_utils/net.py | 83 ++++++++++++++++++++++ 2 files changed, 115 insertions(+), 10 deletions(-) diff --git a/mpm/python/usrp_mpm/periph_manager/x4xx.py b/mpm/python/usrp_mpm/periph_manager/x4xx.py index a8b4a5a68..ae18aaa96 100644 --- a/mpm/python/usrp_mpm/periph_manager/x4xx.py +++ b/mpm/python/usrp_mpm/periph_manager/x4xx.py @@ -33,7 +33,7 @@ from usrp_mpm.periph_manager.x4xx_periphs import ( ) from usrp_mpm.periph_manager.x4xx_rfdc_ctrl import MixerMode, X4xxRfdcCtrl from usrp_mpm.rpc_utils import no_claim, no_rpc -from usrp_mpm.sys_utils import dtoverlay, ectool +from usrp_mpm.sys_utils import dtoverlay, ectool, net from usrp_mpm.sys_utils.gpio import Gpio from usrp_mpm.sys_utils.sysfs_hwmon import HwmonTempSensors from usrp_mpm.sys_utils.udev import dt_symbol_get_spidev @@ -47,10 +47,7 @@ X400_DEVICE_DNA_COMPAT = (8, 3) X400_DEFAULT_ENABLE_PPS_EXPORT = True X400_DEFAULT_TRIG_DIRECTION = ClockingAuxBrdControl.DIRECTION_OUTPUT X400_MONITOR_THREAD_INTERVAL = 1.0 # seconds -# How long to cache the xport info. The result is stable per session -# (init() invalidates it on IP changes), so it can be cached well beyond -# the RPC deadline window. -X400_XPORT_INFO_CACHE_TTL = 10.0 # seconds +X400_XPORT_INFO_CACHE_TTL = 300.0 # seconds; also invalidated by OS net changes QSFPModuleConfig = namedtuple("QSFPModuleConfig", "modprs modsel devsymbol") X400_QSFP_I2C_CONFIGS = [ QSFPModuleConfig(modprs="QSFP0_MODPRS", modsel="QSFP0_MODSEL_n", devsymbol="qsfp0_i2c"), @@ -313,6 +310,7 @@ class x4xx(ZynqComponents, PeriphManagerBase): self.dio_control = None self._xport_info_cache = None self._xport_info_cache_ts = 0.0 + self._xport_info_cache_epoch = None self._xport_info_cache_lock = threading.Lock() try: self._init_peripherals(args) @@ -746,28 +744,52 @@ class x4xx(ZynqComponents, PeriphManagerBase): get_xport_info() enumerates network interfaces via netlink/pyudev, which is slow and doesn't yield to the gevent loop. GetDeviceInfo is unclaimed - and gets flooded during device discovery, so caching this briefly (for + and gets flooded during device discovery, so caching this (for X400_XPORT_INFO_CACHE_TTL seconds) keeps it from starving other RPCs and makes device discovery faster. + The cache is also invalidated whenever the OS reports a network change + (via net.get_iface_change_epoch()), so a longer TTL stays correct: real + interface/IP changes force an immediate refresh regardless of the TTL. + gRPC serves RPCs from a thread pool, so concurrent GetDeviceInfo calls can all see a stale cache at once; the lock ensures only one of them redoes the enumeration while the rest wait for that result instead of each redundantly enumerating in parallel. """ + epoch = net.get_iface_change_epoch() now = monotonic() if ( self._xport_info_cache is None + or epoch != self._xport_info_cache_epoch or now - self._xport_info_cache_ts > X400_XPORT_INFO_CACHE_TTL ): with self._xport_info_cache_lock: + epoch = net.get_iface_change_epoch() now = monotonic() - if ( - self._xport_info_cache is None - or now - self._xport_info_cache_ts > X400_XPORT_INFO_CACHE_TTL - ): + if self._xport_info_cache is None: + reason = "cold cache" + elif epoch != self._xport_info_cache_epoch: + reason = "iface-change epoch {}->{}".format( + self._xport_info_cache_epoch, epoch + ) + elif now - self._xport_info_cache_ts > X400_XPORT_INFO_CACHE_TTL: + reason = "TTL expired ({:.1f}s)".format( + now - self._xport_info_cache_ts + ) + else: + reason = None # another thread already refreshed while we waited + if reason is not None: self._xport_info_cache = self._xport_mgrs["udp"].get_xport_info() self._xport_info_cache_ts = now + self._xport_info_cache_epoch = epoch + self.log.debug( + "Refreshed CHDR xport info cache (%s): %s", + reason, + self._xport_info_cache, + ) + else: + self.log.trace("CHDR xport info cache hit (epoch %s)", epoch) return self._xport_info_cache @no_claim diff --git a/mpm/python/usrp_mpm/sys_utils/net.py b/mpm/python/usrp_mpm/sys_utils/net.py index 429cd9275..4691c9605 100644 --- a/mpm/python/usrp_mpm/sys_utils/net.py +++ b/mpm/python/usrp_mpm/sys_utils/net.py @@ -9,6 +9,7 @@ Network utilities for MPM import itertools import os import socket +import threading import pyudev from pyroute2 import IPRoute @@ -37,6 +38,88 @@ def _get_udev_context(): return _udev_ctx +# Interface-change epoch: increments on rtnetlink address or link events so +# callers can detect stale cached interface information. If the listener cannot +# start, callers fall back to TTL-based caching. +NETLINK_ROUTE = 0 +RTMGRP_LINK = 0x1 +RTMGRP_IPV4_IFADDR = 0x10 + +_iface_epoch = 0 # single-writer (the listener greenlet); readers just read it +_iface_monitor_started = False +_iface_monitor_pid = None +_iface_monitor_lock = threading.Lock() + + +def _netlink_listen_loop(sock): + """Bump the iface-change epoch on every rtnetlink link/address message.""" + global _iface_epoch + log = get_logger("net") + try: + while True: + data = sock.recv(65536) + if not data: + break + _iface_epoch += 1 + log.debug( + "OS net change (rtnetlink); iface-change epoch -> %d", _iface_epoch + ) + except OSError as ex: + log.warning("rtnetlink interface-change listener stopped: %s", ex) + finally: + try: + sock.close() + except OSError: + pass + + +def _ensure_iface_monitor(): + """Start the rtnetlink listener once per process (best-effort). + + Failure is non-fatal: we log and leave the epoch fixed, so callers degrade to + TTL-only cache behavior. + """ + global _iface_monitor_started, _iface_monitor_pid + pid = os.getpid() + if _iface_monitor_started and _iface_monitor_pid == pid: + return + with _iface_monitor_lock: + if _iface_monitor_started and _iface_monitor_pid == pid: + return + try: + import gevent + from gevent import socket as gevent_socket + + sock = gevent_socket.socket( + socket.AF_NETLINK, socket.SOCK_RAW, NETLINK_ROUTE + ) + sock.bind((0, RTMGRP_LINK | RTMGRP_IPV4_IFADDR)) + gevent.spawn(_netlink_listen_loop, sock) + _iface_monitor_started = True + _iface_monitor_pid = pid + get_logger("net").info( + "Started rtnetlink interface-change monitor (RTMGRP_IPV4_IFADDR|" + "RTMGRP_LINK; OS-driven xport cache invalidation active)." + ) + except Exception as ex: # pylint: disable=broad-except + get_logger("net").warning( + "Could not start rtnetlink interface-change monitor: %s; " + "falling back to TTL-only interface caching.", + ex, + ) + + +def get_iface_change_epoch(): + """Return a counter that increments whenever the OS reports a net change. + + Lazily starts the rtnetlink listener on first use. A caller that caches + interface info can store this value alongside its cache and refresh when the + epoch changes (in addition to any TTL). + """ + _ensure_iface_monitor() + return _iface_epoch + + def get_hostname(): """Return the current device's hostname""" return socket.gethostname()