使用地址在 python 进程之间共享 ctype 内存答案

【问题标题】：Share ctype memory across python processes using the address使用地址在 python 进程之间共享 ctype 内存
【发布时间】：2019-02-03 20:01:02
【问题描述】：

我正在尝试在 python 中跨多个进程发送动态数组。我的第一个解决方案是直接通过多处理类的队列/管道发送数据。问题是它受到以太网连接带宽的限制。因此，我尝试使用 ctype 数组并仅传递对象的地址。当我尝试从第二个进程（A.raw 或 A.value）访问数组时，进程无异常退出。有人知道发生了什么吗？可能是锁等问题。

from multiprocessing import Process,Queue
from ctypes import c_char,addressof

from time import sleep
import os




class ProcessIn(Process):
    def __init__(self,QueueI):
        super().__init__(daemon=True)
        self.QueueI=QueueI

    def run(self):
        Array=[]
        while True:
            N=100000
            A=(c_char*N)()
            A.value=b'\x01'
            Address=addressof(A)
            Array.append(A)
            print(os.getpid(),'putted',Address)
            self.QueueI.put((Address,N))
            sleep(2)





class ProcessOut(Process):
    def __init__(self,QueueI):
        super().__init__(daemon=True)
        self.QueueI=QueueI

    def run(self):
        while True:
            print(os.getpid(),'step 1')
            Address,N=self.QueueI.get()
            print(os.getpid(),'step 2',Address)
            A=(c_char*N).from_address(Address)      
            print(os.getpid(),'step 3')
            Value=A.raw         #This will fail 
            print(os.getpid(),'step 4',Value)   
            sleep(1)

if __name__ == '__main__':
    QueueI=Queue()

    In=ProcessIn(QueueI)
    Out=ProcessOut(QueueI)
    print(os.getpid(),'main')
    In.start()
    Out.start()
    input('press key to finish\n')

【问题讨论】：

你读过多处理模块中的共享内存吗？ docs.python.org/3/library/…
是的，问题是数组有静态大小，我想用动态数组
ctype 数组总是固定大小的，你有一个名为Array 的python 列表并添加c_char 到它
当然可以，但想法是所需的数组由第一个进程创建（具有所需的可变大小），然后通过其地址和大小传递给第二个进程。如果我是正确的，多处理值和数组的标准方法假设对象是在主进程中创建的，然后传递给子进程 - 因此没有更改其大小的选项
pyarrow 您可能会感兴趣

标签： python ctypes shared-memory

【解决方案1】：

好的，我明白了 - 使用带有标签的 mmap：

from multiprocessing import Process,Queue
from ctypes import c_char,addressof
import pyarrow as pa
import numpy as np
from time import sleep
import os
import datetime
import mmap
from sys import getsizeof



class ProcessIn(Process):
    def __init__(self,QueueI):
        super().__init__(daemon=True)
        self.QueueI=QueueI

    def run(self):  
        i=0
        while True:
            N=np.random.randint(10,14)*100000
            data = b'abcdefghijklmnopqrstuvwxyz'        
            Tag='Data_'+str(i)
            buf = mmap.mmap(-1, N*len(data),tagname=Tag)            
            buf[0]=i
            NN=N*len(data)
            # print(buf[0:10])

            print(os.getpid(),'putted',Tag,NN)
            if self.QueueI.qsize()==0:
                self.QueueI.put((Tag,NN,datetime.datetime.now()))           


            i+=1
            sleep(1)


class ProcessOut(Process):
    def __init__(self,QueueI):
        super().__init__(daemon=True)
        self.QueueI=QueueI

    def run(self):
        while True:
            # print(os.getpid(),'step 1')
            Tag,N,start=self.QueueI.get()           
            buf =  mmap.mmap(-1, N,tagname=Tag)

            print('got',buf[0:10],Tag)

            # data=buf.read()
            dt=(datetime.datetime.now()-start).total_seconds()
            if dt!=0:
                print(os.getpid(),N/dt/1024**2,'MBs',dt*1000,'ms',N/1024**2,'MB',N) 
            else:
                print(os.getpid(),np.nan,'MBs',dt*1000,'ms',N/1024**2,'MB',N)   
            buf=None

if __name__ == '__main__':
    QueueI=Queue()

    In=ProcessIn(QueueI)
    Out=ProcessOut(QueueI)
    print(os.getpid(),'main')
    In.start()
    Out.start()
    input('press key to finish\n')

【讨论】：