Asyncio: Bir Giriş

Tanımlar

asyncio modülü, olay döngüsü etrafında dönen bir çerçeve sunar. Bir olay döngüsü temelde bir şeylerin olmasını bekler ve ardından olaya göre hareket eder. G/Ç ve sistem olaylarını ele almak onun sorumluluğundadır.

asyncio aslında birden fazla döngü uygulamasına sahiptir. Modül, çalıştığı işletim sistemi için en verimli olması muhtemel olanı varsayılan olarak seçer; ancak isterseniz olay döngüsünü açıkça belirleyebilirsiniz. Bir olay döngüsü kabaca “A olayı olduğunda, B işleviyle tepki ver” der.

Bir sunucuyu, bir web sayfası gibi bir kaynağı talep eden biri gelene kadar beklerken düşünün. Web sitesi çok popüler değilse, sunucu uzun süre boşta kalacaktır. Ancak bir istek geldiğinde sunucunun tepki vermesi gerekir. Bu tepki, olay işleme olarak bilinir. Bir kullanıcı web sayfasını yüklediğinde, sunucu bir veya daha fazla olay işleyiciyi kontrol eder ve çağırır. Bu olay işleyiciler bittiğinde, kontrolü yeniden olay döngüsüne iade etmeleri gerekir. Python’da bunu yapmak için asyncio korotinleri kullanır.

Korotin, durumunu kaybetmeden çağıranına kontrolü bırakabilen özel bir işlevdir. Bir korotin, bir üretecin tüketicisi ve uzantısıdır. İş parçacıklarına göre en büyük faydalarından biri, çalıştırılmak için çok az bellek kullanmalarıdır. Bir korotin işlevini çağırdığınızda, aslında çalıştırılmadığını unutmayın. Bunun yerine, hemen veya daha sonra çalıştırılması için olay döngüsüne verebileceğiniz bir korotin nesnesi döndürür.

asyncio modülünü kullanırken muhtemelen karşınıza çıkacak bir diğer terim de future’dır. Future, tamamlanmamış bir işin sonucunu temsil eden bir nesnedir. Olay döngünüz future nesnelerini izleyebilir ve bitmelerini bekleyebilir. Bir future bittiğinde durumu done olarak ayarlanır. Asyncio ayrıca kilitleri ve semaforları da destekler.

Son olarak bahsetmek istediğim parça Task’tır. Task, bir korotin için bir sarmalayıcı ve Future’ın bir alt sınıfıdır. Hatta bir Task’ı olay döngüsüyle zamanlayabilirsiniz.

async ve await

async ve await anahtar sözcükleri, Python 3.5’te yerel bir korotin tanımlamak ve bunları üreteç tabanlı korotinlerden farklı bir tür olarak ayırmak için eklendi. async ve await hakkında derinlemesine bir açıklama isterseniz, PEP 492’ye göz atın.

Python 3.4’te bir korotini şu şekilde oluştururdunuz:

# Python 3.4 coroutine example
# Import `asyncio`
import _______
 
@asyncio.coroutine
def my_coro():
    yield from func()

Bu dekoratör Python 3.5’te de çalışır, ancak types modülü artık bir korotin işlevi kontrolü alarak etkileştiğiniz şeyin yerel bir korotin olup olmadığını söyleyebilir. Python 3.5’ten itibaren, sözdizimsel olarak bir korotin işlevi tanımlamak için async def kullanabilirsiniz.

Yukarıdaki işlev bu şekilde görünür:

import _______
 
async def _______():
    await func()

Bir korotini bu şekilde tanımladığınızda, korotin işlevinin içinde yield kullanamazsınız. Bunun yerine, çağırana değer döndürmek için kullanılan bir return veya await deyimi içermelidir. await anahtar sözcüğünün yalnızca bir async def işlevinin içinde kullanılabileceğini unutmayın.

async/await anahtar sözcükleri, eşzamanlı olmayan programlama için kullanılacak bir API olarak düşünülebilir. asyncio modülü ise tesadüfen eşzamanlı olmayan programlama için async/await kullanan bir çerçevedir. Bu kavramı kanıtlayan, async/await’i kaputun altında kullanan ayrı bir olay döngüsü uygulaması olan curio adlı bir proje de vardır.

Kötü Bir Korotin Örneği

Tüm bunların nasıl çalıştığına dair bolca arka plan bilgisinin olması yararlı olsa da, bazen sadece sözdizimine ve parçaları nasıl bir araya getireceğinize dair bir fikir edinmek için örnekler görmek istersiniz.

Öyleyse bunu akılda tutarak basit bir örnekle başlayalım!

Oldukça yaygın bir görev, dahili bir kaynaktan veya İnternet’ten bir dosya indirmektir. Genellikle birden fazla dosya indirmek isteyeceksiniz.

O halde bunu yapabilecek bir çift korotin oluşturalım:

import asyncio
import os
import urllib.request
 
async def download_coroutine(url):
    #"A coroutine to download the specified url"
    request = urllib.request.urlopen(url)
    filename = os.path.basename(url)
 
    with open(filename, 'wb') as file_handle:
        while True:
            chunk = request.read(1024)
            if not chunk:
                break
            file_handle.write(chunk)
    msg = 'Finished downloading {filename}'.format(filename=filename)
    return msg
 
async def main(urls):
    """
    Creates a group of coroutines and waits for them to finish
    """
    coroutines = [download_coroutine(url) for url in urls]
    completed, pending = await asyncio.wait(coroutines)
    for item in completed:
        print(item.result())
 
 
if __name__ == '__main__':
    urls = ["http://www.irs.gov/pub/irs-pdf/f1040.pdf",
            "http://www.irs.gov/pub/irs-pdf/f1040a.pdf",
            "http://www.irs.gov/pub/irs-pdf/f1040ez.pdf",
            "http://www.irs.gov/pub/irs-pdf/f1040es.pdf",
            "http://www.irs.gov/pub/irs-pdf/f1040sb.pdf"]
 
    event_loop = asyncio.get_event_loop()
    try:
        event_loop.run_until_complete(main(urls))
    finally:
        event_loop.close()Altyapı:

Bu kodda, ihtiyaç duyduğumuz modülleri içe aktarıyor ve ardından async sözdizimini kullanarak ilk korotinimizi oluşturuyoruz. Bu korotinin adı download_coroutine ve Python’un urllib modülünü, kendisine iletilen URL’yi indirmek için kullanıyor. İş bittiğinde, bunu belirten bir mesaj döndürüyor.

Diğer korotin ise ana korotinimiz. Temel olarak bir veya daha fazla URL listesini alır ve sıraya koyar. Korotinlerin bitmesini beklemek için asyncio’nun wait işlevini kullanıyoruz. Elbette korotinleri gerçekten başlatmak için olay döngüsüne eklenmeleri gerekir. Bunu en sonda, bir olay döngüsü alıp onun run_until_complete metodunu çağırdığımız yerde yaparız. Ana korotini olay döngüsüne ilettiğimize dikkat edin. Bu, ikinci korotini sıraya alan ve onu çalıştıran ana korotini başlatır. Buna zincirli korotin denir.

Bu örneğin sorunu, aslında bir korotin olmamasıdır. Nedeni, download_coroutine işlevinin eşzamanlı olmamasıdır. Sorun, urllib’nin eşzamanlı olmaması ve ayrıca await veya yield kullanmıyor olmamdır. Bunu yapmanın daha iyi bir yolu aiohttp paketini kullanmaktır. Hadi ona bakalım!

Daha İyi Bir Korotin Örneği

aiohttp paketi, eşzamanlı HTTP istemci ve sunucuları oluşturmak için tasarlanmıştır. pip ile şu şekilde kurabilirsiniz:

pip install aiohttpAltyapı:

Kurulum tamamlandıktan sonra, dosyaları indirebilmek için kodumuzu aiohttp kullanacak şekilde güncelleyelim:

import aiohttp
import asyncio
import async_timeout
import os
 
async def download_coroutine(session, url):
    with async_timeout.timeout(10):
        async with session.get(url) as response:
            filename = __.path.basename(url)
            with open(________, 'wb') as f_handle:
                while True:
                    chunk = await response.content.read(1024)
                    if not chunk:
                        break
                    f_handle.write(_____)
            return await response.release()
 
async def main(loop):
    urls = ["http://www.irs.gov/pub/irs-pdf/f1040.pdf",
        "http://www.irs.gov/pub/irs-pdf/f1040a.pdf",
        "http://www.irs.gov/pub/irs-pdf/f1040ez.pdf",
        "http://www.irs.gov/pub/irs-pdf/f1040es.pdf",
        "http://www.irs.gov/pub/irs-pdf/f1040sb.pdf"]
 
    async with aiohttp.ClientSession(loop=loop) as session:
        tasks = [download_coroutine(session, url) for url in urls]
        await asyncio.gather(*tasks)
 
if __name__ == '__main__':
    loop = asyncio.get_event_loop()
    loop.run_until_complete(main(loop))

Burada birkaç yeni öğe içe aktardığımıza dikkat edeceksiniz: aiohttp ve async_timeout. İkincisi aslında aiohttp’nun bir bağımlılığıdır ve bir zaman aşımı bağlam yöneticisi oluşturmamıza olanak tanır. Koddaki en alt kısımdan başlayıp yukarı doğru ilerleyelim. Alttaki koşullu ifadede, eşzamanlı olay döngümüzü başlatıyor ve ana işlevimizi çağırıyoruz.

Ana işlevde, indirmek istediğimiz her url için indirme korotin işlevimize ilettiğimiz bir ClientSession nesnesi oluşturuyoruz. download_coroutine içinde, temelde X saniyelik bir sayaç oluşturan bir async_timeout.timeout() bağlam yöneticisi oluşturuyoruz. Saniyeler dolduğunda bağlam yöneticisi sona erer ya da zaman aşımına uğrar. Bu durumda zaman aşımı 10 saniyedir. Sonra oturumumuzun get() metodunu çağırıyoruz; bu da bize bir response nesnesi verir. Şimdi biraz sihirli olan kısma geliyoruz. Yanıt nesnesinin content özniteliğini kullandığınızda, bize dosyayı istediğimiz boyutta parçalar halinde indirmemize olanak tanıyan bir aiohttp.StreamReader örneği döner. Dosyayı okudukça yerel diske yazarız. Son olarak yanıtın release() metodunu çağırırız; bu, yanıt işlemesini tamamlar.

aiohttp dokümantasyonuna göre, yanıt nesnesi bir bağlam yöneticisi içinde oluşturulduğundan teknik olarak release() örtük olarak çağrılır. Ancak Python’da açık olan genelde daha iyidir ve dokümantasyonda bağlantının kendiliğinden kapanacağına güvenmememiz gerektiğine dair bir not vardır; bu nedenle bu durumda açıkça serbest bırakmanın daha iyi olduğuna inanıyorum.

Burada hâlâ engelleyen bir kısım var; o da diske yazma kısmı. Dosyayı yazarken hâlâ engelliyoruz. Dosya yazmayı da eşzamanlı hâle getirmeye çalışmak için kullanabileceğimiz aiofiles adlı başka bir kütüphane var; fakat o güncellemeyi okura bırakıyorum.

Çağrıları Zamanlama

asyncio olay döngüsünü kullanarak normal işlevlere de çağrılar zamanlayabilirsiniz. Bakacağımız ilk yöntem call_soon. call_soon yöntemi, geri çağrınızı veya olay işleyicinizi mümkün olan en kısa sürede çağırır. FIFO kuyruğu gibi çalışır; dolayısıyla geri çağrıların bazıları çalışmak için zaman alırsa, diğerleri öncekiler bitene kadar gecikir.

Bir örneğe bakalım:

import ______
import functools
 
def event_handler(loop, stop=False):
    print('Event handler called')
    if stop:
        print('stopping the loop')
        loop.stop()

if __name__ == '__main__':
    loop = asyncio.get_event_loop()
    try:
        loop.call_soon(functools.partial(_____________, ____))
        print('starting event loop')
        loop.call_soon(functools.partial(_____________, ____, stop=True))
 
        loop.__________()
    finally:
        print('closing event loop')
        loop._____()

asyncio işlevlerinin çoğu anahtar kelime argümanı kabul etmediğinden, olay işleyicimize anahtar kelimeler iletmemiz gerektiğinde functools modülüne ihtiyaç duyarız. Normal işlevimiz her çağrıldığında stdout’a bazı metinler yazdıracaktır. Eğer stop argümanını True yaparsanız, olay döngüsünü de durdurur.

İlk çağrımızda döngüyü durdurmuyoruz. İkinci çağrımızda döngüyü durduruyoruz. Döngüyü durdurmak istememizin nedeni, döngüye run_forever demiş olmamızdır; bu, olay döngüsünü sonsuz döngüye sokar. Döngü durduğunda ise kapatabiliriz.

Bu kodu çalıştırırsanız, aşağıdaki çıktıyı görmelisiniz:

starting event loop
Event handler called
Event handler called
stopping the loop
closing event loopAltyapı:

call_soon_threadsafe adlı ilişkili bir işlev daha vardır. Adından da anlaşılacağı gibi call_soon ile aynı şekilde çalışır, ancak iş parçacığı güvenlidir. Bir çağrıyı gelecekteki bir zamana kadar geciktirmek isterseniz, call_later işlevini kullanabilirsiniz. Bu durumda, call_soon imzamızı aşağıdaki gibi değiştirebiliriz:

loop.call_later(1, event_handler, loop)Altyapı:

Bu, olay işleyicimizi bir saniye geciktirecek, ardından onu çağıracak ve döngüyü ilk parametre olarak iletecektir. Gelecekte belirli bir zamanı planlamak isterseniz, bilgisayarın zamanı yerine döngünün zamanını almalısınız. Bunu şu şekilde yapabilirsiniz:

current_time = loop.time()Altyapı:

Bunu aldıktan sonra, call_at işlevini kullanıp, olay işleyicinizin çağrılmasını istediğiniz zamanı iletebilirsiniz. Diyelim ki olay işleyicimizi şu andan beş dakika sonra çağırmak istiyoruz. Bunu şöyle yapabilirsiniz:

loop.call_at(current_time + 300, event_handler, loop)Altyapı:

Bu örnekte, aldığımız geçerli zamana 300 saniye yani beş dakika ekliyoruz. Böylece, olay işleyicimizi beş dakika boyunca geciktiriyoruz! Oldukça hoş!

Görevler (Tasks)

Görevler, Future’ın bir alt sınıfıdır ve bir korotinin etrafında bir sarmalayıcıdır. Ne zaman işlemeyi bitirdiklerini takip etmenizi sağlarlar. Future türünde oldukları için, diğer korotinler bir görevi bekleyebilir ve ayrıca görev işleme bittiğinde sonucunu alabilirsiniz.

Basit bir örneğe bakalım:

import _______

async def my_task(seconds):
    """
    A task to do for a number of seconds
    """
    print('This task is taking {} seconds to complete'.format(
        seconds))
    await asyncio.sleep(_______)
    return 'task finished'
 
 
if __name__ == '__main__':
    my_event_loop = asyncio.get_event_loop()
    try:
        print('task creation started')
        task_obj = my_event_loop.create_task(my_task(seconds=2))
        my_event_loop.run_until_complete(task_obj)
    finally:
        my_event_loop._____()
 
    print("The task's result was: {}".format(task_obj.result()))

Burada, çalışması için kaç saniye gerekeceğini alan eşzamanlı bir işlev oluşturuyoruz. Bu, uzun süre çalışan bir işlemi simüle eder. Ardından olay döngümüzü oluşturur ve olay döngüsü nesnesinin create_task işlevini çağırarak bir görev nesnesi yaratırız. create_task, göreve dönüştürmek istediğimiz işlevi kabul eder. Sonra olay döngüsüne, görev tamamlanana kadar çalışmasını söyleriz. En sonda, görev tamamlandığı için sonucunu alırız.

Görevler, cancel metodunu kullanarak çok kolay bir şekilde iptal edilebilir. Bitirmek istediğinizde sadece çağırın. Bir görev başka bir işlemi beklerken iptal edilirse, görev bir CancelledError yükseltir.

Toparlama

Bu noktada, asyncio kütüphanesiyle kendi başınıza çalışmaya başlayacak kadar bilgi sahibi olmalısınız. asyncio kütüphanesi oldukça güçlüdür ve birçok havalı ve ilginç görevi yapmanıza olanak tanır. asyncio kütüphanesi ağ soketleri için tasarlanmıştır.

asyncio’dan önce eşzamanlı soket programlamaya olanak tanıyan harika bir kütüphane Twisted framework’tür. Bir diğer ilginç proje ise analitik hesaplama için esnek bir paralel hesaplama kütüphanesi olan Dask’tır. asyncio modülünün dokümantasyonunu biraz kurcalayın ve fikirlerinizi akmaya bırakın!