برنامه نویسی موازی و پردازش همزمان با زبان پایتون (multithreaded programming)

برنامه نویسی موازی و پردازش همزمان با زبان پایتون (multithreaded programming)

اجرای همزمان چندین thread به منزله ی اجرای همزمان چندین برنامه در آن واحد است که مزایای زیر را به دنبال دارد:

• چندین thread که در بستر یک پردازه یا فرایند (process) اجرا شده و data space یکسانی را دارند، می توانند داده ها را بهتر با یکدیگر به اشتراک گذاشته و بایکدیگر تعامل بهتری داشته باشند، نسبت به زمانی که این thread ها در فرایندهای مجزا قرار دارند.
• thread ها که گاهی پردازه یا فرایندهای سبک نیز نامیده می شوند سربار و memory overhead کمتری نسبت به فرایندهای واقعی داشته و کم هزینه تر می باشند.

هر thread یک نقطه ی آغاز،یک ترتیب یا توالی اجرا و یک نقطه ی پایان دارد. علاوه بر آن، یک instruction pointer دارد که دقیقا مشخص می کند برنامه در کجای بستر (context) جاری در حال اجرا بود و به کدام مرحله و نقطه اجرای دستور رسیده است.

• • می توان thread را مختل یا متوقف (pre-empt) کرد.
• • می توان thread را در حالی که دیگر thread ها فعال هستند، به طور موقت به حالت تعلیق درآورد. از این کار تحت عنوان yielding نیز یاد می شود.

راه اندازی و اجرای thread جدید

جهت آغاز یک thread جدید، بایستی متد زیر که داخل ماژول thread کپسوله شده را فراخوانی نمایید:

thread.start_new_thread ( function, args[, kwargs] )

به وسیله ی این متد می توانید به روش سریع و کارا در هر دو محیط ویندوز و لینوکس thread های جدید ایجاد نمایید.

متد مورد نظر بلافاصله بازگشته و thread فرزند آغاز می شود که متعاقبا function را با پارامتر args صدا می زند. زمانی که function به return می رسد، thread خاتمه می یابد.

در این تابع، پارامتر args مجموعه ی چندتایی از آرگومان ها (tuple) است. اگر می خواهید تابع را بدون آرگومان صدا بزنید، بایستی یک tuple خالی به عنوان پارامتر ارسال کنید. kwargs یک آرگومان از نوع dictionary تشکیل شده از کلیدواژه ها بوده و استفاده از آن اختیاری است.

مثال

#!/usr/bin/python
import thread
import time
# Define a function for the thread
def print_time( threadName, delay):
   count = 0
   while count < 5:
      time.sleep(delay)
      count += 1
      print "%s: %s" % ( threadName, time.ctime(time.time()) )
# Create two threads as follows
try:
   thread.start_new_thread( print_time, ("Thread-1", 2, ) )
   thread.start_new_thread( print_time, ("Thread-2", 4, ) )
except:
   print "Error: unable to start thread"
while 1:
   pass

خروجی:

Thread-1: Thu Jan 22 15:42:17 2009
Thread-1: Thu Jan 22 15:42:19 2009
Thread-2: Thu Jan 22 15:42:19 2009
Thread-1: Thu Jan 22 15:42:21 2009
Thread-2: Thu Jan 22 15:42:23 2009
Thread-1: Thu Jan 22 15:42:23 2009
Thread-1: Thu Jan 22 15:42:25 2009
Thread-2: Thu Jan 22 15:42:27 2009
Thread-2: Thu Jan 22 15:42:31 2009
Thread-2: Thu Jan 22 15:42:35 2009

اگرچه توصیه می شود که برای پردازش موازی سطح پایین (low-level threading) از ماژول thread استفاده نمایید، با این وجود ماژول مزبور در مقایسه با ماژول جدید برنامه نویسی موازی (threading module) از قابلیت های بسیار کمتری برخوردار است.

ماژول Threading

ماژول جدیدتری که برای برنامه نویسی موازی همراه با ویرایش 2.4 زبان Python ارائه شد، قابلیت های بسیار بیشتر و سطح بالاتری جهت بهره گیری از پردازش موازی و thread ها در اختیار برنامه نویس قرار داد.

ماژول threading علاوه بر تمامی توابع ماژول قبلی (thread)، تعدادی متد نوین و کارای دیگر جهت پیاده سازی مفهوم برنامه نویسی موازی ارائه می دهد.

• threading.activeCount(): تعداد آبجکت های thread که فعال و در حال اجرا هستند را بازگردانی می نماید.
• threading.currentThread(): thread جاری را به عنوان خروجی برمی گرداند.
• لیستی از تمامی آبجکت های thread که هم اکنون فعال هستند را بازمی گرداند.

علاوه بر متدها، ماژول نام برده کلاس Thread را شامل می شود که threading و پردازش موازی را پیاده سازی می کند. متدهایی که که کلاس Thread در اختیار توسعه دهنده قرار می دهد، به شرح زیر می باشد:

• run(): متد جاری در واقع entry point یا نقطه ی شروع اجرای thread می باشد.
• start(): thread را با فراخوانی تابع run() راه اندازی و اجرا می کند.
• join([time]): متد join() منتظر می ماند تا thread ها خاتمه یابند.
• isAlive(): متد حاضر بررسی می کند آیا یک thread هنوز در حل اجرا است یا خیر.
• getName(): متد جاری اسم آبجکت thread را بازیابی می کند.
• setName(): این متد اسم thread را مقداردهی می کند.

ایجاد آبجکت جدید Thread از ماژول Threading

به منظور پیاده سازی یک thread جدید از ماژول threading، کافی است مراحل زیر را دنبال نمایید:

• یک کلاس مشتق از کلاس Thread ایجاد نمایید (از آن ارث بری کنید).
• متد __init__(self [,args]) را جهت تزریق آرگومان های بیشتر به کلاس، بازنویسی (override) نمایید.
• در پایان، متد run(self [,args]) را پیاده سازی و در بدنه ی آن مشخص کنید که thread به هنگام اجرا چه عملیاتی را انجام دهد.

پس از ایجاد کلاس مشتق (ارث بری) از Thread، می توانید یک نمونه از آن ایجاد کرده و سپس با فراخوانی متد start() یک thread یا ریزپردازه ی دیگر آغاز نمایید که متعاقبا متد run() را صدا می زند.

مثال

#!/usr/bin/python
import threading
import time
exitFlag = 0
class myThread (threading.Thread):
    def __init__(self, threadID, name, counter):
        threading.Thread.__init__(self)
        self.threadID = threadID
        self.name = name
        self.counter = counter
    def run(self):
        print "Starting " + self.name
        print_time(self.name, self.counter, 5)
        print "Exiting " + self.name
def print_time(threadName, delay, counter):
    while counter:
        if exitFlag:
            threadName.exit()
        time.sleep(delay)
        print "%s: %s" % (threadName, time.ctime(time.time()))
        counter -= 1
# Create new threads
thread1 = myThread(1, "Thread-1", 1)
thread2 = myThread(2, "Thread-2", 2)
# Start new Threads
thread1.start()
thread2.start()
print "Exiting Main Thread"

زمانی که کد فوق به اجرا گذاشته می شود، خروجی به صورت زیر خواهد بود:

Starting Thread-1
Starting Thread-2
Exiting Main Thread
Thread-1: Thu Mar 21 09:10:03 2013
Thread-1: Thu Mar 21 09:10:04 2013
Thread-2: Thu Mar 21 09:10:04 2013
Thread-1: Thu Mar 21 09:10:05 2013
Thread-1: Thu Mar 21 09:10:06 2013
Thread-2: Thu Mar 21 09:10:06 2013
Thread-1: Thu Mar 21 09:10:07 2013
Exiting Thread-1
Thread-2: Thu Mar 21 09:10:08 2013
Thread-2: Thu Mar 21 09:10:10 2013
Thread-2: Thu Mar 21 09:10:12 2013
Exiting Thread-2

همزمان سازی thread ها

ماژول threading که در بالا به آن اشاره کردیم، یک مکانیزم اعمال قفل با قابلیت پیاده سازی آسان در اختیار توسعه دهنده قرار می دهد که به واسطه ی آن می توان به راحتی thread ها و اجرای آن ها را هماهنگ ساخت. جهت ساخت و اعمال قفل جدید کافی است متد Lock() فراخوانی شود که در خروجی نمونه ی جدید از آبجکت lock را بازگردانی می نماید.

متد acquire(blocking) از نمونه ی (آّبجکت) جدید lock، این قابلیت را دارد که با مدیریت thread ها آن ها را به طور همزمان (موازی) اجرا کند. به واسطه ی پارامتر اختیاری blocking توسعه دهنده قادر خواهد بود کنترل اینکه آیا قفل بر روی thread اعمال شود یا خیر را بدست گیرد.

زمانی که مقدار پارامتر blocking برابر 0 باشد، اگر قفل یا lock بر روی thread اعمال شد،متد با 1 و اگر نشد با 0 برمی گردد. هنگامی که blocking روی 1 تنظیم شده باشد، thread مسدود (block) شده و منتظر می ماند تا lock آزاد شود.

زمانی که متد release() از نمونه ی جدید lock صدا خورده می شود، lock یا قفل اعمال شده بر روی thread که دیگر کاربردی ندارد، آزاد می شود.

مثال

#!/usr/bin/python
import threading
import time
class myThread (threading.Thread):
    def __init__(self, threadID, name, counter):
        threading.Thread.__init__(self)
        self.threadID = threadID
        self.name = name
        self.counter = counter
    def run(self):
        print "Starting " + self.name
        # Get lock to synchronize threads
        threadLock.acquire()
        print_time(self.name, self.counter, 3)
        # Free lock to release next thread
        threadLock.release()
def print_time(threadName, delay, counter):
    while counter:
        time.sleep(delay)
        print "%s: %s" % (threadName, time.ctime(time.time()))
        counter -= 1
threadLock = threading.Lock()
threads = []
# Create new threads
thread1 = myThread(1, "Thread-1", 1)
thread2 = myThread(2, "Thread-2", 2)
# Start new Threads
thread1.start()
thread2.start()
# Add threads to thread list
threads.append(thread1)
threads.append(thread2)
# Wait for all threads to complete
for t in threads:
    t.join()
print "Exiting Main Thread"

خروجی:

Starting Thread-1
Starting Thread-2
Thread-1: Thu Mar 21 09:11:28 2013
Thread-1: Thu Mar 21 09:11:29 2013
Thread-1: Thu Mar 21 09:11:30 2013
Thread-2: Thu Mar 21 09:11:32 2013
Thread-2: Thu Mar 21 09:11:34 2013
Thread-2: Thu Mar 21 09:11:36 2013
Exiting Main Thread

پیاده سازی queue و قرار دادن آیتم ها در صف بر اساس اولویت در پردازش موازی (Multithreaded Priority Queue)

ماژول Queue به توسعه دهنده این امکان را می دهد تا یک آبجکت queue جدید ایجاد کند که همزمان چندین آیتم را به طور صف بندی شده در خود کپسوله می نماید. برای مدیریت Queue می توانید از توابع زیر استفاده نمایید:

• get(): متد جاری یک آیتم را حذف و از queue بازیابی می کند.
• put(): این متد یک آیتم جدید را به queue اضافه می کند.
• qsize(): تعداد آیتم هایی که در زمان حاضر داخل صف قرار دارند را به عنوان خروجی بازمی گرداند.
• empty(): چنانچه صف یا آبجکت queue خالی بود، مقدار بولی True و در غیر این صورت False را بازمی گرداند.
• full(): چنانچه آبجکت queue ظرفیت خالی نداشت، مقدار بولی True و در غیر این صورت False را بازگردانی می نماید.

مثال

#!/usr/bin/python
import Queue
import threading
import time
exitFlag = 0
class myThread (threading.Thread):
    def __init__(self, threadID, name, q):
        threading.Thread.__init__(self)
        self.threadID = threadID
        self.name = name
        self.q = q
    def run(self):
        print "Starting " + self.name
        process_data(self.name, self.q)
        print "Exiting " + self.name
def process_data(threadName, q):
    while not exitFlag:
        queueLock.acquire()
        if not workQueue.empty():
            data = q.get()
            queueLock.release()
            print "%s processing %s" % (threadName, data)
        else:
            queueLock.release()
        time.sleep(1)
threadList = ["Thread-1", "Thread-2", "Thread-3"]
nameList = ["One", "Two", "Three", "Four", "Five"]
queueLock = threading.Lock()
workQueue = Queue.Queue(10)
threads = []
threadID = 1
# Create new threads
for tName in threadList:
    thread = myThread(threadID, tName, workQueue)
    thread.start()
    threads.append(thread)
    threadID += 1
# Fill the queue
queueLock.acquire()
for word in nameList:
    workQueue.put(word)
queueLock.release()
# Wait for queue to empty
while not workQueue.empty():
    pass
# Notify threads it's time to exit
exitFlag = 1
# Wait for all threads to complete
for t in threads:
    t.join()
print "Exiting Main Thread"

نتیجه:

Starting Thread-1
Starting Thread-2
Starting Thread-3
Thread-1 processing One
Thread-2 processing Two
Thread-3 processing Three
Thread-1 processing Four
Thread-2 processing Five
Exiting Thread-3
Exiting Thread-1
Exiting Thread-2
Exiting Main Thread