使用 C 扩展共享 Python 多处理共享值

Question

我在 python 中有两个进程共享一个布尔标志：

from multiprocessing import Process, Value

class MyProcess(Process):
   def __init__(self):
       self.flag = Value('B',false)
       # [...]

   def run(self):
       while self.active:
           # do_something()
           if some_condition:
               self.work_to_be_extended__()


   def work_to_be_extended__(self) -> bool:
       while some_internal_loop_condition:
           if self.flag.value:
               # do something
      return result
  
if __name__ == '__main__':
    my_proc = MyProcess()
    my_proc_flag = my_proc.flag
    my_proc.start()
    # [...] Some work
    if condition:
        my_proc_flag.value = True

我需要将MyProcess.work_to_be_extended 放在扩展模块中，以便在 C 代码中执行。就像是：

bool extended_work(void):
{
    while (some_condition) {
        if (my_proc_flag) {
            do_something()
        }
    return result
}

我还没有设计扩展，因为我需要先了解如何共享MyProcess.flag 变量。请，请注意，我不需要经过变量值，我需要通过它的参考为了让扩展程序看到在扩展程序不存在的主进程中操作的标志值的变化`。

希望我已经很清楚了**

Answer 1

Multiprocessing 有一个用于 ctypes 数组和值的 sharedctypes 子模块。您可以使用它来创建共享 ctypes（在我的示例中为 int）。然后使用 ctypes.byref 发送指向该 int 的指针。 由于底层机制是 SHM（不是一些隐藏在引擎盖下的管道），这个引用指向的内存在两个过程中实际上是相同的。 shval.value 是 *p 由传递的 p 参数指向的，即 byref(shval)。

因此，不需要我之前答案的大小为 1 的数组，更重要的是，不需要附带它的免责声明。

#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include <stdint.h>

void myf(volatile uint32_t *p){
    for(;;){
        printf("<c>%d</c>
", *p);
        if(*p==100) return;
        (*p)++;
        sleep(1);
    }
}

import multiprocessing as mp
import multiprocessing.sharedctypes as st
import ctypes

mylib=ctypes.CDLL("libtoto.so")
mylib.myf.argtypes=[ctypes.c_void_p]
shval=st.RawValue(st.ctypes.c_uint32,12)

class MyProcess(mp.Process):
    def __init__(self):
        super().__init__()

    def run(self):
        mylib.myf(st.ctypes.byref(shval))

if __name__=="__main__":
    myproc=MyProcess()
    myproc.start()
    while True:
        i=int(input("val>"))
        shval.value=i

因此，对您的问题的简短回答是：使用 multiprocessing.sharedctypes 并将 byref(sharedval) 传递给您的函数。

Answer 2

前提

这个答案来自对@chrslg给出的好解决方案的改编。这扩展了该用法以适用于 Python/C 编程的其他范例，例如 C 扩展 API、Cython 和 Boost::Python。

请先阅读该答案以获得更深入的背景。

概述：

使用sharedctypes.RawValue 作为所需的布尔标志，可以通过sharedctypes.ctypes.addressof 方法访问底层数据地址。

因此，可以将变量的地址作为long long int（64 位）传递，并将其转换为指向所需数据的指针。例如，对于一个uint8_t 变量，一个包含在 C 扩展中

 int64_t address; // This is initialized in some way, depending on the C interface to python
 
 // Pointer to shared data
 uint8_t* pointer = reinterpret_cast<uint8_t*>(address);

 printf("Current value of shared data: %u
", pointer);

应用到不同的 Python - C/C++ 接口

通用 C 共享库

让我们创建一个基本的、简单的 C 库，它每秒只读取 1 次共享变量的值：

// cshare_data/cshare_data.c

#include "cshare_data.h"

#include <time.h>
#include <unistd.h>
#include <stdio.h>


void cshare_data(uint8_t* data, char from_where_called) {
    char *s = NULL;
    if (from_where_called == 0) {
        s = "cTypes CDLL";
    } else if (from_where_called == 1)
    {
        s = "Python C Extension";
    } else if (from_where_called == 2)
    {
        s = "Boost::Python";
    } else if (from_where_called == 3)
    {
        s = "Cython";
    }
    for (int i = 0; i < 10; i++) {
        printf("C code read from %s a value of: %u
", s, *data);
        sleep(1);
    }
}

标题：

// cshare_data/cshare_data.h

#ifndef CSHARE_DATA_H
#define CSHARE_DATA_H

#include <stdint.h>
#include <stddef.h>

extern void cshare_data(uint8_t*, char);

#endif

Python共享数据编辑流程

对于其余示例，我将参考以下修改共享数据的 Python 进程（示例中为unsigned char）：

from multiprocessing.sharedctypes import RawValue, Value
import multiprocessing.sharedctypes as st
from multiprocessing import Process


class MyProcess(Process):
    def __init__(self):
        Process.__init__(self)
        self.int_val = RawValue(st.ctypes.c_ubyte, 0)
    def run(self) -> None:
        import time
        for _ in range(10):
            print('Value in Python Process: ', self.int_val.value)
            self.int_val.value += 1
            time.sleep(1)

my_proc = MyProcess()
my_proc.start()

注：下文不再赘述。

Python C 扩展

使用上述模式的 Python C 扩展 API 如下：

#include <Python.h>

#include <stdio.h>
#include <time.h>

#include "cshare_data.h"

static PyObject *cshare_data_wrapper(PyObject *self, PyObject *args)
{
    PyObject *val = NULL;

    // This will store the address of the uchar variable being passed from Python
    int64_t address = 0;

    // Convert the single element tuple into a 8-byte int (address)
    if(!PyArg_ParseTuple(args, "L", &address)) {
        printf("Error parsing Tuple
");
        return NULL;
    }

    // Now address is reinterpreted as the shared variable pointer
    uint8_t *pointer = (uint8_t *)(address);
    
    // Call the library function
    cshare_data(pointer, 1);

    return Py_None;
}

static PyMethodDef CShapreDataMethods[] = {
    {"cshare_data", cshare_data_wrapper, METH_VARARGS, "Python interface for sharedata C library function"},
    {NULL, NULL, 0, NULL}
};

static struct PyModuleDef cshareddata_module = {
    PyModuleDef_HEAD_INIT,
    "csharedata_module",
    "Python interface for the fputs C library function",
    -1,
    CShapreDataMethods
};

PyMODINIT_FUNC PyInit_cshare_data_pyext(void) {
    return PyModule_Create(&cshareddata_module);
}

请参考official documentation 和这个非常好的tutorial，以更深入地了解 Python C-API

提升::Python

与 Python C-API 所做的非常相似，boost 包装器看起来像：

extern "C" {
    #include "cshare_data.h"
}
#include <boost/python.hpp>

void cshare_data_boost_wrapper(long long int data_address) {
    uint8_t* data = reinterpret_cast<uint8_t*>(data_address);
    cshare_data(data, 2);
}

BOOST_PYTHON_MODULE(ctrigger) {
    using namespace boost::python;
    def("cshare_data", cshare_data_boost_wrapper);
}

CMake - 图书馆建筑

从具有以下树结构的项目移动：

```
project_root
|   cshare_data.py
|---clibs
|   |   cshare_data_boost.so
|   |   cshare_data_pyext.so
|   |   cshare_data.so
|
|---cshare_data
|   |   cshare_data.c
|   |   cshare_data.h
|   
|   CMakeList.txt
```

使用了以下编译 CMake 脚本：

cmake_minimum_required (VERSION 2.6)

project (cshare_data)

set(CMAKE_SHARED_MODULE_PREFIX "")
set(CMAKE_SHARED_LIBRARY_PREFIX "")

# Common C shared library

set(CMAKE_LIBRARY_OUTPUT_DIRECTORY ${CMAKE_SOURCE_DIR}/clibs)

include_directories(${CMAKE_SOURCE_DIR}/cshare_data)
link_directories(${CMAKE_SOURCE_DIR}/clibs)
# --- Common C shared library ---
add_library(cshare_data SHARED cshare_data/cshare_data.c)

# Needed for Python C Extension Module and Boost::Python
include_directories("/usr/include/python3.8")

# --- Python C Extension Module library ---
add_library(cshare_data_pyext MODULE cshare_data_pyinterface/cshare_data_pyext.c)
target_link_libraries(cshare_data_pyext python3.8)
target_link_libraries(cshare_data_pyext cshare_data)


# --- Python C Extension Module library ---
include_directories("/home/buzz/boost_1_80_0")
link_directories("/home/buzz/boost_1_80_0/build/lib")
add_library(cshare_data_boost MODULE cshare_data_pyinterface/cshare_data_boost.cpp)
target_link_libraries(cshare_data_boost python3.8)
target_link_libraries(cshare_data_boost boost_python38)
target_link_libraries(cshare_data_boost cshare_data)

Python - 调用 C 包装器

仅出于演示的目的，我编写了 3 个不同的进程，它们共享相同的 int_val（由上述 MyProcess 处理）并调用 C 函数来打印此变量的值。请注意，尽管代码行相同，但必须在每次进程调用时撤回地址，因为 multiprocessing.sharedctypes 在底层封装了 int_val 的 IPC 同步架构，这意味着每个实际的 int_val 都存在于正确的进程中。

my_proc = MyProcess()
my_proc.start()

l = []


class FromWhere(IntEnum):
    ctype = 0
    python_c_extension = 1
    boost_python = 2

def from_ctype_import_dll(int_val: RawValue):
    import ctypes
    reference = st.ctypes.byref(my_proc.int_val)
    mylib=ctypes.CDLL("clibs/cshare_data.so")
    mylib.cshare_data.argtypes=[ctypes.c_void_p, ctypes.c_char]
    mylib.cshare_data(reference, FromWhere.ctype.value)


def from_python_c_extension(int_val: RawValue):
    from clibs import cshare_data_pyext
    address = st.ctypes.addressof(int_val)
    cshare_data_pyext.cshare_data(address)

def from_boost_python(int_val: RawValue):
    from clibs import cshare_data_boost
    address = st.ctypes.addressof(int_val)
    cshare_data_boost.cshare_data(address)

ps: List[Process] = []

ps.append(Process(target=from_ctype_import_dll, args=(my_proc.int_val,)))
ps.append(Process(target=from_python_c_extension, args=(my_proc.int_val,)))
ps.append(Process(target=from_boost_python, args=(my_proc.int_val,)))

for p in ps:
    p.start()

for p in ps:
    p.join()

结果达到：

Value in Python Process:  0
C code read from cTypes CDLL a value of: 1
C code read from Python C Extension a value of: 1
C code read from Boost::Python a value of: 1
Value in Python Process:  1
C code read from cTypes CDLL a value of: 2
C code read from Boost::Python a value of: 2
C code read from Python C Extension a value of: 2
Value in Python Process:  2
C code read from cTypes CDLL a value of: 3
C code read from Boost::Python a value of: 3
C code read from Python C Extension a value of: 3
C code read from cTypes CDLL a value of: 3
Value in Python Process:  3
C code read from Boost::Python a value of: 4
C code read from Python C Extension a value of: 4
C code read from cTypes CDLL a value of: 4
Value in Python Process:  4
C code read from Boost::Python a value of: 5
C code read from Python C Extension a value of: 5
C code read from cTypes CDLL a value of: 5
Value in Python Process:  5
C code read from Boost::Python a value of: 6
C code read from Python C Extension a value of: 6
C code read from cTypes CDLL a value of: 6
Value in Python Process:  6
C code read from Python C Extension a value of: 7
C code read from Boost::Python a value of: 7
C code read from cTypes CDLL a value of: 7
Value in Python Process:  7
C code read from Python C Extension a value of: 8
C code read from Boost::Python a value of: 8
C code read from cTypes CDLL a value of: 8
Value in Python Process:  8
C code read from Python C Extension a value of: 9
C code read from Boost::Python a value of: 9
C code read from cTypes CDLL a value of: 9
Value in Python Process:  9
C code read from Python C Extension a value of: 10
C code read from Boost::Python a value of: 10