pyPKIPC

pyPKIPC 2 是用于父子进程通信的轻量事件运行时。父进程负责启动子进程，双方通过一条长期 gRPC 双向流收发任意 JSON 事件。

运行时不使用 stdin/stdout 传输，不定义应用层握手，也没有旧协议兼容代码。

安装

python -m pip install PKIPC

本地开发和打包依赖：

python -m pip install -e .[test,build]

子进程：Server

import threading

from pkipc import Server

server = Server()
stopped = threading.Event()


@server.on("solver.add")
def add(data):
    return {"result": data["a"] + data["b"]}


@server.on("worker.stop")
def stop(_data):
    stopped.set()


config = server.start()
server.send(
    "pkipc.log",
    {
        "level": "INFO",
        "message": f"worker started with {config}",
    },
)
while not stopped.wait(0.1) and not server.stop_requested():
    pass
server.close()

start() 建立 gRPC 流并返回父进程下发的配置。事件回调在单一接收线程中按顺序执行。

父进程：Client

import sys

from pkipc import Client

client = Client(
    sys.executable,
    args=("worker.py",),
    config={"value": 42},
)


@client.on("pkipc.log")
def log_received(data):
    print(data["level"], data["message"])


client.start()
result = client.request(
    "solver.add",
    {
        "a": 10,
        "b": 20,
    },
    timeout=5.0,
)
print(result)
client.send("worker.stop")
client.wait()
client.close()

父子两端使用完全相同的 send(event, data)、request(event, data, timeout) 和 on(event)。send() 是不等待结果的单向事件；request() 阻塞等待对应 handler 的返回值，请求 ID 和响应匹配完全由 Runtime 管理。框架不提供 send_data、send_log、on_data 或 on_log；业务层可以按需用普通函数做薄封装。

request() 不能在 @on handler 内调用，因为 handler 运行在单一接收线程中；Runtime 会直接抛出 RuntimeError，避免嵌套同步请求死锁。请求超时抛出 TimeoutError，断连时所有等待中的请求抛出 ConnectionError，远端 handler 异常则在调用端表现为 RuntimeError。

框架不定义自有异常类型：无效事件抛出 ValueError，断连抛出 ConnectionError，启动等待超时抛出 TimeoutError，子进程提前退出抛出 ChildProcessError，重复启动等生命周期错误抛出 RuntimeError。

运行模型

1. 父进程在 127.0.0.1:0 启动专属 gRPC Server。
2. 父进程生成随机 token，通过环境变量把 endpoint 和 token 传给子进程。
3. 子进程建立 Runtime.Run 双向流。
4. 父进程发送的第一帧固定为 pkipc.config，由 Server.start() 消费。
5. 配置完成后，父子进程通过同样的 Frame 双向收发任意命名事件。
6. 流结束即表示停止，错误由 gRPC status 表达。

传输层只有一个无类型消息和一个 RPC：

service Runtime {
  rpc Run(stream Frame) returns (stream Frame);
}

message Frame {
  bytes payload = 1;
}

Frame.payload 是 UTF-8 JSON，统一由 Runtime 层编码、校验和分发：

{"event":"pkipc.config","data":{"threads":8}}
{"event":"pkipc.control","data":{"action":"PAUSE"}}
{"event":"pkipc.data","data":{"progress":0.5}}
{"event":"pkipc.log","data":{"level":"INFO","message":"solver started"}}

同步请求增加 id，响应使用内部事件 pkipc.response 和 reply_to：

{"event":"solver.add","data":{"a":10,"b":20},"id":"request-id"}
{"event":"pkipc.response","data":{"result":30},"reply_to":"request-id"}

event 必须是非空字符串，data 必须存在且可以是任意合法 JSON。id、reply_to 和错误响应由 Runtime 独占管理。pkipc.config 由启动流程管理；pkipc.control、pkipc.data 和 pkipc.log 是经过校验的标准事件约定，但不绑定专用 Python API。自定义事件名会原样传输，不需要修改或重新生成 Protobuf。

协议定义在：

• pkipc/proto/pkipc.proto

Python 生成代码和 .proto 一起发布。C++ 实现只需要从同一份 .proto 生成传输 stub，并在 Runtime dispatcher 中实现相同的 JSON 信封约定。

完整示例

• examples/01_request_response/：同步请求响应，实现计算器
• examples/02_commands/：单向命令，实现暂停、恢复和停止
• examples/03_progress/：长任务持续推送进度，最后返回结果
• examples/04_reverse_request/：Server 反向请求 Client
• examples/05_errors/：请求超时、连接断开和远端 handler 异常

每个目录包含一组 client.py 和 server.py。只运行 client.py，Client 会负责启动 Server 子进程：

python examples/01_request_response/client.py
python examples/02_commands/client.py
python examples/03_progress/client.py
python examples/04_reverse_request/client.py
python examples/05_errors/client.py

完整说明见 examples/README.md。

测试

python -m pytest -q

测试覆盖单一 Frame 协议、信封校验、双向通用事件分发、五组完整示例、启动失败、 stdout 独立性、主动关闭和子进程异常退出。