Как оптимизировать третий проект в профессии «Python-разработчик» с помощью потоков и конкурентного выполнения кода

Рассказываем, как оптимизировать третий проект по Python и ускорить работу утилиты «Загрузчик страниц». Спойлер — статья подходит только для тех, кто уже закончил проходить третий проект в профессии Python-разработчик, но хочет доработать свой код. Остальные же могут почитать наш большой материал про прокрастинацию у программистов.

Для кого эта статья?

Материал подходит для студентов Хекслета, которые сделали третий проект в профессии Python-разработчик и при этом заинтересовались темой потоков, процессов и конкурентного выполнения кода в Python. При этом, мы не ставим перед собой задачу создать полноценное руководство по использованию потоков в Python — в статье мы показываем, как при помощи этого инструмента конкретно решить задачу оптимизации третьего проекта.

Простая реализация

Посмотрим на самую простую реализацию этой утилиты. При этом мы не будем показывать весь код, а только основные кусочки, которые отвечают за скачивание.

import logging
import os

from progress.bar import IncrementalBar
from requests.exceptions import RequestException

import page_loader.url
from page_loader import html, resources, storage


def download(url, output="."):
    url = url.rstrip("/")
    assets_dir = page_loader.url.to_dir_name(url)
    assets_path = os.path.join(output, assets_dir)

    html_page_path = os.path.join(output, page_loader.url.to_file_name(url))

    logging.info(f"Saving {url} to the {output} ...")

    page_content = resources.get(url)
    page, assets = html.prepare(page_content, url, assets_dir)
    logging.info('Saving html file: %s', html_page_path)
    storage.save(html_page_path, page)
    download_assets(assets, assets_path)
    return html_page_path


def download_assets(assets, assets_path):
    if not assets:
        return

    if not os.path.exists(assets_path):
        logging.info('Create directory for assets: %s', assets_path)
        os.mkdir(assets_path)

    bar_width = len(assets)

    with IncrementalBar("Downloading:", max=bar_width) as bar:
        bar.suffix = "%(percent).1f%% (eta: %(eta)s)"

        for url, file_name in assets:
            try:
                asset_content = resources.get(url)
                storage.save(os.path.join(assets_path, file_name),
                             asset_content)
                bar.next()

            except (RequestException, OSError) as e:
                cause_info = (e.__class__, e, e.__traceback__)
                logging.debug(str(e), exc_info=cause_info)
                logging.warning(
                    f"Page resource {url} wasn't downloaded"
                )

Давайте посмотрим, как работает этот код.

Основная функция download отвечает за главную верхнеуровневую логику:

• Она скачивает главную HTML-страницу
• Вызывает функцию html.prepare(), которая подготавливает новую HTML-страницу и собирает все ресурсы
• Сохраняет HTML-страницу
• Вызывает функцию download_assets для скачивания всех ресурсов.

Функция download_assets проходит по списку всех ресурсов, скачивает при помощи resources.get(), а затем просто сохраняет на диск вызовом storage.save().

Читайте также: Как сохранять фокус на протяжении всего обучения: советы от Хекслета

Кажется, что наш код выглядит достаточно логично и никакой специальной оптимизации не требует. Однако стоит отметить один нюанс нашей программы — скачивание одного ресурса не влияет на скачивание другого ресурса, и вообще — это абсолютно независимые друг от друга действия с точки зрения логики утилиты. Например, если какая-то картинка не загрузилась, то нужно просто скачать остальные ресурсы — в этом нет ничего страшного. В то же время на уровне кода эти действия идут в определенном порядке. Пока не будет скачан текущий ресурс, скачивание следующего не начнется. А если картинка на странице весит пару мб, и ее скачивание происходит достаточно медленно? Тогда следующий ресурс будет очень долго ждать, пока не произойдет скачивание предыдущих ресурсов.

В итоге мы получаем, что на скачивание всех ресурсов уйдет суммарное время скачивания ресурсов по одному. Для маленьких страниц это может быть не так критично. Но если вы скачиваете страницу с большим количеством объемных ресурсов, то возможно придется подождать несколько десятков секунд.

Тогда возникает вопрос: как сделать быстрее?

Реализация при помощи потоков

Представим, что можно было бы скачивать ресурсы параллельно, независимо друг от друга уже на уровне кода. То есть скачивать ресурсы так, чтобы скачивание одного ресурса не задерживало скачивание другого? Это можно сделать, для этого в Python есть специальное понятие — потоки.

Каждая запущенная программа на компьютере — это какой-то процесс, в который входит код и память, выделенная под него. Внутри каждого процесса есть свои потоки выполнения, их может быть от одного до нескольких десятков или даже сотен. Когда Python запускает ваш код, то запускается процесс с одним потоком выполнения внутри. Именно он по одному выполняет команды программы.

Если в вашем процессе несколько потоков, то ОС (точнее планировщик задач) сама решает, в каком порядке их запускать, когда и какой нужно приостановить, и когда снова перезапустить.

Потоки в Python

В Python объекты-потоки можно создавать при помощи класса Thread из стандартного модуля threading.

import logging
import threading
import time

def thread_function(name):
    logging.info("Thread %s: starting", name)
    time.sleep(2)
    logging.info("Thread %s: finishing", name)


if __name__ == "__main__":
    format = "%(asctime)s: %(message)s"
    logging.basicConfig(format=format, level=logging.INFO,
                        datefmt="%H:%M:%S")

    logging.info("Main    : before creating thread")
    x = threading.Thread(target=thread_function, args=(1,))
    logging.info("Main    : before running thread")
    x.start()
    logging.info("Main    : wait for the thread to finish")
    x.join()
    logging.info("Main    : all done")

Метод start() запускает поток и начинает его выполнять. Метод join заставляет основной поток подождать, пока второй не закончит свою работу. При этом можно не блокировать основной поток — тогда он независимо от потока x продолжит свое выполнение, и сообщения после start() смогут немного поменять свой порядок. Поэтому мы не сможем узнать, в каком порядке они будут выполняться — ОС сама решает, когда запустить поток, а когда его прервать.

Потоки — сложная тема для новичков (да и не для новичков тоже). Поэтому мы кратко опишем на нашем кейсе, как вообще они работают.

Аналогия такая: представьте, что каждый ресурс это какая-то посылка на почте, которая вас ждет. Предположим, что за одну поездку вы сможете принести с собой ровно одну посылку. Для того, чтобы привезти все, вам придется съездить на почту несколько раз. Но зачем? Вы можете попросить друга, брата и соседа пойти с вами. Тогда вы перевезете все гораздо быстрее.

С нашей реализацией дела обстоят примерно похожим образом. В ней существует ровно один поток выполнения, и он делает все совсем один. Но никто не запрещает создать несколько потоков, а ту часть программы, которая отвечает за скачивание ресурсов, распределить между этими потоками. Пускай они одновременно скачивают ресурсы независимо друг от друга.

import logging
import os
from concurrent import futures

from progress.bar import IncrementalBar
from requests.exceptions import RequestException

import page_loader.url
from page_loader import html, resources, storage

bar = None


def download(url, output="."):
    url = url.rstrip("/")
    assets_dir = page_loader.url.to_dir_name(url)
    assets_path = os.path.join(output, assets_dir)

    html_page_path = os.path.join(output, page_loader.url.to_file_name(url))

    logging.info(f"Saving {url} to the {output} ...")

    page_content = resources.get(url)
    page, assets = html.prepare(page_content, url, assets_dir)
    logging.info('Saving html file: %s', html_page_path)
    storage.save(html_page_path, page)
    download_assets(assets, assets_path)
    return html_page_path


def download_assets(assets, assets_path):
    if not assets:
        return

    if not os.path.exists(assets_path):
        logging.info('Create directory for assets: %s', assets_path)
        os.mkdir(assets_path)

    bar_width = len(assets)
    global bar
    bar = IncrementalBar("Downloading:", max=bar_width)
    with futures.ThreadPoolExecutor(max_workers=8) as executor, bar:
        tasks = [executor.submit(download_and_save_asset, url, file_name, assets_path, bar) for url, file_name in assets]
        result = [task.result() for task in tasks]
        logging.info(f"All assets was downloaded: {result}")


def download_and_save_asset(url, file_name, assets_path, bar):
    try:
        asset_content = resources.get(url)
        storage.save(os.path.join(assets_path, file_name),
                     asset_content)
        bar.next()
    except (RequestException, OSError) as e:
        cause_info = (e.__class__, e, e.__traceback__)
        logging.debug(str(e), exc_info=cause_info)
        logging.warning(
            f"Page resource {url} wasn't downloaded"
        )
    return os.path.join(assets_path, file_name)

Код поменялся не очень сильно. Появилась новая функция download_and_save_asset, которая скачивает ресурс по-нужному url и пишет в нужное место. Тут надо отметить, что функция storage.save() тоже перешла в эту функцию. Под капотом она просто записывает данные на диск.

После этого мы создаем объект ThreadPoolExecutor, который под капотом запускает несколько потоков (максимум их 8 — за это отвечает параметр max_workers) и поручает им разные задачи. В нашем случае это вызов функции download_and_save_asset для конкретного ресурса.

Сравнение

Чтобы проверить, насколько отличается время работы утилиты с одним потоком и с несколькими, надо скачать одну и ту же страницу двумя реализациями.

Для скачивания страницы из «Википедии» про Машину Тьюринга реализация с несколькими потоками потратила около 2,5 сек.

Вывод терминала будет таким:

╰─$ time page-loader -o ./data https://en.wikipedia.org/wiki/Turing_machine > /dev/null 2>&1
page-loader -o ./data https://en.wikipedia.org/wiki/Turing_machine > /dev/nul  0.78s user 0.10s system 35% cpu 2.450 total

Эту же страницу однопоточный вариант скачал за почти 8,5 сек.

╰─$ time page-loader -o ./data https://en.wikipedia.org/wiki/Turing_machine > /dev/null 2>&1
page-loader -o ./data https://en.wikipedia.org/wiki/Turing_machine > /dev/nul  0.83s user 0.12s system 11% cpu 8.405 total

Можете сами попробовать переписать свои решения и убедиться, что они начнут работать намного быстрее.

Все ли так хорошо?

Потоки в Python не являются серебряной пулей — совсем не всегда с их помощью можно увеличить производительность вашей программы. В Python есть механизм GIL, который ограничивает использование потоков, который блокирует параллельное выполнение и заставляет работать их последовательно, один за другим.

В нашем примере потоки работают, потому что им поручают I/O-bound (Input Output)-задачи: скачать что-то по сети, записать файл на диск.

Но если потокам поручить какую-то вычислительную задачу, например, вычисление квадратного корня, либо сортировку списка, то они не будут работать эффективно. Такие задачи называются CPU-bound-задачами, и невозможность их распараллеливания при помощи потоков как раз является следствием упомянутого механизма GIL. Мы в этой статье не будем подробно углубляться в эту тему, но если вам хочется самим подробнее в ней разобраться, то вы можете почитать эти материалы:

• Про особенности работы GIL. Статья на английском
• Доклад про GIL. На русском
• Статья про использование потоков в Python

Никогда не останавливайтесь: В программировании говорят, что нужно постоянно учиться даже для того, чтобы просто находиться на месте. Развивайтесь с нами — на Хекслете есть сотни курсов по разработке на разных языках и технологиях