为什么`pickle.dump`+`pickle.load` IPC 这么慢，有没有快速的替代方案？

Question

我正在使用 python 子进程进行 IPC。现在，让我们假设我必须使用subprocess.Popen来生成另一个进程，所以我不能multiprocessing.Pipe用于通信。我想到的第一件事是将他们的 STDIO 流与pickle.load+一起使用pickle.dump（现在不要担心安全性）。

但是，我注意到，传输速率非常糟糕：我的机器上的顺序为 750KB/s！这比通过通信慢multiprocessing.Pipe95 倍pickle，据我所知，它也使用。使用cPickle也没有任何好处。

（更新：注意，我意识到，这只是在 python2 上的情况！在 python3 上它工作正常。）

为什么这么慢？我怀疑原因是在.dump/.load通过 python 文件对象而不是 C 文件描述符中执行 IO 的方式。也许它与GIL有关？

有什么方法可以跨平台获得与 相同的速度multiprocessing.Pipe吗？

我已经发现，在 linux 上可以使用_multiprocessing.Connection（或multiprocessing.connection.Connection在 python3 上）来包装子进程的 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排序的top调用是：

      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}

这让我担心，使用readline可能是性能不佳的原因。

以下连接仅将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()

事实上，它的速度只比multiprocessing.Pipe. （这还是很可怕的）

现在在父级中进行分析：

      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}

所以，我仍然没有真正的线索，用什么来代替。

Answer 1

对此答案的评论建议使用：

pickle.dump(data, file, -1)

即，将协议设置为可用的最新版本。事实上，这只会比我的机器上的速度差multiprocessing.Pipe大约 1.7 倍。使用cPickle将其提高到大约 1.4 倍。