从 Python 调用 Rust

GOTO 会议上理查德-费尔德曼（Richard Feldman）的演讲《跨语言调用函数》（Calling Functions Across Languages）。

该演示表明，从HTTP迁移到IPC，然后从 IPC 迁移到 FFI，删除了样板代码。它使用 JavaScript 作为客户端堆栈，使用 Ruby 作为底层堆栈。虽然 JavaScript 很有意义，但恕我直言，使用 Ruby（一种不以性能闻名的动态语言）并不合适。

演示

我想使用 Python 作为客户端堆栈，并使用 Rust 作为目标堆栈。我想提高我在这两方面的知识，它完全符合上面提到的性能原因。

该示例应将复数计算从 Python 委托给 Rust。尽管 Python 本身就支持它们，但它对于我们的目标来说已经足够好了。此外，它允许解析和显示它们而无需额外的代码。这个想法是使用CLI来进行计算，例如：

python main.py --add --arg1 1+3j --arg2 -5j

我们期望得到以下输出：

(1-2j)

对于命令处理，我们将利用该click库：

Click 是一个 Python 包，用于以可组合的方式使用尽可能少的代码创建漂亮的命令行界面。它是“命令行界面创建工具包”。它具有高度可配置性，但具有开箱即用的合理默认值。

这是"wrapper"代码:

@click.command()

@click.option('--add', 'command', flag_value='add')

@click.option('--sub', 'command', flag_value='sub')

@click.option('--mul', 'command', flag_value='mul')

@click.option('--arg1', help='First complex number in the form x+yj')

@click.option('--arg2', help='Second complex number in the form x\'+y\'j')

def cli(command: Optional[str], arg1: Optional[str], arg2: Optional[str]) -> None:

#result: complex = compute result

#print(result)

if __name__ == '__main__':

cli()

现在已经解决了，让我们检查一下上面提到的选项。

HTTP

从开发人员的角度来看，HTTP 是最直接的方法，尽管它的性能最差：我们需要遍历许多 OSI 模型层。

即使客户端和服务器以及客户端实现相同的堆栈，HTTP 也会强制要求序列化格式。尽管我很喜欢 XML，但 JSON 是最广泛使用的一种。

在设计层面上，我们将使用 HTTPPOST方法来随请求发送正文数据。我们将每个操作映射到一个 URL，例如、add和sub。

Python客户端

虽然 Python 通过http开箱即用的包支持 HTTP，但requests库提供了巨大的帮助。

json: dict[str, list[float]] = dict(a=[n1.real, n1.imag], b=[n2.real, n2.imag]) # 创建请求征文

req = requests.post(f'http://localhost:3000/{command}', json=json) # 使用有效负载将POST请求发送到Rust服务

body: dict[str: list[float]] = req.json() # 获取响应

if body and 'result' in body:

real: Optional[str] = body['result'][0]

imaginary: Optional[str] = body['result'][1]

return complex(float(real), float(imaginary)) # complex从响应正文创建Python

raise Exception()

Rust服务器

我将使用axum，因为它是我最熟悉的框架：

使用macro免费API将请求路由到处理程序。

使用提取器以声明方式解析请求。

简单且可预测的错误处理模型。

使用最少的样板生成响应。

充分利用中间件、服务和实用程序组成的生态系统tower和tower-http。

我们首先对输入和输出进行建模：

#[derive(Deserialize)]

struct Input {

a: Complex,

b: Complex,

}

#[derive(Serialize)]

struct Output { result: Complex }

接下来，我们需要提供一个async函数来管理操作：

async fn add(Json(payload): Json) -> impl IntoResponse {

Json(Output { result: payload.a + payload.b })

}

最后，我们可以搭建路由和服务器本身：

#[tokio::main] # tokio宏，将异步主函数作为入口点

async fn main() {

let app = axum::Router::new() # 创建路由器对象

.route("/add", post(add)) # 将每个函数添加到其专用子路径下

.route("/sub", post(sub))

.route("/mul", post(mul));

axum::Server::bind(&"0.0.0.0:3000".parse().unwrap()) # 创建绑定到端口的 HTTP 服务器`3000`

.serve(app.into_make_service()) # 让魔法发生

.await

.unwrap()

}

进程间通信

我们不需要HTTP，因为客户端和服务器在同一台机器上。今后，我们可以用IPC代替 HTTP 。

在基于 Nix 的系统上，我们可以利用域套接字：

Unix 域套接字的 API 与 Internet 套接字的 API 类似，但不是使用底层网络协议，而是所有通信完全发生在操作系统内核内。Unix 域套接字可以使用文件系统作为它们的地址名称空间。（某些操作系统，如 Linux，提供额外的命名空间。）进程将 Unix 域套接字引用为文件系统 inode，因此两个进程可以通过打开同一个套接字进行通信。--- Wikipedia

Python客户端

Python 通过“socket”包原生支持Unix套接字。

with socket.socket(socket.AF_UNIX) as client: # 创建socket客户端

data: dict[str, list[float]] = dict(command=command,a=[n1.real, n1.imag], b=[n2.real, n2.imag])

encoded: bytes = json.dumps(data).encode('UTF-8') # 将 JSON 字符串编码为字节

client.connect("/tmp/socket") # 连接到socket

client.send(encoded) # 发送数据

从网络的角度来看，这要简单得多。从开发人员的角度来看，我们只需要将 JSON 转换为字节即可。

Rust 服务器

Rust 本身也支持 Unix 套接字

let srv_path = "/tmp/socket"; # Unix 套接字是一种特定类型的文件

if metadata(srv_path).is_ok() {

remove_file(srv_path).unwrap(); # 清理以前潜在的套接字

}

let listener = UnixListener::bind(srv_path).unwrap(); # 将套接字绑定到路径

loop {

let (mut stream, _) = listener.accept().unwrap(); # 监听套接字的数据

let mut payload = String::new(); # 创建一个新的可变的可调整大小的字符串

let _ = stream.read_to_string(&mut payload); # 将套接字数据读入字符串

println! {"Received {}", payload};

let input = serde_json::from_str::(&payload).unwrap(); # 将输入字符串反序列化为 JSON

let output = &command(input);

let result = serde_json::to_string(output).unwrap(); # 将输出字符串序列化为 JSON

stream.write(result.as_bytes()).unwrap_or_default(); # 将数据发送回套接

println! {"Sent {}", result};

}

我以前没有使用 Unix 套接字的经验。我可以将数据从 Python 发送到 Rust，但无法返回数据。任何有关该主题的帮助将不胜感激。

对外函数接口

到目前为止，从 Python 调用 Rust 需要通过网络（可能）远程或本地传递数据。FFI允许将所有内容保留在一个进程中。然而，我们需要一种在不同技术堆栈之间传递数据的方法。由于历史原因，大多数都提供了调用基于 C 的库的桥梁。事实上，我们可以将 Rust 代码编译为 C 兼容库并从 Python 调用它。

Python 允许通过包加载 C 库ctypes：

ctypes 是 Python 的外部函数库。它提供与 C 兼容的数据类型，并允许调用 DLL 或共享库中的函数。它可用于将这些库包装在纯 Python 中。

根据平台的不同，该库可以是.dll(Windows)、.so(Nix) 或.dylib(OSX)。我发现自己属于后者。因此，我可以加载它：

rust = ctypes.CDLL('/path/to/lib/my.dylib')

从现在开始，我们可以通过变量调用库中定义的任何rust函数。剩下的问题是 Python 类型不是 Rust 类型。因此，我们需要在两边都使用 C 类型。

from ctypes import c_double

rust.compute(command.encode("UTF-8"), c_double(n1.real), c_double(n1.imag), c_double(n2.real), c_double(n2.imag))

该c_double函数将 Python 浮点数转换为 C 双精度值。

我们甚至可以通过类重用我们的自定义类Structure：

结构和联合必须派生自模块中定义的Structure和基类。每个子类必须定义一个属性。必须是一个二元组列表，包含字段名称和字段类型。Union``ctypes``*fields*``*fields*

正如我们所知 Rust 的形式Complex，这很容易提供：

class RustComplex(Structure):

_fields_ = [("re", c_double), # 根据文档

("im", c_double)]

def __init__(self, c: complex, *args: Any, **kw: Any) -> None: # 易于使用的构造函数

super().__init__(*args, **kw)

self.re = c.real

self.im = c.imag

最后一点是设置函数的返回类型compute。默认情况下，它是int：

默认情况下，假定函数返回 Cint类型。其他返回类型可以通过设置函数对象的 restype 属性来指定。

rust.compute.restype = RustComplex # 设置`compute`返回类型

return rust.compute(command.encode("UTF-8"), RustComplex(n1), RustComplex(n2)) # 利用上面创建的类

Rust 库

在 Rust 中，该std::ffi模块支持 FFI：

与 FFI 绑定相关的实用程序。

该模块提供了跨非 Rust 接口处理数据的实用程序，就像其他编程语言和底层操作系统一样。它主要用于 FFI（外部函数接口）绑定和需要与其他语言交换类 C 字符串的代码。

与之前步骤的主要区别在于，我们不再需要应用程序，而是需要库。我们需要将main.rs文件重命名为lib.rs. 此外，我们必须配置项目来创建一个库：

Cargo.toml

[lib]

name = "my"

crate-type = ["dylib"]

此外，我们唯一需要的依赖是complex.

然而，代码变得非常简单：

lib.rs

#[no_mangle] # 保留符号名称以供外部使用

# 使用兼容的类型。第一个参数是指向 C 字符串的指针，其他参数是常规结构，因为 Python 将以预期格式传递它们

fn compute(c_string_ptr: *const c_char, a: Complex, b: Complex) -> Complex {

# 将 C 字符串转换为 Rust 字符串

let bytes = unsafe { CStr::from_ptr(c_string_ptr).to_bytes() };

let command = str::from_utf8(bytes).unwrap();

match command {

"add" => a + b,

"sub" => a - b,

"mul" => a * b,

_ => panic!("Unknown command"),

}

}

此时，我们现在可以在Python 进程中使用 Rust 代码了。

结论

在这篇文章中，我们详细介绍了从 Python 调用 Rust 的三种不同方法：HTTP、IPC 和 FFI。

FFI 是性能最好的一种，但只是有时可行。IPC 替代方案在单台机器上效果很好。当两者都失败时，我们始终可以相信HTTP。