mirror of
https://github.com/ThearchyHelios/AgentLab.git
synced 2026-11-15 00:11:59 +00:00
复核 R-b14（PostgreSQL 单进程墙钟一跳就隔离自己）：取得到计入休眠的钟（SLEEP_AWARE）时 stale() 只看它，墙钟只在取不到时当后备（两只里走得多的那只）；两只钟对不上（jumped）不再直接隔离，先续一次租——续租那一句按库的时钟核对「自己的租约还没过期」，以它为准：续上了只记一条日志照常，续不上才隔离。README「多进程」补一句：库服务器自己的时钟往前跳超过一个租约时所有进程一起隔离、怎么恢复，往回跳时真没了的进程要多等，本机墙钟跳了不隔离。补测试：墙钟往前往后跳一小时不算续晚了、只算对不上，计入休眠的钟走过两个租约照样算，取不到时墙钟当后备；PostgreSQL 单进程墙钟 ±1 小时续租照常续上、不隔离，真过期仍隔离；执行到一半墙钟 ±1 小时，运行照常跑完、世代不变（改前的写法往前跳那条不过）
This commit is contained in:
ThearchyHelios
@@ -1188,7 +1188,7 @@ asyncpg 方言一律这样做，同一个事务里、在同一条服务端连接
|
||||
SQL 形状。各进程各自的并发闸门（口径比对、自动对比、起草的模型调用、上传解析）按进程算，进程多了总量跟着多。
|
||||
|
||||
**多台机器。** 几台机器要用同一份数据目录：放在共享存储（NFS 之类）上，每台挂在**同一个绝对路径**（库里记着上传表格各版本库文件的
|
||||
绝对路径），各台机器的时钟仍要对时（NTP）：租约按库的时钟算，机器之间差多少都不影响谁算活着，但「停了两个租约时长」这类判断看的是各机器按本机时钟写下的时间。文件都是先写临时文件再改名，读的一方看不到写了一半的；写上传版本库、记录表
|
||||
绝对路径），各台机器的时钟仍要对时（NTP）：租约按库的时钟算，机器之间差多少都不影响谁算活着，但「停了两个租约时长」这类判断看的是各机器按本机时钟写下的时间。库服务器自己的时钟往前跳超过一个租约时（少见），所有进程续租时都会发现自己的租约已经过期、一起隔离：执行到一半的运行标成中断，在运行详情里点「继续运行」从断点接着执行；后台任务按「属主不在了」的规矩处理；各进程自动换身份重新登记，之后发起的照常。往回跳时不隔离，只是真没了的进程要多等跳回去的那么久才被接手。本机墙钟跳了（NTP 一步校时、手动改时间）不会隔离：判续没续晚看的是计入休眠的单调钟，对不上时以库里的租约为准。文件都是先写临时文件再改名，读的一方看不到写了一半的；写上传版本库、记录表
|
||||
在库里的锁下一个接一个。但要说实话：数据目录里还有 SQLite 文件——上传表格的各版本库（发布后只读）、填报的记录表（`records/*.db`，
|
||||
原地追加写）、沙箱里代码自己建的库——SQLite 文件放在网络文件系统上，读写一致性靠文件锁，而 NFS 上的文件锁不可靠：一台机器在追加时
|
||||
另一台读，可能读到不一致的页。所以多台机器时，记录表写得多的部署建议用一台机器多进程，或者把数据目录放在支持可靠文件锁的存储上。
|
||||
|
||||
@@ -11,12 +11,14 @@
|
||||
- 一旦发现续晚了（trip）：世代加一，手上的执行全部停掉（停的时候不写终态，那些行交给接手的进程按「属主不在了」处理），换一个新的
|
||||
身份重新登记；新身份登记上之前新开始的执行先等着（ready），登记上之后照常。
|
||||
|
||||
**用哪只钟（复核 R-b7）。** 别的进程判「租约过期」用的是库的时钟（cluster/workers：续租写 lease_until = 库的 now() + 租约），
|
||||
跟各台机器的墙钟差多少无关。这边判「续晚了」用计入休眠的钟：time.monotonic() 在 macOS（mach_absolute_time）、Linux
|
||||
**用哪只钟（复核 R-b7、R-b14）。** 别的进程判「租约过期」用的是库的时钟（cluster/workers：续租写 lease_until = 库的 now() +
|
||||
租约），跟各台机器的墙钟差多少无关。这边判「续晚了」用计入休眠的钟：time.monotonic() 在 macOS（mach_absolute_time）、Linux
|
||||
（CLOCK_MONOTONIC）上都不计休眠——笔记本合盖、虚拟机挂起超过两个租约，对面早已接手，这边还以为刚续过。所以 Linux 用
|
||||
CLOCK_BOOTTIME，macOS 用 clock_gettime(CLOCK_MONOTONIC)（这个计入休眠），都取不到时退回 monotonic；不论哪种，另外同时比墙钟，
|
||||
两只钟里走得多的那只说过了一个租约就算续晚了。续租时再比一下两只钟（jumped）：差出一个租约以上（墙钟跳了、或者计入休眠的钟
|
||||
其实没计）就隔离。续租那一句本身也按库的时钟核对：库里自己的租约已经过期（或者登记不见了），就不再续这个身份。
|
||||
CLOCK_BOOTTIME，macOS 用 clock_gettime(CLOCK_MONOTONIC)（这个计入休眠），只看它；墙钟会被 NTP 一步校时、手动改时间、虚拟机
|
||||
对时拨来拨去，取得到计入休眠的钟时不拿它判（否则单进程墙钟一跳就把自己隔离了）。两只都取不到时退回 monotonic，墙钟当后备：
|
||||
两只里走得多的那只说过了一个租约就算续晚了。两只钟对不上（jumped，差出一个租约以上）时不直接隔离：先续一次租，续租那一句按库的
|
||||
时钟核对「自己的租约还没过期」，以它为准——续上了记一条日志照常，续不上（库里自己的租约已经过期、或者登记不见了）才隔离。
|
||||
虚拟机被宿主挂起时客户机的 CLOCK_BOOTTIME 也可能停着：那时靠续租那一句（最迟一拍心跳之后）发现过期。
|
||||
|
||||
Fenced 是 BaseException：节点、工具里常见的 except Exception 拦不住它，一路退到执行的最外层。没起协调（测试、脚本）时什么都不查。
|
||||
"""
|
||||
@@ -98,8 +100,8 @@ def bind() -> None:
|
||||
|
||||
|
||||
def stale() -> bool:
|
||||
"""离上次续上租约是不是已经超过一个租约时长：两只钟（计入休眠的单调钟、墙钟）里走得多的那只说了算。隔离之后、新身份还没
|
||||
登记上的那一段也算（pending）。"""
|
||||
"""离上次续上租约是不是已经超过一个租约时长：取得到计入休眠的钟时只看它；取不到时它（monotonic）和墙钟里走得多的那只说了算。
|
||||
隔离之后、新身份还没登记上的那一段也算（pending）。"""
|
||||
if not _active:
|
||||
return False
|
||||
since = _since()
|
||||
@@ -107,7 +109,7 @@ def stale() -> bool:
|
||||
return True
|
||||
from app.cluster.workers import lease_s
|
||||
|
||||
return max(since) > lease_s()
|
||||
return (since[0] if SLEEP_AWARE else max(since)) > lease_s()
|
||||
|
||||
|
||||
def pending() -> bool:
|
||||
@@ -131,7 +133,8 @@ async def ready(timeout: float | None = None) -> None:
|
||||
|
||||
|
||||
def jumped() -> bool:
|
||||
"""两只钟差出一个租约以上（墙钟往前或往后跳了，或者单调钟没计入休眠）：续租时问，是就隔离。"""
|
||||
"""两只钟差出一个租约以上（墙钟往前或往后跳了，或者单调钟没计入休眠）。续租时问：是的话不直接隔离，以续租那一句按库的时钟
|
||||
核对的结果为准（cluster/service.renew）。"""
|
||||
if not _active:
|
||||
return False
|
||||
since = _since()
|
||||
|
||||
@@ -131,15 +131,15 @@ class _Service:
|
||||
async def renew(self) -> bool:
|
||||
"""续一拍租约（在专用连接上，不走连接池：池子被占满时取连接要等，等得比租约还长）。按库的时钟续、按库的时钟核对（复核
|
||||
R-b7）：库里自己的租约已经过期（或者登记不见了），别的进程可能已经接手，不再续这个身份——隔离（cluster/fence.py），换身份
|
||||
重新登记。本地的两只钟说续晚了、或者两只钟差出一个租约（墙钟跳了），同样隔离。交回续没续上。"""
|
||||
重新登记。计入休眠的钟说续晚了，同样隔离；两只钟差出一个租约（墙钟跳了）只记一条日志，以库的核对为准（复核 R-b14）。
|
||||
交回续没续上。"""
|
||||
from app.cluster import fence
|
||||
|
||||
if fence.stale():
|
||||
fence.trip("续租约的时候发现已经续晚了")
|
||||
return False
|
||||
if fence.jumped():
|
||||
fence.trip("续租约的时候发现本机的墙钟和单调钟差出一个租约以上（墙钟跳了，或者休眠没计进单调钟）")
|
||||
return False
|
||||
# 两只钟对不上（墙钟跳了，或者单调钟没计入休眠）：不直接隔离，以下面这一句按库的时钟核对的结果为准（复核 R-b14）
|
||||
jumped = fence.jumped()
|
||||
try:
|
||||
conn = await self._lease_conn_open()
|
||||
renewed = await asyncio.wait_for(conn.fetchval(
|
||||
@@ -159,8 +159,12 @@ class _Service:
|
||||
self._lease_conn = None
|
||||
return False
|
||||
if renewed is None:
|
||||
fence.trip("按库的时钟本进程的租约已经过期" if expired else "本进程的登记不见了（被当成不在了）")
|
||||
why = "按库的时钟本进程的租约已经过期" if expired else "本进程的登记不见了（被当成不在了）"
|
||||
fence.trip(why + ("；本机的墙钟和单调钟也差出了一个租约以上" if jumped else ""))
|
||||
return False
|
||||
if jumped:
|
||||
logger.warning("本机的墙钟和单调钟差出一个租约以上（墙钟跳了：NTP 校时、手动改时间、虚拟机对时），按库的时钟本进程的租约"
|
||||
"还没过期：照常续上，不隔离")
|
||||
fence.renewed()
|
||||
return True
|
||||
|
||||
|
||||
@@ -7,6 +7,7 @@ import asyncio
|
||||
import os
|
||||
import subprocess
|
||||
import sys
|
||||
import time
|
||||
from datetime import timedelta
|
||||
from pathlib import Path
|
||||
|
||||
@@ -412,25 +413,34 @@ async def test_leadership_is_handed_over_when_the_lease_runs_out():
|
||||
assert (await leader_mod.current("test-lead"))["owner"] is None
|
||||
|
||||
|
||||
def test_the_fence_counts_what_the_monotonic_clock_missed(monkeypatch):
|
||||
"""复核 R-b7：单调钟不走（休眠不计、虚拟机挂起）而墙钟过了两个租约——照样算续晚了；两只钟差出一个租约算跳了。"""
|
||||
def test_the_fence_judges_by_the_sleep_aware_clock_and_falls_back_to_the_wall(monkeypatch):
|
||||
"""复核 R-b7、R-b14：取得到计入休眠的钟时只看它——它走过两个租约（合盖、挂起）算续晚了，墙钟往前、往后跳一小时不算（只算
|
||||
两只钟对不上，jumped）；取不到计入休眠的钟时，monotonic 和墙钟里走得多的那只说了算（墙钟当后备）。"""
|
||||
from app.cluster import fence
|
||||
|
||||
clocks = {"mono": 1000.0, "wall": 5000.0}
|
||||
monkeypatch.setattr(fence, "_clock", lambda: clocks["mono"])
|
||||
monkeypatch.setattr(fence, "_wall", lambda: clocks["wall"])
|
||||
monkeypatch.setattr(fence, "SLEEP_AWARE", True)
|
||||
fence.activate()
|
||||
try:
|
||||
assert not fence.stale() and not fence.jumped()
|
||||
clocks["wall"] += 2 * workers.lease_s() # 合盖两个租约：单调钟没计，墙钟走了
|
||||
assert fence.stale() and fence.jumped()
|
||||
for jump in (3600.0, -3600.0): # NTP 一步校时、手动改时间
|
||||
fence.renewed()
|
||||
clocks["wall"] += jump
|
||||
assert not fence.stale() and fence.jumped(), "只是墙钟跳了：不算续晚了"
|
||||
fence.renewed()
|
||||
clocks["mono"] += 2 * workers.lease_s() # 反过来：墙钟被往回拨了，单调钟照走
|
||||
assert fence.stale() and fence.jumped()
|
||||
clocks["mono"] += 2 * workers.lease_s() # 计入休眠的钟走过两个租约（合盖）
|
||||
assert fence.stale()
|
||||
fence.renewed()
|
||||
clocks["mono"] += workers.lease_s() / 2
|
||||
clocks["wall"] += workers.lease_s() / 2
|
||||
assert not fence.stale() and not fence.jumped()
|
||||
|
||||
monkeypatch.setattr(fence, "SLEEP_AWARE", False) # 取不到计入休眠的钟：monotonic 不计休眠，墙钟当后备
|
||||
fence.renewed()
|
||||
clocks["wall"] += 2 * workers.lease_s()
|
||||
assert fence.stale()
|
||||
finally:
|
||||
fence.deactivate()
|
||||
|
||||
@@ -464,6 +474,14 @@ async def test_leases_use_the_database_clock_not_the_local_wall_clock(monkeypatc
|
||||
before = fence.generation()
|
||||
try:
|
||||
assert await service.renew(), "按库的时钟没过期：续上"
|
||||
monkeypatch.setattr(fence, "SLEEP_AWARE", True)
|
||||
for jump in (3600.0, -3600.0): # 单进程、墙钟往前往后跳一小时（复核 R-b14）
|
||||
monkeypatch.setattr(fence, "_wall", lambda jump=jump: time.time() + jump)
|
||||
assert not fence.stale() and fence.jumped()
|
||||
fence.check()
|
||||
assert await service.renew() and fence.generation() == before, "库的时钟说租约还在：续上、不隔离"
|
||||
assert not fence.jumped(), "续上之后两只钟重新对齐"
|
||||
monkeypatch.setattr(fence, "_wall", time.time)
|
||||
async with SessionLocal() as session:
|
||||
row = await session.get(ClusterWorker, identity.owner())
|
||||
row.lease_until = real_utcnow() - timedelta(seconds=1) # 库的时钟已经过了它的租约，本机的两只钟都没觉得
|
||||
|
||||
@@ -149,6 +149,27 @@ async def test_a_run_continued_elsewhere_finishes_once_and_the_stalled_side_stay
|
||||
await _no_repeated_seq(run_id)
|
||||
|
||||
|
||||
@pytest.mark.parametrize("jump", [3600.0, -3600.0])
|
||||
async def test_a_wall_clock_jump_does_not_fence_a_running_run(slow, monkeypatch, jump):
|
||||
"""复核 R-b14：执行到一半本机墙钟往前 / 往后跳一小时（NTP 一步校时、手动改时间）——计入休眠的钟没问题，不隔离，运行照常跑完。"""
|
||||
import time as _time
|
||||
|
||||
monkeypatch.setattr(fence, "SLEEP_AWARE", True)
|
||||
run = await run_manager.start(graph=GRAPH, input_payload={"question": "q"})
|
||||
await asyncio.wait_for(slow["entered"].wait(), timeout=15)
|
||||
before = fence.generation()
|
||||
monkeypatch.setattr(fence, "_wall", lambda: _time.time() + jump)
|
||||
slow["gate"].set()
|
||||
for _ in range(300):
|
||||
async with SessionLocal() as session:
|
||||
row = await session.get(Run, run.id)
|
||||
if row.status in ("succeeded", "failed", "interrupted") and not run_manager.is_active(run.id):
|
||||
break
|
||||
await asyncio.sleep(0.05)
|
||||
assert row.status == "succeeded", row.error
|
||||
assert fence.generation() == before, "没有隔离"
|
||||
|
||||
|
||||
async def test_the_final_write_checks_the_owner():
|
||||
async with SessionLocal() as session:
|
||||
run = Run(status="running", owner="peer-host:1:abcd", graph={}, input={})
|
||||
|
||||
Reference in New Issue
Block a user