Critical Section
查看 Markdown当同一个 Flow 的多个并行分支必须轮流执行一段由多个 Step 组成的流程时，使用 Critical Section。同一时刻只能有一个分支处在这段流程中。
例如多位审核人编辑同一份合同。每次编辑需要三个 Step：第一个拿到编辑权，第二个修改合同，第三个保存并把编辑权交给下一位。两次编辑绝不能交错。
为什么不用 Attribute lock
Attribute lock（例如 ExecuteLockAttributes）只覆盖一个 Step 方法。它不覆盖 WaitFor，也不能跨越多个 Step。受保护的工作能放进一次 Execute 时，用 lock。这段流程跨越多个 Step 时，用 token。
工作方式
一个专用 Channel 里恰好只有一条消息，这条消息就是 token。
- 在任何分支进入之前，由一个 Step 发布一次 token。
- 一个交接 Step 用 ForOne 等待 token。一条消息只满足一个等待，所以只有一个分支能拿到它，其余分支继续等待。
- 交接 Step 不做任何受保护的工作。它的 Execute 只转到第一个做受保护工作的 Step。
- 这段流程的最后一个 Step 把 token 发布回去。某个正在等待的分支消费它并进入。
StartContractEdits.execute(edits)
ContractEditToken.publish("contract")
goToMany(for each edit: EnterContractEdit(edit), FinishContractEdits(count))
EnterContractEdit.waitFor -> Until(ContractEditToken.ForOne())
EnterContractEdit.execute -> goTo(EditContract) // handoff only, no protected work
EditContract.execute -> edit the contract; goTo(SaveContractEdit)
SaveContractEdit.execute -> save the edit; ContractEditToken.publish("contract"); deadEnd()
ReleaseContractEditToken.execute -> ContractEditToken.publish("contract"); deadEnd()
// the ExecuteFailure route of EnterContractEdit, EditContract, and SaveContractEdit
Dex 不保证下一个拿到 token 的是哪个等待中的分支。不要依赖进入顺序。
每个出口都要归还 token
token 只是一条普通消息。如果某个分支离开这段流程时没有把它发布回去，其他所有分支会永远等待。列出离开这段流程的每一种方式，并在每一种方式上发布 token。
- 正常退出。 这段流程的最后一个 Step 发布 token。
- Execute 失败。 用 ExecuteFailure 把失败路由到一个发布 token 的 Step。这段流程的每个 Step 都要配置这条路由。在示例中，EnterContractEdit、EditContract 和 SaveContractEdit 都会进入 ReleaseContractEditToken。
- WaitFor 失败。 如果这段流程中后面的某个 Step 有自己的 WaitFor，给它配置继续执行 Execute 的 WaitForFailure 策略。那个 Execute 看到失败后发布 token。
- 提前退出。 用业务决定提前离开这段流程的 Execute，要在同一个 Execute 中发布 token。
- 交接 Step 被取消。 取消交接 Step 是安全的，因为它不做受保护的工作。如果取消发生在它的等待已经消费 token、而 Dex 还没保存它的 Execute 结果时，Dex 会把 token 放回 Channel 的最前面，另一个分支随后进入。
- 在段内被取消。 绝不要取消做受保护工作的 Step。被取消的 Step 不会发布任何消息，它的失败路由也不会运行。发起取消的 Step 或 RPC 也不能替它发布 token。Dex 要到保存发起取消的那个结果时，才确定取消哪些 Step。那时被取消的 Step 可能已经发布了 token，再发一个 token 会让两个分支同时进入这段流程。要提前停止一个分支，给它发送一个它会检查的停止请求，例如由 RPC 设置的 Attribute。看到请求的 Step 通过发布 token 的提前退出离开这段流程。
- 关闭 Flow。 ForceComplete、ForceFail、Flow timeout 和 Client 的 StopFlow 会结束所有分支，不再有分支等待 token。
Dex 把一个方法的 Channel 发布和它的结果一起保存，要么都保存，要么都不保存。失败的 attempt 不会发布任何消息，所以重试不会产生第二个 token。
Dex 无法阻止被取消的方法继续调用外部系统。Dex 归还 token 之后，被取消的 Execute 仍可能继续运行并写入。这是绝不要取消做受保护工作的 Step 的另一个原因。
规则
- 只发布一个 token，在启动各分支的 Step 中发布，或者更早。多一个 token，就会有两个分支同时进入这段流程。
- token Channel 只用于 token。
- 让这段流程尽量短。一个分支在段内时，其余分支都在等待。
- token 的值无关紧要。受保护的状态放在 Attribute 中。
示例
ContractEditFlow 以一组编辑请求启动，每位审核人一个。StartContractEdits 发布 token，为每个编辑请求启动一个 EnterContractEdit，并启动汇合 Step FinishContractEdits。EnterContractEdit 交接给 EditContract，由它记录编辑开始的时间。汇合 Step 用每次编辑的记录完成 Flow，这些编辑的时间段互不重叠。
编辑请求上的标志用来模拟其他出口。交接失败，以及 SaveContractEdit 失败两次的编辑，都会进入 ReleaseContractEditToken。被撤回的编辑从 EditContract 提前离开。每种情况都会归还 token，其他审核人仍能轮到。
CONTRACT_EDIT_TOKEN = Channel("contractEditToken", str)
CONTRACT_EDIT_FINISHED = Channel("contractEditFinished", ContractEditRecord)
CONTRACT_EDIT_RETRY = RetryPolicy(
initial_interval=timedelta(seconds=1), maximum_attempts=2
)
class EnterContractEdit(Step[ContractEdit]):
"""Takes the token and hands off; it does no protected work, so canceling it is safe."""
def get_step_options(self) -> StepOptions:
return StepOptions(execute_retry=CONTRACT_EDIT_RETRY).on_execute_failure_proceed_to(
ReleaseContractEditToken
)
def wait_for(self, context: Context, edit: ContractEdit) -> Wait:
return Wait.until(CONTRACT_EDIT_TOKEN.for_one())
def execute(self, context: Context, edit: ContractEdit) -> StepDecision:
if edit.should_fail_handoff:
raise ContractEditError("contract handoff failed")
return go_to(EditContract, edit)
class EditContract(Step[ContractEdit]):
"""The first Step that works on the contract. Never cancel it."""
def get_step_options(self) -> StepOptions:
return StepOptions(execute_retry=CONTRACT_EDIT_RETRY).on_execute_failure_proceed_to(
ReleaseContractEditToken
例子: examples/python/dex_examples/patterns/critical-section/contract_edit_flow.py
反模式
- 用一个布尔 Attribute（例如 "isEditing"）让 Step 检查并设置。没有任何 Step 能等待它，两个分支也可能同时看到它未被设置。
- 在旧 token 可能仍然存在时，发布新 token 来“找回”丢失的 token。两个 token 会让两个分支同时进入这段流程。