Асинхронность в C# 5. Часть 3: Композиция
Как то в 6:45 утра я шел на остановку к своему автобусу. Прямо на углу 45-й улицы, молодой парень, без футболки, весь в крови промчался мимо меня. За ним гнался другой парень, размахивая бейсбольной битой. Я сразу же подумал: «Боже мой! Нужно немедленно вызвать полицию!»
Затем я увидел, что за парнем с битой гнался Граф Дракула, небольшая толпа зомби, группа пиратов, один средневековый рыцарь, а за ними по пятам гнался огромный шмель. Видимо на тех выходных кто-то проводил забег в честь Хеллоуина.
Еще новости: мы получили множество отличных комментариев по новой возможности языка async; пожалуйста, продолжайте в том же духе. Они все будут прочитаны; на их систематизацию и оценку уйдут недели, за что я сразу хочу извиниться, но именно так работает эта система.
Сегодня я хочу немного поговорить о композиции асинхронного кода и о том, почему он оказался таким сложным в CPS и насколько он упростился с “await” в C# 5.
Мне нужно дать этой штуке какое-то название. LINQ расшифровывается как Language Integrated Query. Давайте пока назовем эту возможность TAP , TaskAsynchronyPattern. Я уверен, что мы придумаем название получше; но помните, что это всего лишь прототип (*).
Пример получения и сохранения документов на ленту, который мы до этого рассматривали, был выбран специально. Как мы видели, при использовании стиля передачи продолжений, объединение даже двух методов может быть весьма сложной задачей. Сегодня я хочу поговорить о композиции асинхронных методов. Наш метод ArchiveDocuments возвращал void, что сильно упрощает задачу. Давайте предположим, что метод ArchiveDocument возвращает некоторое значение, например, суммарное количество байт, сохраненных на ленту. Синхронная версия элементарна:
long ArchiveDocuments(List<Url> urls)
{
long count = 0;
for(int i = 0; i < urls.Count; ++i)
{
var document = Fetch(urls[i]);
count += document.Length;
Archive(document);
}
return count;
}
А теперь давайте подумаем, как мы можем переписать ее в асинхронном виде. Если метод ArchiveDocuments должен немедленно вернуть управление после первого вызова FetchAsync, и продолжить выполнение только после завершения получения первого документа, тогда, когда же будет выполнено “return count”? Простая асинхронная версия метода ArchiveDocments не может возвращать количество сохраненных байтов; она также должна быть переписана в стиле CPS:
void ArchiveDocumentsAsync(List<Url> urls, Action<long> continuation)
{
// Каким-то образом выполнить сохранение документов на ленту асинхронно,
// а затем вызвать продолжение
}
И пошло поехало. Теперь вызывающий код метода ArchiveDocmentsAsync должен быть переписан в CPS, чтобы его продолжение можно было передать в этот метод. А что если этот код возвращает значение? Это приведет к беспределу; так всю программу придется переписать и вывернуть наизнанку.
В модели TAP, мы говорим, что тип Task<T> является типом, который представляет результаты выполнения асинхронной операции. В C# 5 вы можете просто написать:
async Task<long> ArchiveDocumentsAsync(List<Url> urls)
{
long count = 0;
Task archive = null;
for(int i = 0; i < urls.Count; ++i)
{
var document = await FetchAsync(urls[i]);
count += document.Length;
if (archive != null)
await archive;
archive = ArchiveAsync(document);
}
return count;
}
И компилятор позаботится обо всем остальном за вас. Он в точности знает, что здесь происходит и преобразует этот код во что-то такое:
Task<long> ArchiveDocuments(List<Url> urls)
{
var taskBuilder = AsyncMethodBuilder<long>.Create();
State state = State.Start;
TaskAwaiter<Document> fetchAwaiter = null;
TaskAwaiter archiveAwaiter = null;
int i;
long count = 0;
Task archive = null;
Document document;
Action archiveDocuments = () =>
{
switch(state)
{
case State.Start: goto Start;
case State.AfterFetch: goto AfterFetch;
case State.AfterArchive: goto AfterArchive;
}
Start:
for(i = 0; i < urls.Count; ++i)
{
fetchAwaiter = FetchAsync(urls[i]).GetAwaiter();
state = State.AfterFetch;
if (fetchAwaiter.BeginAwait(archiveDocuments))
return;
AfterFetch:
document = fetchAwaiter.EndAwait();
count += document.Length;
if (archive != null)
{
archiveAwaiter = archive.GetAwaiter();
state = State.AfterArchive;
if (archiveAwaiter.BeginAwait(archiveDocuments))
return;
AfterArchive:
archiveAwaiter.EndAwait();
}
archive = ArchiveAsync(document);
}
taskBuilder.SetResult(count);
return;
};
archiveDocuments();
return taskBuilder.Task;
}
(Обратите внимание, что у нас все еще есть проблема, поскольку метки находятся вне области видимости. Помните, что компилятор C# не должен следовать всем правилам при генерации кода; давайте просто представим, что метки находятся в нужной области видимости. И обратите также внимание на отсутствие обработки исключений. Как я уже писал ранее, исключения вносят дополнительную сложность, поскольку требуют *два* продолжения: нормальное продолжение и продолжение в случае возникновения ошибки. Как мы с этим справляемся? Об этом я расскажу в другой раз.)
Давайте удостоверимся, что поток управления понятен. Вначале давайте рассмотрим простой случай: список пуст. Что тогда произойдет? Мы создадим объект taskBuilder. Мы создадим делегат типа Action. Затем мы вызовем его синхронно. Он проинициализирует внешнюю переменную “count” нулем, перейдет к метке ”Start”, пропустит цикл, скажет объекту taskBuilder о том, что получен результат и вернет управление. Выполнение делегата завершено. У объекта taskBuilder будет запрошена задача; он знает, что работа уже выполнена, поэтому он вернет завершенную задачу, которая просто содержит 0.
Если вызывающий код попытается дождаться завершения задачи, тогда при попытке начать асинхронную операцию awaiter вернет false, поскольку задача уже завершена. Если вызывающий код не будет ждать завершения выполнения задачи, тогда …, что ж, он может делать все, что захочет с объектом класса Task. В конце концов, он запросит результат выполнения задачи или проигнорирует его, если он не интересен.
Теперь давайте рассмотрим более сложный случай; существует несколько документов. Мы, опять-таки, создаем taskBuilder и делегат, который вызываем синхронно. Проходим первый раз по циклу, начинаем получение документов асинхронно, устанавливаем делегат в качестве продолжения и выходим из делегата. К этому моменту мы создали задачу, которая представляет собой «асинхронное выполнение тела метода ArchiveDocumentAsync». После асинхронного завершения задачи получения документа и вызова продолжения, снова вызывается делегат «с того момента, где он был приостановлен», благодаря магии конечного автомата. Выполнение продолжается, как и в случае возвращения void; единственное отличие заключается в том, что задача Task<long> метода ArchiveDocumentsAsync скажет о своем завершении (путем вызова продолжения), только после вызова метода taskBuilder.SetResult внутри делегата.
Разумно?
Прежде чем продолжить рассуждения о композиции задач, небольшое замечание о расширяемости TAP. Мы сделали LINQ очень расширяемым; любой тип, реализующий методы Select, Where и др., или содержащий такие методы расширения, может быть использован в выражениях запросов (query comprehensions). С TAP мы поступили аналогично: любой тип, содержащий метод GetAwaiter, который в свою очередь содержит методы BeginAwait, EndAwait и др., может быть использован в выражениях “await”. Однако методы с модификатором async могут возвращать только void, Task или Task<T>. Мы прилагаем все силы для расширяемости использования существующего асинхронного кода, но у нас нет ни малейшего желания позволять создавать асинхронные методы с экзотическими типами. Внимательный читатель может заметить, что я не обсуждаю вопросы расширяемости построителя задач (task builder). Позднее я расскажу о том, откуда появился построитель задач.
Продолжаем: (ха-ха-ха)
В LINQ, бывают ситуации, когда более естественно использовать оператор “where” языка, а иногда использовать fluent-синтаксис типа “Where(c=>…)”. Ситуация с TAP аналогична: наша цель – создать синтаксис языка, который позволит использовать несколько асинхронных задач, однако иногда вам может понадобиться более сложная логика «объединения». Для этого будут предназначены методы вроде “WhenAll” “WhenAny” для объединения задач следующим образом:
List<List<Url>> groupsOfUrls = whatever;
Task<long[]> allResults = Task.WhenAll(from urls in groupsOfUrls select ArchiveDocumentsAsync(urls));
long[] results = await allResults;
Что делает этот код? Итак, метод ArchiveDocumentdsAsync возвращает Task<long>, таким образом, выражение запроса возвращает IEnumerable<Task<long>>. Метод WhenAll принимает на вход последовательность задач и возвращает новую задачу, которая асинхронно ожидает завершение каждой из задач, помещает результаты в массив, и вызывает свое продолжение, когда результат будет доступен.
Метод WhenAny, аналогично, принимает последовательность задач и возвращает новую задачу, которая вызывает свое продолжение с первым результатом, когда одна из задач завершится. (Интересный вопрос заключается в том, что произойдет, если первая задача завершится успешно, а все остальные сгенерируют исключения, но давайте поговорим об этом позднее.)
Кроме этих методов будут и другие функции комбинирования заданий и вспомогательные методы; см. примеры в CTP. Обратите внимание, что в CTP мы не могли модифицировать класс Task; вместо этого мы добавили эту функциональность в класс TaskEx. В конечном выпуске мы наверняка перенесем эту функциональность в класс Task.
В следующий раз: нет, серьезно, асинхронность не требует многопоточности.
(*) Я подчеркиваю, что это название временное и предназначено исключительно для обсуждения, а не официальное название. Пожалуйста, не называйте свою книгу «Основы TAP» или «Выучить TAP за 21 день» или как-то так. У меня на полке есть книга «Instant DHTML Scriplets»; Дино Эспозито (Dino Esposito) пишет настолько быстро, что он успел выпустить книгу после того, как мы сказали ему кодовое имя продукта и до того, как мы объявили его настоящее имя (настоящее имя было таким: «Windows Script Components»).