C# IAsyncEnumerable, await foreach ve Async Dispose

Sürüm notu: Bu yazıdaki kodlar C# 10 ve .NET 6 API'leriyle yazılmıştır; .NET 10 SDK ile derlenip denenmiştir.

Önce Tahmin Et

Katalog API'si kitapları sayfa sayfa veriyor: her sayfada iki kitap var ve her sayfa isteği 200 ms sürüyor. Üç sayfanın hepsini bir listeye toplayıp döndüren bir method yazdık ve sonucu ekrana basıyoruz. İlk kitabın adı ekrana kaçıncı milisaniyede gelir?

Stopwatch sw = Stopwatch.StartNew();
foreach (Book book in await GetAllBooksAsync())
{
    Console.WriteLine($"{sw.ElapsedMilliseconds,4} ms  {book.Title}");
}

static async Task<List<Book>> GetAllBooksAsync()
{
    List<Book> all = new();
    for (int page = 1; page <= 3; page++)
    {
        all.AddRange(await Catalog.GetPageAsync(page));
    }
    return all;
}

record Book(string Isbn, string Title);

static class Catalog
{
    private static readonly Book[] Books =
    {
        new("978-0441172719", "Dune"),
        new("978-0553293357", "Foundation"),
        new("978-0261102217", "The Hobbit"),
        new("978-0547928227", "The Lord of the Rings"),
        new("978-0451524935", "1984"),
        new("978-0060850524", "Brave New World"),
    };

    // her sayfa 200 ms süren bir API çağrısı
    public static async Task<Book[]> GetPageAsync(int page, CancellationToken cancellationToken = default)
    {
        await Task.Delay(200, cancellationToken);
        return Books.Skip((page - 1) * 2).Take(2).ToArray();
    }
}

Hepsini Toplayıp Dönmek

Çıktı (dosyanın başında using System.Diagnostics; satırı var):

 607 ms  Dune
 609 ms  Foundation
 609 ms  The Hobbit
 609 ms  The Lord of the Rings
 609 ms  1984
 609 ms  Brave New World

İlk kitap 607. milisaniyede geldi, yani üç sayfanın hepsi bittikten sonra. Dune aslında 200. milisaniyede elimizdeydi, ama Task<List<Book>> tek bir sonuç demek: liste tamamlanmadan dışarıya hiçbir şey çıkamaz. Kullanıcı 600 ms boyunca boş bir ekrana bakar ve bütün kitaplar aynı anda bellekte durur. Altı kitapta sorun değil; on bin sayfalık bir dışa aktarımda hem bekleme hem bellek sorun olur.

Düz bir IEnumerable<Book> ile yield return da işe yaramaz, çünkü MoveNext() senkron bir method'dur: bir sonraki element için ağı beklemek zorundaysa thread'i bloklar. Bize hem "parça parça ver" hem de "beklerken thread'i bırak" diyebilen bir tip lazım.

Bir diziyi izlemeyi düşün. Bir platform bütün sezonu çekip montajını bitirdikten sonra tek seferde yayınlayabilir, ya da her bölümü hazır olduğu hafta yayınlar. İlkinde ilk bölümü izlemek için sezonun bitmesini beklersin. İkincisinde ilk bölüm hazır olduğu gün izlemeye başlarsın. Task<List<T>> birincisi, IAsyncEnumerable<T> ikincisi.

Async Iterator: yield return ve await Bir Arada

C# 8 ve .NET Core 3.0 ile gelen IAsyncEnumerable<T>, bu ihtiyacın cevabı. Method'u async yapıp dönüş tipini IAsyncEnumerable<Book> yazdığında içinde hem await hem yield return kullanabilirsin:

static async IAsyncEnumerable<Book> GetBooksAsync()
{
    for (int page = 1; page <= 3; page++)
    {
        foreach (Book book in await Catalog.GetPageAsync(page))
        {
            yield return book; // sayfa gelir gelmez kitapları tek tek ver
        }
    }
}

Tüketen taraf da foreach'in async karşılığını kullanır:

await foreach (Book book in GetBooksAsync())
{
    Console.WriteLine($"{sw.ElapsedMilliseconds,4} ms  {book.Title}");
}

 202 ms  Dune
 202 ms  Foundation
 404 ms  The Hobbit
 404 ms  The Lord of the Rings
 606 ms  1984
 606 ms  Brave New World

İlk iki kitap 202. milisaniyede geldi, çünkü ilk sayfa iner inmez yield return onları tek tek dışarı verdi. Toplam süre yine 600 ms civarı; daha hızlı bitirmedik, ama sonucu beklerken boş durmadık. Bellekte de her an en fazla bir sayfa var.

Compiler böyle bir method için hem iterator'ın hem async method'un işini yapan tek bir state machine üretir. yield return'de method duraklar ve element'i consumer'a verir. Consumer bir sonrakini istediğinde method kaldığı yerden devam eder; o sırada bir await'e gelirse thread'i bırakır ve bekler.

await foreach Perde Arkası

await foreach, compiler'ın senin yerine yazdığı birkaç satırın kısaltmasıdır. Elle yazınca şuna benzer:

IAsyncEnumerator<Book> enumerator = GetBooksAsync().GetAsyncEnumerator();
try
{
    while (await enumerator.MoveNextAsync()) // MoveNextAsync bir ValueTask<bool> döner
    {
        Book book = enumerator.Current;
        Console.WriteLine(book.Title);
    }
}
finally
{
    await enumerator.DisposeAsync();
}

Dune
Foundation

İki ayrıntı göze çarpıyor. MoveNextAsync bir ValueTask<bool> döner. Async bir stream'de çoğu element zaten buffer'da hazırdır; örneğimizde bir sayfa gelince ikinci kitap için beklemeye gerek yok. Her element için bir Task ayırmamak tam da bir önceki yazıda gördüğümüz ValueTask kullanım alanı. Kurallara da uyuluyor: her ValueTask bir kez, hemen await ediliyor.

İkincisi, döngü nasıl biterse bitsin finally bloğunda DisposeAsync çağrılıyor. Bunun neden önemli olduğunu biraz sonra erken çıkışta göreceğiz.

İptal: EnumeratorCancellation

Bir async iterator'a token iletmenin doğru yolu iki parçadan oluşur. Method'un parametresine [EnumeratorCancellation] attribute'u koyarsın ve token'ı consumer tarafında WithCancellation ile verirsin:

static async IAsyncEnumerable<Book> GetBooksAsync(
    [EnumeratorCancellation] CancellationToken cancellationToken = default)
{
    for (int page = 1; page <= 3; page++)
    {
        foreach (Book book in await Catalog.GetPageAsync(page, cancellationToken))
        {
            yield return book;
        }
    }
}

using CancellationTokenSource cts = new(TimeSpan.FromMilliseconds(500));

try
{
    await foreach (Book book in GetBooksAsync().WithCancellation(cts.Token))
    {
        Console.WriteLine(book.Title);
    }
}
catch (OperationCanceledException)
{
    Console.WriteLine("Canceled while waiting for the next page");
}

Dune
Foundation
The Hobbit
The Lord of the Rings
Canceled while waiting for the next page

500 ms'de token iptal oldu. O sırada üçüncü sayfa bekleniyordu ve Task.Delay iptali fark edip exception fırlattı. Dört kitap ekrana geldi, beşincisi gelmedi. Attribute için System.Runtime.CompilerServices namespace'ini eklemek gerekiyor.

Attribute'u unutursan ne olur? Aynı programı [EnumeratorCancellation] olmadan derlediğimizde compiler şu uyarıyı verdi ve iptal hiç çalışmadı:

Program.cs(16,37): warning CS8425: Async-iterator 'GetBooksAsync(CancellationToken)' has one or more parameters of type 'CancellationToken' but none of them is decorated with the 'EnumeratorCancellation' attribute, so the cancellation token parameter from the generated 'IAsyncEnumerable<>.GetAsyncEnumerator' will be unconsumed
Dune
Foundation
The Hobbit
The Lord of the Rings
1984
Brave New World

WithCancellation ile verdiğin token enumerator'a gider, ama method'un parametresine bağlanmaz. Parametre default değerinde kalır ve altı kitabın hepsi gelir. Bu uyarıyı her zaman bir hata gibi ele al. Attribute varken token'ı doğrudan GetBooksAsync(cts.Token) diye de verebilirsin; iki yol da aynı parametreye ulaşır.

Erken Çıkış ve finally

Async iterator'lar çoğu zaman arkada bir kaynak tutar: bir veritabanı bağlantısı, bir dosya, bir HTTP response'u. Consumer döngüden erken çıkarsa o kaynak ne olur?

await foreach (Book book in GetBooksAsync())
{
    Console.WriteLine($"Found: {book.Title}");
    break; // ilk kitap yeterli
}
Console.WriteLine("After the loop");

static async IAsyncEnumerable<Book> GetBooksAsync()
{
    Console.WriteLine("Opening the catalog connection");
    try
    {
        for (int page = 1; page <= 3; page++)
        {
            Console.WriteLine($"Loading page {page}");
            await Task.Delay(200);
            yield return new Book("978-0441172719", $"Book from page {page}");
        }
    }
    finally
    {
        Console.WriteLine("Closing the catalog connection"); // break'te de çalışır
    }
}

record Book(string Isbn, string Title);

Opening the catalog connection
Loading page 1
Found: Book from page 1
Closing the catalog connection
After the loop

Sadece ilk sayfa yüklendi ve break ile döngüden çıktığımız anda "Closing the catalog connection" yazıldı. await foreach döngü bittiğinde, break ile çıkıldığında ya da exception fırladığında enumerator'ın DisposeAsync'ini çağırır; o da iterator'daki finally bloğunu çalıştırır. Kaynağı iterator'ın içinde try/finally ya da await using ile tutarsan consumer'ın ne yaptığından bağımsız olarak doğru kapanır. Iterator'ı await foreach olmadan elle dolaşıyorsan DisposeAsync'i çağırmak senin sorumluluğun.

IAsyncDisposable ve await using

Bazı kaynakların kapanması da zaman alır: buffer'ı diske yazmak, ağ bağlantısını düzgünce kapatmak. IDisposable.Dispose senkron olduğu için bu işi bloklayarak yapmak zorundadır. C# 8 ile gelen IAsyncDisposable ve await using bunun async karşılığıdır:

sealed class OrderExportWriter : IAsyncDisposable
{
    private readonly string _path;
    private readonly List<string> _buffer = new();

    public OrderExportWriter(string path) => _path = path;

    public Task WriteAsync(string row)
    {
        _buffer.Add(row);
        return Task.CompletedTask;
    }

    public async ValueTask DisposeAsync()
    {
        await File.WriteAllLinesAsync(_path, _buffer); // buffer'ı diske async yaz
        Console.WriteLine($"Flushed {_buffer.Count} rows to {_path}");
    }
}

await using (OrderExportWriter writer = new("orders.csv"))
{
    await writer.WriteAsync("42,Dune,9.99");
    await writer.WriteAsync("43,Foundation,8.50");
    Console.WriteLine("Rows written");
}
Console.WriteLine("After the using block");

Rows written
Flushed 2 rows to orders.csv
After the using block

Blok bittiğinde compiler await writer.DisposeAsync() çağrısını ekledi; buffer diske async olarak yazıldı ve "After the using block" ancak ondan sonra geldi. DisposeAsync'in dönüş tipine dikkat: ValueTask. Çoğu dispose işlemi ya hiç beklemez ya da kısa sürer, yine aynı gerekçe.

Süslü parantezsiz yazım da var: await using OrderExportWriter writer = new("orders.csv");. Bu durumda dispose, değişkenin bulunduğu scope bittiğinde çalışır. Bir class hem senkron hem async kullanılacaksa ikisini de implemente edebilir; Stream ve DbContext böyledir.

Gerçek Hayatta: JSON ve ASP.NET Core

Async stream'lerle en sık JSON tarafında karşılaşırsın. .NET 6 ile gelen JsonSerializer.DeserializeAsyncEnumerable, büyük bir JSON array'ini bütünüyle belleğe almadan, element element okur:

using System.Text;
using System.Text.Json;

byte[] json = Encoding.UTF8.GetBytes(
    "[{\"isbn\":\"978-0441172719\",\"title\":\"Dune\"},{\"isbn\":\"978-0553293357\",\"title\":\"Foundation\"}]");
using MemoryStream stream = new(json); // gerçekte bir HTTP response stream'i

JsonSerializerOptions options = new(JsonSerializerDefaults.Web);
await foreach (Book? book in JsonSerializer.DeserializeAsyncEnumerable<Book>(stream, options))
{
    Console.WriteLine(book?.Title);
}

record Book(string Isbn, string Title);

Dune
Foundation

Gerçek kodda MemoryStream yerine HttpClient'tan aldığın response stream'i verirsin; binlerce siparişlik bir yanıtı ilk element gelir gelmez işlemeye başlarsın.

Ters yönde de çalışır. ASP.NET Core'da bir endpoint IAsyncEnumerable<T> döndürebilir:

var builder = WebApplication.CreateBuilder(args);
var app = builder.Build();

app.MapGet("/books", () => GetBooksAsync());

app.Run();

static async IAsyncEnumerable<Book> GetBooksAsync()
{
    string[] titles = { "Dune", "Foundation", "The Hobbit" };
    foreach (string title in titles)
    {
        await Task.Delay(500); // her kitap yavaş bir kaynaktan geliyor
        yield return new Book(title);
    }
}

record Book(string Title);

Uygulamayı çalıştırıp yanıtı küçük bir istemciyle parça parça okuduk ve her parçanın geliş zamanını yazdırdık:

  535 ms  [{"title":"Dune"}
 1037 ms  ,{"title":"Foundation"}
 1537 ms  ,{"title":"The Hobbit"}]

JSON array'i tek seferde değil, her kitap hazır oldukça parça parça geldi. Proje net6.0 hedefliyor; bu makinede .NET 6 runtime'ı olmadığı için roll-forward ile ASP.NET Core 8 üzerinde çalıştı. EF Core'da da AsAsyncEnumerable() ile bir sorgunun sonuçlarını satır satır akıtabilirsin.

Bir eksik var: .NET 6'nın BCL'inde IAsyncEnumerable<T> için Where, Select, ToListAsync gibi LINQ operatörleri yok. O dönemde bunun için topluluğun System.Linq.Async paketi kullanılıyordu. .NET 10 bu operatörleri System.Linq.AsyncEnumerable ile platforma taşıdı; onu bu serinin .NET 9–10 yazısında göreceğiz.

Sık Düşülen Tuzaklar

Async iterator'ın token parametresine [EnumeratorCancellation] koymamak. WithCancellation sessizce etkisiz kalır; CS8425 uyarısı tam bunu söyler.

Stream olarak gelebilecek veriyi Task<List<T>> ile toplamak. İlk element en sonuncusu kadar geç gelir ve hepsi aynı anda bellekte durur.

Kaynağı iterator'ın dışında açıp içinde kullanmak. Consumer erken çıktığında ya da hiç dolaşmadığında kapatacak kimse kalmaz. Kaynağı iterator'ın içinde try/finally veya await using ile yönet.

Async bir kaynağı düz using ile kapatmak. Sadece IAsyncDisposable implemente eden bir tipte using derlenmez. İkisini de implemente eden bir tipte ise derlenir ama senkron Dispose çağrılır ve async kapanışın sağladığı avantajı kaybedersin.

Aynı async iterator'ı iki kez dolaşmak. Her await foreach method'u baştan çalıştırır; sayfalı bir API'de bu, bütün sayfaların ikinci kez istenmesi demektir. LINQ'daki deferred execution'ın async hâli.

Küçük Bir Deneme

★☆☆ İlk örnekteki GetBooksAsync'i tüketirken sadece başlığı "The" ile başlayan kitapları yazdır ve ikinci böyle kitabı bulduğunda break ile çık. Kaç sayfa istendi?

★★☆ System.Linq.Async paketi olmadan kendi ToListAsync extension method'unu yaz: bir IAsyncEnumerable<T> alsın, bir CancellationToken desteklesin ve bir List<T> döndürsün.

★★★ OrderExportWriter'ı gerçek bir StreamWriter ile yeniden yaz ve bir async iterator'dan gelen siparişleri satır satır dosyaya akıt. İterasyonun ortasında bir exception fırlat ve dosyanın yine de düzgün kapandığını kontrol et.

Async stream'lerde elementler sırayla, birer birer işlendi. Bazen tam tersini istersin: elindeki işleri aynı anda birden fazla çekirdekte ya da sınırlı bir concurrency ile işlemek. Bir sonraki yazıda Parallel.For, Parallel.ForEach, .NET 6 ile gelen Parallel.ForEachAsync, PLINQ ve Task.Run'ın ne zaman doğru araç olduğunu görüyoruz.