Python на сверхзвуке

[L.Luciano]

Речь пойдёт о способе достижения максимальной скорости обработки удалённых данных при использовании Python + различных библиотек.

Спойлер

asyncio+aiohttp

Мы будем пытаться добиться максимальной эффективности: скачивать много, по возможности быстро, ну и как можно менее затратно по ресурсам.

Предыстория: исторически образовался клуб любителей GIL-многопоточности* - людей, знающих о неблокирующих сокетах ровным счётом ничего и свято уверенным, что единственный способ распаралелливания каких-либо процессов - мультипоточность. На деле же мультипоточность имеет свои ограничения и сферу применения. А именно: сложные вычислительные процессы, обработка данных в несколько потоков, потребность в синхронизации доступа к данным (и вытекающие из этого блокировки, паузы в работе) и подобное.
*Достаточно набрать "multithreading" в поиске на форуме

В чём проблема с GIL-потоками? Они кушают память, 8 MB RAM на 1 поток, в большем кол-ве это заметнее (например: 8*100=800 MB RAM), нужно писать thread-safe код, низкая продуктивность в сравнении с другими решениями.

Мной был проведён ряд тестов, результаты одного из них приведены ниже. Примерный разброс такой: multithread (100, 150, 200, 1000 потоков) - 25-33 sec; greqs - 35-41 sec; async - 15-18 sec. Если эти числа умножить на 10, то можно получить примерное кол-во секунд, требуемое для отправки 100к запросов. Разброс получится весьма большой и эти секунды превратятся в минуты при миллионе запросов.

Тесты проводились в локальной сети на нешифрованном, а затем и шифрованном соединении с использованием наиболее популярных библиотек для python. Для эксперимента создавалось до 1000 одновременных соединений, целью было создание 10000 запросов. Лучше всех(по скорости) показала себя asyncio, затем threading и grequests. Подводя итоги, можно узреть лидерство асинхронных библиотек и неоспоримое преимущество при больших нагрузках.

Результаты:

user@host:~/pylibs$ time python3 greqs.py
36.9738130569458 seconds

real    0m37.453s
user    0m32.000s
sys     0m4.740s
user@host:~/pylibs$ time python3 async.py
18.482529640197754 seconds

real    0m18.845s
user    0m11.452s
sys     0m0.560s
user@host:~/pylibs$ time python3 multithread.py 1000
Threads: 1000
32.848010778427124 seconds (time)

real    0m33.182s
user    0m24.024s
sys     0m3.024s

Спойлер

threading

[CODE]from threading import Thread
import requests
import sys
from queue import Queue
import time

if len(sys.argv) != 2:
    print("Using:", sys.argv[0], "[threads]")
    sys.exit(1)
concurrent = int(sys.argv[1])
print("Threads: "+sys.argv[1])
start_time = time.time()

def doWork():
    while True:
        url = q.get()
        status, url = getStatus(url)
        doSomethingWithResult(status, url)
        q.task_done()

def getStatus(url):
    try:
        res = requests.get(url)
        return res.status_code, url
    except Exception as e:
        return "error", e

def doSomethingWithResult(status, url):
    #print(status, url)
    pass

q = Queue(concurrent * 2)
for i in range(concurrent):
    t = Thread(target=doWork)
    t.daemon = True
    t.start()
urls = ("http://localhost/ "*10000).split(" ")
try:
    for url in urls:#open('urllist.txt'):
        q.put(url.strip())
    q.join()
except KeyboardInterrupt:
    sys.exit(1)

print(time.time() - start_time, "seconds (time)")

grequests

import grequests
import time

start_time = time.time()
urls = ("http://localhost/ "*10000).split(" ")
urls.pop()

def exception_handlerr(request, exception):
    print("Request failed", request.url)

rs = (grequests.get(u) for u in urls)
for r in grequests.imap(rs, size=1000, exception_handler=exception_handlerr):
    #print(r.status_code, r.url)
    pass

print(time.time() - start_time, "seconds")

asyncio+aiohttp

import random
import asyncio
import time
from aiohttp import ClientSession

async def fetch(url, session):
    try:
        async with session.get(url) as response:
            return
    except Exception as e:
        #print(e)
        pass


async def bound_fetch(sem, url, session):
    # Getter function with semaphore.
    async with sem:
        await fetch(url, session)


async def run(r):
    url = "http://localhost:80/ "
    tasks = []
    # create instance of Semaphore
    sem = asyncio.Semaphore(1000)
    urls = (url*r).split(" ")
    urls.pop()

    # Create client session that will ensure we dont open new connection
    # per each request.
    async with ClientSession() as session:
        for url in urls:
            # pass Semaphore and session to every GET request
            task = asyncio.ensure_future(bound_fetch(sem, url, session))
            tasks.append(task)

        responses = asyncio.gather(*tasks)
        await responses

start_time = time.time()

number = 10000
loop = asyncio.get_event_loop()

future = asyncio.ensure_future(run(number))
loop.run_until_complete(future)

print(time.time() - start_time, "seconds")

Заключение: каждый способ хорош для своих задач, но не следует использовать какой-либо инструмент, пытаясь решить все задачи. Не существует "панацеи", это давно признали медики и пора бы уже признать программистам. Для каждой задачи нужен свой, индивидуальный подход, который позволит решить задачу максимально эффективно, как в плане трудозатрат, так и в плане эффективности работы софта. В данном случае было доказано, что для парсинга веб-страниц, гораздо эффективнее будет использовать asyncio+aiohttp, чем другие методы.

Выражаю огромную благодарность SilverT за помощь в написании статьи, а также за его просветительскую деятельность.

Интересные линки:
Документация по AsyncIO
Документация по AsyncIO с юзабельными примерами
1 Million Requests with python-aiohttp
Python's Web Framework Benchmarks
Также нашёл интересную статью, но там всё на сокетах

P.S. Позже может будут (а может и не будут) результаты тестов с HTTPS, где всё гораздо более однозначно
P.S.S. Если кого-то интересует, то также можно написать небольшую статью по использованию AsyncIO (это займёт время, поэтому, чем больше желающих, тем больше вероятность появления в свет статьи)