fix(provider): sqlite连接改thread-local根治跨日存活跨线程崩——09-01 09:30 live_19实录:引擎首次跨日存活后回调线程变化(连接建于线程19932/用在线程4184),_connect单例连接跨线程复用直接ProgrammingError→momentum_timing牛熊分界不可用跳过当日调仓(安全降级但每日复发直到重启);修=threading.local每线程各持连接(fetchers四处ctx._connect()零改动);影子引擎0例中招;2新测试复现同款报错文案钉死 [vps]
This commit is contained in:
@@ -15,6 +15,7 @@ import logging
|
||||
import os
|
||||
import re
|
||||
import sqlite3
|
||||
import threading
|
||||
from concurrent.futures import ThreadPoolExecutor
|
||||
from datetime import datetime
|
||||
from typing import Any, Dict, List, Optional, Union
|
||||
@@ -161,16 +162,20 @@ class LocalUnifiedProvider(DataProvider): # type: ignore[misc]
|
||||
cfg = config or {}
|
||||
self.db_path: str = cfg.get("db_path", _DEFAULT_DB)
|
||||
self.data_dir: str = cfg.get("data_dir", _DEFAULT_DATA_DIR)
|
||||
self._conn: Optional[sqlite3.Connection] = None
|
||||
# thread-local:引擎跨日存活后回调线程会变(2026-09-01 live_19 事故),
|
||||
# 单例连接跨线程复用直接 ProgrammingError;每线程各持一条连接
|
||||
self._tls = threading.local()
|
||||
self._val_bs_cache: Dict[int, pd.DataFrame] = {} # year -> valuation_baostock
|
||||
self._lpp_helper: Any = None
|
||||
|
||||
def _connect(self) -> sqlite3.Connection:
|
||||
"""惰性连接 dbbardata sqlite(单连接复用)。"""
|
||||
if self._conn is None:
|
||||
self._conn = sqlite3.connect(self.db_path, timeout=30)
|
||||
self._conn.execute("PRAGMA busy_timeout = 30000")
|
||||
return self._conn
|
||||
"""惰性连接 dbbardata sqlite(thread-local 连接复用)。"""
|
||||
conn = getattr(self._tls, "conn", None)
|
||||
if conn is None:
|
||||
conn = sqlite3.connect(self.db_path, timeout=30)
|
||||
conn.execute("PRAGMA busy_timeout = 30000")
|
||||
self._tls.conn = conn
|
||||
return conn
|
||||
|
||||
@staticmethod
|
||||
def _to_date_str(value: Optional[Union[str, datetime]]) -> Optional[str]:
|
||||
|
||||
@@ -12,6 +12,7 @@ Mac 本地 TDD: sqlite tmp_path + tmp parquet fixture,零 VPS 依赖,零网络
|
||||
from __future__ import annotations
|
||||
|
||||
import sqlite3
|
||||
import threading
|
||||
from typing import Any, Dict, List
|
||||
|
||||
import pandas as pd
|
||||
@@ -863,3 +864,45 @@ class TestMixedDatetimeFormat:
|
||||
p = LocalUnifiedProvider({"db_path": str(db), "data_dir": str(tmp_path)})
|
||||
df = p.get_price("600519.XSHG", start_date="2024-09-25", end_date="2024-09-27", fq="raw")
|
||||
assert len(df) == 3 # 不崩 + 返 3 行(混合格式解析 OK)
|
||||
|
||||
|
||||
# ======================== Task N: _connect thread-local ========================
|
||||
class TestConnectThreadLocal:
|
||||
"""2026-09-01 live_19 事故:引擎首次跨日存活后回调线程变化(连接建于线程A、
|
||||
用在线程B),单例 sqlite 连接跨线程复用直接 ProgrammingError → 牛熊分界
|
||||
不可用跳过调仓。_connect 改 thread-local 后每线程自建连接,互不共享。"""
|
||||
|
||||
def test_connect_cross_thread_usable(self, tmp_path):
|
||||
db = _make_security_info_fixture(tmp_path)
|
||||
p = LocalUnifiedProvider({"db_path": str(db), "data_dir": str(tmp_path)})
|
||||
conn_main = p._connect() # 主线程先建连(旧实现会缓存单例)
|
||||
|
||||
errs: List[Any] = []
|
||||
|
||||
def worker():
|
||||
try:
|
||||
c = p._connect()
|
||||
c.execute("SELECT 1").fetchall() # 子线程直接用,不得抛错
|
||||
except Exception as exc: # noqa: BLE001 - 收集给断言看
|
||||
errs.append(exc)
|
||||
|
||||
t = threading.Thread(target=worker)
|
||||
t.start()
|
||||
t.join()
|
||||
assert not errs, f"跨线程取连接报错: {errs}"
|
||||
# 主线程连接仍是原对象(thread-local 各持各的)
|
||||
assert p._connect() is conn_main
|
||||
|
||||
def test_two_threads_get_distinct_connections(self, tmp_path):
|
||||
db = _make_security_info_fixture(tmp_path)
|
||||
p = LocalUnifiedProvider({"db_path": str(db), "data_dir": str(tmp_path)})
|
||||
got: Dict[str, Any] = {}
|
||||
|
||||
def worker():
|
||||
got["conn"] = p._connect()
|
||||
|
||||
t = threading.Thread(target=worker)
|
||||
t.start()
|
||||
t.join()
|
||||
assert got["conn"] is not None
|
||||
assert got["conn"] is not p._connect() # 不是同一个连接对象
|
||||
|
||||
Reference in New Issue
Block a user