Асинхронное программирование: концепция Deferred

Асинхронная концепция программирования заключается в том, что результат выполнения функции доступен не сразу же, а через некоторое время в виде

некоторого асинхронного (нарушающего обычный порядок выполнения) вызова. Зачем

такое может быть полезно? Рассмотрим несколько примеров.

Первый пример - сетевой сервер, веб-приложение. Чаще всего как таковых

вычислений на процессоре такие приложения не выполняют. Большая часть времени

(реального, не процессорного) тратится на ввод-вывод: чтение запроса от

клиента, обращение к диску за данными, сетевые обращение к другим подсистемам

(БД, кэширующие сервера, RPC и т.п.), запись ответа клиенту. Во время этих

операций ввода-вывода процессор простаивает, его можно загрузить обработкой

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

процесс на каждое соединение (Apache mpm_prefork, PostgreSQL, PHP FastCGI),

отдельный поток (нить) на каждое соединение или комбинированный вариант

процесс/нить (Apache mpm_worker, MySQL). Подход с использованием процессов или

нитей перекладывает мультиплексирование процессора между обрабатываемыми

соединениями на ОС, при этом расходуется относительно много ресурсов (память,

переключения контекста и т.п.), такой вариант не подходит для обработки

большого количества одновременных соединений, но идеален для ситуации, когда

объем вычислений достаточно высок (например, в СУБД). К плюсам модели нитей и

процессов можно добавить потенциальное использование всех доступных процессоров

в многопроцессорной архитектуре.

Альтернативой является использование однопоточной модели с использованием

примитивов асинхронного ввода-вывода, предоставляемых ОС (select, poll, и

т.п.). При этом объем ресурсов на каждое новое обслуживаемое соединение не

такой большой (новый сокет, какие-то структуры в памяти приложения). Однако

программирование существенно усложняется, т.к. данные из сетевых сокетов

поступают некоторыми "отрывками", причем за один цикл обработки данные поступают от разных

соединений, находящихся в разных состояниях, часть соединений могут быть

входящими от клиентов, часть - исходящими к внешним ресурсам (БД, другой сервер

и т.п.). Для упрощения разработки используются различные концепции: callback,

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

ввод-вывод: nginx, lighttpd,

HAProxy, pgBouncer, и т.д. Именно при такой

однопоточной модели возникает необходимость в асинхронном программировании.

Например, мы хотим выполнить запрос в БД. С точки зрения программы выполнение

запроса - это сетевой ввод-вывод: соединение с сервером, отправка запроса,

ожидание ответа, чтение ответа сервера БД. Поэтому если мы вызываем

функцию "выполнить запрос БД", то она сразу вернуть результат не сможет

(иначе она должна была бы заблокироваться), а вернет лишь нечто, что позволит

впоследствие получить результат запроса или, возможно, ошибку (нет соединения

с сервером, некорректный запрос и т.п.) Этим возвращаемым значением удобно

сделать именно Deferred.

Второй пример связан с разработкой обычных десктопных приложений. Предположим,

мы решили сделать аналог Miranda (QIP, MDC, ...), то есть свой мессенджер.

В интерфейсе программы есть контакт-лист, где можно удалить контакт. Когда

пользователь выбирает это действие, он ожидает что контакт исчезнет на экране

и что он действительно удалится из контакт-листа. На самом деле операция

удаления из серверного контакт-листа опирается на сетевое взаимодействие

с сервером, при этом пользовательский интерфейс не должен быть заблокирован

на время выполнения этой операции, поэтому в любом случае после выполнения

операции потребуется некоторое асинхронное взаимодействие с результатом операции.

Можно использовать механизм сигналов-слотов, callback'ов или что-то еще,

но лучше всего подойдет Deferred: операция удаления из контакт-листа возвращает

Deferred, в котором обратно придет либо положительный результат (всё хорошо),

либо исключение (точная ошибка, которую надо сообщить пользователю): в случае

ошибки контакт надо восстановить контакт в контакт-листе.

Примеры можно приводить долго и много, теперь о том, что же такое Deferred.

Deferred - это сердце framework'а асинхронного сетевого программирования

Twisted в Python. Это простая и стройная концепция, которая

позволяет перевести синхронное программирование в асинхронный код, не изобретая

велосипед для каждой ситуации и обеспечивая высокое качества кода.

Deferred - это просто возвращаемый результат функции, когда этот результат

неизвестен (не был получен, будет получен в другой нити и т.п.) Что мы можем

сделать с Deferred? Мы можем "подвеситься" в цепочку обработчиков, которые

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

только положительный результат выполнения, но и исключения, сгенерированные

функцией или обработчиками, есть возможность исключения обработать, перевыкинуть

и т.д. Фактически, для синхронного кода есть более-менее однозначная параллель в

терминах Deferred. Для эффективной разработки с Deferred оказываются полезными

такие возможности языка программирования, как замыкания, лямбда-функци.

Приведем пример синхронного кода и его альтернативу в терминах Deferred:

try:
    # Скачать по HTTP некоторую страницу
    page = downloadPage(url)
    # Распечатать содержимое
    print page
except HTTPError, e:
    # Произошла ошибка
    print "An error occured: %s", e

В асинхронном варианте с Deferred он был бы записан следующим образом:

def printContents(contents):
    """
    Callback, при успешном получении текста страницы,
    распечатываем её содержимое.
    """
    print contents


def handleError(failure):
    """
    Errback (обработчик ошибок), просто распечатываем текст ошибки.
    """

    # Мы готовы обработать только HTTPError, остальные исключения
    # "проваливаются" ниже.
    failure.trap(HTTPError)
    # Распечатываем само исключение
    print "An error occured: %s", failure


# Теперь функция выполняется асинхронно и вместо непосредственного

# результата мы получаем Deferred

deferred = downloadPage(url)

# Навешиваем на Deferred-объект обработчики успешных результатов

# и ошибок (callback, errback).

deferred.addCallback(printContents)

deferred.addErrback(handleError)

На практике обычно мы возвращаем Deferred из функций, которые получают

Deferred в процессе своей работы, навешиваем большое количество обработчиков,

обрабатываем исключения, некоторые исключения возвращаем через Deferred (выбрасываем

наверх). В качестве более сложного примера приведем код в асинхронном

варианте для примера атомарного счетчика из статьи про структуры данных в memcached,

здесь мы предполагаем, что доступ к memcached как сетевому сервису идет через Deferred, т.е.

методы класса Memcache возвращают Deferred (который вернет либо результат операции, либо ошибку):

class MCCounter(MemcacheObject):
    def __init__(self, mc, name):
        """
        Конструктор.

        @param name: имя счетчика
        @type name: C{str}
        """
        super(MCCounter, self).__init__(mc)
        self.key = 'counter' + name

    def increment(self, value=1):
        """
        Увеличить значение счетчика на указанное значение.

        @param value: инкремент
        @type value: C{int}
        @return: Deferred, результат операции
        """
        def tryAdd(failure):
            # Обрабатываем только KeyError, всё остальное "вывалится"
            # ниже
            failure.trap(KeyError)

            # Пытаемся создать ключ, если раз его еще нет
            d = self.mc.add(self.key, value, 0)
            # Если вдруг кто-то еще создаст ключ раньше нас,
            # мы это обработаем
            d.addErrback(tryIncr)
            # Возвращаем Deferred, он "вклеивается" в цепочку
            # Deferred, в контексте которого мы выполняемся
            return d

        def tryIncr(failure):
            # Всё аналогично функции tryAdd
            failure.trap(KeyError)

            d = self.mc.incr(self.key, value)
            d.addErrback(tryAdd)
            return d

        # Пытаемся выполнить инкремент, получаем Deferred
        d = self.mc.incr(self.key, value)
        # Обрабатываем ошибку
        d.addErrback(tryAdd)
        # Возвращаем Deferred вызывающему коду, он может тем самым:
        #  а) узнать, когда операция действительно завершится
        #  б) обработать необработанные нами ошибки (например, разрыв соединения)
        return d

    def value(self):
        """
        Получить значение счетчика.

        @return: текущее значение счетчика
        @rtype: C{int}
        @return: Deferred, значение счетчика
        """
        def handleKeyError(failure):
            # Обрабатываем только KeyError
            failure.trap(KeyError)

            # Ключа нет - возвращаем 0, он станет результатом
            # вышележащего Deferred
            return 0

        # Пытаемся получить значение ключа
        d = self.mc.get(self.key)
        # Будем обрабатывать ошибку отсутствия ключа
        d.addErrback(handleKeyError)
        # Возвращаем Deferred, наверх там можно будет повеситься
        # на его callback и получить искомое значение счетчика
        return d

На практике приведенный выше код можно написать "короче", объединяя часто используемые операции, например:

    return self.mc.get(self.key).addErrback(handleKeyError)

Практически для каждой конструкции синхронного кода можно найти аналог в асинхронной концепции с Deferred:

• последовательности синхронных операторов соответствует цепочка callback с асинхронными вызовами;
• вызову одной подпрграммы с вводом-выводом из другой соответствует возврат Deferred из Deferred (ветвление Deferred);
• глубокой цепочки вложенности, распространению исключений по стеку соответствует цепочка функций, возвращающие друг другу Deferred;
• блокам try..except соответствуют обработчики ошибок (errback), которые могут "пробрасывать" исключения дальше, любое исключение в callback переводит выполнение в errback;
• для "параллельного" выполнения асинхронных операций есть DeferredList.

Нити часто применяются в асинхронных программах для осуществления вычислительных процедур, осуществления блокирующегося

ввода-вывода (когда не существует асинхронного аналога). Всё это легко моделируется в простой модели 'worker', тогда

нет необходимости при грамотной архитектуре в явной синхронизации, при этом всё элегантно включается в общий поток

вычислений с помощью Deferred:

def doCalculation(a, b):
    """
    В этой функции осуществляются вычисления, синхронные операции ввода-вывода,
    не затрагивающие основной поток.
    """

    return a/b


def printResult(result):
    print result


def handleDivisionByZero(failure):
    failure.trap(ZeroDivisionError)

    print "Ooops! Division by zero!"


deferToThread(doCalculation, 3, 2).addCallback(printResult).addCallback(
    lambda _: deferToThread(doCalculation, 3, 0).addErrback(handleDivisionByZero))

В приведенном выше примере функция deferToThread переносит выполнение указанной функции в отдельную нить и возвращает

Deferred, через который будет асинхронно получен результат выполнения функции или исключение, если они будет выброшено.

Первое деление (3/2) выполняется в отдельной нити, затем распечатывается его результат на экран, а затем запускается

еще одно вычисление (3/0), которое генерирует исключение, обрабатываемое функцией handleDivisionByZero.

В одной статье не описать и части того, что хотелось бы сказать о Deferred, мне удалось не написать ни слова о том,

как же они работают. Если успел заинтересовать - читайте материалы ниже, а я обещаю написать еще.

Дополнительные материалы

• Документация Twisted Framework:
• Основы асинхронного программирования
• Использование Deferred, Откуда берутся Deferred?, Детальное описание Deferred
• Deferred в других языках программирования:
• JavaScript: Использование Deferred в JavaScript, Deferred в qooxdoo, Deferred в Dojo, Deferred в MochiKit
• C++: 1, 2, 3
• ...
• Александр Бурцев о Twisted