# cpp_python

**Repository Path**: liu-chenfan/cpp_

## Basic Information

- **Project Name**: cpp_python
- **Description**: No description available
- **Primary Language**: Unknown
- **License**: MIT
- **Default Branch**: main
- **Homepage**: None
- **GVP Project**: No

## Statistics

- **Stars**: 1
- **Forks**: 0
- **Created**: 2026-05-14
- **Last Updated**: 2026-05-19

## Categories & Tags

**Categories**: Uncategorized

**Tags**: None

## README

# cpp_run_python

C++ 应用程序通过 Win32 管道启动一个长驻 Python 子进程，使用换行分隔的 JSON 协议进行双向通信。支持通过 Windows 命名共享内存进行零拷贝图像数据传输。仅支持 Windows 平台。

## 项目结构

```
cpp_run_python/
├── core/                            # 核心动态库 (cpp_run_python_core.dll)
│   ├── include/
│   │   ├── core_export.h            # DLL 导出宏
│   │   ├── protocol.h               # 请求/响应结构体（iguana 反射序列化）
│   │   ├── python_worker_client.h   # Python 子进程管理客户端（单 worker）
│   │   ├── python_worker_pool.h     # 多 worker 线程安全连接池（round-robin + per-worker mutex）
│   │   ├── shared_memory.h          # Win32 命名共享内存封装类
│   │   └── json_utils.h              # StepsBuilder 类和动态 JSON 辅助函数
│   └── src/
│       ├── python_worker_client.cpp # CreateProcess + 管道读写实现
│       ├── python_worker_pool.cpp   # 连接池 Start/Stop/Call 实现
│       └── shared_memory.cpp        # CreateFileMapping / OpenFileMapping / MapViewOfFile
├── test/                            # 测试可执行文件 (cpp_run_python.exe)
│   ├── include/
│   │   └── test_runner.h            # CallAndPrint 模板（Client/Pool 双重载）+ 测试函数声明
│   └── src/
│       ├── main.cpp                 # 入口：解析路径和 -w N 参数，单 worker 或 Pool 模式运行测试
│       ├── demo_test.cpp            # 演示脚本测试（run, greet, status）
│       ├── math_test.cpp            # 数学脚本测试（run, square, 组合调用）
│       ├── counter_test.cpp         # Counter 类实例测试（new, call_method, dispose）
│       ├── image_test.cpp           # 共享内存图像测试（invert, brightness, grayscale, blur, float32, ImageProcessor）
│       ├── pipeline_test.cpp         # Pipeline 函数调用链测试（数学链, 跨脚本, 图像链, $result 链）
│       └── pool_test.cpp             # Pool 模式测试（全部测试的 Pool 重载 + 多线程并发验证）
├── scripts/
│   ├── python_worker.py             # Worker 入口（2 行引导代码）
│   ├── worker/
│   │   ├── __init__.py
│   │   ├── main.py                  # stdin/stdout JSON 主循环
│   │   ├── dispatcher.py            # action 路由分发（call/new/call_method/dispose/pipeline）
│   │   ├── loader.py                # 动态加载用户 .py 脚本模块
│   │   ├── instance_cache.py        # 类实例缓存（new/call_method/dispose）
│   │   └── shm_helper.py            # Win32 共享内存 ctypes 封装 + numpy 零拷贝视图
│   ├── demo_script.py               # 演示脚本：函数调用 + Counter 类
│   ├── math_script.py               # 数学运算脚本
│   └── image_script.py              # 图像处理脚本：使用 cv2 处理共享内存中的图像
└── thirdparty/
    └── iguana/                      # Header-only C++ JSON 库
```

## 构建与运行

### 前置条件

- Visual Studio 2019+ (MSVC)
- CMake ≥ 3.16
- OpenCV 4.x（`CMakeLists.txt` 中设置 `OpenCV_DIR`，默认 `E:/opencv/build`；`opencv_world4XXX.dll` 需复制到 `build/bin/Release/`）
- Python 3.x（默认使用 `python`，可通过命令行参数指定路径）
- Python 依赖：`numpy`、`opencv-python`（图像处理功能所需）

### 构建

```sh
cmake -B build -G "Visual Studio 17 2022" -A x64
cmake --build build --config Release
```

构建产物输出到 `build/bin/Release/`，DLL 与 EXE 同目录。

### 运行

```sh
# 单 worker 模式（默认）
build\bin\Release\cpp_run_python.exe
# 或指定 Python 解释器路径
build\bin\Release\cpp_run_python.exe C:\Python312\python.exe

# 多 worker 连接池模式（N 个并发 Python 进程）
build\bin\Release\cpp_run_python.exe -w 3
```

命令行参数：
- 第一个位置参数（可选）：Python 解释器路径，默认 `python`
- `-w N` 或 `--workers N`：worker 数量，默认 1。`-w 1` 使用单 `PythonWorkerClient`；`-w N>1` 使用 `PythonWorkerPool` 启动 N 个 worker 并发执行

> 可执行文件通过向上三级目录定位 `scripts/`。exe 需位于 `build/bin/Release/` 下，移动后脚本将无法找到。

## 多线程并发

`PythonWorkerPool` 支持从多个线程安全地并发调用 Python worker：

- **单线程模式** (`-w 1`)：使用 `PythonWorkerClient`，与原有行为完全一致
- **多线程模式** (`-w N`)：使用 `PythonWorkerPool`，管理 N 个独立的 Python worker 进程，每个 worker 有自己的管道和进程
- **分发策略**：`atomic<int>` round-robin 选择 worker，每个 worker 有独立的 `std::mutex` 保护管道读写
- **实例亲和性**：`new`/`call_method`/`dispose` 的实例缓存在单个 worker 进程内。Pool 模式下实例可能路由到不同 worker，导致 `KeyError`。对有状态类实例请使用单 worker 模式，或确保同一实例的请求始终路由到同一 worker

## 通信协议

C++ 主进程与 Python worker 之间通过 **单行 JSON** 通信（一行请求，一行响应）。

### 请求格式

```json
{
  "request_id": "unique-id",
  "script": "path/to/script.py",
  "target": "function_or_class_name",
  "action": "call",
  "timeout_ms": 5000,
  "instance_id": "",
  "...": "其他字段作为 inputs 传入 Python"
}
```

### 响应格式

```json
{
  "request_id": "unique-id",
  "success": true,
  "debug": "call",
  "error_code": 1,
  "instance_id": "",
  "...": "Python 返回的结果字段"
}
```

### Action 类型

| action | `target` 含义 | `instance_id` | 说明 |
|--------|--------------|---------------|------|
| `call`（默认） | 函数名 | 忽略 | 调用模块级函数 |
| `new` | 类名 | 响应中返回 | 实例化类并缓存，返回 `instance_id` |
| `call_method` | 方法名 | 必填 | 调用缓存实例的方法 |
| `dispose` | 忽略 | 必填 | 销毁缓存的实例 |
| `pipeline` | "pipeline" | 忽略 | 按顺序执行一组函数调用链 |

### 错误码

| 错误码 | 含义 |
|--------|------|
| `1` | 成功 |
| `1000` | JSON 解析失败 |
| `1001` | 脚本运行时错误 |
| 负数 | C++ 端错误 |

## 共享内存图像传输

C++ 和 Python 之间通过 Windows 命名共享内存传递大型图像数据，避免管道序列化开销。

### 数据流

```
C++ 端                                    Python 端
──────                                    ──────
1. SharedMemory::Create(name, size)
2. 写入 cv::Mat 像素数据
3. 通过管道发送 JSON 请求
   {shm_name, width, height, channels, dtype}
                                          4. OpenFileMapping → MapViewOfFile
                                          5. 包装为 numpy 数组（零拷贝）
                                          6. 使用 cv2/numpy 处理图像
                                          7. 写回共享内存
                                          8. 通过管道发送 JSON 响应
9. 从共享内存读取结果
10. SharedMemory::Close()
```

### C++ 端使用

```cpp
// 创建共享内存并写入 cv::Mat
cv::Mat img = cv::imread("image.png");
SharedMemory shm;
shm.Create("my_image", img.total() * img.elemSize());
shm.Write(img.data, img.total() * img.elemSize());

// 发送处理请求
auto req = std::make_shared<ImageProcessRequest>();
req->shm_name = "my_image";
req->width = img.cols;
req->height = img.rows;
req->channels = img.channels();
req->dtype = "uint8";
// ... 调用 Python 处理函数

// 读取结果
auto result = shm.Read(0, buf_size);
cv::Mat result_img(height, width, CV_8UC3);
std::memcpy(result_img.data, result.data(), result.size());

shm.Close();
```

### Python 端使用

```python
from worker.shm_helper import open_shared_memory

def my_filter(inputs: dict) -> dict:
    shm_name = inputs["shm_name"]
    width = int(inputs["width"])
    height = int(inputs["height"])
    channels = int(inputs.get("channels", 3))

    buf_size = width * height * channels
    shm = open_shared_memory(shm_name, buf_size)
    arr = shm.numpy_array(width, height, channels, "uint8")
    try:
        # 使用 cv2 处理（零拷贝，修改 arr 即修改共享内存）
        cv2.GaussianBlur(arr, (5, 5), 0, arr)
    finally:
        shm.close()

    return {
        "shm_name": shm_name,
        "width": width, "height": height,
        "channels": channels, "dtype": "uint8",
    }
```

> 共享内存名称自动添加 `Local\` 前缀。C++ 负责分配和清理，Python 仅打开视图。

## Pipeline（函数调用链）

`action = "pipeline"` 允许在单次请求中按顺序执行多个函数调用，支持跨脚本、跨 action 类型、以及数据链传递。

### 请求格式

```json
{
  "request_id": "pipe-001",
  "script": "path/to/script.py",
  "target": "pipeline",
  "action": "pipeline",
  "steps_json": "[{\"target\":\"square\",\"a\":4},{\"target\":\"square\",\"a\":\"$result.result\"}]"
}
```

`steps_json` 是一个 JSON 数组字符串，每个元素是一个 step 对象，支持以下字段：

- `target`（必选）— 函数名、类名或方法名
- `script`（可选）— 缺省使用外层 `script`，支持跨脚本调用
- `action`（可选）— 缺省 `"call"`，支持 `call`/`new`/`call_method`/`dispose`
- `instance_id`（可选）— `call_method`/`dispose` 需要
- 其他字段 — 作为 `inputs` 传递给目标函数

### 数据链传递

step 中任意字段的值以 `$result` 开头时，会被替换为上一步的返回值：

- `"$result"` → 上一步返回的整个 dict
- `"$result.field"` → 上一步返回 dict 中 `field` 对应的值

示例：
```
Step 1: square(a=4)           → 返回 {"result": 16.0}
Step 2: square(a="$result.result") → a 被替换为 16.0 → 返回 {"result": 256.0}
```

### 响应格式

成功时 `result_json` 包含最后一步的完整返回值：

```json
{
  "request_id": "pipe-001",
  "success": true,
  "debug": "pipeline",
  "error_code": 1,
  "steps_total": 2,
  "steps_executed": 2,
  "failed_step": -1,
  "result_json": "{\"request_id\":\"pipe-001\",\"success\":true,\"result\":256.0,...}"
}
```

失败时立即停止并返回失败步骤信息：

```json
{
  "request_id": "pipe-001",
  "success": false,
  "debug": "<traceback>",
  "error_code": 1001,
  "steps_total": 3,
  "steps_executed": 1,
  "failed_step": 1,
  "result_json": "{...}"
}
```

### C++ 端构建 steps_json

**方式一：StepsBuilder（推荐，简单场景）**

```cpp
StepsBuilder sb;
sb.Begin().Str("script", path).Str("target", "square").Dbl("a", 4.0).End();
sb.Begin().Str("script", path).Str("target", "square").Str("a", "$result.result").End();
req->steps_json = sb.Build();
```

**方式二：iguana::jobject / jarray（复杂嵌套结构）**

```cpp
iguana::jarray steps;
iguana::jobject step;
step["script"] = path;
step["target"] = std::string("invert");
step["shm_name"] = std::string("my_image");
step["width"] = 64;
steps.push_back(step);

std::string steps_json = "[";
for (size_t i = 0; i < steps.size(); ++i) {
  if (i > 0) steps_json += ",";
  steps_json += JsonValueToString(steps[i]);
}
steps_json += "]";
```

**解析 result_json**

```cpp
iguana::jvalue last;
iguana::parse(last, resp->result_json.begin(), resp->result_json.end());
double result = JsonGetNumber(last, "result");  // 安全处理 int/double
std::string message = JsonGetString(last, "message");
```

## 编写用户脚本

### 模块级函数

```python
# scripts/my_script.py

def hello(inputs: dict) -> dict:
    name = inputs.get("name", "world")
    return {"message": f"Hello, {name}!"}
```

C++ 端调用：

```cpp
auto req = std::make_shared<MyRequest>();
req->script = my_script.string();
req->target = "hello";
req->action = "call";
```

### 带状态的类

```python
# scripts/my_script.py

class Accumulator:
    def __init__(self, inputs: dict):
        self.total = float(inputs.get("initial", 0))

    def add(self, inputs: dict) -> dict:
        self.total += float(inputs.get("value", 0))
        return {"total": self.total}
```

C++ 端使用流程：

```cpp
// 1. 实例化（action="new"）→ 获得 instance_id
req->target = "Accumulator";
req->action = "new";

// 2. 调用方法（action="call_method"）
req->target = "add";
req->action = "call_method";
req->instance_id = instance_id;

// 3. 销毁（action="dispose"）
req->action = "dispose";
req->instance_id = instance_id;
```

> 类实例缓存在 worker 进程的 `_instances` 字典中。worker 退出后实例丢失，无自动清理机制，需手动 `dispose`。在多 worker 池模式下（`-w N>1`），实例亲和性不保证——`call_method` 可能路由到不同于 `new` 的 worker，导致 `KeyError`。对有状态实例请使用单 worker 模式（`-w 1`）。

## 添加新命令

1. 在 `core/include/protocol.h` 中定义请求/响应结构体，继承 `RequestCommon`/`ResponseCommon`，添加 `iguana::base_impl` 和 `YLT_REFL` 宏
2. 在 Python 脚本中编写函数或类
3. 在 `test/src/` 中创建或扩展测试文件，在 `test/include/test_runner.h` 中声明函数，从 `test/src/main.cpp` 中调用

## 设计要点

- **核心库分离**：`core/` 编译为独立 DLL，`test/` 作为可执行文件链接核心库 + OpenCV，方便复用和独立迭代
- **DLL 导出**：通过 `core_export.h` 的 `CORE_API` 宏控制导出，构建时定义 `CPP_RUN_PYTHON_CORE_BUILDING`。`PythonWorkerClient`、`PythonWorkerPool`、`SharedMemory` 均使用相同导出机制
- **多线程安全**：`PythonWorkerPool::Call()` 使用 `atomic fetch_add` 做 round-robin 选择 worker + per-worker `std::mutex` 保护管道读写，支持多线程并发调用
- **共享内存零拷贝**：C++ 通过 Win32 `CreateFileMapping` 分配命名共享内存，Python 通过 `ctypes` 调用 `OpenFileMapping` 打开并包装为 numpy 数组，双方直接操作同一块物理内存
- **热更新**：Worker 每次处理 `call`/`new` 请求时通过 `importlib.util` 重新加载 Python 脚本，修改脚本无需重启或重新编译
- **类型安全**：C++ 端使用 iguana 的 `base_impl` + `YLT_REFL` 宏自动生成 `to_json`/`from_json`，无需手动拼接 JSON 字符串（Pipeline 的 `steps_json` 除外，需用 `StepsBuilder` 或 `jobject`/`jarray` 构建）
- **动态 JSON 构建**：`json_utils.h` 提供 `StepsBuilder`（流式 API）和 `JsonValueToString()`（jvalue 序列化）两种方式构建 `steps_json`；`JsonGetNumber()` 安全处理 iguana 的 int/double 类型差异
- **UTF-8 支持**：MSVC 编译选项 `/utf-8` + 运行时 `chcp 65001`，支持中文字符的传输和显示
- **stderr 安全**：Python worker 的 stderr 与 stdout 共用同一管道，`SharedMemoryView.__del__` 等清理方法必须静默捕获所有异常，避免异常输出污染 JSON 响应流
- **编码规范**：C++ 代码遵循 Google C++ Style Guide（2 空格缩进、`PascalCase` 函数名、`snake_case_` 成员变量带尾下划线、`snake_case` 局部变量/参数）