Pipe brisée lors de l'utilisation de gestionnaires multiprocesseurs Python (BaseManager/SyncManager) pour partager une file d'attente avec des machines distantes

Question

Le mois dernier, nous avons rencontré un problème persistant avec le package multitraitement Python 2.6.x lorsque nous avons essayé de l'utiliser pour partager une file d'attente entre plusieurs ordinateurs (linux) différents. J'ai posé cette question directement à Jesse Noller, puisque nous n'avons pas encore trouvé quelque chose qui élucide le problème sur StackOverflow, les docs Python, le code source ou ailleurs en ligne.

Notre équipe d'ingénieurs n'a pas été en mesure de résoudre celle-ci, et nous avons posé la question à un certain nombre de personnes dans les groupes d'utilisateurs python en vain. J'espérais que quelqu'un pourrait donner un aperçu, car j'ai l'impression que nous faisons quelque chose d'incorrect mais nous sommes trop proches du problème pour le voir pour ce qu'il est.

est ici le symptôme:

Traceback (most recent call last): 
    File "/var/django_root/dev/com/brightscope/data/processes/daemons/deferredupdates/servers/queue_server.py", line 65, in get_from_queue 
    return queue, queue.get(block=False) 
    File "<string>", line 2, in get 
    File "/usr/local/lib/python2.6/multiprocessing/managers.py", line 725, in _callmethod 
    conn.send((self._id, methodname, args, kwds)) 
IOError: [Errno 32] Broken pipe 

(je montre où notre code appelle queue.get() sur un objet de file d'attente partagée, hébergé par un gestionnaire qui va SyncManger). Ce qui est curieux à propos du problème, c'est que si nous nous connectons à cette file d'attente partagée sur une seule machine (appelons cela machine A), même à partir de nombreux processus concurrents, nous ne semblons jamais rencontrer de problème. C'est seulement quand nous nous connectons à la file d'attente (encore une fois, en utilisant une classe qui étend SyncManager multiprocesseur et n'ajoute actuellement aucune fonctionnalité supplémentaire) d'autres machines (appelons ces machines B and C) et courons un volume élevé d'articles dans et hors de la file d'attente temps que nous éprouvons un problème. Tout se passe comme si le paquetage multitraitement de python traitait les connexions locales (même si elles utilisent toujours la même méthode de connexion manager.connect()) d'une manière qui fonctionne à partir de machine A mais lorsque les connexions à distance sont faites simultanément à partir d'au moins machines B or C nous obtenons une erreur de tuyau cassé.

Dans toutes les lectures que mon équipe a faites, nous avons pensé que le problème était lié au verrouillage. Nous avons pensé que nous ne devrions peut-être pas utiliser Queue.Queue, mais plutôt multiprocessing.Queue, mais nous avons changé et le problème a persisté (nous avons également remarqué que la propre file d'attente partagée de SyncManager est une instance de Queue.Queue). Nous essayons de résoudre le problème, car il est difficile à reproduire, mais arrive assez souvent (plusieurs fois par jour si nous insérons et .get() beaucoup d'éléments de la file d'attente) . La méthode que nous avons créée get_from_queue tente de réessayer d'acquérir l'élément à partir d'une file d'attente ~ 10 fois avec des intervalles de sommeil aléatoires, mais il semble que si elle échoue une fois, elle échouera dix fois (ce qui me porte à croire que .register() et .connect() à un gestionnaire ne donne peut-être pas une autre connexion socket au serveur, mais je ne pouvais pas le confirmer en lisant les docs ou en regardant le code source interne de Python). Quelqu'un peut-il nous donner un aperçu de l'endroit où nous pourrions regarder ou comment nous pourrions suivre ce qui se passe réellement?

Comment pouvons-nous commencer une nouvelle connexion en cas d'un bris de tuyau en utilisant multiprocessing.BaseManager ou multiprocessing.SyncManager?

Comment pouvons-nous empêcher le tuyau cassé en premier lieu?

Answer 1

FYI Si quelqu'un d'autre exécute cette même erreur, après une longue consultation avec Ask Solem et Jesse Noller de l'équipe de développement de base de Python, il semble que ce soit un bug dans python 2.6.x (et éventuellement 2.7+ et éventuellement 3.x).Ils cherchent des solutions possibles et un correctif sera probablement inclus dans une future version de Python.

Answer 2

J'ai rencontré le même problème, même si je me suis connecté sur localhost en python 2.7.1. Après une journée de débogage je l'ai trouvé la cause et une solution de contournement:

Cause: classe BaseProxy a fil stockage local qui met en cache la connexion, qui est réutilisée pour les connexions futures provoquant des erreurs de « broken pipe », même sur la création d'un nouveau directeur

Solution: Supprimez la connexion en cache avant de reconnecter

if address in BaseProxy._address_to_local: 
    del BaseProxy._address_to_local[address][0].connection 

Answer 3

vous pouvez également essayer d'attraper l'exception dans les processus de l'enfant, de sorte qu'il ne devrait pas essayer de fermer la connexion ONU-expectedly. La même chose arrivait à moi et finalement je devais supprimer les erreurs pour que le tuyau ne se ferme pas brusquement.