Paralel Programlama

Paralel programlama, bir problemi birden fazla bilgisayar veya işlemci kullanarak daha kısa zamanda hesaplamayı hedefler. Paralel programlama ya da paralelleştirme sayesinde, donanımın ya da işletim sisteminin yetersizliği nedeniyle sıralı yöntemlerle hesaplanamayan problemler, paralel bilgisayarlar ve çoklu-threadler sayesinde kabul edilebilir zamanda veya gerçek zamanda hesaplanabilmektedir. DNA modellemesi, hava tahmini vs. gibi yoğun hesaplama gerektiren problemler, paralel programlama sayesinde kabul edilebilir zamanlarda hesaplanabilirler. Diğer yandan, değişik alanlardaki birçok araştırmacı hesaplamalarda yüksek hıza ihtiyaç duymaktadır.

Paralel Bilgisayar Sistemleri

Paralel bilgisayar sistemleri, paylaşımlı ve dağıtık bellekli sistemler olmak üzere 2 ana gruba ayrılabilir. Paylaşımlı bellekli sistemlerde işlemciler ortak bir belleğe erişirler. Aynı bellek bölgesine eş zamanlı erişimleri engellemek açısından senkronizasyon büyük önem taşır. Dağıtık bellekli sistemlerde ise her bir işlemci kendi belleğine sahiptir ve bir işlemci başka bir işlemcinin belleğindeki veriyi mesaj geçme (message passing) ile elde eder. Her iki sistemin de birbirlerine göre avantajları ve dezavantajları mevcuttur. Kısaca, paylaşımlı bellekli sistemler kolay programlanabilir, dağıtık bellekli bilgisayarlar ise ölçeklenebilir paralel bilgisayarlardır.

Python ile Paralel programlama

Paralel programlama, büyük sayısal sorunları daha küçük alt görevlere bölerek çözer ve böylece multi-processor ve/veya multi-corei makinelerde toplam hesaplama süresini azaltır. Paralel programlama, C ve FORTRAN gibi “heavy-duty” hesaplama işleri için uygun olan geleneksel programlama dillerinde desteklenmektedir. Geleneksel olarak, Python’un kısmen global interpreter lock (GIL) nedeniyle paralel programlamayı çok iyi desteklemediği düşünülmektedir. Ancak zengin çeşitlilikte kütüphane ve paketlerin geliştirilmesi sayesinde Python’da paralel programlama desteği eskiye kıyasla artık çok daha iyi. Paralel programlama için MPI4PY, pyMPI, multiprocessing, threading, parallel gibi kütüphaneler geliştirilmiştir.

Yazının geri kalanında Pi sayısının hesaplanması örneği ile sıralı ve paralel (multi-threading, multi-processing) yaklaşımları verilecektir.

π Sayısı (3.141592653589793238462643383279502884197169399375105820974…)

Pi sayısının hesaplanması için farklı metotlar geliştirilmiştir. Bu yazıda $\inline \fn_cm \LARGE \pi = \int_{0}^{1}\frac{4}{1+x^2}dx$ formülünün seri açılımı olan $\inline \fn_cm \LARGE \sum_{i=1}^{interval}\left ( \frac{4}{((i-0.5)*width)^2} \right )*width$ kullanılacaktır. Bu formül şu şekilde de düşünülebilir;

Yanda ki şekildeki gibi bir dairenin bir karenin içerisine yerleştirilmesi sayesinde karenin alanının dairenin alanına oranından π sayısı çekilerek hesaplanabilir. Bu mantığa göre yapılması gereken şekilden rastgele noktaları çekip bu noktaların dairenin içinde ve dışında olması durumlarına göre M ve N alanlarını artırıp oranlayarak bulabiliriz. Ne kadar fazla nokta seçersek o kadar π sayısının gerçek değerine yaklaşırız. Fakat fazla nokta seçmek ve bu noktanın dairenin içinde olup olmadığını sorgulamak on binlerce noktanın hesabı düşünüldüğünde verimli değildir. Bu verimsiz yaklaşımı paralel programlama yaklaşımlarıyla ile hızlandırabiliriz. Altta ki kod [(-1, 1), (-1, 1)] koordinatlarında verilmiş daire ve kareye göre π sayısını bulmaktadır.

import random
def calc_pi(N):
  M = 0
  for i in range(N):
    # Simulate impact coordinates
    x = random.uniform(-1, 1)
    y = random.uniform(-1, 1)
 
    # True if impact happens inside the circle 
    if x**2 + y**2 < 1.0:
      M += 1
  return 4 * M / N

import random

def calc_pi(N):

M = 0

for i in range(N):

# Simulate impact coordinates

x = random.uniform(-1, 1)

y = random.uniform(-1, 1)

# True if impact happens inside the circle

if x**2 + y**2 < 1.0:

M += 1

return 4 * M / N

Şimdi bu örneği farklı paralel formlarda verip çalışma zamanlarını kıyaslayalım.

import time

def timing(f):
    def wrap(*args):
        time1 = time.time()
        ret = f(*args)
        time2 = time.time()
        print('{:s} function took {:.3f} ms'.format(f.__name__, (time2 - time1) * 1000.0))

        return ret

    return wrap


@timing
def calculate_Pi():
    nsteps = 100000000
    dx = 1.0 / nsteps
    pi = 0.0
    for i in range(nsteps):
        x = (i + 0.5) * dx
        pi += 4.0 / (1.0 + x * x)
    pi *= dx
    print("Pi Sayısı: ", pi)


calculate_Pi()

import time

def timing(f):

def wrap(*args):

time1 = time.time()

ret = f(*args)

time2 = time.time()

print('{:s} function took {:.3f} ms'.format(f.__name__, (time2 - time1) * 1000.0))

return ret

return wrap

@timing

def calculate_Pi():

nsteps = 100000000

dx = 1.0 / nsteps

pi = 0.0

for i in range(nsteps):

x = (i + 0.5) * dx

pi += 4.0 / (1.0 + x * x)

pi *= dx

print("Pi Sayısı: ", pi)

calculate_Pi()

Üste deki sıralı kodun çıktısı şu şekildedir:

Pi Sayısı: 3.1415926535904264
calculate_Pi function took 12288.980 ms

Şimdi multi-threading örneğine ve sonucuna bakalım.

import multiprocessing as mp
from concurrent.futures import ThreadPoolExecutor
from functools import wraps
import time

_DEFAULT_POOL = ThreadPoolExecutor()

def threadpool(f, executor=None):
    @wraps(f)
    def wrap(*args, **kwargs):
        return (executor or _DEFAULT_POOL).submit(f, *args, **kwargs)

    return wrap

@threadpool
def calc_partial_pi(rank):
    partial_pi = 0.0
    nsteps = 100000000
    dx = 1.0 / nsteps
    nprocs = mp.cpu_count()

    for i in range(rank, nsteps, nprocs):
        x = (i + 0.5) * dx
        partial_pi += 4.0 / (1.0 + x * x)
    partial_pi *= dx
    return partial_pi

pi = 0

time1 = time.time()
for rank in range(mp.cpu_count()):
    a = calc_partial_pi(rank)
    pi = pi + a.result()

time2 = time.time()

print("Pi Sayısı: ", pi)
print('{:.3f} ms'.format((time2 - time1) * 1000.0))

import multiprocessing as mp

from concurrent.futures import ThreadPoolExecutor

from functools import wraps

import time

_DEFAULT_POOL = ThreadPoolExecutor()

def threadpool(f, executor=None):

@wraps(f)

def wrap(*args, **kwargs):

return (executor or _DEFAULT_POOL).submit(f, *args, **kwargs)

return wrap

@threadpool

def calc_partial_pi(rank):

partial_pi = 0.0

nsteps = 100000000

dx = 1.0 / nsteps

nprocs = mp.cpu_count()

for i in range(rank, nsteps, nprocs):

x = (i + 0.5) * dx

partial_pi += 4.0 / (1.0 + x * x)

partial_pi *= dx

return partial_pi

pi = 0

time1 = time.time()

for rank in range(mp.cpu_count()):

a = calc_partial_pi(rank)

pi = pi + a.result()

time2 = time.time()

print("Pi Sayısı: ", pi)

print('{:.3f} ms'.format((time2 - time1) * 1000.0))

multi-threading için sonuç

Pi Sayısı: 3.14159265358974
15569.067 ms

Görüldüğü gibi bir iyileşme olmasına rağmen yeteri kadar iyi bir performans olmadı. Programın dallandırılması gibi durumlar gerektiğinde tercih edilebilir. Ayrıca threading için bir çok farklı yol olmasına rağmen ben threadpool yaklaşımını ve bu kod düzenini sevdim. Pratik olarak threadpool fonksiyonunu koda dahil edip kullanabilirsiniz. Şimdi çok işlemcili (multiprocessing) yaklaşım örneklerine bakalım. Bu yazıda iki yaklaşım olan starmap ve async ile ilgili örnek verdim. starmap alternatifi olarak ‘map’ fonksiyonu da kullanılmaktadır fakat map de çoklu parametre verilmesi olmadığı için starmap’i tercih ettim. starmap ve async arasında ki fark ise starmap de tüm prcosesslerin eş zamanlı çalışması ve async de ise processlerin çalışma sırasının olmamasıdır. Performansları aşağı yukarı bu örnek için aynı çıktı. Şimdi kodlarına bakalım.

import multiprocessing as mp
import time


def calc_partial_pi(rank, nprocs, nsteps, dx):
    partial_pi = 0.0
    for i in range(rank, nsteps, nprocs):
        x = (i + 0.5) * dx
        partial_pi += 4.0 / (1.0 + x * x)
    partial_pi *= dx
    return partial_pi


nsteps = 100000000
dx = 1.0 / nsteps

if __name__ == '__main__':
    nprocs = mp.cpu_count()

    inputs = []
    for rank in range(nprocs):
        inputs.append([rank, nprocs, nsteps, dx])

    pool = mp.Pool(processes=nprocs)

    time1 = time.time()

    result = pool.starmap(calc_partial_pi, inputs)
    pi = sum(result)

    time2 = time.time()

    print("Pi Sayısı: ", pi)
    print('{:.3f} ms'.format((time2 - time1) * 1000.0))

import multiprocessing as mp

import time

def calc_partial_pi(rank, nprocs, nsteps, dx):

partial_pi = 0.0

for i in range(rank, nsteps, nprocs):

x = (i + 0.5) * dx

partial_pi += 4.0 / (1.0 + x * x)

partial_pi *= dx

return partial_pi

nsteps = 100000000

dx = 1.0 / nsteps

if __name__ == '__main__':

nprocs = mp.cpu_count()

inputs = []

for rank in range(nprocs):

inputs.append([rank, nprocs, nsteps, dx])

pool = mp.Pool(processes=nprocs)

time1 = time.time()

result = pool.starmap(calc_partial_pi, inputs)

pi = sum(result)

time2 = time.time()

print("Pi Sayısı: ", pi)

print('{:.3f} ms'.format((time2 - time1) * 1000.0))

starmap için sonuç:

Pi Sayısı: 3.14159265358974
3225.882 ms

——————————

import multiprocessing as mp
import time


def calc_partial_pi(rank, nprocs, nsteps, dx):
    partial_pi = 0.0
    for i in range(rank, nsteps, nprocs):
        x = (i + 0.5) * dx
        partial_pi += 4.0 / (1.0 + x * x)
    partial_pi *= dx
    return partial_pi


nsteps = 100000000
dx = 1.0 / nsteps

if __name__ == '__main__':
    nprocs = mp.cpu_count()

    inputs = []
    for rank in range(nprocs):
        inputs.append([rank, nprocs, nsteps, dx])

    pool = mp.Pool(processes=nprocs)

    time1 = time.time()

    multi_result = [pool.apply_async(calc_partial_pi, inp) for inp in inputs]
    result = [p.get() for p in multi_result]
    pi = sum(result)

    time2 = time.time()

    print("Pi Sayısı: ", pi)
    print('{:.3f} ms'.format((time2 - time1) * 1000.0))

import multiprocessing as mp

import time

def calc_partial_pi(rank, nprocs, nsteps, dx):

partial_pi = 0.0

for i in range(rank, nsteps, nprocs):

x = (i + 0.5) * dx

partial_pi += 4.0 / (1.0 + x * x)

partial_pi *= dx

return partial_pi

nsteps = 100000000

dx = 1.0 / nsteps

if __name__ == '__main__':

nprocs = mp.cpu_count()

inputs = []

for rank in range(nprocs):

inputs.append([rank, nprocs, nsteps, dx])

pool = mp.Pool(processes=nprocs)

time1 = time.time()

multi_result = [pool.apply_async(calc_partial_pi, inp) for inp in inputs]

result = [p.get() for p in multi_result]

pi = sum(result)

time2 = time.time()

print("Pi Sayısı: ", pi)

print('{:.3f} ms'.format((time2 - time1) * 1000.0))

async için sonuç:

Pi Sayısı: 3.14159265358974
3303.169 ms

Ayrıca async için doğru çıktı garantisi verilmemektedir. Bu örnek için sonuç doğru olsa da ek yöntemler uygulanabilir.

Python ile Paralel Kodlama örneği (Pi Sayısı)

Paralel Programlama

Paralel Bilgisayar Sistemleri

Python ile Paralel programlama

π Sayısı (3.141592653589793238462643383279502884197169399375105820974…)

Comments

Write a Reply or Comment Cevabı iptal et