Як async/await дійсно працює в C#. Частина 1

Підписуйтеся на Telegram-канал «DOU #tech», щоб не пропустити нові технічні статті

Вітаю, мене звати Юрій Рожков. Займаюсь автоматизацією тестування, розробкою тестових фреймворків, по можливості закриваю задачі девелоперів і налаштовую інфраструктуру. Так сталось, що крайні 7 років мій техстек роботи — це .NET/C# і все, з ним пов’язане. Нещодавно зіткнувся з цікавою статтею про async/await, про що й буде далі мова. Стаття є перекладом, оригінал знаходиться тут.

Нещодавно .NET Blog опублікували статтю Що таке .NET, і чому вам варто його обрати? (‘What is .NET, and why should you choose it?’). В ній надається поверхневий огляд платформи, різних компонентів і рішень дизайну, з подальшим поглибленим вивченням тем. Ця стаття є першою з циклу, з детальним оглядом історії, прийнятих рішень та імплементації async/await в C# та .NET.

Підтримка async/await існує вже понад 10 років. На сьогодні це трансформувалось в те, як пишеться масштабований код на .NET, і доволі життєво стало користуватись функціоналом без розуміння, що відбувається «під капотом». Почнемо з синхронного методу («синхронний» тому, що при його виклику нічого не можливо зробити, поки вся операція не завершиться і контроль не перейде до основної програми):

// Synchronously copy all data from source to destination.
public void CopyStreamToStream(Stream source, Stream destination)
{
    var buffer = new byte[0x1000];
    int numRead;
    while ((numRead = source.Read(buffer, 0, buffer.Length)) != 0)
    {
        destination.Write(buffer, 0, numRead);
    }
}

Додаємо пару фраз, міняємо назви методів, і ви зупинитесь на асинхронному методі («асинхронний» тому, що контроль досить швидко повертається до основної програми, ймовірно до того, як робота метода буде завершена):

// Asynchronously copy all data from source to destination.
public async Task CopyStreamToStreamAsync(Stream source, Stream destination)
{
    var buffer = new byte[0x1000];
    int numRead;
    while ((numRead = await source.ReadAsync(buffer, 0, buffer.Length)) != 0)
    {
        await destination.WriteAsync(buffer, 0, numRead);
    }
}

Майже ідентичний синтаксис, ті ж самі конструкції, але тепер неблокуюча поведінка, зі значно різною моделлю виконання, і з величезною зміною у роботі C# компілятора та бібліотек ядра.

Хоча достатньо поширене використання чогось без знання роботи «під капотом», твердо вірю, що розуміння того, як це щось дійсно працює, допоможе зробити речі краще. Для async/await зокрема, розуміння механізмів дійсно корисне, коли ви дебажите проблемні місця або намагаєтесь покращити перформанс. Для цього нам треба повернутись в часі назад, до появи async/await.

На початку...

В .NET Framework 1.0 був патерн асинхронної програмної моделі (APM), інакше відомий як Begin/End pattern, або IAsyncResult патерн. Поверхнево, патерн простий. Для синхронної операції DoStuff:

class Handler
{
    public int DoStuff(string arg);
}

буде 2 відповідних метода BeginDoStuff, EndDoStuff:

class Handler
{
    public int DoStuff(string arg);

    public IAsyncResult BeginDoStuff(string arg, AsyncCallback? callback, object? state);
    public int EndDoStuff(IAsyncResult asyncResult);
}

BeginDoStuff приймає ті ж параметри що і DoStuff + делегат AsyncCallback та об’єкт стану object, які можуть бути null. Begin метод відповідає за початок асинхронної дії, при наявному callback параметру (продовження основної операції) також відповідає за його виклик після закінчення асинхронної операції. Begin метод також будує інстанс інтерфейсу IAsyncResult з опціональним полем стану AsyncState:

namespace System
{
    public interface IAsyncResult
    {
        object? AsyncState { get; }
        WaitHandle AsyncWaitHandle { get; }
        bool IsCompleted { get; }
        bool CompletedSynchronously { get; }
    }

    public delegate void AsyncCallback(IAsyncResult ar);
}

IAsyncResult інстанс буде повернутий з Begin методу і переданий в AsyncCallback, коли його буде врешті викликано. При готовності обробити результати операції програма передає інстанс IAsyncResult в End метод, котрий відповідає за перевірку завершення операції (синхронно очікуючи), і наприкінці повертає будь-який результат, в т.ч. будь-які помилки/ ексепшени. Таким чином, замість написання синхронного коду:

try
{
    int i = handler.DoStuff(arg); 
    Use(i);
}
catch (Exception e)
{
    ... // handle exceptions from DoStuff and Use
}

Begin/End методи можуть бути використані для тієї ж операції асинхронно:

try
{
    handler.BeginDoStuff(arg, iar =>
    {
        try
        {
            Handler handler = (Handler)iar.AsyncState!;
            int i = handler.EndDoStuff(iar);
            Use(i);
        }
        catch (Exception e2)
        {
            ... // handle exceptions from EndDoStuff and Use
        }
    }, handler);
}
catch (Exception e)
{
    ... // handle exceptions thrown from the synchronous call to BeginDoStuff
}

Для всіх, хто мав справу з callback-based API, це має бути знайомим.

Далі все стає складніше. Наприклад, проблема «занурення у стек» (stack dives). Це коли код повторно робить виклики, що ведуть далі і далі по стеку до моменту потенційного переповнення стека (stack overflow). Begin викликає callback синхронно, тільки якщо операція завершена синхронно, виклик Begin може сам по собі викликати callback.

Асинхронні операції що завершуються синхронно, доволі поширені; для них асинхронне завершення скоріше дозволено, ніж гарантовано. Уявіть асинхронне читання операції з мережі, наприклад з сокета. Якщо вам потрібна невелика кількість даних, таких як читання заголовків (header data) з респонсу, ви можете додати буфер, щоб уникнути частих системних викликів.

Замість читання невеликих об’ємів необхідних даних, виконується читання великого об’єму в буфер і читання з нього, допоки він не буде вичерпаний; це дозволяє скоротити кількість важких викликів по сокету. Буфер може бути за будь-якою асинхронною абстракцією, так перша «асинхронна» операція (заповнення буфера) завершується асинхронно, але потім всі наступні операції, поки буфер не буде вичерпаний, не потрібні, і можуть бути завершені синхронно. Коли Begin метод виконує одну з таких операцій, і вона виявляється завершеною синхронно, він може викликати колбек синхронно.

Це значить, що є 1 стек фрейм (stack frame) з назвою метода Begin, інший стек фрейм для самого метода, і ще один — для колбеку. Що трапиться, якщо колбек викликає Begin знову? Якщо операція завершується синхронно і її колбек був викликаний синхронно, маємо на декілька стек-фреймів більше. І так, допоки не стикнемось з переповненням.

Це доволі легко репродьюсити. Спробуйте наступне:

using System.Net;
using System.Net.Sockets;

using Socket listener = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp);
listener.Bind(new IPEndPoint(IPAddress.Loopback, 0));
listener.Listen();

using Socket client = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp);
client.Connect(listener.LocalEndPoint!);

using Socket server = listener.Accept();
_ = server.SendAsync(new byte[100_000]);

var mres = new ManualResetEventSlim();
byte[] buffer = new byte[1];

var stream = new NetworkStream(client);

void ReadAgain()
{
    stream.BeginRead(buffer, 0, 1, iar =>
    {
        if (stream.EndRead(iar) != 0)
        {
            ReadAgain(); // uh oh!
        }
        else
        {
            mres.Set();
        }
    }, null);
};
ReadAgain();
mres.Wait();

Тут я засетапив прості клієнт і сервер для сокета, підключені між собою. Сервер відправляє 100,000 байтів клієнту, котрі обробляються «асинхронно» в методах BeginRead/EndRead по одному (що вкрай неефективно). В методі BeginRead колбек закінчує роботу викликом метода EndRead, якщо байт даних прочитаний (що не кінець стріму), він викликає BeginRead ще раз рекурсивно в локальній функції ReadAgain.

Хоча в .NET Core операції з сокетами відпрацьовують набагато швидше, ніж це було в .NET Framework, і вони будуть завершені синхронно, якщо це підтримується ядром ОС, отримаємо переповнення стеку:

Є 2 шляхи виправити це:

1. Не дозволяти викликати AsyncCallback синхронно. Якщо він асинхронний, не зважаючи на синхронну операцію, ризик переповнення стеку зникає, але так само просідає перформанс, додавання колбеку в чергу накладає свої витрати.
2. Задіяти механізм, при якому основна операція замість колбеку виконує роботу по завершенню синхронно. В такому випадку отримаємо ще один фрейм у стеку.

APM слідує пункту 2. IAsyncResult інтерфейс прокидає IsCompleted та CompletedSynchronously. IsCompleted вказує, чи завершена операція; можна перевіряти його декілька разів, врешті значення перейде з false на true. На відмінність від нього, CompletedSynchronously ніколи не змінює стан (або це дуже брудний баг); він використовується для комунікації між основним методом Begin та AsyncCallback, котрий відповідає за роботу по завершенню.

Якщо CompletedSynchronously = false, тоді операція завершується асинхронно і вся робота після цього має бути залишена для колбеку; після всього, якщо робота не завершена синхронно, виклик методу Begin не може обробити його, тому що невідомо, чи завершена операція (і якщо був викликаний метод End, він блокує поки операція не завершена).

Якщо CompletedSynchronously = true, і якщо колбек обробляє роботу після завершення, то існує ризик занурення у стек. Таким чином, будь-яка імплементація у зв’язку з проблемою «stack dives» має перевіряти CompletedSynchronously і виклик методу Begin має виконувати роботу після завершення, тільки якщо воно = true, тобто колбек не має виконувати роботу після завершення. Тому CompletedSynchronously ніколи не має змінюватись: основний виклик і колбек мають бачити одне й теж саме значення для впевненості, що робота по завершенню виконана один і тільки один раз, незважаючи на race conditions.

Наш попередній приклад DoStuff стає:

try
{
    IAsyncResult ar = handler.BeginDoStuff(arg, iar =>
    {
        if (!iar.CompletedSynchronously)
        {
            try
            {
                Handler handler = (Handler)iar.AsyncState!;
                int i = handler.EndDoStuff(iar);
                Use(i);
            }
            catch (Exception e2)
            {
                ... // handle exceptions from EndDoStuff and Use
            }
        }
    }, handler);
    if (ar.CompletedSynchronously)
    {
        int i = handler.EndDoStuff(ar);
        Use(i);
    }
}
catch (Exception e)
{
    ... // handle exceptions that emerge synchronously from BeginDoStuff and possibly EndDoStuff/Use
}

Без коментарів. Поки що ми тільки розглянули використання патерну, не імплементацію... Хоча більшості девелоперів не потрібно перейматись реалізацією таких операцій як Socket.BeginReceive/EndReceive, багатьом з них необхідно думати про композицію таких операцій (виконання декількох асинхронних операцій), що означає не тільки використання інших Begin/End методів, але й їхня власна імплементація. Ви могли помітити відсутність контролю виконання в попередньому прикладі DoStuff. Додайте до цього пару простих операцій, як цикли, і це стає зоною страждань, або спробою автора блогу довести свою точку зору.

Давайте для повної картини реалізуємо весь приклад. На початку, я показав метод CopyStreamToStream, котрий копіює всі дані з одного стріма в інший (а-ля Stream.CopyTo, але заради експерименту уявимо, що його нема):

public void CopyStreamToStream(Stream source, Stream destination)
{
    var buffer = new byte[0x1000];
    int numRead;
    while ((numRead = source.Read(buffer, 0, buffer.Length)) != 0)
    {
        destination.Write(buffer, 0, numRead);
    }
}

По кроках: ми поступово читаємо з одного стріма і пишемо результат в інший, читаємо-пишемо, і так далі, поки більше не маємо даних для читання. Як нам реалізувати це асинхронно через АРМ паттерн? Щось подібне:

public IAsyncResult BeginCopyStreamToStream(
    Stream source, Stream destination,
    AsyncCallback callback, object state)
{
    var ar = new MyAsyncResult(state);
    var buffer = new byte[0x1000];

    Action<IAsyncResult?> readWriteLoop = null!;
    readWriteLoop = iar =>
    {
        try
        {
            for (bool isRead = iar == null; ; isRead = !isRead)
            {
                if (isRead)
                {
                    iar = source.BeginRead(buffer, 0, buffer.Length, static readResult =>
                    {
                        if (!readResult.CompletedSynchronously)
                        {
                            ((Action<IAsyncResult?>)readResult.AsyncState!)(readResult);
                        }
                    }, readWriteLoop);

                    if (!iar.CompletedSynchronously)
                    {
                        return;
                    }
                }
                else
                {
                    int numRead = source.EndRead(iar!);
                    if (numRead == 0)
                    {
                        ar.Complete(null);
                        callback?.Invoke(ar);
                        return;
                    }

                    iar = destination.BeginWrite(buffer, 0, numRead, writeResult =>
                    {
                        if (!writeResult.CompletedSynchronously)
                        {
                            try
                            {
                                destination.EndWrite(writeResult);
                                readWriteLoop(null);
                            }
                            catch (Exception e2)
                            {
                                ar.Complete(e);
                                callback?.Invoke(ar);
                            }
                        }
                    }, null);

                    if (!iar.CompletedSynchronously)
                    {
                        return;
                    }

                    destination.EndWrite(iar);
                }
            }
        }
        catch (Exception e)
        {
            ar.Complete(e);
            callback?.Invoke(ar);
        }
    };

    readWriteLoop(null);

    return ar;
}

public void EndCopyStreamToStream(IAsyncResult asyncResult)
{
    if (asyncResult is not MyAsyncResult ar)
    {
        throw new ArgumentException(null, nameof(asyncResult));
    }

    ar.Wait();
}

private sealed class MyAsyncResult : IAsyncResult
{
    private bool _completed;
    private int _completedSynchronously;
    private ManualResetEvent? _event;
    private Exception? _error;

    public MyAsyncResult(object? state) => AsyncState = state;

    public object? AsyncState { get; }

    public void Complete(Exception? error)
    {
        lock (this)
        {
            _completed = true;
            _error = error;
            _event?.Set();
        }
    }

    public void Wait()
    {
        WaitHandle? h = null;
        lock (this)
        {
            if (_completed)
            {
                if (_error is not null)
                {
                    throw _error;
                }
                return;
            }

            h = _event ??= new ManualResetEvent(false);
        }

        h.WaitOne();
        if (_error is not null)
        {
            throw _error;
        }
    }

    public WaitHandle AsyncWaitHandle
    {
        get
        {
            lock (this)
            {
                return _event ??= new ManualResetEvent(_completed);
            }
        }
    }

    public bool CompletedSynchronously
    {
        get
        {
            lock (this)
            {
                if (_completedSynchronously == 0)
                {
                    _completedSynchronously = _completed ? 1 : -1;
                }

                return _completedSynchronously == 1;
            }
        }
    }

    public bool IsCompleted
    {
        get
        {
            lock (this)
            {
                return _completed;
            }
        }
    }
}

Трясця! І навіть ця абракадабра — не найкраща реалізація. Наприклад, імплементація IAsyncResult блокує кожну операцію, замість того, щоб працювати без блокувань де це можливо, Exception передається в чистому вигляді замість ExceptionDispatchInfo, котрий дозволяє накопичувати стек викликів, багато розподілу пам’яті в кожній окремій операції (напр. виділений делегат на кожен виклик BeginWrite) тощо. Уявіть все це для кожного реюзабельного метода з використанням асинхронних операцій. А також комбінації з декількома дискретними IAsyncResults (згадаємо Task.WhenAll). Кожна операція реалізує своє специфічне АПІ, що означає відсутність спільної мови між ними (хоча деякі розробники намагались полегшити цей тягар через окремий шар колбеків).

Вся ця складність означає, що дехто намагався це робити, і скоріш за все стикався з лютими багами. По правді кажучи, це не критика АРМ патерна, а скоріше асинхронності на колбеках в цілому.

Ми так звикли до міці і простоти в сучасних мовах програмування що контролює потік виконання, що підходи на колбеках зазвичай стикаються з такими складними конструкціями. Жодна інша популярна мова не мала кращої альтернативи.

Нам було потрібне інше рішення, з урахуванням, що було зроблено правильно і неправильно. Цікаво, що АРМ патера це тільки патерн; ані runtime, ані бібліотеки ядра або компілятор не надають підтримки в його використанні або реалізації.

Патерн на асинхронних подіях (Event-Based Asynchronous Pattern)

.NET Framework 2.0 надав пару АПІ з інших підходом до обробки асинхронних операцій, один з яких призначений для роботи в контексті клієнтських застосунків. Це Event-based Asynchronous Pattern, або EAP, з методом для початку асинхронної операції, і івентом для очікування її завершення. Таким чином, наш попередній приклад з DoStuff стає:

class Handler
{
    public int DoStuff(string arg);

    public void DoStuffAsync(string arg, object? userToken);
    public event DoStuffEventHandler? DoStuffCompleted;
}

public delegate void DoStuffEventHandler(object sender, DoStuffEventArgs e);

public class DoStuffEventArgs : AsyncCompletedEventArgs
{
    public DoStuffEventArgs(int result, Exception? error, bool canceled, object? userToken) :
        base(error, canceled, usertoken) => Result = result;

    public int Result { get; }
}

Робота по завершенню реєструється в івенті DoStuffCompleted, потім викликається метод DoStuffAsync; він стартує операцію, і до її завершення DoStuffCompleted буде асинхронно викликатись з основного метода. Потім хендлер може запустити роботу по завершенню, напр. валідацію userToken, при цьому дозволяючи запуск декількох хендлерів одночасно.

Цей паттерн спрощує одні і суттєво ускладнює інші кейси (і попередній приклад з АРМ CopyStreamToStream). Він не отримав широкого розповсюдження, з’явився в одному релізі .NET Framework і залишив після себе АПІ, наприклад, Ping.SendAsync/Ping.PingCompleted:

public class Ping : Component
{
    public void SendAsync(string hostNameOrAddress, object? userToken);
    public event PingCompletedEventHandler? PingCompleted;
    ...
}

Однак він залишив одну помітну річ, якою ми користуємось дотепер: SynchronizationContext. Він був представлений в .NET Framework 2.0 як абстракція для загального планувальника. Зокрема, найбільш вживаний метод SynchronizationContext-у — метод Post, котрий додає в чергу будь-що, представлене планувальником. Наприклад, базова імплементація SynchronizationContext — ThreadPool, базова імплементація SynchronizationContext.Post — ThreadPool.QueueUserWorkItem, котрий викликає з ThreadPool колбек зі статусом одного з потоків пула. З усім тим, основа основ SynchronizationContext-у — скоріше в підтримці планувань у необхідному вигляді для різних застосунків.

Згадайте UI-фреймворк Windows Forms. Як і в більшості фреймворків на Windows, контроли асоційовані з відповідним потоком, і цей потік запускає повідомлення до відповідної роботи з цими контролами: тільки цей потік має керувати цими контролами, і будь-який інший потік має відправити повідомлення в купу UI потоків для обробки. У Windows Forms це легко робиться завдяки Control.BeginInvoke, котрий передає в чергу відповідний делегат з аргументами які асоціюють потік з цим контролом. Наприклад:

private void button1_Click(object sender, EventArgs e)
{
    ThreadPool.QueueUserWorkItem(_ =>
    {
        string message = ComputeMessage();
        button1.BeginInvoke(() =>
        {
            button1.Text = message;
        });
    });
}

Робота методу ComputeMessage() виконується в ThreadPool (задля збереження UI активним під час виконання потоку), і коли робота завершена, передається делегат назад в потік, пов’язаний з button1 для оновлення лейбла кнопки. Досить легко. WPF має щось аналогічне, з його типом Dispatcher:

private void button1_Click(object sender, RoutedEventArgs e)
{
    ThreadPool.QueueUserWorkItem(_ =>
    {
        string message = ComputeMessage();
        button1.Dispatcher.InvokeAsync(() =>
        {
            button1.Content = message;
        });
    });
}

І .NET MAUI має щось схоже. Далі, збережемо цю логіку в хелпер метод.

// Call ComputeMessage and then invoke the update action to update controls.
internal static void ComputeMessageAndInvokeUpdate(Action<string> update) { ... }

І використаємо його:

private void button1_Click(object sender, EventArgs e)
{
    ComputeMessageAndInvokeUpdate(message => button1.Text = message);
}

Постає питання: як використати ComputeMessageAndInvokeUpdate у всіх інших застосунках? Хардкодити його для кожного UI-фреймворку? Тут і з’являється SynchronizationContext. Додаємо метод:

internal static void ComputeMessageAndInvokeUpdate(Action<string> update)
{
    SynchronizationContext? sc = SynchronizationContext.Current;
    ThreadPool.QueueUserWorkItem(_ =>
    {
        string message = ComputeMessage();
        if (sc is not null)
        {
            sc.Post(_ => update(message), null);
        }
        else
        {
            update(message);
        }
    });
}

Він використовує SynchronizationContext, щоб вказати, як повернути планувальник до необхідного середовища для взаємодії з UI. Кожна модель реалізує свій тип, похідний від SynchronizationContext, наприклад, Windows Forms має такий тип:

public sealed class WindowsFormsSynchronizationContext : SynchronizationContext, IDisposable
{
    public override void Post(SendOrPostCallback d, object? state) =>
        _controlToSendTo?.BeginInvoke(d, new object?[] { state });
    ...
}

WPF :

public sealed class DispatcherSynchronizationContext : SynchronizationContext
{
    public override void Post(SendOrPostCallback d, Object state) =>
        _dispatcher.BeginInvoke(_priority, d, state);
    ...
}

ASP.NET раніше мав аналогічне і не займався запуском потоку, а скоріше забезпечував, щоб декілька потоків одночасно не мали доступу до HttpContext при серіалізації конкретного реквеста:

internal sealed class AspNetSynchronizationContext : AspNetSynchronizationContextBase
{ 
    public override void Post(SendOrPostCallback callback, Object state) =>
        _state.Helper.QueueAsynchronous(() => callback(state));
    ...
}

Цим не обмежується. Наприклад, популярний юніт-тест фреймворк xunit також реалізує декілька своїх SynchronizationContext. Завдяки цьому, ви можете запускати тести паралельно, але обмежити кількість одночасно запущених тестів:

public class MaxConcurrencySyncContext : SynchronizationContext, IDisposable
{
    public override void Post(SendOrPostCallback d, object? state)
    {
        var context = ExecutionContext.Capture();
        workQueue.Enqueue((d, state, context));
        workReady.Set();
    }
}

У даному кейсі Post метод тільки додає роботу у свою внутрішню чергу, котра запускається в своєму окремому потоці, де метод контролює, як багато одночасних потоків запускати.

Як це пов’язано з ЕАР патерном? Обидва (ЕАР та SynchronizationContext) були представлені одночасно, і ЕАР вказував, що завершення івентів має бути додано в чергу незалежно від того, який був SynchronizationContext, коли була розпочата асинхронна операція. Щоб трохи спростити це, в System.ComponentModel були введені допоміжні типи, зокрема AsyncOperation та AsyncOperationManager.

Перший був кортежем (тьюплом), обгорнутий в об’єкт стану і опрацьований в SynchronizationContext, другий — проста фабрика для створення і опрацювання інстансу AsyncOperation. Імплементації ЕАР використовують це, наприклад, Ping.SendAsync викликає AsyncOperationManager.CreateOperation для перехоплення методу SynchronizationContext, і коли операція завершена, з методу AsyncOperation.PostOperationCompleted буде викликаний SynchronizationContext.Post.

SynchronizationContext також надає кілька цінних дрібниць. Зокрема, він видає методи OperationStarted та OperationCompleted. Базова реалізація цих віртуальних методів порожня, але похідні можуть перезаписувати її. Імплементації ЕАР будуть також викликати ці перезаписані методи OperationStarted/OperationCompleted на початку і наприкінці кожної дії, щоб інформувати будь-який наявний SynchronizationContext і дозволити йому відстежувати роботу. Це релевантно для ЕАР патерну, тому що методи, що ініціюють асинхронні операції, повертають void. Ми ще повернемось до цього.

Отже, нам було необхідно щось краще за АРМ, і наступний за ним ЕАР представив деякі нові речі, але насправді не вирішив головних проблем. Нам досі необхідно щось краще.