Попытка использовать многопроцессорный модуль завершается неудачно с TypeError: __new __ () отсутствует 2 обязательных позиционных аргумента: 'tag' и 'attribute'

Question

Я пытаюсь использовать модуль Pool из multiprocessing для параллельного вызова функции, и она не работает с этими ошибками. Поиск вокруг не дает мне понятия, как решить проблему. здесь - похожий вопрос без ответа.

Вот мой ответ:

Process ForkPoolWorker-1:
Traceback (most recent call last):
  File "/usr/lib/python3.6/multiprocessing/process.py", line 258, in _bootstrap
    self.run()
  File "/usr/lib/python3.6/multiprocessing/process.py", line 93, in run
    self._target(*self._args, **self._kwargs)
  File "/usr/lib/python3.6/multiprocessing/pool.py", line 108, in worker
    task = get()
  File "/usr/lib/python3.6/multiprocessing/queues.py", line 337, in get
    return _ForkingPickler.loads(res)
TypeError: __new__() missing 2 required positional arguments: 'tag' and 'attributes'
Process ForkPoolWorker-2:
Traceback (most recent call last):
  File "/usr/lib/python3.6/multiprocessing/process.py", line 258, in _bootstrap
    self.run()
  File "/usr/lib/python3.6/multiprocessing/process.py", line 93, in run
    self._target(*self._args, **self._kwargs)
  File "/usr/lib/python3.6/multiprocessing/pool.py", line 108, in worker
    task = get()
  File "/usr/lib/python3.6/multiprocessing/queues.py", line 337, in get
    return _ForkingPickler.loads(res)
TypeError: __new__() missing 2 required positional arguments: 'tag' and 'attributes'

---

То, что я делаю, просто. У меня есть список идентификаторов, которые я передавал функции, используя для l oop, и все работало нормально:

def do_the_job(self, id):
  pass
  # Process ID
  # Add ID to a list
  # Do a network activity related to id    

def main(self):
  for id in self.lst:
    self.do_the_job(id)

И поскольку он не смог использовать все мои сетевые возможности, я изменил часть l oop, как показано ниже, после импорта Pool из multiprocessing, чтобы запустить его параллельно. Но это терпит неудачу, поскольку это поражает p.map(...) с ошибками, сообщенными выше Есть ли способ диагностировать проблему?

def main(self):
  with Pool(2) as p:
    p.map(self.do_the_job, self.lst)

Answer 1

Посмотрев на Traceback, мы можем определить:

return _ForkingPickler.loads(res)
TypeError: __new__() missing 2 required positional arguments: 'tag' and 'attributes'

Вот что происходит, когда вы передаете тип данных, который не может быть обработан, что необходимо для работы Pool.

By проверяя элементы списка, убедитесь, что все его элементы состоят из стандартного типа данных Python.

Это была настоящая проблема здесь.

Answer 2

Решение простое - поместите функцию в глобальное пространство имен.

from multiprocessing import Pool

def func(x):
    return x

class A():
    def __init__(self):
        self.lst = [1,2,3,4,5]
        self.func = func

    def main(self):
        with Pool(2) as p:
            result = p.map(self.func, self.lst)
        return result

if __name__ == '__main__':
    print(A().main())