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.
This commit is contained in:
Abhishek Verma 2026-09-15 14:20:34 +02:00 committed by mkoop-ni
parent 5745c8a640
commit b1e77b24e4
2 changed files with 115 additions and 10 deletions

View File

@ -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

View File

@ -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()