有没有一种很好的方法可以在不使用磁盘的情况下在两个python子进程之间传递大量数据？这是我希望完成的动画示例：
import sys, subprocess, numpy
cmdString = """
import sys, numpy
done = False
while not done:
cmd = raw_input()
if cmd == 'done':
done = True
elif cmd == 'data':
##Fake data. In real life, get data from hardware.
data = numpy.zeros(1000000, dtype=numpy.uint8)
data.dump('data.pkl')
sys.stdout.write('data.pkl' + '\\n')
sys.stdout.flush()"""
proc = subprocess.Popen( #python vs. pythonw on Windows?
[sys.executable, '-c %s'%cmdString],
stdin=subprocess.PIPE,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE)
for i in range(3):
proc.stdin.write('data\n')
print proc.stdout.readline().rstrip()
a = numpy.load('data.pkl')
print a.shape
proc.stdin.write('done\n')
这将创建一个子进程，该子进程生成numpy数组并将数组保存到磁盘。然后父进程从磁盘加载数组。有用！
问题是，我们的硬件可以生成比磁盘可读/写快10倍的数据。有没有办法将数据从一个python进程传输到另一个纯内存中，甚至可能没有复制数据？我可以做一些像传递参考的东西吗？
我第一次尝试纯粹在内存中传输数据是非常糟糕的：
import sys, subprocess, numpy
cmdString = """
import sys, numpy
done = False
while not done:
cmd = raw_input()
if cmd == 'done':
done = True
elif cmd == 'data':
##Fake data. In real life, get data from hardware.
data = numpy.zeros(1000000, dtype=numpy.uint8)
##Note that this is NFG if there's a '10' in the array:
sys.stdout.write(data.tostring() + '\\n')
sys.stdout.flush()"""
proc = subprocess.Popen( #python vs. pythonw on Windows?
[sys.executable, '-c %s'%cmdString],
stdin=subprocess.PIPE,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE)
for i in range(3):
proc.stdin.write('data\n')
a = numpy.fromstring(proc.stdout.readline().rstrip(), dtype=numpy.uint8)
print a.shape
proc.stdin.write('done\n')
这非常慢（比保存到磁盘慢得多）而且非常非常脆弱。必须有更好的方法！
我没有与'subprocess'模块结合，只要数据获取过程不会阻止父应用程序。我简单地尝试了“多处理”，但到目前为止还没有成功。
背景：我们有一块硬件可以在一系列ctypes缓冲区中生成高达~2 GB / s的数据。处理这些缓冲区的python代码完全处理大量信息。我希望将这个信息流与在“主”程序中同时运行的其他几个硬件进行协调，而不会使子进程相互阻塞。我目前的方法是在保存到磁盘之前将数据在子进程中稍微降低一点，但将完整的monty传递给'master'进程会很好。
在谷歌搜索有关Joe Kington发布的代码的更多信息时，我找到了numpy-sharedmem包。从这个numpy/multiprocessing tutorial来看，它似乎拥有相同的知识遗产（可能大部分是同一作者？ - 我不确定）。
使用sharedmem模块，您可以创建一个共享内存numpy数组（太棒了！），并将它与multiprocessing一起使用，如下所示：
import sharedmem as shm
import numpy as np
import multiprocessing as mp
def worker(q,arr):
done = False
while not done:
cmd = q.get()
if cmd == 'done':
done = True
elif cmd == 'data':
##Fake data. In real life, get data from hardware.
rnd=np.random.randint(100)
print('rnd={0}'.format(rnd))
arr[:]=rnd
q.task_done()
if __name__=='__main__':
N=10
arr=shm.zeros(N,dtype=np.uint8)
q=mp.JoinableQueue()
proc = mp.Process(target=worker, args=[q,arr])
proc.daemon=True
proc.start()
for i in range(3):
q.put('data')
# Wait for the computation to finish
q.join()
print arr.shape
print(arr)
q.put('done')
proc.join()
运行收益率
rnd=53
(10,)
[53 53 53 53 53 53 53 53 53 53]
rnd=15
(10,)
[15 15 15 15 15 15 15 15 15 15]
rnd=87
(10,)
[87 87 87 87 87 87 87 87 87 87]
基本上，你只想在进程之间共享一块内存并将其视为一个numpy数组，对吧？
在这种情况下，看看这个（由Nadav Horesh发布到numpy讨论一段时间，而不是我的工作）。有几个类似的实现（一些更灵活），但它们都基本上使用这个原则。
# "Using Python, multiprocessing and NumPy/SciPy for parallel numerical computing"
# Modified and corrected by Nadav Horesh, Mar 2010
# No rights reserved
import numpy as N
import ctypes
import multiprocessing as MP
_ctypes_to_numpy = {
ctypes.c_char : N.dtype(N.uint8),
ctypes.c_wchar : N.dtype(N.int16),
ctypes.c_byte : N.dtype(N.int8),
ctypes.c_ubyte : N.dtype(N.uint8),
ctypes.c_short : N.dtype(N.int16),
ctypes.c_ushort : N.dtype(N.uint16),
ctypes.c_int : N.dtype(N.int32),
ctypes.c_uint : N.dtype(N.uint32),
ctypes.c_long : N.dtype(N.int64),
ctypes.c_ulong : N.dtype(N.uint64),
ctypes.c_float : N.dtype(N.float32),
ctypes.c_double : N.dtype(N.float64)}
_numpy_to_ctypes = dict(zip(_ctypes_to_numpy.values(), _ctypes_to_numpy.keys()))
def shmem_as_ndarray(raw_array, shape=None ):
address = raw_array._obj._wrapper.get_address()
size = len(raw_array)
if (shape is None) or (N.asarray(shape).prod() != size):
shape = (size,)
elif type(shape) is int:
shape = (shape,)
else:
shape = tuple(shape)
dtype = _ctypes_to_numpy[raw_array._obj._type_]
class Dummy(object): pass
d = Dummy()
d.__array_interface__ = {
'data' : (address, False),
'typestr' : dtype.str,
'descr' : dtype.descr,
'shape' : shape,
'strides' : None,
'version' : 3}
return N.asarray(d)
def empty_shared_array(shape, dtype, lock=True):
'''
Generate an empty MP shared array given ndarray parameters
'''
if type(shape) is not int:
shape = N.asarray(shape).prod()
try:
c_type = _numpy_to_ctypes[dtype]
except KeyError:
c_type = _numpy_to_ctypes[N.dtype(dtype)]
return MP.Array(c_type, shape, lock=lock)
def emptylike_shared_array(ndarray, lock=True):
'Generate a empty shared array with size and dtype of a given array'
return empty_shared_array(ndarray.size, ndarray.dtype, lock)
从其他答案来看，似乎numpy-sharedmem是要走的路。
但是，如果您需要一个纯python解决方案，或者安装扩展，cython等是一个（大）麻烦，您可能想要使用以下代码，这是Nadav代码的简化版本：
import numpy, ctypes, multiprocessing
_ctypes_to_numpy = {
ctypes.c_char : numpy.dtype(numpy.uint8),
ctypes.c_wchar : numpy.dtype(numpy.int16),
ctypes.c_byte : numpy.dtype(numpy.int8),
ctypes.c_ubyte : numpy.dtype(numpy.uint8),
ctypes.c_short : numpy.dtype(numpy.int16),
ctypes.c_ushort : numpy.dtype(numpy.uint16),
ctypes.c_int : numpy.dtype(numpy.int32),
ctypes.c_uint : numpy.dtype(numpy.uint32),
ctypes.c_long : numpy.dtype(numpy.int64),
ctypes.c_ulong : numpy.dtype(numpy.uint64),
ctypes.c_float : numpy.dtype(numpy.float32),
ctypes.c_double : numpy.dtype(numpy.float64)}
_numpy_to_ctypes = dict(zip(_ctypes_to_numpy.values(),
_ctypes_to_numpy.keys()))
def shm_as_ndarray(mp_array, shape = None):
'''Given a multiprocessing.Array, returns an ndarray pointing to
the same data.'''
# support SynchronizedArray:
if not hasattr(mp_array, '_type_'):
mp_array = mp_array.get_obj()
dtype = _ctypes_to_numpy[mp_array._type_]
result = numpy.frombuffer(mp_array, dtype)
if shape is not None:
result = result.reshape(shape)
return numpy.asarray(result)
def ndarray_to_shm(array, lock = False):
'''Generate an 1D multiprocessing.Array containing the data from
the passed ndarray. The data will be *copied* into shared
memory.'''
array1d = array.ravel(order = 'A')
try:
c_type = _numpy_to_ctypes[array1d.dtype]
except KeyError:
c_type = _numpy_to_ctypes[numpy.dtype(array1d.dtype)]
result = multiprocessing.Array(c_type, array1d.size, lock = lock)
shm_as_ndarray(result)[:] = array1d
return result
你会像这样使用它：
sa = ndarray_to_shm(a)。a（和multiprocessing.Array，也许是multiprocessing.Process(target = somefunc, args = (sa, )）在单独的start中调用join，传递共享数组。somefunc中，使用process获取指向共享数据的ndarray。 （实际上，您可能希望在创建somefunc之后立即在原始过程中执行相同操作，以便有两个ndarray引用相同的数据。）AFAICS，您不需要将lock设置为True，因为a = shm_as_ndarray(sa)无论如何都不会使用锁定。如果需要锁定，可以将lock设置为True并在sa上调用acquire / release。
此外，如果您的数组不是一维的，您可能希望将形状与sa一起传输（例如，使用shm_as_ndarray）。
该解决方案的优点是除了多处理（在标准库中）之外，它不需要额外的包或扩展模块。
使用线程。但我猜你会遇到GIL的问题。
相反：选择你的sa。
我从我使用的MPI实现中了解到，他们使用共享内存进行节点间通信。在这种情况下，您必须编写自己的同步代码。
2 GB / s听起来像你会遇到大多数“简单”方法的问题，具体取决于你的实时约束和可用的主存储器。
使用线程。您可能不会遇到GIL问题。
GIL只影响Python代码，而不影响C / Fortran / Cython支持的库。大多数numpy操作和C-backed Scientific Python堆栈的很大一部分都会释放GIL，并且可以在多个内核上运行良好。 args = (sa, a.shape)更深入地讨论了GIL和科学Python。
使用线程的简单方法包括poison模块和This blogpost。
考虑的一种可能性是使用threading临时存储要在进程之间共享的文件。 RAM驱动器将RAM的一部分视为逻辑硬盘驱动器，可以像使用常规驱动器一样写入/读取文件，但RAM读/写速度。
本文介绍如何使用ImDisk软件（用于MS Win）创建此类磁盘并获得6-10 GB /秒的文件读/写速度：multiprocessing.pool.ThreadPool
Ubuntu中的一个例子：RAM drive
另一个显着的好处是具有任意格式的文件可以通过这种方法传递：例如， Picke，JSON，XML，CSV，HDF5等......
请记住，重新启动时会擦除RAM磁盘上存储的所有内容。