Python Executor

​

职责与边界

• 通过一条 UDS 长连接接收 ExecuteCall / ResumeCall / CancelCall / HeartbeatAck，发出 CallWaiting / CallCompleted / CallFailed / Heartbeat。
• 按源码 digest 编译并缓存函数模块，按 task 隔离模块命名空间，缺源码时按 digest 向 Worker 回拉。
• 用 greenlet 让同步风格的用户代码在 SDK 调用点挂起、在 ResumeCall 后恢复。
• 在安全点执行取消与 deadline / budget 检查，并用 SIGALRM 定时器抢占纯 CPU 循环。
• 每个心跳周期上报进程指标（RSS、CPU 利用率、累计 CPU 秒、各状态 call 计数）。

• 决定 call 派发顺序或选择哪个 Executor（Worker dispatcher 的事）。
• 直接访问 BlockDB / 节点 RPC / Function Code View。SDK 的所有 IO 都被拦截后经 Worker 出去。
• 持久化或跨进程恢复任何 call。进程重启后一切上下文重建。

• 用户代码只在主线程的 greenlet 里跑，每次只跑一个；UDS reader / writer / heartbeat 是独立线程，不切换 greenlet。
• 一个 ExecContext 同一时刻最多一个 pending requestId；ResumeCall 必须同时匹配 callId、状态对应的 resumeKind 和 requestId，否则按 stale 丢弃。
• 同一 callId 再次 ExecuteCall 视为新 attempt：旧 greenlet 标 stale 并终止、不再发终态；新 attempt 用新 greenlet。
• 每个被丢弃的 context 恰好发出一条终态消息（CallCompleted 或 CallFailed），attempt 替换除外。
• 编译产物按 digest 跨 task 共享；exec 出来的模块命名空间按 task_id 隔离，task 最后一个 call 结束时释放。

​

代码位置

​

核心类型与接口

class Provider(Protocol):
    def read_req(self, backend: str, operation: str, params: Mapping[str, Any], *, cache_key: Optional[str] = None) -> Any: ...
    def write_req(self, backend: str, operation: str, params: Mapping[str, Any]) -> Any: ...
    def capability_req(self, backend: str, operation: str, params: Mapping[str, Any], *, cache_key: str = "") -> Any: ...
    def subcall(self, function_id: str, args: Any) -> Any: ...
    def sleep(self, seconds: float) -> None: ...

​

用户函数编程模型

def _(p):
    child = function.call("adder", {"left": p["left"], "right": p["right"]})
    total = Decimal(str(child["sum"])) + Decimal("0.50")
    return {
        "child": child,
        "encoded": json.dumps({"sum": child["sum"]}, sort_keys=True, separators=(",", ":")),
        "total": str(total),
        "day": date(2026, 6, 10).isoformat(),
        "stamp": datetime(2026, 6, 10, 9, 30).isoformat(),
    }

from leafage import ChainState

def _(p):
    cs = ChainState(p["chain"], p["block_id"])
    code = cs.get_address_code(p["address"])
    bal = cs.get_address_balance(p["address"])
    return len(code) + bal

​

数据流 / 执行流程

​

状态与生命周期

• 恢复时状态不会先回到 RUNNABLE：resume_call 只写 resume_payload 并置 queued=True 入队，_perform_wait 从 switch() 返回后直接改成 RUNNING。等待中的 context 是否已排队由 queued 标志表达，不由状态表达。
• DONE 的 context 立即从 _contexts 移除，所以 inflight_count() 就是 len(_contexts)。
• WAITING_FUNCTION_CODE 用于按 digest 回拉源码。同一 digest 的并发 miss 在 SdkBridge._function_code_fetches 里做本地 singleflight，只有 leader 发 CallWaiting(waitKind=function_code)。

​

配置

​

扩展点

• 接一个新的外部 SDK 到 Worker IO 通道：写一个 bridge 模块（参考 blockdb_bridge.py / noderpc_bridge.py），把 SDK 的传输接缝替换成对 provider.read_req / write_req 的调用，在 ExecutorRuntime.__init__ 里安装；backend 名要与 Worker 侧 IO 子系统注册的 backend 对齐（见 IO 访问子系统）。若 SDK 是 gRPC unary，直接复用 BridgeChannel(provider_getter, backend=..., streaming_methods=frozenset())。
• 给用户代码新增 SDK 入口：只加 blockx_sdk 下的薄封装并让它调 get_provider()；不要在 SDK 里持有连接。受限模式还要同步 blockx_audit/tables.py 的白名单与 restricted_runtime facade。
• 新增 ResumeCall 错误类型：在 sdk_bridge.py 加 RemoteCallError 子类并扩展 _map_resume_error；backend 专属映射用 register_io_error_kinds。
• 新增心跳字段或消息类型：messages.py 加构造、wire_keys.py 加常量、runtime._handle_inbound_message_typed 加分支，并同步 api/uds/types.go。
• 改安全点或计费：所有过期判断经 ExecContext.deadline_expired_reason 与 raise_if_deadline_expired，不要在别处另写判断。

​

测试

uv sync --project python                                    # 首次或 uv.lock 变更后；Go E2E 也用这个 .venv
PYTHONPATH=python uv run --project python python -m unittest discover -s python/tests
PYTHONPATH=python uv run --project python python -m unittest python.tests.test_executor_protocol_contract   # 单模块

• Runtime：test_executor_runtime_scheduler.py、test_executor_runtime_termination.py、test_executor_runtime_attempt.py、test_executor_budget.py、test_executor_subcall.py、test_executor_oom_hardening.py、test_terminal_bounds.py。
• 协议：test_executor_protocol_contract.py（ExecuteCall 校验、入站控制消息、心跳字段）、test_uds_codec.py、test_uds_session.py、test_envelope_*.py。
• SDK hook：test_executor_sdk_hook.py、test_sdk_bridge_provider.py、test_sdk_bridge_capability.py、test_perform_wait_emit_failure.py、test_blockdb_bridge.py、test_noderpc_bridge.py、test_sleep_patch.py。
• 装载与隔离：test_module_registry.py、test_restricted_runtime.py、test_executor_audit_gate.py。
• 上下文与指标：test_exec_context_*.py、test_process_metrics_sampler.py、test_logging_setup.py、test_tracing_setup.py、test_debug_capture.py。
• 审计器：test_audit_*.py。

​

相关文档

• docs/specs/python-function-executor.md — Executor 进程内设计：调度、状态、attempt / requestId、取消与 deadline、资源边界（重点）。
• docs/specs/worker-executor-connection-and-python-sdk-hook.md §6–§8 — SDK hook 方案、SdkBridge 同步 contract、挂起恢复时序。
• docs/specs/python-executor-process-metrics.md — 心跳里进程指标的采样与降级。
• docs/specs/2026-07-23-execute-call-function-code-reference.md — 按 digest 回拉源码的协议。
• docs/specs/executor-sandbox-isolation.md — containerd 沙箱 spawn 模式。
• docs/specs/2026-07-30-python-executor-cpu-optimization.md — CPU 优化方案。
• docs/specs/python-test-organization.md — 单测组织标准。
• docs/specs/function-code-python-whitelist.md、docs/specs/function-code-audit-design.md — 用户代码白名单与审计。

• Call 执行子系统 — Worker 侧 dispatcher、executor adapter、子函数。
• 协议与接口 — UDS 帧格式与消息字段。
• Function Code View 与代码审计 — 源码来源与 blockx_audit。
• IO 访问子系统 — CallWaiting(waitKind=io) 在 Worker 侧的去向。
• 可观测性 — 日志字段、指标、tracing。