Многопоточное программированиеВ этой главе
 Скачать 0.74 Mb.
|
|
Пример 4.10. Сценарий, в котором предусмотрены блокировки и введены дополнительные средства случайного выбора ( mtsleepF.py) В этом примере демонстрируется использование блокировок и других инструмен- тов обеспечения многопоточного функционирования. 1 #!/usr/bin/env python 2 3 from atexit import register 4 from random import randrange 5 from threading import Thread, Lock, currentThread 6 from time import sleep, ctime 7 8 class CleanOutputSet(set): 9 def __str__(self): 10 return ', '.join(x for x in self) 11 12 lock = Lock() 13 loops = (randrange(2,5) for x in xrange(randrange(3,7))) 14 remaining = CleanOutputSet() 15 16 def loop(nsec): 17 myname = currentThread().name 18 lock.acquire() 19 remaining.add(myname) 20 print '[%s] Started %s' % (ctime(), myname) 21 lock.release() 22 sleep(nsec) 23 lock.acquire() 24 remaining.remove(myname) 25 print '[%s] Completed %s (%d secs)' % ( 26 ctime(), myname, nsec) 27 print ' (remaining: %s)' % (remaining or 'NONE') 28 lock.release() 29 30 def _main(): 31 for pause in loops: 32 Thread(target=loop, args=(pause,)).start() 33 34 @register 35 def _atexit(): 36 print 'all DONE at:', ctime() 37 38 if __name__ == '__main__': 39 main() 06_ch04.indd 217 22.01.2015 22:00:47 Глава 4 Многопоточное программирование 218 Построчное объяснение Строки 1–6 Это обычные строки запуска и импорта. Следует учитывать, что начиная с версии 2.6 метод threading.currentThread() переименован в threading. current_thread(), но первый компонент имени метода остался неизмен- ным в целях обеспечения обратной совместимости. Строки 8–10 Это подкласс множества, о котором речь шла выше. Он содержит реализацию ме- тода __str__(), который позволяет вместо формата вывода, применяемого по умол- чанию, перейти к формату, представляющему собой строку элементов, разделенную запятыми. Строки 12–14 В состав применяемых глобальных переменных входят блокировка, экземпляр по- заимствованного из предыдущего кода откорректированного множества и случайно выбранного числа потоков (от трех до шести), каждый из которых останавливается (вернее, приостанавливается) на время от двух до четырех секунд. Строки 16–28 Функция loop() сохраняет имя текущего потока, в котором она выполняется, за- тем захватывает блокировку, что позволяет сделать неразрывным (атомарным) сле- дующий ряд операций: добавление имени к множеству remaining и формирование вывода с указанием на начало потока (с этого момента больше никакой другой поток не сможет войти в критический участок кода). После освобождения блокировки этот поток приостанавливается на время в секундах, выбранное случайным образом, затем повторно захватывает блокировку, чтобы сформировать свой окончательный вывод перед ее освобождением. Строки 30–39 Функция _main() выполняется, только если этот сценарий не был импортирован для использования в другой программе. Назначение этой функции состоит в том, что она порождает и запускает каждый из потоков. Как было указано выше, применяется метод atexit.register() для регистрации функции _atexit(), которую интерпре- татор может выполнить перед выходом из программы. В качестве альтернативного по отношению к варианту, в котором поддерживается собственное множество выполняющихся в настоящее время потоков, можно приме- нить вариант с использованием метода threading.enumerate(), который возвращает список всех потоков, работающих в настоящее время (этот список включает потоки, действующие в качестве демона, но, безусловно, в него не входят потоки, которые еще не запущены). В рассматриваемом примере этот вариант не используется, поскольку его применение приводит к созданию двух дополнительных потоков, которые прихо- дится удалять для сокращения объема вывода, в том числе текущего потока (посколь- ку он еще не завершен) и основного потока (который так или иначе не должен быть показан). 06_ch04.indd 218 22.01.2015 22:00:47 219 4.7. Практическое применение многопоточной обработки Не следует также забывать, что вместо оператора форматирования строки можно использовать метод str.format(), при условии, что для работы применяется вер- сия Python 2.6 или более новая (включая версии 3.x). Иными словами, следующая инструкция print: print '[%s] Started %s' % (ctime(), myname) может быть заменена в версии 2.6 и последующей таким вызовом: print '[{0}] Started {1}'.format(ctime(), myname) или (в версии 3.x) таким вызовом print(): print('[{0}] Started {1}'.format(ctime(), myname)) Если задача состоит лишь в том, чтобы подсчитать число потоков, выполняю- щихся в настоящее время, то вместо этого можно использовать метод threading. activeCount() (переименованный в active_count(), начиная с версии 2.6). Применение управления контекстом Программисты, работающие в версии Python 2.5 и более новой, могут вос- пользоваться еще одним вариантом, который вообще не требует вызова ме- тодов acquire() и release() применительно к блокировке, что способству- ет еще большему упрощению кода. С помощью инструкции with можно вводить в действие для любого объекта диспетчер контекста, который обе- спечит вызов метода acquire() перед входом в критический участок и мето- да release(), когда этот блок завершит выполнение. Диспетчеры контекста предусмотрены для всех объектов модуля threading, та- ких как Lock, RLock, Condition, Semaphore и BoundedSemaphore, а это означает, что для работы с этими объектами может применяться инструкция with. С помощью инструкции with можно еще больше упростить код функции loop() следующим образом: from __future__ import with_statement # только в версии 2.5 def loop(nsec): myname = currentThread().name with lock: remaining.add(myname) print '[%s] Started %s' % (ctime(), myname) sleep(nsec) with lock: remaining.remove(myname) print '[%s] Completed %s (%d secs)' % ( ctime(), myname, nsec) print ' (remaining: %s)' % ( remaining or 'NONE',) 06_ch04.indd 219 22.01.2015 22:00:47 Глава 4 Многопоточное программирование 220 Перенос приложения в версию Python 3 Теперь можно сравнительно легко перенести приведенный выше сценарий в версию Python 3.x, применив к нему инструмент 2to3 (следующий вывод приведен в сокращенном виде, поскольку полные результаты применения diff уже были показаны ранее): $ 2to3 -w mtsleepF.py RefactoringTool: Skipping implicit fixer: buffer RefactoringTool: Skipping implicit fixer: idioms RefactoringTool: Skipping implicit fixer: set_literal RefactoringTool: Skipping implicit fixer: ws_comma : RefactoringTool: Files that were modified: RefactoringTool: mtsleepF.py После переименования mtsleepF.py в mtsleepF3.py и mtsleep.py.bak в mtsleepF.py обнаруживается, что этот сценарий относится к той замечательной ка- тегории программ, перенос которых в новую версию интерпретатора происходит идеально, без малейших проблем: $ python3 mtsleepF3.py [Sun Apr 3 23:29:39 2011] Started Thread-1 [Sun Apr 3 23:29:39 2011] Started Thread-2 [Sun Apr 3 23:29:39 2011] Started Thread-3 [Sun Apr 3 23:29:41 2011] Completed Thread-3 (2 secs) (remaining: Thread-2, Thread-1) [Sun Apr 3 23:29:42 2011] Completed Thread-2 (3 secs) (remaining: Thread-1) [Sun Apr 3 23:29:43 2011] Completed Thread-1 (4 secs) (remaining: NONE) all DONE at: Sun Apr 3 23:29:43 2011 Теперь, после ознакомления с блокировками, перейдем к изучению семафоров и рассмотрим пример, в котором используется и то и другое. 4.7.4. Пример семафора Как уже было сказано, блокировки являются довольно простыми для понимания и могут быть легко реализованы. Кроме того, можно довольно легко определить, в каком случае они действительно необходимы. Однако ситуация может оказаться сложнее, и тогда вместо блокировки потребуется более мощный примитив синхро- низации. Если в приложении приходится иметь дело с ограниченными ресурсами, то семафоры могут оказаться более приемлемыми. Семафоры относятся к числу примитивов синхронизации, которые были введе- ны в действие раньше других. Семафор, по существу, представляет собой счетчик, значение которого уменьшается после захвата ресурса (и снова увеличивается после освобождения ресурса). Семафоры, представляющие закрепленные за ними ресурсы, можно рассматривать как доступные или недоступные. Действие по захвату ресурса и уменьшению значения счетчика принято обозначать как P() (от голландского слова probeer/proberen), но для обозначения этого действия применяются также термины “переход в состояние ожидания”, “осуществление попытки”, “захват”, “приоста- новка” или “получение”. И наоборот, после завершения работы потока с ресурсом 06_ch04.indd 220 22.01.2015 22:00:47 221 4.7. Практическое применение многопоточной обработки должен быть произведен возврат ресурса в пул ресурсов. Для этого применяется действие, по традиции обозначаемое V() (от голландского слова verhogen/verhoog). Это действие обозначается также как “сигнализация”, “наращивание”, “отпускание”, “отправка”, “освобождение”. В языке Python вместо всех этих вариантов именования применяются упрощенные обозначения, согласно которым функции и (или) методы обозначаются как те, что служат для работы с блокировками: acquire и release. Се- мафоры являются более гибкими, чем блокировки, поскольку обеспечивают работу с несколькими потоками, в каждом из которых используется один из экземпляров конечного ресурса. В качестве следующего примера рассмотрим предельно упрощенную модель ав- томата для торговли конфетами. В данном конкретном автомате имеются в наличии только пять карманов, заполняемых запасом товара (шоколадными батончиками). Если заполнены все карманы, то в автомат больше нельзя заправить ни одной кон- феты и, аналогично, при отсутствии в автомате шоколадных батончиков какого-то конкретного типа покупатели, желающие приобрести именно их, будут вынуждены возвратиться с пустыми руками. В данном случае ресурсы (карманы для конфет) яв- ляются конечными, и для отслеживания их состояния можно использовать семафор. Исходный код сценария (candy.py) приведен в примере 4.11. Пример 4.11. Автомат для торговли конфетами, управляемый с помощью семафоров ( candy.py) В этом сценарии блокировки и семафоры используются для моделирования авто- мата для торговли конфетами. 1 #!/usr/bin/env python 2 3 from atexit import register 4 from random import randrange 5 from threading import BoundedSemaphore, Lock, Thread 6 from time import sleep, ctime 7 8 lock = Lock() 9 MAX = 5 10 candytray = BoundedSemaphore(MAX) 11 12 def refill(): 13 lock.acquire() 14 print 'Refilling candy...', 15 try: 16 candytray.release() 17 except ValueError: 18 print 'full, skipping' 19: else: 20: print 'OK' 21: lock.release() 22: 23: def buy(): 24: lock.acquire() 25: print 'Buying candy...', 26: if candytray.acquire(False): 27: print 'OK' 28: else: 29: print 'empty, skipping' 06_ch04.indd 221 22.01.2015 22:00:48 Глава 4 Многопоточное программирование 222 30: lock.release() 31: 32: def producer(loops): 33: for i in xrange(loops): 34: refill() 35: sleep(randrange(3)) 36: 37: def consumer(loops): 38: for i in xrange(loops): 39: buy() 40: sleep(randrange(3)) 41: 42: def _main(): 43: print 'starting at:', ctime() 44: nloops = randrange(2, 6) 45: print 'THE CANDY MACHINE (full with %d bars)!' % MAX 46: Thread(target=consumer, args=(randrange( 47: nloops, nloops+MAX+2),)).start() # покупатель 48: Thread(target=producer, args=(nloops,)).start() #продавец 49: 50: @register 51: def _atexit(): 52: print 'all DONE at:', ctime() 53: 54: if __name__ == '__main__': 55: _main() Построчное объяснение Строки 1–6 Строки запуска и импорта практически полностью совпадают с теми, которые были приведены в предыдущих примерах этой главы. Добавлено лишь объявление семафора. В модуле threading предусмотрены два класса семафоров, Semaphore и BoundedSemaphore. Как уже было сказано, в действительности семафоры — просто счетчики; при запуске семафора задается некоторое постоянное число, которое опре- деляет конечное количество единиц ресурса. Значение этого счетчика уменьшается на 1 после потребления одной единицы из конечного количества единиц ресурса, а после возврата этой единицы в пул значение счетчика увеличивается. Объект BoundedSemaphore предоставляет дополнительную возможность, которая состоит в том, что значение счетчика не может быть увеличено свыше установленного для него начального значения; иными словами, позволяет пре- дотвратить возникновение такой абсурдной ситуации, при которой количество опе- раций освобождения семафора превышает количество операций его захвата. Строки 8–10 Глобальные переменные в этом сценарии определяют блокировку, константу, представляющую максимальное количество единиц ресурса, которые могут составить потребляемый запас, а также лоток с конфетами. 06_ch04.indd 222 22.01.2015 22:00:48 223 4.7. Практическое применение многопоточной обработки Строки 12–21 Оператор, обслуживающий эти вымышленные торговые автоматы, время от вре- мени пополняет запасы товара. Для моделирования этого действия применяется функция refill(). Вся процедура представляет критический участок кода; именно поэтому захват блокировки становится обязательным условием, при котором могут быть беспрепятственно выполнены все строки от начала до конца. В коде предусмо- трен вывод журнала со сведениями о выполненных действиях, с которым может оз- накомиться пользователь. Кроме того, вырабатывается предупреждение при попытке превысить максимально допустимый объем запасов (строки 17-18). Строки 23–30 Функция buy() противоположна по своему назначению функции refill(); она позволяет покупателю каждый раз получать по одной единице из хранимых запасов. Для обнаружения того, не исчерпаны ли уже конечные ресурсы, применяется услов- ное выражение (строка 26). Значение счетчика ни в коем случае не может стать отри- цательным, поэтому после достижения нулевого значения количества запасов вызов функции buy() блокируется до того момента, когда значение счетчика будет снова увеличено. Если этой функции передается значение неблокирующего флага, False, то вместо блокирования функция возвращает False, указывая на то, что ресурсы ис- черпаны. Строки 32–40 Функции producer() и consumer(), моделирующие пополнение и потребление запаса конфет, просто повторяются в цикле и обеспечивают выполнение соответству- ющих вызовов функций refill() и buy(), приостанавливаясь на короткое время между вызовами. Строки 42–55 В последней части кода сценария содержится вызов функции _main(), выпол- няемый при запуске сценария из командной строки, предусмотрена регистрация функции выхода, а также приведено определение функции _main(), в которой пред- усмотрена инициализация вновь созданной пары потоков, которые моделируют по- полнение и потребление конфет. В коде создания потока, который представляет покупателя (потребителя), допол- нительно предусмотрена возможность вводить установленное случайным образом положительное смещение, при котором покупатель фактически может попытаться получить больше шоколадных батончиков, чем заправлено в торговый автомат по- ставщиком/производителем (в противном случае в коде никогда не возникла бы си- туация, в которой покупателю разрешалось бы совершать попытку купить шоколад- ный батончик из пустого автомата). Выполнение сценарием приводит к получению примерно такого вывода: $ python candy.py starting at: Mon Apr 4 00:56:02 2011 THE CANDY MACHINE (full with 5 bars)! Buying candy... OK Refilling candy... OK Refilling candy... full, skipping Buying candy... OK 06_ch04.indd 223 22.01.2015 22:00:49 Глава 4 Многопоточное программирование 224 Buying candy... OK Refilling candy... OK Buying candy... OK Buying candy... OK Buying candy... OK all DONE at: Mon Apr 4 00:56:08 2011 В процессе отладки этого сценария может потребоваться вмешательство вручную Если в какой-то момент нужно будет приступить к отладке сценария, в котором ис- пользуются семафоры, прежде всего необходимо точно знать, какое значение находит- ся в счетчике семафора в любое заданное время. В одном из упражнений в конце дан- ной главы предлагается реализовать такое решение в сценарии candy.py, возможно, переименовав его в candydebug.py, и предусмотреть в нем возможность отображать значение счетчика. Для этого может потребоваться рассмотреть исходный код модуля threading.py (причем, вероятно, и в версии Python 2, и в версии Python 3). При этом следует учитывать, что имена примитивов синхронизации модуля threading не являются именами классов, даже притом, что в них используется вы- деление прописными буквами, как в ВерблюжьемСтиле, что делает их похожими на классы. В действительности эти примитивы представляют собой всего лишь одностроч- ные функции, применяемые для порождения необходимых объектов. При работе с мо- дулем threading необходимо учитывать две сложности: во-первых, невозможно поро- ждать подклассы создаваемых объектов (поскольку они представляют собой функции), а во-вторых, при переходе от версий 2.x к версиям 3.x изменились имена переменных. Обе эти сложности можно было бы легко преодолеть, если бы в обеих версиях объек- ты предоставляли свободный и единообразный доступ к счетчику, что в действитель- ности не обеспечивается. Непосредственный доступ к значению счетчика может быть получен, поскольку он представляет собой всего лишь атрибут класса, но, как уже было сказано, имя переменной self.__value изменилось. Это означает, что переменная self._Semaphore__value в Python 2 была переименована в self._value в Python 3. Что касается разработчиков, то наиболее удобный вариант применения интерфейса прикладного программирования API (по крайней мере, по мнению автора) состоит в создании подкласса класса threading._BoundedSemaphore и реализации метода __len__(). Но если планируется поддерживать сценарий и в версии 2.x, и в версии 3.x, то следует использовать правильное значение счетчика, как было только что описано. |