mirror of
https://github.com/ThearchyHelios/AgentLab.git
synced 2026-11-15 00:11:59 +00:00
复核 R-b9(Fenced 打到不该打的地方):check() 续晚了谁问到都隔离,但只对绑定了世代的执行(运行的一段执行、后台任务)抛 Fenced,HTTP 处理、调度那一拍、进程间请求问到时只隔离不抛;调度循环、心跳、办进程间请求也接住 Fenced;续租的任务不再因为一下出错就退出——续租、隔离之后重新登记都包住,照实记日志、退避着重试(最长一个租约时长)。隔离时马上换身份(之后新建的行记新身份),新身份登记上之前心跳不争当班、不接手,新开始的运行、后台任务在 fence.ready() 上等(最多一个租约时长,等不到就不开始,那几行由别的进程按属主不在了处理,这边同时放弃这个一直没登记上的身份、再换一个)。日志写出新旧完整身份。补测试:没绑定的调用者只隔离不抛、绑定在隔离之前的抛、新身份登记上之前 ready 等着;调度循环被 Fenced 打到一拍之后照常;重新登记先被版本守卫拦、再碰上库连不上,续租任务都不退出,登记上之后照常续租
This commit is contained in:
@@ -6,9 +6,10 @@
|
||||
|
||||
- 续租约在单独的任务、单独的连接上做(cluster/service.py),不被同一拍别的活拖住;最近一次续上的时刻记在这里(renewed)。
|
||||
- 本进程的执行在动手写之前问一句 check():离上次续上已经超过一个租约时长(别的进程按两个租约时长才接手,中间留着一段),或者这段
|
||||
执行是在上一次隔离之前开始的(世代不对),就抛 Fenced——写事件、写断点、写终态、写待办和动作、发外部写入之前都问。
|
||||
执行是在上一次隔离之前开始的(世代不对),就抛 Fenced——写事件、写断点、写终态、写待办和动作、发外部写入之前都问。只对绑定了
|
||||
世代的执行(运行的一段执行、后台任务)抛;别的调用者问到时只隔离、不抛。
|
||||
- 一旦发现续晚了(trip):世代加一,手上的执行全部停掉(停的时候不写终态,那些行交给接手的进程按「属主不在了」处理),换一个新的
|
||||
身份重新登记,之后发起的照常。
|
||||
身份重新登记;新身份登记上之前新开始的执行先等着(ready),登记上之后照常。
|
||||
|
||||
**用哪只钟(复核 R-b7)。** 别的进程判「租约过期」用的是库的时钟(cluster/workers:续租写 lease_until = 库的 now() + 租约),
|
||||
跟各台机器的墙钟差多少无关。这边判「续晚了」用计入休眠的钟:time.monotonic() 在 macOS(mach_absolute_time)、Linux
|
||||
@@ -21,6 +22,7 @@ Fenced 是 BaseException:节点、工具里常见的 except Exception 拦不
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import contextvars
|
||||
import logging
|
||||
import sys
|
||||
@@ -96,17 +98,38 @@ def bind() -> None:
|
||||
|
||||
|
||||
def stale() -> bool:
|
||||
"""离上次续上租约是不是已经超过一个租约时长:两只钟(计入休眠的单调钟、墙钟)里走得多的那只说了算。"""
|
||||
"""离上次续上租约是不是已经超过一个租约时长:两只钟(计入休眠的单调钟、墙钟)里走得多的那只说了算。隔离之后、新身份还没
|
||||
登记上的那一段也算(pending)。"""
|
||||
if not _active:
|
||||
return False
|
||||
since = _since()
|
||||
if since is None:
|
||||
return False
|
||||
return True
|
||||
from app.cluster.workers import lease_s
|
||||
|
||||
return max(since) > lease_s()
|
||||
|
||||
|
||||
def pending() -> bool:
|
||||
"""隔离过、换的新身份还没登记上。"""
|
||||
return _active and _renewed is None
|
||||
|
||||
|
||||
async def ready(timeout: float | None = None) -> None:
|
||||
"""新开始的一段执行(运行、后台任务)在 bind() 之前等一等:隔离之后、新身份还没登记上时,它认领的行在别的进程眼里属主
|
||||
不在,会被当成没人管接手(复核 R-b9)。最多等一个租约时长(timeout),等不到抛 Fenced——不开始,那几行由别的进程照「属主
|
||||
不在了」处理(cluster/service 那边同时会放弃这个一直没登记上的身份)。"""
|
||||
if not pending():
|
||||
return
|
||||
from app.cluster.workers import lease_s
|
||||
|
||||
deadline = _clock() + (lease_s() if timeout is None else timeout)
|
||||
while pending():
|
||||
if _clock() >= deadline:
|
||||
raise Fenced()
|
||||
await asyncio.sleep(0.05)
|
||||
|
||||
|
||||
def jumped() -> bool:
|
||||
"""两只钟差出一个租约以上(墙钟往前或往后跳了,或者单调钟没计入休眠):续租时问,是就隔离。"""
|
||||
if not _active:
|
||||
@@ -120,22 +143,22 @@ def jumped() -> bool:
|
||||
|
||||
|
||||
def fenced_here() -> bool:
|
||||
"""当前这段执行还能不能写:世代不对(隔离之前开始的)或者租约续晚了。"""
|
||||
"""当前这段执行还能不能写:世代不对(隔离之前开始的)或者租约续晚了。只说绑定了世代的执行(bind 过的)。"""
|
||||
if not _active:
|
||||
return False
|
||||
bound = _bound.get()
|
||||
return (bound is not None and bound != _generation) or stale()
|
||||
return bound is not None and (bound != _generation or stale())
|
||||
|
||||
|
||||
def check() -> None:
|
||||
"""写之前问一句:不许写就抛 Fenced(续晚了的话顺手隔离)。"""
|
||||
"""写之前问一句。续晚了就隔离(谁先问到都一样);抛 Fenced 只对绑定了世代的执行(运行的一段执行、后台任务:bind 过的)——
|
||||
别的调用者(HTTP 处理、调度那一拍、进程间请求)不是认领了行在跑的执行,抛给它们只会把调度循环之类的打停(复核 R-b9)。"""
|
||||
if not _active:
|
||||
return
|
||||
if stale():
|
||||
if _renewed is not None and stale():
|
||||
trip("执行要写的时候发现租约已经续晚了")
|
||||
raise Fenced()
|
||||
bound = _bound.get()
|
||||
if bound is not None and bound != _generation:
|
||||
if bound is not None and (bound != _generation or stale()):
|
||||
raise Fenced()
|
||||
|
||||
|
||||
@@ -162,4 +185,4 @@ def trip(why: str) -> bool:
|
||||
|
||||
|
||||
__all__ = ["Fenced", "SLEEP_AWARE", "activate", "bind", "check", "deactivate", "fenced_here", "generation", "jumped", "now",
|
||||
"on_trip", "renewed", "stale", "trip"]
|
||||
"on_trip", "pending", "ready", "renewed", "stale", "trip"]
|
||||
|
||||
@@ -94,9 +94,11 @@ _running: set[asyncio.Future[Any]] = set()
|
||||
|
||||
async def _handle(fn: Callable[..., Awaitable[Any]], kind: str, target: str, target_id: str, actor: str | None,
|
||||
payload: dict[str, Any]) -> None:
|
||||
from app.cluster.fence import Fenced
|
||||
|
||||
try:
|
||||
await fn(target_id, actor, payload)
|
||||
except Exception: # noqa: BLE001
|
||||
except (Exception, Fenced): # noqa: BLE001 - Fenced 是 BaseException,不接住就成了没人取的任务异常
|
||||
logger.exception("办进程间请求 %s/%s %s 出错", kind, target, target_id)
|
||||
|
||||
|
||||
|
||||
@@ -65,6 +65,11 @@ class _Service:
|
||||
#: SQLite:守卫已经确认只有本进程——登记只写一次、不续租、本进程直接当班、接手只在启动时做(不为协调多写库)
|
||||
self.single = False
|
||||
self._lease_conn: Any = None
|
||||
self.lease_wake: asyncio.Event | None = None
|
||||
#: 隔离之后、新身份还没登记上:从什么时候起(cluster/fence.now 的读数)
|
||||
self.pending_since: float | None = None
|
||||
#: 隔离掉(或者一直没登记上、放弃了)的旧身份:新身份登记上之后收拾它们留下的登记、名额、请求、当班租约
|
||||
self.retired: list[str] = []
|
||||
|
||||
async def start(self) -> None:
|
||||
"""登记、开始收发通知、起心跳。SQLite 时先取单进程的守卫(拿不到抛 guard.SingleProcessOnly)。"""
|
||||
@@ -94,6 +99,7 @@ class _Service:
|
||||
await self._lead()
|
||||
fence.activate()
|
||||
fence.on_trip(self._on_fenced)
|
||||
self.lease_wake = asyncio.Event()
|
||||
self.lease_task = asyncio.create_task(self._lease_loop(), name="agentlab-cluster-lease")
|
||||
self.task = asyncio.create_task(self._loop(), name="agentlab-cluster-heartbeat")
|
||||
logger.info("工作进程 %s 已登记(%s,租约 %g 秒、心跳 %g 秒;现在另有 %d 个进程)", identity.describe(identity.owner()),
|
||||
@@ -159,27 +165,62 @@ class _Service:
|
||||
return True
|
||||
|
||||
async def _lease_loop(self) -> None:
|
||||
"""续租约;隔离之后改成重新登记新身份。哪一下出了错(库一抖、版本守卫拦下)都不让这个任务退出——它一退,之后就再也没有
|
||||
隔离保护了(复核 R-b9):照实记日志,退避着重试(最长一个租约时长)。"""
|
||||
from app.cluster import fence
|
||||
|
||||
delay = workers.heartbeat_s()
|
||||
while True:
|
||||
await self.renew()
|
||||
await asyncio.sleep(workers.heartbeat_s())
|
||||
try:
|
||||
if self.pending_since is not None:
|
||||
await self.register_again()
|
||||
else:
|
||||
await self.renew()
|
||||
delay = workers.heartbeat_s()
|
||||
except asyncio.CancelledError:
|
||||
raise
|
||||
except (Exception, fence.Fenced): # noqa: BLE001
|
||||
delay = min(max(delay * 2, workers.heartbeat_s()), workers.lease_s())
|
||||
logger.exception("%s出错,%g 秒后再试", "隔离之后重新登记" if self.pending_since is not None else "续租约", delay)
|
||||
if self.lease_wake is None:
|
||||
await asyncio.sleep(delay)
|
||||
continue
|
||||
with contextlib.suppress(asyncio.TimeoutError):
|
||||
await asyncio.wait_for(self.lease_wake.wait(), timeout=delay)
|
||||
self.lease_wake.clear()
|
||||
|
||||
def _on_fenced(self) -> None:
|
||||
"""隔离了:手上的执行由各处自己停(runner、job_runtime 登记的处理),这里换身份重新登记。"""
|
||||
asyncio.ensure_future(self._reidentify())
|
||||
|
||||
async def _reidentify(self) -> None:
|
||||
"""隔离了:手上的执行由各处自己停(runner、job_runtime 登记的处理);这里马上换身份(之后新建的行记新身份),叫续租的任务
|
||||
去登记它。"""
|
||||
from app.cluster import fence
|
||||
|
||||
old = identity.owner()
|
||||
self.retired.append(old)
|
||||
new = identity.regenerate()
|
||||
self.pending_since = fence.now()
|
||||
leader.scheduler.drop()
|
||||
try:
|
||||
await workers.register()
|
||||
fence.renewed()
|
||||
logger.warning("隔离之后换了身份重新登记:%s → %s(之前认领的行由当班进程按「属主不在了」接手)",
|
||||
identity.describe(old), identity.describe(new))
|
||||
except Exception: # noqa: BLE001
|
||||
logger.exception("隔离之后重新登记失败,下一拍再试")
|
||||
logger.warning("隔离:换成新身份 %s(旧身份 %s),登记上之前新开始的执行先等着", new, old)
|
||||
if self.lease_wake is not None:
|
||||
self.lease_wake.set()
|
||||
|
||||
async def register_again(self) -> None:
|
||||
"""隔离之后登记新身份。一个租约时长都没登记上(库连不上、版本守卫拦着)就放弃这个身份、再换一个:这期间等不到的执行已经
|
||||
放弃开始(fence.ready),它们的行记着这个身份——它永远不登记,别的进程就照「属主不在了」处理那几行。"""
|
||||
from app.cluster import fence
|
||||
|
||||
assert self.pending_since is not None
|
||||
if fence.now() - self.pending_since > workers.lease_s():
|
||||
gave_up = identity.owner()
|
||||
self.retired.append(gave_up)
|
||||
identity.regenerate()
|
||||
self.pending_since = fence.now()
|
||||
logger.error("新身份 %s 一个租约时长都没登记上,放弃它、换成 %s 再试", gave_up, identity.owner())
|
||||
await workers.register()
|
||||
retired, self.retired = self.retired, []
|
||||
self.pending_since = None
|
||||
fence.renewed()
|
||||
logger.warning("隔离之后换了身份重新登记:%s → %s(之前认领的行由当班进程按「属主不在了」接手)",
|
||||
"、".join(retired), identity.owner())
|
||||
|
||||
# ---------------- 当班、心跳的其余活 ----------------
|
||||
|
||||
@@ -203,8 +244,9 @@ class _Service:
|
||||
return now_leading
|
||||
|
||||
async def beat(self) -> None:
|
||||
"""心跳的一拍(续租约另在 _lease_loop;测试里直接调)。SQLite 只做本进程自己攒着的活。"""
|
||||
if self.single:
|
||||
"""心跳的一拍(续租约另在 _lease_loop;测试里直接调)。SQLite 只做本进程自己攒着的活;隔离之后新身份还没登记上时也只做
|
||||
这些(不争当班、不接手)。"""
|
||||
if self.single or self.pending_since is not None:
|
||||
await self._own_work()
|
||||
return
|
||||
live = await workers.liveness()
|
||||
@@ -254,6 +296,8 @@ class _Service:
|
||||
return out
|
||||
|
||||
async def _loop(self) -> None:
|
||||
from app.cluster import fence
|
||||
|
||||
assert self.wake is not None
|
||||
while True:
|
||||
with contextlib.suppress(asyncio.TimeoutError):
|
||||
@@ -263,7 +307,7 @@ class _Service:
|
||||
await self.beat()
|
||||
except asyncio.CancelledError:
|
||||
raise
|
||||
except Exception: # noqa: BLE001 - 一拍出错(库一时不通)不该让心跳停下
|
||||
except (Exception, fence.Fenced): # noqa: BLE001 - 一拍出错(库一时不通)不该让心跳停下
|
||||
logger.exception("心跳这一拍出错,下一拍照常进行")
|
||||
|
||||
async def _light_loop(self) -> None:
|
||||
@@ -284,6 +328,7 @@ class _Service:
|
||||
with contextlib.suppress(asyncio.CancelledError, Exception):
|
||||
await task
|
||||
self.task = self.lease_task = None
|
||||
self.pending_since, self.retired = None, []
|
||||
fence.deactivate()
|
||||
with contextlib.suppress(Exception):
|
||||
await leader.scheduler.release()
|
||||
|
||||
@@ -543,6 +543,11 @@ class JobRun:
|
||||
async def run(self) -> None:
|
||||
from app.cluster import fence
|
||||
|
||||
try:
|
||||
await fence.ready() # 隔离之后新身份还没登记上时先等一等(cluster/fence.ready)
|
||||
except fence.Fenced:
|
||||
logger.error("%s %s:本进程隔离之后一直没重新登记上,不开始(由别的进程按「属主不在了」处理)", self.kind, self.job_id)
|
||||
return
|
||||
fence.bind()
|
||||
try:
|
||||
try:
|
||||
|
||||
@@ -794,7 +794,12 @@ class RunManager:
|
||||
# 等待从 run.interrupted 算起——以前执行一直计到 finally,收尾那一截执行和等待两边都算,
|
||||
# active + wait 比墙钟还长(机器一忙就差出几十毫秒)
|
||||
paused: float | None = None
|
||||
# 这段执行属于现在这一世代(cluster/fence.py):本进程隔离之后,它不再写
|
||||
# 这段执行属于现在这一世代(cluster/fence.py):本进程隔离之后,它不再写。隔离之后新身份还没登记上时先等一等
|
||||
try:
|
||||
await fence.ready()
|
||||
except Fenced:
|
||||
logger.error("运行 %s:本进程隔离之后一直没重新登记上,不开始执行(由别的进程按「属主不在了」处理)", run_id)
|
||||
return
|
||||
fence.bind()
|
||||
mine = identity.owner()
|
||||
#: 被隔离了(续不上租约):停手,不写终态(这次运行由接手的进程按「属主不在了」处理)
|
||||
|
||||
@@ -201,6 +201,7 @@ class _Runner:
|
||||
|
||||
async def _loop(self, interval: float, stopping: asyncio.Event) -> None:
|
||||
from app.cluster import leading
|
||||
from app.cluster.fence import Fenced
|
||||
from app.scheduler import webhook
|
||||
|
||||
self._poked = asyncio.Event()
|
||||
@@ -218,7 +219,7 @@ class _Runner:
|
||||
self.notify = asyncio.create_task(_notify(webhook), name="agentlab-scheduler-notify")
|
||||
except asyncio.CancelledError:
|
||||
raise
|
||||
except Exception: # noqa: BLE001 - 这一拍出错不该让循环停下
|
||||
except (Exception, Fenced): # noqa: BLE001 - 这一拍出错不该让循环停下(Fenced 是 BaseException:复核 R-b9)
|
||||
logger.exception("定时任务调度这一拍出错,下一拍照常进行")
|
||||
await self._wait(stopping, interval)
|
||||
|
||||
|
||||
@@ -4,7 +4,7 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import time
|
||||
import contextlib
|
||||
|
||||
import pytest
|
||||
from sqlalchemy import func, select
|
||||
@@ -168,3 +168,108 @@ async def test_a_repeated_seq_is_refused_by_the_database():
|
||||
RunEvent(id=new_id(), run_id=run.id, seq=1, type="log", ts=0, data={})])
|
||||
with pytest.raises(IntegrityError):
|
||||
await session.commit()
|
||||
|
||||
|
||||
# ---------------- 复核 R-b9:谁会被打停、谁不会 ----------------
|
||||
|
||||
|
||||
async def test_only_executions_bound_to_a_generation_are_stopped_by_check():
|
||||
"""续晚之后第一个问 check() 的不一定是运行本身:没绑定世代的调用者(HTTP 处理、调度那一拍)只隔离、不抛;绑定了的抛。
|
||||
隔离之后新身份还没登记上,新开始的执行在 ready() 上等,等不到抛 Fenced;登记上了就放行。"""
|
||||
gate = asyncio.Event()
|
||||
seen: list[str] = []
|
||||
|
||||
async def bound_writer() -> None:
|
||||
fence.bind()
|
||||
await gate.wait()
|
||||
try:
|
||||
fence.check()
|
||||
seen.append("wrote")
|
||||
except fence.Fenced:
|
||||
seen.append("fenced")
|
||||
|
||||
task = asyncio.create_task(bound_writer())
|
||||
await asyncio.sleep(0)
|
||||
before = fence.generation()
|
||||
fence.renewed(ago=lease_s() + 1)
|
||||
fence.check() # 没绑定世代:只隔离
|
||||
assert fence.generation() == before + 1 and fence.pending()
|
||||
gate.set()
|
||||
await task
|
||||
assert seen == ["fenced"], "绑定在隔离之前的那一段执行:不许再写"
|
||||
|
||||
with pytest.raises(fence.Fenced):
|
||||
await fence.ready(timeout=0.1) # 新身份还没登记上:新开始的执行等不到就不开始
|
||||
waiting = asyncio.create_task(fence.ready(timeout=5))
|
||||
await asyncio.sleep(0.1)
|
||||
assert not waiting.done()
|
||||
fence.renewed() # 登记上了
|
||||
await asyncio.wait_for(waiting, timeout=2)
|
||||
fence.bind()
|
||||
fence.check() # 新世代照常
|
||||
|
||||
|
||||
async def test_the_scheduler_loop_survives_a_fenced_tick(monkeypatch):
|
||||
from app.scheduler import loop as sloop
|
||||
|
||||
calls = {"n": 0}
|
||||
|
||||
async def fire_due(now=None):
|
||||
calls["n"] += 1
|
||||
if calls["n"] == 1:
|
||||
raise fence.Fenced()
|
||||
return []
|
||||
|
||||
monkeypatch.setattr(sloop, "fire_due", fire_due)
|
||||
await sloop.runner.start(interval=0.02)
|
||||
try:
|
||||
for _ in range(200):
|
||||
if calls["n"] >= 3:
|
||||
break
|
||||
await asyncio.sleep(0.02)
|
||||
assert calls["n"] >= 3 and sloop.runner.running(), "被 Fenced 打到的那一拍之后照常往下"
|
||||
finally:
|
||||
await sloop.runner.stop()
|
||||
|
||||
|
||||
async def test_the_lease_task_keeps_going_when_registering_again_fails(monkeypatch):
|
||||
"""隔离之后重新登记:库一抖、版本守卫拦下都不让续租的任务退出,退避着重试;登记上了照常续租。"""
|
||||
from app.cluster import identity, service as service_mod, workers
|
||||
from app.cluster.version import VersionMismatch
|
||||
|
||||
monkeypatch.setattr(identity, "OWNER", identity.OWNER) # 测试完换回原来的身份
|
||||
monkeypatch.setattr(workers, "heartbeat_s", lambda: 0.01)
|
||||
monkeypatch.setattr(workers, "lease_s", lambda: 5.0)
|
||||
svc = service_mod._Service()
|
||||
tries = {"register": 0, "renew": 0}
|
||||
|
||||
async def register(*, exclusive: bool = False) -> None: # noqa: ARG001
|
||||
tries["register"] += 1
|
||||
if tries["register"] == 1:
|
||||
raise VersionMismatch("库里还有版本不同的进程活着")
|
||||
if tries["register"] == 2:
|
||||
raise OSError("库一时连不上")
|
||||
|
||||
async def renew() -> bool:
|
||||
tries["renew"] += 1
|
||||
return True
|
||||
|
||||
monkeypatch.setattr(workers, "register", register)
|
||||
monkeypatch.setattr(svc, "renew", renew)
|
||||
old = identity.owner()
|
||||
fence.trip("测试")
|
||||
svc._on_fenced()
|
||||
assert identity.owner() != old and svc.pending_since is not None
|
||||
task = asyncio.create_task(svc._lease_loop())
|
||||
try:
|
||||
for _ in range(300):
|
||||
if tries["renew"] >= 2:
|
||||
break
|
||||
await asyncio.sleep(0.01)
|
||||
assert not task.done(), "续租的任务还在"
|
||||
assert tries["register"] == 3 and svc.pending_since is None and not fence.pending()
|
||||
assert tries["renew"] >= 2, "登记上之后照常续租"
|
||||
finally:
|
||||
task.cancel()
|
||||
with contextlib.suppress(asyncio.CancelledError):
|
||||
await task
|
||||
|
||||
Reference in New Issue
Block a user