Программа работает с map(), но вызывает TypeError с pool.map()

Question

У меня есть программа, которая долгое время не использовалась, но она уже используется и работает. Он использует многопроцессорность, поскольку одна и та же задача должна выполняться для разных данных много раз.

Теперь я коснулся программы, чтобы добавить новый параметр, протестировать его и заметить, что он вызывает ошибку. Также более ранняя версия с версией контролирует ту же ошибку. Полная ошибка выглядит так:

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) 
TypeError: 'NoneType' object is not callable 

Это все. Честно говоря, это не говорит мне многого. При попытке отладить это, я пришел к идее попробовать использовать обычный map() вместо объединенной версии pool.map. Скрипт работает нормально.

Я не могу придумать минимальный пример, который воспроизводит ошибку, но я могу придумать пример, где все прекрасно работает, так же, как и ожидалось:

import random 
import time 
from multiprocessing import Pool 


def do_work(x, y, z): 
    time.sleep(random.random() * 2) 
    print x + y + z 

def do_one(arguments): 
    print "doing one" 
    do_work(*arguments) 

def do_many(x, y, zs): 
    map(do_one, [(x, y, z) for z in zs]) 

def do_many_pooled(x, y, zs): 
    pool = Pool(2) 
    pool.map(do_one, [(x, y, z) for z in zs]) 
    pool.close() 
    pool.join() 

def main(): 
    x = 1 
    y = 2 
    zs = range(10) 
    print "doing many" 
    do_many(x, y, zs) 
    print "doing many pooled" 
    do_many_pooled(x, y, zs) 


if __name__ == '__main__': 
    main() 

Настоящая программа делает много запросов базы данных, вычислений с использованием numpy и сохранения результатов обратно в базу данных. В реальной программе программа выходит с ошибкой, прежде чем печатать «делать один», когда используется с объединенной версией, но работает отлично, используя не объединенную версию.

Кто-нибудь знает, как правильно читать и отслеживать трассировку и/или может мне сказать, что может вызвать это исключение?

Answer 1

Я бы сказал, что это выглядит как put настоящее время затерт в значение Нет из этой части отслеживающий:

File "/usr/lib64/python2.7/multiprocessing/pool.py", line 342, in _handle_tasks 
    put(task) 
TypeError: 'NoneType' object is not callable 

Глядя на источник Python, Pool.__init__() является создание _task_handler , который сделает звоните в _handle_tasks и предоставляют аргументы для сказал вызов:

self._task_handler = threading.Thread(
    target=Pool._handle_tasks, 
    args=(self._taskqueue, self._quick_put, self._outqueue, self._pool) 
    ) 

Если посмотреть на _handle_tasks, то вы видите, что self._quick_put является то, что заканчивается вверх будучи put переменным:

@staticmethod 
def _handle_tasks(taskqueue, put, outqueue, pool, cache): 
    thread = threading.current_thread() 

    for taskseq, set_length in iter(taskqueue.get, None): 
     i = -1 
     for i, task in enumerate(taskseq): 
      if thread._state: 
       debug('task handler found thread._state != RUN') 
       break 
      try: 
       put(task) 
      except Exception as e: 
       job, ind = task[:2] 
       try: 
        cache[job]._set(ind, (False, e)) 
       except KeyError: 
        pass 
     else: 
      if set_length: 
       debug('doing set_length()') 
       set_length(i+1) 
      continue 
     break 
    else: 
     debug('task handler got sentinel') 

Кроме того, вы можете увидеть, что все исключения ловятся здесь и спрятаны отчетности позже. Но, если вы голову обратно на Python 2.7.6, вы увидите это:

@staticmethod 
def _handle_tasks(taskqueue, put, outqueue, pool): 
    thread = threading.current_thread() 

    for taskseq, set_length in iter(taskqueue.get, None): 
     i = -1 
     for i, task in enumerate(taskseq): 
      if thread._state: 
       debug('task handler found thread._state != RUN') 
       break 
      try: 
       put(task) 
      except IOError: 
       debug('could not put task on queue') 
       break 
     else: 
      if set_length: 
       debug('doing set_length()') 
       set_length(i+1) 
      continue 
     break 
    else: 
     debug('task handler got sentinel') 

Обратите внимание, что здесь TypeError может уйти. Оказывается, это было исправлено в результате ошибки #19425. Как ни странно, это было , заявив, что это не проблема в Python 2.7, но набор изменений был все еще backported.

Во всяком случае, в любом случае, put() должно быть известное значение, и не , кажется, каким-либо образом установить put в этом коде. Итак, для меня это пахнет как ошибка в Python. Любой шанс вы можете запустить тот же код в более новой версии Python?

Некоторые другая полезная информация

Быстрый поиск Google возвращает некоторые интересные результаты тоже:

• Python ошибка #9755 - Похожие, но разные трассировки стека.

• Ошибка Python #15881 - Аналогично вышесказанному, , но немного отличается трассировкой стека.

Обе проблемы, где очистка влияла состояние модуля и вызывающие вещи неудачу с исключением «объект„NoneType“не вызываемая».

Другие, некоторые связанные с этим ошибки существуют и в Python. В какой-то момент вы могли бы столкнуться с аналогичным исключением с использованием потоков демона и выйти из главного потока вашего приложения. Я забыл, какая версия была исправлена. I просто хотел показать, что эта проблема не является неслыханной и является ошибкой в ​​ Python.