python - Python:Numpy.min替换了内置函数:导致Pyro4错误:回复序列不同步

Question

编辑：原来是一个莳萝错误，请参阅答案。

编辑：寻找我的错误排查进度的底部，原来是由numpy.min取代了内置的min函数引起的

我正在使用Pyro（版本4.43，Python 3.5.1，Windows 10）并尝试建立一个简单的集群，服务器进程在其中等待工作程序，工作程序请求工作并发回结果。服务器收到结果后，将对其进行进一步处理。

目前，我只是想让它在一台计算机上工作（使用localhost并从同一台计算机生成工作进程）。

到目前为止，我已经能够使服务器进程运行，并且工作进程能够连接到服务器以请求数据，处理该数据，但是当工作进程尝试将结果发送回服务器时，工作进程会出错。

我遇到一个奇怪的错误消息：

File "worker.py", line 90, in <module>
    main()
  File "worker.py", line 87, in main
    worker.send_result()
  File "worker.py", line 49, in send_result
    self.server.recieve(result)
  File "C:\Anaconda3\lib\site-packages\Pyro4\core.py", line 171, in __call__
    return self.__send(self.__name, args, kwargs)
  File "C:\Anaconda3\lib\site-packages\Pyro4\core.py", line 418, in _pyroInvoke
    self.__pyroCheckSequence(msg.seq)
  File "C:\Anaconda3\lib\site-packages\Pyro4\core.py", line 448, in __pyroCheckSequence
    raise errors.ProtocolError(err)
Pyro4.errors.ProtocolError: invoke: reply sequence out of sync, got 0 expected 2

彻底搜索之后，我只能找到一个 other person who has had this error，但是响应是这是一个纯粹的Pyro错误，他需要更新Pyro，但是我的版本远远超出了编写该版本时的当前版本。

更进一步，我无法在生产代码之外重现此错误。我试图创建一个简单的版本来缩小错误的出处，但无法得到此错误。我什至从工作人员发送的结果与在生产代码中发送的结果的准确形式相同，没有错误。

这是简化的代码，仅是为了介绍我的设置结构。下面的代码不会重现该错误。我不确定下一步如何使它更接近生产代码而不会过于复杂。

服务器代码：

#simple_server.py

import Pyro4
import sys, dill

class SimpleServer:

    def serve(self):
        with open('served data.pkl', 'rb') as f:
            data = dill.load(f) #actual data coming from production code
        return data

    def recieve(self, result):
        print(result)

def main():
    Pyro4.config.SERIALIZER = 'dill' #default serpent serializer doesn't work
    dill.settings['recurse'] = True #dill won't work without this option

    server = SimpleServer()
    daemon = Pyro4.Daemon()
    server_uri = daemon.register(server)
    ns = Pyro4.locateNS()
    ns.register("test", server_uri)
    print('Server running.')
    daemon.requestLoop()

if __name__ == '__main__':
    main()    

工作人员代码：

#simple_worker.py

import Pyro4
import sys, dill
import numpy as np
import scipy.optimize as opt

class SimpleWorker:

    def __init__(self, server):
        self.server = server

    def recieve_data(self):
        self.data = self.server.serve()

    def send_result(self):
        res = opt.basinhopping(lambda x: sum(x), np.arange(11), niter=2, minimizer_kwargs={'options':{'maxiter':2}})
        #This below data structure is the same that I send in production
        result = ('ABCD', 'filename.csv', res, 6)
        self.server.recieve(result) #creates error in production code but not here

def main():
    sys.excepthook = Pyro4.util.excepthook #gives a more meaningful stack trace
    Pyro4.config.SERIALIZER = 'dill' #default serpent serializer doesn't work
    dill.settings['recurse'] = True #dill won't work without this option

    server = Pyro4.Proxy('PYRONAME:test') #connects to pinest server
    worker = SimpleWorker(server)
    worker.recieve_data() 
    worker.send_result()

if __name__ == '__main__':
    main()

Windows CMD代码：

#run_simple_server.bat
set PYRO_SERIALIZERS_ACCEPTED=serpent,json,marshal,pickle,dill
start cmd /C python -m Pyro4.naming
python simple_server.py
pause

#run_simple_worker.bat
python simple_worker.py
pause

注意：我需要将Dill与递归选项一起使用以发送这些类型的数据

如果在辅助主程序中打印 Pyro4.current_context.seq，它将返回0。如果尝试 Pyro4.current_context.seq = 2，则不会影响该错误。

有谁知道如何处理此错误，或者在尝试进行故障排除时下一步该怎么做？

编辑：在审查了Pyro4源之后，似乎由于Pyro4中的编码错误而引发了该错误。在core.Daemon.handleRequest中，如果接收消息时发生错误，则会将其自己的消息序列设置为零，并尝试将错误作为消息进行传输。但是当core.Proxy._pyroInvoke接收到该消息时，如果序列为零，则它不具有将其视为错误的功能。因此，引发了应答序列不同步错误。

我已经弄清楚了导致错误消息的根本原因。 socketutil.receiveData具有一个接收循环，该循环的一行选择最小值为60000，其余部分为消息 min(60000, size - msglen)。当执行该命令时，它使用的是numpy.min而不是内置的min，并且会出错，因为numpy.min的第二个参数应该是轴号。这是令人惊讶的，因为我只在代码中导入了 numpy as np，而从不导入 from numpy import *或直接导入min函数。

更令人惊讶的是，我无法通过将其替换为内置函数来修复它。我先尝试 import builtins然后 min = builtins.min，错误仍然存​​在。如果运行 inspect.getfile(builtins.min)，它将指向Numpy文件。

我试图通过切换 min([60000, size - msglen])的行来完全避免该问题，该行适用于numpy和内置的min，但是min分配仍然保留在我的服务器代码中，并且在那里也弄乱了功能。

作为一个相当棘手的修复程序，我保留了min函数的上述更改，但在服务器类初始化时，我存储了内置函数：

#Store builtin functions as they later get replaced for some unknown reason
b = [t for t in ().__class__.__base__.__subclasses__() if t.__name__ == 'Sized'][0].__len__.__globals__['__builtins__']
self.real_builtins = copy.copy(b) #copy so that dict doesn't get updated

然后，每次服务器接收或发送数据时，我首先运行此函数：

def fix_builtins(self):
    global builtins
    import builtins
    __builtins__ = self.real_builtins
    #These are all of [i for i in dir(builtins) if i in dir(numpy)]
    builtins.abs = __builtins__['abs']
    builtins.all = __builtins__['all']
    builtins.any = __builtins__['any']
    builtins.bool = __builtins__['bool']
    builtins.complex = __builtins__['complex']
    builtins.float = __builtins__['float']
    builtins.int = __builtins__['int']
    builtins.max = __builtins__['max']
    builtins.min = __builtins__['min']
    builtins.object = __builtins__['object']
    builtins.round = __builtins__['round']
    builtins.str = __builtins__['str']
    builtins.sum = __builtins__['sum']

这似乎现在正在工作。但这显然不是解决此问题的好方法，我宁愿从一开始就阻止它取代内置函数...这是某些Pyro特定的问题吗？

Answer 1

这是一个莳萝错误，它是由腌制经过lambdified的Sympy表达式引起的。下面的代码重现该错误：

from sympy import symbols, lambdify
import dill, inspect

def check_if_builtin(func):
    try:
        file = inspect.getsourcefile(func) #will throw TypeError for builtin
        return file
    except TypeError:
        return True



dill.settings['recurse'] = True #without this option, throws PicklingError

a, b, c = symbols("a b c")
expr = a + b + c
lambda_expr = lambdify([a, b, c], expr)

print(check_if_builtin(min))

dill.dump(lambda_expr, open('test.p', 'wb'))

print(check_if_builtin(min))

返回：

True
C:\Anaconda3\lib\site-packages\numpy\core\fromnumeric.py

我已将此提交为莳萝 issue #167。