thread_ – ایجاد ترد

مقدمه

در python 2 برای برنامه‌نویسی multithreaded استفاده می‌شد. این ماژول اکنون در python 3 با نام _thread شناخته می‌شود.

این ماژول برای برنامه‌نویسی با تکنیک multithreading استفاده می‌شود. نسبت به ماژول threading امکانات کمتری دارد بنابراین برنامه‌نویسی با آن کمی پیچیده تر است. کدنویسی با این ماژول نسبت به ماژول threading بیشتر مستعد خطاست زیرا بسیاری از امکانات پیشرفته‌تر در آن وجود نداشته و برنامه‌نویس خودش باید آن ها را پیاده سازی کند. این ماژول قبلا در python 2 با نام thread استفاده می‌شد.

استفاده از این ماژول مدت‌هاست که به دلیل محدودیت‌هایی که دارد منسوخ شده است. اما یادگیری آن همچنان می‌تواند مفید باشد زیرا ممکن است در برنامه‌ای قدیمی استفاده شده باشد و برای نگهداری آن برنامه، ناچار به سروکله زدن با این ماژول باشیم.

نحوه استفاده

به کد زیر توجه کنید. فرض کنید اجرای تابع say_hi از نظر زمانی پرهزینه است. این پرهزینه بودن را به کمک تابع time.sleep شبیه‌سازی شده است.

import time

def say_hi(name):
    time.sleep(1)
    print('hi', name)

names = ['Ali', 'Reza', 'Mahboobeh', 'Saeed']
for name in names:
    say_hi(name)

import time

def say_hi(name):
    time.sleep(1)
    print('hi', name)

names = ['Ali', 'Reza', 'Mahboobeh', 'Saeed']
for name in names:
    say_hi(name)

با کمی تغییر در این کد و استفاده از ماژول _thread زمان اجرا به طرز چشمگیری کاهش می‌یابد.

import time
import _thread

def say_hi(name):
    time.sleep(1)
    print('hi', name)

names = ['Ali', 'Reza', 'Mahboobeh', 'Saeed']
for name in names:
    _thread.start_new_thread(say_hi, (name, ))

import time
import _thread

def say_hi(name):
    time.sleep(1)
    print('hi', name)

names = ['Ali', 'Reza', 'Mahboobeh', 'Saeed']
for name in names:
    _thread.start_new_thread(say_hi, (name, ))

اما بعد از اجرای کد بالا متوجه یک مشکل عجیب خواهید شد. اجرای برنامه بدون اینکه چیزی چاپ شود به پایان می‌رسد. فعلا با صرف نظر از علت این مشکل، یک input اضافه می‌کنیم تا روی اتمام اجرای برنامه کنترل پیدا کنیم.

import time
import _thread

def say_hi(name):
    time.sleep(1)
    print('hi', name)

names = ['Ali', 'Reza', 'Mahboobeh', 'Saeed']
for name in names:
    _thread.start_new_thread(say_hi, (name, ))

input('Wait for the result and then press enter to exit!\n')

import time
import _thread

def say_hi(name):
    time.sleep(1)
    print('hi', name)

names = ['Ali', 'Reza', 'Mahboobeh', 'Saeed']
for name in names:
    _thread.start_new_thread(say_hi, (name, ))

input('Wait for the result and then press enter to exit!\n')

حال با اجرای کد بالا می‌توانید خروجی را ببینید و متوجه شوید که چقدر سرعت اجرای برنامه افزایش پیدا کرده است.

بررسی مشکل ماژول _thread

همیشه همه برنامه‌ها صرف نظر از اینکه در چه پارادایمی و با چه تکنیکی توسعه داده شده اند، در. یک thread اصلی به نام mail thread اجرا می‌شوند. هر وقت تسک هایی که در main thread مشخص شده اند به پایان برسد برنامه بسته می‌شود.

در تکه کد زیر تنها تسکی که برای main thread مشخص شده است این است که صرفا با استفاده از _thread.start_new_thread تعدادی thread بسازد. دقت داشته باشید که در main thread مشخص نشده است که آیا باید برای اتمام اجرای تسک های thread هایی که ساخته شده صبر کنیم یا خیر. به همین دلیل main thread به محض اینکه thread های فرعی را می‌سازد تسک هایش تمام می‌شود و به همین دلیل برنامه با خاتمه تسک های mail thread بسته می‌شود. این در صورتی است که سایر thread ها هنوز تسکشان تمام نشده است.

از آنجایی که عملیات print در thread های فرعی قرار دارد، قبل از اینکه چیزی چاپ شود اجرای برنامه پایان می‌یابد.

import time
import _thread

def say_hi(name):
    time.sleep(1)
    print('hi', name)

names = ['Ali', 'Reza', 'Mahboobeh', 'Saeed']
for name in names:
    _thread.start_new_thread(say_hi, (name, ))

import time
import _thread

def say_hi(name):
    time.sleep(1)
    print('hi', name)

names = ['Ali', 'Reza', 'Mahboobeh', 'Saeed']
for name in names:
    _thread.start_new_thread(say_hi, (name, ))

در زیر یک نمودار قرار داده‌ام که چگونگی عملکرد کد بالا را مشخص می‌کند.

برای اینکه کاری کنیم تا طول عمر main thread افزایش یابد و بعد از اجرای thread های دیگر متوقف شود نه قبل از آن، می‌توانیم از یک تابع input استفاده کنیم. در این صورت تا زمانی که کاربر دکمه Enter را نزند، main thread زنده خواهد بود.

import time
import _thread

def say_hi(name):
    time.sleep(1)
    print('hi', name)

names = ['mmreza', 'Fatima', 'Mahboobeh', 'Saeed']
for name in names:
    _thread.start_new_thread(say_hi, (name, ))

input('Press enter to exit!\n')

import time
import _thread

def say_hi(name):
    time.sleep(1)
    print('hi', name)

names = ['mmreza', 'Fatima', 'Mahboobeh', 'Saeed']
for name in names:
    _thread.start_new_thread(say_hi, (name, ))

input('Press enter to exit!\n')

درست است که توانستیم طول عمر main thread را کنترل کنیم اما این روش اصلا روش مناسبی نیست. مثلا در برنامه‌های تحت وب که با فریمورک‌هایی مثل django و flask توسعه می‌یابند امکان ندارد که با این روش بتوانیم طول عمر main thread را کنترل کنیم. چیزی که در اینجا نیاز داریم شبیه به کاریست که متد join در ماژول threading انجام می‌دهد. متاسفانه چنین چیزی در ماژول _thread نداریم و خودمان باید پیاده‌سازی کنیم.

شبیه سازی join برای _thread

برای شبیه سازی متد join سه روش داریم:

• استفاده از شرط
• استفاده از شرط به همراه حلقه while
• قفل کردن منابع مشترک

شبیه‌سازی join با استفاده از شرط

اگر بتوانیم برای دو چالش زیر راه حلی پیدا کنیم مشکل حل می‌شود.

• چالش اول: فهمیدن اینکه چه زمانی اجرای همه thread های فرعی با موفقیت به پایان رسیده است.
• چالش دوم: وقتی اجرای همه thread های فرعی به پایان رسید، اجرای برنامه را با بستن mail thread پایان دهیم.

import os
import time
import _thread

def say_hi(name):
    global names
    global counter
    time.sleep(1)
    print('hi', name)
    counter += 1
    if counter == len(names):
        os._exit(0)

names = ['mmreza', 'Fatima', 'Mahboobeh', 'Saeed']
counter = 0
for name in names:
    _thread.start_new_thread(say_hi, (name, ))

input('Please wate untill the program execute completely!\n')

import os
import time
import _thread

def say_hi(name):
    global names
    global counter
    time.sleep(1)
    print('hi', name)
    counter += 1
    if counter == len(names):
        os._exit(0)

names = ['mmreza', 'Fatima', 'Mahboobeh', 'Saeed']
counter = 0
for name in names:
    _thread.start_new_thread(say_hi, (name, ))

input('Please wate untill the program execute completely!\n')

برای حل چالش اول، یک شمارنده به نام counter تعریف کرده و مقدار آن را هر بار که یک thread اجرا می‌شود یک عدد افزایش می‌دهیم. و برای حل چالش دوم، هرگاه مقدار counter با تعداد thread ها برابر شد با استفاده از os._exit(0) از برنامه خارج می‌شویم.

این روش دو عیب دارد:

• قابل اطمینان نیست زیرا مطمئن نیستم که os._exit همه جا در همه فریمورک ها و در همه سخت افزار ها بدون مشکل کار کند.
• پیچیده است زیرا نیاز به پیاده‌سازی الگوریتمی دارد

شبیه سازی join با استفاده از حلقه while

به با کمی تغییر در کد بالا می‌توان عملیات خروج از mail thread را با کمک یک حلقه while به بیرون از تابع say_hi منتقل کرد. این روش مزیت خاصی نسبت به روش قبل ندارد. صرفا حاوی یک خلاقیت الگوریتمی است به آن اشاره کرده ام.

import os
import time
import _thread

def say_hi(name):
    global names
    global counter
    time.sleep(1)
    print('hi', name)
    counter += 1

names = ['mmreza', 'Fatima', 'Mahboobeh', 'Saeed']
counter = 0
for name in names:
    _thread.start_new_thread(say_hi, (name, ))

# منتظر ماندن برای پایان نخ‌ها
while True:
    if counter == len(names):
        break
    time.sleep(0.1)  # خواب کوتاه پردازشگر برای جلوگیری از اشغال بیش از حد

import os
import time
import _thread

def say_hi(name):
    global names
    global counter
    time.sleep(1)
    print('hi', name)
    counter += 1

names = ['mmreza', 'Fatima', 'Mahboobeh', 'Saeed']
counter = 0
for name in names:
    _thread.start_new_thread(say_hi, (name, ))

# منتظر ماندن برای پایان نخ‌ها
while True:
    if counter == len(names):
        break
    time.sleep(0.1)  # خواب کوتاه پردازشگر برای جلوگیری از اشغال بیش از حد

شبیه سازی join با قفل کردن منابع مشترک

قفل‌گذاری علاوه بر اینکه به ما کمک می‌کند تا مشکل Race Condition را حل کنیم، می‌تواند در شبیه‌سازی متد join نیز کمک کند. برای درک بهتر این موضوع پیشنهاد میکنم نگاهی به مفهوم قفل‌گذاری در Multithreaded Programming داشته باشید.

import time
import _thread

def say_hi(name, lock):
    time.sleep(1)
    print('hi', name)
    lock.release()

if __name__ == '__main__':
    names = ['mmreza', 'Fatima', 'Mahboobeh', 'Saeed']
    locks = []

    for name in names:
        lock = _thread.allocate_lock()
        lock.acquire()
        locks.append(lock)
        _thread.start_new_thread(say_hi, (name, lock))

    # انتظار برای تکمیل تمام نخ‌ها
    for lock in locks:
        lock.acquire()

    print('Successfully Done!')

import time
import _thread

def say_hi(name, lock):
    time.sleep(1)
    print('hi', name)
    lock.release()

if __name__ == '__main__':
    names = ['mmreza', 'Fatima', 'Mahboobeh', 'Saeed']
    locks = []

    for name in names:
        lock = _thread.allocate_lock()
        lock.acquire()
        locks.append(lock)
        _thread.start_new_thread(say_hi, (name, lock))

    # انتظار برای تکمیل تمام نخ‌ها
    for lock in locks:
        lock.acquire()

    print('Successfully Done!')

سوالات

• زمانی که از ماژول _thread استفاده می‌کنیم، چطور می‌توانیم از اجرای همه thread ها اطمینان حاصل کنیم و سپس دسته بعدی thread ها را اجرا کنیم؟

تمرین ها

تمرین اول – ایجاد thread

اجرای برنامه زیر زمان‌بر است. با استفاده از ماژول _thread آن را در پارادایتم concurrent programming بازنویسی کنید.

import time

def say_hi(name):
    time.sleep(1)
    print('hi', name)

names = ['mmreza', 'Fatima', 'Mahboobeh', 'Saeed']
for name in names:
    say_hi(name)

import time

def say_hi(name):
    time.sleep(1)
    print('hi', name)

names = ['mmreza', 'Fatima', 'Mahboobeh', 'Saeed']
for name in names:
    say_hi(name)

تمرین دوم- دانلودر

تکه کد زیر برنامه‌ای است که لیستی از آدرس فایل‌ها را می‌گیرد و دانلود می‌کند. این برنامه را باز نویسی کرده و سرعت دانلود فایل‌ها را با استفاده از ماژول _thread افزایش دهید.

import requests
import os

# فهرست آدرس‌های فایل‌ها
file_urls = [f"https://dummyimage.com/{i}.png" for i in range(300, 400)]

# تابع برای دانلود هر فایل
def download_file(file_url, cookies, headers):
    file_name = os.path.basename(file_url)  # نام فایل از آخرین بخش آدرس استخراج می‌شود
    file_path = os.path.join("downloads", file_name)  # مسیر محلی برای ذخیره فایل‌ها
    print(f"Downloading {file_name}...")
    # دانلود فایل با استفاده از کتابخانه requests
    response = requests.get(file_url, cookies=cookies, headers=headers)
    with open(file_path, 'wb') as f:
        f.write(response.content)
    print(f"{file_name} downloaded successfully.")

# ایجاد پوشه downloads اگر وجود نداشته باشد
if not os.path.exists("downloads"):
    os.makedirs("downloads")

# شروع دانلود هر فایل به صورت متوالی
for file_url in file_urls:
    download_file(file_url, {}, {})

print("All files downloaded successfully.")

import requests
import os

# فهرست آدرس‌های فایل‌ها
file_urls = [f"https://dummyimage.com/{i}.png" for i in range(300, 400)]

# تابع برای دانلود هر فایل
def download_file(file_url, cookies, headers):
    file_name = os.path.basename(file_url)  # نام فایل از آخرین بخش آدرس استخراج می‌شود
    file_path = os.path.join("downloads", file_name)  # مسیر محلی برای ذخیره فایل‌ها
    print(f"Downloading {file_name}...")
    # دانلود فایل با استفاده از کتابخانه requests
    response = requests.get(file_url, cookies=cookies, headers=headers)
    with open(file_path, 'wb') as f:
        f.write(response.content)
    print(f"{file_name} downloaded successfully.")

# ایجاد پوشه downloads اگر وجود نداشته باشد
if not os.path.exists("downloads"):
    os.makedirs("downloads")

# شروع دانلود هر فایل به صورت متوالی
for file_url in file_urls:
    download_file(file_url, {}, {})

print("All files downloaded successfully.")

تمرین سوم – گزارش ساز برای دانلودر

پس از اینکه کد بالا را با استفاده از _thread بازنویسی کردید، یک گزارش‌ساز به برنامه اضافه کنید به نحوی که وضعیت دانلود شدن فایل ها را در یک فایل متنی ذخیره کند.