آموزش ماژول multiprocessing – ویژگی ها

start

اجرای این متد به سیستم‌عامل اطلاع می‌دهد که در اولین فرصت ممکن process مربوط به آن باید اجرا شود.

from multiprocessing import Process

def task(name):
    print(f'{name} is running')

Process(target=task, args=('process1', )).start()
Process(target=task, args=('process2', )).start()
print('main process is running')

خروجی کد بالا مشابه زیر خواهد بود.

[tabby title=”خروجی اول”]

process1 is running
main process is running
process2 is running

[tabby title=”خروجی دوم”]

main process is running
process1 is running
process2 is running

[tabbyending]

خروجی بالا کاملا نشان دهنده این است که لزما اجرای process ها ترتیب خاصی ندارد. ترتیب اجرای آن‌ها را سیستم‌عامل مشخص می‌کند.

join

همیشه برای یک process تعریف می‌شود و مادامی که آن process به اتمام نرسیده باشد از اجرای کدهای خطوط بعدی جلوگیری می‌کند.

به تکه کد زیر توجه کنید.

from time import sleep
from multiprocessing import Process

def task():
    print(f'Task started')
    sleep(1)
    print(f'Task completed')

p = Process(target=task)
p.start()
print('End of the program!')

خروجی این کد مشابه زیر است:

End of the program!
Task started
Task completed

مشکلی که کد بالا دارد این است که ابتدا خروجی خط ۱۱ نمایش داده می‌شود و سپس نتیجه process ها. ما میخواهیم پیام خط ۱۱ آخرین چیزی باشد که print می‌شود. به عبارتی دیگر می‌خواهیم خروجی مشابه زیر باشد:

Task started
Task completed
End of the program!

بدین منظور تغییر زیر را در کد بالا ایجاد می‌کنیم.

from time import sleep
from multiprocessing import Process

def task():
    print(f'Task started')
    sleep(1)
    print(f'Task completed')

p = Process(target=task)
p.start()
p.join()
print('End of the program!')

current_process

به ما اجاره می‌دهد که اطلاعات process ای که در حال اجراست را بگیریم و در صورت نیاز با آن تعاملاتی نیز داشته باشیم. به تکه کد زیر توجه کنید.

from time import sleep
from multiprocessing import current_process, Process

def process1():
    available_methods_and_attrs = [i for i in dir(current_process()) if not i.startswith('_')]
    print(
        'Available methods and attrs for the current_thread() are:',
        ', '.join(available_methods_and_attrs)
    )
    print(f'Process {current_process().name} started')
    sleep(1)
    print(f'Process {current_process().name} completed')

Process(target=process1, name='Process_1').start()

توضیح کد بالا:

• خط ۵: لیست همه متدها و ویژگی‌هایی که current_process در اختیار ما قرار می‌دهد را با استفاده از یک list comprehension به دست آورده ایم.
• خط ۶ تا ۹: لیستی که در خط ۵ به دست آوردیم را چاپ کرده‌ایم.
• خط ۱۰: نام process را در یک پیام چاپ کرده‌ایم.
• خط ۱۱: نام process را در یک پیام چاپ کرده‌ایم.

خروجی کد بالا مشابه زیر است:

Available methods and attrs for the current_process() are: authkey, close, daemon, exitcode, ident, is_alive, join, kill, name, pid, run, sentinel, start, terminate
Process Process_1 started
Process Process_1 completed

authkey

در پایتون، زمانی که از ماژول multiprocessing برای ایجاد پروسس‌ها استفاده می‌کنیم، ممکن است نیاز به ارتباط بین پروسس‌ها داشته باشیم. یکی از مکانیزم‌های امنیتی که ماژول multiprocessing ارائه می‌دهد، استفاده از کلیدهای احراز هویت یا authkey است.

کلید احراز هویت (authkey) در multiprocessing به عنوان یک لایه امنیتی عمل می‌کند که اطمینان می‌دهد که تنها پروسس‌هایی که دارای کلید مشترک هستند، می‌توانند با یکدیگر ارتباط برقرار کنند. این کلید به صورت پیش‌فرض به شکل یک رشته بایت‌های تصادفی ایجاد می‌شود، اما می‌توان آن را به صورت دستی تنظیم کرد.

close

فراخوانی متد join تضمینی برای آزاد سازی منابع مورد استفاده از process را نمی‌دهد. مثلا ممکن است pipe ها و queue هایی که در آن process استفاده می‌شدند هنوز آزاد نشده باشند.

فراخوانی متد close منابع این اطمینان را به ما می‌دهد که منابع درگیر آزاد شده و queue ها و pipe ها نیز دیگر قابل استفاده نیستند.

exitcode

این ویژگی کد خروج (exit code) پروسه را بعد از اتمام اجرای آن برمی‌گرداند. مقدار None نشان می‌دهد که پروسه هنوز در حال اجرا است. کد خروج 0 به معنای خاتمه موفقیت‌آمیز پروسه است.

ident

این ویژگی شناسه (identifier) منحصر به فرد پروسه را برمی‌گرداند. این مقدار می‌تواند برای تشخیص پروسه‌ها از یکدیگر مفید باشد.

name

این ویژگی نام پروسه را برمی‌گرداند. می‌توانید نام پروسه را به صورت دلخواه تنظیم کنید تا تشخیص پروسه‌ها آسان‌تر شود.

pid

این ویژگی شناسه پروسه (process ID) را برمی‌گرداند. این شناسه یک عدد صحیح است که به طور منحصر به فرد پروسه را در سیستم عامل شناسایی می‌کند.

run

این متد شامل کدی است که پروسه باید اجرا کند. شما معمولاً این متد را به صورت مستقیم فراخوانی نمی‌کنید، بلکه آن را در یک کلاس سفارشی که از Process ارث‌بری می‌کند، بازنویسی می‌کنید.

sentinel

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

import time
import multiprocessing
import multiprocessing.connection

def task(name, weight):
    print(f"Task {name} started")
    time.sleep(weight)
    print(f"Task {name} finished")

if __name__ == '__main__':
    p1 = multiprocessing.Process(target=task, args=('1', 1))
    p2 = multiprocessing.Process(target=task, args=('2', 2))
    p3 = multiprocessing.Process(target=task, args=('3', 3))
    
    p1.start()
    p2.start()
    p3.start()

    print('p1 sentinel is:', p1.sentinel)
    print('p2 sentinel is:', p2.sentinel)
    print('p3 sentinel is:', p3.sentinel)

    completed_processes_sentinel = multiprocessing.connection.wait([
        p1.sentinel,
        p2.sentinel,
        p3.sentinel,
    ])

    print(completed_processes_sentinel)

توضیح کد بالا:

• خط ۵ تا ۸: یک تابع تعریف شده که name و weight را به عنوان ورودی می‌گیرد. از name به عنوان نام task استفاده می‌کند تا بتوانم بگوید که چه چیزی را start و finish کرده. از weight نیز به عنوان مدت زمانی که باید به طول بیانجامد استفاده می‌کند.
• خط ۱۱ تا ۱۳: ۳ پردازش ایجاد کرده‌ایم.
• خط ۱۵ تا ۱۷: ۳ پردازشی که ایجاد کرده‌ایم را start کرده‌ایم.
• خط ۱۹ تا ۲۱: مقدار sentinel را برای هر کدام از process ها نمایش‌داده‌ایم. این مقدار متناسب با سیستم‌عامل و شرایط محیطی در هر دستگاهی ممکن است متفاوت باشد.
• خط ۲۳ تا ۲۷: صبر می‌کنیم تا ببینیم بین process های موجود کدام یک زودتر از بقیه کامل اجرا خواهد شد. completed_processes_sentinel یک لیست است که مقدارش با sentinel پردازش هایی که زودتر از بقیه اجرا شده اند پر خواهد شد.

خروجی کد بالا مشابه زیر است:

Task 1 started
p1 sentinel is: 3
p2 sentinel is: 4
p3 sentinel is: 5
Task 2 started
Task 3 started
Task 1 finished
[3]
Task 2 finished
Task 3 finished

همانطور که ملاحظه می‌کنید مقدار sentinel برای پردازش های p1 و p2 و p3 به ترتیب برابر ۳ و ۴ و ۵ است. همانطور که اشاره کردم این مقادیر را سیستم‌عامل مقداردهی می‌کند و ممکن است در دستگاه دیگری مقادیر دیگری داشته باشند.

همچنین همانطور که انتظار می‌رفت، p1 زودتر از سایرین اجرا شده است زیرا weight برای آن ۱ ثانیه تنظیم شده بود. پس لیست completed_processes_sentinel حاوی مقدار sentinel برای p1 است که برابر [3] است.

اکنون بیاید یک سناریو واقعی تر را بررسی کنیم. لطفا به مثال زیر توجه کنید.

مثال – پیدا کردن وضعیت

فرض کنید یک فایل متنی بزرگ داریم و می‌خواهیم بفهمیم که یک کلمه خاص در آن وجود دارد یا خیر. بدین منظور این فایل را باید به چند قسمت شکسته و هر قسمت را با استفاده از یک process پردازش کنیم. در صورتی که یکی از این process ها موفق به پیدا کردن شد سایر process ها را باید ببندیم. بدین منظور از متد terminate استفاده می‌کنیم. لطفا به کد زیر توجه کنید.

import time
import multiprocessing
import multiprocessing.connection

def find(name, weight):
    print(f"Task {name} started")
    time.sleep(weight)
    print(f"Task {name} finished")

if __name__ == '__main__':
    p1 = multiprocessing.Process(target=find, args=('1', 1))
    p2 = multiprocessing.Process(target=find, args=('2', 2))
    p3 = multiprocessing.Process(target=find, args=('3', 3))
    
    p1.start()
    p2.start()
    p3.start()

    print('p1 sentinel is:', p1.sentinel)
    print('p2 sentinel is:', p2.sentinel)
    print('p3 sentinel is:', p3.sentinel)

    completed_processes_sentinel = multiprocessing.connection.wait([
        p1.sentinel,
        p2.sentinel,
        p3.sentinel,
    ])
    
    if p1.sentinel in completed_processes_sentinel:
        print('p1 found the word')
    if p2.sentinel in completed_processes_sentinel:
        print('p2 found the word')
    if p3.sentinel in completed_processes_sentinel:
        print('p3 found the word')
    
    p1.terminate()
    p2.terminate()
    p3.terminate()
    
    print("Worker has finished")

توضیح کد بالا:

• خط ۲۹ تا ۳۴: در این قسمت بررسی می‌کنیم که کدام process توانسته است کلمه مورد نظر را پیدا کند. توجه کنید که در این بخش از if استفاده شده است نه از elif، دلیلش این است که ممکن است چندین process با هم در یک زمان خاص توانسته باشند که با موفقیت اجرا شوند در این صورت completed_process_sentinel می‌تواند چندین مقدار داشته باشد و همه مقادیر باید بررسی شوند.
• خط ۳۶ تا ۳۸: در این بخش همه پردازش ها را می‌بندیم تا منابع سیستم را هدر ندهیم.

terminate: این متد برای خاتمه پروسه استفاده می‌شود. این متد یک سیگنال خاتمه به پروسه ارسال می‌کند که ممکن است پروسه را به صورت تمیز (clean-up) خاتمه دهد.

استفاده از os.getpid() و os.getppid()

توابع os.getpid() و os.getppid() به ترتیب شناسه فرآیند جاری و شناسه والد (پدر) آن را ارائه می‌دهند. این اطلاعات ممکن است برای پیگیری فرآیند‌ها در داخل سیستم مفید باشد.

import os
from multiprocessing import Process

def process1():
    print(f'My process id is {os.getpid()} and my parent process id is {os.getppid()}')

p1 = Process(target=process1)

p1.start()
p1.join()

استفاده از is_alive:

متد is_alive() بر روی یک شیء فرآیند صدا زده می‌شود و بررسی می‌کند که آیا فرآیند مربوطه هنوز در حال اجرا است یا خیر. این مورد مفید است زمانی که نیاز است تا برنامه‌ی اصلی منتظر پایان یک فرآیند باشد.

from time import sleep
from multiprocessing import Process

def process1():
    print('Process 1 started')
    sleep(2)
    print('Process 1 completed')

def process2():
    print('Process 2 started')
    sleep(2)
    print('Process 2 completed')

p1 = Process(target=process1)
p2 = Process(target=process2)

p1.start()
p2.start()

while p1.is_alive() or p2.is_alive():
    print("At least one process is still running...")
    sleep(1)

print("All processes are done.")

terminate

در سیستم‌های یونیکس (مانند لینوکس و مک)، terminate() سیگنال SIGTERM را به پردازش ارسال می‌کند. این سیگنال به طور پیش‌فرض باعث خاتمه پردازش می‌شود، اما پردازش می‌تواند این سیگنال را مدیریت کند (مثلاً برای انجام عملیات پاکسازی).

در سیستم‌عامل ویندوز، terminate() پردازش را فوراً متوقف می‌کند و معادل TerminateProcess در API ویندوز است که باعث پایان بلافاصله پردازش می‌شود.

این متد مناسب زمانی است که می‌خواهید پردازشی را فوراً متوقف کنید، اما به آن اجازه دهید که اگر سیگنال‌ها را مدیریت می‌کند، بتواند عملیات پاکسازی انجام دهد.

from time import sleep
from multiprocessing import Process

def process1():
    print('Process 1 started')
    sleep(2)
    print('Process 1 completed')


p1 = Process(target=process1)

p1.start()
sleep(1)
p1.terminate()

kill

در سیستم‌های یونیکس، kill() سیگنال SIGKILL را به پردازش ارسال می‌کند. این سیگنال قابل مدیریت نیست و پردازش را بدون آزادسازی منابع در حال استفاده بلافاصله متوقف می‌کند.در ویندوز، kill() نیز مشابه terminate() عمل می‌کند و پردازش را فوراً متوقف می‌کند.

این روش معمولاً توصیه نمی‌شود زیرا می‌تواند منجر به از دست رفتن اطلاعات یا مشکلات دیگری شود.

from time import sleep
from multiprocessing import Process

def process1():
    print('Process 1 started')
    sleep(2)
    print('Process 1 completed')


p1 = Process(target=process1)

p1.start()
sleep(1)
p1.kill()

pid

ویژگی pid شناسه فرآیند را برای یک شیء فرآیند ارائه می‌دهد. این ویژگی می‌تواند مفید باشد زمانی که نیاز به شناسایی یک فرآیند در سطح سیستم عامل دارید.

import os
from time import sleep
from multiprocessing import Process

def process1():
    print('Process 1 started with pid:', os.getpid())
    sleep(2)
    print('Process 1 completed with pid:', os.getpid())

Process(target=process1).start()

exitcode

ویژگی exitcode کد خروجی فرآیند را برمی‌گرداند. این کد خروجی نشان می‌دهد که فرآیند به درستی اجرا شده و خاتمه یافته است یا خیر، و می‌تواند برای بررسی موفقیت یا شکست اجرای یک فرآیند مفید باشد.

import os
from time import sleep
from multiprocessing import Process

def process1():
    print('Process 1 started with pid:', os.getpid())
    sleep(2)
    print('Process 1 completed with pid:', os.getpid())


p1 = Process(target=process1)
p1.start()
p1.join() # اگر این خط را ننویسید، کد خارج شدن از برنامه صفر خواهد شد
print('process 1 finished with exit code:', p1.exitcode)

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

کد خروج صفر (0):

• نشان می‌دهد که پروسه به طور موفقیت‌آمیز خاتمه یافته است و هیچ خطایی رخ نداده است.

کدهای خروج مثبت (بزرگتر از صفر):

• نشان‌دهنده این است که پروسه با خطای خاصی خاتمه یافته است.
• این کدها معمولاً بیانگر نوع خاصی از خطا یا وضعیت خروج هستند که در برنامه خود می‌توانید تعریف کنید.
• مثال: اگر در یک برنامه خطای خاصی را با کد خروج 1 نمایش دهید، exitcode برابر 1 خواهد بود.

کدهای خروج منفی (کمتر از صفر):

• نشان می‌دهد که پروسه با سیگنال خاصی خاتمه یافته است.
• این کدها معمولاً نشان‌دهنده سیگنال‌هایی هستند که به پروسه ارسال شده‌اند تا آن را خاتمه دهند.
• مثال:
  • کد -9: پروسه با سیگنال SIGKILL خاتمه یافته است.
  • کد -15: پروسه با سیگنال SIGTERM خاتمه یافته است.

در کد زیر، پروسه‌ای ایجاد می‌شود که با موفقیت خاتمه می‌یابد (کد خروج صفر):

import os
from time import sleep
from multiprocessing import Process

def process1():
    print('Process 1 started with pid:', os.getpid())
    sleep(2)
    print('Process 1 completed with pid:', os.getpid())

p1 = Process(target=process1)
p1.start()
p1.join()
print('process 1 finished with exit code:', p1.exitcode)  # Output: 0

در کد زیر، پروسه‌ای ایجاد می‌شود که با استفاده از یک سیگنال خاتمه می‌یابد (کد خروج منفی):

import os
from time import sleep
from multiprocessing import Process

def process1():
    print('Process 1 started with pid:', os.getpid())
    sleep(10)
    print('Process 1 completed with pid:', os.getpid())

p1 = Process(target=process1)
p1.start()

# بعد از 2 ثانیه پروسه را خاتمه می‌دهیم
sleep(2)
p1.terminate()
p1.join()
print('process 1 finished with exit code:', p1.exitcode)  # Output: -15 (if terminated with SIGTERM)

جمع‌بندی

• 0: پروسه به طور موفقیت‌آمیز خاتمه یافته است.
• کدهای مثبت: پروسه با خطای خاصی خاتمه یافته است.
• کدهای منفی: پروسه با سیگنال خاصی خاتمه یافته است.

daemon

مشخص می‌کند که یک process باید پر پس‌زمینه اجرا شود یا خیر

import os
from time import sleep
from multiprocessing import Process

def process1():
    print('Process 1 started')
    sleep(2)
    print('Process 1 completed')

p1 = Process(target=process1, daemon=True)
p1.start()

cpu_count

در دست نگارش …

تفاوت is_alive و sentinel چیست؟

• نوع بررسی:
  • is_alive به صورت فوری و همزمان بررسی می‌کند که آیا پروسه هنوز در حال اجرا است یا خیر.
  • sentinel برای شناسایی پایان یافتن پروسه به صورت ناهمگام و در مکانیزم‌های انتخاب و حلقه‌های رویداد استفاده می‌شود.
• نحوه استفاده:
  • is_alive یک متد است که باید فراخوانی شود و نتیجه بولین برمی‌گرداند.
  • sentinel یک آبجکت فایل‌مانند است که می‌تواند در مکانیزم‌های ناهمگام استفاده شود تا پایان یافتن پروسه را شناسایی کند.
• کاربردها:
  • is_alive برای بررسی سریع و همزمان وضعیت پروسه‌ها مناسب است.
  • sentinel برای برنامه‌های پیچیده‌تر که نیاز به هماهنگی و بررسی پایان یافتن پروسه‌ها به صورت ناهمگام دارند، مفید است.

جمع‌بندی

• از is_alive زمانی استفاده کنید که نیاز به بررسی فوری و همزمان وضعیت پروسه دارید.
• از sentinel زمانی استفاده کنید که نیاز به شناسایی پایان یافتن پروسه به صورت ناهمگام و در مکانیزم‌های انتخاب دارید.

تفاوت ident و pid چیست؟

در استفاده از ماژول multiprocessing در پایتون، موارد استفاده‌ی ident و pid متفاوت است و بستگی به نیاز شما دارد. در زیر توضیح می‌دهم که در کجاها ident و در کجاها pid بیشتر مناسب هستند:

موارد مناسب برای استفاده از ident

1. تشخیص پروسه‌ها در سطح پایتون:
   • زمانی که نیاز دارید پروسه‌ها را در کد پایتون خود شناسایی و مدیریت کنید، ident بیشتر مناسب است. این شناسه به طور داخلی توسط multiprocessing استفاده می‌شود و برای شناسایی پروسه‌ها در سطح کد پایتون مفید است.
   • مثال: در یک برنامه چند نخی (multithreaded)، ممکن است از ident برای تشخیص و مدیریت نخ‌های مختلف استفاده کنید.
2. استفاده در کتابخانه‌های پایتون:
   • اگر از کتابخانه‌هایی استفاده می‌کنید که نیاز به شناسایی پروسه‌ها یا نخ‌ها دارند، احتمالاً از ident استفاده خواهید کرد. بسیاری از کتابخانه‌ها و ابزارهای پایتون از ident برای مدیریت و شناسایی پروسه‌ها استفاده می‌کنند.

موارد مناسب برای استفاده از pid

1. مدیریت پروسه‌ها در سطح سیستم عامل:
   • اگر نیاز دارید پروسه‌ها را در سطح سیستم عامل مدیریت کنید، مانند ارسال سیگنال‌ها به پروسه‌ها، مانیتورینگ پروسه‌ها، یا خاتمه دادن به پروسه‌ها، pid مناسب‌تر است. pid به طور جهانی در سیستم عامل یکتا است و به شما امکان می‌دهد با پروسه‌ها در سطح سیستم عامل تعامل داشته باشید.
   • مثال: اگر نیاز دارید پروسه‌ای را خاتمه دهید یا وضعیت پروسه‌ای را مانیتور کنید، از pid استفاده می‌کنید.
2. دیباگینگ و مانیتورینگ پروسه‌ها:
   • در مواقعی که نیاز دارید پروسه‌ها را دیباگ کنید یا مانیتورینگ انجام دهید، pid بسیار مفید است. ابزارهای مانیتورینگ سیستم مانند ps، top و htop از pid برای شناسایی پروسه‌ها استفاده می‌کنند.
   • مثال: برای بررسی مصرف منابع پروسه‌ها و عملکرد آنها، از pid در ابزارهای مانیتورینگ سیستم استفاده می‌کنید.

جمع‌بندی

• از ident زمانی استفاده کنید که نیاز دارید پروسه‌ها را در سطح کد پایتون شناسایی و مدیریت کنید.
• از pid زمانی استفاده کنید که نیاز دارید پروسه‌ها را در سطح سیستم عامل مدیریت، دیباگ یا مانیتور کنید.

تمرین ها

تمرین اول – تکه کد زیر حاوی دو تابع است. یکی فایل را فشرده می‌کند و دیگری یک تایمر نمایش می‌دهد. کد زیر را به نحوی تغییر دهید که تایمر در زمان فشرده سازی اجرا شود تا بتوانیم بفهمیم که فشرده سازی چند ثانیه طول کشیده است.

import time
from time import sleep
from multiprocessing import Process

def compress_file():
    sleep(5)

def show_timer():
    start_time = time.time()
    while True:
        sleep(1)

تمرین دوم – تکه کد زیر حاوی یک تابع به نام do_something است که بر اساس ورودی ای که میگیرد مدت زمان اجرایش تغییر می‌کند. تابع terminate_after_2_seconds را به نحوی توسعه دهید که پس از دو ثانیه تکلیف را مشخص کند. این تابع پس از دو ثانیه از اجرای process ها باید اجرا شود و process هایی که هنوز باز هستند را باید ببندد.

import time
from time import sleep
from multiprocessing import Process

def do_something(sleep_time):
    print('starting process with sleep time', sleep_time)
    time.sleep(sleep_time)
    print('ending process with sleep time', sleep_time)

def terminate_all_after_2_seconds(process_list):
    pass

process_list = [
    Process(target=do_something, args=(1,)),
    Process(target=do_something, args=(2,)),
    Process(target=do_something, args=(3,)),
    Process(target=do_something, args=(4,)),
]

for process in process_list:
    process.start()