mirror of
https://github.com/firestar5683/StarPilot.git
synced 2026-10-05 05:44:03 +08:00
firestar5683
326 lines
9.4 KiB
Python
326 lines
9.4 KiB
Python
import threading
|
|
|
|
import pytest
|
|
|
|
from openpilot.selfdrive.ui.lib import ui_param_cache
|
|
from openpilot.selfdrive.ui.lib.ui_param_cache import UIParamCache
|
|
|
|
|
|
class FakeParams:
|
|
def __init__(self):
|
|
self.values = {"enabled": False, "count": 1}
|
|
self.calls = []
|
|
|
|
def get(self, key, **_kwargs):
|
|
self.calls.append(("get", key))
|
|
return self.values.get(key)
|
|
|
|
def get_bool(self, key, **_kwargs):
|
|
self.calls.append(("get_bool", key))
|
|
return bool(self.values.get(key, False))
|
|
|
|
def get_int(self, key, **_kwargs):
|
|
self.calls.append(("get_int", key))
|
|
return int(self.values.get(key, 0))
|
|
|
|
def get_float(self, key, **_kwargs):
|
|
self.calls.append(("get_float", key))
|
|
return float(self.values.get(key, 0.0))
|
|
|
|
def put(self, key, value, **_kwargs):
|
|
self.values[key] = value
|
|
|
|
def put_bool(self, key, value, **_kwargs):
|
|
self.values[key] = value
|
|
|
|
def put_int(self, key, value, **_kwargs):
|
|
self.values[key] = value
|
|
|
|
def put_float(self, key, value, **_kwargs):
|
|
self.values[key] = value
|
|
|
|
def put_nonblocking(self, key, value):
|
|
self.values[key] = value
|
|
|
|
def put_bool_nonblocking(self, key, value):
|
|
self.values[key] = value
|
|
|
|
def remove(self, key):
|
|
self.values.pop(key, None)
|
|
|
|
def clear_all(self, marker=None):
|
|
self.calls.append(("clear_all", marker))
|
|
self.values.clear()
|
|
|
|
|
|
def test_reads_are_shared_until_ttl():
|
|
now = [0.0]
|
|
params = FakeParams()
|
|
cached = UIParamCache(params, ttl=0.1, clock=lambda: now[0])
|
|
|
|
assert not cached.get_bool("enabled")
|
|
assert not cached.get_bool("enabled")
|
|
assert params.calls == [("get_bool", "enabled")]
|
|
|
|
now[0] = 0.11
|
|
params.values["enabled"] = True
|
|
assert cached.get_bool("enabled")
|
|
assert params.calls.count(("get_bool", "enabled")) == 2
|
|
|
|
|
|
def test_writes_invalidate_immediately():
|
|
params = FakeParams()
|
|
cached = UIParamCache(params, ttl=10.0)
|
|
|
|
assert not cached.get_bool("enabled")
|
|
cached.put_bool("enabled", True)
|
|
assert cached.get_bool("enabled")
|
|
assert params.calls.count(("get_bool", "enabled")) == 2
|
|
|
|
|
|
@pytest.mark.parametrize("background", [False, True])
|
|
def test_zero_ttl_disables_caching(background):
|
|
params = FakeParams()
|
|
cached = UIParamCache(params, ttl=0.0)
|
|
if background:
|
|
cached.start()
|
|
try:
|
|
assert not cached.get_bool("enabled")
|
|
params.values["enabled"] = True
|
|
assert cached.get_bool("enabled")
|
|
assert params.calls.count(("get_bool", "enabled")) == 2
|
|
finally:
|
|
cached.stop()
|
|
|
|
|
|
@pytest.mark.parametrize("method", ["put", "put_bool", "put_int", "put_float", "put_nonblocking", "put_bool_nonblocking"])
|
|
def test_all_write_paths_invalidate(method):
|
|
params = FakeParams()
|
|
cached = UIParamCache(params, ttl=10.0)
|
|
assert not cached.get_bool("enabled")
|
|
getattr(cached, method)("enabled", True)
|
|
assert cached.get_bool("enabled")
|
|
assert params.calls.count(("get_bool", "enabled")) == 2
|
|
|
|
|
|
def test_remove_and_clear_all_invalidate():
|
|
params = FakeParams()
|
|
cached = UIParamCache(params, ttl=10.0)
|
|
|
|
assert cached.get_int("count") == 1
|
|
cached.remove("count")
|
|
assert cached.get_int("count") == 0
|
|
|
|
assert not cached.get_bool("enabled")
|
|
params.values["enabled"] = True
|
|
cached.clear_all("flag")
|
|
params.values["enabled"] = True
|
|
assert cached.get_bool("enabled")
|
|
assert ("clear_all", "flag") in params.calls
|
|
|
|
|
|
class ControlledParams(FakeParams):
|
|
def __init__(self):
|
|
super().__init__()
|
|
self.block_next = False
|
|
self.fail_next = False
|
|
self.read_started = threading.Event()
|
|
self.release_read = threading.Event()
|
|
self.reader_threads = []
|
|
|
|
def get_int(self, key, *args, **kwargs):
|
|
self.reader_threads.append(threading.current_thread())
|
|
value = super().get_int(key, **kwargs)
|
|
if self.block_next:
|
|
self.block_next = False
|
|
self.read_started.set()
|
|
if not self.release_read.wait(3):
|
|
raise TimeoutError("Blocked parameter test read was not released")
|
|
if self.fail_next:
|
|
self.fail_next = False
|
|
raise OSError("Parameter temporarily unavailable")
|
|
return value
|
|
|
|
|
|
@pytest.fixture
|
|
def async_cache():
|
|
params = ControlledParams()
|
|
now = [0.0]
|
|
cache = UIParamCache(params, ttl=0.1, clock=lambda: now[0])
|
|
cache.start()
|
|
try:
|
|
yield params, cache, now
|
|
finally:
|
|
params.release_read.set()
|
|
cache.stop()
|
|
|
|
|
|
def test_background_cache_keeps_cold_reads_synchronous(async_cache):
|
|
params, cache, _ = async_cache
|
|
assert cache.get_int("count") == 1
|
|
assert params.reader_threads == [threading.current_thread()]
|
|
|
|
|
|
def test_stale_read_returns_while_refresh_is_blocked_and_deduplicates(async_cache):
|
|
params, cache, now = async_cache
|
|
assert cache.get_int("count") == 1
|
|
params.values["count"] = 2
|
|
params.block_next = True
|
|
now[0] = 0.2
|
|
|
|
assert cache.get_int("count") == 1
|
|
assert params.read_started.wait(1)
|
|
for _ in range(100):
|
|
assert cache.get_int("count") == 1
|
|
assert len(params.calls) == 2
|
|
assert len(cache._pending) == 1
|
|
|
|
now[0] = 0.5
|
|
params.release_read.set()
|
|
cache._refresh_queue.join()
|
|
assert cache.get_int("count") == 2
|
|
assert len(params.calls) == 2
|
|
assert not cache._pending
|
|
|
|
|
|
@pytest.mark.parametrize("clear_all", [False, True])
|
|
def test_invalidating_an_inflight_refresh_preserves_new_reads(async_cache, clear_all):
|
|
params, cache, now = async_cache
|
|
assert cache.get_int("count") == 1
|
|
params.block_next = True
|
|
now[0] = 0.2
|
|
assert cache.get_int("count") == 1
|
|
assert params.read_started.wait(1)
|
|
|
|
if clear_all:
|
|
cache.clear_all()
|
|
params.values["count"] = 3
|
|
else:
|
|
cache.put_int("count", 3)
|
|
assert cache.get_int("count") == 3
|
|
params.release_read.set()
|
|
cache._refresh_queue.join()
|
|
assert cache.get_int("count") == 3
|
|
|
|
|
|
def test_background_errors_keep_value_and_allow_later_refresh(async_cache):
|
|
params, cache, now = async_cache
|
|
assert cache.get_int("count") == 1
|
|
params.fail_next = True
|
|
params.values["count"] = 2
|
|
now[0] = 0.2
|
|
assert cache.get_int("count") == 1
|
|
cache._refresh_queue.join()
|
|
assert cache.get_int("count") == 1
|
|
assert cache._worker.is_alive()
|
|
assert len(params.calls) == 2
|
|
|
|
now[0] = 0.4
|
|
assert cache.get_int("count") == 1
|
|
cache._refresh_queue.join()
|
|
assert cache.get_int("count") == 2
|
|
|
|
|
|
def test_blocking_reads_remain_synchronous(async_cache):
|
|
params, cache, now = async_cache
|
|
assert cache.get_int("count", block=True) == 1
|
|
params.values["count"] = 2
|
|
now[0] = 0.2
|
|
assert cache.get_int("count", block=True) == 2
|
|
assert params.reader_threads == [threading.current_thread()] * 2
|
|
assert not cache._pending
|
|
|
|
|
|
def test_background_cache_does_not_refresh_unused_settings(async_cache):
|
|
params, cache, now = async_cache
|
|
assert cache.get_bool("enabled") is False
|
|
assert cache.get_int("count") == 1
|
|
now[0] = 0.2
|
|
cache.get_int("count")
|
|
cache._refresh_queue.join()
|
|
assert params.calls.count(("get_bool", "enabled")) == 1
|
|
assert params.calls.count(("get_int", "count")) == 2
|
|
|
|
|
|
def test_stop_releases_worker_and_restores_synchronous_refresh(async_cache):
|
|
params, cache, now = async_cache
|
|
assert cache.get_int("count") == 1
|
|
worker = cache._worker
|
|
cache.start()
|
|
assert cache._worker is worker
|
|
cache.stop()
|
|
assert not worker.is_alive()
|
|
params.values["count"] = 2
|
|
now[0] = 0.2
|
|
assert cache.get_int("count") == 2
|
|
assert params.reader_threads == [threading.current_thread()] * 2
|
|
|
|
|
|
def test_timed_out_worker_cannot_publish_after_restart(async_cache):
|
|
params, cache, now = async_cache
|
|
assert cache.get_int("count") == 1
|
|
params.values["count"] = 2
|
|
params.block_next = True
|
|
now[0] = 0.2
|
|
assert cache.get_int("count") == 1
|
|
assert params.read_started.wait(1)
|
|
old_worker, old_queue = cache._worker, cache._refresh_queue
|
|
cache.stop(timeout=0.01)
|
|
assert old_worker.is_alive()
|
|
|
|
params.values["count"] = 3
|
|
cache.start()
|
|
assert cache.get_int("count") == 1
|
|
cache._refresh_queue.join()
|
|
assert cache.get_int("count") == 3
|
|
params.release_read.set()
|
|
old_queue.join()
|
|
old_worker.join(1)
|
|
assert not old_worker.is_alive()
|
|
assert cache.get_int("count") == 3
|
|
|
|
|
|
def test_worker_lowers_its_own_realtime_priority(monkeypatch):
|
|
calls = []
|
|
configured = threading.Event()
|
|
|
|
def set_scheduler(pid, policy, priority):
|
|
calls.append((pid, policy, priority, threading.current_thread()))
|
|
configured.set()
|
|
|
|
monkeypatch.setattr(ui_param_cache, "PC", False)
|
|
monkeypatch.setattr(ui_param_cache.os, "SCHED_OTHER", 0, raising=False)
|
|
monkeypatch.setattr(ui_param_cache.os, "sched_param", lambda priority: priority, raising=False)
|
|
monkeypatch.setattr(ui_param_cache.os, "sched_setscheduler", set_scheduler, raising=False)
|
|
cache = UIParamCache(FakeParams())
|
|
cache.start()
|
|
try:
|
|
assert configured.wait(1)
|
|
assert calls == [(0, 0, 0, cache._worker)]
|
|
finally:
|
|
cache.stop()
|
|
|
|
|
|
def test_worker_priority_failure_restores_synchronous_reads(monkeypatch):
|
|
def set_scheduler(*_):
|
|
raise PermissionError("Unable to change scheduler")
|
|
|
|
monkeypatch.setattr(ui_param_cache, "PC", False)
|
|
monkeypatch.setattr(ui_param_cache.os, "SCHED_OTHER", 0, raising=False)
|
|
monkeypatch.setattr(ui_param_cache.os, "sched_param", lambda priority: priority, raising=False)
|
|
monkeypatch.setattr(ui_param_cache.os, "sched_setscheduler", set_scheduler, raising=False)
|
|
monkeypatch.setattr(ui_param_cache.cloudlog, "exception", lambda *_: None)
|
|
params = FakeParams()
|
|
now = [0.0]
|
|
cache = UIParamCache(params, clock=lambda: now[0])
|
|
assert cache.get_int("count") == 1
|
|
cache.start()
|
|
worker = cache._worker
|
|
if worker is not None:
|
|
worker.join(1)
|
|
assert not worker.is_alive()
|
|
params.values["count"] = 2
|
|
now[0] = 0.2
|
|
assert cache.get_int("count") == 2
|
|
assert cache._worker is None
|