我正在使用 python子进程进行IPC.现在,让我们假设我必须使用subprocess.Popen来生成其他进程,所以我不能使用multiprocessing.Pipe进行通信.首先,我想到的是将他们的STDIO流与pickle.load pickle.dump一起
但是,我注意到,传输速率非常糟糕：我的机器上的订单量为750KB / s！这比通过多处理进行通信慢.管道使用95因子,据我所知,它也使用了pickle.使用cPickle也没有任何好处.
(更新：注意,我意识到,这只是python2上的情况！在python3上它工作正常.)
为什么这么慢？我怀疑原因是在.dump / .load中通过python文件对象而不是C文件描述符执行IO的方式.也许它与GIL有关？
是否有任何方式跨平台的方式来获得与多处理相同的速度.管道？
我已经发现,在linux上可以使用_multiprocessing.Connection(或python3上的multiprocessing.connection.Connection)来包装子进程的STDIO文件描述符并得到我想要的东西.但是,这在win32上是不可能的,我甚至不知道Mac.
基准测试：
from __future__ import print_function
from timeit import default_timer
from subprocess import Popen, PIPE
import pickle
import sys
import os
import numpy
try:
from _multiprocessing import Connection as _Connection
except ImportError:
from multiprocessing.connection import Connection as _Connection
def main(args):
if args:
worker(connect(args[0], sys.stdin, sys.stdout))
else:
benchmark()
def worker(conn):
while True:
try:
amount = conn.recv()
except EOFError:
break
else:
conn.send(numpy.random.random(amount))
conn.close()
def benchmark():
for amount in numpy.arange(11)*10000:
pickle = parent('pickle', amount, 1)
pipe = parent('pipe', amount, 1)
print(pickle[0]/1000, pickle[1], pipe[1])
def parent(channel, amount, repeat):
start = default_timer()
proc = Popen([sys.executable, '-u', __file__, channel],
stdin=PIPE, stdout=PIPE)
conn = connect(channel, proc.stdout, proc.stdin)
for i in range(repeat):
conn.send(amount)
data = conn.recv()
conn.close()
end = default_timer()
return data.nbytes, end - start
class PickleConnection(object):
def __init__(self, recv, send):
self._recv = recv
self._send = send
def recv(self):
return pickle.load(self._recv)
def send(self, data):
pickle.dump(data, self._send)
def close(self):
self._recv.close()
self._send.close()
class PipeConnection(object):
def __init__(self, recv_fd, send_fd):
self._recv = _Connection(recv_fd)
self._send = _Connection(send_fd)
def recv(self):
return self._recv.recv()
def send(self, data):
self._send.send(data)
def close(self):
self._recv.close()
self._send.close()
def connect(channel, recv, send):
recv_fd = os.dup(recv.fileno())
send_fd = os.dup(send.fileno())
recv.close()
send.close()
if channel == 'pipe':
return PipeConnection(recv_fd, send_fd)
elif channel == 'pickle':
return PickleConnection(os.fdopen(recv_fd, 'rb', 0),
os.fdopen(send_fd, 'wb', 0))
else:
raise ValueError("Invalid channel: %s" % channel)
if __name__ == '__main__':
main(sys.argv[1:])
结果：
非常感谢阅读,
托马斯
更新：
好的,所以我按照@martineau的建议对其进行了分析.对于具有固定值amount = 500000的单次运行的独立调用,获得以下结果.
在父进程中,按tottime排序的热门调用是：
11916 function calls (11825 primitive calls) in 5.382 seconds
Ordered by: internal time
ncalls tottime percall cumtime percall filename:lineno(function)
35 4.471 0.128 4.471 0.128 {method 'readline' of 'file' objects}
52 0.693 0.013 0.693 0.013 {method 'read' of 'file' objects}
4 0.062 0.016 0.063 0.016 {method 'decode' of 'str' objects}
在子流程中：
11978 function calls (11855 primitive calls) in 5.298 seconds
Ordered by: internal time
ncalls tottime percall cumtime percall filename:lineno(function)
52 4.476 0.086 4.476 0.086 {method 'write' of 'file' objects}
73 0.552 0.008 0.552 0.008 {repr}
3 0.112 0.037 0.112 0.037 {method 'read' of 'file' objects}
这让我很担心,使用读取线可能是性能不佳的原因.
以下连接仅使用pickle.dumps / pickle.loads和write / read.
class DumpsConnection(object):
def __init__(self, recv, send):
self._recv = recv
self._send = send
def recv(self):
raw_len = self._recvl(4)
content_len = struct.unpack('>I', raw_len)[0]
content = self._recvl(content_len)
return pickle.loads(content)
def send(self, data):
content = pickle.dumps(data)
self._send.write(struct.pack('>I', len(content)))
self._send.write(content)
def _recvl(self, size):
data = b''
while len(data) < size:
packet = self._recv.read(size - len(data))
if not packet:
raise EOFError
data += packet
return data
def close(self):
self._recv.close()
self._send.close()
实际上,它的速度只比多处理速度差14倍.管道. (哪个仍然很糟糕)
现在分析,在父母：
11935 function calls (11844 primitive calls) in 1.749 seconds
Ordered by: internal time
ncalls tottime percall cumtime percall filename:lineno(function)
2 1.385 0.692 1.385 0.692 {method 'read' of 'file' objects}
4 0.125 0.031 0.125 0.031 {method 'decode' of 'str' objects}
4 0.056 0.014 0.228 0.057 pickle.py:961(load_string)
在孩子：
11996 function calls (11873 primitive calls) in 1.627 seconds
Ordered by: internal time
ncalls tottime percall cumtime percall filename:lineno(function)
73 1.099 0.015 1.099 0.015 {repr}
3 0.231 0.077 0.231 0.077 {method 'read' of 'file' objects}
2 0.055 0.028 0.055 0.028 {method 'write' of 'file' objects}
所以,我仍然没有真正的线索,而是使用什么.
pickle / cPickle序列化numpy数组有一些问题：In [14]: timeit cPickle.dumps(numpy.random.random(1000)) 1000 loops, best of 3: 727 us per loop In [15]: timeit numpy.random.random(1000).dumps() 10000 loops, best of 3: 31.6 us per loop
问题只发生在序列化,反序列化很好：
In [16]: timeit cPickle.loads(numpy.random.random(1000).dumps()) 10000 loops, best of 3: 40 us per loop
你可以使用marshal模块,巫婆甚至更快(但不安全)：
In [19]: timeit marshal.loads(marshal.dumps(numpy.random.random(1000))) 10000 loops, best of 3: 29.8 us per loop
好吧我推荐msgpack,但是它没有numpy的支持,而且有一个拥有它的lib很慢,反正python-msgpack不支持缓冲区也没有zerocopy功能所以它不可能对numpy做有效的支持.