From 63a95361162569327e2f233ee3c468f0a9c8ba0c Mon Sep 17 00:00:00 2001 From: chakcy <947105045@qq.com> Date: Fri, 28 Aug 2026 16:20:01 +0800 Subject: [PATCH] =?UTF-8?q?modify:=20worker=20=E7=AE=A1=E7=90=86=E5=A4=9A?= =?UTF-8?q?=E8=BF=9B=E7=A8=8B=E5=89=AF=E6=9C=AC?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- pyproject.toml | 12 +- src/chongming_worker/process.py | 68 ++++- src/chongming_worker/runtime.py | 515 +++++++++++++++++++++++++++----- 3 files changed, 517 insertions(+), 78 deletions(-) diff --git a/pyproject.toml b/pyproject.toml index fdb3b1c..e0e389c 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,16 +4,16 @@ requires = ["hatchling"] [project] name = "chongming-worker" -version = "0.4.0" +version = "0.4.1" description = "chongming worker" readme = "README.md" requires-python = ">=3.12" dependencies = [ - "eclipse-zenoh>=1.9.0", - "pydantic>=2.0", - "tomli>=2.4.1", - "tomli-w>=1.0.0", - "minio>=7.2.0", + "eclipse-zenoh", + "pydantic", + "tomli", + "minio", + "chongming-cache", ] [project.scripts] diff --git a/src/chongming_worker/process.py b/src/chongming_worker/process.py index 3e086cd..25adb89 100644 --- a/src/chongming_worker/process.py +++ b/src/chongming_worker/process.py @@ -43,7 +43,7 @@ def load_plugin_config(worker_type: str, config_dir: str = "config/workers") -> Returns: {"type", "module", "factory", "cleanup", "count", "load_mode", - "mqtt_topics", "config": 展平后的运行时配置} + "mqtt_topics", "cache"([workers.cache] 段), "config": 展平后的运行时配置} """ import pathlib @@ -67,7 +67,7 @@ def load_plugin_config(worker_type: str, config_dir: str = "config/workers") -> mqtt_topics.append(t) workers_flat = {k: v for k, v in workers_section.items() - if k not in ("config", "mqtt_topics", "load_mode")} + if k not in ("config", "mqtt_topics", "load_mode", "cache")} runtime_config = { **plugin_conf, **workers_flat, @@ -82,10 +82,24 @@ def load_plugin_config(worker_type: str, config_dir: str = "config/workers") -> "count": workers_section.get("count", 1), "load_mode": load_mode, "mqtt_topics": mqtt_topics, + "cache": dict(workers_section.get("cache") or {}), "config": runtime_config, } +def build_worker_cache(cache_cfg: dict): + """按插件 [workers.cache] 段建立配置缓存（redis/diskcache），返回 (cache, mode)。 + + 从配置中剔除内部控制键 launch_worker_count（决定配置优先来源 toml/cache）， + 避免其被当作后端连接参数传给 Cache.from_config。 + """ + from chongming_cache import Cache + + mode = str((cache_cfg or {}).get("launch_worker_count", "toml")).lower() + clean = {k: v for k, v in (cache_cfg or {}).items() if k != "launch_worker_count"} + return Cache.from_config(clean), mode + + def worker_serve(): """独立 Worker 进程入口（console script: chongming-worker / worker_serve）。""" # 独立进程不依赖网关的日志配置，兜底 basicConfig（已有 handler 时不覆盖） @@ -99,6 +113,12 @@ def worker_serve(): parser.add_argument("--type", required=True, help="插件类型（如 demo / example）") parser.add_argument("--count", type=int, default=None, help="副本数（覆盖配置中的 count）") parser.add_argument("--config", default="config.toml", help="全局配置文件路径（读取 [zenoh] 节）") + parser.add_argument( + "--instance", default=None, + help="外部进程实例标识（多实例全局分摊/逐实例精细扩缩用）。缺省用 " + "CHONGMING_WORKER_INSTANCE 或 {hostname}:{pid}（同机多实例默认即自动区分）；" + "跨重启稳定适逐实例缩放请显式指定", + ) args = parser.parse_args() # console script 启动时 sys.path 不含当前目录，显式加入以保证 @@ -136,6 +156,36 @@ def worker_serve(): session = zenoh.open(zconfig) logger.info("[%s] Zenoh 会话已建立", args.type) + # 1.5 配置缓存（[workers.cache] 段，进程插件自行管理）+ 配置优先来源登记 + # [workers.config] 的单一事实来源解析由 worker 运行时（create_workers）统一完成。 + # 多实例场景下还登记「外部进程实例参与方」，使扩缩容按存活实例数分摊全局 count。 + from chongming_worker.runtime import ( + register_instance, + run_instance_heartbeat, + set_current_instance_id, + set_worker_cache, + set_worker_mode, + unregister_instance, + ) + set_current_instance_id(args.instance) + + cache = None + mode = "toml" + hb_task = None # 实例参与方心跳任务（外部多实例全局分摊用） + if wc.get("cache"): + cache, mode = build_worker_cache(wc["cache"]) + set_worker_cache(args.type, cache) + set_worker_mode(args.type, mode) + register_instance(args.type, cache, args.instance) + hb_task = asyncio.create_task( + run_instance_heartbeat(args.type, cache, instance_id=args.instance) + ) + logger.info( + "[%s] 已启用配置缓存（backend=%s, mode=%s）并登记外部实例参与方", + args.type, getattr(getattr(cache, "backend", None), "name", "?"), mode, + ) + worker_config = wc["config"] + try: # 2. import 插件（触发 @rpc_method 注册），然后契约校验 module = __import__(wc["module"], fromlist=[wc["factory"]]) @@ -145,9 +195,9 @@ def worker_serve(): factory = getattr(module, wc["factory"]) count = args.count or wc.get("count", 1) if inspect.iscoroutinefunction(factory): - tasks = await factory(session, wc["config"], count, worker_manager=None) + tasks = await factory(session, worker_config, count, worker_manager=None) else: - tasks = factory(session, wc["config"], count, worker_manager=None) + tasks = factory(session, worker_config, count, worker_manager=None) # 4. 发布路由描述符（网关 RouteRegistry 订阅后动态挂载 HTTP 路由） publisher = RoutePublisher(session, args.type) @@ -173,6 +223,9 @@ def worker_serve(): await stop_event.wait() announce_task.cancel() await asyncio.gather(announce_task, return_exceptions=True) + if hb_task is not None: + hb_task.cancel() + await asyncio.gather(hb_task, return_exceptions=True) # 6. 优雅退出：摘除路由 → cleanup → 取消副本任务 → 关闭会话 publisher.unpublish() @@ -187,9 +240,16 @@ def worker_serve(): t.cancel() if tasks: await asyncio.gather(*tasks, return_exceptions=True) + if cache is not None: + unregister_instance(args.type, cache, args.instance) logger.info("[%s] Worker 进程已退出", args.type) finally: session.close() + if cache is not None: + try: + cache.close() + except Exception: # noqa: BLE001 + pass try: asyncio.run(_run()) diff --git a/src/chongming_worker/runtime.py b/src/chongming_worker/runtime.py index 0d1dbda..4d64ecf 100644 --- a/src/chongming_worker/runtime.py +++ b/src/chongming_worker/runtime.py @@ -9,13 +9,18 @@ import inspect import logging import pathlib import random +import socket +import time import tomllib import orjson -import tomli_w from typing import Any, Awaitable, Callable, Dict, List, Optional import zenoh +# 配置缓存（chongming-cache）用于 [workers.config] 与 count 的单一事实来源读写； +# 由 process.py build_worker_cache（进程插件）或网关 manager.cache（inline）注入， +# 本模块不自建，仅按注入的 cache 句柄读写缓存键。 + from chongming_worker.decorators import ( RPCMethod, get_rpc_methods, @@ -42,6 +47,7 @@ ADMIN_CMD_RESTART = "restart" ADMIN_CMD_STOP = "stop" ADMIN_CMD_DRAIN = "drain" ADMIN_CMD_CONFIG_SET = "config_set" +ADMIN_CMD_GET_CONFIG = "get_config" ADMIN_CMD_RESCALE = "rescale" # 优雅排空时，等待在途请求完成的最大时长（秒），超时则强制关闭该副本 @@ -77,65 +83,365 @@ def _plugin_config_path(worker_type: str) -> pathlib.Path: return pathlib.Path("config/workers") / worker_type / "config.toml" -def _write_plugin_config_section(worker_type: str, updates: Dict[str, Any]) -> Dict[str, Any]: - """把 updates 合并写入插件 config.toml 的 [workers.config] 段（原子写回）。 +# ── 单事实来源缓存（chongming-cache）────────────────────────────── +# 运行时可写配置（[workers.config]）与副本目标数（count）以缓存为准，不再回写 config.toml： +# - key = "worker:cm:configs:{type}"：全局一份 [workers.config] dict +# - key = "worker:cm:counts:{type}"：全局目标副本数（已有，由网关 scale 统一写） +# TOML 仅在启动时作为种子（默认模式用 toml 修正缓存；cache 模式缓存缺失时用 toml 引导）， +# 之后读写都在缓存侧进行。缓存可选：网关 inline 用 manager.cache，进程插件用自身 [workers.cache]。 - 仅改动 [workers.config] 下的键，保留文件其它段与段落。 - 返回写入后的 [workers.config] 完整 dict（供回显）。 - Raises: - ValueError: 包含保留键 / config.toml 缺失 / 无 [workers] 段 +def _configs_cache_key(worker_type: str) -> str: + return f"worker:cm:configs:{worker_type}" + + +# ── 全局副本数（chongming-cache 单一事实来源，与网关共享同一套键）── +# 「count」= 全局目标总量；worker（inline/external 统一）从缓存读在线网关列表，把总量 +# 分摊成本网关应承担的副本数（external 无网关身份则全额承担），并写缓存。 +_SCALE_COUNTS_KEY = "worker:cm:counts:{type}" # 全局目标总量（int） +_SCALE_GATEWAYS_KEY = "worker:cm:gateways" # 在线网关列表：[[id, ts], ...]（JSON） +_SCALE_QUOTA_KEY = "worker:cm:quota:{type}:{gw}" # 各网关分摊配额（int） +_GATEWAY_LEASE = 60 + +# ── 外部进程实例参与方（多实例全局协同）──────────────────── +# 同一类型的外部进程可能并行运行多个（多台机器/多个容器）。为避免"每个进程都全额起 +# 副本"导致副本翻倍，每个外部进程用稳定的 instance_id 登记到共享缓存，扩缩容时按存活 +# 实例数分摊全局 count（与 inline 按在线网关分摊同构）。租约/心跳复用网关参数。 +_SCALE_INSTANCES_KEY = "worker:cm:instances:{type}" # 参与方注册：[[inst_id, ts], ...]（JSON） +_INSTANCE_LEASE = 60 +# 逐实例绝对副本数覆盖表：{instance_id: count}。有记录的实例按绝对值扩缩（精细定制）， +# 无记录的回退到「全局 count 均分」；全局 /scale 会清除此表回到对称分摊。 +_SCALE_INSTCOUNTS_KEY = "worker:cm:instcounts:{type}" + + +def _read_online_gateways(cache) -> List[str]: + """从缓存读在线网关 id（按心跳租约清理死网关），排序返回。无缓存返回 []。""" + if cache is None: + return [] + now = time.time() + deadline = now - _GATEWAY_LEASE + alive = [] + for entry in cache.get(_SCALE_GATEWAYS_KEY) or []: + if not isinstance(entry, (list, tuple)) or len(entry) != 2: + continue + gw_id, ts = entry + if isinstance(gw_id, str) and isinstance(ts, (int, float)) and ts >= deadline: + alive.append(gw_id) + return sorted(set(alive)) + + +# ── 外部进程实例参与方助手 ───────────────────────────────── +def _read_instances_raw(cache, worker_type: str) -> List[List]: + """读同类型外部实例原始列表 [[inst_id, ts], ...]；无缓存返回 []。""" + if cache is None: + return [] + return list(cache.get(_SCALE_INSTANCES_KEY.format(type=worker_type)) or []) + + +def _prune_instances_raw(cache, worker_type: str, raw: List[List]) -> List[List]: + """剔除超过租约未心跳的死实例，写回清理结果以保持列表准确。""" + deadline = time.time() - _INSTANCE_LEASE + alive = [] + for entry in raw: + if not isinstance(entry, (list, tuple)) or len(entry) != 2: + continue + iid, ts = entry + if isinstance(iid, str) and isinstance(ts, (int, float)) and ts >= deadline: + alive.append([iid, ts]) + if len(alive) != len(raw): + cache[_SCALE_INSTANCES_KEY.format(type=worker_type)] = alive + return alive + + +def _read_instances(cache, worker_type: str) -> List[str]: + """读同类型存活外部实例 id（按租约清理死实例、保序去重），排序返回。""" + raw = _prune_instances_raw(cache, worker_type, _read_instances_raw(cache, worker_type)) + return sorted({iid for iid, _ in raw}) + + +# 进程内显式设定的实例身份（process.py 启动时注入）；None 则回退 env/hostname。 +# 与 register_instance/逐一绝对扩缩共用同一 id，保证命中判断一致。 +_current_instance_id: Optional[str] = None + + +def set_current_instance_id(iid: Optional[str]) -> None: + """设定当前外部 worker 进程的实例身份（应与 register_instance 使用的 id 一致）。""" + global _current_instance_id + _current_instance_id = iid + + +def current_instance_id() -> str: + """外部 worker 进程的稳定全局实例标识。 + + 优先取进程内显式设定的身份（process.py --instance）；其次 env + CHONGMING_WORKER_INSTANCE（编排时显式给定、跨重启稳定）；缺省回退 + "{hostname}:{pid}"——同一台机器上并行多个未指定身份的实例时，凭 pid 区分， + 避免它们被误并为一个参与方、导致副本翻倍或 UI 只显示一个实例。 + 注意：缺省 id 随进程重启变化；若需跨重启稳定且被逐实例精细扩缩持续命中， + 请显式设 --instance / CHONGMING_WORKER_INSTANCE。 """ - bad = [k for k in updates if k in CONFIG_FORBIDDEN_KEYS] + if _current_instance_id: + return _current_instance_id + iid = os.environ.get("CHONGMING_WORKER_INSTANCE") + if not iid: + iid = f"{socket.gethostname()}:{os.getpid()}" + return iid + + +def _write_instance_heartbeats(cache, worker_type: str, iid: str, *, keep_self: bool) -> None: + """按 keep_self 保留/移除本实例，并刷新本实例心跳 ts，写回参与方列表。""" + if cache is None: + return + now = time.time() + raw = _prune_instances_raw(cache, worker_type, _read_instances_raw(cache, worker_type)) + alive = [[i, ts] for i, ts in raw if i != iid] + if keep_self: + alive.append([iid, now]) + cache[_SCALE_INSTANCES_KEY.format(type=worker_type)] = alive + + +def register_instance(worker_type: str, cache, instance_id: Optional[str] = None) -> None: + """外部进程启动时登记自身为参与方（带心跳时间戳）。进程退出后租约过期自动清除。""" + _write_instance_heartbeats( + cache, worker_type, instance_id or current_instance_id(), keep_self=True + ) + + +def heartbeat_instance(worker_type: str, cache, instance_id: Optional[str] = None) -> None: + """周期性刷新本实例心跳 ts，并剔除超租约死实例。""" + _write_instance_heartbeats( + cache, worker_type, instance_id or current_instance_id(), keep_self=True + ) + + +def unregister_instance(worker_type: str, cache, instance_id: Optional[str] = None) -> None: + """关闭时从参与方列表移除本实例。""" + _write_instance_heartbeats( + cache, worker_type, instance_id or current_instance_id(), keep_self=False + ) + + +# ── 逐实例绝对副本数（精细扩缩）──────────────────────────── +def _read_instance_counts(cache, worker_type: str) -> Dict[str, int]: + """读逐实例绝对副本数覆盖表 {instance_id: count}；无缓存/非对象返回 {}。""" + if cache is None: + return {} + v = cache.get(_SCALE_INSTCOUNTS_KEY.format(type=worker_type)) + return dict(v) if isinstance(v, dict) else {} + + +def _write_own_instance_count( + cache, worker_type: str, count: int, instance_id: Optional[str] = None +) -> None: + """把本实例的绝对副本数写入覆盖表（精细扩缩；进程重启后依旧生效）。""" + if cache is None: + return + iid = instance_id or current_instance_id() + key = _SCALE_INSTCOUNTS_KEY.format(type=worker_type) + counts = dict(cache.get(key) or {}) + counts[iid] = int(count) + cache[key] = counts + + +def _clear_instance_counts(cache, worker_type: str) -> None: + """清除所有逐实例绝对覆盖，回到「全局 count 均分」对称语义。""" + if cache is None: + return + cache[_SCALE_INSTCOUNTS_KEY.format(type=worker_type)] = {} + + +def _own_instance_count( + cache, worker_type: str, instance_id: Optional[str] = None +) -> Optional[int]: + """读本实例在覆盖表中的绝对副本数；未定制返回 None（回退全局均分）。""" + if cache is None: + return None + counts = _read_instance_counts(cache, worker_type) + iid = instance_id or current_instance_id() + v = counts.get(iid) + return int(v) if isinstance(v, (int, float)) else None + + +async def run_instance_heartbeat( + worker_type: str, cache, *, + interval: float = 15.0, instance_id: Optional[str] = None, +) -> None: + """外部进程心跳常驻循环。启动后 create_task；无需显式反注册，租约过期自清。""" + if cache is None: + return + while True: + try: + heartbeat_instance(worker_type, cache, instance_id) + except Exception as e: # noqa: BLE001 + logger.warning("[instance] %s 心跳失败: %s", worker_type, e) + await asyncio.sleep(interval) + + +def distribute_quota(cache, worker_type: str, total: int, gw_id: Optional[str]) -> int: + """在「count=全局总量」语义下，算当前主体应承担的副本数。 + + gw_id 非空（inline，具备网关身份）→ 按在线网关 id 排序均分，余量按序补前若干。 + gw_id 为空（external 独立进程）→ 按同一类型「存活实例 N」分摊（余量补靠前实例）， + 避免每个外部进程都全额起副本导致翻倍；无参与方时兜底全额。 + 结果写回 cache（observability，与网关共享键）。 + """ + total = max(0, int(total)) + if gw_id: + gws = _read_online_gateways(cache) + if not gws: + quota = total + else: + base, rem = divmod(total, len(gws)) + idx = gws.index(gw_id) if gw_id in gws else 0 + quota = base + (1 if idx < rem else 0) + quota_key = _SCALE_QUOTA_KEY.format(type=worker_type, gw=gw_id) + else: + iid = current_instance_id() + inst = _read_instances(cache, worker_type) + if not inst: + quota = total + else: + base, rem = divmod(total, len(inst)) + idx = inst.index(iid) if iid in inst else 0 + quota = base + (1 if idx < rem else 0) + quota_key = _SCALE_QUOTA_KEY.format(type=worker_type, gw=iid) + if cache is not None: + cache[quota_key] = quota + return quota + + +def resolve_worker_count( + worker_type: str, default: int, cache, mode: str, gw_id: Optional[str] = None +) -> int: + """启动时解析本进程应承载的副本数（count 收归 worker 后由 WorkerPool 调用）。 + + 计算全局总量：toml（默认）用 default 覆盖写回缓存；cache 模式缓存有则用、缺则 default 引导写回。 + 再按在线网关分摊为写给本进程的副本数（external/no gw 按存活实例分摊； + 外部实例若在 instcounts 有绝对覆盖，直接采用该值）。 + """ + if cache is None: + return int(default) + # 逐实例绝对覆盖（精细扩缩）：外部进程若被单独定制过，重启后应恢复到该绝对数， + # 而非被全局均分覆盖。inline（有网关身份）不适用。 + if gw_id is None: + own = _own_instance_count(cache, worker_type) + if own is not None: + return own + total = max(0, int(default)) + if mode == "cache": + v = cache.get(_SCALE_COUNTS_KEY.format(type=worker_type)) + if isinstance(v, int) and v >= 0: + total = v + cache[_SCALE_COUNTS_KEY.format(type=worker_type)] = total + return distribute_quota(cache, worker_type, total, gw_id) + + +# 当前进程内建立的缓存（key=worker_type）。进程插件在 process.py 启动时注入， +# 供 admin（config_set/get_config/rescale）读写缓存，无需改 create_workers 签名。 +_worker_caches: Dict[str, Any] = {} + +# 各 worker 类型使用的「配置优先来源」模式（toml 默认 / cache）。 +# 进程插件在 build_worker_cache 时登记；网关 inline 走 manager 的 _launch_worker_count。 +_worker_modes: Dict[str, str] = {} + + +def set_worker_cache(worker_type: str, cache) -> None: + """为某 worker 类型登记其配置缓存句柄（进程插件自建缓存后调用）。""" + _worker_caches[worker_type] = cache + + +def set_worker_mode(worker_type: str, mode: str) -> None: + """登记某 worker 类型的配置优先来源模式（"toml" 默认 / "cache"）。""" + _worker_modes[worker_type] = ("cache" if str(mode).lower() == "cache" else "toml") + + +def _resolve_cache(worker_type: str, worker_manager=None): + """取某 worker 类型的配置缓存：显式登记 > 网关 manager.cache > 无。""" + cache = _worker_caches.get(worker_type) + if cache is not None: + return cache + if worker_manager is not None: + return getattr(worker_manager, "cache", None) + return None + + +def _resolve_mode(worker_type: str, worker_manager=None) -> str: + """取某 worker 类型的配置优先来源模式：网关 inline 用 manager > 进程插件登记 > toml。""" + if worker_manager is not None: + return "cache" if str(getattr(worker_manager, "_launch_worker_count", "toml")).lower() == "cache" else "toml" + return _worker_modes.get(worker_type, "toml") + + +def _read_worker_config_section(cache, worker_type: str) -> Dict[str, Any]: + """读该类型 [workers.config]（缓存优先，缓存缺失回退读 toml 种子）。""" + if cache is not None: + val = cache.get(_configs_cache_key(worker_type)) + if isinstance(val, dict): + return dict(val) + try: + with open(_plugin_config_path(worker_type), "rb") as f: + data = tomllib.load(f) + workers = data.get("workers") or {} + cfg = workers.get("config") if isinstance(workers, dict) else None + return dict(cfg) if isinstance(cfg, dict) else {} + except Exception: # noqa: BLE001 - toml 缺失/损坏按空处理 + return {} + + +def _persist_worker_config_section(cache, worker_type: str, updates: Dict[str, Any]) -> Dict[str, Any]: + """把 updates 合并写入该类型 [workers.config]（缓存侧，全局唯一）。 + + 仅改动 [workers.config] 下的键，不再写回 config.toml（取消回写机制）。 + 缓存缺失（未配置缓存后端）时不持久化，仅返回合并后的配置供热更/重启使用。 + """ + bad = sorted(k for k in updates if k in CONFIG_FORBIDDEN_KEYS) if bad: raise ValueError(f"禁止修改保留配置键: {bad}（路由/加载控制配置除外，不在此功能范围）") - path = _plugin_config_path(worker_type) - with open(path, "rb") as f: - data = tomllib.load(f) - - workers = data.get("workers") - if not isinstance(workers, dict): - raise ValueError(f"{path} 缺少 [workers] 节") - - cfg = workers.setdefault("config", {}) - if not isinstance(cfg, dict): - raise ValueError(f"{path} 的 [workers.config] 不是表，无法更新") - cfg.update(updates) - - # 原子写回：临时文件 + rename，避免写一半被读 - tmp = path.with_suffix(".toml.tmp") - with open(tmp, "w", encoding="utf-8") as f: - f.write(tomli_w.dumps(data)) - os.replace(tmp, path) - return dict(cfg) + current = _read_worker_config_section(cache, worker_type) + merged = {**current, **updates} + if cache is not None: + cache[_configs_cache_key(worker_type)] = merged + return dict(merged) -def _write_plugin_count(worker_type: str, count: int) -> int: - """把某类型的副本目标数写入插件 config.toml 的 [workers].count（原子写回）。 +def resolve_runtime_config( + worker_type: str, + config: Dict[str, Any], + cache, + mode: str = "toml", +) -> Dict[str, Any]: + """启动时按模式对齐 [workers.config] 的单一事实来源，返回覆盖后的运行时配置。 - 由 rescale 指令的处理副本在**其所在进程的工作目录**下执行，因此 inline（网关进程） - 与 external（独立进程/容器、配置目录已挂载）都能持久化到自己那侧的配置，重启后保持。 - 返回写回的计数值。 - - Raises: - ValueError: 文件缺失 / 无 [workers] 节 + 缓存配置（[workers.config]）在传入插件前被展平进运行配置顶层（与 _discover_plugins 一致）。 + - "toml"（默认）：用 toml 的 [workers.config] 覆盖并修正缓存（缓存缺失时写入）； + - "cache"：缓存有值则用缓存覆盖；缓存缺失才用 toml 引导并同步到缓存。 + cache 为 None 时保持传入的 config 原样（不启用缓存语义）。 """ - path = _plugin_config_path(worker_type) - with open(path, "rb") as f: - data = tomllib.load(f) + try: + with open(_plugin_config_path(worker_type), "rb") as f: + data = tomllib.load(f) + workers = data.get("workers") or {} + toml_cfg = workers.get("config") if isinstance(workers, dict) else None + toml_cfg = dict(toml_cfg) if isinstance(toml_cfg, dict) else {} + except Exception: # noqa: BLE001 - toml 缺失/损坏 + toml_cfg = {} - workers = data.get("workers") - if not isinstance(workers, dict): - raise ValueError(f"{path} 缺少 [workers] 节") + if cache is None: + # 无缓存：保持 toml 种子（经典行为，无回写） + return {**config, **toml_cfg} - workers["count"] = int(count) - - tmp = path.with_suffix(".toml.tmp") - with open(tmp, "w", encoding="utf-8") as f: - f.write(tomli_w.dumps(data)) - os.replace(tmp, path) - return int(count) + key = _configs_cache_key(worker_type) + if mode == "cache": + cached = cache.get(key) + if not isinstance(cached, dict): + cached = toml_cfg + cache[key] = cached + return {**config, **cached} + # toml（默认）：toml 覆盖并修正缓存 + toml_cfg = dict(toml_cfg) + cache[key] = toml_cfg + return {**config, **toml_cfg} def _merge_runtime_config(config: Dict[str, Any], updates: Dict[str, Any]) -> Dict[str, Any]: @@ -161,6 +467,8 @@ class WorkerController: self.epoch = 0 self.exit_request: str = "" self.current_worker_id: str = "" + # 该副本所属类型的配置缓存句柄（由 WorkerPool 注入，供 admin 读写缓存） + self.cache = None # 配置变更后待应用的新运行时配置（由 config_set 设置，重启时生效） self.pending_config: Optional[Dict[str, Any]] = None self.pending_updates: Optional[Dict[str, Any]] = None @@ -224,22 +532,44 @@ async def _handle_admin_command( controller.exit_request = ADMIN_CMD_STOP controller.draining = True logger.info("[Worker] %s 收到优雅排空指令（将隔离新请求并等待在途请求完成）", worker_id) + elif cmd == ADMIN_CMD_GET_CONFIG: + # 查询该类型的 [workers.config]（缓存优先，无缓存回退读 toml 种子）。 + cache = getattr(controller, "cache", None) + cfg = _read_worker_config_section(cache, worker_type) + query.reply( + key_expr=query.key_expr, + payload=orjson.dumps({ + "status": "ok", + "cmd": cmd, + "worker_type": worker_type, + "worker_id": worker_id, + "config": cfg, + "source": "cache" if cache is not None else "toml", + }), + ) + return elif cmd == ADMIN_CMD_CONFIG_SET: updates = raw.get("config") if not isinstance(updates, dict) or not updates: setup_error = "config_set 需要 data.config 字段（非空对象）" else: try: - # 仅 index0 落盘写回，其余副本不落盘（同一进程多副本/共享文件避免并发写） + # 写回改为缓存侧（不再回写 config.toml）。缓存为全局唯一事实来源， + # 仅 index0 执行写+热重载，避免多副本并发 read-modify-write；其余副本只应用。 + cache = getattr(controller, "cache", None) if controller.index == 0: - new_section = _write_plugin_config_section(worker_type, updates) + new_section = _persist_worker_config_section(cache, worker_type, updates) applied_updates = { k: v for k, v in updates.items() if k not in CONFIG_FORBIDDEN_KEYS } + persist_note = ( + f"{len(new_section)} 键已写缓存（{_configs_cache_key(worker_type)}）" + if cache is not None else "缓存未启用，仅内存热更" + ) logger.info( - "[Worker] %s 已将配置写入 %s: %s", - worker_id, _plugin_config_path(worker_type), new_section, + "[Worker] %s 已将配置写入缓存 %s: %s", + worker_id, _configs_cache_key(worker_type), new_section, ) else: applied_updates = { @@ -264,12 +594,16 @@ async def _handle_admin_command( "[Worker] %s 收到配置变更指令，将以新配置重启（%d 个键）", worker_id, len(applied_updates), ) - except Exception as e: # noqa: BLE001 - 写回/校验失败按指令失败返回 + except Exception as e: # noqa: BLE001 - 校验/写缓存失败按指令失败返回 setup_error = f"配置写入失败: {e}" logger.error("[Worker] %s %s", worker_id, setup_error) elif cmd == ADMIN_CMD_RESCALE: - # 调整当前进程内该类型的副本槽位数（缩容走优雅排空）。作用于本进程 Worker 池， - # 不退出本副本自身；跨进程（external）指令经 Zenoh 到达其容器进程内的指定副本 → 其进程池扩缩。 + # 调整本进程内该类型的副本槽位数（缩容走优雅排空）。支持两种语义： + # - 逐实例绝对（data.instance_id + data.absolute=true 且命中本实例 id）： + # 直接把本实例池设为 data.scale，并写入 instcounts 覆盖（精细扩缩， + # 进程重启后仍生效）；目标是其它实例的指令被安全忽略，避免误缩。 + # - 全局（无 instance_id）：先清除所有逐实例覆盖回到「均分」对称语义，再把 + # 全局 target 按在线网关/存活实例分摊为本进程应承载数并写回缓存。 try: target = int(raw.get("scale", -1)) except (TypeError, ValueError): @@ -281,18 +615,47 @@ async def _handle_admin_command( if pool is None: setup_error = f"当前进程无 {worker_type} 类型的 Worker 池，无法缩放" else: - asyncio.ensure_future(pool.set_target(target)) - logger.info( - "[Worker] %s 收到缩放指令，目标副本数=%d（当前池=%d）", - worker_id, target, len(pool._slots), - ) - # 持久化目标数量到本进程侧配置（重启后保持），失败不影响本次内存缩放 - try: - _write_plugin_count(worker_type, target) - persist_note = f"count={target} 已持久化" - except Exception as e: # noqa: BLE001 - persist_note = f"count 持久化失败: {e}" - logger.warning("[Worker] %s %s", worker_id, persist_note) + cache = getattr(controller, "cache", None) + gw_id = getattr(pool.worker_manager, "_gateway_id", None) \ + if getattr(pool, "worker_manager", None) is not None else None + total = max(0, target) + target_inst = raw.get("instance_id") + if target_inst and raw.get("absolute"): + # 逐实例绝对扩缩：只对命中本实例生效 + if target_inst == current_instance_id(): + _write_own_instance_count(cache, worker_type, total, target_inst) + app_count = total + asyncio.ensure_future(pool.set_target(app_count)) + persist_note = f"实例 {target_inst} 绝对 count={total}" + logger.info( + "[Worker] %s 实例级缩放 count=%d（当前池=%d）", + worker_id, total, len(pool._slots), + ) + else: + # 目标为其它实例：不缩放本池，安全忽略 + app_count = len(pool._slots) + persist_note = f"实例级缩放目标 {target_inst} 非本实例，忽略" + logger.info( + "[Worker] %s 忽略其它实例的缩放指令（target=%s）", + worker_id, target_inst, + ) + else: + # 全局对称：清除逐实例覆盖，回到均分 + _clear_instance_counts(cache, worker_type) + if cache is not None: + cache[_SCALE_COUNTS_KEY.format(type=worker_type)] = total + app_count = distribute_quota(cache, worker_type, total, gw_id) + persist_note = ( + f"全局 count={total}，本进程应用={app_count} 已写缓存" + ) + else: + app_count = total + persist_note = "缓存未启用，count 仅内存缩放" + asyncio.ensure_future(pool.set_target(app_count)) + logger.info( + "[Worker] %s 收到缩放指令，全局目标=%d，本进程应用=%d（当前池=%d）", + worker_id, total, app_count, len(pool._slots), + ) else: query.reply( key_expr=query.key_expr, @@ -301,6 +664,7 @@ async def _handle_admin_command( "worker_id": worker_id, "index": controller.index, "epoch": controller.epoch, + "instance_id": current_instance_id(), "status": "running", }), ) @@ -682,6 +1046,10 @@ def create_workers( session, worker_type, config, worker_manager=worker_manager, reload_cb=reload_cb ) _register_process_pool(worker_type, pool) + # count 收归 worker：按缓存全局总量与分摊后启动本进程副本数（inline 分摊/external 全额） + if pool.cache is not None: + gw_id = getattr(worker_manager, "_gateway_id", None) if worker_manager is not None else None + count = resolve_worker_count(worker_type, count, pool.cache, pool.mode, gw_id) return pool.start(count) @@ -710,9 +1078,17 @@ class WorkerPool: ): self.session = session self.worker_type = worker_type - self.config = config or {} + self.config = dict(config) if isinstance(config, dict) else {} self.worker_manager = worker_manager self.reload_cb = reload_cb + # 本类型配置缓存句柄：进程插件经 set_worker_cache 注入 > 网关 manager.cache > 无 + self.cache = _resolve_cache(worker_type, worker_manager) + self.mode = _resolve_mode(worker_type, worker_manager) + # [workers.config] 单一事实来源：启动时按模式解析进运行配置（不再由网关/进程入口直写缓存） + if self.cache is not None: + self.config = resolve_runtime_config( + worker_type, self.config, self.cache, self.mode + ) # index -> controller / task（仅记录当前活动槽位；drain 结束后由 done 回调移除） self._slots: Dict[int, WorkerController] = {} self._tasks: Dict[int, asyncio.Task] = {} @@ -724,6 +1100,7 @@ class WorkerPool: index = self._next_index self._next_index += 1 controller = WorkerController(self.worker_type, index) + controller.cache = self.cache task = asyncio.create_task( _managed_worker( self.session, self.worker_type, controller, self.config, @@ -794,4 +1171,6 @@ __all__ = [ "create_workers", "WorkerPool", "get_process_pool", + "set_worker_cache", + "resolve_runtime_config", ]