modify: worker 管理多进程副本
This commit is contained in:
@@ -0,0 +1,20 @@
|
||||
"""
|
||||
原生 FastAPI 对照服务（无 Zenoh / 无网关中间件 / 无 worker）
|
||||
|
||||
用于与 chongming 网关的 /api/v1/demo/hello 做性能对照：
|
||||
同样的 GET 返回一个 JSON（字段形状与 demo/hello 一致），排除业务与 RPC 链路，
|
||||
仅度量纯 FastAPI+Uvicorn 框架本身的吞吐与延迟。
|
||||
|
||||
启动:
|
||||
uv run uvicorn native_fastapi:app --host 0.0.0.0 --port 8001 --no-access-log
|
||||
"""
|
||||
import time
|
||||
|
||||
from fastapi import FastAPI
|
||||
|
||||
app = FastAPI()
|
||||
|
||||
|
||||
@app.get("/hello")
|
||||
async def hello(name: str = "bench"):
|
||||
return {"greeting": f"Hello, {name}!", "worker_id": "native", "timestamp": time.time()}
|
||||
@@ -0,0 +1,187 @@
|
||||
"""
|
||||
外部进程多实例全局分摊 —— distribute_quota 逻辑单测（无需网关/缓存后端）。
|
||||
|
||||
用 dict 作缓存桩，模拟同类型多个外部进程实例注册，验证扩缩容时全局 count 按
|
||||
「存活实例数 N」分摊（而非每个进程都全额），以及实例退出/租约过期后的接管行为。
|
||||
"""
|
||||
import os
|
||||
import sys
|
||||
import time
|
||||
|
||||
sys.path.insert(0, os.path.join(os.path.dirname(__file__), "../utils/worker/src"))
|
||||
|
||||
import chongming_worker.runtime as r
|
||||
|
||||
PASS = 0
|
||||
FAIL = 0
|
||||
|
||||
|
||||
def report(name, cond, detail=""):
|
||||
global PASS, FAIL
|
||||
if cond:
|
||||
PASS += 1
|
||||
print(f"✅ {name}" + (f" — {detail}" if detail else ""))
|
||||
else:
|
||||
FAIL += 1
|
||||
print(f"❌ {name}" + (f" — {detail}" if detail else ""))
|
||||
|
||||
|
||||
def fake_cache():
|
||||
return {}
|
||||
|
||||
|
||||
def as_instance(iid, cache, worker_type="demo"):
|
||||
"""把当前身份设为 iid，并在缓存登记该实例参与方。"""
|
||||
old = os.environ.get("CHONGMING_WORKER_INSTANCE")
|
||||
os.environ["CHONGMING_WORKER_INSTANCE"] = iid
|
||||
try:
|
||||
r.register_instance(worker_type, cache)
|
||||
finally:
|
||||
if old is None:
|
||||
os.environ.pop("CHONGMING_WORKER_INSTANCE", None)
|
||||
else:
|
||||
os.environ["CHONGMING_WORKER_INSTANCE"] = old
|
||||
|
||||
|
||||
def quota_as(iid, worker_type, total, cache):
|
||||
"""以指定实例身份计算其分摊配额（external 分支）。"""
|
||||
old = os.environ.get("CHONGMING_WORKER_INSTANCE")
|
||||
os.environ["CHONGMING_WORKER_INSTANCE"] = iid
|
||||
try:
|
||||
return r.distribute_quota(cache, worker_type, total, gw_id=None)
|
||||
finally:
|
||||
if old is None:
|
||||
os.environ.pop("CHONGMING_WORKER_INSTANCE", None)
|
||||
else:
|
||||
os.environ["CHONGMING_WORKER_INSTANCE"] = old
|
||||
|
||||
|
||||
def unreg_instance(iid, cache, worker_type="demo"):
|
||||
"""以指定实例身份从参与方列表移除它。"""
|
||||
old = os.environ.get("CHONGMING_WORKER_INSTANCE")
|
||||
os.environ["CHONGMING_WORKER_INSTANCE"] = iid
|
||||
try:
|
||||
r.unregister_instance(worker_type, cache)
|
||||
finally:
|
||||
if old is None:
|
||||
os.environ.pop("CHONGMING_WORKER_INSTANCE", None)
|
||||
else:
|
||||
os.environ["CHONGMING_WORKER_INSTANCE"] = old
|
||||
|
||||
|
||||
def run():
|
||||
cache = fake_cache()
|
||||
wt = "demo"
|
||||
|
||||
# 1. 单实例（未启用实例注册，参与方为空）→ 兜底全额（兼容旧行为）
|
||||
q = r.distribute_quota(cache, wt, 6, gw_id=None)
|
||||
report("单实例无参与方 → 全额承担 6", q == 6, f"got={q}")
|
||||
|
||||
# 2. 两个实例并存：全局 6 分摊 → 各 3
|
||||
as_instance("inst-a", cache, wt)
|
||||
as_instance("inst-b", cache, wt)
|
||||
qa = quota_as("inst-a", wt, 6, cache)
|
||||
qb = quota_as("inst-b", wt, 6, cache)
|
||||
report("两个实例并存 → 各分 3", qa == 3 and qb == 3, f"a={qa}, b={qb}")
|
||||
|
||||
# 3. 三个实例、总数不能整除：7 // 3 余 1 → 按序给第一个补 1
|
||||
as_instance("inst-c", cache, wt)
|
||||
qs = {i: quota_as(i, wt, 7, cache) for i in ("inst-a", "inst-b", "inst-c")}
|
||||
report(
|
||||
"三个实例 ≠整除 → 总和=7 且余量补靠前实例",
|
||||
sum(qs.values()) == 7 and set(qs.values()) == {2, 2, 3},
|
||||
f"{qs}",
|
||||
)
|
||||
|
||||
# 4. 一个实例退出（unregister）→ 其余接管全部
|
||||
unreg_instance("inst-c", cache, wt)
|
||||
qa2 = quota_as("inst-a", wt, 6, cache)
|
||||
qb2 = quota_as("inst-b", wt, 6, cache)
|
||||
report(
|
||||
"实例退出后由剩余接管 → 各分 3",
|
||||
qa2 == 3 and qb2 == 3,
|
||||
f"a={qa2}, b={qb2}",
|
||||
)
|
||||
|
||||
# 5. 租约过期清理：伪造一个过期的参与方，读实例列表应排除它
|
||||
cache[r._SCALE_INSTANCES_KEY.format(type=wt)] = [
|
||||
["inst-stale", time.time() - r._INSTANCE_LEASE - 10],
|
||||
["inst-fresh", time.time()],
|
||||
]
|
||||
ids = r._read_instances(cache, wt)
|
||||
report(
|
||||
"死实例按租约剔除",
|
||||
ids == ["inst-fresh"],
|
||||
f"instances={ids}",
|
||||
)
|
||||
|
||||
# 6. 实例数=1 时全额（不误分摊）
|
||||
cache[r._SCALE_INSTANCES_KEY.format(type=wt)] = [
|
||||
[r.current_instance_id(), time.time()]
|
||||
]
|
||||
q1 = r.distribute_quota(cache, wt, 5, gw_id=None)
|
||||
report("单存活实例 → 全额 5", q1 == 5, f"got={q1}")
|
||||
|
||||
# 7. inline（有网关身份）路径回归：按在线网关分摊不受影响
|
||||
cache[r._SCALE_GATEWAYS_KEY] = [["gw-a", time.time()], ["gw-b", time.time()]]
|
||||
qa_in = r.distribute_quota(cache, wt, 6, "gw-a")
|
||||
qb_in = r.distribute_quota(cache, wt, 6, "gw-b")
|
||||
report(
|
||||
"inline 按网关分摊回归 → 各 3",
|
||||
qa_in == 3 and qb_in == 3,
|
||||
f"gw-a={qa_in}, gw-b={qb_in}",
|
||||
)
|
||||
|
||||
# 8. 逐实例绝对覆盖：写入后能被读回且供启动解析（external）使用
|
||||
r._write_own_instance_count(cache, wt, 4, "inst-a")
|
||||
r._write_own_instance_count(cache, wt, 2, "inst-b")
|
||||
report(
|
||||
"instcounts 覆盖写入/读回",
|
||||
r._own_instance_count(cache, wt, "inst-a") == 4
|
||||
and r._own_instance_count(cache, wt, "inst-b") == 2,
|
||||
f"{r._read_instance_counts(cache, wt)}",
|
||||
)
|
||||
# 启动解析：external（gw_id=None）有绝对覆盖时直接采用，而非被全局均分
|
||||
old = os.environ.get("CHONGMING_WORKER_INSTANCE")
|
||||
os.environ["CHONGMING_WORKER_INSTANCE"] = "inst-a"
|
||||
try:
|
||||
restored = r.resolve_worker_count(wt, 6, cache, "cache", gw_id=None)
|
||||
finally:
|
||||
if old is None:
|
||||
os.environ.pop("CHONGMING_WORKER_INSTANCE", None)
|
||||
else:
|
||||
os.environ["CHONGMING_WORKER_INSTANCE"] = old
|
||||
report("resolve_worker_count 启动时采用本实例绝对数", restored == 4, f"got={restored}")
|
||||
|
||||
# 9. 清除覆盖后回退全局均分
|
||||
r._clear_instance_counts(cache, wt)
|
||||
report(
|
||||
"清除覆盖后覆盖表为空",
|
||||
r._read_instance_counts(cache, wt) == {},
|
||||
f"{r._read_instance_counts(cache, wt)}",
|
||||
)
|
||||
|
||||
# 10. 缺省实例身份唯一（hostname:pid）：同机未指定 --instance 的多进程不会并成一个参与方
|
||||
r.set_current_instance_id(None)
|
||||
old_env = os.environ.pop("CHONGMING_WORKER_INSTANCE", None)
|
||||
try:
|
||||
default_iid = r.current_instance_id()
|
||||
finally:
|
||||
if old_env is not None:
|
||||
os.environ["CHONGMING_WORKER_INSTANCE"] = old_env
|
||||
else:
|
||||
os.environ.pop("CHONGMING_WORKER_INSTANCE", None)
|
||||
import socket as _socket
|
||||
report(
|
||||
"缺省实例身份含进程区分（{host}:{pid}）",
|
||||
default_iid == f"{_socket.gethostname()}:{os.getpid()}",
|
||||
f"got={default_iid}",
|
||||
)
|
||||
|
||||
print("\n" + "=" * 56)
|
||||
print(f"测试结果: ✅ {PASS} 通过 ❌ {FAIL} 失败")
|
||||
return 0 if FAIL == 0 else 1
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(run())
|
||||
Reference in New Issue
Block a user