fix: 修复缓存导致内存暴增

Author: tianxiu2b2t

assets/js/index.js

@@ -177,10 +177,12 @@ class Channel {
             total: 0,
             current: 0,
         };
+        this.timeout = 10000
         if (!this.support_websocket) return;
         this._ws_init();
         this._ws_initizalized = false;
         this._ws_callbacks = {};
+        this._ws_timeouts = {}
     }
     // websocket
     _ws_init() {
@@ -191,21 +193,23 @@ class Channel {
         this._ws.onmessage = (event) => {
             var data = JSON.parse(event.data);
             if (data.echo_id) {
+                clearTimeout(this._ws_timeouts[data.echo_id])
+                delete this._ws_timeouts[data.echo_id]
                 this._ws_callbacks[data.echo_id].resolve(data.data);
                 delete this._ws_callbacks[data.echo_id];
                 return;
             }
-            console.log(data)
+            window.dispatchEvent(new CustomEvent(`channel_${data.event}`, { detail: data.data }))
         }
         this._ws.onclose = () => {
             this._ws_initizalized = false;
-            this._Ws_reconnect();
+            this._ws_reconnect();
         }
         this._ws.onerror = (event) => {
             console.log("websocket error", event)
         }
     }
-    _Ws_reconnect() {
+    _ws_reconnect() {
         if (this._ws_reconnect_task) return;
         this._ws_reconnect_task = setTimeout(() => {
             this._ws_init();
@@ -223,6 +227,10 @@ class Channel {
                 data,
                 echo_id
             }))
+            this._ws_timeouts[echo_id] = setTimeout(() => {
+                delete this._ws_callbacks[echo_id];
+                reject("timeout")
+            }, this.timeout)
         })
     }
 
@@ -250,6 +258,7 @@ class Channel {
             })
             xhr.addEventListener("readystatechange", (event) => {
                 if (xhr.readyState == 4) {
+                    clearTimeout(timer)
                     if (xhr.statusText == "OK") {
                         resolve(JSON.parse(xhr.responseText)[0].data)
                     } else {
@@ -261,6 +270,10 @@ class Channel {
                 event,
                 data: data
             }))
+            var timer = setTimeout(() => {
+                xhr.abort();
+                reject("timeout")
+            }, this.timeout)
         })
     }
 

core/cache.py

@@ -1,95 +1,63 @@
 import abc
+import collections
 from dataclasses import dataclass
 import inspect
 import time
 from typing import Any, Optional, TypeVar
 
-from . import scheduler
+from apscheduler.schedulers.background import BackgroundScheduler
+from apscheduler.job import Job
 
 
 T = TypeVar("T")
 EMPTY = inspect._empty
 
-@dataclass
-class StorageValue:
-    value: Any
-    expires: Optional[float]
-    timestamp: float
-
-
-class Storage(metaclass=abc.ABCMeta):
-    def __init__(self) -> None:
-        self.cache: dict[str, StorageValue] = {}
-        self._clean_task: int = -1
-    @abc.abstractmethod
-    def set(self, key: str, value: object, expires: Optional[float] = None) -> None:
-        raise NotImplementedError
-    @abc.abstractmethod
-    def get(self, key: str, _def: Any = EMPTY) -> Any:
-        raise NotImplementedError
-    
-    @abc.abstractmethod
-    def delete(self, key: str) -> None:
-        raise NotImplementedError
-    
-    def __setitem__(self, key: str, value: object) -> Any:
-        return self.set(key, value)
-
-    def __getitem__(self, key: str) -> Any:
-        value = self.get(key)
-        if value == EMPTY:
-            raise IndexError
-        return value
-    def exists(self, key: str) -> bool:
-        obj = self.cache.get(key, None)
-        return obj is not None and (obj.expires is not None and obj.expires + obj.timestamp >= time.time() or not obj.expires)
-    def __contains__(self, key: str) -> bool:
-        return self.exists(key)
-    
-    def clean(self):
-        for k, v in filter(lambda x: (x[1].expires is not None and x[1].expires + x[1].timestamp >= time.time()), list(self.cache.items())):
-            self.cache.pop(k)
-            print(k)
-        self._start_clean()
-
+class CacheValue[T]:
+    def __init__(self, value: T, expires: Optional[float] = None) -> None:
+        self.value: T = value
+        self.expires: Optional[float] = expires
+        self.timestamp: float = time.monotonic_ns()
+        self.job: Optional[Job] = None
+
+
+# Time out cache    
+class TimeoutCache[T]:
+    def __init__(self):
+        self.cache: collections.OrderedDict[str, CacheValue] = collections.OrderedDict()
+        self.background = BackgroundScheduler()
+    
+    def set(self, key: str, value: T, timeout: Optional[float] = None) -> None:
+        self._delete_job(key)
+        self.cache[key] = CacheValue(
+            value,
+            timeout
+        )
+        if timeout is not None:
+            self.cache[key].job = self.background.add_job(self.delete, 'interval', seconds=timeout, args=[key])
 
-    def _start_clean(self):
-        scheduler.cancel(self._clean_task)
-        ready = filter(lambda x: (x.expires is not None and x.expires + x.timestamp >= time.time()), list(self.cache.values()))
-        if not ready:
+    def _delete_job(self, key: str):
+        current = self.cache.get(key, None)
+        if current is None or current.job is None:
             return
-        next_time = max(ready, key=lambda x: (x.expires + x.timestamp) - time.time()) # type: ignore
-        self._clean_task = scheduler.run_later(self.clean, delay=(next_time.expires + next_time.timestamp) - time.time()) # type: ignore
+        current.job.remove()
+        current.job = None
 
-    def get_keys(self) -> list[str]:
-        return list(self.cache.keys())
-
-    def get_startswith_all(self, key: str) -> dict[str, Any]:
-        return {k: v for k, v in self.cache.items() if k.startswith(key)}
-    
-    def get_endswith_all(self, key: str) -> dict[str, Any]:
-        return {k: v for k, v in self.cache.items() if k.endswith(key)}
-    
-    def get_contains_all(self, key: str) -> dict[str, Any]:
-        return {k: v for k, v in self.cache.items() if key in k}
-
-class MemoryStorage(Storage):
-    def __init__(self) -> None:
-        super().__init__()
-    
-    def set(self, key: str, value: object, expires: float | None = None) -> None:
-        data = value
-        obj = StorageValue(
-            data,
-            expires,
-            time.time()
-        )
-        self.cache[key] = obj
-    
     def get(self, key: str, _def: Any = EMPTY) -> Any:
-        if not self.exists(key):
+        current = self.cache.get(key, None)
+        if current is None:
+            return _def
+        if current.expires is not None and current.expires + current.timestamp < time.monotonic_ns():
+            self.delete(key)
             return _def
-        return self.cache[key].value or _def
+        return current.value
 
     def delete(self, key: str) -> None:
-        self.cache.pop(key, None)
+        self._delete_job(key)
+        self.cache.pop(key, None)
+
+    def __contains__(self, key: str) -> bool:
+        if key in self.cache:
+            current = self.cache[key]
+            if current.expires is not None and current.expires + current.timestamp < time.monotonic_ns():
+                self.delete(key)
+        return key in self.cache

core/config.py

@@ -170,7 +170,7 @@ def rank_clusters_url(self):
 
 const = Const()
 
-VERSION = "3.3.2"
+VERSION = "3.3.3"
 API_VERSION = "1.13.1"
 USER_AGENT = f"openbmclapi/{API_VERSION} python-openbmclapi/{VERSION}"
 PYTHON_VERSION = ".".join(map(str, (sys.version_info.major, sys.version_info.minor, sys.version_info.micro)))

core/dashboard.py

@@ -566,9 +566,9 @@ async def handle_api(
                 for item in resp:
                     clusters[item["_id"]] = item["name"]
 
-        return [
-            clusters.get(c.id, c.id) for c in cluster.clusters.clusters
-        ]
+        return {
+            c.id: clusters.get(c.id, c.id) for c in cluster.clusters.clusters
+        }
 
     if event == "rank":
         async with aiohttp.ClientSession() as session:
@@ -644,18 +644,17 @@ async def init():
     global task
     task = threading.Thread(target=record)
     task.start()
-    if config.const.auto_sync_assets:
-        scheduler.run_repeat_later(
-            sync_assets,
-            5,
-            interval=3600
-        )
+    scheduler.run_repeat_later(
+        fetch_github_repo,
+        5,
+        interval=3600
+    )
 
 async def unload():
     global running
     running = 0
 
-async def sync_assets():
+async def fetch_github_repo():
     async def get_dir_list(path: str = "/"):
         result = []
         if not path.startswith("/"):
@@ -699,15 +698,8 @@ async def get_file(file: GithubPath):
                 return data
             return io.BytesIO()
 
-    headers = {
-        "User-Agent": config.USER_AGENT
-    }
-    if config.const.github_token:
-        headers["Authorization"] = f"Bearer {config.const.github_token}"
-    async with aiohttp.ClientSession(
-        GITHUB_BASEURL,
-        headers=headers
-    ) as session:
+    
+    async def sync_assets():
         res: list[GithubPath] = await get_dir_list("/assets")
         if not res:
             return
@@ -738,4 +730,32 @@ async def get_file(file: GithubPath):
             path.parent.mkdir(exist_ok=True, parents=True)
             with open(path, "wb") as f:
                 f.write(files[file].getvalue())
-        logger.info(f"Synced {len(files)} files")
+        logger.info(f"Synced {len(files)} files")
+    
+    async def check_update():
+        async with session.get(
+            f"/repos/{GITHUB_REPO}/releases/latest"
+        ) as resp:
+            if resp.status == 200:
+                json_data = await resp.json()
+                tag_name = json_data["tag_name"][1:]
+                if tag_name != config.VERSION:
+                    logger.tinfo("dashboard.info.new_version", current=config.VERSION, latest=tag_name)
+                    return
+
+
+    headers = {
+        "User-Agent": config.USER_AGENT
+    }
+    if config.const.github_token:
+        headers["Authorization"] = f"Bearer {config.const.github_token}"
+    async with aiohttp.ClientSession(
+        GITHUB_BASEURL,
+        headers=headers
+    ) as session:
+        tasks = [
+            check_update()
+        ]
+        if config.const.auto_sync_assets:
+            tasks.append(sync_assets())
+        await asyncio.gather(*tasks)

core/scheduler.py

@@ -8,7 +8,7 @@
     AsyncIOScheduler as AsyncBackground
 )
 from apscheduler.job import Job
-from . import logger
+from .logger import logger
 from weakref import WeakValueDictionary
 
 from . import units

core/storages/__init__.py

@@ -39,7 +39,6 @@ def __init__(self, path: str, weight: int) -> None:
         self.path = path
         self.weight = weight
         self.current_weight = 0
-        self.cache_files = cache.MemoryStorage()
 
     @property
     def unique_id(self) -> str:
@@ -221,7 +220,7 @@ def __init__(self, path: str, weight: int, url: str, username: Optional[str], pa
         self.wait_token.acquire()
         self.tcp_connector = aiohttp.TCPConnector(limit=256)
         self.download_connector = aiohttp.TCPConnector(limit=256)
-        self.cache = cache.MemoryStorage()
+        self.cache = cache.TimeoutCache()
         self.link_cache_timeout = utils.parse_time(link_cache_expires).to_seconds if link_cache_expires is not None else None
         self.link_cache: defaultdict[str, AlistLink] = defaultdict(lambda: AlistLink())
         scheduler.run_repeat_later(self._check_token, 0, 3600)
@@ -312,7 +311,7 @@ async def _action_data(self, action: str, url: str, data: Any, headers: dict[str
                     logger.debug(data)
                     logger.debug(result)
                 else:
-                    self.cache.set(hash, result, 30)
+                    self.cache.set(hash, result, 600)
                 return result
             except:
                 logger.terror("storage.error.alist", status=resp.status, message=await resp.text())
@@ -399,7 +398,7 @@ async def get_files(root_id: int):
                         if result.code != 200:
                             logger.tdebug("storage.debug.error_alist", status=result.code, message=result.message)
                         else:
-                            self.cache.set(f"listfile_{root}", result, 60)
+                            self.cache.set(f"listfile_{root}", result, 600)
                 except:
                     logger.traceback()
                     return []
@@ -506,7 +505,7 @@ def __init__(self, path: str, weight: int, url: str, username: str, password: st
             "User-Agent": config.USER_AGENT
         })
         self.client_lock = asyncio.Lock()
-        self.cache = cache.MemoryStorage()
+        self.cache = cache.TimeoutCache()
         self.session = aiohttp.ClientSession(
             auth=aiohttp.BasicAuth(
                 self.username,
@@ -549,7 +548,7 @@ async def get_files(root_id: int):
                         name=r["name"],
                         size=int(r["size"])
                     ) for r in result if not r["isdir"]]
-                    self.cache.set(f"listfile_{root}", res, 60)
+                    self.cache.set(f"listfile_{root}", res, 600)
                 except webdav3_exceptions.RemoteResourceNotFound:
                     return []
                 except:

core/web.py

@@ -374,7 +374,7 @@ async def check_server():
         for server in privates.values():
             servers.append(CheckServer(server.server.sockets[0].getsockname()[1], start_private_server, server.key))
 
-    logger.tdebug("web.debug.check_server", servers=len(servers))
+    #logger.tdebug("web.debug.check_server", servers=len(servers))
     results = await asyncio.gather(*[asyncio.create_task(_check_server(server)) for server in servers])
     for server, result in zip(servers, results):
         if result:

i18n/zh_cn.json

@@ -55,5 +55,6 @@
     "cluster.error.configuration": "无法获取 [${id}] 下载配置，原因：",
     "cluster.warning.measure_storage": "尝试重定向至存储测速失败，已自动切换至本地测速",
     "database.error.unable.to.decompress": "无法解压数据库，数据：%{data}",
-    "storage.error.check_available": "存储 [${type}] [${path}] [${url}] 检查可用性出错，原因："
+    "storage.error.check_available": "存储 [${type}] [${path}] [${url}] 检查可用性出错，原因：",
+    "dashboard.info.new_version": "当前版本 [${current}] 发现新版本 [${latest}]"
 }