原文地址:
http://xiaorui.cc/archives/3290
============================================================
摘抄内容:
接着粗略的聊聊multiprocessing共享内存的实现方法.
multiprocessing提前设定的ctypes映射表,这样对你来说只需要传递符号就可以了。
typecode_to_type = { 'c': ctypes.c_char, 'u': ctypes.c_wchar, 'b': ctypes.c_byte, 'B': ctypes.c_ubyte, 'h': ctypes.c_short, 'H': ctypes.c_ushort, 'i': ctypes.c_int, 'I': ctypes.c_uint, 'l': ctypes.c_long, 'L': ctypes.c_ulong, 'f': ctypes.c_float, 'd': ctypes.c_double }
下面的代码通过mmap ctypes实现了一个简单的数据共享, 这样对于我们来说,可以像操作python对象那样操作映射的内存地址
(经过修改后的Python3.7代码)
### a.py
import ctypes import mmap import os import struct import time def main(): fd = os.open('/tmp/mmaptest', os.O_CREAT | os.O_TRUNC | os.O_RDWR) assert os.write(fd, b'x00' * mmap.PAGESIZE) == mmap.PAGESIZE buf = mmap.mmap(fd, 0, mmap.MAP_SHARED, mmap.PROT_WRITE) i = ctypes.c_int.from_buffer(buf) # Before we create a new value, we need to find the offset of the next free # memory address within the mmap # print(offset, ctypes.sizeof(i)) # byte length offset = struct.calcsize(i._type_) # The offset should be uninitialized ('x00') #assert buf[offset] == b'x00' #print(type(buf[offset])) assert buf[offset] == 0x00 # Now ceate a string containing 'foo' by first creating a c_char array s_type = ctypes.c_char * len('foot') # Now create the ctypes instance s = s_type.from_buffer(buf, offset) s.raw = b'foot' i.value = 10 i.value *= i.value print( 'Changing i : {}'.format(i.value) ) s.raw = b'bara' print( 'Changing s : {}'.format(s.raw) ) if __name__ == '__main__': main() time.sleep(60)
运行结果:
### b.py
import mmap import os import struct import time def main(): # Open the file for reading fd = os.open('/tmp/mmaptest', os.O_RDONLY) # Memory map the file buf = mmap.mmap(fd, 0, mmap.MAP_SHARED, mmap.PROT_READ) i = None s = None new_i, = struct.unpack('i', buf[:4]) new_s, = struct.unpack('4s', buf[4:8]) if i != new_i or s != new_s: print( 'i: %s => %d' % (i, new_i) ) print( 's: %s => %s' % (s, new_s) ) i = new_i s = new_s if __name__ == '__main__': main()
运行结果: