Программирование иногда не даёт нормально спать.

Придёт в голову какая-нибудь идея, присмотришься к ней — вроде простая и красивая. Потом начинаешь прикидывать подробности, и вроде тоже всё складывается. А потом садишься писать код... Тут-то и получается, что в реальности есть ещё и другие подробности, которые мозг при умозрительном разглядывании удобно опускал. Компилятор же, не обладающей такой моральной гибкостью, упрямо твердит, что нет, оно так не работает. И так не работает. И вот так тоже.

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

У меня так было на прошедших выходных. Увидел комментарий от Yuppy про использование yield для асинхронных вызовов, вспомнил, что когда-то где-то уже видел такую штуку (может как раз в том же Twisted), и на свою голову решил прикинуть в подробностях, как оно работает. Понятно, что можно просто пойти посмотреть в код Twisted и в код Yuppy (что я, в итоге, тоже делал), но хотелось именно построить всё руками, чтобы лучше осознать.

В итоге, промучавшись несколько дней и три раза всё переписав, я это осилил. И теперь у меня два положительных результата: подробное описание принципа работы и небольшая библиотечка, которую можно использовать для таких вещей, независимо от каких-либо фреймворков (ссылка "скачать" в конце).

Чего хочется

В общем и целом, хочется писать асинхронный код без callback'ов. Потому что они плохо читаются, потому что выстраиваются в цепочки в неестественном порядке, потому что заставляют передавать параметры через глобальные значения и т.д. Хочется же, чтобы было как-то так:

def get_url(url):
    # ... асинхронный запрос по HTTP

def process_url(url):
    result = yield get_url(url) # здесь process_url теряет управление
    print result                # сюда управление возвращается после завершения
                                # асинхронного вызова

process_url('http://...')
print 'This line runs before get_url completes'

Есть функция get_url, которая делает асинхронный запрос по HTTP. Это значит, что при её вызове она не ждёт завершения запроса, а возвращает управление вызвавшему сразу. А функция process_url, которая вызывает get_url, хочет получить управление тогда, когда асинхронный вызов завершился, не раньше. Принципиальную возможность так сделать даёт yield, который для функции process_data работает как return — она возвращает значение куда-то во Внешний Эфир. А начиная с версии Питона 2.5 этот Внешний Эфир может вернуть какое-то значение обратно внутрь process_data. Соответственно наша задача состоит в том, чтобы этот Внешний Эфир реализовать так, чтобы:

• вызов process_url не блокировал дальнейшее выполнение программы ('This line runs before...')
• в process_url не надо было писать никаких callback'ов: неявно callback'ом является кусок кода после yield
• get_url не была сильно ограниченной в реализации: запускает ли она тред, делает ли асинхронный IO-вызов сама, передаёт ли эту работу в какой-то фреймворк (Twisted, Tornado)

Начал я как раз с последнего. Поскольку get_url возвращается быстрее, чем делает собственную работу, единственный способ для неё что-то сообщить внешнему миру — это принимать callback, который потом вызывать. Чтобы говорить конкретней, давайте эту функцию реально напишем, сделав асинхронность например с помощью треда (хотя это не принципиально):

def get_url(url, callback):
    def read():
        s = urllib2.urlopen(url).read()
        callback(s)
    threading.Thread(target=read).start()

Однако, когда мы зовём её из process_url, та не может (да и не хочет) передать ей никакого явного callback'а — у ней его нет. Следовательно нам нужно как-то задекорировать get_url. Я назвал этот декоратор "async":

def async(func):
    def wrapper(*args, **kwargs):
        def caller(callback):
            return func(callback=callback, *args, **kwargs)
        return caller
    return wrapper

Он подменяет клиентскую функцию на свой wrapper, который не вызывает функцию сразу. Вместо этого он делает её ленивой: отдаёт наружу caller, который сможет позже вызвать func, передав ей callback. Работать это должно так:

1. Функция process_url вызывает yield get_url(url), где get_url задекорирована в async.
2. Значение под yield — это caller, который знает, как вызвать get_url(url).
3. Через yield этот объект попадает во Внешний Эфир, который тут же вызывает caller, передавая ему какой-то конкретный callback.

Здесь уже должно быть понятно, что callback Внешнего Эфира должен возвращать управление обратно в генератор process_data, пропихивая ему то значение, которое получит по завершению асинхронного вызова. Внешний эфир реализуется небольшим объектом и навешивается на функцию process_data опять же декоратором с названием "process":

class CallbackDispatcher(object):
    def __init__(self, generator):
        self.g = generator
        caller = self.g.next()
        caller(callback=self.callback)

    def callback(self, arg):
        try:
            caller = self.g.send(arg)
            caller(callback=self.callback)
        except StopIteration:
            pass

def process(func):
    def wrapper(*args, **kwargs):
        CallbackDispatcher(func(*args, **kwargs))
    return wrapper

Последовательность работы:

1. Декоратор process нужен только для того, чтобы вызвать клиентскую функцию, и передать полученный генератор в CallbackDispatcher.
2. CallbackDispatcher прямо в конструкторе запускает генератор через self.g.next()
3. В process_data при этом выполняется кусок кода до первого yield, и через него возвращается значение — caller для get_url.
4. Этот caller приезжает в CallbackDispatcher в качестве возвратного значения next'а.
5. caller тут же вызывается с callback'ом — методом самого CallbackDispatcher.
6. На это моменте заканчивается конструктор диспетчера, а следовательно и работа декоратора process — соответственно программа тут же продолжается дальше на print 'This line runs ...'.
7. Тем временем где-то в get_url происходит асинхронный HTTP-вызов, заканчивается, и она вызывает CallbackDispatcher.callback с каким-то результатом своей работы.
8. CallbackDispatcher.callback пропихивает результат обратно в генератор process_data, тем самым запуская его следующий кусок.

И вот, собственно, теперь у нас есть базовая функциональность:

• Обрабатывать результаты асинхронных вызовов можно последовательно, сколько угодно раз в функции. Для этого её достаточно завернуть в @process и вызывать асинхронные функции через yield.
• Асинхронная функция должна быть завёрнута в @async, должна принимать параметр callback и возвращать в него одно значение (если надо несколько, их можно упаковать в tuple например).
• Это неочевидно, но async-функции сами могут быть process'ами и вызывать из себя другие async-функции. Единственное, при навешивании обоих декораторов async должен быть внешним.

Бонус: мультивызовы

Конечно я слукавил, когда сказал, что это всё было чисто теоретическим упражнением. На самом деле, у нас тут в Яндексе есть вполне конкретный юзкейс: небольшой HTTP-прокси, написанный на Tornado, который асинхронно обрабатывает входящие запросы и асинхронно же делает свои. Код там очень трудно читается из-за явных callback'ов, и всё это придумывалось для его упрощения.

Так вот там есть задачка, в которой надо запустить несколько асинхронных функций одновременно, дождаться их всех и список всех результатов обработать целиком (map-reduce, ага). Сейчас оно написано примерно так:

for item in items:
    async_fetch(item, callback)
results = []

def callback(item, result):
    items.remove(item)
    results.append(result)
    if not items:
        process_results(results)

Вместо такого вороха обслуживающего кода хочется иметь что-то такое:

results = yield map(async_fetch, items)
# обработка results

Чтобы так было, надо всего лишь научить Внешний Эфир (CallbackDispatcher) таким вещам:

• понимать, что через yield может прийти не только одинокий caller, но и их последовательность
• посчитать caller'ы и подготовить список для их результатов
• в callback передавать ещё и порядковый номер задачи, чтобы порядок результатов совпадал с порядком задач

Выглядит это вот так (__init__ опущен для краткости):

class CallbackDispatcher(object):
    def call(self, callers):

        # приведение параметра к последовательности
        single = not hasattr(callers, '__iter__')
        if single:
            callers = [callers]

        # запуск перенумерованных caller'ов с прилеплением
        # к callback'у номера задачи и признака одинокого результата
        for count, caller in enumerate(callers):
            caller(callback=partial(self.callback, count, single))

        # создание места под результаты
        self.call_count = count + 1
        self.results = [None] * self.call_count

    def callback(self, index, single, arg):

        # сохранение результата
        self.call_count -= 1
        self.results[index] = arg

        if self.call_count > 0: # возврат, если есть ещё кого ждать
            return

        # отсылка результата в генератор в виде списка или одного значения
        try:
            result = self.results[0] if single else self.results
            self.call(self.g.send(result))
        except StopIteration:
            pass

В примерах работает :-)

Код

• Проект на Launchpad: adisp
• Архив: adisp-0.1.zip
• Просмотр кода