Многопоточный доступ к ресурсам - куда я могу поставить свои блокировки?

Question

У меня есть многопоточный код, где каждый поток должен записывать в один и тот же файл. Чтобы предотвратить проблемы с параллелизмом, я использую объект Lock.

Мой вопрос заключается в том, правильно ли я использую замок. Если я устанавливаю блокировку внутри каждого потока, является ли эта блокировка глобальной или специфичной только для этого конкретного потока?

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

import time
from threading import Thread, Lock

def main():
    for i in range(20):
        agent = Agent(i)
        agent.start()

class Agent(Thread):
    def __init__(self, thread_num):
        Thread.__init__(self)
        self.thread_num = thread_num

    def run(self):
        while True:
            print 'hello from thread %s' % self.thread_num
            self.write_result()   

    def write_result(self):
        lock = Lock()
        lock.acquire()
        try:
            f = open('foo.txt', 'a')
            f.write('hello from thread %s\n' % self.thread_num)
            f.flush()
            f.close()
        finally:
            lock.release()

if __name__ == '__main__':
    main()

Answer 1

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

class LockedWrite(file):
    """ Wrapper class to a file object that locks writes """
    def __init__(self, *args, **kwds):
        super(LockedWrite, self).__init__(*args, **kwds)
        self._lock = Lock()

    def write(self, *args, **kwds):
        self._lock.acquire()
        try:
            super(LockedWrite, self).write(*args, **kwds)
        finally:
            self._lock.release()

Для использования в вашем коде просто замените следующие функции:

def main():
    f = LockedWrite('foo.txt', 'a')

    for i in range(20):
        agent = Agent(i, f)
        agent.start()

class Agent(Thread):
    def __init__(self, thread_num, fileobj):
        Thread.__init__(self)
        self.thread_num = thread_num
        self._file = fileobj    

#   ...

    def write_result(self):
        self._file.write('hello from thread %s\n' % self.thread_num)

Этот подход помещает блокировку файла в сам файл, который кажется более чистым ИМХО

Answer 2

Создать блокировку вне метода.

class Agent(Thread):
    mylock = Lock()
    def write_result(self):
        self.mylock.acquire()
        try:
            ...
        finally:
            self.mylock.release()

или при использовании python> = 2.5:

class Agent(Thread):
    mylock = Lock()
    def write_result(self):
        with self.mylock:
            ...

Чтобы использовать это с Python 2.5, вы должны импортировать оператор из будущего:

from __future__ import with_statement

Answer 3

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

Очереди имеют все встроенные функции блокировки, поэтому любой поток может безопасно вызвать Queue.put() в любое время. Средство записи файлов будет единственным потоком, вызывающим Queue.get(), и, вероятно, может тратить большую часть своего времени на блокировку этого вызова (с разумным тайм-аутом, позволяющим потоку безошибочно отвечать на запрос на отключение). Все проблемы с синхронизацией будут решаться в очереди, и вам не придется беспокоиться о том, что вы забыли какую-то блокировку получения / снятия блокировки ...:)

Answer 4

Экземпляр блокировки должен быть связан с экземпляром файла.

Другими словами, вы должны одновременно создать и блокировку, и файл и передать оба потока.

Answer 5

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

Answer 6

Метод lock () возвращает объект блокировки для каждого вызова. Таким образом, каждый поток (фактически каждый вызов write_result) будет иметь свой объект блокировки. И не будет блокировки.

Answer 7

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

import time
from threading import Thread, Lock

def main():
    lock = Lock()
    for i in range(20):
        agent = Agent(i, lock)
        agent.start()

class Agent(Thread, Lock):
    def __init__(self, thread_num, lock):
        Thread.__init__(self)
        self.thread_num = thread_num
        self.lock = lock

    def run(self):
        while True:
            print 'hello from thread %s' % self.thread_num
            self.write_result()   

    def write_result(self):
        self.lock.acquire()
        try:
            f = open('foo.txt', 'a')
            f.write('hello from thread %s\n' % self.thread_num)
            f.flush()
            f.close()
        finally:
            lock.release()

if __name__ == '__main__':
    main()