diff --git a/.gitignore b/.gitignore index c4ecd16..b656058 100644 --- a/.gitignore +++ b/.gitignore @@ -23,3 +23,7 @@ pytest-of-*/ # 被强杀(例如 `timeout` 截断跑批)时会留下残骸 chongming-store-*/ chongming-jsonstore-*/ + +# 本机环境里 tempfile 的临时目录解析到了项目根,进程被强杀(如 timeout 截断跑批) +# 时会留下孤儿临时文件(正常结束不会) +tmp* diff --git a/README.md b/README.md index 0f5b97a..d3dfe45 100644 --- a/README.md +++ b/README.md @@ -857,6 +857,7 @@ outcome = run_task( | 文档 | 内容 | |---|---| +| [最佳实践](docs/best-practices.md) | **照什么做生产上不会踩坑**:选型、任务写法、并发与资源、排障、反模式清单、上线检查清单 | | [用户指南](docs/user-guide.md) | 概念、任务编写、控制语义、超时、执行器、FAQ | | [命令行控制台](docs/cli.md) | 子命令、退出码、任务模块约定、完整走查(真实输出) | | [持久层实现指南](docs/store-guide.md) | **换掉默认持久层要实现哪些接口**、逐方法语义、并发要求、陷阱清单 | diff --git a/docs/best-practices.md b/docs/best-practices.md new file mode 100644 index 0000000..6438847 --- /dev/null +++ b/docs/best-practices.md @@ -0,0 +1,364 @@ +# 最佳实践 + +**这份文档只回答一个问题:照什么做,生产上不会踩坑。** + +它不是入门教程(那是 [用户指南](user-guide.md)),也不是接口规格(那是 +[持久层实现指南](store-guide.md) 与 [设计文档](task-control-design.md))。这里给的是 +**默认答案**——绝大多数场景下该选什么、任务该怎么写、哪些坑已经有人踩过了。 + +一条贯穿全文的原则:**这个库是 at-least-once 的**。停止、超时、进程崩溃之后,任务 +可能已经产生过一部分副作用。所有"最佳实践"最终都落回同一句话——**把任务写成幂等的**。 + +--- + +## 0. 一分钟版 + +```python +from chongming_task import ( + BucketPolicy, ConcurrencyPolicy, Controller, ExponentialBackoff, RetryPolicy, + SqliteTaskStore, TaskContext, task, +) + +NUM_BUCKETS = 64 # 部署级约定:所有节点必须一致 + + +@task(tags={"sync", "db"}, retry=RetryPolicy(max_attempts=3, backoff=ExponentialBackoff())) +def sync_tenant(ctx: TaskContext, tenant: str) -> dict[str, object]: + """同步一个租户的数据。""" + pages = list_pages(tenant) # 你的业务代码 + for index, page in enumerate(pages, start=1): + ctx.checkpoint() # ← 必须 + ctx.progress(index, total=len(pages), message=tenant) # 进度走旁路,不碰版本号 + apply_page(tenant, page) # ← 必须幂等 + return {"tenant": tenant, "pages": len(pages)} + + +controller = Controller( + store=SqliteTaskStore("tasks.db"), # 跨机换成 PostgresTaskStore(dsn) + # owner 默认就是 f"{主机名}:{进程号}",要区分同一台机上的多个节点再显式给。 + bucket_policy=BucketPolicy(num_buckets=NUM_BUCKETS), + owned_buckets=range(0, 32), # 本节点负责的桶(连续分区) + concurrency=ConcurrencyPolicy(max_parallel=8, limits={"db": 4}), + max_parallel_per_bucket=1, # 同一租户串行 +) +controller.register_all(sync_tenant) +controller.reconcile() # 启动时清理上次的僵尸任务 +``` + +五条不可省略的: + +| 必须 | 为什么 | +|---|---| +| 循环里调 `ctx.checkpoint()` | 不调 = 停不掉、也会被 `idle` 判成卡死 | +| 任务**幂等** | 超时/停止/崩溃之后可能重跑,副作用会再来一次 | +| 返回值可 JSON 序列化(跨进程再加"可 pickle") | 返回值要落库;不满足会在返回时响亮失败 | +| 进程执行器的目标函数**在模块里** | spawn 的子进程靠"模块 + 函数名"找回它 | +| `NUM_BUCKETS` 各节点一致 | 不一致会让同一个键落到不同的桶,且**不报错** | + +--- + +## 1. 按规模选型 + +| 场景 | 存储 | 执行器 | 起点 | +|---|---|---|---| +| 脚本里跑一个任务 | 不要存储 | 线程(默认) | `run_task(fn, *args)` | +| 单机常驻服务 | `SqliteTaskStore("tasks.db")` | 线程 | 本文 §0 | +| 多机 / 多进程共用一套状态 | `PostgresTaskStore(dsn)` | 线程(按需进程) | [示例 11](../examples/11_best_practice.py) | + +**内存存储只用于测试。** `InMemoryTaskStore` 不共享、不持久化——用它跑生产等于把 +"跨进程可见"和"重启恢复"这两件事一起放弃。 + +**能单机就别多机。** 多节点带来的是运维成本(桶划分、告警巡检),不是性能魔法; +单机的 `max_parallel` 通常先到瓶颈,先把它调明白。 + +--- + +## 2. 任务怎么写 + +### 2.1 签名与位置 + +```python +@task(tags={"db"}) +def sync_tenant(ctx: TaskContext, tenant: str, *, batches: int = 24) -> dict[str, object]: + ... +``` + +`ctx` 由执行器**注入**为第一个位置参数,提交时不要自己传。需要跨进程执行时,任务函数 +必须是**模块级**的(不能是闭包、`lambda`、局部函数)——这一点在进程执行器下是硬约束, +违反时提交那一刻就报错。 + +### 2.2 检查点频率要与 `idle` 对齐 + +`idle` 超时衡量的是**两次 `checkpoint()` 之间的间隔**,不是"CPU 有没有在干活"。 + +```python +controller.submit("calc", timeout=Timeout(idle=30)) # 30 秒没有检查点 = 判卡死 +``` + +- 任务每批最多 2 秒 → `idle=30` 很宽裕; +- 任务有一段"一口气算 5 分钟"的循环 → 要么**在里面塞 checkpoint**,要么把 `idle` + 调到 6 分钟以上。漏了这一步的症状是:任务跑得好好的,被判 `timeout_idle`。 + +**`ctx.progress()` 不刷新心跳。** 这是刻意的:进度是"到哪了",心跳是"还在推进"。 +只报进度不调检查点的任务会被如实判成卡死——那正是 `idle` 要抓的东西。 + +### 2.3 幂等:把"重跑一次"当常态 + +会重跑的情况不止失败重试: + +| 情况 | 之后会发生什么 | +|---|---| +| `stop()` 落在任务收尾窗口里 | 状态 `stopped`,但副作用可能已发生一半 | +| 优雅窗口用尽(线程) | 状态 `lost`,**任务可能还在后台跑**,随后可能重试 | +| 优雅窗口用尽(进程) | 子进程被 `kill()`,已写入的东西不会回滚 | +| 进程崩溃 / 被 OOM 杀掉 | 同上,且状态是 `lost`(结果未知) | + +所以:**写操作要能重放**——`INSERT ... ON CONFLICT`、先查后写、写临时路径再原子改名、 +带幂等键调用外部 API。不要指望"库会保证只跑一次",它保证不了。 + +### 2.4 返回值与参数 + +```python +@task +def extract(ctx) -> dict[str, object]: + return {"rows": 3, "path": "s3://bucket/x.parquet"} # ✅ 传指针 + # return {"frame": df} # ❌ DataFrame 不可 JSON 序列化 +``` + +返回值会写进存储(`outcome` 列): + +- **所有执行器**:必须可 JSON 序列化,否则任务在返回时失败(`result_invalid`); +- **进程执行器**:还要能 pickle(先跨进程传回来,再落库)。两个条件都要满足:`pathlib.Path` + 能被 pickle 但**不能** JSON 化,所以它做返回值一样会被拒。 + +参数则相反:**线程执行器下随便传**(`Path`、连接对象、句柄都行),进程执行器下必须 +可 pickle。所以"参数里塞一个打开的数据库连接"这种写法会锁死执行器的选择。 + +### 2.5 别把停止信号吃掉 + +```python +try: + while work(): + ctx.checkpoint() +except TaskCancelled: + raise # ← 必须放行:取消/停止是控制流,不是错误 +except Exception as exc: # 业务异常照常处理 + log(exc) +``` + +`TaskCancelled` 继承 `BaseException` 就是为了防止 `except Exception` 把它吞掉。它一旦 +被吞,任务就**永远停不下来**。 + +--- + +## 3. 并发与资源 + +```python +controller = Controller( + concurrency=ConcurrencyPolicy( + max_parallel=20, # 同时执行上限;超出排队(不是拒绝) + limits={"db": 4, "gpu": 1}, # 具名信号量:按任务标签记账 + priority=0, # 默认优先级 + ), + max_parallel_per_bucket=1, # 每桶各自限流 +) +controller.submit("sync", "tenant-A", tags=["db"], bucket_key="tenant-A") +controller.submit("report", priority=10) # 急件先出队 +``` + +三条经验: + +1. **保护外部资源用 `limits`,不要靠调小 `max_parallel`。** 混合型负载里,为了保住 + 4 个数据库连接把全局并发压到 4,等于让纯 CPU 的任务跟着一起挨饿。 +2. **`max_parallel_per_bucket=1` 是"同一份数据不自我竞争"的最省事做法**:同一个租户 + (同一个 `bucket_key`)串行,不同租户仍然并行。 +3. **名额与优先级都是本进程的账**,不做跨进程协调:多节点各自配自己的份额 + (例如"每个节点最多 4 个数据库连接")。`limits` 统计的是活跃任务(含被暂停的), + 因为暂停不释放执行槽。 + +`concurrency=` 与 `max_parallel=` **只能给一个**(同时给会报错)——两份配置谁说了算 +必须明确。 + +--- + +## 4. 执行器怎么选 + +| | 线程(默认) | 进程 | +|---|---|---| +| 共享内存 / 连接池 / 缓存 | ✅ 直接复用 | ❌ 各自一份 | +| 闭包、`lambda`、局部函数 | ✅ | ❌ 必须模块级 + 可 pickle | +| 参数的额外要求 | 无 | 必须可 pickle | +| 停止 / 暂停 | 即时 | 约 20ms 轮询延迟 | +| 不响应停止的任务 | 只能放弃,记 `lost`(未知) | `terminate()` → `kill()`,**真的停掉** | +| CPU 密集 | 受 GIL 限制 | 真并行 | +| 原生扩展段错误 | 带走整个监督进程 | 只带走子进程 | + +**默认线程。** 只在三种场景换 `ProcessExecutor()`:需要"停得住"的保证(例:任务里跑 +别人写的、可能不响应取消的代码)、CPU 密集、或跑第三方原生扩展。 + +```python +controller = Controller(executor=ProcessExecutor()) # 整批 +outcome = run_task(fn, executor=ProcessExecutor()) # 单个 +``` + +CLI 也有:`chongming-task --db tasks.db run --tasks myapp.tasks heavy --executor process`。 + +进程执行器的三条代价(可 pickle、模块级目标 + `__main__` 守卫、轮询延迟)见 +[用户指南 §7.4](user-guide.md#74-进程执行器硬性停止--绕开-gil--崩溃隔离)。注意:**别为了 +"更安全"无脑上进程**——它的约束会渗透到任务签名与参数里,撤回来的成本很高。 + +--- + +## 5. 持久化与重启 + +### 5.1 心智模型:内存是缓存,存储是事实 + +`Controller(store=...)` 之后:每次状态迁移**写穿**到存储,查询**读穿**存储(因此能看到 +别的进程的任务),控制动作对不属于本进程的任务**指令化**。 + +存储出问题**不会让任务失败**(可观测性坏了不是任务的错),但也不会静默: + +```python +controller.persist_failures # 写穿失败的历史(去重、有上限) +controller.desynced_tasks # 检测到"外部写入者"而停止写穿的任务 +``` + +上线后把这两个属性接进你的告警——它们是"存储里的事实与内存里的状态分叉了"的唯一信号。 + +### 5.2 启动时跑一次 `reconcile()` + +```python +report = controller.reconcile() # 先看一眼:reconcile(dry_run=True) +if report.lost: + log.warning("上次进程留下的 %d 个任务判为丢失:%s", len(report.lost), report.lost) +``` + +判据是**监督者租约**("还有人在管它吗"),不是任务心跳——所以长时间不调 checkpoint 的 +健康任务不会被误杀。`lease_ttl`(默认 30 秒)决定崩溃多久之后被发现。 + +**它不会重新执行任何任务。** 存储里记的是状态,不是配方:任务的函数与参数从未落库。 +重跑必须由调用方重新 `submit`,而那需要调用方知道参数——这是设计边界,不是遗漏。 + +### 5.3 规模上来之后 + +- `prune(keep_terminal=N)` 只清内存里的终态记录,存储里的历史仍在; +- `transitions` 只增不减,长期跑要按月归档或分区(数据库侧的事,库不代劳); +- 跨机把 `SqliteTaskStore("tasks.db")` 换成 `PostgresTaskStore(dsn)` 即可,业务代码 + 一行不用改;只读运维账号也能连(检测到表已存在就不做 DDL)。 + +--- + +## 6. 多节点与桶划分 + +**桶是节点之间的划分单位**:每个节点自治(自己的任务定义与触发方式),但共用一套存储 +与控制层,靠 `owned_buckets` 划界。 + +```python +def buckets_for_node(index: int, total: int, num_buckets: int) -> range: + """连续分区——不是取模。取模会让"扩一个节点"把所有桶都重新洗一遍。""" + per_node = num_buckets // total + start = index * per_node + stop = num_buckets if index == total - 1 else start + per_node + return range(start, stop) +``` + +三条铁律: + +1. **`num_buckets` 是部署级约定,各节点必须一致**(CLI 也要用 `--num-buckets` 显式给), + 不一致不会报错,只会让任务落到没人拥有的桶里; +2. **提交时校验是保护,不是障碍**:落在别人桶里的任务不会有节点执行,它会在 `queued` + 上静默堆积——所以 `submit` 会直接报错; +3. **`reconcile()` 只看自己的桶**,跨节点停止走指令通道(`stop()` 返回 `False` 表示 + "已下发、未确认",看结果要查 `status`)。 + +**没有自动告警**:两个节点配了同一个桶(会重复执行)、或某个桶没人管(任务静默堆积), +都不会报错。这是已知缺口,用你自己的巡检兜住:定期 `buckets` 视图比对节点配置。 + +--- + +## 7. 运维 + +### 7.1 排障三件套 + +```bash +chongming-task --db tasks.db ps --all # 现在都在什么状态 +chongming-task --db tasks.db status # 详情 + 迁移历史(谁在什么时候改的) +chongming-task --db tasks.db buckets # 哪个桶在堆积(星号是本节点负责的) +``` + +任务卡住时先看快照里的 `pid`:进程执行器下它就是那个子进程的进程号,`ps` 能直接对上号。 + +### 7.2 失败类型决定下一步 + +| `failure.kind` | 含义 | 建议 | +|---|---|---| +| `exception` | 任务自己抛异常 | 看 traceback 修逻辑 | +| `timeout_start` / `timeout_attempt` / `timeout_idle` | 三种超时(启动 / 单次尝试 / 静默) | 调阈值或加检查点;`idle` 多半是缺 checkpoint | +| `nonzero_exit` | 子进程非零退出或崩溃(带 `exit_code`) | 查子进程 stderr;原生扩展崩溃就该用进程执行器隔离 | +| `result_invalid` | 返回值不满足持久化/跨进程契约 | 改返回值为可序列化/可 pickle,**重试无用** | +| `cancelled` | 停止或取消(不是错误) | 不用管 | +| `lost` | 结果**未知**(载体被放弃) | 重跑前先人工核对副作用:可能已经生效过一次 | + +### 7.3 优雅退出 + +```python +try: + controller.wait_all(timeout=60) +finally: + controller.close(timeout=30, close_store=True) # 关闸 → 停任务 → 补写穿 → 关存储 +``` + +`close()` 会先把排队中的任务"启动后立刻停止"(不执行任务体),再等收尾,最后把因瞬时 +故障积压的写穿补一次。配 systemd 时用 `SIGTERM` → 这个路径;**别用 `kill -9`**, +那会留下一堆 `running` 僵尸任务,下次启动靠 `reconcile()` 收拾。 + +--- + +## 8. 反模式清单 + +| 反模式 | 症状 | 正确做法 | +|---|---|---| +| 任务里不调 `ctx.checkpoint()` | 停不掉、被判 `timeout_idle` | 每个循环都调;频率与 `idle` 对齐 | +| 用 `except Exception` 包住循环 | 停止请求被吞,任务永远停不下来 | `except TaskCancelled: raise` 放行 | +| 任务写成非幂等 | 重跑后数据重复/重复扣款 | 幂等键、`ON CONFLICT`、原子替换 | +| 返回值塞大对象 / DataFrame | `result_invalid` 失败 | 传指针(对象存储 key、表名 + 主键) | +| 参数里传数据库连接 | 换进程执行器时全部报错 | 任务内部自建连接池,用 `limits` 限流 | +| 全局并发设为 1 来保护数据库 | 所有任务串行,吞吐崩掉 | 用 `limits={"db": N}` 只限那一类任务 | +| 忘了 `max_parallel_per_bucket` | 同一租户的多个任务互相抢锁/打架 | 同租户串行:`max_parallel_per_bucket=1` | +| 用内置 `hash()` 分桶 | 跨进程桶位不一致,且**静默** | 用 `BucketPolicy`(`blake2b`,跨进程稳定) | +| 节点桶数不一致 | 任务落到没人拥有的桶,静默堆积 | `num_buckets` 当部署配置管,写进巡检 | +| 多个进程共用 SQLite 文件 | 网络盘上锁不可靠 | 同机用 SQLite 文件、跨机换 Postgres | +| `Controller` 用完不 `close()` | 文件句柄泄漏,Windows 下数据库文件删不掉 | `close(close_store=True)` | +| 忘了启动时 `reconcile()` | 看板上永远挂着几个假的"运行中" | 启动第一件事就是它 | +| 把 `lost` 当"没跑过" | 放心重跑 → 副作用来两次 | `lost` = 结果未知,先核对再决定 | +| 无脑上进程执行器 | 任务签名/参数被 pickle 约束锁死 | 需要那三种能力时才用(§4) | + +--- + +## 9. 上线检查清单 + +- [ ] 存储选好并接上(生产用 SQLite 文件或 Postgres,**不用内存**) +- [ ] 启动时跑一次 `reconcile()`,`lost` 结果进日志/告警 +- [ ] 退出路径是 `close(close_store=True)`,systemd 用 `SIGTERM` +- [ ] 每个任务的循环里都有 `ctx.checkpoint()`,且与 `idle` 阈值匹配 +- [ ] 任务幂等(写操作可重放),或明确接受重复副作用 +- [ ] 返回值可 JSON 序列化(跨进程还需可 pickle) +- [ ] `max_parallel` 定过压测;外部资源用 `limits` 单独限流 +- [ ] 需要同租户串行的地方配了 `max_parallel_per_bucket` +- [ ] `num_buckets` 与各节点的 `owned_buckets` 对得上(连续分区、无重叠、无空洞) +- [ ] `persist_failures` / `desynced_tasks` 接进了告警 +- [ ] 桶重复/空洞有自己的巡检(库不会替你报) +- [ ] 执行器选择有理由:默认线程;进程只在需要强杀/GIL/隔离时用 + +--- + +## 相关文档 + +| 想要 | 去哪 | +|---|---| +| 完整概念与 API | [用户指南](user-guide.md) | +| 命令行控制台 | [命令行](cli.md) | +| 换/写持久层 | [持久层实现指南](store-guide.md) | +| 多节点架构、边界与风险 | [分布式设计说明](distributed-design.md) | +| 状态模型、转移表、取舍理由 | [设计文档](task-control-design.md) | +| 可直接运行的代码 | [examples/](../examples/)(11 号是集成主模板,12 号讲进程执行器) |