perf: per-key overlay cache locks and quantized bbox keys
Nearby pans share a 0.25° cache cell. Slow NWS/WFIGS factories no longer hold a process-wide lock that stalls trains/aircraft/radar fills.
This commit is contained in:
parent
07638288a9
commit
84532d505a
2 changed files with 91 additions and 8 deletions
|
|
@ -58,7 +58,9 @@ _COMPASS = {
|
|||
}
|
||||
|
||||
_cache: dict[str, tuple[float, Any]] = {}
|
||||
_cache_lock = asyncio.Lock()
|
||||
_key_locks: dict[str, asyncio.Lock] = {}
|
||||
_key_locks_guard = asyncio.Lock()
|
||||
_QUANT = 0.25 # degrees — pan jitter inside a cell reuses the TTL entry
|
||||
|
||||
# Last-known AIS positions (MMSI -> marker). Filled by ais_stream worker.
|
||||
vessel_last_known: dict[str, dict] = {}
|
||||
|
|
@ -116,6 +118,33 @@ def parse_bbox(bbox: str) -> tuple[float, float, float, float]:
|
|||
return minlon, minlat, maxlon, maxlat
|
||||
|
||||
|
||||
def quantize_bbox(
|
||||
minlon: float, minlat: float, maxlon: float, maxlat: float,
|
||||
step: float = _QUANT,
|
||||
) -> tuple[float, float, float, float]:
|
||||
"""Snap a viewport to a coarse cell so nearby pans share a cache key.
|
||||
|
||||
The returned envelope is expanded to cover the original box.
|
||||
"""
|
||||
def q_down(v: float, lo: float, hi: float) -> float:
|
||||
v = max(lo, min(hi, v))
|
||||
return math.floor(v / step) * step
|
||||
|
||||
return (
|
||||
round(q_down(minlon, -180.0, 180.0), 4),
|
||||
round(q_down(minlat, -90.0, 90.0), 4),
|
||||
round(q_down(maxlon, -180.0, 180.0) + step, 4),
|
||||
round(q_down(maxlat, -90.0, 90.0) + step, 4),
|
||||
)
|
||||
|
||||
|
||||
def bbox_cell_key(bbox: str | None) -> str:
|
||||
"""Stable cache-key fragment for a viewport (or 'all')."""
|
||||
if not bbox:
|
||||
return "all"
|
||||
return ",".join(f"{v:.4f}" for v in quantize_bbox(*parse_bbox(bbox)))
|
||||
|
||||
|
||||
def bbox_center_radius_nm(
|
||||
minlon: float, minlat: float, maxlon: float, maxlat: float,
|
||||
) -> tuple[float, float, int]:
|
||||
|
|
@ -404,12 +433,22 @@ def _headers() -> dict[str, str]:
|
|||
return {"User-Agent": OSINT_USER_AGENT, "Accept": "application/json"}
|
||||
|
||||
|
||||
async def _lock_for(key: str) -> asyncio.Lock:
|
||||
async with _key_locks_guard:
|
||||
lock = _key_locks.get(key)
|
||||
if lock is None:
|
||||
lock = asyncio.Lock()
|
||||
_key_locks[key] = lock
|
||||
return lock
|
||||
|
||||
|
||||
async def _ttl_get(key: str, ttl: float, factory: Callable[[], Awaitable[Any]]) -> Any:
|
||||
now = time.monotonic()
|
||||
hit = _cache.get(key)
|
||||
if hit and now - hit[0] < ttl:
|
||||
return hit[1]
|
||||
async with _cache_lock:
|
||||
lock = await _lock_for(key)
|
||||
async with lock:
|
||||
hit = _cache.get(key)
|
||||
if hit and time.monotonic() - hit[0] < ttl:
|
||||
return hit[1]
|
||||
|
|
@ -428,7 +467,8 @@ async def _get_json(url: str, params: dict | None = None) -> Any:
|
|||
|
||||
async def fetch_aircraft(bbox: str, limit: int = DEFAULT_LIMIT) -> list[dict]:
|
||||
minlon, minlat, maxlon, maxlat = parse_bbox(bbox)
|
||||
lat, lon, radius = bbox_center_radius_nm(minlon, minlat, maxlon, maxlat)
|
||||
qminlon, qminlat, qmaxlon, qmaxlat = quantize_bbox(minlon, minlat, maxlon, maxlat)
|
||||
lat, lon, radius = bbox_center_radius_nm(qminlon, qminlat, qmaxlon, qmaxlat)
|
||||
cache_key = f"adsb:{lat:.2f}:{lon:.2f}:{radius}"
|
||||
|
||||
async def _load():
|
||||
|
|
@ -493,7 +533,7 @@ def _wfigs_params(bbox: str | None) -> dict:
|
|||
"resultRecordCount": 2000,
|
||||
}
|
||||
if bbox:
|
||||
minlon, minlat, maxlon, maxlat = parse_bbox(bbox)
|
||||
minlon, minlat, maxlon, maxlat = quantize_bbox(*parse_bbox(bbox))
|
||||
params["geometry"] = f"{minlon},{minlat},{maxlon},{maxlat}"
|
||||
params["geometryType"] = "esriGeometryEnvelope"
|
||||
params["inSR"] = "4326"
|
||||
|
|
@ -511,7 +551,7 @@ async def fetch_fire_incidents(bbox: str | None, limit: int = DEFAULT_LIMIT) ->
|
|||
async def _load():
|
||||
return transform_wfigs_incidents(await _get_json(WFIGS_INCIDENTS, params))
|
||||
|
||||
rows = await _ttl_get(f"wfigs:inc:{bbox or 'all'}", 600.0, _load)
|
||||
rows = await _ttl_get(f"wfigs:inc:{bbox_cell_key(bbox)}", 600.0, _load)
|
||||
return rows[:limit]
|
||||
|
||||
|
||||
|
|
@ -525,7 +565,7 @@ async def fetch_fire_perimeters(bbox: str | None) -> dict:
|
|||
async def _load():
|
||||
return await _get_json(WFIGS_PERIMETERS, params)
|
||||
|
||||
fc = await _ttl_get(f"wfigs:per:{bbox or 'all'}", 600.0, _load)
|
||||
fc = await _ttl_get(f"wfigs:per:{bbox_cell_key(bbox)}", 600.0, _load)
|
||||
if not isinstance(fc, dict):
|
||||
return {"type": "FeatureCollection", "features": []}
|
||||
return fc
|
||||
|
|
@ -539,7 +579,7 @@ async def fetch_weather_alerts(area: str | None, bbox: str | None) -> dict:
|
|||
if area:
|
||||
nws_params["area"] = area.upper()
|
||||
elif bbox:
|
||||
minlon, minlat, maxlon, maxlat = parse_bbox(bbox)
|
||||
minlon, minlat, maxlon, maxlat = quantize_bbox(*parse_bbox(bbox))
|
||||
nws_params["bbox"] = f"{minlon},{minlat},{maxlon},{maxlat}"
|
||||
nws_fc: dict = {"features": []}
|
||||
sbw_fc: dict = {"features": []}
|
||||
|
|
@ -571,7 +611,7 @@ async def fetch_weather_alerts(area: str | None, bbox: str | None) -> dict:
|
|||
features.append(feat)
|
||||
return {"type": "FeatureCollection", "features": features}
|
||||
|
||||
key = f"alerts:{area or ''}:{bbox or ''}"
|
||||
key = f"alerts:{area or ''}:{bbox_cell_key(bbox) if bbox else ''}"
|
||||
return await _ttl_get(key, 30.0, _load)
|
||||
|
||||
|
||||
|
|
|
|||
|
|
@ -5,6 +5,7 @@ from live_layers import (
|
|||
bbox_center_radius_nm,
|
||||
filter_points_bbox,
|
||||
parse_bbox,
|
||||
quantize_bbox,
|
||||
rainviewer_tile_url,
|
||||
to_marker,
|
||||
transform_adsb_lol,
|
||||
|
|
@ -12,6 +13,8 @@ from live_layers import (
|
|||
transform_amtraker,
|
||||
transform_nhc_storms,
|
||||
transform_wfigs_incidents,
|
||||
_cache,
|
||||
_ttl_get,
|
||||
)
|
||||
|
||||
from camera_scraper import parse_caltrans_json
|
||||
|
|
@ -244,3 +247,43 @@ def test_parse_caltrans_skips_oos_and_maps_jpeg_hls():
|
|||
assert "I-80" in cam["location_name"]
|
||||
assert "rtsp://" not in cam["source_url"].lower()
|
||||
assert "rtsp://" not in cam["snapshot_url"].lower()
|
||||
|
||||
|
||||
def test_quantize_bbox_stable_under_jitter():
|
||||
a = quantize_bbox(*parse_bbox("-78.7912,35.7711,-78.6101,35.9102"))
|
||||
b = quantize_bbox(*parse_bbox("-78.7900,35.7700,-78.6110,35.9090"))
|
||||
assert a == b
|
||||
minlon, minlat, maxlon, maxlat = a
|
||||
assert minlon <= -78.7912
|
||||
assert minlat <= 35.7700
|
||||
assert maxlon >= -78.6101
|
||||
assert maxlat >= 35.9102
|
||||
|
||||
|
||||
def test_ttl_get_does_not_block_other_keys():
|
||||
import asyncio
|
||||
|
||||
_cache.clear()
|
||||
order = []
|
||||
|
||||
async def slow():
|
||||
order.append("slow-start")
|
||||
await asyncio.sleep(0.2)
|
||||
order.append("slow-end")
|
||||
return "S"
|
||||
|
||||
async def fast():
|
||||
order.append("fast")
|
||||
return "F"
|
||||
|
||||
async def run():
|
||||
t1 = asyncio.create_task(_ttl_get("slow", 5, slow))
|
||||
await asyncio.sleep(0.01)
|
||||
t2 = asyncio.create_task(_ttl_get("fast", 5, fast))
|
||||
await asyncio.gather(t1, t2)
|
||||
|
||||
asyncio.run(run())
|
||||
assert order.index("fast") < order.index("slow-end")
|
||||
assert _cache["slow"][1] == "S"
|
||||
assert _cache["fast"][1] == "F"
|
||||
_cache.clear()
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue