From bd88f717fb83ff94f242d7776cc2b60bf515c7c5 Mon Sep 17 00:00:00 2001 From: "2569718930@qq.com" <2569718930@qq.com> Date: Tue, 9 Jun 2026 22:36:23 +0800 Subject: [PATCH] Reduce observation update latency --- docker-compose.yml | 2 +- tests/test_deployment_runtime_config.py | 6 ++- tests/test_observation_collector.py | 60 ++++++++++++++++++++++++- web/observation_collector_service.py | 59 ++++++++++++++++++++++++ 4 files changed, 124 insertions(+), 3 deletions(-) diff --git a/docker-compose.yml b/docker-compose.yml index 50a7cc91..3f04e6dd 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -112,7 +112,7 @@ services: POLYWEATHER_OBSERVATION_COLLECTOR_INITIAL_DELAY_SEC: ${POLYWEATHER_OBSERVATION_COLLECTOR_INITIAL_DELAY_SEC:-5} POLYWEATHER_OBSERVATION_COLLECTOR_MADIS_SEC: ${POLYWEATHER_OBSERVATION_COLLECTOR_MADIS_SEC:-300} POLYWEATHER_OBSERVATION_COLLECTOR_TICK_SEC: ${POLYWEATHER_OBSERVATION_COLLECTOR_TICK_SEC:-30} - POLYWEATHER_SCAN_TERMINAL_PREWARM_ENABLED: 'false' + POLYWEATHER_SCAN_TERMINAL_PREWARM_ENABLED: ${POLYWEATHER_SCAN_TERMINAL_PREWARM_ENABLED:-true} POLYWEATHER_SERVICE_ROLE: web UVICORN_WORKERS: ${UVICORN_WORKERS:-2} healthcheck: diff --git a/tests/test_deployment_runtime_config.py b/tests/test_deployment_runtime_config.py index a8d42b2d..626d4790 100644 --- a/tests/test_deployment_runtime_config.py +++ b/tests/test_deployment_runtime_config.py @@ -91,7 +91,11 @@ def test_docker_compose_isolates_collector_from_web_and_bot_services(): assert "POLYWEATHER_SERVICE_ROLE: bot" in compose assert "POLYWEATHER_SERVICE_ROLE: collector" in collector_block assert "POLYWEATHER_SCAN_TERMINAL_PREWARM_ENABLED: 'false'" in bot_block - assert "POLYWEATHER_SCAN_TERMINAL_PREWARM_ENABLED: 'false'" in web_block + assert ( + "POLYWEATHER_SCAN_TERMINAL_PREWARM_ENABLED: " + "${POLYWEATHER_SCAN_TERMINAL_PREWARM_ENABLED:-true}" + in web_block + ) assert "POLYWEATHER_SCAN_TERMINAL_PREWARM_ENABLED: 'false'" in collector_block assert "POLYWEATHER_OBSERVATION_COLLECTOR_ENABLED: 'false'" in bot_block assert "POLYWEATHER_OBSERVATION_COLLECTOR_ENABLED: 'false'" in web_block diff --git a/tests/test_observation_collector.py b/tests/test_observation_collector.py index d9e9761b..d746f9b2 100644 --- a/tests/test_observation_collector.py +++ b/tests/test_observation_collector.py @@ -1,5 +1,6 @@ -from concurrent.futures import ThreadPoolExecutor +from concurrent.futures import ThreadPoolExecutor, TimeoutError import sqlite3 +import threading import time @@ -119,6 +120,7 @@ def test_observation_collector_run_due_once_refreshes_panel_cache(): ) ], cache_refresher=lambda city: refreshed.append(city), + async_cache_refresh=False, ) assert collector.run_due_once(now_ts=1000.0) == 1 @@ -133,6 +135,62 @@ def test_observation_collector_run_due_once_refreshes_panel_cache(): assert refreshed == ["qingdao", "qingdao"] +def test_observation_collector_cache_refresh_does_not_block_source_polling(): + from web.observation_collector_service import ( + ObservationCollector, + ObservationSourceProfile, + ) + + calls = [] + refresh_started = threading.Event() + release_refresh = threading.Event() + refreshed = [] + + class FakeWeather: + def _uses_fahrenheit(self, city): + return False + + def _attach_china_amsc_awos_data(self, results, city, use_fahrenheit): + calls.append(city) + results["amsc_awos"] = {"source": "amsc_awos", "temp_c": 24.0} + + def slow_cache_refresher(city): + refresh_started.set() + release_refresh.wait(timeout=2) + refreshed.append(city) + + collector = ObservationCollector( + weather=FakeWeather(), + profiles=[ + ObservationSourceProfile( + source="amsc_awos", + cities=("qingdao", "beijing"), + interval_sec=180, + ) + ], + cache_refresher=slow_cache_refresher, + ) + + with ThreadPoolExecutor(max_workers=1) as executor: + future = executor.submit(collector.run_due_once, now_ts=1000.0) + assert refresh_started.wait(timeout=1) + try: + try: + result = future.result(timeout=0.1) + except TimeoutError: + result = None + finally: + release_refresh.set() + + assert result == 2 + assert calls == ["qingdao", "beijing"] + deadline = time.time() + 2 + while set(refreshed) != {"qingdao", "beijing"} and time.time() < deadline: + time.sleep(0.01) + assert set(refreshed) == {"qingdao", "beijing"} + collector.close() + + def test_observation_collector_records_source_status_to_runtime_state(tmp_path): from src.database.runtime_state import ObservationCollectorStatusRepository, RuntimeStateDB from web.observation_collector_service import ( diff --git a/web/observation_collector_service.py b/web/observation_collector_service.py index dc0ffa6b..49e5a95f 100644 --- a/web/observation_collector_service.py +++ b/web/observation_collector_service.py @@ -5,6 +5,7 @@ from __future__ import annotations import os import threading import time +from concurrent.futures import ThreadPoolExecutor from dataclasses import dataclass from typing import Any, Callable, Iterable, List, Optional, Sequence, Tuple @@ -53,6 +54,8 @@ class ObservationCollector: profiles: Sequence[ObservationSourceProfile], cache_refresher: Optional[Callable[[str], Any]] = None, status_recorder: Optional[ObservationCollectorStatusRepository] = None, + async_cache_refresh: Optional[bool] = None, + cache_refresh_workers: Optional[int] = None, ) -> None: self.weather = weather self.profiles = list(profiles) @@ -60,6 +63,29 @@ class ObservationCollector: self.status_recorder = status_recorder self._last_run_ts: dict[tuple[str, str], float] = {} self._lock = threading.Lock() + self._cache_refresh_lock = threading.Lock() + self._cache_refresh_inflight: set[str] = set() + self._cache_refresh_async = ( + _env_bool("POLYWEATHER_OBSERVATION_COLLECTOR_CACHE_REFRESH_ASYNC", True) + if async_cache_refresh is None + else bool(async_cache_refresh) + ) + worker_count = max( + 1, + min( + 4, + int( + cache_refresh_workers + if cache_refresh_workers is not None + else _env_int("POLYWEATHER_OBSERVATION_COLLECTOR_CACHE_REFRESH_WORKERS", 1) + ), + ), + ) + self._cache_refresh_executor: Optional[ThreadPoolExecutor] = ( + ThreadPoolExecutor(max_workers=worker_count) + if callable(cache_refresher) and self._cache_refresh_async + else None + ) def run_due_once(self, *, now_ts: Optional[float] = None) -> int: now = float(time.time() if now_ts is None else now_ts) @@ -166,11 +192,44 @@ class ObservationCollector: def _refresh_city_cache(self, city: str) -> None: if not callable(self.cache_refresher): return + normalized_city = str(city or "").strip().lower() + if not normalized_city: + return + if self._cache_refresh_executor is None: + self._refresh_city_cache_inline(normalized_city) + return + with self._cache_refresh_lock: + if normalized_city in self._cache_refresh_inflight: + return + self._cache_refresh_inflight.add(normalized_city) + try: + self._cache_refresh_executor.submit(self._refresh_city_cache_task, normalized_city) + except Exception as exc: + with self._cache_refresh_lock: + self._cache_refresh_inflight.discard(normalized_city) + logger.warning( + "observation collector cache refresh queue failed city={}: {}", + normalized_city, + exc, + ) + + def _refresh_city_cache_task(self, city: str) -> None: + try: + self._refresh_city_cache_inline(city) + finally: + with self._cache_refresh_lock: + self._cache_refresh_inflight.discard(city) + + def _refresh_city_cache_inline(self, city: str) -> None: try: self.cache_refresher(city) except Exception as exc: logger.warning("observation collector cache refresh failed city={}: {}", city, exc) + def close(self) -> None: + if self._cache_refresh_executor is not None: + self._cache_refresh_executor.shutdown(wait=False) + def build_observation_source_profiles() -> List[ObservationSourceProfile]: us_madis_cities = [