Python: Numpy.min ersetzt Builtin: verursacht Pyro4 Fehler: Antwort-Sequenz nicht synchron

Question

EDIT: Es stellte sich heraus, dass es sich um einen Dill-Fehler handelt, siehe Antwort.

EDIT: Blick auf den Grund für meine Fortschritte bei der Problembehandlung, die durch numpy.min verursacht werden stellt sich heraus, das intern min Funktion gebaut

ich mit Pyro arbeite (Version 4.43, Python 3.5 .1, Windows 10) und versucht, einen einfachen Cluster einzurichten, in dem ein Serverprozess auf die Arbeit von Arbeitern und Arbeitsprozessen wartet und Ergebnisse sendet. Sobald der Server das Ergebnis erhält, führt er eine weitere Verarbeitung durch.

Momentan versuche ich nur, es funktioniert auf einem einzigen Computer (mit localhost und launching Worker-Prozesse vom selben Computer).

Bis jetzt kann ich den Serverprozess ausführen, und der Arbeitsprozess ist in der Lage, eine Verbindung zum Server herzustellen, um Daten anzufordern, diese Daten zu verarbeiten, aber dann verarbeitet der Worker Fehler, wenn er versucht, das Ergebnis zurückzusenden zum Server.

Ich bin in eine seltsame Fehlermeldung ausgeführt wird:

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 

Nach gründlicher Suche, kann ich nur eine other person who has had this error finden, aber die Antwort war, dass dies ein reiner Pyro Fehler war, und er brauchte Pyro zu aktualisieren , aber meine Version ist weit jenseits der aktuellen, als das geschrieben wurde.

Noch weiter, ich habe Probleme, diesen Fehler außerhalb meines Produktionscodes zu reproduzieren. Ich habe versucht, eine einfache Version zu erstellen, um einzugrenzen, woher der Fehler kommt und konnte diesen Fehler nicht bekommen. Ich schickte sogar ein Ergebnis von einem Arbeiter der genauen Form des Ergebnisses, das im Produktionscode ohne Fehler gesendet wurde.

Hier ist der vereinfachte Code, nur um eine Vorstellung von der Struktur meines Setups zu geben. Dieser Code reproduziert den Fehler nicht. Ich bin mir nicht sicher, was der nächste Schritt ist, um es näher an den Produktionscode zu bringen, ohne übermäßig zu komplizieren.

Server Code:

#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()  

Worker Code:

#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-Code:

#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 

Hinweis: Ich brauche Dill mit der rekursiven Option zu verwenden, um diese Arten zu senden von Daten

Wenn ich Pyro4.current_context.seq innerhalb des Worker-Hauptgeräts drucke, wird 0 zurückgegeben. Wenn ich Pyro4.current_context.seq = 2 versuche, hat dies keine Auswirkungen auf den Fehler.

Weiß jemand, wie man mit diesem Fehler umgeht oder was ich als nächstes tun sollte, um Fehler zu beheben?

EDIT: Nach einer Überprüfung der Pyro4-Quelle scheint es, dass dieser Fehler aufgrund eines Codierungsfehlers in Pyro4 ausgelöst wird. Wenn es in core.Daemon.handleRequest einen Fehler beim Empfang der Nachricht hat, setzt es seine eigene Nachrichtensequenz auf Null und versucht, den Fehler als Nachricht zu senden. Aber wenn core.Proxy._pyroInvoke empfängt die Nachricht, es hat keine Möglichkeit, es als einen Fehler zu behandeln, wenn die Sequenz Null ist. Somit wird der Fehler "Synchronisierungsfehler der Antwortsequenz" ausgelöst.

Ich habe herausgefunden, das zugrunde liegende Problem, das den Fehler beim Empfang der Nachricht verursacht. socketutil.receiveData hat eine Empfangsschleife mit einer Zeile, die das Minimum von 60000 und die verbleibende Größe der Nachricht min(60000, size - msglen) annimmt. Irgendwie, wenn dies ausgeführt wird, verwendet es numpy.min anstelle der eingebauten Min. Und Fehler, weil das zweite Argument von numpy.min die Achsennummer sein soll. Das ist überraschend, da ich nur numpy as np in meinen Code importiere und nie from numpy import * oder die Min-Funktion direkt importiere.

Was noch überraschender ist, dass ich es nicht reparieren kann, indem ich es durch die eingebaute Funktion ersetze. Ich versuche import builtins dann min = builtins.min, und der Fehler bleibt bestehen. Wenn ich inspect.getfile(builtins.min) ausführen, zeigt es auf die Numpy-Datei.

Ich habe versucht, das Problem vollständig zu vermeiden, indem Sie die Zeile für min([60000, size - msglen]), die sowohl für numpy als auch für eingebaute Min funktioniert, aber die Min-Zuweisung bleibt in meinem Server Code und vermasselt Funktionen dort auch.

Als eher hackish fix, ich die obige Änderung der Min.-Funktion gehalten, sondern auch bei der Initialisierung meiner Serverklasse, speichere ich die eingebauten Funktionen:

#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 

Dann wird der Server jedes Mal empfängt oder sendet Daten, ich führe diese Funktion zuerst aus:

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'] 

Dies scheint jetzt zu arbeiten. Aber das ist offensichtlich kein guter Weg, um das Problem zu beheben, ich würde es lieber davon abhalten, die eingebauten Funktionen zu ersetzen ... Ist das ein Pyro-spezifisches Problem?

Answer 1

Dies ist ein Dill-Fehler, der durch das Beizen eines lambifizierten Sympy-Ausdrucks verursacht wird. Der folgende Code reproduziert den Fehler:

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)) 

kehrt:

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

ich diese issue #167 wie Dill eingereicht haben.