پردازش موازی در پایتون

معرفی

وقتی یک برنامه را راه اندازی می کنید روی دستگاه شما در “حباب” خود اجرا می شود که کاملاً جدا از سایر برنامه هایی است که همزمان فعال هستند. این “حباب” a نامیده می شود processو شامل همه چیزهایی است که برای مدیریت این تماس برنامه نیاز است.

به عنوان مثال، این به اصطلاح process محیط شامل صفحات حافظه را process در حال استفاده است، فایل این را مدیریت می کند process حقوق دسترسی کاربر و گروه و کل فراخوانی خط فرمان آن، از جمله پارامترهای داده شده را باز کرده است.

این اطلاعات در process سیستم فایل سیستم یونیکس/لینوکس شما که یک سیستم فایل مجازی است و از طریق آن قابل دسترسی است /proc فهرست راهنما. مدخل ها بر اساس مرتب شده اند process شناسه، که برای هر کدام منحصر به فرد است process. مثال 1 این را برای یک انتخاب دلخواه نشان می دهد process که دارد process شناسه شماره 177.

مثال 1: اطلاعاتی که در دسترس الف است process

root@system:/proc/177
attr         cpuset   limits      net            projid_map   statm
autogroup    cwd      loginuid    ns             root         status
auxv         environ  map_files   numa_maps      sched        syscall
cgroup       exe      maps        oom_adj        sessionid    task
clear_refs   fd       mem         oom_score      setgroups    timers
cmdline      fdinfo   mountinfo   oom_score_adj  smaps        uid_map
comm         gid_map  mounts      pagemap        stack        wchan
coredump_filter       io          mountstats     personality  stat

ساختار کد برنامه و داده ها

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

• یک مجرد process یک قطعه کد را پوشش می دهد که می تواند به طور جداگانه اجرا شود
• بخش های خاصی از کد را می توان به طور همزمان اجرا کرد و در اصل اجازه موازی سازی را می دهد
• استفاده از ویژگی‌های پردازنده‌ها و سیستم‌عامل‌های مدرن، به‌عنوان مثال، تمام هسته‌های پردازنده‌ای که در دسترس داریم برای کاهش کل زمان اجرای یک برنامه.
• برای کاهش پیچیدگی برنامه/کد خود و برون سپاری قطعات کار به عوامل تخصصی که به عنوان فرآیندهای فرعی عمل می کنند.

استفاده از فرآیندهای فرعی مستلزم این است که در روش اجرای برنامه خود از خطی به موازی تجدید نظر کنید. این شبیه تغییر دیدگاه کاری شما در یک شرکت از یک کارگر معمولی به یک مدیر است – شما باید مراقب باشید روی چه کسی چه کاری انجام می دهد، یک قدم چقدر طول می کشد، و چه وابستگی هایی بین نتایج میانی وجود دارد.

این به شما کمک می کند کد خود را به تکه های کوچکتر تقسیم کنید که می تواند توسط یک عامل فقط برای این کار اجرا شود. اگر هنوز این کار را انجام نداده اید، به این فکر کنید که مجموعه داده شما چگونه ساختار یافته است تا بتواند به طور موثر توسط عوامل فردی پردازش شود. این منجر به این سؤالات می شود:

• چرا می خواهید کد را موازی کنید؟ در مورد خاص شما و از نظر تلاش، آیا فکر کردن به آن منطقی است؟
• آیا برنامه شما قرار است فقط یک بار اجرا شود یا به طور منظم اجرا خواهد شد روی یک مجموعه داده مشابه؟
• آیا می توانید الگوریتم خود را به چند مرحله اجرا تقسیم کنید؟
• آیا داده های شما اصلاً اجازه موازی سازی را می دهد؟ اگر هنوز نه، سازماندهی داده های شما باید به چه شکلی تطبیق داده شود؟
• کدام نتایج میانی محاسبات شما بستگی دارد روی یکدیگر؟
• کدام تغییر در سخت افزار برای آن مورد نیاز است؟
• آیا در سخت افزار یا الگوریتم تنگنا وجود دارد و چگونه می توان از تأثیر این عوامل جلوگیری کرد یا آن را به حداقل رساند؟
• کدام عوارض جانبی دیگر موازی سازی ممکن است رخ دهد؟

یک مورد استفاده احتمالی یک مورد اصلی است processو دیمونی که در پس‌زمینه اجرا می‌شود (master/slave) که منتظر فعال شدن است. همچنین، این می تواند یک اصلی باشد process که فرآیندهای کارگری را شروع می کند روی تقاضا در عمل، اصلی process تغذیه کننده است process که دو یا چند عامل را کنترل می کند که بخش هایی از داده ها را تغذیه می کنند و محاسبات را انجام می دهند روی بخش داده شده

به خاطر داشته باشید که موازی سازی به دلیل سربار فرآیندهای فرعی که سیستم عامل شما به آن نیاز دارد، هم پرهزینه و هم زمان بر است. در مقایسه با اجرای دو یا چند کار به صورت خطی، انجام این کار به صورت موازی ممکن است بین 25 تا 30 درصد در زمان هر فرعی صرفه جویی کنید. روی مورد استفاده شما به عنوان مثال، دو کار که هر کدام 5 ثانیه زمان می برند، اگر به صورت سری اجرا شوند، در مجموع به 10 ثانیه نیاز دارند و ممکن است حدود 8 ثانیه زمان نیاز داشته باشند. روی میانگین روی یک ماشین چند هسته ای در صورت موازی شدن. ممکن است 3 ثانیه از آن 8 ثانیه در سربار گم شود و بهبود سرعت شما را محدود کند.

اجرای یک تابع به صورت موازی با پایتون

پایتون چهار راه ممکن برای مدیریت آن ارائه می دهد. ابتدا می توانید توابع را به صورت موازی با استفاده از چند پردازشی مدول. دوم، یک جایگزین برای فرآیندها، موضوعات هستند. از نظر فنی، اینها فرآیندهای سبک وزن هستند و خارج از محدوده این مقاله هستند. برای مطالعه بیشتر می توانید به پایتون نگاهی بیندازید ماژول threading. سوم، می توانید با استفاده از برنامه های خارجی تماس بگیرید system() روش از os ماژول، یا روش های ارائه شده توسط subprocess ماژول، و سپس نتایج را جمع آوری کنید.

را multiprocessing ماژول مجموعه خوبی از روش ها را برای مدیریت موازی اجرای روتین ها پوشش می دهد. این شامل فرآیندها، مجموعه‌ای از عوامل، صف‌ها و لوله‌ها می‌شود.

لیست 1 با مجموعه ای از پنج عامل کار می کند که process یک تکه از سه مقدار به طور همزمان. مقادیر برای تعداد عوامل، و برای chunksize خودسرانه برای اهداف نمایشی انتخاب می شوند. این مقادیر را بر اساس تعداد هسته های پردازنده خود تنظیم کنید.

روش Pool.map() به سه پارامتر نیاز دارد – یک تابع باید فراخوانی شود روی هر عنصر از مجموعه داده، خود مجموعه داده، و chunksize. که در لیست 1 ما از تابعی استفاده می کنیم که نام دارد square و مربع مقدار صحیح داده شده را محاسبه می کند. علاوه بر این، chunksize را می توان حذف کرد. در صورت عدم تنظیم صریح پیش فرض chunksize 1 است.

لطفا توجه داشته باشید که دستور اجرای نمایندگان تضمین نمی شود، اما مجموعه نتایج به ترتیب درست است. این شامل مقادیر مربع با توجه به ترتیب عناصر مجموعه داده اصلی است.

فهرست 1: اجرای توابع به صورت موازی

from multiprocessing import Pool

def square(x):
    
    return x*x

if __name__ == '__main__':

    
    dataset = (1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14)

    
    print ('Dataset: ' + str(dataset))

    
    agents = 5
    chunksize = 3
    with Pool(processes=agents) as pool:
        result = pool.map(square, dataset, chunksize)

    
    print ('Result:  ' + str(result))

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

$ python3 pool_multiprocessing.py 
Dataset: (1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14)
Result:  (1, 4, 9, 16, 25, 36, 49, 64, 81, 100, 121, 144, 169, 196)

توجه داشته باشید: ما از Python 3 برای این مثال ها استفاده خواهیم کرد.

اجرای چندین تابع با استفاده از صف

به عنوان یک ساختار داده، صف بسیار متداول است و به طرق مختلف وجود دارد. آن را به عنوان هر دو سازماندهی شده است اولین ورودی اولین خروجی (FIFO) یا Last In First Out (LIFO)/پشتهو همچنین با و بدون اولویت (صف اولویت). ساختار داده به صورت یک آرایه با تعداد ورودی ثابت یا به صورت فهرستی که تعداد متغیری از عناصر منفرد را در خود جای داده است پیاده سازی می شود.

که در لیست های 2.1-2.7 ما از یک صف FIFO استفاده می کنیم. این به صورت لیستی پیاده سازی می شود که قبلاً توسط کلاس مربوطه از آن ارائه شده است multiprocessing مدول. علاوه بر این، time ماژول بارگذاری می شود و برای تقلید از بار کاری استفاده می شود.

فهرست 2.1: ماژول های مورد استفاده

import multiprocessing
from time import sleep

سپس یک تابع کارگر تعریف می شود (لیست 2.2). این تابع در واقع عامل را نشان می دهد و به سه آرگومان نیاز دارد. را process نام نشان می دهد که کدام process آن است، و هر دو tasks و results به صف مربوطه مراجعه کنید.

در داخل تابع کارگر یک نامتناهی است while حلقه هر دو tasks و results صف هایی هستند که در برنامه اصلی تعریف می شوند. tasks.get() وظیفه فعلی را از صف وظیفه برای پردازش برمی گرداند. مقدار کار کوچکتر از 0 از آن خارج می شود while حلقه، و مقدار -1 را برمی گرداند. هر مقدار وظیفه دیگری یک محاسبه (مربع) را انجام می دهد و این مقدار را برمی گرداند. برگرداندن یک مقدار به برنامه اصلی به صورت پیاده سازی می شود results.put(). این مقدار محاسبه شده را در انتهای مقدار اضافه می کند results صف

فهرست 2.2: تابع کارگر

def calculate(process_name, tasks, results):
    print('(%s) evaluation routine starts' % process_name)

    while True:
        new_value = tasks.get()
        if new_value < 0:
            print('(%s) evaluation routine quits' % process_name)

            
            results.put(-1)
            break
        else:
            
            compute = new_value * new_value
            sleep(0.02*new_value)

            
            
            print('(%s) received value: %i' % (process_name, new_value))
            print('(%s) calculated value: %i' % (process_name, compute))

            
            results.put(compute)

    return

مرحله بعدی حلقه اصلی است (نگاه کنید به لیست 2.3). اول، یک مدیر برای بینprocess ارتباط (IPC) تعریف شده است. بعد، دو صف اضافه می شود – یکی که وظایف را نگه می دارد، و دیگری برای نتایج.

لیست 2.3: IPC و صف ها

if __name__ == "__main__":
    
    manager = multiprocessing.Manager()

    
    tasks = manager.Queue()
    results = manager.Queue()

پس از انجام این تنظیمات، a را تعریف می کنیم process استخر با چهار فرآیند کارگری (عامل). ما از کلاس استفاده می کنیم multiprocessing.Pool()، و یک نمونه از آن ایجاد کنید. در مرحله بعد، ما یک لیست خالی از فرآیندها را تعریف می کنیم (نگاه کنید به لیست 2.4).

فهرست 2.4: تعریف الف process استخر

num_processes = 4
pool = multiprocessing.Pool(processes=num_processes)
processes = ()

در مرحله زیر، چهار فرآیند کارگر (عامل) را آغاز می کنیم. برای سادگی، آنها از “P0” تا “P3” نامگذاری شده اند. ایجاد چهار فرآیند کارگر با استفاده از آن انجام می شود multiprocessing.Process(). این امر هر یک از آنها را به تابع worker و همچنین وظیفه و صف نتایج متصل می کند. در نهایت، مقدار تازه اولیه را اضافه می کنیم process در انتهای لیست فرآیندها، و جدید را شروع کنید process استفاده کردن new_process.start() (دیدن لیست 2.5).

فهرست 2.5: فرآیندهای کارگر را آماده کنید

for i in range(num_processes):

    
    process_name = 'P%i' % i

    
    new_process = multiprocessing.Process(target=calculate, args=(process_name,tasks,results))

    
    processes.append(new_process)

    
    new_process.start()

فرآیندهای کارگری ما منتظر کار هستند. ما لیستی از وظایف را تعریف می کنیم که در مورد ما به صورت دلخواه اعداد صحیح انتخاب شده اند. با استفاده از این مقادیر به لیست وظایف اضافه می شود tasks.put(). هر کارگر process منتظر وظایف می ماند و کار موجود بعدی را از لیست وظایف انتخاب می کند. این توسط خود صف اداره می شود (نگاه کنید به لیست 2.6).

لیست 2.6: صف کار را آماده کنید

task_list = (43, 1, 780, 256, 142, 68, 183, 334, 325, 3)
for single_task in task_list:
    tasks.put(single_task)


sleep(5)

بعد از مدتی دوست داریم که کارگزاران ما کار را تمام کنند. هر کارگر process واکنش نشان می دهد روی یک کار با مقدار -1. این مقدار را به عنوان سیگنال پایان تفسیر می کند و پس از آن می میرد. به همین دلیل است که به تعداد -1 در صف وظایف به تعداد پردازش هایی که در حال اجرا هستیم قرار می دهیم. قبل از مرگ، الف process که به پایان می رسد یک -1 را در صف نتایج قرار می دهد. این به معنای یک سیگنال تایید به حلقه اصلی است که عامل در حال خاتمه است.

در حلقه اصلی از آن صف می خوانیم و عدد -1 را می شماریم. حلقه اصلی به محض اینکه به تعداد پروسه های تایید خاتمه شمارش کردیم، از بین می رود. در غیر این صورت، نتیجه محاسبه را از صف خروجی می گیریم.

فهرست 2.7: خاتمه و خروجی نتایج

for i in range(num_processes):
    tasks.put(-1)


num_finished_processes = 0
while True:
    
    new_result = results.get()

    
    if new_result == -1:
        
        num_finished_processes += 1

        if num_finished_processes == num_processes:
            break
    else:
        
        print('Result:' + str(new_result))

مثال 2 خروجی برنامه پایتون را نمایش می دهد. با اجرای بیش از یک بار برنامه، ممکن است متوجه شوید که ترتیب شروع فرآیندهای کارگر به همان اندازه غیر قابل پیش بینی است. process خودش که وظیفه ای را از صف انتخاب می کند. با این حال، پس از اتمام ترتیب عناصر صف نتایج با ترتیب عناصر صف وظایف مطابقت دارد.

مثال 2

$ python3 queue_multiprocessing.py 
(P0) evaluation routine starts
(P1) evaluation routine starts
(P2) evaluation routine starts
(P3) evaluation routine starts
(P1) received value: 1
(P1) calculated value: 1
(P0) received value: 43
(P0) calculated value: 1849
(P0) received value: 68
(P0) calculated value: 4624
(P1) received value: 142
(P1) calculated value: 20164
result: 1
result: 1849
result: 4624
result: 20164
(P3) received value: 256
(P3) calculated value: 65536
result: 65536
(P0) received value: 183
(P0) calculated value: 33489
result: 33489
(P0) received value: 3
(P0) calculated value: 9
result: 9
(P0) evaluation routine quits
(P1) received value: 334
(P1) calculated value: 111556
result: 111556
(P1) evaluation routine quits
(P3) received value: 325
(P3) calculated value: 105625
result: 105625
(P3) evaluation routine quits
(P2) received value: 780
(P2) calculated value: 608400
result: 608400
(P2) evaluation routine quits

توجه داشته باشید: همانطور که قبلا ذکر شد، ممکن است خروجی شما دقیقاً با آنچه در بالا نشان داده شده مطابقت نداشته باشد زیرا ترتیب اجرا غیرقابل پیش بینی است.

با استفاده از روش os.system()

را system() روش بخشی از ماژول OS، که اجازه می دهد تا برنامه های خط فرمان خارجی را به صورت جداگانه اجرا کنید process از برنامه پایتون شما را system() روش یک تماس مسدود کننده است و باید منتظر بمانید تا تماس تمام شود و برگردد. به عنوان یک فتیشیست یونیکس/لینوکس می دانید که یک دستور می تواند در پس زمینه اجرا شود و نتیجه محاسبه شده را در جریان خروجی که به فایلی مانند این هدایت می شود بنویسید (نگاه کنید به مثال 3):

مثال 3: فرمان با تغییر مسیر خروجی

$ ./program >> outputfile &

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

لیست 3: تماس سیستمی ساده با استفاده از ماژول OS

import os

os.system("./program >> outputfile &")

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

استفاده از این روش بسیار گرانتر از روش های قبلی است که توضیح دادم. اول اینکه سربار بسیار بزرگتر است (process سوئیچ) و دوم اینکه داده ها را روی حافظه فیزیکی مانند دیسک می نویسد که بیشتر طول می کشد. اگرچه، اگر حافظه محدودی دارید (مانند RAM) این گزینه بهتر است و در عوض می توانید داده های خروجی عظیمی را روی یک دیسک حالت جامد نوشته باشید.

با استفاده از ماژول زیر فرآیند

این ماژول برای جایگزینی در نظر گرفته شده است os.system() و os.spawn() تماس می گیرد. ایده از فرآیند فرعی ساده سازی فرآیندهای تخم ریزی، برقراری ارتباط با آنها از طریق لوله ها و سیگنال ها، و جمع آوری خروجی تولید شده از جمله پیام های خطا است.

با شروع پایتون 3.5، فرآیند فرعی حاوی متد است subprocess.run() برای شروع یک فرمان خارجی، که پوششی برای زیرین است subprocess.Popen() کلاس به عنوان مثال دستور UNIX/Linux را اجرا می کنیم df -h برای اینکه بفهمید هنوز چقدر فضای دیسک در دسترس است روی را /home پارتیشن دستگاه شما در یک برنامه پایتون این فراخوانی را مطابق شکل زیر انجام می دهید (لیست 4).

لیست 4: مثال اصلی برای اجرای یک دستور خارجی

import subprocess

ret = subprocess.run(("df", "-h", "/home"))
print(ret)

این فراخوانی اصلی است و بسیار شبیه به دستور است df -h /home در حال اعدام در الف terminal. توجه داشته باشید که پارامترها به جای یک رشته به صورت یک لیست از هم جدا می شوند. خروجی مشابه خواهد بود مثال 4. در مقایسه با اسناد رسمی پایتون برای این ماژول، نتیجه فراخوانی را به خروجی می دهد stdout، علاوه بر مقدار برگشتی تماس.

مثال 4 خروجی تماس ما را نشان می دهد. آخرین خط خروجی اجرای موفقیت آمیز دستور را نشان می دهد. صدا زدن subprocess.run() نمونه ای از کلاس را برمی گرداند CompletedProcess که دارای دو صفت نامگذاری شده است args (آگومان های خط فرمان)، و returncode (مقدار برگشتی دستور).

مثال 4: اجرای اسکریپت پایتون از لیست 4

$ python3 diskfree.py
Filesystem   Size   Used  Avail Capacity  iused   ifree %iused  Mounted روی
/dev/sda3  233Gi  203Gi   30Gi    88% 53160407 7818407   87%   /home
CompletedProcess(args=('df', '-h', '/home'), returncode=0)

برای سرکوب خروجی به stdoutو هم خروجی و هم مقدار بازگشتی را برای ارزیابی بیشتر، فراخوانی بگیرید subprocess.run() باید کمی اصلاح شود بدون تغییر بیشتر، subprocess.run() خروجی فرمان اجرا شده را به stdout که کانال خروجی پایتون زیرین است process. برای گرفتن خروجی، باید این را تغییر دهیم و کانال خروجی را روی مقدار از پیش تعریف شده تنظیم کنیم subprocess.PIPE. لیست 5 روش انجام آن را نشان می دهد.

فهرست 5: گرفتن خروجی در یک لوله

import subprocess


output = subprocess.run(("df", "-h", "/home"), stdout=subprocess.PIPE)


print ("Return code: %i" % output.returncode)
print ("Output data: %s" % output.stdout)

همانطور که قبلا توضیح داده شد subprocess.run() نمونه ای از کلاس را برمی گرداند CompletedProcess. که در لیست 5، این نمونه یک متغیر با نام ساده است output. کد برگشتی دستور در ویژگی نگهداری می شود output.returncode، و خروجی چاپ شده در stdout را می توان در صفت یافت output.stdout. به خاطر داشته باشید که این پیام های خطای مدیریت را پوشش نمی دهد زیرا ما کانال خروجی را برای آن تغییر نداده ایم.

نتیجه

پردازش موازی یک فرصت عالی برای استفاده از قدرت سخت افزارهای معاصر است. پایتون به شما امکان دسترسی به این روش ها را در سطح بسیار پیچیده ای می دهد. همانطور که قبلا هر دو دیده اید multiprocessing و subprocess ماژول به شما امکان می دهد به راحتی در آن موضوع شیرجه بزنید.

سپاسگزاریها

نویسنده مایل است تشکر کند جرولد روپرشت برای حمایت او و منتقدان در تهیه این مقاله.