17. async в ASP.NET Core¶

// Глава 17. Minimal API с типичными ошибками.
// Задание: найдите и исправьте проблемы (README, §17.2):
//   - блокирующее ожидание в обработчике;
//   - нет CancellationToken;
//   - fire-and-forget, использующий scoped-сервис запроса после завершения запроса.
// Запуск сервера:  dotnet run -c Release --project start        (затем GET http://localhost:5000/report, POST http://localhost:5000/orders/42)
// Самопроверка:    dotnet run -c Release --project start -- selftest   (сервер сам нагружает себя и печатает числа)

var builder = WebApplication.CreateBuilder(args);
builder.Logging.SetMinimumLevel(LogLevel.Warning);
builder.Services.AddScoped<ReportService>();
builder.Services.AddScoped<Mailer>();
var app = builder.Build();

app.MapGet("/report", (ReportService reports) =>
{
    string report = reports.BuildAsync(CancellationToken.None).Result;
    return Results.Text(report);
});

app.MapPost("/orders/{id:int}", (int id, Mailer mailer) =>
{
    _ = Task.Run(() => mailer.SendAsync($"заказ {id} принят"));
    return Results.Accepted();
});

if (args.Contains("selftest"))
{
    app.Urls.Add("http://127.0.0.1:5057");
    await app.StartAsync();
    await SelfTest.RunAsync("http://127.0.0.1:5057");
    await app.StopAsync();
}
else
{
    app.Run("http://localhost:5000");
}

sealed class ReportService
{
    public async Task<string> BuildAsync(CancellationToken ct)
    {
        try
        {
            await Task.Delay(500, ct);                 // имитация запроса к БД
            Stats.ReportsCompleted++;
            return $"отчёт на {DateTime.Now:T}";
        }
        catch (OperationCanceledException)
        {
            Stats.ReportsCancelled++;
            throw;
        }
    }
}

sealed class Mailer(ILogger<Mailer> log) : IDisposable
{
    private bool _disposed;

    public async Task SendAsync(string text)
    {
        await Task.Delay(1000);
        if (_disposed)
        {
            Interlocked.Increment(ref Stats.MailsFailed);
            throw new ObjectDisposedException(nameof(Mailer), "scope запроса уже закрыт");
        }
        Interlocked.Increment(ref Stats.MailsSent);
        log.LogInformation("письмо отправлено: {Text}", text);
    }

    public void Dispose() => _disposed = true;
}

static class Stats
{
    public static int ReportsCompleted, ReportsCancelled, MailsSent, MailsFailed;
}

using System.Diagnostics;

// Сервер нагружает сам себя и печатает числа, чтобы не нужен был внешний генератор нагрузки.
static class SelfTest
{
    public static async Task RunAsync(string baseUrl)
    {
        using var http = new HttpClient { BaseAddress = new Uri(baseUrl), Timeout = TimeSpan.FromSeconds(60) };
        await http.GetStringAsync("/report");                                   // прогрев
        Stats.ReportsCompleted = 0;

        // 1. 64 одновременных запроса к /report (каждый «ждёт БД» 500 мс).
        int peakThreads = 0;
        using var stop = new CancellationTokenSource();
        var sampler = Task.Run(async () =>
        {
            while (!stop.IsCancellationRequested)
            {
                peakThreads = Math.Max(peakThreads, ThreadPool.ThreadCount);
                await Task.Delay(50);
            }
        });
        var sw = Stopwatch.StartNew();
        await Task.WhenAll(Enumerable.Range(0, 64).Select(_ => http.GetStringAsync("/report")));
        Console.WriteLine($"1) 64 одновременных /report: {sw.ElapsedMilliseconds} мс, потоков в пуле до {peakThreads} (минимум {Environment.ProcessorCount})");
        stop.Cancel();
        await sampler;

        // 2. Клиент отваливается через 100 мс: останавливает ли сервер работу?
        Stats.ReportsCompleted = 0; Stats.ReportsCancelled = 0;
        using (var cts = new CancellationTokenSource(100))
        {
            try { await http.GetStringAsync("/report", cts.Token); } catch (OperationCanceledException) { }
        }
        await Task.Delay(1000);
        Console.WriteLine($"2) клиент ушёл через 100 мс: сервер довёл отчёт до конца = {Stats.ReportsCompleted}, отменил = {Stats.ReportsCancelled}");

        // 3. Письма в fire-and-forget с scoped-сервисом запроса.
        for (int i = 0; i < 10; i++) await http.PostAsync($"/orders/{i}", null);
        await Task.Delay(2000);
        Console.WriteLine($"3) 10 заказов: писем отправлено {Stats.MailsSent}, упало {Stats.MailsFailed}");
    }
}

// Глава 17, итог.
//   - обработчики async, принимают CancellationToken (это HttpContext.RequestAborted);
//   - фоновая работа идёт через ограниченную очередь на Channel и BackgroundService;
//   - воркер создаёт свой scope на каждое задание и не падает от ошибки одного задания.
//   - потоковый ответ через IAsyncEnumerable; синхронный ввод-вывод в ответ запрещён.
// Запуск сервера:  dotnet run -c Release --project final
// Самопроверка:    dotnet run -c Release --project final -- selftest
using System.Threading.Channels;

var builder = WebApplication.CreateBuilder(args);
builder.Logging.SetMinimumLevel(LogLevel.Warning);
if (args.Contains("selftest")) builder.Logging.AddFilter("Microsoft.AspNetCore", LogLevel.None);   // не засорять вывод ожидаемой ошибкой /sync
builder.Services.AddScoped<ReportService>();
builder.Services.AddScoped<Mailer>();
builder.Services.AddSingleton<BackgroundQueue>();
builder.Services.AddHostedService<QueueWorker>();
var app = builder.Build();

app.MapGet("/report", async (ReportService reports, CancellationToken ct) =>
    Results.Text(await reports.BuildAsync(ct)));

app.MapPost("/orders/{id:int}", async (int id, BackgroundQueue queue, CancellationToken ct) =>
{
    await queue.EnqueueAsync(new MailJob($"заказ {id} принят"), ct);   // при переполнении ждём (обратное давление)
    return Results.Accepted();
});

app.MapGet("/ctx", async () =>
{
    bool before = SynchronizationContext.Current is null;
    int threadBefore = Environment.CurrentManagedThreadId;
    await Task.Delay(50);
    return Results.Text($"контекста нет: {before}; поток до await {threadBefore}, после {Environment.CurrentManagedThreadId}; поток пула: {Thread.CurrentThread.IsThreadPoolThread}");
});

app.MapGet("/stream", () => Numbers());                      // клиент получает элементы по мере готовности

app.MapGet("/sync", (HttpContext ctx) =>
{
    ctx.Response.Body.Write("синхронная запись"u8);           // запрещено по умолчанию (AllowSynchronousIO = false)
    return Results.Empty;
});

if (args.Contains("selftest"))
{
    app.Urls.Add("http://127.0.0.1:5058");
    await app.StartAsync();
    await SelfTest.RunAsync("http://127.0.0.1:5058", app);
}
else
{
    app.Run("http://localhost:5000");
}

static async IAsyncEnumerable<int> Numbers()
{
    for (int i = 1; i <= 5; i++)
    {
        await Task.Delay(200);
        yield return i;
    }
}

sealed record MailJob(string Text);

sealed class BackgroundQueue
{
    private readonly Channel<MailJob> _channel = Channel.CreateBounded<MailJob>(100);

    public ValueTask EnqueueAsync(MailJob job, CancellationToken ct) => _channel.Writer.WriteAsync(job, ct);
    public IAsyncEnumerable<MailJob> ReadAllAsync(CancellationToken ct) => _channel.Reader.ReadAllAsync(ct);
}

sealed class QueueWorker(BackgroundQueue queue, IServiceScopeFactory scopes, ILogger<QueueWorker> log) : BackgroundService
{
    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
    {
        try
        {
            await foreach (var job in queue.ReadAllAsync(stoppingToken))
            {
                try
                {
                    await using var scope = scopes.CreateAsyncScope();          // свой scope, не scope запроса
                    var mailer = scope.ServiceProvider.GetRequiredService<Mailer>();
                    await mailer.SendAsync(job.Text, stoppingToken);
                }
                catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested) { throw; }
                catch (Exception ex) { log.LogError(ex, "задание не выполнено"); }   // не уронить цикл
            }
        }
        catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested)
        {
            // штатная остановка хоста
        }
    }
}

sealed class ReportService
{
    public async Task<string> BuildAsync(CancellationToken ct)
    {
        try
        {
            await Task.Delay(500, ct);
            Stats.ReportsCompleted++;
            return $"отчёт на {DateTime.Now:T}";
        }
        catch (OperationCanceledException)
        {
            Stats.ReportsCancelled++;
            throw;
        }
    }
}

sealed class Mailer(ILogger<Mailer> log) : IDisposable
{
    private bool _disposed;

    public async Task SendAsync(string text, CancellationToken ct)
    {
        await Task.Delay(1000, ct);
        if (_disposed) throw new ObjectDisposedException(nameof(Mailer));
        Interlocked.Increment(ref Stats.MailsSent);
        log.LogInformation("письмо отправлено: {Text}", text);
    }

    public void Dispose() => _disposed = true;
}

static class Stats
{
    public static int ReportsCompleted, ReportsCancelled, MailsSent, MailsFailed;
}

using System.Diagnostics;

// Сервер нагружает сам себя и печатает числа, чтобы не нужен был внешний генератор нагрузки.
static class SelfTest
{
    public static async Task RunAsync(string baseUrl, WebApplication app)
    {
        using var http = new HttpClient { BaseAddress = new Uri(baseUrl), Timeout = TimeSpan.FromSeconds(60) };
        await http.GetStringAsync("/report");                                   // прогрев
        Stats.ReportsCompleted = 0;

        // 1. 64 одновременных запроса к /report (каждый «ждёт БД» 500 мс).
        int peakThreads = 0;
        using var stop = new CancellationTokenSource();
        var sampler = Task.Run(async () =>
        {
            while (!stop.IsCancellationRequested)
            {
                peakThreads = Math.Max(peakThreads, ThreadPool.ThreadCount);
                await Task.Delay(50);
            }
        });
        var sw = Stopwatch.StartNew();
        await Task.WhenAll(Enumerable.Range(0, 64).Select(_ => http.GetStringAsync("/report")));
        Console.WriteLine($"1) 64 одновременных /report: {sw.ElapsedMilliseconds} мс, потоков в пуле до {peakThreads} (минимум {Environment.ProcessorCount})");
        stop.Cancel();
        await sampler;

        // 2. Клиент отваливается через 100 мс: останавливает ли сервер работу?
        Stats.ReportsCompleted = 0; Stats.ReportsCancelled = 0;
        using (var cts = new CancellationTokenSource(100))
        {
            try { await http.GetStringAsync("/report", cts.Token); } catch (OperationCanceledException) { }
        }
        await Task.Delay(1000);
        Console.WriteLine($"2) клиент ушёл через 100 мс: сервер довёл отчёт до конца = {Stats.ReportsCompleted}, отменил = {Stats.ReportsCancelled}");

        // 3. Письма через очередь и BackgroundService.
        for (int i = 0; i < 10; i++) await http.PostAsync($"/orders/{i}", null);
        await Task.Delay(2500);
        Console.WriteLine($"3) 10 заказов: писем отправлено {Stats.MailsSent}, упало {Stats.MailsFailed}");

        // 4. Потоковый ответ: когда приходит первый элемент и когда всё.
        sw.Restart();
        using var resp = await http.GetAsync("/stream", HttpCompletionOption.ResponseHeadersRead);
        using var stream = await resp.Content.ReadAsStreamAsync();
        var buf = new byte[256];
        int n = await stream.ReadAsync(buf);
        long first = sw.ElapsedMilliseconds;
        while (await stream.ReadAsync(buf) > 0) { }
        Console.WriteLine($"4) /stream: первые данные через {first} мс, всё через {sw.ElapsedMilliseconds} мс");

        // 5. Синхронная запись в ответ.
        var r = await http.GetAsync("/sync");
        Console.WriteLine($"5) /sync (Response.Body.Write): HTTP {(int)r.StatusCode}");

        // 6. Контекст синхронизации в обработчике.
        Console.WriteLine($"6) /ctx: {await http.GetStringAsync("/ctx")}");

        // 7. Остановка хоста с ещё не обработанной очередью: что происходит с заданиями?
        Stats.MailsSent = 0;
        for (int i = 0; i < 40; i++) await http.PostAsync($"/orders/{100 + i}", null);
        await app.StopAsync();
        Console.WriteLine($"7) 40 заказов поставлены, хост остановлен сразу: отправлено {Stats.MailsSent} из 40 (остальные потеряны вместе с очередью в памяти)");
    }
}