Предотвращение состояния гонки (race condition) в Python

В этом руководстве вы узнаете о состояниях гонки (race condition) и о том, как использовать объект Lock модуля  threading в Python для их предотвращения.

• Что такое состояние гонки?
• Пример race condition
• Предотвращение состояния гонки
• Использование блокировки потоков с оператором with
• Определение thread-safe класса Counter, который использует объект Lock

Что такое состояние гонки?

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

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

Окончательное значение общей переменной зависит от того, какой поток завершает обновление последним. Какой бы поток ни изменил значение последним, он выиграет гонку.

Пример race condition

Следующий пример иллюстрирует состояние гонки:

from threading import Thread
from time import sleep


counter = 0

def increase(by):
    global counter

    local_counter = counter
    local_counter += by

    sleep(0.1)

    counter = local_counter
    print(f'counter={counter}')


# create threads
t1 = Thread(target=increase, args=(10,))
t2 = Thread(target=increase, args=(20,))

# start the threads
t1.start()
t2.start()


# wait for the threads to complete
t1.join()
t2.join()


print(f'The final counter is {counter}')

В этой программе оба потока пытаются одновременно изменить значение переменной-счетчика. Значение переменной зависит от того, какой поток завершается последним.

Если поток t1 завершится раньше потока t2, вы увидите следующий результат:

counter=10
counter=20
The counter is 20

В противном случае вы увидите следующий вывод:

counter=20
counter=10
The final counter is 10

Как это работает.

• Сначала импортируйте класс Thread из модуля threading и функцию Sleep() из модуля time:

from threading import Thread
from time import sleep

• Во-вторых, определите глобальную переменную с именем counter, значение которой равно нулю:

counter = 0

• В-третьих, определите функцию, которая увеличивает значение переменной счетчика на число:

def increase(by):
    global counter

    local_counter = counter
    local_counter += by

    sleep(0.1)

    counter = local_counter
    print(f'counter={counter}')

• В-четвертых, создайте два потока. Первый поток увеличивает счетчик на 10, а второй поток увеличивает счетчик на 20:

t1 = Thread(target=increase, args=(10,))
t2 = Thread(target=increase, args=(20,))

• В-пятых, запустите темы:

t1.start()
t2.start()

• В-шестых, в основном потоке дождитесь завершения потоков t1 и t2:

t1.join()
t2.join()

• Наконец, покажите окончательное значение переменной counter:

print(f'The final counter is {counter}')

Предотвращение состояния гонки

Чтобы предотвратить условия гонки, вы можете использовать блокировку потоков (threading lock).

Объект threading lock — это примитив синхронизации, который обеспечивает монопольный доступ к общему ресурсу в многопоточном приложении. Блокировка потока также известна как mutex, что означает взаимное исключение.

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

В Python вы можете использовать класс Lock из модуля threading для создания объекта блокировки:

Сначала создайте экземпляр класса Lock:

lock = Lock()

По умолчанию Lock разблокирована до тех пор, пока поток не завладеет ею.

• Во-вторых, получите блокировку, вызвав методacquire():

lock.acquire()

• В-третьих, отпустите блокировку, как только поток завершит изменение общей переменной:

lock.release()

В следующем примере показано, как использовать объект Lock для предотвращения состояния гонки в предыдущей программе:

from threading import Thread, Lock
from time import sleep


counter = 0


def increase(by, lock):
    global counter

    lock.acquire()

    local_counter = counter
    local_counter += by

    sleep(0.1)

    counter = local_counter
    print(f'counter={counter}')

    lock.release()


lock = Lock()

# create threads
t1 = Thread(target=increase, args=(10, lock))
t2 = Thread(target=increase, args=(20, lock))

# start the threads
t1.start()
t2.start()


# wait for the threads to complete
t1.join()
t2.join()


print(f'The final counter is {counter}')

Выход:

counter=10
counter=30
The final counter is 30

Как это работает.

• Сначала добавьте второй параметр в функцию increase().
• Во-вторых, создайте экземпляр класса Lock.
• В-третьих, установите блокировку перед доступом к переменной-счетчику и отпустите ее после обновления нового значения.

Использование блокировки потоков с оператором with

Объект lock проще использовать с оператором with для получения и снятия блокировки внутри блока кода:

import threading

# Create a lock object
lock = threading.Lock()

# Perform some operations within a critical section
with lock:
    # Lock was acquired within the with block
    # Perform operations on the shared resource
    # ...

# the lock is released outside the with block

Например, вы можете использовать оператор with без необходимости вызова методов acquire() и Release() в приведенном выше примере следующим образом:

from threading import Thread, Lock
from time import sleep


counter = 0

def increase(by, lock):
    global counter

    with lock:
        local_counter = counter
        local_counter += by

        sleep(0.1)

        counter = local_counter
        print(f'counter={counter}')


lock = Lock()

# create threads
t1 = Thread(target=increase, args=(10, lock))
t2 = Thread(target=increase, args=(20, lock))

# start the threads
t1.start()
t2.start()


# wait for the threads to complete
t1.join()
t2.join()


print(f'The final counter is {counter}')

Определение thread-safe класса Counter, который использует объект Lock

Ниже показано, как определить потокобезопасный класс Counter с помощью объекта Lock:

from threading import Thread, Lock
from time import sleep


class Counter:
    def __init__(self):
        self.value = 0
        self.lock = Lock()

    def increase(self, by):
        with self.lock:
            current_value = self.value
            current_value += by

            sleep(0.1)

            self.value = current_value
            print(f'counter={self.value}')

def main():
    counter = Counter()
    # create threads
    t1 = Thread(target=counter.increase, args=(10, ))
    t2 = Thread(target=counter.increase, args=(20, ))

    # start the threads
    t1.start()
    t2.start()


    # wait for the threads to complete
    t1.join()
    t2.join()


    print(f'The final counter is {counter.value}')

if __name__ == '__main__':
    main()