diff --git a/api.py b/api.py index 67ebfe5..3b206cc 100644 --- a/api.py +++ b/api.py @@ -71,8 +71,15 @@ def plan_dates(limit: int = 30): # --------------------------------------------------------------------------- _PLAN_CACHE: dict = {} _PLAN_CACHE_LOCK = threading.Lock() -PLAN_CACHE_SEC = int(os.environ.get("PLAN_CACHE_SEC", "600")) +# 2026-09-10: 默认从 600 秒（10 分钟）改成 12 小时。 +# 起因: 09-10 开盘前 PMS 在 08:40 预热了一次（花 39 秒），缓存 08:50 就过期；用户 09:02 +# 打开页面正好撞上冷启动，页面几路并发一起等，叠加破了 60 秒超时。 +# 为什么可以放这么长: 缓存键里带的是**真实数据日**，上游换日后旧那份立刻失效，不靠时间 +# 保新鲜。12 小时是为了覆盖整个交易日（08:40 预热能管到 20:40），不是为了省算力。 +# 当天要强刷仍然走 nocache 参数。 +PLAN_CACHE_SEC = int(os.environ.get("PLAN_CACHE_SEC", "43200")) _PLAN_CACHE_MAX = 8 # 不同参数组合最多留几份，防内存慢涨 +_PLAN_BUILD_LOCKS: dict = {} # 每个缓存键一把「正在算」的锁，见 _plan_cached 里的说明 def _plan_cached(date, top, obs_top, theme_cap, nocache: bool): @@ -93,14 +100,32 @@ def _plan_cached(date, top, obs_top, theme_cap, nocache: bool): hit = _PLAN_CACHE.get(key) if hit and now - hit[0] < PLAN_CACHE_SEC: return hit[1], True - data = plan.collect(date, top, obs_top, theme_cap) - data.pop("_full", None) - if PLAN_CACHE_SEC > 0: - with _PLAN_CACHE_LOCK: - _PLAN_CACHE[key] = (now, data) - if len(_PLAN_CACHE) > _PLAN_CACHE_MAX: - for k in sorted(_PLAN_CACHE, key=lambda x: _PLAN_CACHE[x][0])[:-_PLAN_CACHE_MAX]: - _PLAN_CACHE.pop(k, None) + + # 同一个键同时只算一次 (2026-09-10)。原来 plan.collect 在锁外面, 缓存冷的时候页面 + # 几路并发进来会**各跑一遍**这趟三十多秒的装配, 还互相抢同一批库连接, 实际耗时远超 + # 单跑一次 —— 这正是 09-10 早上页面破 60 秒超时的那一半原因。 + # 现在让第一个请求去算, 后到的等它算完直接吃缓存。等待用的是每个键自己的锁, + # 不同参数组合互不阻塞。 + with _PLAN_CACHE_LOCK: + lock = _PLAN_BUILD_LOCKS.get(key) + if lock is None: + lock = _PLAN_BUILD_LOCKS[key] = threading.Lock() + with lock: + # 双重检查: 排在后面的请求进到这里时, 前一个多半已经把结果写进缓存了 + if not nocache and PLAN_CACHE_SEC > 0: + with _PLAN_CACHE_LOCK: + hit = _PLAN_CACHE.get(key) + if hit and time.time() - hit[0] < PLAN_CACHE_SEC: + return hit[1], True + data = plan.collect(date, top, obs_top, theme_cap) + data.pop("_full", None) + if PLAN_CACHE_SEC > 0: + with _PLAN_CACHE_LOCK: + _PLAN_CACHE[key] = (time.time(), data) + if len(_PLAN_CACHE) > _PLAN_CACHE_MAX: + for k in sorted(_PLAN_CACHE, key=lambda x: _PLAN_CACHE[x][0])[:-_PLAN_CACHE_MAX]: + _PLAN_CACHE.pop(k, None) + _PLAN_BUILD_LOCKS.pop(k, None) return data, False diff --git a/test_plan_cache.py b/test_plan_cache.py new file mode 100644 index 0000000..66c1856 --- /dev/null +++ b/test_plan_cache.py @@ -0,0 +1,160 @@ +"""选股计划的接口缓存（2026-09-10 故障修复）：数据日换了立刻失效、同键只算一次、强刷绕开缓存。 + +09-10 早上页面报「上游的选股计划读不到 · timeout of 60000ms exceeded」，根因两半： +缓存只活 10 分钟，盘前 08:40 预热那份 08:50 就过期；而装配又在锁外面，09:02 页面几路 +并发进来各跑一遍三十多秒的装配、互相抢库连接，叠加破了 60 秒。这里把两条都钉住。 + +离线，不连库不起服务。跑法：python3 test_plan_cache.py，预期最后一行是 ALL OK。 +""" +import sys +import threading +import time +import types + +for _n in ("pandas", "pymysql", "psycopg"): + if _n not in sys.modules: + try: + __import__(_n) + except Exception: # noqa: BLE001 + _m = types.ModuleType(_n) + if _n == "pandas": + _m.DataFrame = type("DataFrame", (), {}) + sys.modules[_n] = _m + +import api # noqa: E402 +import plan # noqa: E402 + + +def t(name, cond, extra=""): + print((" ok " if cond else " FAIL ") + name + ((" " + str(extra)) if not cond else "")) + assert cond, name + + +class _Spy: + """替身：记录 collect 被调了几次、每次算多久。""" + + def __init__(self, delay=0.0): + self.calls = [] + self.delay = delay + self._lock = threading.Lock() + + def collect(self, date, top, obs_top, theme_cap): + with self._lock: + self.calls.append((date, top, obs_top, theme_cap)) + time.sleep(self.delay) + return {"date": date or "2026-09-09", "main": [], "observe": [], "_full": "该被丢掉"} + + +def _reset(): + api._PLAN_CACHE.clear() + api._PLAN_BUILD_LOCKS.clear() + + +def test_hit_and_miss(): + print("[命中与未命中]") + spy = _Spy() + _reset() + old_collect, old_latest = plan.collect, plan._latest_date + plan.collect = spy.collect + plan._latest_date = lambda tbl: "2026-09-09" + try: + d1, hit1 = api._plan_cached(None, 300, 100, 2, False) + d2, hit2 = api._plan_cached(None, 300, 100, 2, False) + t("第一次未命中、第二次命中", hit1 is False and hit2 is True) + t("只算了一次", len(spy.calls) == 1, spy.calls) + t("内部大字段被丢掉，不进缓存也不出接口", "_full" not in d1 and "_full" not in d2) + + # 参数不同 = 不同的键。PMS 用 300/100，人工联调常用 30/20，两份互不干扰 + api._plan_cached(None, 30, 20, 2, False) + t("参数不同要各算各的", len(spy.calls) == 2, spy.calls) + + # 数据日换了，旧那份必须立刻失效，不能等缓存到期 + plan._latest_date = lambda tbl: "2026-09-10" + _d, hit3 = api._plan_cached(None, 300, 100, 2, False) + t("上游换日后旧缓存立刻失效", hit3 is False and len(spy.calls) == 3) + + # 强刷绕开缓存，但要把新结果写回去 + n = len(spy.calls) + _d, hit4 = api._plan_cached(None, 300, 100, 2, True) + t("强刷绕开缓存", hit4 is False and len(spy.calls) == n + 1) + _d, hit5 = api._plan_cached(None, 300, 100, 2, False) + t("强刷之后缓存是新的", hit5 is True and len(spy.calls) == n + 1) + finally: + plan.collect, plan._latest_date = old_collect, old_latest + + +def test_single_flight(): + print("[同键只算一次 —— 09-10 超时的那一半]") + spy = _Spy(delay=0.4) # 装配很慢，真实环境是三十多秒 + _reset() + old_collect, old_latest = plan.collect, plan._latest_date + plan.collect = spy.collect + plan._latest_date = lambda tbl: "2026-09-09" + try: + results, errs = [], [] + + def one(): + try: + results.append(api._plan_cached(None, 300, 100, 2, False)) + except Exception as e: # noqa: BLE001 + errs.append(e) + + ths = [threading.Thread(target=one) for _ in range(8)] + t0 = time.time() + for x in ths: + x.start() + for x in ths: + x.join() + el = time.time() - t0 + + t("八路并发一个都没出错", not errs, errs) + t("八路都拿到了结果", len(results) == 8, len(results)) + t("装配只跑了一遍（原来是八遍）", len(spy.calls) == 1, spy.calls) + t("总耗时接近单跑一次，不是八次叠加", el < 0.4 * 3, "%.2f 秒" % el) + t("只有一路是未命中，其余都吃了缓存", + sum(1 for _d, h in results if not h) == 1, [h for _d, h in results]) + finally: + plan.collect, plan._latest_date = old_collect, old_latest + + +def test_different_keys_not_blocked(): + print("[不同参数互不阻塞]") + spy = _Spy(delay=0.4) + _reset() + old_collect, old_latest = plan.collect, plan._latest_date + plan.collect = spy.collect + plan._latest_date = lambda tbl: "2026-09-09" + try: + def one(top): + api._plan_cached(None, top, 100, 2, False) + ths = [threading.Thread(target=one, args=(x,)) for x in (300, 30, 50)] + t0 = time.time() + for x in ths: + x.start() + for x in ths: + x.join() + el = time.time() - t0 + t("三种参数各算一次", len(spy.calls) == 3, spy.calls) + t("是并行不是排队", el < 0.4 * 2.5, "%.2f 秒" % el) + finally: + plan.collect, plan._latest_date = old_collect, old_latest + + +def test_ttl_covers_trading_day(): + print("[存活时间要盖住整个交易日]") + t("默认至少 8 小时（08:40 预热要能管到收盘）", + api.PLAN_CACHE_SEC >= 8 * 3600, api.PLAN_CACHE_SEC) + t("锁字典与缓存字典一起清理，不会只涨不落", + "_PLAN_BUILD_LOCKS.pop" in open(api.__file__.replace(".pyc", ".py"), encoding="utf-8").read()) + + +def main(): + test_hit_and_miss() + test_single_flight() + test_different_keys_not_blocked() + test_ttl_covers_trading_day() + print("ALL OK — 选股计划缓存：命中与失效 / 同键只算一次 / 不同参数不互相阻塞 / 存活盖住交易日") + + +if __name__ == "__main__": + main()