Блог

 >>> import concurrent.futures
>>> from collections import namedtuple
>>> #1. Initialise namedtuple here
>>> # tm = namedtuple("tm", ["pk"])  
>>> class T:  
...     #2. Initialise named tuple here
...     #tm = namedtuple("tm", ["pk"]) 
...     def __init__(self): 
...         #3: Initialise named tuple here
...         tm = namedtuple("tm", ["pk"])                       
...         self.x = {'key': [tm('value')]}  
...     def test1(self):  
...         with concurrent.futures.ProcessPoolExecutor(max_workers=1) as executor:  
...             results = executor.map(self.test, ["key"])  
...         return results  
...     def test(self, s): 
...         print(self.x[s])   
... 
>>> t = T().test1()
  

Это застревает здесь.

 ^CTraceback (most recent call last):
  File "<stdin>", line 1, in <module>
Process ForkProcess-1:
  File "<stdin>", line 10, in test1
  File "/Library/Frameworks/Python.framework/Versions/3.7/lib/python3.7/concurrent/futures/_base.py", line 623, in __exit__
    self.shutdown(wait=True)
  File "/Library/Frameworks/Python.framework/Versions/3.7/lib/python3.7/concurrent/futures/process.py", line 681, in shutdown
    self._queue_management_thread.join()
  File "/Library/Frameworks/Python.framework/Versions/3.7/lib/python3.7/threading.py", line 1044, in join
Traceback (most recent call last):
  File "/Library/Frameworks/Python.framework/Versions/3.7/lib/python3.7/multiprocessing/process.py", line 297, in _bootstrap
    self.run()
  File "/Library/Frameworks/Python.framework/Versions/3.7/lib/python3.7/multiprocessing/process.py", line 99, in run
    self._target(*self._args, **self._kwargs)
  File "/Library/Frameworks/Python.framework/Versions/3.7/lib/python3.7/concurrent/futures/process.py", line 233, in _process_worker
    call_item = call_queue.get(block=True)
  File "/Library/Frameworks/Python.framework/Versions/3.7/lib/python3.7/multiprocessing/queues.py", line 94, in get
    res = self._recv_bytes()
  File "/Library/Frameworks/Python.framework/Versions/3.7/lib/python3.7/multiprocessing/connection.py", line 216, in recv_bytes
    buf = self._recv_bytes(maxlength)
  File "/Library/Frameworks/Python.framework/Versions/3.7/lib/python3.7/multiprocessing/connection.py", line 407, in _recv_bytes
    buf = self._recv(4)
  File "/Library/Frameworks/Python.framework/Versions/3.7/lib/python3.7/multiprocessing/connection.py", line 379, in _recv
    chunk = read(handle, remaining)
KeyboardInterrupt
    self._wait_for_tstate_lock()
  File "/Library/Frameworks/Python.framework/Versions/3.7/lib/python3.7/threading.py", line 1060, in _wait_for_tstate_lock
    elif lock.acquire(block, timeout):
KeyboardInterrupt
  

Если я инициализирую именованный кортеж вне класса (в # 1), в этом случае это работает нормально. Может кто-нибудь, пожалуйста, сообщить мне, в чем проблема, если я инициализирую в соответствии с # 2 или # 3?

Вы не меняете, где вы инициализируете namedtuple. Вы меняете место создания класса namedtuple.

Когда вы создаете класс namedtuple с именем «x» в модуле «y» с помощью collections.namedtuple , для него __module__ устанавливается значение 'y' , а для его __qualname__ — значение 'x' . Травление и удаление зависит от того, что этот класс фактически доступен в y.x местоположении, указанном этими атрибутами, но в случаях 2 и 3 вашего примера это не так.

Python не может выбрать namedtuple, что нарушает межпроцессную связь с рабочими. Выполнение self.test в рабочем процессе зависит от травления self.test и удаления его копии в рабочем процессе, а этого не может произойти, если self.x является экземпляром класса, который нельзя травить.