Szintaxis
Több processzt és valamilyen IPC-t használó párhuzamos programot minden nyelven írhatunk. Python programokban használhatunk viszont szálakat is, melyek a különálló processzekkel szemben közös adatterülettel rendelkeznek. Emiatt a szálak használatánál szükség van egy olyan mechanizmusra, ami a közösen használt adatokat védi attól, hogy egyszerre több szál is módosítsa, vagy, hogy valaki olvassa, miközben módosítás történik. A Python lehetőséget ad ezeknek az adatok a védelmére lockokkal, szemaforokkal.
Szálak használatára ad lehetőséget például a thread modul. Ez a modul POSIX szálakat használó operációs rendszereken támogatott. A támogatott platformok között van a Windows NT/95, SGI IRIX, Solaris 2.x, Linux. Ez a modul minimális lehetőségeket ad a szálak kezelésére. Általában, ha egy programban a főszál abortál, akkor minden szál futása véget ér. Nem így SGI IRIX operációs rendszeren, ahol a szálak tovább élnek. Ezek miatt érdemes figyelni arra, hogy a szálak futnak-e, amikor a fő szál éppen kilép. A thread modul nem ad lehetőséget ennek lekérdezésére, ezt nekünk kell valamilyen módon megvalósítani.
Egy lehetséges megoldást mutat be a következő Python program. A program működését leírják a dokumentációs sztringek és a kommentek a programban:
#!/usr/bin/python
import thread
import random
from time import sleep
class ThreadControl:
'''Ez az osztály felügyeli a szálakat. Minden szál hozzá regisztálja be
magát, ebből a felügyelő tudni fogja, hogy hány szál fut még.
A fő szál ezek után megvárhatja, míg minden szál befejeződik, és csak
utána lép ki, esélyt adva ezzel arra, hogy minden szál tisztességesen
befejezze a működését.
'''
threads_should_quit = 0
counter = 0 # szál számláló
# lock objektum a számláló thread-safe módosításához
lock = thread.allocate_lock()
def add(self):
'Regisztrál egy új szálat. Nyilvántartás nincs, csak számolás.'
self.lock.acquire()
self.counter += 1
self.lock.release()
def remove(self):
'Eggyel csökkenti a számolt szálak számát.'
self.lock.acquire()
self.counter -= 1
self.lock.release()
def anyThreadRunning(self):
'Igazat ad vissza, ha van még futó szál, egyébként hamisat.'
self.lock.acquire()
count = self.counter > 0
self.lock.release()
return count
def threadsShouldQuit(self, value=None):
'''Ez a függvény visszaadja a threads_should_quit változó
értékét, amivel a főprogram jelzi a szálaknak, hogy szeretné, ha
befejeződnének. Ha van második paraméter megadva, akkor
beállítja ezt az értéket.'''
if not value is None:
self.threads_should_quit = value
return self.threads_should_quit
def thread_control_function(control):
'''Ez a függvény fogja a szál működését meghatározni. A szál addig fut,
amíg ez a függvény fut. Minden szálban külön névtér tartozik ehhez a
függvényhez, amit bizonyít az, hogy a rand objektum azonosítója
mindhárom szálnál különböző.'''
# a korrekt működés egyik feltétele a körültekintően működő szál
# jelzés a felügyelőnek, hogy új szál indul
control.add()
# a szál belső azonosítójának kiírása
print "%d: thread running" % thread.get_ident()
# Random példány későbbi használatra, és a példány id-jának kiírása
rand = random.Random()
print "%d: random object's id is %d"\
% (thread.get_ident(), id(rand))
# jelzésig nyugodtan futhat a szál
while not control.threadsShouldQuit():
sleep_time = rand.random() * 5 + 1
print "%d: sleeping for %1.2f seconds"\
% (thread.get_ident(), sleep_time)
# véletlen ideig tétlenek leszünk
sleep(sleep_time)
# a jelzés megérkezett, a szál befejeződik
print "%d: THREAD EXITING (random object at %d)"\
% (thread.get_ident(), id(rand))
# a felügyelő csökkentheti a szálak számát
control.remove()
# MAIN
# ThreadControl példányosítás
thread_control = ThreadControl()
# 3 szál létrehozása
for i in range(3):
# új szál indítása, a szál a thread_control_function függvényt fogja
# futtatni, és a második paraméterben adott tuple tartalmazza a neki
# átadandó paramétereket, pillanatnyilag egyedül a ThreadControl
# példányt
thread.start_new_thread(thread_control_function, (thread_control,))
# a szálak kapnak egy kis időt, hogy fussanak
sleep(10)
# jelezzük a szálaknak, hogy új műveletbe már ne kezdjenek bele
thread_control.threadsShouldQuit(1)
print "THREADS SHOULD QUIT!"
# türelmesen megvárjuk, míg minden szál befejezi a futását, 0.1
# másodpercenként leellenőrizve ezt
while thread_control.anyThreadRunning():
sleep(0.1)
import thread
import random
from time import sleep
class ThreadControl:
'''Ez az osztály felügyeli a szálakat. Minden szál hozzá regisztálja be
magát, ebből a felügyelő tudni fogja, hogy hány szál fut még.
A fő szál ezek után megvárhatja, míg minden szál befejeződik, és csak
utána lép ki, esélyt adva ezzel arra, hogy minden szál tisztességesen
befejezze a működését.
'''
threads_should_quit = 0
counter = 0 # szál számláló
# lock objektum a számláló thread-safe módosításához
lock = thread.allocate_lock()
def add(self):
'Regisztrál egy új szálat. Nyilvántartás nincs, csak számolás.'
self.lock.acquire()
self.counter += 1
self.lock.release()
def remove(self):
'Eggyel csökkenti a számolt szálak számát.'
self.lock.acquire()
self.counter -= 1
self.lock.release()
def anyThreadRunning(self):
'Igazat ad vissza, ha van még futó szál, egyébként hamisat.'
self.lock.acquire()
count = self.counter > 0
self.lock.release()
return count
def threadsShouldQuit(self, value=None):
'''Ez a függvény visszaadja a threads_should_quit változó
értékét, amivel a főprogram jelzi a szálaknak, hogy szeretné, ha
befejeződnének. Ha van második paraméter megadva, akkor
beállítja ezt az értéket.'''
if not value is None:
self.threads_should_quit = value
return self.threads_should_quit
def thread_control_function(control):
'''Ez a függvény fogja a szál működését meghatározni. A szál addig fut,
amíg ez a függvény fut. Minden szálban külön névtér tartozik ehhez a
függvényhez, amit bizonyít az, hogy a rand objektum azonosítója
mindhárom szálnál különböző.'''
# a korrekt működés egyik feltétele a körültekintően működő szál
# jelzés a felügyelőnek, hogy új szál indul
control.add()
# a szál belső azonosítójának kiírása
print "%d: thread running" % thread.get_ident()
# Random példány későbbi használatra, és a példány id-jának kiírása
rand = random.Random()
print "%d: random object's id is %d"\
% (thread.get_ident(), id(rand))
# jelzésig nyugodtan futhat a szál
while not control.threadsShouldQuit():
sleep_time = rand.random() * 5 + 1
print "%d: sleeping for %1.2f seconds"\
% (thread.get_ident(), sleep_time)
# véletlen ideig tétlenek leszünk
sleep(sleep_time)
# a jelzés megérkezett, a szál befejeződik
print "%d: THREAD EXITING (random object at %d)"\
% (thread.get_ident(), id(rand))
# a felügyelő csökkentheti a szálak számát
control.remove()
# MAIN
# ThreadControl példányosítás
thread_control = ThreadControl()
# 3 szál létrehozása
for i in range(3):
# új szál indítása, a szál a thread_control_function függvényt fogja
# futtatni, és a második paraméterben adott tuple tartalmazza a neki
# átadandó paramétereket, pillanatnyilag egyedül a ThreadControl
# példányt
thread.start_new_thread(thread_control_function, (thread_control,))
# a szálak kapnak egy kis időt, hogy fussanak
sleep(10)
# jelezzük a szálaknak, hogy új műveletbe már ne kezdjenek bele
thread_control.threadsShouldQuit(1)
print "THREADS SHOULD QUIT!"
# türelmesen megvárjuk, míg minden szál befejezi a futását, 0.1
# másodpercenként leellenőrizve ezt
while thread_control.anyThreadRunning():
sleep(0.1)
A program kimenete valami ilyesmi lesz:
1026: thread running
1026: random object's id is 135084972
1026: sleeping for 1.58 seconds
2051: thread running
2051: random object's id is 135310788
2051: sleeping for 1.69 seconds
3076: thread running
3076: random object's id is 135305820
3076: sleeping for 1.80 seconds
1026: sleeping for 3.23 seconds
2051: sleeping for 2.55 seconds
3076: sleeping for 1.87 seconds
3076: sleeping for 1.04 seconds
2051: sleeping for 2.19 seconds
3076: sleeping for 3.49 seconds
1026: sleeping for 3.34 seconds
2051: sleeping for 5.23 seconds
1026: sleeping for 1.97 seconds
3076: sleeping for 5.66 seconds
THREADS SHOULD QUIT!
1026: THREAD EXITING (random object at 135084972)
2051: THREAD EXITING (random object at 135310788)
3076: THREAD EXITING (random object at 135305820)
1026: random object's id is 135084972
1026: sleeping for 1.58 seconds
2051: thread running
2051: random object's id is 135310788
2051: sleeping for 1.69 seconds
3076: thread running
3076: random object's id is 135305820
3076: sleeping for 1.80 seconds
1026: sleeping for 3.23 seconds
2051: sleeping for 2.55 seconds
3076: sleeping for 1.87 seconds
3076: sleeping for 1.04 seconds
2051: sleeping for 2.19 seconds
3076: sleeping for 3.49 seconds
1026: sleeping for 3.34 seconds
2051: sleeping for 5.23 seconds
1026: sleeping for 1.97 seconds
3076: sleeping for 5.66 seconds
THREADS SHOULD QUIT!
1026: THREAD EXITING (random object at 135084972)
2051: THREAD EXITING (random object at 135310788)
3076: THREAD EXITING (random object at 135305820)
Szinkronizáció
A mutex modul is lehetőséget ad kölcsönös kizárás megvalósítására. Bár nem kizárólag többszálas programoknál használható, de ezekben a programokban különösen hasznos lehet. Ez a modul egy zárolási kérés esetén egy sorba gyűjti a végrehajtandó függvényeket. A modul lehetőséget ad arra is, hogy egy erőforrás zárolt voltát ellenőrizzük, és arra is, hogy az ellenőrzéssel egyidőben atomi művelettel zároljuk is az erőforrást. Ennek akkor van jelentősége, ha egy erőforrás zároltsága esetén más műveletet szeretnénk végezni, és később esetleg ismét megpróbálkozni a zárolással.
Másik lehetőség a threading modul használata. Ez a modul szálak egy magasabb szintű kezelését teszi lehetővé, mint a thread modul. Nincs szükség rá, hogy mi tartsuk nyilván a futó szálakat. Használható egyszerre mindkét modul is - de a gyakorlatban ezt inkább kerüljük. Ugyanis, ha nem a threading modulon keresztül hoztuk létre a szálakat, akkor a threading modul nyilvántartása nem fogja tartalmazni ezeket a szálakat, így ha például a currentThread() függvényt használva szeretnénk elkérni az aktuális szál objektumot, akkor egy ún. "dummy" szál objektumot hoz létre, ami csak korlátozott funkciókkal rendelkezik. Ezzel csak feleslegesen bonyolítanánk a kódot.
Ezen kívűl a modul sok hasznos osztályt definiál:
- Event - ennek egy példánya egy eseményt reprezentálhat, ami lehet bekövetkezett, be nem következett, és a wait() metódussal blokkolhatunk egy szálat az esemény bekövetkeztéig.
- Lock - ennek egy példánya egy olyan zárat implementál, bárki elengedhet, nem csak az, aki a zárat kérte.
- RLock - ennek egy példánya egy olyan zárat implementál, amit csak az a szál engedhet el, amelyik a zárolást kérte. Egy szál többször is kérheti a zárolást egymás után anélkül, hogy blokkolódna, de pontosan annyiszor kell elengednie is a zárat.
- Semaphore - Dijkstra jól ismert szemaforjának implementációja. Egy zár felengedése esetén a számlálót növeli, ha nincs szál, ami a szemafor által védett erőforrásra várakozna.
- BoundedSemaphore - hasonló az előzőhöz, azzal a különbséggel, hogy figyeli a számlálót, hogy nem megy-e valamikor a példányosításkor megadott kezdeti érték fölé. Ugyanis, ha túl sok a felengedés, akkor az mindenképpen valamilyen programozási hiba jele. Ha túl nagyra nőne ez az érték, akkor egy ValueError kivételt dob.
- Condition - ennek egy példánya egy olyan feltételt valósít meg, ami lehetővé teszi egy vagy több szálnak, hogy egy másik szál jelzésére várjanak.
- Timer - egy függvény futtatását teszi lehetővé bizonyos idő eltelte után.
- Thread - ebből az osztályból örökölve definiálhatjuk saját szálainkat. Csak az __init__ (konstruktor) és a run() metódust szabad felüldefiniálni.
Ezzel a modullal például nem kell már arra figyelnünk arra, hogy van-e még futó szál, amikor a fő szál már kilépne. Minden szálban egy flag mondja meg, hogy a szál démoni-e (daemonic). A főszál terminálás előtt megvár minden nem démon szálat. A főszál nem démoni, minden további szál pedig a létrehozó száltól örökli ennek a flagnek a kezdeti értékét (ami később természetesen módosítható). Az előző példa threading modult használó változatában például nincs démoni szál, így minden további erőfeszítés nélkül megkapja az esélyt minden szál arra, hogy "kitakarítson maga körül" (cleanup).
#!/usr/bin/python
from threading import *
import random
from time import sleep
# öröklés a Thread osztályból
class MyThread(Thread):
# konstruktor
def __init__(self, event, name):
Thread.__init__(self, name=name)
self.event = event
self.rand = random.Random()
def run(self):
# a szálak konstruktorban megadott nevét írjuk, nem azonosítókat
print "%s: RUNNING" % self.getName()
# a konstruktorban eltárolt Event objektum jelzi majd, ha
# bekövetkezik a "be kell fejeződni felhívás" esemény :)
while not event.isSet():
sleep_time = self.rand.random() * 5 + 1
print "%s: sleeping for %1.2f seconds"\
% (self.getName(), sleep_time)
# véletlen ideig tétlenek leszünk
sleep(sleep_time)
# a jelzés megérkezett, a szál befejeződik
print "%s: THREAD EXITING" % self.getName()
# MAIN
# 3 szál létrehozása
event = Event()
for i in range(3):
thread = MyThread(event, "Thread-" + str(i))
thread.start()
# a szálak kapnak egy kis időt, hogy fussanak
sleep(10)
# jelezzük a szálaknak, hogy új műveletbe már ne kezdjenek bele
event.set()
print "THREADS SHOULD QUIT!"
# a threading modulnak köszönhetően, mivel egyik szál sem démoni (daemonic),
# ezért a főszál megvárja a többi szál terminálását
from threading import *
import random
from time import sleep
# öröklés a Thread osztályból
class MyThread(Thread):
# konstruktor
def __init__(self, event, name):
Thread.__init__(self, name=name)
self.event = event
self.rand = random.Random()
def run(self):
# a szálak konstruktorban megadott nevét írjuk, nem azonosítókat
print "%s: RUNNING" % self.getName()
# a konstruktorban eltárolt Event objektum jelzi majd, ha
# bekövetkezik a "be kell fejeződni felhívás" esemény :)
while not event.isSet():
sleep_time = self.rand.random() * 5 + 1
print "%s: sleeping for %1.2f seconds"\
% (self.getName(), sleep_time)
# véletlen ideig tétlenek leszünk
sleep(sleep_time)
# a jelzés megérkezett, a szál befejeződik
print "%s: THREAD EXITING" % self.getName()
# MAIN
# 3 szál létrehozása
event = Event()
for i in range(3):
thread = MyThread(event, "Thread-" + str(i))
thread.start()
# a szálak kapnak egy kis időt, hogy fussanak
sleep(10)
# jelezzük a szálaknak, hogy új műveletbe már ne kezdjenek bele
event.set()
print "THREADS SHOULD QUIT!"
# a threading modulnak köszönhetően, mivel egyik szál sem démoni (daemonic),
# ezért a főszál megvárja a többi szál terminálását
A program kimenete hasonló lesz ehhez:
Thread-0: RUNNING
Thread-0: sleeping for 3.67 seconds
Thread-1: RUNNING
Thread-1: sleeping for 3.67 seconds
Thread-2: RUNNING
Thread-2: sleeping for 3.67 seconds
Thread-1: sleeping for 1.14 seconds
Thread-0: sleeping for 1.14 seconds
Thread-2: sleeping for 1.14 seconds
Thread-0: sleeping for 1.82 seconds
Thread-1: sleeping for 1.82 seconds
Thread-2: sleeping for 1.82 seconds
Thread-1: sleeping for 3.12 seconds
Thread-0: sleeping for 3.12 seconds
Thread-2: sleeping for 3.12 seconds
Thread-0: sleeping for 4.19 seconds
Thread-1: sleeping for 4.19 seconds
Thread-2: sleeping for 4.19 seconds
THREADS SHOULD QUIT!
Thread-1: THREAD EXITING
Thread-0: THREAD EXITING
Thread-2: THREAD EXITING
Thread-0: sleeping for 3.67 seconds
Thread-1: RUNNING
Thread-1: sleeping for 3.67 seconds
Thread-2: RUNNING
Thread-2: sleeping for 3.67 seconds
Thread-1: sleeping for 1.14 seconds
Thread-0: sleeping for 1.14 seconds
Thread-2: sleeping for 1.14 seconds
Thread-0: sleeping for 1.82 seconds
Thread-1: sleeping for 1.82 seconds
Thread-2: sleeping for 1.82 seconds
Thread-1: sleeping for 3.12 seconds
Thread-0: sleeping for 3.12 seconds
Thread-2: sleeping for 3.12 seconds
Thread-0: sleeping for 4.19 seconds
Thread-1: sleeping for 4.19 seconds
Thread-2: sleeping for 4.19 seconds
THREADS SHOULD QUIT!
Thread-1: THREAD EXITING
Thread-0: THREAD EXITING
Thread-2: THREAD EXITING
Ezekben a példákban a standard outputra való írást nem kezeltük korlátozott erőforrásként, pedig az. Az egyszerűség kedvéért most vállaltuk a kockázatot, hogy a szálak egymás soraiba piszkítanak, egy körültekintő programnak azonban figyelnie kell erre is.
A Thread osztály metódusaival hasonló szálkezelést használhatunk, mint ami a Javaban is megtalálgató, noha számos onnan ismert funkciót nem támogat a Python, így például nincs lehetőség szálak felfüggesztésére és későbbi újraindítására, nem lehet egy szálat megállítani vagy megszakítani. Talán a legfontosabb, hogy míg Javaban nyelvi elem van a kölcsönös kizárás megvalósítására, addig Pythonban ezt különálló objektumokkal kell megvalósítani. Van lehetőségünk viszont arra, hogy egy szál join metódusát hívva a hívó szál futása blokkolódjon, amíg a hívott objektum által reprezentált szál futása be nem fejeződik. Ennek a hívásnak egy időkorlátos változata is van, ami az opcionális paraméterrel adható meg.