TypeError: ne peut pas pickle _thread.verrouiller des objets

Essayez d'exécuter deux fonctions différentes en même temps avec des files d'attente partagées et obtenez une erreur...comment puis-je exécuter les deux fonctions en même temps avec une file d'attente partagée? C'est une version de Python 3.6 sur Windows 7.

from multiprocessing import Process
from queue import Queue
import logging

def main():
    x = DataGenerator()
    try:
        x.run()
    except Exception as e:
        logging.exception("message")


class DataGenerator:

    def __init__(self):
        logging.basicConfig(filename='testing.log', level=logging.INFO)

    def run(self):
        logging.info("Running Generator")
        queue = Queue()
        Process(target=self.package, args=(queue,)).start()
        logging.info("Process started to generate data")
        Process(target=self.send, args=(queue,)).start()
        logging.info("Process started to send data.")

    def package(self, queue): 
        while True:
            for i in range(16):
                datagram = bytearray()
                datagram.append(i)
                queue.put(datagram)

    def send(self, queue):
        byte_array = bytearray()
        while True:
            size_of__queue = queue.qsize()
            logging.info(" queue size %s", size_of_queue)
            if size_of_queue > 7:
                for i in range(1, 8):
                    packet = queue.get()
                    byte_array.append(packet)
                logging.info("Sending datagram ")
                print(str(datagram))
                byte_array(0)

if __name__ == "__main__":
    main()

Les journaux indiquent une erreur, j'ai essayé de lancer la console en tant qu'administrateur et je reçois le même message...

INFO:root:Running Generator
ERROR:root:message
Traceback (most recent call last):
  File "test.py", line 8, in main
    x.run()
  File "test.py", line 20, in run
    Process(target=self.package, args=(queue,)).start()
  File "C:\ProgramData\Miniconda3\lib\multiprocessing\process.py", line 105, in start
    self._popen = self._Popen(self)
  File "C:\ProgramData\Miniconda3\lib\multiprocessing\context.py", line 223, in _Popen
    return _default_context.get_context().Process._Popen(process_obj)
  File "C:\ProgramData\Miniconda3\lib\multiprocessing\context.py", line 322, in _Popen
    return Popen(process_obj)
  File "C:\ProgramData\Miniconda3\lib\multiprocessing\popen_spawn_win32.py", line 65, in __init__
    reduction.dump(process_obj, to_child)
  File "C:\ProgramData\Miniconda3\lib\multiprocessing\reduction.py", line 60, in dump
    ForkingPickler(file, protocol).dump(obj)
TypeError: can't pickle _thread.lock objects

queue.Queue est pour la communication inter-thread. multiprocessing.Queue est pour envoyer des choses entre les processus.
J'ai fait le changement de multiprocessing.Queue et qui a résolu le problème. Je vous remercie.

OriginalL'auteur Jonathan Kittell | 2017-05-23

python-3.x

4

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

Déplacer la file d'attente pour soi plutôt que comme un argument à vos fonctions package et send

OriginalL'auteur PvdL
1

Vous avez besoin de changer from queue import Queue à from multiprocessing import Queue.

La raison profonde est l'ancien de la File d'attente est conçu pour le filetage module File d'attente tandis que le dernier est pour le multitraitement.Module de processus.

Pour plus de détails, vous pouvez lire certains de code source ou de me contacter!

OriginalL'auteur user8675687

J'ai eu le même problème avec Pool() en Python 3.6.3.

D'erreur reçus: TypeError: can't pickle _thread.RLock objects

Disons que nous voulons ajouter un certain nombre num_to_add à chaque élément de la liste de certains num_list en parallèle. Le code est schématiquement comme ceci:

class DataGenerator:
    def __init__(self, num_list, num_to_add)
        self.num_list = num_list # e.g. [4,2,5,7]
        self.num_to_add = num_to_add # e.g. 1 

        self.run()

    def run(self):
        new_num_list = Manager().list()

        pool = Pool(processes=50)
        results = [pool.apply_async(run_parallel, (num, new_num_list)) 
                      for num in num_list]
        roots = [r.get() for r in results]
        pool.close()
        pool.terminate()
        pool.join()

    def run_parallel(self, num, shared_new_num_list):
        new_num = num + self.num_to_add # uses class parameter
        shared_new_num_list.append(new_num)

Le problème ici est que self en fonction run_parallel() ne peuvent pas être traitées comme c'est une instance de classe. Le déplacement de ce parallélisée fonction run_parallel() de la classe aidé. Mais ce n'est pas la meilleure solution en tant que cette fonction est probablement pas nécessaire d'utiliser les paramètres de classe comme self.num_to_add et ensuite vous devez passer en argument.

Solution:

def run_parallel(num, shared_new_num_list, to_add): # to_add is passed as an argument
    new_num = num + to_add
    shared_new_num_list.append(new_num)

class DataGenerator:
    def __init__(self, num_list, num_to_add)
        self.num_list = num_list # e.g. [4,2,5,7]
        self.num_to_add = num_to_add # e.g. 1

        self.run()

    def run(self):
        new_num_list = Manager().list()

        pool = Pool(processes=50)
        results = [pool.apply_async(run_parallel, (num, new_num_list, self.num_to_add)) # num_to_add is passed as an argument
                      for num in num_list]
        roots = [r.get() for r in results]
        pool.close()
        pool.terminate()
        pool.join()

D'autres suggestions ci-dessus ne m'a pas aidé.

OriginalL'auteur Marina

Vous devez vous connecter pour publier un commentaire.