fix(duty-cycle): synchronize per-radio budget state

This commit is contained in:
agessaman committed 2026-09-13 12:13:49 -07:00
1 parent d4aca403a5
commit d6f1a4c73b
6 files changed
+466 -193

No files matched your search

+244 -154
View File
@@ -1,5 +1,7 @@
import logging
import threading
import time
from dataclasses import dataclass
from typing import Optional, Tuple
from openhop_core.protocol.packet_utils import calculate_lora_airtime_ms
@@ -142,6 +144,100 @@ class AirtimeManager:
}
class _RadioAirtimeBudget:
"""A radio-specific calculator backed by a possibly shared ledger."""
def __init__(
self,
ledger: AirtimeManager,
calculator: AirtimeManager,
lock: threading.RLock,
) -> None:
self._ledger = ledger
self._calculator = calculator
self._lock = lock
def _replace(self, ledger: AirtimeManager, calculator: AirtimeManager) -> None:
self._ledger = ledger
self._calculator = calculator
def calculate_airtime(self, *args, **kwargs) -> float:
with self._lock:
return self._calculator.calculate_airtime(*args, **kwargs)
def can_transmit(self, airtime_ms: float) -> Tuple[bool, float]:
with self._lock:
return self._ledger.can_transmit(airtime_ms)
def record_tx(self, airtime_ms: float) -> None:
with self._lock:
self._ledger.record_tx(airtime_ms)
def record_rx(self, airtime_ms: float) -> None:
with self._lock:
self._ledger.record_rx(airtime_ms)
def get_stats(self) -> dict:
with self._lock:
return self._ledger.get_stats()
@property
def max_airtime_per_minute(self) -> float:
with self._lock:
return self._ledger.max_airtime_per_minute
@property
def tx_history(self) -> list:
with self._lock:
return self._ledger.tx_history
@property
def total_airtime_ms(self) -> float:
with self._lock:
return self._ledger.total_airtime_ms
@property
def total_rx_airtime_ms(self) -> float:
with self._lock:
return self._ledger.total_rx_airtime_ms
@property
def radio_config(self) -> dict:
with self._lock:
return self._calculator.radio_config
@property
def spreading_factor(self):
with self._lock:
return self._calculator.spreading_factor
@property
def bandwidth(self):
with self._lock:
return self._calculator.bandwidth
@property
def coding_rate(self):
with self._lock:
return self._calculator.coding_rate
@property
def preamble_length(self):
with self._lock:
return self._calculator.preamble_length
@dataclass
class _BudgetState:
by_radio: dict
order: list
by_channel: dict
channel_of: dict
profile_backed: bool
default_radio_id: Optional[str]
default: _RadioAirtimeBudget
class AirtimeBudgets:
"""The duty-cycle budgets a node meters against: one per channel.
@@ -164,22 +260,14 @@ class AirtimeBudgets:
def __init__(self, config: dict):
self.config = config
# radio_id -> manager, in configured order.
self._managers: dict = {}
self._order: list = []
# Channel key -> manager, and radio_id -> channel key. Both are how a
# rebuild works out what carries over to what.
self._by_channel: dict = {}
self._channel_of: dict = {}
self._profile_backed = False
self._default_manager: Optional[AirtimeManager] = None
self._build()
self._lock = threading.RLock()
self._state = self._build_state()
# ------------------------------------------------------------------
# Construction
# ------------------------------------------------------------------
def _build(self) -> None:
def _build_state(self) -> _BudgetState:
from .config import build_metering_profiles
try:
@@ -188,49 +276,61 @@ class AirtimeBudgets:
logger.warning("Could not read radio profiles for duty cycle: %s", exc)
profiles = []
self._managers = {}
self._order = []
self._by_channel = {}
self._channel_of = {}
self._profile_backed = bool(profiles)
if not profiles:
# Nothing to read: meter exactly as before, keyed on None so a packet
# with no radio id still finds it.
self._order = [None]
self._channel_of = {None: None}
manager = AirtimeManager(self.config)
self._managers = {None: manager}
self._by_channel = {None: manager}
self._default_manager = manager
return
ledger = AirtimeManager(self.config)
view = _RadioAirtimeBudget(ledger, ledger, self._lock)
return _BudgetState(
by_radio={None: view},
order=[None],
by_channel={None: ledger},
channel_of={None: None},
profile_backed=False,
default_radio_id=None,
default=view,
)
shared = bool(self.config.get("duty_cycle", {}).get("shared_budget", False))
# One budget for every radio when asked for, otherwise one per channel:
# same frequency and bandwidth is the same channel.
order = []
channel_of = {}
for profile in profiles:
radio_id = str(profile["radio_id"])
self._order.append(radio_id)
self._channel_of[radio_id] = (
order.append(radio_id)
channel_of[radio_id] = (
"shared" if shared else (profile.get("frequency_hz"), profile.get("bandwidth_hz"))
)
by_channel = {}
by_radio = {}
for profile in profiles:
radio_id = str(profile["radio_id"])
key = self._channel_of[radio_id]
manager = self._by_channel.get(key)
if manager is None:
manager = AirtimeManager(
key = channel_of[radio_id]
ledger = by_channel.get(key)
if ledger is None:
ledger = AirtimeManager(
self.config,
radio_config=self._air_settings(profile),
max_airtime_per_minute=self._channel_budget(
[rid for rid, rid_key in self._channel_of.items() if rid_key == key]
[rid for rid, rid_key in channel_of.items() if rid_key == key]
),
)
self._by_channel[key] = manager
self._managers[radio_id] = manager
by_channel[key] = ledger
calculator = AirtimeManager(
self.config,
radio_config=self._air_settings(profile),
max_airtime_per_minute=ledger.max_airtime_per_minute,
)
by_radio[radio_id] = _RadioAirtimeBudget(ledger, calculator, self._lock)
self._default_manager = self._managers[self._default_radio_id()]
default_radio_id = self._default_radio_id(by_radio, order)
return _BudgetState(
by_radio=by_radio,
order=order,
by_channel=by_channel,
channel_of=channel_of,
profile_backed=True,
default_radio_id=default_radio_id,
default=by_radio[default_radio_id],
)
def _air_settings(self, profile: dict) -> dict:
"""A profile in the shape AirtimeManager reads air settings from.
@@ -309,93 +409,68 @@ class AirtimeBudgets:
prove it was the quiet one; a second refresh then finds each side's own
key and stops there rather than doubling again.
"""
previous_channel_of = dict(self._channel_of)
previous_default = self._default_manager
snapshots = {
key: (
list(manager.tx_history),
manager.total_airtime_ms,
manager.total_rx_airtime_ms,
)
for key, manager in self._by_channel.items()
}
self._build()
for key, manager in self._by_channel.items():
sources = {
previous_channel_of[radio_id]
for radio_id, radio_key in self._channel_of.items()
if radio_key == key and radio_id in previous_channel_of
}
if not sources:
# A channel nothing was metering: a radio added, or one retuned
# onto a band the node was not using. An empty window on a node
# that has been transmitting would let the next minute's budget
# be spent twice, so it starts from the busiest channel the node
# was already keeping. The busiest single one, not the sum of
# them: summing is what turns repeated rebuilds into 100 -> 400
# -> 1600 ms and wedges transmission for the rest of the window.
busiest = max(
snapshots.values(),
key=lambda carried: sum(at for _, at in carried[0]),
default=None,
with self._lock:
previous = self._state
snapshots = {
key: (
list(manager.tx_history),
manager.total_airtime_ms,
manager.total_rx_airtime_ms,
)
if busiest is None:
continue
carried_all = [busiest]
for key, manager in previous.by_channel.items()
}
rebuilt = self._build_state()
for key, manager in rebuilt.by_channel.items():
sources = {
previous.channel_of[radio_id]
for radio_id, radio_key in rebuilt.channel_of.items()
if radio_key == key and radio_id in previous.channel_of
}
if not sources:
busiest = max(
snapshots.values(),
key=lambda carried: sum(at for _, at in carried[0]),
default=None,
)
if busiest is None:
continue
carried_all = [busiest]
else:
carried_all = [snapshots[source] for source in sources]
history: list = []
total_tx = 0.0
total_rx = 0.0
for carried in carried_all:
history.extend(carried[0])
total_tx += carried[1]
total_rx += carried[2]
manager.tx_history = sorted(history, key=lambda entry: entry[0])
manager.total_airtime_ms = total_tx
manager.total_rx_airtime_ms = total_rx
self._reuse_views(previous, rebuilt)
self._state = rebuilt
@staticmethod
def _reuse_views(previous: _BudgetState, rebuilt: _BudgetState) -> None:
"""Keep radio handles valid across the atomic state swap."""
default_id = rebuilt.default_radio_id
for radio_id, new_view in list(rebuilt.by_radio.items()):
if radio_id == default_id:
old_view = previous.default
else:
carried_all = [snapshots[source] for source in sources]
old_view = previous.by_radio.get(radio_id)
if old_view is previous.default:
old_view = None
if old_view is None or old_view is new_view:
continue
old_view._replace(new_view._ledger, new_view._calculator)
rebuilt.by_radio[radio_id] = old_view
rebuilt.default = rebuilt.by_radio[default_id]
history: list = []
total_tx = 0.0
total_rx = 0.0
for carried in carried_all:
history.extend(carried[0])
total_tx += carried[1]
total_rx += carried[2]
manager.tx_history = sorted(history, key=lambda entry: entry[0])
manager.total_airtime_ms = total_tx
manager.total_rx_airtime_ms = total_rx
self._keep_default_identity(previous_default)
def _keep_default_identity(self, was_default: Optional[AirtimeManager]) -> None:
"""Let the object that was the default manager go on being it.
Callers hold onto it: main.py hands it to NeighborScopeHelper at boot,
which keeps the reference for the life of the process. Replacing it with
a new object would leave that helper reading a manager nothing charges
any more, so it would see an empty window and stop throttling itself.
Adopt the rebuilt settings into the original object instead.
Takes the previous default as an argument rather than working it out
again: ``config`` is mutated in place by the daemon, so by the time this
runs ``fabric.default_radio`` may already name a different radio, and
recomputing would preserve the wrong object -- orphaning the one the
helper is holding, which is the failure this exists to prevent.
"""
if was_default is None or self._default_manager is None:
return
rebuilt = self._default_manager
if rebuilt is was_default:
return
was_default.radio_config = rebuilt.radio_config
was_default.refresh_radio_params(rebuilt.radio_config)
was_default.max_airtime_per_minute = rebuilt.max_airtime_per_minute
was_default.tx_history = rebuilt.tx_history
was_default.total_airtime_ms = rebuilt.total_airtime_ms
was_default.total_rx_airtime_ms = rebuilt.total_rx_airtime_ms
for radio_id, manager in self._managers.items():
if manager is rebuilt:
self._managers[radio_id] = was_default
for key, manager in self._by_channel.items():
if manager is rebuilt:
self._by_channel[key] = was_default
self._default_manager = was_default
def _default_radio_id(self) -> Optional[str]:
def _default_radio_id(self, by_radio: dict, order: list) -> Optional[str]:
"""The radio Fabric transmits on by default.
Mirrors build_radio_stack's rule -- ``fabric.default_radio`` when set,
@@ -406,22 +481,24 @@ class AirtimeBudgets:
fabric = self.config.get("fabric")
fabric = fabric if isinstance(fabric, dict) else {}
configured = fabric.get("default_radio") or fabric.get("default_radio_id")
if configured and str(configured) in self._managers:
if configured and str(configured) in by_radio:
return str(configured)
return self._order[0] if self._order else None
return order[0] if order else None
# ------------------------------------------------------------------
# Reading
# ------------------------------------------------------------------
@property
def default(self) -> AirtimeManager:
def default(self) -> _RadioAirtimeBudget:
"""The manager everything that reports one channel's figures reads."""
return self._default_manager
with self._lock:
return self._state.default
@property
def multi(self) -> bool:
return len(self._order) > 1
with self._lock:
return len(self._state.order) > 1
@property
def profile_backed(self) -> bool:
@@ -433,26 +510,36 @@ class AirtimeBudgets:
what happens if a caller tests ``multi`` and takes the legacy path --
undoes the whole point on every radio save from the web UI.
"""
return self._profile_backed
with self._lock:
return self._state.profile_backed
def radio_ids(self) -> list:
"""Configured radio ids, in order. ``[None]`` when nothing was profiled."""
return list(self._order)
with self._lock:
return list(self._state.order)
def for_radio(self, radio_id: Optional[str]) -> AirtimeManager:
def shares_budget(self, first_radio_id: Optional[str], second_radio_id: Optional[str]) -> bool:
"""Whether two radios debit the same channel ledger."""
with self._lock:
first = self.for_radio(first_radio_id)
second = self.for_radio(second_radio_id)
return first._ledger is second._ledger
def for_radio(self, radio_id: Optional[str]) -> _RadioAirtimeBudget:
"""The budget a send on this radio is charged to.
An unknown id falls back to the default radio rather than going
unmetered: an unrecognised label is a reason to be careful, not a reason
to transmit freely.
"""
if radio_id is None:
return self.default
manager = self._managers.get(str(radio_id))
if manager is not None:
return manager
logger.debug("No duty-cycle budget for radio %s; metering on the default", radio_id)
return self.default
with self._lock:
if radio_id is None:
return self._state.default
manager = self._state.by_radio.get(str(radio_id))
if manager is not None:
return manager
logger.debug("No duty-cycle budget for radio %s; metering on the default", radio_id)
return self._state.default
def per_radio_stats(self) -> list:
"""``[{radio_id, ...stats}]`` on a multi-radio node, else an empty list.
@@ -460,12 +547,14 @@ class AirtimeBudgets:
Radios sharing a channel report the same figures, because they are the
same budget: that is the statement, not a duplication.
"""
if not self.multi:
return []
return [
{"radio_id": radio_id, **self._managers[radio_id].get_stats()}
for radio_id in self._order
]
with self._lock:
state = self._state
if len(state.order) <= 1:
return []
return [
{"radio_id": radio_id, **state.by_radio[radio_id].get_stats()}
for radio_id in state.order
]
def node_stats(self) -> dict:
"""One set of figures for the whole node, in AirtimeManager's shape.
@@ -485,15 +574,16 @@ class AirtimeBudgets:
On a single-radio node every figure is that one manager's, unchanged.
"""
managers = list(self._by_channel.values())
if len(managers) == 1:
return managers[0].get_stats()
with self._lock:
managers = list(self._state.by_channel.values())
if len(managers) == 1:
return managers[0].get_stats()
stats = [manager.get_stats() for manager in managers]
return {
"current_airtime_ms": sum(s["current_airtime_ms"] for s in stats),
"max_airtime_ms": sum(s["max_airtime_ms"] for s in stats),
"utilization_percent": max(s["utilization_percent"] for s in stats),
"total_airtime_ms": sum(s["total_airtime_ms"] for s in stats),
"total_rx_airtime_ms": sum(s["total_rx_airtime_ms"] for s in stats),
}
stats = [manager.get_stats() for manager in managers]
return {
"current_airtime_ms": sum(s["current_airtime_ms"] for s in stats),
"max_airtime_ms": sum(s["max_airtime_ms"] for s in stats),
"utilization_percent": max(s["utilization_percent"] for s in stats),
"total_airtime_ms": sum(s["total_airtime_ms"] for s in stats),
"total_rx_airtime_ms": sum(s["total_rx_airtime_ms"] for s in stats),
}
+9 -11
View File
@@ -163,17 +163,9 @@ class ConfigManager:
return
budgets = getattr(repeater_handler, "airtime_budgets", None)
if budgets is not None and getattr(budgets, "profile_backed", False):
# Each radio meters on its own modulation, so rebuild them all from
# the config rather than overwriting the default radio's profile
# with the top-level block it does not use.
#
# Tested on profile_backed, not on how many radios there are: a node
# with one radios[] entry has that entry's own bandwidth and its own
# band's limit, and the legacy branch below would overwrite them with
# the top-level block on every radio save from the web UI -- a
# 62.5 kHz radio silently metered at 500 kHz from the first save
# until the next restart.
if budgets is not None:
# Rebuild the container even on a legacy single-radio node because
# it owns both the cached modulation and the cached duty-cycle limit.
budgets.refresh()
repeater_handler.airtime_mgr = budgets.default
return
@@ -430,11 +422,17 @@ class ConfigManager:
logger.info("radio_type change detected; service restart required")
live_update_ok = False
if "radios" in sections:
logger.info("Non-default radio change detected; service restart required")
live_update_ok = False
if "kiss" in sections and self._kiss_transport_restart_required():
live_update_ok = False
if "radio" in sections:
live_update_ok = self._apply_live_radio_config() and live_update_ok
elif "duty_cycle" in sections:
self._refresh_airtime_radio_params()
if "http" in sections:
live_update_ok = self._apply_live_http_config() and live_update_ok
+37 -23
View File
@@ -4905,26 +4905,41 @@ class APIEndpoints:
if "mesh" not in self.config:
self.config["mesh"] = {}
# Optional multi-radio target. When radios[] exists and radio_id is
# provided, LoRa air settings are written into that entry's radio{}
# instead of (or in addition to) the legacy top-level radio{}.
# A Fabric edit targets the named radio, or its default when the old
# single-radio request shape omits radio_id.
target_radio_id = data.get("radio_id")
radios_list = self.config.get("radios")
target_radio_cfg = self.config["radio"]
target_entry = None
if isinstance(radios_list, list) and target_radio_id:
target_is_default = False
effective_target_id = target_radio_id
if isinstance(radios_list, list) and radios_list:
fabric = (
self.config.get("fabric") if isinstance(self.config.get("fabric"), dict) else {}
)
default_id = fabric.get("default_radio") or fabric.get("default_radio_id")
if default_id is None:
first = radios_list[0] if isinstance(radios_list[0], dict) else {}
default_id = first.get("id") or first.get("radio_id")
if effective_target_id is None:
effective_target_id = default_id
if isinstance(radios_list, list) and effective_target_id:
for entry in radios_list:
if not isinstance(entry, dict):
continue
rid = entry.get("id") or entry.get("radio_id")
if str(rid) == str(target_radio_id):
if str(rid) == str(effective_target_id):
target_entry = entry
if not isinstance(entry.get("radio"), dict):
entry["radio"] = {}
target_radio_cfg = entry["radio"]
break
if target_entry is None:
return self._error(f"Unknown radio_id={target_radio_id!r}")
return self._error(f"Unknown radio_id={effective_target_id!r}")
target_is_default = default_id is not None and str(default_id) == str(
effective_target_id
)
def _set_radio_param(key, value):
target_radio_cfg[key] = value
@@ -4932,21 +4947,10 @@ class APIEndpoints:
# radio or when no radios[] list is active.
if target_entry is None:
self.config["radio"][key] = value
else:
default_id = None
fabric = (
self.config.get("fabric")
if isinstance(self.config.get("fabric"), dict)
else {}
)
default_id = fabric.get("default_radio") or fabric.get("default_radio_id")
if default_id is None and radios_list:
first = radios_list[0] if isinstance(radios_list[0], dict) else {}
default_id = first.get("id") or first.get("radio_id")
if default_id is not None and str(default_id) == str(target_radio_id):
if "radio" not in self.config or not isinstance(self.config["radio"], dict):
self.config["radio"] = {}
self.config["radio"][key] = value
elif target_is_default:
if "radio" not in self.config or not isinstance(self.config["radio"], dict):
self.config["radio"] = {}
self.config["radio"][key] = value
# Update TX power (up to 30 dBm for high-power radios)
if "tx_power" in data:
@@ -5115,8 +5119,18 @@ class APIEndpoints:
if not applied:
return self._error("No valid settings provided")
live_sections = ["repeater", "delays", "radio"]
if target_entry is not None:
live_sections = ["repeater", "delays"]
radio_fields = {
"tx_power",
"frequency",
"bandwidth",
"spreading_factor",
"coding_rate",
}
radio_changed = any(field in data for field in radio_fields)
if radio_changed and (target_entry is None or target_is_default):
live_sections.append("radio")
elif radio_changed:
live_sections.append("radios")
if "mesh" in self.config and any(k in data for k in ("path_hash_mode", "loop_detect")):
live_sections.append("mesh")
+63
View File
@@ -2634,6 +2634,69 @@ def test_update_radio_config_owner_info_and_mesh_fields(cherrypy_ctx):
assert "owner.info" in out["data"].get("applied", [])
def test_update_non_default_radio_requires_restart(cherrypy_ctx):
request, _ = cherrypy_ctx
request.method = "POST"
request.json = {"radio_id": "link", "bandwidth": 500000}
api = _make_api(
{
"radio": {"bandwidth": 62500},
"radios": [
{"id": "local", "radio": {"bandwidth": 62500}},
{"id": "link", "radio": {"bandwidth": 62500}},
],
"fabric": {"default_radio": "local"},
"repeater": {},
"delays": {},
}
)
api.config_manager.update_and_save.return_value = {
"success": True,
"saved": True,
"live_updated": False,
}
out = api.update_radio_config()
assert out["success"] is True
assert out["data"]["restart_required"] is True
sections = api.config_manager.update_and_save.call_args.kwargs["live_update_sections"]
assert "radios" in sections
assert "radio" not in sections
def test_update_radio_without_id_keeps_the_fabric_default_entry_in_sync(cherrypy_ctx):
request, _ = cherrypy_ctx
request.method = "POST"
request.json = {"bandwidth": 500000}
api = _make_api(
{
"radio": {"bandwidth": 62500},
"radios": [
{"id": "local", "radio": {"bandwidth": 62500}},
{"id": "link", "radio": {"bandwidth": 62500}},
],
"fabric": {"default_radio": "local"},
"repeater": {},
"delays": {},
}
)
api.config_manager.update_and_save.return_value = {
"success": True,
"saved": True,
"live_updated": True,
}
out = api.update_radio_config()
assert out["success"] is True
assert api.config["radio"]["bandwidth"] == 500000
assert api.config["radios"][0]["radio"]["bandwidth"] == 500000
sections = api.config_manager.update_and_save.call_args.kwargs["live_update_sections"]
assert "radio" in sections
assert "radios" not in sections
class _FakeIdentityObj:
def __init__(self, first=0x42):
self._pk = bytes([first]) + (b"A" * 31)
+65 -1
View File
@@ -1,6 +1,6 @@
import math
from repeater.airtime import AirtimeManager
from repeater.airtime import AirtimeBudgets, AirtimeManager
from repeater.config_manager import ConfigManager
@@ -216,3 +216,67 @@ def test_failed_live_radio_apply_leaves_airtime_manager_unchanged():
assert airtime_mgr.spreading_factor == 7
assert math.isclose(airtime_mgr.calculate_airtime(50), 97.536, rel_tol=1e-9)
def test_live_duty_cycle_update_refreshes_the_cached_limit():
config = {
"radio": {
"frequency": 868000000,
"bandwidth": 125000,
"spreading_factor": 7,
"coding_rate": 5,
"preamble_length": 8,
},
"duty_cycle": {"max_airtime_per_minute": 3600, "enforcement_enabled": True},
}
budgets = AirtimeBudgets(config)
handler = _DummyRepeaterHandler(config)
handler.airtime_budgets = budgets
handler.airtime_mgr = budgets.default
daemon = _DummyDaemon(config, _DummySX1262Radio())
daemon.repeater_handler = handler
manager = ConfigManager("/tmp/config.yaml", config, daemon)
config["duty_cycle"]["max_airtime_per_minute"] = 600
assert manager.live_update_daemon(["duty_cycle"])
assert budgets.default.max_airtime_per_minute == 600
def test_non_default_radio_change_requires_restart_and_keeps_runtime_metering():
config = {
"radio": {
"frequency": 910100000,
"bandwidth": 500000,
"spreading_factor": 7,
"coding_rate": 5,
"preamble_length": 8,
},
"radios": [
{
"id": "local",
"radio_type": "sx1262",
"radio": {"frequency": 910100000, "bandwidth": 500000},
},
{
"id": "link",
"radio_type": "sx1262",
"radio": {"frequency": 910525000, "bandwidth": 62500},
},
],
"fabric": {"use_fabric": True, "default_radio": "local"},
"duty_cycle": {"max_airtime_per_minute": 3600, "enforcement_enabled": True},
}
budgets = AirtimeBudgets(config)
old_airtime = budgets.for_radio("link").calculate_airtime(50)
handler = _DummyRepeaterHandler(config)
handler.airtime_budgets = budgets
handler.airtime_mgr = budgets.default
daemon = _DummyDaemon(config, _DummySX1262Radio())
daemon.repeater_handler = handler
manager = ConfigManager("/tmp/config.yaml", config, daemon)
config["radios"][1]["radio"]["bandwidth"] = 500000
assert manager.live_update_daemon(["radios"]) is False
assert budgets.for_radio("link").calculate_airtime(50) == old_airtime
+48 -4
View File
@@ -8,6 +8,7 @@ manager charging both at one modulation is wrong in both directions at once.
from __future__ import annotations
import copy
import threading
from collections import OrderedDict
from unittest.mock import AsyncMock, MagicMock, patch
@@ -108,16 +109,29 @@ def test_radios_on_one_channel_share_a_budget():
same = [_radio("north", WIDE), _radio("south", WIDE)]
budgets = AirtimeBudgets(_config(same))
assert budgets.for_radio("north") is budgets.for_radio("south")
assert budgets.shares_budget("north", "south")
budgets.for_radio("north").record_tx(3600)
assert budgets.for_radio("south").can_transmit(10)[0] is False
def test_radios_on_one_channel_keep_their_own_modulation():
slow = dict(WIDE, spreading_factor=12)
budgets = AirtimeBudgets(_config([_radio("fast", WIDE), _radio("slow", slow)]))
assert budgets.shares_budget("fast", "slow")
assert budgets.for_radio("slow").calculate_airtime(50) > budgets.for_radio(
"fast"
).calculate_airtime(50)
def test_shared_budget_restores_one_allowance_for_the_whole_node():
budgets = AirtimeBudgets(_config(BRIDGE, shared_budget=True))
assert budgets.for_radio("local") is budgets.for_radio("link")
assert budgets.shares_budget("local", "link")
assert budgets.for_radio("link").calculate_airtime(50) == pytest.approx(
budgets.for_radio("local").calculate_airtime(50) * 8, rel=0.05
)
budgets.for_radio("link").record_tx(3600)
assert budgets.for_radio("local").can_transmit(10)[0] is False
@@ -689,7 +703,7 @@ def test_radios_merging_onto_one_channel_sum_what_they_spent():
config["radios"][1]["radio"] = dict(WIDE) # link retuned onto local's channel
budgets.refresh()
assert budgets.for_radio("local") is budgets.for_radio("link")
assert budgets.shares_budget("local", "link")
assert budgets.for_radio("local").get_stats()["current_airtime_ms"] == 3600
assert budgets.for_radio("local").can_transmit(10)[0] is False
@@ -808,7 +822,7 @@ def test_an_unprofilable_radio_is_metered_on_the_top_level_block():
budgets = AirtimeBudgets(config)
assert budgets.for_radio("link") is budgets.for_radio("local")
assert budgets.shares_budget("link", "local")
assert budgets.for_radio("link").bandwidth == WIDE["bandwidth"]
@@ -844,6 +858,34 @@ def test_a_rebuild_is_idempotent_for_an_unchanged_config():
assert budgets.for_radio("local").get_stats()["current_airtime_ms"] == 0
def test_readers_cannot_observe_a_partially_rebuilt_state(monkeypatch):
budgets = AirtimeBudgets(_config(BRIDGE))
build_started = threading.Event()
release_build = threading.Event()
read_finished = threading.Event()
original_build = budgets._build_state
def paused_build():
build_started.set()
assert release_build.wait(timeout=2)
return original_build()
monkeypatch.setattr(budgets, "_build_state", paused_build)
refresh_thread = threading.Thread(target=budgets.refresh)
refresh_thread.start()
assert build_started.wait(timeout=2)
read_thread = threading.Thread(target=lambda: (budgets.node_stats(), read_finished.set()))
read_thread.start()
assert not read_finished.wait(timeout=0.05)
release_build.set()
refresh_thread.join(timeout=2)
read_thread.join(timeout=2)
assert not refresh_thread.is_alive()
assert read_finished.is_set()
def test_a_new_radio_starts_from_the_busiest_channel_not_the_sum_of_them():
config = _config([_radio("local", WIDE), _radio("link", NARROW)])
budgets = AirtimeBudgets(config)
@@ -872,6 +914,8 @@ def test_changing_the_default_radio_keeps_the_manager_captured_at_boot():
assert budgets.default is captured
assert budgets.for_radio("link") is captured
assert captured.bandwidth == NARROW["bandwidth"]
assert not budgets.shares_budget("local", "link")
assert budgets.for_radio("local").bandwidth == WIDE["bandwidth"]
def test_the_captured_manager_still_receives_what_is_charged_to_it():