Python ile Asenkron İşlemler

Jusstthe' Alıntı:

Faydalı konu. Keşke thread-safety kavramına değinilseydi

Genişletmek için tıkla ...

Hacknology' Alıntı:

Merhaba. Normalde MVVM başlığı açacaktım ama son zamanlar forumda yazılan kodları görünce Concurrency - Paralellism konularına forumda ihtiyaç olduğunu düşündüm; o yüzden bu rehberde elimden geldiğince asenkron yapılardan bahsetmeye çalışıp Python ile implemente edeceğim. Biraz uzun bir rehber olacak, tam olarak nereden başlayacağımı da planlamadığım için uğraştırıcı olacak ama bu rehberde mutex - semaphor konularına yer vermeyeceğim. Ayrıca thread-process farkları için de yeni bir başlık açacağım. Bu rehberde ise elimden geldiğince işlemci verimini arttırmaktan bahsedeceğim; aksi halde okunmayacak kadar uzun olacak. Başlayalım,

Öncelikle belirtmek gerekir ki ikisi de "Multitasking" başlığı altında yer alan paralellism ve concurrency farklı şeylerdir. Paralellism aynı anda birden çok görevin (rehberin geri kalanında task kelimesini kullanacağım) işlenmesidir. Doğal olarak tek çekirdekli CPU'larda -mesela birçok mikroişlemcide!- paralellism uygulamak mümkün değildir. Concurrency ise tek bir CPU çekirdeğinin birden fazla task'ı belirli oranda paylaştırarak yapması ve toplam harcanan zamanı minimize etmeye çalışmasıdır. CPU, bu tasklar arasından en uygunu bir algoritma yardımıyla -CPU Scheduling adı veriliyor, başka bir rehberde daha detaylı incelenecektir.- seçer ve işini yapar. Multitasking genel anlamda iki felsefe ile çalışır:

1 - Task Splitting:
Thread'ler aynı veriyi kullanırlar, hepsi farklı işlemleri yapar. Sözgelimi büyük bir SQL datası elimizde olsun. Bir thread bu datadaki kullanıcı isimlerini kaydeder, başka bir thread bu sırada telefon numaralarının geçerliliğini kontrol eder. Bu sayede sıra sıra birbirinden bağımsız işleri yapmaktansa aynı veriyi farklı threadler farklı şekillerde işler.

2 - Data Splitting:
Thread'ler büyük bir veriyi eşit parçalar halinde bölüşür ve aynı işlemi yapar. Mesela 100 bin kullanıcıya sms atmak istediğiniz bir uygulamanızda, 100 bin kullanıcı listesini eşit parçaya bölüp her bir thread'e aynı SmsAt() fonksiyonunu çağırtarak bu işlem gerçekleştirilir.

Hayali bir uygulama üzerinden asenkron işlem implemente edelim. Bir "Öğrenci" sınıfımız olsun ve tüm öğrenciler rastgele isimlere sahip olsun, 1'000'000 öğrencilik boş bir öğrenci listesinin tamamına isim kaydı yapmak isteyelim. Bunu Data Splitting ile yapacağız, ardından isim kaydı yaptığımız listenin tamamını ele alalım ve hem yaş hem de şehir bilgileri giren bir fonksiyonumuz olsun. Bu işlemleri hem senkron hem asenkron çalıştırıp performans iyileştirmesine bakalım. Eğer tek thread'de isim kaydedecek olsaydık:

Python:

import threading
import random, string
import time
class Ogrenci(): #Ogrenci sinifini yaratiyoruz.
    def __init__(self):
        self.isim = "" #bu degerlere sahip olacaklar
        self.sehir = ""
        self.yas = 0

def isim_kayit(ogrenci : Ogrenci, isim : str):
    ogrenci.isim = isim

ogrenci_liste = []


for i in range(1000000): #bir milyon ogrenci var
    ogrenci_liste.append(Ogrenci())

print("Isim Kayit Basliyor...")

baslangic = time.time_ns() #baslangic zamani
for ogrenci in ogrenci_liste:
    
    ogrenci_isim = ''.join(random.choices(string.ascii_letters, k=10)) #hepsi rastgele isme sahip.
    isim_kayit(ogrenci, ogrenci_isim) #isimleri kaydet.
bitis = time.time_ns() #bitis zamani.

print(f"Isim Kayit Bitti! Gecen sure {(bitis-baslangic) / (1000000000):.2f} s")

Bash:

PS C:\Users\Hacknology\tht> python .\paralellism.py
Isim Kayit Basliyor...
Isim Kayit Bitti! Gecen sure 1.48 s

Programımızın bütün işlemleri yapması 1.48 saniye almış. Şimdi bu bir milyonluk öğrenci listesini eşit parçalara bölelim ve farklı threadlere verelim. Bunun için belirli bir thread sayısı seçmeliyiz, ben 200 olarak belirledim çünkü 1 milyona tam bölünebiliyor ve yeterli bir sayı. Her bir thread'e ne kadarlık bir liste düşeceğini hesaplayıp pass etmeliyiz, bunun için isim_kayit fonksiyonunu da birazcık yeniledim. Değişiklikleri eklemek kod takibini zorlaştırabileceğinden ötürü (ve farkı net olarak görebilmek için) asenkron çalışan kodun tamamını paylaşıyorum.

Python:

import threading
import random, string
import time

NUM_THREADS = 200
OGRENCI_SAYISI = 1000000
class Ogrenci(): #Ogrenci sinifini yaratiyoruz.
    def __init__(self):
        self.isim = "" #bu degerlere sahip olacaklar
        self.sehir = ""
        self.yas = 0

ogrenci_liste = [] #ogrenci listesi
thread_basi_liste_uzunluk = OGRENCI_SAYISI // NUM_THREADS #her bir threade dusecek eleman sayısı, 1m / 200
thread_list = [] #calisacak threadlerin listesi

def isim_kayit(ogrenci_listesi : list, isim: str): #isim_kayit artik bir ogrenci listesi aliyor, bu listeyi dolaşarak isim kaydediyor.
    for ogrenci in ogrenci_listesi:
        ogrenci.isim = isim

for i in range(OGRENCI_SAYISI): #bir milyon ogrenci var
    ogrenci_liste.append(Ogrenci())

print("Isim Kayit Basliyor...")

baslangic = time.time_ns() #baslangic zamani
for i in range(NUM_THREADS):
    start = i * thread_basi_liste_uzunluk
    end = 0 #her bir threadin payina dusen alt listenin baslangic ve bitis indexleri
    if i < (NUM_THREADS - 1):
        end = (i+1) * thread_basi_liste_uzunluk
    else:
        end = len(ogrenci_liste)
    alt_liste = ogrenci_liste[start:end] #her bir threadin payina dusen alt liste
    ogrenci_isim = ''.join(random.choices(string.ascii_letters, k=10)) #hepsi rastgele isme sahip.

    t = threading.Thread(target=isim_kayit, args=(alt_liste, ogrenci_isim))
    t.start() #threadleri yaratiyoruz.
    thread_list.append(t) #thread listesine ekliyoruz

for t in thread_list:
    t.join() #threadleri baslatiyoruz.
bitis = time.time_ns() #bitis zamani.

print(f"Isim Kayit Bitti! Gecen sure {(bitis-baslangic) / (1000000000):.2f} s")

Bash:

PS C:\Users\Hacknology\tht> python .\paralellism.py
Isim Kayit Basliyor...
Isim Kayit Bitti! Gecen sure 0.05 s

Farkı görüyor musunuz ?! Tam tamına %97 lik bir performas artışı sağladık! Sadece bu bile bize asenkron ve senkron işlemlerin farkını anlatıyor. Bir de task splitting örneğini gösterebilmek adına şimdi bu öğrencilere rastgele şehir ve yaş bilgisi de ekleyelim. Bunun için iki fonksiyon daha yazıyorum

Python:

def yas_kayit(ogrenci: Ogrenci, yas: int):
    ogrenci.yas = yas

def sehir_kayit(ogrenci: Ogrenci, sehir: str):
    ogrenci.sehir = sehir

## codes
## codes
print("Tum Islemler Basliyor...")

baslangic = time.time_ns() #baslangic zamani
## codes
## codes
## codes
for ogrenci in ogrenci_liste:
    sehir = ''.join(random.choices(string.ascii_letters, k=10)) #hepsi rastgele sehir adina sahip.
    yas = random.randint(1,30) # [1-30] yas araligindalar.

    yas_kayit(ogrenci, yas)
    sehir_kayit(ogrenci, sehir)

bitis = time.time_ns() #bitis zamani.

print(f"Islemler Bitti! Gecen sure {(bitis-baslangic) / (1000000000):.2f} s")

Bash:

PS C:\Users\Hacknology\tht> python .\paralellism.py
Tum Islemler Basliyor...
Islemler Bitti! Gecen sure 1.81 s

Gördüğümüz gibi toplam geçen süre 1.81 saniye ve bunun yalnızca 0.03 saniyesi asenkron taraftan geldi. Geri kalan yükümüzün tamamı sıra sıra yaptığımız bu işlemde, şimdi bunu 2 ayrı threade vermeyi deneyelim. (Asıl en verimli yol 2 ayrı process'e bölüp her bir process içinde birden fazla thread ile data splitting yapmak ama bu rehberin hedefinden dışarıya taşıyor.) Bunu yapmak için kayıt fonksiyonlarını güncelledim ve koda birkaç satır ekledim.

Python:

import threading
import random, string
import time

lock = threading.Lock()
##CODES CODES

def yas_kayit(ogrenci_listesi: list):
    yas = random.randint(1,30) # [1-30] yas araligindalar
    for ogrenci in ogrenci_listesi:
        with lock: #lock kullanarak programı thread-safe yaptık ve race conditionlarin onune gectik
            ogrenci.yas = yas
def sehir_kayit(ogrenci_listesi: list):
    sehir = ''.join(random.choices(string.ascii_letters, k=10)) #hepsi rastgele sehir adina sahip.
    for ogrenci in ogrenci_listesi:
        with lock:
            ogrenci.sehir = sehir

for i in range(OGRENCI_SAYISI): #bir milyon ogrenci var
    ogrenci_liste.append(Ogrenci())

print("Tum Islemler Basliyor...")

### CODES CODES

task_thread_list = []

yas_kayit_thread   = threading.Thread(target=yas_kayit,   args=(ogrenci_liste,))
sehir_kayit_thread = threading.Thread(target=sehir_kayit, args=(ogrenci_liste,))

yas_kayit_thread.start()
sehir_kayit_thread.start()

yas_kayit_thread.join()
sehir_kayit_thread.join()

bitis = time.time_ns() #bitis zamani.

print(f"Islemler Bitti! Gecen sure {(bitis-baslangic) / (1000000000):.2f} s")

Bu sefer ise çıktımız aşağıdaki gibi:

Bash:

PS C:\Users\Hacknology\tht> python .\paralellism.py
Tum Islemler Basliyor...
Islemler Bitti! Gecen sure 0.28 s

Gördüğünüz gibi akıllara zarar performans karları kazandık. Buraya kadar okuduğunuz için teşekkür ederim. Tüm kodu aşağıya paylaşıyorum.

Python:

import threading
import random, string
import time

NUM_THREADS = 200
OGRENCI_SAYISI = 1000000
class Ogrenci(): #Ogrenci sinifini yaratiyoruz.
    def __init__(self):
        self.isim = "" #bu degerlere sahip olacaklar
        self.sehir = ""
        self.yas = 0

ogrenci_liste = [] #ogrenci listesi
thread_basi_liste_uzunluk = OGRENCI_SAYISI // NUM_THREADS #her bir threade dusecek eleman sayısı, 1m / 200
thread_list = [] #calisacak threadlerin listesi
lock = threading.Lock()

def isim_kayit(ogrenci_listesi : list, isim: str): #isim_kayit artik bir ogrenci listesi aliyor, bu listeyi dolaşarak isim kaydediyor.
    for ogrenci in ogrenci_listesi:
        ogrenci.isim = isim

def yas_kayit(ogrenci_listesi: list):
    yas = random.randint(1,30) # [1-30] yas araligindalar
    for ogrenci in ogrenci_listesi:
        with lock:
            ogrenci.yas = yas

def sehir_kayit(ogrenci_listesi: list):
    sehir = ''.join(random.choices(string.ascii_letters, k=10)) #hepsi rastgele sehir adina sahip.
    for ogrenci in ogrenci_listesi:
        with lock:
            ogrenci.sehir = sehir

for i in range(OGRENCI_SAYISI): #bir milyon ogrenci var
    ogrenci_liste.append(Ogrenci())

print("Tum Islemler Basliyor...")

baslangic = time.time_ns() #baslangic zamani
for i in range(NUM_THREADS):
    start = i * thread_basi_liste_uzunluk
    end = 0 #her bir threadin payina dusen alt listenin baslangic ve bitis indexleri
    if i < (NUM_THREADS - 1):
        end = (i+1) * thread_basi_liste_uzunluk
    else:
        end = len(ogrenci_liste)
    alt_liste = ogrenci_liste[start:end] #her bir threadin payina dusen alt liste
    ogrenci_isim = ''.join(random.choices(string.ascii_letters, k=10)) #hepsi rastgele isme sahip.

    t = threading.Thread(target=isim_kayit, args=(alt_liste, ogrenci_isim))
    t.start() #threadleri yaratiyoruz.
    thread_list.append(t)
for t in thread_list:
    t.join() #threadleri baslatiyoruz.


task_thread_list = []

yas_kayit_thread   = threading.Thread(target=yas_kayit,   args=(ogrenci_liste,))
sehir_kayit_thread = threading.Thread(target=sehir_kayit, args=(ogrenci_liste,))

yas_kayit_thread.start()
sehir_kayit_thread.start()

yas_kayit_thread.join()
sehir_kayit_thread.join()

bitis = time.time_ns() #bitis zamani.

print(f"Islemler Bitti! Gecen sure {(bitis-baslangic) / (1000000000):.2f} s")

Genişletmek için tıkla ...

Ecstasy07' Alıntı:

Eline sağlık Dostum

Genişletmek için tıkla ...

sametghack68' Alıntı:

Eline saglik

Genişletmek için tıkla ...