Многопоточные вычисления

Класс Thread

Экземпляры класса threading.Thread представляют потоки Python-программы. Задать действия, которые будут выполняться в потоке, можно двумя способами: передать конструктору класса исполняемый объект и аргументы к нему или путем наследования получить новый класс с переопределенным методом run(). Первый способ был рассмотрен в примере выше. Конструктор класса threading.Thread имеет следующие аргументы:

Thread(group, target, name, args, kwargs)

Здесь group - группа потоков (пока что не используется, должен быть равен None ), target - объект, который будет вызван в методе run(), name - имя потока, args и kwargs - последовательность и словарь позиционных и именованных параметров (соответственно) для вызова заданного в параметре target объекта. В примере выше были использованы только позиционные параметры, но то же самое можно было выполнить и с применением именованных параметров:

import threading

            def proc(n):
              print "Процесс", n

            p1 = threading.Thread(target=proc, name="t1", kwargs={"n": "1"})
            p2 = threading.Thread(target=proc, name="t2", kwargs={"n": "2"})
            p1.start()
            p2.start()

То же самое можно проделать через наследование от класса threading.Thread с определением собственного конструктора и метода run():

import threading

            class T(threading.Thread):
              def __init__(self, n):
                threading.Thread.__init__(self, name="t" + n)
                self.n = n
              def run(self):
                print "Процесс", self.n

            p1 = T("1")
            p2 = T("2")
            p1.start()
            p2.start()

Самое первое, что необходимо сделать в конструкторе - вызвать конструктор базового класса. Как и раньше, для запуска потока нужно выполнить метод start() объекта-потока, что приведет к выполнению действий в методе run().

Жизнью потоков можно управлять вызовом методов:

В модуле Thread пока что не реализованы возможности, присущие потокам в Java (определение групп потоков, приостановка и прерывание потоков извне, приоритеты и некоторые другие вещи), однако они, скорее всего, будут созданы в недалеком будущем.

Таймер

Класс threading.Timer представляет действие, которое должно быть выполнено через заданное время. Этот класс является подклассом класса threading.Thread, поэтому запускается также методом start(). Следующий простой пример, печатающий на стандартном выводе Hello, world! поясняет сказанное:

def hello():
              print "Hello, world!"

            t = Timer(30.0, hello)
            t.start()

Замки

Простейший замок может быть реализован на основе класса Lock модуля threading. Замок имеет два состояния: он может быть или открыт, или заперт. В последнем случае им владеет некоторый поток. Объект класса Lock имеет следующие методы:

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

Когда нужны замки?

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

Тупиковая ситуация (deadlock)

Замки применяются для управления доступом к ресурсу, который нельзя использовать совместно. В программе таких ресурсов может быть несколько. При работе с замками важно хорошо продумать, не зайдет ли выполнение программы в тупик (deadlock) из-за того, что двум потокам потребуются одни и те же ресурсы, но ни тот, ни другой не смогут их получить, так как они уже получили замки. Такая ситуация проиллюстрирована в следующем примере:

import threading, time

                resource = {'A': threading.Lock(), 'B': threading.Lock()}

                def proc(n, rs):
                 for r in rs:
                   print "Процесс %s запрашивает ресурс %s" % (n, r)
                   resource[r].acquire()
                   print "Процесс %s получил ресурс %s" % (n, r)
                   time.sleep(1)
                 print "Процесс %s выполняется" % n
                 for r in rs:
                   resource[r].release()
                 print "Процесс %s закончил выполнение" % n

                p1 = threading.Thread(target=proc, name="t1", args=["1", "AB"])
                p2 = threading.Thread(target=proc, name="t2", args=["2", "BA"])
                p1.start()
                p2.start()
                p1.join()
                p2.join()

В этом примере два потока (t1 и t2) запрашивают замки к одним и тем же ресурсам (A и B), но в разном порядке, отчего получается, что ни у того, ни у другого не хватает ресурсов для дальнейшей работы, и они оба безнадежно повисают, ожидая освобождения нужного ресурса. Благодаря операторам print можно увидеть последовательность событий:

Процесс 1 запрашивает ресурс A
                Процесс 1 получил ресурс A
                Процесс 2 запрашивает ресурс B
                Процесс 2 получил ресурс B
                Процесс 1 запрашивает ресурс B
                Процесс 2 запрашивает ресурс A

Существуют методики, позволяющие избежать подобных тупиков, однако их рассмотрение не входит в рамки данной лекции. Можно посоветовать следующие приемы:

Семафоры

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

Конструктор класса threading.Semaphore принимает в качестве (необязательного) аргумента начальное состояние счетчика (по умолчанию оно равно 1, что соответствует замку класса Lock ). Методы acquire() и release() действуют аналогично описанным выше одноименным методам у замков.

Семафор может применяться для охраны ограниченного ресурса. Например, с его помощью можно вести пул соединений с базой данных. Пример такого использования семафора (заимствован из документации к Python) дан ниже:

from threading import BoundedSemaphore
            maxconnections = 5
            # Подготовка семафора
            pool_sema = BoundedSemaphore(value=maxconnections)

            # Внутри потока:

            pool_sema.acquire()
            conn = connectdb()
            # ... использование соединения ...
            conn.close()
            pool_sema.release()

Таким образом, применяется не более пяти соединений с базой данных. В примере использован класс threading.BoundedSemaphore. Экземпляры этого класса отличаются от экземпляров класса threading.Semaphore тем, что не дают сделать release() больше, чем сделан acquire().

События

Еще одним способом коммуникации между объектами являются события. Экземпляры класса threading.Event могут быть использованы для передачи информации о наступлении некоторого события от одного потока одному или нескольким другим потокам. Объекты-события имеют внутренний флаг, который может находиться в установленном или сброшенном состоянии. При своем создании флаг события находится в сброшенном состоянии. Если флаг в установленном состоянии, ожидания не происходит: поток, вызвавший метод wait() для ожидания события, просто продолжает свою работу. Ниже приведены методы экземпляров класса threading.Event:

Составить пример работы с событиями предлагается в качестве упражнения.

Условия

Более сложным механизмом коммуникации между потоками является механизм условий. Условия представляются в виде экземпляров класса threading.Condition и, подобно только что рассмотренным событиям, оповещают потоки об изменении некоторого состояния. Конструктор класса threading.Condition принимает необязательный параметр, задающий замок класса threading.Lock или threading.RLock. По умолчанию создается новый экземпляр замка класса threading.RLock. Методы объекта-условия описаны ниже:

В следующем примере условия используются для оповещения потоков о прибытии новой порции данных (организуется связь производитель - потребитель, producer - consumer):

import threading

            cv = threading.Condition()

            class Item:
              """Класс-контейнер для элементов, которые будут потребляться
              в потоках"""
              def __init__(self):
                self._items = []
              def is_available(self):
                return len(self._items) > 0
              def get(self):
                return self._items.pop()
              def make(self, i):
                self._items.append(i)

            item = Item()

            def consume():
              """Потребление очередного элемента (с ожиданием его появления)"""
              cv.acquire()
              while not item.is_available():
                cv.wait()
              it = item.get()
              cv.release()
              return it

            def consumer():
              while True:
                print consume()

            def produce(i):
              """Занесение нового элемента в контейнер и оповещение потоков"""
              cv.acquire()
              item.make(i)
              cv.notify()
              cv.release()

            p1 = threading.Thread(target=consumer, name="t1")
            p1.setDaemon(True)
            p2 = threading.Thread(target=consumer, name="t2")
            p2.setDaemon(True)
            p1.start()
            p2.start()
            produce("ITEM1")
            produce("ITEM2")
            produce("ITEM3")
            produce("ITEM4")
            p1.join()
            p2.join()

В этом примере условие cv отражает наличие необработанных элементов в контейнере item. Функция produce() "производит" элементы, а consume(), работающая внутри потоков, "потребляет". Стоит отметить, что в приведенном виде программа никогда не закончится, так как имеет бесконечный цикл в потоках, а в главном потоке - ожидание завершения этих потоков. Еще одна особенность - признак демона, установленный с помощью метода setDaemon() объекта-потока до его старта.

Очередь

Процесс, показанный в предыдущем примере, имеет значение, достойное отдельного модуля. Такой модуль в стандартной библиотеке языка Python есть, и он называется Queue.

Помимо исключений - Queue.Full (очередь переполнена) и Queue.Empty (очередь пуста) - модуль определяет класс Queue, заведующий собственно очередью.

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

import threading, Queue

            item = Queue.Queue()

            def consume():
              """Потребление очередного элемента (с ожиданием его появления)"""
              return item.get()

            def consumer():
              while True:
                print consume()

            def produce(i):
              """Занесение нового элемента в контейнер и оповещение потоков"""
              item.put(i)

            p1 = threading.Thread(target=consumer, name="t1")
            p1.setDaemon(True)
            p2 = threading.Thread(target=consumer, name="t2")
            p2.setDaemon(True)
            p1.start()
            p2.start()
            produce("ITEM1")
            produce("ITEM2")
            produce("ITEM3")
            produce("ITEM4")
            p1.join()
            p2.join()

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

Модуль thread

По сравнению с модулем threading, модуль thread предоставляет низкоуровневый доступ к потокам. Многие функции модуля threading, который рассматривался до этого, реализованы на базе модуля thread. Здесь стоит сделать некоторые замечания по применению потоков вообще. Документация по Python предупреждает, что использование потоков имеет особенности:

Визуализация работы потоков

Следующий пример иллюстрирует параллельность выполнения потоков, используя возможности библиотеки графических примитивов Tkinter (она входит в стандартную поставку Python). Несколько потоков наперегонки увеличивают размеры прямоугольника некоторого цвета. Цветом победившего потока окрашивается кнопка Go:

import threading, time, sys
            from Tkinter import Tk, Canvas, Button, LEFT, RIGHT, NORMAL, DISABLED

            global champion

            # Задается дистанция, цвет полосок и другие параметры
            distance = 300
            colors = ["Red","Orange","Yellow","Green","Blue","DarkBlue","Violet"]
            nrunners = len(colors)      # количество дополнительных потоков
            positions = [0] * nrunners  # список текущих позиций
            h, h2 = 20, 10              # параметры высоты полосок

            def run(n):
              """Программа бега n-го участника (потока)"""
              global champion
              while 1:
                for i in range(10000):           # интенсивные вычисления
                  pass
                graph_lock.acquire()
                positions[n] += 1                # передвижение на шаг
                if positions[n] == distance:     # если уже финиш
                  if champion is None:           # и чемпион еще не определен,
                    champion = colors[n]         # назначается чемпион
                  graph_lock.release()
                  break
                graph_lock.release()

            def ready_steady_go():
              """Инициализация начальных позиций и запуск потоков"""
              graph_lock.acquire()
              for i in range(nrunners):
                positions[i] = 0
                threading.Thread(target=run, args=[i,]).start()
              graph_lock.release()

            def update_positions():
              """Обновление позиций"""
              graph_lock.acquire()
              for n in range(nrunners):
                c.coords(rects[n], 0, n*h, positions[n], n*h+h2)
              tk.update_idletasks()  # прорисовка изменений
              graph_lock.release()

            def quit():
              """Выход из программы"""
              tk.quit()
              sys.exit(0)

            # Прорисовка окна, основы для прямоугольников и самих прямоугольников,
            # кнопок для пуска и выхода
            tk = Tk()
            tk.title("Соревнование потоков")
            c = Canvas(tk, width=distance, height=nrunners*h, bg="White")
            c.pack()
            rects = [c.create_rectangle(0, i*h, 0, i*h+h2, fill=colors[i])
                     for i in range(nrunners)]
            go_b = Button(text="Go", command=tk.quit)
            go_b.pack(side=LEFT)
            quit_b = Button(text="Quit", command=quit)
            quit_b.pack(side=RIGHT)

            # Замок, регулирующий доступ к функции пакета Tk
            graph_lock = threading.Lock()

            # Цикл проведения соревнований
            while 1:
              go_b.config(state=NORMAL), quit_b.config(state=NORMAL)
              tk.mainloop()             # Ожидание нажатия клавиш
              champion = None
              ready_steady_go()
              go_b.config(state=DISABLED), quit_b.config(state=DISABLED)
              # Главный поток ждет финиша всех участников
              while sum(positions) < distance*nrunners:
                update_positions()
              update_positions()
              go_b.config(bg=champion)     # Кнопка окрашивается в цвет победителя
              tk.update_idletasks()

Примечание:

Эта программа использует некоторые возможности языка Python 2.3 (встроенную функцию sum() и списковые включения), поэтому для ее выполнения нужен Python версии не меньше 2.3.

Заключение

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

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

Стандартная библиотека Python предоставляет довольно неплохой набор возможностей для многопоточного программирования в модулях threading и thread, а также некоторые полезные вспомогательные модули (например, Queue ).