diff --git a/src/adn_server/application/ports.py b/src/adn_server/application/ports.py index f4e8d7c..8c55dca 100644 --- a/src/adn_server/application/ports.py +++ b/src/adn_server/application/ports.py @@ -298,6 +298,11 @@ class SubscriptionStore(ABC): def legs_in_table(self, table_key: str) -> tuple["Subscription", ...]: ... + @abstractmethod + def has_table(self, table_key: str) -> bool: + """True when at least one leg belongs to ``table_key``; runs per datagram.""" + ... + @abstractmethod def has_active_target_leg(self, system: str, slot: int, tgid: int) -> bool: ... diff --git a/src/adn_server/application/routing/lc_ta.py b/src/adn_server/application/routing/lc_ta.py index fd2a9e3..6c3ca27 100644 --- a/src/adn_server/application/routing/lc_ta.py +++ b/src/adn_server/application/routing/lc_ta.py @@ -49,6 +49,7 @@ from typing import Any from bitarray import bitarray from ...domain import int_id +from ...domain.dmr import bptc from ...domain.dmr.const import LC_OPT_G, LC_OPT_U from ...domain.talker_alias import DMRA_BLOCK_COUNT from ..talker_alias_use_cases import passthrough_complete, talker_alias_settings @@ -57,9 +58,41 @@ from .helpers import EMB_LC_SLICE logger = logging.getLogger(__name__) +# Distinct destination LCs kept encoded; a call start needs one per (TG, source radio). +_LC_SET_CACHE_MAX = 512 + + class LcTaMixin: """Talker Alias DMRA relay and embedded LC overlay on forward legs.""" + _lc_set_cache: dict[bytes, tuple[bitarray, bitarray, dict[int, Any]]] + + def _encode_lc_set(self, dst_lc: bytes) -> tuple[bitarray, bitarray, dict[int, Any]]: + """Header LC, terminator LC and embedded LC for one destination LC. + + About 250us of BPTC/RS work, done once per LC instead of once per target + leg: every leg opened for the same TG and radio gets the same encoding. + The results are shared between legs, and every reader only slices them. + """ + cache = self._lc_set_cache + if not isinstance(dst_lc, bytes): # unhashable (bytearray): encode, don't keep + return ( + bptc.encode_header_lc(dst_lc), + bptc.encode_terminator_lc(dst_lc), + self._encode_emblc(dst_lc), + ) + codes = cache.get(dst_lc) + if codes is None: + codes = ( + bptc.encode_header_lc(dst_lc), + bptc.encode_terminator_lc(dst_lc), + self._encode_emblc(dst_lc), + ) + if len(cache) >= _LC_SET_CACHE_MAX: + del cache[next(iter(cache))] # oldest first, not the whole cache + cache[dst_lc] = codes + return codes + def _get_stream_dmra_blocks(self, source_system: str, stream_id: bytes) -> dict[int, bytes] | None: if not self._get_dmra_blocks: return None diff --git a/src/adn_server/application/routing/obp_forward.py b/src/adn_server/application/routing/obp_forward.py index e959594..4277136 100644 --- a/src/adn_server/application/routing/obp_forward.py +++ b/src/adn_server/application/routing/obp_forward.py @@ -52,6 +52,8 @@ from typing import Any from ...domain import HBPF_DATA_SYNC, HBPF_SLT_VHEAD, int_id from ...domain.dmr import decode from ...domain.dmr.const import LC_OPT +from ..subscription.obp_source_ops import ensure_obp_source_for_tg_store, obp_source_needs_ensure +from ..subscription.subscription_queries import store_has_table from .helpers import group_voice_tg_ingress_collision, obp_is_canonical_ingress, unit_data_reportable logger = logging.getLogger(__name__) @@ -85,11 +87,6 @@ class ObpForwardMixin: return if not (79 <= dst_int < 9990 or dst_int > 9999): return - from ..subscription.obp_source_ops import ( - ensure_obp_source_for_tg_store, - obp_source_needs_ensure, - ) - store = self._subscription_store if not obp_source_needs_ensure(store, system_name, relay_table_key, dst_int): return @@ -456,8 +453,6 @@ class ObpForwardMixin: if self._config.get("GLOBAL", {}).get("GEN_STAT_BRIDGES"): _di = int_id(dst_id) _bk = str(_di) - from ..subscription.subscription_queries import store_has_table - if _di >= 5 and _di != 9 and not store_has_table(self._subscription_store, _bk): logger.debug("(%s) Bridge for STAT TG %s does not exist. Creating", system_name, _di) self.ensure_stat_relay(dst_id) diff --git a/src/adn_server/application/routing/voice_subscription.py b/src/adn_server/application/routing/voice_subscription.py index e762e72..3778b69 100644 --- a/src/adn_server/application/routing/voice_subscription.py +++ b/src/adn_server/application/routing/voice_subscription.py @@ -44,7 +44,9 @@ from __future__ import annotations import logging +from typing import Any, NamedTuple +from ...domain.value_objects import TgId, bytes_3 from ...domain.voice_routing import ForwardLeg, VoiceIngress from ..ports import SubscriptionStore from ..subscription.ingress import build_voice_ingress @@ -52,12 +54,53 @@ from ..subscription.router import SubscriptionRouter logger = logging.getLogger(__name__) +# Same set as ``build_voice_ingress``: other call types resolve to no legs. +_BRIDGE_CALL_TYPES = frozenset({"group", "vcsbk"}) +_FORWARD_PLAN_CACHE_MAX = 4096 + + +class ForwardPlan(NamedTuple): + """Where one group voice frame goes, ready for the forwarding loop.""" + + tables: tuple[str, ...] + legs: tuple[ForwardLeg, ...] + # (relay table key, target entry) per leg; shared between frames, read only. + entries: tuple[tuple[str, dict[str, Any]], ...] + + +class _CachedPlan(NamedTuple): + revision: int + # Every SYSTEMS block the plan read, as the object it read: a reload + # replaces the block, which is what makes the plan stale. Identity only + # catches reloads, though: blocks are also changed in place at runtime (the + # static TGs, for one). That is safe because the plan reads nothing from a + # block but MODE, which only a reload changes. A plan that reads any other + # field must also be keyed on it, or it will be silently stale. + blocks: tuple[tuple[str, Any], ...] + plan: ForwardPlan + + +def _route_ingress(system_name: str, slot: int, dst_int: int, call_type: str, mode: str) -> VoiceIngress: + """An ingress carrying only what ``resolve()`` routes on: no peer, radio or stream. + + Leaving those out is what lets one plan serve every frame of a call; a router + that starts reading them has to be added to the plan cache key first. + """ + return VoiceIngress( + source_system=system_name, + slot=1 if int(slot) == 1 else 2, # type: ignore[arg-type] + dst_tgid=TgId(dst_int), + source_is_obp=mode == "OPENBRIDGE", + call_type=call_type, + ) + class VoiceSubscriptionMixin: """Wire ``SubscriptionRouter`` into ``dmrd_received``.""" _subscription_store: SubscriptionStore _subscription_router: SubscriptionRouter | None + _forward_plan_cache: dict[tuple, _CachedPlan] def _subscription_router_instance(self) -> SubscriptionRouter: router = getattr(self, "_subscription_router", None) @@ -90,6 +133,78 @@ class VoiceSubscriptionMixin: stream_id=stream_id, ) + def _group_voice_forward_plan( + self, + *, + system_name: str, + slot: int, + call_type: str, + source_is_obp: bool, + bridge_match_slot: int, + dst_int: int, + ) -> ForwardPlan: + """The forward plan of a group voice frame, rebuilt only when it can differ. + + It depends on the subscription store, the source and target SYSTEMS blocks + and on nothing in the frame beyond the key, so each frame of a call reuses + the plan of the first until the store is written or a block is reloaded. + Per-frame checks (ENABLED, quench, keepalive, contention) stay in the loop. + + Invariant: from the SYSTEMS blocks it reads only MODE (see ``_CachedPlan``). + """ + systems = self._config.get("SYSTEMS", {}) + revision = getattr(self._subscription_store, "revision", None) + src_block = systems.get(system_name) + mode = "OPENBRIDGE" if source_is_obp else (src_block or {}).get("MODE", "") + routable = call_type in _BRIDGE_CALL_TYPES + # VoiceIngress.bridge_match_slot, which is what resolve() matches on. + match_slot = 1 if mode == "OPENBRIDGE" else (1 if int(slot) == 1 else 2) + key = (system_name, bridge_match_slot, match_slot, dst_int, mode == "OPENBRIDGE", routable) + cache = self._forward_plan_cache + hit = cache.get(key) if revision is not None else None + if ( + hit is not None + and hit.revision == revision + and all(systems.get(name) is block for name, block in hit.blocks) + ): + return hit.plan + + router = self._subscription_router_instance() + tables = router.relay_tables_with_active_source(system_name, bridge_match_slot, dst_int) + legs = router.resolve(_route_ingress(system_name, slot, dst_int, call_type, mode)) if routable else () + # One leg per (target, translated TGID) on MASTER/PEER targets: their + # send_peers() picks each peer's slot itself, so a second leg is the same + # audio twice. OpenBridge targets keep per-slot legs (separate links). + seen_hbp: set[tuple[str, int]] = set() + deduped = [] + for leg in legs: + if systems.get(leg.target_system, {}).get("MODE") != "OPENBRIDGE": + hbp_key = (leg.target_system, int(leg.target_tgid)) + if hbp_key in seen_hbp: + continue + seen_hbp.add(hbp_key) + deduped.append(leg) + table_key = tables[0] if tables else str(dst_int) + entries = tuple( + ( + table_key, + { + "SYSTEM": leg.target_system, + "TS": int(leg.slot), + "TGID": bytes_3(int(leg.target_tgid)), + "ACTIVE": True, + }, + ) + for leg in deduped + ) + plan = ForwardPlan(tables, tuple(deduped), entries) + if revision is not None: + names = {system_name, *(leg.target_system for leg in legs)} + if len(cache) >= _FORWARD_PLAN_CACHE_MAX: + del cache[next(iter(cache))] # oldest first, not the whole cache + cache[key] = _CachedPlan(revision, tuple((n, systems.get(n)) for n in names), plan) + return plan + def _voice_relay_tables_with_active_source( self, system_name: str, diff --git a/src/adn_server/application/routing_use_cases.py b/src/adn_server/application/routing_use_cases.py index 79e4482..34935e8 100644 --- a/src/adn_server/application/routing_use_cases.py +++ b/src/adn_server/application/routing_use_cases.py @@ -64,7 +64,6 @@ from .routing.helpers import ( obp_publish_flat_bridge_tx, obp_status_plugin_voice, obp_sync_flat_bridge_tx_times, - obp_target_bcsq_quenches_stream, resolve_voice_peer_id, slot_has_active_voice, unit_data_hbp_target_idle, @@ -78,6 +77,7 @@ from .routing.subscription_table import SubscriptionTableMixin from .routing.timers import RoutingTimerMixin from .routing.voice_subscription import VoiceSubscriptionMixin from .server_voice import all_server_voice_ids +from .subscription.subscription_queries import store_has_table from .talker_alias_use_cases import TalkerAliasUseCases logger = logging.getLogger(__name__) @@ -124,6 +124,8 @@ class RoutingUseCases( self._config = config self._subscription_store = subscription_store self._subscription_router = None + self._forward_plan_cache = {} # see VoiceSubscriptionMixin._group_voice_forward_plan + self._lc_set_cache = {} # see LcTaMixin._encode_lc_set self._routing_table_legacy_view = None self._send_to_system = send_to_system # (system_name, packet, **kwargs) -> None self._get_protocols = get_protocols # () -> dict[str, protocol] @@ -259,8 +261,6 @@ class RoutingUseCases( # Arm ON in-band rules on VHEAD (echo 9990 and UA bridges); VTERM handled in udp_hbp too. if frame_type == HBPF_DATA_SYNC and dtype_vseq == HBPF_SLT_VHEAD: self.apply_in_band_signalling(system_name, slot, dst_id, pkt_time) - from .subscription.subscription_queries import store_has_table - relay_table_key = str(int_id(dst_id)) dst_int = int_id(dst_id) # Legacy bridge_master to_target: OpenBridge clears TS bit — "all OpenBridge streams are @@ -494,40 +494,16 @@ class RoutingUseCases( source_lc = b"\x00\x00\x20" + dst_id_b + rf_src # Legacy bridge_master routerOBP: _sysIgnore accumulates across each to_target(BRIDGES[_bridge]) # pass; dedupe (SYSTEM, TS) for OpenBridge targets so the same leg is not sent twice per packet. - # SubscriptionRouter.resolve() already applies OBP dedup on OpenBridge targets. - forward_tables, forward_legs = self._voice_forward_plan( + # SubscriptionRouter.resolve() already applies OBP dedup on OpenBridge targets, and the plan + # collapses same-(target, translated TGID) MASTER/PEER legs (see _group_voice_forward_plan). + _plan = self._group_voice_forward_plan( system_name=system_name, - peer_id=peer_id, - rf_src=rf_src, - dst_id=dst_id, slot=slot, call_type=call_type, - stream_id=stream_id, source_is_obp=source_is_obp, bridge_match_slot=bridge_match_slot, dst_int=dst_int, ) - # A BRIDGES scan (legacy parity, kept in SubscriptionRouter.resolve()) can list the - # same MASTER/PEER target on both TS1 and TS2 (e.g. an inject-only proxy whose - # connected hotspots collectively use both slots for one TG). Legacy's dumb - # send_peers() broadcast made that harmless — each hotspot's own radio dropped the - # slot it didn't want. This server's send_peers() instead resolves each peer's - # actual listen slot from its OPTIONS (iter_downlink_voice_slots) regardless of the - # wire slot, so a second identical leg to the same target delivers the same audio - # twice — doubling the downlink rate and making unpaced bridges (e.g. ysf2dmr) sound - # slow. OpenBridge targets keep distinct per-slot legs (real separate links); collapse - # only same-(target, translated TGID) duplicates for MASTER/PEER targets. - if forward_legs: - _seen_hbp_leg: set[tuple[str, int]] = set() - _deduped_legs = [] - for _leg in forward_legs: - if systems_cfg.get(_leg.target_system, {}).get("MODE") != "OPENBRIDGE": - _hbp_key = (_leg.target_system, int(_leg.target_tgid)) - if _hbp_key in _seen_hbp_leg: - continue - _seen_hbp_leg.add(_hbp_key) - _deduped_legs.append(_leg) - forward_legs = tuple(_deduped_legs) _tx_report_peer = int_id(peer_id) if not source_is_obp and not synthetic_announcement: _tx_report_peer = int_id( @@ -543,18 +519,7 @@ class RoutingUseCases( _src_slot_st = src_proto.STATUS.get(slot, {}) if isinstance(_src_slot_st, dict) and _src_slot_st.get("_suppress_uplink"): _suppress_uplink = True - _leg_iter: list[tuple[str, dict[str, Any]]] = [ - ( - forward_tables[0] if forward_tables else str(dst_int), - { - "SYSTEM": leg.target_system, - "TS": int(leg.slot), - "TGID": bytes_3(int(leg.target_tgid)), - "ACTIVE": True, - }, - ) - for leg in forward_legs - ] + _leg_iter = _plan.entries for _relay_table_key, entry in _leg_iter: if _suppress_uplink: @@ -575,10 +540,10 @@ class RoutingUseCases( if isinstance(target_tgid, int): target_tgid = bytes_3(target_tgid) # If target has quenched us, don't send (~1856-1859). - if obp_target_bcsq_quenches_stream(self._config, entry["SYSTEM"], dst_id_b, stream_id): + _target_session = self._obp_session(entry["SYSTEM"]) + if _target_session.quenches(dst_id_b, stream_id): continue # If target has missed keepalives (ENHANCED_OBP), don't send (~1861-1863) - _target_session = self._obp_session(entry["SYSTEM"]) if _target_system.get("ENHANCED_OBP") and not _target_session.keepalive_ok(pkt_time): continue # Talkgroup ACL (global + per-system TG1) (~1865-1873) @@ -622,9 +587,11 @@ class RoutingUseCases( logger.exception("(to_target) caught exception") _target_status[stream_id]["LAST"] = pkt_time return - _target_status[stream_id]["H_LC"] = bptc.encode_header_lc(dst_lc) - _target_status[stream_id]["T_LC"] = bptc.encode_terminator_lc(dst_lc) - _target_status[stream_id]["EMB_LC"] = self._encode_emblc(dst_lc) + ( + _target_status[stream_id]["H_LC"], + _target_status[stream_id]["T_LC"], + _target_status[stream_id]["EMB_LC"], + ) = self._encode_lc_set(dst_lc) self._init_talker_alias_embed( _target_status[stream_id], system_name, @@ -835,9 +802,11 @@ class RoutingUseCases( _bridge_tx_leg["TX_STREAM_ID"] = stream_id _bridge_tx_leg["TX_RFS"] = rf_src _bridge_tx_leg["TX_PEER"] = peer_id - _bridge_tx_leg["TX_H_LC"] = bptc.encode_header_lc(dst_lc) - _bridge_tx_leg["TX_T_LC"] = bptc.encode_terminator_lc(dst_lc) - _bridge_tx_leg["TX_EMB_LC"] = self._encode_emblc(dst_lc) + ( + _bridge_tx_leg["TX_H_LC"], + _bridge_tx_leg["TX_T_LC"], + _bridge_tx_leg["TX_EMB_LC"], + ) = self._encode_lc_set(dst_lc) if obp_flat_bridge_tx_idle(_ts_st, pkt_time) or _ts_st.get("TX_STREAM_ID") == stream_id: obp_publish_flat_bridge_tx(_ts_st, _bridge_tx_leg) self._dispatch_talker_alias_on_bridge_open( @@ -856,9 +825,7 @@ class RoutingUseCases( _ts_st["TX_STREAM_ID"] = stream_id _ts_st["TX_RFS"] = rf_src _ts_st["TX_PEER"] = peer_id - _ts_st["TX_H_LC"] = bptc.encode_header_lc(dst_lc) - _ts_st["TX_T_LC"] = bptc.encode_terminator_lc(dst_lc) - _ts_st["TX_EMB_LC"] = self._encode_emblc(dst_lc) + _ts_st["TX_H_LC"], _ts_st["TX_T_LC"], _ts_st["TX_EMB_LC"] = self._encode_lc_set(dst_lc) self._dispatch_talker_alias_on_bridge_open( _ts_st, system_name, diff --git a/src/adn_server/application/subscription/in_band_signalling_ops.py b/src/adn_server/application/subscription/in_band_signalling_ops.py index 87b9c2f..ba8677b 100644 --- a/src/adn_server/application/subscription/in_band_signalling_ops.py +++ b/src/adn_server/application/subscription/in_band_signalling_ops.py @@ -30,7 +30,7 @@ from adn_server.application.routing.helpers import is_special_tg from adn_server.application.subscription.routing_table_export import _legacy_to_type from adn_server.application.subscription.trigger_bytes import dst_in_triggers from adn_server.domain import bytes_3, int_id -from adn_server.domain.subscription import ActivationPolicy, Subscription, SubscriptionPhase +from adn_server.domain.subscription import ActivationPolicy, Subscription, SubscriptionPhase, SystemId logger = logging.getLogger(__name__) @@ -52,10 +52,7 @@ def apply_in_band_signalling_store( dst_group = int_id(dst_id) dst_id_b = dst_id if isinstance(dst_id, bytes) and len(dst_id) >= 3 else bytes_3(dst_group) - for sub in store.snapshot(): - if sub.system.value != system_name: - continue - + for sub in store.list_by_system(SystemId(system_name)): relay_table_key = sub.table_key() if relay_table_key[:1] == "#" and dst_group != 9: continue diff --git a/src/adn_server/application/subscription/obp_source_ops.py b/src/adn_server/application/subscription/obp_source_ops.py index e029c2b..2371f33 100644 --- a/src/adn_server/application/subscription/obp_source_ops.py +++ b/src/adn_server/application/subscription/obp_source_ops.py @@ -58,7 +58,7 @@ def obp_source_needs_ensure( active_tables = set(store.relay_tables_with_active_source(system_name, 1, dst_int)) pending_keys: list[str] = [] for key in (relay_table_key, "#" + relay_table_key): - if store.legs_in_table(key): + if store.has_table(key): pending_keys.append(key) if not pending_keys: return False @@ -75,7 +75,7 @@ def ensure_obp_source_for_tg_store( ) -> None: """Ensure OBP has ACTIVE TS1 source row in main and #reflector tables.""" for key in (relay_table_key, "#" + relay_table_key): - if not any(sub.table_key() == key for sub in store.snapshot()): + if not store.has_table(key): continue channel_tgid = dst_int patched = False diff --git a/src/adn_server/application/subscription/subscription_queries.py b/src/adn_server/application/subscription/subscription_queries.py index 3d7c0b1..59e2e07 100644 --- a/src/adn_server/application/subscription/subscription_queries.py +++ b/src/adn_server/application/subscription/subscription_queries.py @@ -31,7 +31,7 @@ def store_has_table(store: SubscriptionStore, table_key: str) -> bool: Indexed: this runs per datagram, and the scan it replaces was building a tuple of every subscription to answer a yes/no question. """ - return bool(store.legs_in_table(table_key)) + return store.has_table(table_key) def system_has_active_leg_in_store( diff --git a/src/adn_server/application/subscription/subscription_table_ops.py b/src/adn_server/application/subscription/subscription_table_ops.py index 1446dfa..6f5cd05 100644 --- a/src/adn_server/application/subscription/subscription_table_ops.py +++ b/src/adn_server/application/subscription/subscription_table_ops.py @@ -54,13 +54,12 @@ def _effective_tmout_minutes(tgid_int: int, tmout: float) -> float: def _table_has_legs(store: SubscriptionStore, table_key: str) -> bool: - return any(sub.table_key() == table_key for sub in store.snapshot()) + return store.has_table(table_key) def _remove_table(store: SubscriptionStore, table_key: str) -> None: - for sub in list(store.snapshot()): - if sub.table_key() == table_key: - store.remove(sub.subscription_id) + for sub in store.legs_in_table(table_key): + store.remove(sub.subscription_id) def _find_leg( diff --git a/src/adn_server/infrastructure/subscription_store.py b/src/adn_server/infrastructure/subscription_store.py index 116696c..23984b1 100644 --- a/src/adn_server/infrastructure/subscription_store.py +++ b/src/adn_server/infrastructure/subscription_store.py @@ -42,12 +42,20 @@ class InMemorySubscriptionStore(SubscriptionStore): def __init__(self) -> None: self._items: dict[SubscriptionId, Subscription] = {} + # Per system, in the same order as _items (an upsert keeps its place). + self._by_system: dict[str, dict[SubscriptionId, Subscription]] = {} self._by_table: dict[str, list[Subscription]] = defaultdict(list) self._source_tables: dict[_IndexKey, set[str]] = {} self._active_target_counts: dict[_IndexKey, int] = {} # What each leg was indexed under. Callers change a leg in place and then # upsert it, so by the time it is unindexed it may no longer say where it is. self._indexed: dict[SubscriptionId, tuple[str, _IndexKey | None]] = {} + self._revision = 0 + + @property + def revision(self) -> int: + """Bumped on every write, so readers can keep what they derived until it moves.""" + return self._revision def get(self, sub_id: SubscriptionId) -> Subscription | None: return self._items.get(sub_id) @@ -57,27 +65,39 @@ class InMemorySubscriptionStore(SubscriptionStore): if old is not None: self._unindex(old) self._items[subscription.subscription_id] = subscription + self._by_system.setdefault(subscription.system.value, {})[subscription.subscription_id] = subscription self._index(subscription) + self._revision += 1 def remove(self, sub_id: SubscriptionId) -> bool: old = self._items.pop(sub_id, None) if old is None: return False + of_system = self._by_system.get(old.system.value) + if of_system is not None: + of_system.pop(sub_id, None) + if not of_system: + del self._by_system[old.system.value] self._unindex(old) + self._revision += 1 return True def clear(self) -> None: self._items.clear() + self._by_system.clear() self._by_table.clear() self._source_tables.clear() self._active_target_counts.clear() self._indexed.clear() + self._revision += 1 def replace_all(self, subscriptions: Sequence[Subscription]) -> None: self.clear() for sub in subscriptions: self._items[sub.subscription_id] = sub + self._by_system.setdefault(sub.system.value, {})[sub.subscription_id] = sub self._index(sub) + self._revision += 1 def snapshot(self) -> tuple[Subscription, ...]: return tuple(self._items.values()) @@ -86,7 +106,7 @@ class InMemorySubscriptionStore(SubscriptionStore): return tuple(sub for sub in self._items.values() if sub.channel == channel) def list_by_system(self, system: SystemId) -> tuple[Subscription, ...]: - return tuple(sub for sub in self._items.values() if sub.system == system) + return tuple(self._by_system.get(system.value, {}).values()) def list_active(self) -> tuple[Subscription, ...]: return tuple(sub for sub in self._items.values() if sub.is_active()) @@ -106,6 +126,10 @@ class InMemorySubscriptionStore(SubscriptionStore): return () return tuple(sorted(keys)) + def has_table(self, table_key: str) -> bool: + """O(1): the table index instead of a scan of every subscription.""" + return bool(self._by_table.get(table_key)) + def legs_in_table(self, table_key: str) -> tuple[Subscription, ...]: """All legs for a relay table key (indexed).""" return tuple(self._by_table.get(table_key, ())) diff --git a/tests/routing/test_forward_plan_cache.py b/tests/routing/test_forward_plan_cache.py new file mode 100644 index 0000000..cd40ddc --- /dev/null +++ b/tests/routing/test_forward_plan_cache.py @@ -0,0 +1,221 @@ +# ADN DMR Peer Server - tests routing forward plan cache +# +# Copyright (C) 2026 Rodrigo Pérez, CE5RPY +# +############################################################################### +# This program is free software; you can redistribute it and/or modify +# it under the terms of the GNU General Public License as published by +# the Free Software Foundation; either version 3 of the License, or +# (at your option) any later version. +# +# This program is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU General Public License for more details. +# +# You should have received a copy of the GNU General Public License +# along with this program; if not, write to the Free Software Foundation, +# Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA +############################################################################### + +"""A group voice call reuses its forward plan, and the plan follows the store and the config.""" + +from __future__ import annotations + +import copy + +from tests.harness.deterministic import ( + DeterministicScenario, + PacketSpec, + active_routing_table, + add_openbridge_system, + minimal_config, +) + +from adn_server.application.subscription.routing_table_import import subscriptions_from_routing_table +from adn_server.domain.dmr import bptc +from adn_server.domain.subscription import SubscriptionPhase, SystemId +from adn_server.infrastructure.subscription_store import InMemorySubscriptionStore + +TG = 213 +OBPS = ("OBP-0", "OBP-1") + + +def _scenario() -> DeterministicScenario: + config = minimal_config(("MASTER-A", "MASTER-B")) + for i, name in enumerate(OBPS): + add_openbridge_system(config, name) + config["SYSTEMS"][name]["NETWORK_ID"] = (100 + i).to_bytes(4, "big") + table = active_routing_table( + TG, (("MASTER-A", 2), ("MASTER-B", 2)) + tuple((n, 1) for n in OBPS), timeout_minutes=10**6 + ) + sc = DeterministicScenario(config=config, routing_table=table) + sc.routing.apply_startup_subscriptions() + return sc + + +def _call(sc: DeterministicScenario, stream_id: int, bursts: int = 3) -> None: + base = PacketSpec(dst_id=TG, stream_id=stream_id, slot=2) + sc.inject_hbp("MASTER-A", DeterministicScenario.voice_head_spec(base)) + for seq in range(1, bursts + 1): + sc.clock.advance(0.06) + sc.inject_hbp("MASTER-A", DeterministicScenario.voice_burst_spec(base, seq=seq, dtype_vseq=seq % 6)) + + +def _plan(sc: DeterministicScenario): + return sc.routing._group_voice_forward_plan( + system_name="MASTER-A", slot=2, call_type="group", source_is_obp=False, + bridge_match_slot=2, dst_int=TG, + ) + + +def _targets(sc: DeterministicScenario) -> set[str]: + return {p.target_system for p in sc.capture.packets} + + +def test_frames_of_a_call_share_one_plan() -> None: + sc = _scenario() + first = _plan(sc) + assert _plan(sc) is first + assert {entry["SYSTEM"] for _, entry in first.entries} == {"MASTER-B", *OBPS} + + +def test_plan_is_rebuilt_after_a_store_write() -> None: + sc = _scenario() + _call(sc, 0x1001) + assert _targets(sc) == {"MASTER-B", *OBPS} + + store = sc.subscription_store + before = _plan(sc) + leg = next(s for s in store.legs_in_table(str(TG)) if s.system.value == "OBP-1") + leg.state.phase = SubscriptionPhase.IDLE + store.upsert(leg) + + after = _plan(sc) + assert after is not before + assert "OBP-1" not in {entry["SYSTEM"] for _, entry in after.entries} + sc.capture.packets.clear() + sc.clock.advance(0.06) + base = PacketSpec(dst_id=TG, stream_id=0x1001, slot=2) + sc.inject_hbp("MASTER-A", DeterministicScenario.voice_burst_spec(base, seq=9, dtype_vseq=3)) + assert _targets(sc) == {"MASTER-B", "OBP-0"} + + +def test_plan_is_rebuilt_when_a_reload_replaces_a_system_block() -> None: + sc = _scenario() + before = _plan(sc) + # config_reload assigns a merged copy per system; the old block is gone. + sc.config["SYSTEMS"]["OBP-0"] = copy.deepcopy(sc.config["SYSTEMS"]["OBP-0"]) + assert _plan(sc) is not before + assert _plan(sc) is _plan(sc) + + +def test_cached_plan_matches_a_fresh_one() -> None: + sc = _scenario() + cached = _plan(sc) + sc.routing._forward_plan_cache.clear() + fresh = _plan(sc) + assert fresh is not cached + assert fresh == cached + + +def test_has_table_agrees_with_a_full_scan_through_every_write() -> None: + table = active_routing_table(TG, (("MASTER-A", 2), ("OBP-0", 1))) + table |= active_routing_table(91, (("MASTER-A", 1),)) + subs = subscriptions_from_routing_table(table) + store = InMemorySubscriptionStore() + + def agrees() -> None: + for key in (str(TG), "91", "4000"): + scan = any(s.table_key() == key for s in store.snapshot()) + assert store.has_table(key) is scan, key + + revisions = [store.revision] + store.replace_all(subs) + agrees() + revisions.append(store.revision) + store.remove(subs[0].subscription_id) + agrees() + revisions.append(store.revision) + store.upsert(subs[0]) + agrees() + revisions.append(store.revision) + for sub in store.legs_in_table("91"): + store.remove(sub.subscription_id) + agrees() + revisions.append(store.revision) + store.clear() + agrees() + revisions.append(store.revision) + assert revisions == sorted(set(revisions)), "every write moves the revision" + + +def test_removing_a_missing_subscription_keeps_the_revision() -> None: + store = InMemorySubscriptionStore() + subs = subscriptions_from_routing_table(active_routing_table(TG, (("MASTER-A", 2),))) + before = store.revision + assert store.remove(subs[0].subscription_id) is False + assert store.revision == before + + +def test_list_by_system_keeps_the_order_of_a_full_scan() -> None: + table = active_routing_table(TG, (("MASTER-A", 2), ("OBP-0", 1), ("MASTER-B", 2))) + table |= active_routing_table(91, (("MASTER-A", 1), ("OBP-0", 1))) + table |= active_routing_table(92, (("MASTER-A", 2),)) + subs = subscriptions_from_routing_table(table) + store = InMemorySubscriptionStore() + + def agrees() -> None: + for system in ("MASTER-A", "OBP-0", "MASTER-B", "NONE"): + sid = SystemId(system) + scan = tuple(s for s in store.snapshot() if s.system == sid) + assert store.list_by_system(sid) == scan, system + + store.replace_all(subs) + agrees() + moved = next(s for s in subs if s.system.value == "MASTER-A") + moved.state.phase = SubscriptionPhase.IDLE + store.upsert(moved) # an upsert keeps its place + agrees() + store.remove(moved.subscription_id) + agrees() + store.upsert(moved) # back in, now last + agrees() + for sub in store.list_by_system(SystemId("OBP-0")): + store.remove(sub.subscription_id) + agrees() + store.clear() + agrees() + + +def test_legs_of_one_call_share_one_lc_encoding() -> None: + sc = _scenario() + _call(sc, 0x2002, bursts=1) + rows = [sc.protocols[name].STATUS[(0x2002).to_bytes(4, "big")] for name in OBPS] + assert rows[0]["H_LC"] is rows[1]["H_LC"] + assert rows[0]["EMB_LC"] is rows[1]["EMB_LC"] + lc = b"\x00\x00\x20" + TG.to_bytes(3, "big") + PacketSpec(dst_id=TG).rf_src.to_bytes(3, "big") + assert rows[0]["H_LC"] == bptc.encode_header_lc(lc) + assert rows[0]["T_LC"] == bptc.encode_terminator_lc(lc) + assert rows[0]["EMB_LC"] == bptc.encode_emblc(lc) + + +def test_lc_set_of_a_bytearray_is_encoded_without_caching() -> None: + sc = _scenario() + lc = bytearray(b"\x00\x00\x20" + TG.to_bytes(3, "big") + (2130035).to_bytes(3, "big")) + header, terminator, emb = sc.routing._encode_lc_set(lc) + assert header == bptc.encode_header_lc(bytes(lc)) + assert terminator == bptc.encode_terminator_lc(bytes(lc)) + assert emb == bptc.encode_emblc(bytes(lc)) + assert not sc.routing._lc_set_cache + + +def test_full_lc_set_cache_drops_only_its_oldest_entry(monkeypatch) -> None: + from adn_server.application.routing import lc_ta + + monkeypatch.setattr(lc_ta, "_LC_SET_CACHE_MAX", 3) + sc = _scenario() + lcs = [b"\x00\x00\x20" + TG.to_bytes(3, "big") + (2130000 + i).to_bytes(3, "big") for i in range(4)] + for lc in lcs: + sc.routing._encode_lc_set(lc) + assert list(sc.routing._lc_set_cache) == lcs[1:]