Асинхронные стримы (IAsyncEnumerable) - .NET

Коллеги, кто уже гонял сабж в продакшн или достаточно внимательно в тестах?

Хочу понять, каковы накладные расходы.

Простой воображаемый пример — я, допустим, читаю данные из (относительно медленного) файла.
Можно сделать честный синхронный енумератор, который будет блокировать поток каждый раз, как нам надо пополнить буфер:

public IEnumerable<string> Lines
{ 
  get
  {
    while(true)
    {
      var s = _reader.ReadLine();

      if(s!=null) 
        yield return s;
      else
        break;
    }
  }
}

А можно попробовать поэкономить простаивающие потоки, и сделать

public IAsyncEnumerable<string> Lines
{ 
  get
  {
    while(true)
    {
      var s = await _reader.ReadLineAsync();

      if(s!=null) 
        yield return s;
      else
        break;
    }
  }
}

Внимание, вопрос: начиная с каких времён ожидания внутри ReadLine имеет смысл вторая схема?

Здравствуйте, Sinclair, Вы писали:

S>Внимание, вопрос: начиная с каких времён ожидания внутри ReadLine имеет смысл вторая схема?

Здесь вижу два аспекта, которые могут давать смысл.
1) — общая продуктивность. Здесь можно сослаться на аналогичную задачу, где копируются данные из медленного источника в медленный приемник.
Синхронное копирование приводит к решению, которое займет сумму времен (чтение всех байт и запись всех байт). Даже, если мы будем читать циклично в некий буфер.
Асинхронное копирование позволяет приблизиться ко времени самого медленного из источника или приемника. Если разница между суммой времен и самым медленным из приемника и источника значительна, то смысл есть.

2) — речь об фризе на время выполнения всего кода. Даже если время каждого ReadLine принебрежительно мало, у нас всегда может быть велико кол-во вызовов ReadLine. Тогда асинхронный поток позволит не фризить форму (например), пока идет обработка большого кол-ва линий. Не заморачиваясь при этом с фоновыми потоками и т.п. В таком случае смысл в удобстве.

Здравствуйте, samius, Вы писали:

S>Здравствуйте, Sinclair, Вы писали:

S>>Внимание, вопрос: начиная с каких времён ожидания внутри ReadLine имеет смысл вторая схема?

S>Здесь вижу два аспекта, которые могут давать смысл.
S>1) — общая продуктивность. Здесь можно сослаться на аналогичную задачу, где копируются данные из медленного источника в медленный приемник.
S>Синхронное копирование приводит к решению, которое займет сумму времен (чтение всех байт и запись всех байт). Даже, если мы будем читать циклично в некий буфер.
S>Асинхронное копирование позволяет приблизиться ко времени самого медленного из источника или приемника. Если разница между суммой времен и самым медленным из приемника и источника значительна, то смысл есть.

S>2) — речь об фризе на время выполнения всего кода. Даже если время каждого ReadLine принебрежительно мало, у нас всегда может быть велико кол-во вызовов ReadLine. Тогда асинхронный поток позволит не фризить форму (например), пока идет обработка большого кол-ва линий. Не заморачиваясь при этом с фоновыми потоками и т.п. В таком случае смысл в удобстве.

Меня больше интересует сценарий вида "я веб сервис, превращаю OData запрос в SQL, а результаты его отдаю клиенту в виде Json". Со всеми прелестями c10k/c100k.

То есть "фриз", как таковой, не очень важен. Важно — не блокировать поток из пула на время ожидания DB или HTTP. Но при этом понятно, что чрезмерно мелкогранулированные таски сами по себе сожрут много ресурсов.
Вот и есть вопрос, какой способ организации конвеера самый правильный. То ли мы дёргаем SQL запрос, асинхронно дожидаемся его завершения (предполагаем, что "завершение" означает окончание копирования всех данных в адресное пространство апп-сервера), потом синхронно конвертируем результат в JSON, потом асинхронно пихаем его в HTTP output stream.
То ли мы строим IAsyncEnumerable над строками результата, предполагая, что он всё ещё едет с сервера в потоковом режиме, и по мере готовности строчек перекладываем их в Json по одной, точно так же асинхронно отправляя в HTTP соединение.

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

Ну, и попутно появляется вопрос — нельзя ли построить что-то типа переходника, который сочетает особенности обоих подходов?
Что-то типа IAsyncEnumerable<IEnumerable<T>>, где мы асинхронно выгребаем некоторые "чанки" данных комфортного размера; при этом каждый чанк разгребается синхронно.
Может быть, это и вовсе не нужно, если IAsyncEnumerable способен избежать лишней асинхронности, если запрошенные данные уже готовы. Но я пока не понимаю, как такое сделать.

Здравствуйте, Sinclair, Вы писали:

S>Внимание, вопрос: начиная с каких времён ожидания внутри ReadLine имеет смысл вторая схема?
Используя ValueTask разницы тo особо и не будет.

Здравствуйте, Sinclair, Вы писали:

S>Здравствуйте, samius, Вы писали:

S>То есть "фриз", как таковой, не очень важен. Важно — не блокировать поток из пула на время ожидания DB или HTTP. Но при этом понятно, что чрезмерно мелкогранулированные таски сами по себе сожрут много ресурсов.
S>Вот и есть вопрос, какой способ организации конвеера самый правильный.
Боюсь, что нет самого правильного ответа в сферических обстоятельствах.
S>То ли мы дёргаем SQL запрос, асинхронно дожидаемся его завершения (предполагаем, что "завершение" означает окончание копирования всех данных в адресное пространство апп-сервера), потом синхронно конвертируем результат в JSON, потом асинхронно пихаем его в HTTP output stream.
S>То ли мы строим IAsyncEnumerable над строками результата, предполагая, что он всё ещё едет с сервера в потоковом режиме, и по мере готовности строчек перекладываем их в Json по одной, точно так же асинхронно отправляя в HTTP соединение.
Здесь принятие решения может зависеть от того, комфортно ли будет клиенту ждать формирования полного ответа на сервере, или он захочет получать его кусками по мере подготовки, вероятно, отказываясь от получения дальнейших результатов...

S>У первого способа потенциальный недостаток — высокие требования к свободной памяти. У второго — накладные расходы на гранулярность.

S>Ну, и попутно появляется вопрос — нельзя ли построить что-то типа переходника, который сочетает особенности обоих подходов?
S>Что-то типа IAsyncEnumerable<IEnumerable<T>>, где мы асинхронно выгребаем некоторые "чанки" данных комфортного размера; при этом каждый чанк разгребается синхронно.
S>Может быть, это и вовсе не нужно, если IAsyncEnumerable способен избежать лишней асинхронности, если запрошенные данные уже готовы. Но я пока не понимаю, как такое сделать.
Как я себе интуитивно представляю, разница между IEnumerable и IAsyncEnumerable не должна быть значительна в случае работы с уже готовыми элементами. В частном случае, думаю что можно исходить из того, что время на проекцию строки БД в JSON будет превосходить накладные расходы на работу с асинком.

Здравствуйте, samius, Вы писали:
S>Как я себе интуитивно представляю, разница между IEnumerable и IAsyncEnumerable не должна быть значительна в случае работы с уже готовыми элементами. В частном случае, думаю что можно исходить из того, что время на проекцию строки БД в JSON будет превосходить накладные расходы на работу с асинком.
Ок, спасибо.

Здравствуйте, Sinclair, Вы писали:

S>Вот и есть вопрос, какой способ организации конвеера самый правильный. То ли мы дёргаем SQL запрос, асинхронно дожидаемся его завершения (предполагаем, что "завершение" означает окончание копирования всех данных в адресное пространство апп-сервера), потом синхронно конвертируем результат в JSON, потом асинхронно пихаем его в HTTP output stream.

А почему не асинхронно, как ниже? Т.е. все шаги конвеера делать асинхронно.

S>То ли мы строим IAsyncEnumerable над строками результата, предполагая, что он всё ещё едет с сервера в потоковом режиме, и по мере готовности строчек перекладываем их в Json по одной, точно так же асинхронно отправляя в HTTP соединение.

Никогда не задумывался, а json может работать потоково? Т.е. когда сериализумемая структура имеется не полностью?

Здравствуйте, Sharov, Вы писали:

S>Никогда не задумывался, а json может работать потоково? Т.е. когда сериализумемая структура имеется не полностью?
Да в Newton есть ReadAsync
https://www.newtonsoft.com/json/help/html/T_Newtonsoft_Json_JsonTextReader.htm

в Utf8JsonReader нет. Но там работа с памятью
https://docs.microsoft.com/ru-ru/dotnet/api/system.text.json.utf8jsonreader.-ctor?view=net-6.0

Здравствуйте, Sharov, Вы писали:

S>А почему не асинхронно, как ниже? Т.е. все шаги конвеера делать асинхронно.
Ну вот я лично не работал со всей async-магией, и не представляю себе уровень накладных расходов.
То есть как работает обычный enumerable pattern я себе хорошо представляю. Понимаю, как можно минимизировать накладные расходы на итерирование через итератор (относительно, скажем, итерирования по массиву или односвязному списку).
А вот что там с IAsyncEnumerable — хз. Не получится ли так, что асинхронное выгребание тысячи строк займёт втрое больше времени, чем их же синхронное выгребание?

S>Никогда не задумывался, а json может работать потоково? Т.е. когда сериализумемая структура имеется не полностью?
Ну, встроенного метода постепенной выгрузки Json нету, но он, вроде бы, очень просто устроен.
Тупо берём стрим, скидываем в него что-то вроде { "items": [, и фигачим цикл.

await foreach(var p in Db.GetAsync<Person>())
{
  await serializer.SerializeAsync(stream, p); // тут ещё надо запихать запятую
}

И в конце закидываем ему "]}".

Здравствуйте, Sinclair, Вы писали:

S>

S>await foreach(var p in Db.GetAsync<Person>())
S>{
S>  await serializer.SerializeAsync(stream, p); // тут ещё надо запихать запятую
S>}
S>

Общепринятый паттерн: сепаратор втавлять перед элементом, а не после.
То есть не так:

loop
{
    write(item);
    if (not last)
        add(separator);
}

А так:

loop
{
    if (not first)
        add(separator);
    write(item);
}

А вообще для такого сценария можно использовать формат JSON Lines.

Здравствуйте, Qbit86, Вы писали:

Q>Общепринятый паттерн: сепаратор втавлять перед элементом, а не после.
Да, конечно. Мне в основном не хотелось лезть в подробности преобразований кодировок перед работой со Stream.

Q>А вообще для такого сценария можно использовать формат JSON Lines.
Для этого надо договориться с клиентом об используемом протоколе. В примере предполагалось, что мы формируем валидный JSON-документ.
Впрочем, это всё несущественные детали с точки зрения обсуждаемого вопроса

Здравствуйте, Sinclair, Вы писали:

S>Коллеги, кто уже гонял сабж в продакшн или достаточно внимательно в тестах?

S>Хочу понять, каковы накладные расходы.

Легко получить накладные расходы бенчмарками, да

1. если под await сеть, то сравнить await foreach и foreach для TcpListener на localhost и 127.0.0.1. разница будет потому что MTA разный
2. если под await диск, то сравнить await foreach и foreach для RAM-диска или оптан диска. На всякий случай упомяну что асинхронная работа в FileStream сильно переписана в 6м Core. В 5м много чего в Async-методах всё еще реализовано абы как (синхронно)
3. если под await потоки (например ManualResetEventSlim.Wait — самая быстрая реализация) то я уже делился задержками. Они большие: 1 микросекунда в Core на Windows, 2 — Core на линукс, 4 — на NetFramework под Windows, и 16 — в моно. на кажую итерацию. Это всё на Intel Xeon 6го поколения.

4й вариант — сравнить в Dapper await foreach и foreach. Тут разница конечно сильно большая будет в SQLite и MS SQL Shared Memory и MS SQL TCP IP. Вобщем работы много.

Здравствуйте, VladCore, Вы писали:

VC>накладные расходы легко посчитать:
VC>1. если под await сеть, то сравнить await foreach и foreach для TcpListener на localhost и 127.0.0.1. разница будет потому что MTA разный
VC>2. если под await диск, то сравнить await foreach и foreach для RAM-диска или оптан диска.
VC>3. если под await потоки (например ManualResetEventSlim.Wait — самая быстрая реализация) то я уже делился задержками. Они большие: 1 микросекунда в Core на Windows, 2 — Core на линукс, 4 — на NetFramework и 16 — в моно. на кажую итерацию. Это всё на Intel Xeon 6го поколения.
Ок, я, кажется, понял, что нужно делать.

Вот нашёл прямо-таки идеально совпадающий с моими размышлениями пример:

public async IAsyncEnumerable<Product> GetAllProducts()
{
    var iterator = cosmosClient.GetContainer(DatabaseId, ContainerId).GetItemQueryIterator<Product>("SELECT * FROM c");
    while (iterator.HasMoreResults)
        foreach (var product in await iterator.ReadNextAsync())
            yield return product;
}

Тут как раз на входе — набор страниц, на выходе — плоский IAsyncEnumerable.

В случае, если GetItemQueryIterator<Product> возвращает нам ровно один FeedResponse<Product>, то это всё вырождается в линейный код типа такого:

public async IAsyncEnumerable<Product> GetAllProducts()
{
  foreach (var product in await iterator.ReadNextAsync())
    yield return product;
}

Тут нам не очень важен внутренний await.
Можно просто построить асинхронный стрим:

public static async IAsyncEnumerable<T> ToAsync(this IEnumerable<T> enumerable)
{
  foreach(var t in enumerable) yield return t;
}

После этого можно сравнить что-нибудь простое, типа вычисления суммы массива синхронно и асинхронно:

public async Task<int> SumSync(IEnumerable<int> enumerable)
{
  var s = 0;
  foreach(var i in enumerable) s+=i;
  return s;
}

public async Task<int> SumAsync(IEnumerable<int> enumerable)
{
  var s = 0;
  await foreach(var i in enumerable.ToAsync()) s+=i;
  return s;
}

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

Здравствуйте, Sinclair, Вы писали:

S>Вот нашёл прямо-таки идеально совпадающий с моими размышлениями пример:
S>

S>public async IAsyncEnumerable<Product> GetAllProducts()
S>{
S>    var iterator = cosmosClient.GetContainer(DatabaseId, ContainerId).GetItemQueryIterator<Product>("SELECT * FROM c");
S>    while (iterator.HasMoreResults)
S>        foreach (var product in await iterator.ReadNextAsync())
S>            yield return product;
S>}
S>

S>Тут как раз на входе — набор страниц, на выходе — плоский IAsyncEnumerable.

По ссылке не ходил, но зачем тут await, если все продукты будут в памяти? Или драйвер бд будет потоково гнать продукты
через сеть,например?

Здравствуйте, Sharov, Вы писали:

S>Здравствуйте, Sinclair, Вы писали:

S>>Вот нашёл прямо-таки идеально совпадающий с моими размышлениями пример:
S>>

S>>public async IAsyncEnumerable<Product> GetAllProducts()
S>>{
S>>    var iterator = cosmosClient.GetContainer(DatabaseId, ContainerId).GetItemQueryIterator<Product>("SELECT * FROM c");
S>>    while (iterator.HasMoreResults)
S>>        foreach (var product in await iterator.ReadNextAsync())
S>>            yield return product;
S>>}
S>>

S>>Тут как раз на входе — набор страниц, на выходе — плоский IAsyncEnumerable.

S>По ссылке не ходил, но зачем тут await, если все продукты будут в памяти? Или драйвер бд будет потоково гнать продукты
S>через сеть,например?
Да, конечно, здесь драйвер гонит всё через сеть. Но никто через сеть килобайтные айтемы по одному не тащит — тащат пачками. CosmosClient наружу выставляет интерфейс, эквивалентный IEnumerable<Task<IEnumerable<T>>>.
Но потреблять этот интерфейс надо через двойной цикл, что как бы не вполне удобно.

Вот у меня и возникает вопрос: можно ли вообще отказаться в клиентском API от вот этой вот конструкции IEnumerable<Task<IEnumerable<T>>>, полностью перейдя на IAsyncEnumerable<T>, или всё-таки он будет портить производительность на небольших запросах (которые удалось качнуть из сети в один приём).

Здравствуйте, Sinclair, Вы писали:

Посмотри https://andrey.moveax.ru/post/csharp-features-v8-0-async-streams
Там кстати ValueTask<bool> MoveNextAsync() возвращает.
Обычно тебе нужно асинхронно подождать пакет, а затем синхронно пройтись по пакету.
Обычно в таком случае быдет возвращаться ValueTask с минмальными задержками

Здравствуйте, Sharov, Вы писали:

S>>Вот нашёл прямо-таки идеально совпадающий с моими размышлениями пример:
S>>

S>>public async IAsyncEnumerable<Product> GetAllProducts()
S>>{
S>>    var iterator = cosmosClient.GetContainer(DatabaseId, ContainerId).GetItemQueryIterator<Product>("SELECT * FROM c");
S>>    while (iterator.HasMoreResults)
S>>        foreach (var product in await iterator.ReadNextAsync())
S>>            yield return product;
S>>}
S>>

S>>Тут как раз на входе — набор страниц, на выходе — плоский IAsyncEnumerable.

S>По ссылке не ходил, но зачем тут await, если все продукты будут в памяти? Или драйвер бд будет потоково гнать продукты
S>через сеть,например?

это издержки горизонтально масштабируемего NoSQL хранилища. клиенское API подогнано под тот факт что одним запросом всё нельзя получить что попадает под фильтр. С разных партиций поступают свои наборы "строчек" и клиент обязан вызывать ReadNext в цикле

await foreach сюда идеально ложится но всё же это не всем нужно и await foreach не только для такого случая.

Здравствуйте, VladCore, Вы писали:

VC>это издержки горизонтально масштабируемего NoSQL хранилища. клиенское API подогнано под тот факт что одним запросом всё нельзя получить что попадает под фильтр. С разных партиций поступают свои наборы "строчек" и клиент обязан вызывать ReadNext в цикле

А классический sql бд не могут батчами данные гнать, ну т.е. данных до фига, канал не резиновый, загнали 1Гб,
получили все возможные Person, прочитали, подогнали еще 1Гб данных и т.д. Это как раз про то, что выше говорилось --
IEnumerable<Task<IEnumerable<T>>>. Т.е. сценарий NoSql понятен, но вот когда реально много данных и все сразу в память
не прочитаешь тоже сюда относится.

Здравствуйте, Sharov, Вы писали:

VC>>это издержки горизонтально масштабируемего NoSQL хранилища. клиенское API подогнано под тот факт что одним запросом всё нельзя получить что попадает под фильтр. С разных партиций поступают свои наборы "строчек" и клиент обязан вызывать ReadNext в цикле

S>А классический sql бд не могут батчами данные гнать, ну т.е. данных до фига, канал не резиновый, загнали 1Гб,
S>получили все возможные Person, прочитали, подогнали еще 1Гб данных и т.д. Это как раз про то, что выше говорилось --
S>IEnumerable<Task<IEnumerable<T>>>. Т.е. сценарий NoSql понятен, но вот когда реально много данных и все сразу в память
S>не прочитаешь тоже сюда относится.

Классический SQL клиент никогда не грузит в память весь 1Гб и на сервере иногда "грузит" весь возвращаемый набор иногда нет.

SqlConnection con = ;
var persons = await con.QueryAsync<Person>("Select * From Person Order By Surname");
foreach(Person p in persons) {
}

вот тут весь 1 гигабайт на клиенте всегда НЕ грузится целиком в озу.
И на стороне сервера этот 1 гигабайт не копируется ни в ОЗУ ни во временные "таблицы" если есть индекс по Surname

Не знал? 😉

P.S. Точнее говоря весь 1 гиг в примере выше все же попадут в ОЗУ верменно, потому что GC лень сразу освободить все экземпляры Person p если ОЗУ хватает

Здравствуйте, VladCore, Вы писали:

S>>А классический sql бд не могут батчами данные гнать, ну т.е. данных до фига, канал не резиновый, загнали 1Гб,
S>>получили все возможные Person, прочитали, подогнали еще 1Гб данных и т.д. Это как раз про то, что выше говорилось --
S>>IEnumerable<Task<IEnumerable<T>>>. Т.е. сценарий NoSql понятен, но вот когда реально много данных и все сразу в память
S>>не прочитаешь тоже сюда относится.

VC>Классический SQL клиент никогда не грузит в память весь 1Гб и на сервере иногда "грузит" весь возвращаемый набор иногда нет.

VC>

VC>SqlConnection con = ;
VC>var persons = await con.QueryAsync<Person>("Select * From Person Order By Surname");
VC>foreach(Person p in persons) {
VC>}
VC>

VC>вот тут весь 1 гигабайт на клиенте всегда НЕ грузится целиком в озу.

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

В любом случае, речь о том, где будут данные между

var persons = await con.QueryAsync<Person>("Select * From Person Order By Surname");
и
foreach(Person p in persons)

Как я понимаю, когда мы уже итерируемся по persons, то мы получили все данные от сервера, где
бы на клиенте они не находились. Я же говорил, про возможность итерироваться по полученным
persons, когда мы получаем только часть данных. Т.е. что-то получили, десериализовали, прошлись циклом,
читаем сл. порцию и т.д.

Т.е. в одном случае (как в примере выше) foreach начнет работу после окончания работы сервера,
а возможен ли случай, когда foreach начнет работу параллельно с работой сервера?

VC>И на стороне сервера этот 1 гигабайт не копируется ни в ОЗУ ни во временные "таблицы" если есть индекс по Surname
VC>Не знал? 😉

Нет, но догадывался

VC>P.S. Точнее говоря весь 1 гиг в примере выше все же попадут в ОЗУ верменно, потому что GC лень сразу освободить все экземпляры Person p если ОЗУ хватает

Это от драйвера зависит, может сразу в память все перегнать.

От:	Sinclair	https://github.com/evilguest/
Дата:	21.09.21 06:00
Оценка:

От:	samius	http://sams-tricks.blogspot.com
Дата:	21.09.21 06:40
Оценка:	85 (3) +1

От:	Sinclair	https://github.com/evilguest/
Дата:	21.09.21 07:31
Оценка:	5 (1)

От:	Serginio1	https://habrahabr.ru/users/serginio1/topics/
Дата:	21.09.21 08:28
Оценка:

От:	samius	http://sams-tricks.blogspot.com
Дата:	21.09.21 08:34
Оценка:

От:	Qbit86	https://twitter.com/qbit86
Дата:	22.09.21 16:11
Оценка:	38 (1)

	От:	VladCore
	Дата:	23.09.21 12:54
	Оценка:

	От:	Sharov
	Дата:	23.09.21 13:52
	Оценка:

	От:	Sharov
	Дата:	23.09.21 20:40
	Оценка:

	От:	Sharov
	Дата:	24.09.21 11:43
	Оценка: