multitraitement.Piscine - PicklingError: ne Peut pas pickle <type de fil conducteur.lock'>: attribut de recherche de fil.serrure a échoué

multiprocessing.Pool me rend fou...

Je veux mettre à jour de nombreux paquets, et pour chacun d'entre eux, je dois vérifier si il y a plus de version ou pas. Ceci est fait par le check_one fonction.

Le code principal est dans le Updater.update méthode: j'ai créer le Pool objet et l'appel de la map() méthode.

Voici le code:

def check_one(args):
res, total, package, version = args
i = res.qsize()
logger.info('\r[{0:.1%} - {1}, {2} /{3}]',
i / float(total), package, i, total, addn=False)
try:
json = PyPIJson(package).retrieve()
new_version = Version(json['info']['version'])
except Exception as e:
logger.error('Error: Failed to fetch data for {0} ({1})', package, e)
return
if new_version > version:
res.put_nowait((package, version, new_version, json))
class Updater(FileManager):
# __init__ and other methods...
def update(self):    
logger.info('Searching for updates')
packages = Queue.Queue()
data = ((packages, self.set_len, dist.project_name, Version(dist.version)) \
for dist in self.working_set)
pool = multiprocessing.Pool()
pool.map(check_one, data)
pool.close()
pool.join()
while True:
try:
package, version, new_version, json = packages.get_nowait()
except Queue.Empty:
break
txt = 'A new release is avaiable for {0}: {1!s} (old {2}), update'.format(package,
new_version,
version)
u = logger.ask(txt, bool=('upgrade version', 'keep working version'), dont_ask=self.yes)
if u:
self.upgrade(package, json, new_version)
else:
logger.info('{0} has not been upgraded', package)
self._clean()
logger.success('Updating finished successfully')

Quand je le lance, j'obtiens cette erreur bizarre:

Searching for updates
Exception in thread Thread-1:
Traceback (most recent call last):
File "/usr/lib/python2.7/threading.py", line 552, in __bootstrap_inner
self.run()
File "/usr/lib/python2.7/threading.py", line 505, in run
self.__target(*self.__args, **self.__kwargs)
File "/usr/local/lib/python2.7/dist-packages/multiprocessing/pool.py", line 225, in _handle_tasks
put(task)
PicklingError: Can't pickle <type 'thread.lock'>: attribute lookup thread.lock failed
InformationsquelleAutor rubik | 2011-10-23
  1. 28

    multitraitement passe des tâches (qui comprennent check_one et data) pour les processus de travail par le biais d'un mp.SimpleQueue. Contrairement à Queue.Queues, tout mettre dans le mp.SimpleQueue doit être cliquable. Queue.Queues ne sont pas cliquable:

    import multiprocessing as mp
import Queue
def foo(queue):
pass
pool=mp.Pool()
q=Queue.Queue()
pool.map(foo,(q,))

    les rendements de cette exception:

    UnpickleableError: Cannot pickle <type 'thread.lock'> objects

    Votre data comprend packages, qui est une File d'attente.La file d'attente. Que pourrait être la source du problème.

    Voici une solution possible: Le Queue est utilisé à deux fins:

    1. pour trouver la taille approximative (en appelant qsize)
    2. pour stocker les résultats pour une récupération ultérieure.

    Au lieu de l'appeler qsize, à part une valeur entre plusieurs processus, nous pourrions utiliser un mp.Value.

    Au lieu de stocker les résultats dans une file d'attente, on peut (et devrait) être juste retour des valeurs à partir des appels à check_one. Le pool.map recueille les résultats dans une file d'attente de sa propre fabrication, et renvoie les résultats que la valeur de retour de pool.map.

    Par exemple:

    import multiprocessing as mp
import Queue
import random
import logging
# logger=mp.log_to_stderr(logging.DEBUG)
logger = logging.getLogger(__name__)
qsize = mp.Value('i', 1)
def check_one(args):
total, package, version = args
i = qsize.value
logger.info('\r[{0:.1%} - {1}, {2} /{3}]'.format(
i / float(total), package, i, total))
new_version = random.randrange(0,100)
qsize.value += 1
if new_version > version:
return (package, version, new_version, None)
else:
return None
def update():    
logger.info('Searching for updates')
set_len=10
data = ( (set_len, 'project-{0}'.format(i), random.randrange(0,100))
for i in range(set_len) )
pool = mp.Pool()
results = pool.map(check_one, data)
pool.close()
pool.join()
for result in results:
if result is None: continue
package, version, new_version, json = result
txt = 'A new release is avaiable for {0}: {1!s} (old {2}), update'.format(
package, new_version, version)
logger.info(txt)
logger.info('Updating finished successfully')
if __name__=='__main__':
logging.basicConfig(level=logging.DEBUG)
update()
    • Merci!!!! Mais maintenant, comment puis-je remplir la file d'attente? Je ne peut pas le faire à l'échelle mondiale... j'ai aussi essayé de faire check_one une méthode (avec ce hack: stackoverflow.com/questions/1816958/...), mais encore une fois, cela n'a pas fonctionné...
    • J'ai trouvé la solution ici: stackoverflow.com/questions/3217002/.... J'ai utiliser multiprocessing.Manager().Queue() au lieu de multiprocessing.Queue.
    • Comme pour les solutions de contournement, je pense que vous avez raison. Je vais corriger mon code. Merci encore!!
    • Merci beaucoup pour la correction.
    InformationsquelleAutor unutbu
  2. 7

    Après beaucoup de creuser sur un problème similaire...

    Il s'avère également que TOUT objet qui contient un filetage.Condition() de l'objet ne sera JAMAIS ne JAMAIS travailler avec le multitraitement.Piscine.

    Voici un exemple

    import multiprocessing as mp
import threading
class MyClass(object):
def __init__(self):
self.cond = threading.Condition()
def foo(mc):
pass
pool=mp.Pool()
mc=MyClass()
pool.map(foo,(mc,))

    Je suis de l'exécution de cette avec Python 2.7.5 et frappé de la même erreur:

    Exception in thread Thread-2:
Traceback (most recent call last):
File "/usr/lib64/python2.7/threading.py", line 811, in __bootstrap_inner
self.run()
File "/usr/lib64/python2.7/threading.py", line 764, in run
self.__target(*self.__args, **self.__kwargs)
File "/usr/lib64/python2.7/multiprocessing/pool.py", line 342, in _handle_tasks
put(task)
PicklingError: Can't pickle <type 'thread.lock'>: attribute lookup thread.lock failed

    Mais ensuite, il a couru sur python 3.4.1 et ce problème a été résolu.

    Bien que je n'ai pas trouvé utile solutions de contournement encore pour ceux d'entre nous sont encore sur la 2.7.x.

    • Comme de Python 3.4.3, threading.Lock ne peut toujours pas être marinés (même si, comme vous l'avez mentionné, threading.Condition peut être).
    • Je confirme que threading.Lock objets ne peuvent toujours pas être traitées en Python 3.6.2.
    • hmm...et encore 3.7
    InformationsquelleAutor jersey bean
  3. 1

    J'ai connu ce problème avec la version de python 3.6 sur docker. Changé la version 3.7.3 et il a été résolu.

    InformationsquelleAutor Akalanka Weerasooriya