На этом шаге мы рассмотрим один из способов реализации такой блокировки.

Задача

    Ваша программа использует потоки, и вы хотите заблокировать критически важные участки кода, чтобы избежать состояния гонки (race condition).

Решение

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

import threading


class SharedCounter:
    '''
    Объект счетчика, который может быть общим для нескольких потоков.
    '''
    def __init__(self, initial_value=0):
        self._value = initial_value
        self._value_lock = threading.Lock()

    def incr(self, delta=1):
        '''
        Инкрементирует счетчик с блокировкой.
        '''
        with self._value_lock:
            self._value += delta

    def decr(self, delta=1):
        '''
        Декрементирует счетчик с блокировкой.
        '''
        with self._value_lock:
            self._value -= delta

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

Обсуждение

    Расписание потоков является врожденно недетерминистичным. По этой причине неудача с использованием блокировок в основанных на потоках программах может привести к порче данных и странному поведению, известному как "состояние гонки" (race condition). Чтобы избежать этого, блокировки должны быть применены во всех случаях, когда к изменяемому состоянию (mutable state) осуществляется доступ со стороны многих потоков.

    В старом коде Python часто можно увидеть явное получение и освобождение блокировок. Например, вот вариант последнего примера:

import threading


class SharedCounter:
    '''
    Объект счетчика, который может быть общим для нескольких потоков.
    '''
    def __init__(self, initial_value=0):
        self._value = initial_value
        self._value_lock = threading.Lock()

    def incr(self, delta=1):
        '''
        Инкрементирует счетчик с блокировкой.
        '''
        self._value_lock.acquire()
        self._value += delta
        self._value_lock.release()

    def decr(self, delta=1):
        '''
        Декрементирует счетчик с блокировкой.
        '''
        self._value_lock.acquire()
        self._value -= delta
        self._value_lock.release()

    Инструкция with более элегантна и менее подвержена ошибкам - особенно в ситуациях, когда программист может забыть вызвать метод release(), или если в программе возбуждается исключение во время удерживания блокировки (а инструкция with гарантирует, что блокировки будут освобождены в обоих этих случаях).

    Чтобы избежать потенциального дедлока, программы, использующие блокировки, должны быть написаны таким образом, чтобы каждому потоку разрешалось получать только одну блокировку за раз. Если это невозможно, вам может потребоваться добавить более продвинутую схему избежания дедлоков.

    В библиотеке threading вы найдете другие примитивы синхронизации, такие как RLock и объекты Semaphore. Опыт показывает, что они имеют более специальное назначение и не должны быть использованы для простых блокировок изменяемого состояния. RLock (объект многократной блокировки) - это блокировка, которая может приобретаться несколько раз одним и тем же потоком. В основном она используется для реализации основанной на коде блокировки (code based locking) или синхронизации, основанной на конструкции, известной как "монитор". С блокировкой такого типа только одному потоку разрешается применять функцию или методы класса, пока блокировка удерживается. Например, вы можете реализовать класс SharedCounter таким образом:

import threading


class SharedCounter:
    '''
    Объект счетчика, который может быть общим для нескольких потоков.
    '''
    _lock = threading.RLock()

    def __init__(self, initial_value=0):
        self._value = initial_value

    def incr(self, delta=1):
        '''
        Инкрементирует счетчик с блокировкой.
        '''
        with SharedCounter._lock:
            self._value += delta

    def decr(self, delta=1):
        '''
        Декрементирует счетчик с блокировкой.
        '''
        with SharedCounter._lock:
            self.incr(-delta)

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

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

    Объект Semaphore - это примитив синхронизации, основанный на общем счетчике. Если счетчик ненулевой, инструкция with уменьшает его значение, и потоку разрешается работать. Счетчик инкрементируется после завершения выполнения блока with. Если счетчик равен нулю, работа блокируется, пока счетчик не увеличится другим потоком. Хотя семафор может быть использован так же, как и стандартный Lock, добавленная сложность в реализации отрицательно влияет на производительность. Объекты Semaphore более полезны в качестве замены простой блокировки в приложениях, использующих обмен сигналами между потоками или троттлинг (throttling). Например, если вы хотите ограничить конкурентность в каком-то конкретном участке кода, то вы можете использовать семафор:

from threading import Semaphore
import urllib.request


# Самое большее 5 потоков могут выполняться одновременно
_fetch_url_sema = Semaphore(5)
def fetch_url(url):
    with _fetch_url_sema:
        return urllib.request.urlopen(url)

    Если вы заинтересованы в теории и реализации потоковых примитивов синхронизации, обратитесь к любому учебнику по операционным системам.

    На следующем шаге мы рассмотрим блокировку с избежанием дедлока.

Предыдущий шаг Содержание Следующий шаг