Проблемы с многопоточностью в Python 2.5, KeyError: 51, Помощь в отладке?

Question

У меня есть сценарий Python, который запускает определенный сценарий большое количество раз (для целей Монте-Карло) и способ, которым я написал сценарий, заключается в том, что я ставлю сценарий в очередь желаемое количество раз, которое он должен быть запущен, затем я порождаю потоки и каждый поток запускает сценарий один раз и снова, когда все готово.

Как только скрипт в определенном потоке завершен, вывод записывается в файл путем доступа к блокировке (поэтому я предположил, что только один поток получает доступ к блокировке в данный момент времени). Как только блокировка снята одним потоком, следующий поток получает к ней доступ, добавляет свой вывод в ранее записанный файл и перезаписывает его.

Я не сталкиваюсь с проблемой, когда число итераций мало, например 10 или 20, но когда оно большое, например 50 или 150, python возвращает KeyError: 51 говорит, что элемент не существует, и ошибка, на которую он указывает, в замке, который меня озадачивает, так как только один поток должен получить доступ к замку одновременно, и я не ожидаю ошибки.

Это класс, который я использую:

class errorclass(threading.Thread):

    def __init__(self, queue):
        self.__queue=queue
        threading.Thread.__init__(self)

    def run(self):
        while 1:
              item = self.__queue.get()
              if item is None: break
              result = myfunction()
              lock = threading.RLock()
              lock.acquire()
              ADD entries from current thread to entries in file and
              REWRITE FILE
              lock.release()

queue = Queue.Queue()

for i in range(threads):
    errorclass(queue).start()

for i in range(desired iterations):
    queue.put(i)
for i in range(threads):
    queue.put(None)

Python возвращает с KeyError: 51 для большого числа желаемых итераций во время операции добавления / записи файла после доступа к блокировке, мне интересно, является ли это правильным способом использования блокировки, поскольку каждый поток имеет операцию блокировки, а не каждый поток доступ к общей блокировке? Как бы это исправить?

Answer 1

Сейчас у вас есть новая блокировка для каждой итерации в методе run каждого потока. По сути, блокировки вообще не происходит. Если вы хотите защитить записи в файл, вам нужно убедиться, что все потоки, которые обращаются к одному и тому же файлу, используют один и тот же объект блокировки. Самый простой способ сделать это - создать его на глобальном уровне:

lock = threading.RLock()
class errorclass(...):
    ...
    def run(self)
        ... # get items and process
        with lock:
            ADD entries from current thread to entries in file and REWRITE FILE

Вам нужно будет использовать from __future__ import with_statement, чтобы использовать оператор with в Python 2.5, но он гарантирует, что вы никогда не забудете разблокировать блокировку, даже когда происходит ошибка. (Если вам нужна совместимость с Python 2.4 и более ранними версиями, вам придется использовать вместо этого try / finally.)

Answer 2

Создайте блокировку и передайте ее в errorclass.__init__, чтобы они делили экземпляр. В противном случае каждый поток блокируется от повторного входа в свою собственную критическую секцию, которая является просто запретом.