21. Память и сборка мусора в async-коде: утечки и их поиск¶

// Глава 21. Сервис с тремя утечками. Запуск: dotnet run -c Release --project start
// Программа имитирует работу и печатает, сколько памяти и объектов осталось ПОСЛЕ неё.
// PREDICT: сколько экранов и ожидающих запросов останется живыми? Сколько мегабайт?
// TODO 1: найдите все три места утечки. Для этого запустите с паузой (аргумент hang) и посмотрите dotnet-dump: dumpheap -stat, gcroot.
// TODO 2: исправьте так, чтобы после «нагрузки» живых экранов и записей реестра было 0, а куча ≈ 1 МБ.
// TODO 3: как бы вы поймали эту утечку тестом? (подсказка: WeakReference + GC.Collect)
using System.Collections.Concurrent;

const int N = 10_000;
var bus = new Bus();
var gateway = new Gateway();
var screens = new List<WeakReference>();

double before = Mb();
using var cts = new CancellationTokenSource();

// Утечка 1 и 2: запросы без ответа и «keep-alive» без токена.
var requests = new List<Task>();
for (int i = 0; i < N; i++) requests.Add(gateway.RequestAsync(i, new byte[1000], cts.Token));
await gateway.KeepAliveAsync(100);                    // 100 ожиданий по 10 минут: «забытый» таймер

// Утечка 3: экраны подписываются на шину и «закрываются».
for (int i = 0; i < N; i++)
{
    var screen = new Screen(bus);
    screens.Add(new WeakReference(screen));
    screen.Close();
}

// Конец «нагрузки»: клиенты ушли.
cts.Cancel();                                         // «разрыв соединений»: токен отменён
requests.Clear();                                     // про задачи запросов мы больше не помним
await Task.Delay(200);                                // даём стеку завершения размотаться

GC.Collect(); GC.WaitForPendingFinalizers(); GC.Collect();
Console.WriteLine($"Живых экранов: {screens.Count(s => s.IsAlive)} из {N}");
Console.WriteLine($"Записей в реестре шлюза: {gateway.Pending}");
Console.WriteLine($"Куча: +{Mb() - before:F1} МБ");

if (args.FirstOrDefault() == "hang")
{
    Console.WriteLine($"PID {Environment.ProcessId}: подключайтесь dotnet-dump / dotnet-gcdump. Завершится через 120 с.");
    await Task.Delay(TimeSpan.FromSeconds(120));
}

static double Mb()
{
    GC.Collect(2, GCCollectionMode.Forced, blocking: true, compacting: true);
    var info = GC.GetGCMemoryInfo(GCKind.FullBlocking);
    return (info.HeapSizeBytes - info.FragmentedBytes) / 1024.0 / 1024.0;
}

sealed class Gateway
{
    private readonly ConcurrentDictionary<int, TaskCompletionSource<string>> _pending = new();
    public int Pending => _pending.Count;

    public async Task RequestAsync(int id, byte[] context, CancellationToken ct)
    {
        var tcs = new TaskCompletionSource<string>(TaskCreationOptions.RunContinuationsAsynchronously);
        _pending[id] = tcs;
        string reply = await tcs.Task;               // ответ приходит из другого места; он может не прийти
        _pending.TryRemove(id, out _);
        Console.Write(context.Length + reply.Length);
    }

    public async Task KeepAliveAsync(int count)
    {
        var waits = new List<Task>();
        for (int i = 0; i < count; i++) waits.Add(KeepOneAsync(new byte[10_000]));
        await Task.Delay(10);                        // «работаем дальше», не дожидаясь keep-alive
    }

    private static async Task KeepOneAsync(byte[] state)
    {
        await Task.Delay(TimeSpan.FromMinutes(10));
        Console.Write(state.Length);
    }
}

sealed class Bus
{
    public event Func<string, Task>? Message;
    public void Subscribe(Func<string, Task> handler) => Message += handler;
}

sealed class Screen
{
    private readonly byte[] _state = new byte[1000];

    public Screen(Bus bus) => bus.Subscribe(async text => { await Task.Yield(); _state[0] = (byte)text.Length; });

    public void Close() { /* экран закрыт, пользователь ушёл */ }
}

// Глава 21. Память и сборка мусора в async-коде. Режимы (после -- ):
//   hoist    — что держит локальная переменная async-метода, пока он приостановлен;
//   delay    — «брошенные» Task.Delay и отмена токеном;
//   tcs      — запросы, ждущие ответ, который не придёт (реестр ожидающих);
//   event    — async-обработчики событий долгоживущего издателя;
//   channel  — неограниченная и ограниченная очередь при медленном потребителе;
//   gen      — поколения: во что превращается долго ждущий async-метод;
//   service  — сервис из start/ с исправленными утечками;
//   hang     — утечка из режима delay и пауза, чтобы подключиться dotnet-gcdump / dotnet-dump;
//   hang-cancelled — «исправленный» delay, но одна отменённая задача осталась в статическом поле: найдите gcroot'ом, что держит память.
// Запуск:  dotnet run -c Release --project final -- delay
switch (args.FirstOrDefault() ?? "hoist")
{
    case "hoist": await Scenarios.Hoist(); break;
    case "delay": await Scenarios.Delay(); break;
    case "tcs": await Scenarios.Tcs(); break;
    case "event": await Scenarios.Events(); break;
    case "channel": await Scenarios.ChannelGrowth(); break;
    case "gen": await Scenarios.Generations(); break;
    case "hang-cancelled": Scenarios.HangCancelled(args.Length > 1 ? int.Parse(args[1]) : 120); break;
    case "service": await Service.RunAsync(); break;
    case "hang": await Scenarios.Hang(args.Length > 1 ? int.Parse(args[1]) : 120); break;
}

using System.Threading.Channels;

static class Scenarios
{
    static Pending? _leakyRegistry;                    // «долгоживущий» реестр: статическое поле

    // Прогон GC и измерение: сколько мегабайт в управляемой куче.
    // Живые данные: после полной блокирующей сборки берём размер кучи без фрагментации.
    // (GC.GetTotalMemory(true) в этом месте завышает на дыры в куче: см. §21.5.)
    static double Mb()
    {
        GC.Collect(2, GCCollectionMode.Forced, blocking: true, compacting: true);
        var info = GC.GetGCMemoryInfo(GCKind.FullBlocking);
        return (info.HeapSizeBytes - info.FragmentedBytes) / 1024.0 / 1024.0;
    }

    static bool Alive(WeakReference w)
    {
        GC.Collect(); GC.WaitForPendingFinalizers(); GC.Collect();
        return w.IsAlive;
    }

    // ---------- 1. Что держит локальная переменная ----------
    public static async Task Hoist()
    {
        var gate = new TaskCompletionSource();
        var usedAfter = new WeakReference(null);
        var notUsedAfter = new WeakReference(null);
        var nulled = new WeakReference(null);

        var t1 = UsedAfterAwait(gate.Task, usedAfter);
        var t2 = NotUsedAfterAwait(gate.Task, notUsedAfter);
        var t3 = NulledBeforeAwait(gate.Task, nulled);

        Console.WriteLine("Три метода приостановлены на await, буфер по 1 МБ у каждого.");
        Console.WriteLine($"  буфер нужен после await : жив = {Alive(usedAfter)}");
        Console.WriteLine($"  буфер после await не нужен: жив = {Alive(notUsedAfter)}");
        Console.WriteLine($"  буфер обнулён до await  : жив = {Alive(nulled)}");

        gate.SetResult();
        await Task.WhenAll(t1, t2, t3);
        Console.WriteLine($"После завершения методов: буфер из первого жив = {Alive(usedAfter)}");
    }

    static async Task UsedAfterAwait(Task wait, WeakReference probe)
    {
        var buffer = new byte[1_000_000];
        probe.Target = buffer;
        buffer[0] = 1;
        await wait;
        Console.Write(buffer[0] > 0 ? "" : "?");     // буфер нужен после await → станет полем машины состояний
    }

    static async Task NotUsedAfterAwait(Task wait, WeakReference probe)
    {
        var buffer = new byte[1_000_000];
        probe.Target = buffer;
        buffer[0] = 1;
        await wait;                                    // после await буфер не используется → остаётся локальной в MoveNext
    }

    static async Task NulledBeforeAwait(Task wait, WeakReference probe)
    {
        var buffer = new byte[1_000_000];
        probe.Target = buffer;
        buffer[0] = 1;
        byte first = buffer[0];
        buffer = null;
        await wait;
        Console.Write(first > 5 ? "?" : "");
    }

    // ---------- 2. Брошенные Task.Delay ----------
    public static async Task Delay()
    {
        const int N = 20_000;

        using var cts = new CancellationTokenSource();
        double before = Mb();
        var tasks = new List<Task>();
        for (int i = 0; i < N; i++) tasks.Add(WaitLongAsync(new byte[1000], cts.Token));
        double held = Mb();
        cts.Cancel();
        await DrainCancelledAsync(tasks);
        await Task.Delay(100);                         // дать стеку, на котором шло завершение задач, размотаться (см. §21.3)
        double afterCancel = Mb();
        Console.WriteLine($"С токеном:  {N} вызовов Delay(1 ч) с буфером 1 КБ → +{held - before:F1} МБ; после Cancel +{afterCancel - before:F1} МБ");
        cts.Dispose();
        Console.WriteLine($"            после cts.Dispose() +{Mb() - before:F1} МБ");

        before = Mb();
        var leaked = new List<Task>();
        for (int i = 0; i < N; i++) leaked.Add(WaitLongAsync(new byte[1000], CancellationToken.None));
        held = Mb();
        leaked.Clear();                                // выбрасываем свои ссылки на задачи
        double afterDrop = Mb();
        Console.WriteLine($"Без токена: {N} вызовов Delay(1 ч) → +{held - before:F1} МБ; после отказа от ссылок на задачи +{afterDrop - before:F1} МБ");
        Console.WriteLine("  (память держит очередь таймеров, а не вы: ссылок на задачи у вас уже нет)");
    }

    // Дождаться отменённых задач в ОТДЕЛЬНОМ методе: после выхода из него в кадре вызывающего не остаётся ссылки на последнюю задачу.
    static async Task DrainCancelledAsync(List<Task> tasks)
    {
        foreach (var t in tasks) { try { await t; } catch (OperationCanceledException) { } }
        tasks.Clear();
    }

    static async Task WaitLongAsync(byte[] payload, CancellationToken ct)
    {
        await Task.Delay(TimeSpan.FromHours(1), ct);
        Console.Write(payload.Length);
    }

    // ---------- 3. Реестр ожидающих запросов ----------
    public static async Task Tcs()
    {
        const int N = 20_000;

        var fixedReg = new Pending();
        using var cts = new CancellationTokenSource();
        double before = Mb();
        var w2 = new List<Task>();
        for (int i = 0; i < N; i++) w2.Add(fixedReg.WaitAsync(i, new byte[1000], cts.Token));
        double held = Mb() - before;
        cts.Cancel();
        await DrainCancelledAsync(w2);
        await Task.Delay(100);
        Console.WriteLine($"Реестр с токеном:    +{held:F1} МБ пока ждут, после отмены +{Mb() - before:F1} МБ, в реестре {fixedReg.Count} записей");

        _leakyRegistry = new Pending();
        before = Mb();
        var waiting = new List<Task>();
        for (int i = 0; i < N; i++) waiting.Add(_leakyRegistry.WaitAsync(i, new byte[1000], CancellationToken.None));
        waiting.Clear();
        Console.WriteLine($"Реестр без таймаута: ответы потерялись → +{Mb() - before:F1} МБ, в реестре {_leakyRegistry.Count} записей");
    }

    sealed class Pending
    {
        private readonly Dictionary<int, TaskCompletionSource<string>> _byId = new();
        public int Count => _byId.Count;

        public async Task WaitAsync(int id, byte[] context, CancellationToken ct)
        {
            var tcs = new TaskCompletionSource<string>(TaskCreationOptions.RunContinuationsAsynchronously);
            _byId[id] = tcs;
            using var reg = ct.Register(() =>
            {
                _byId.Remove(id);                      // снять запись из реестра…
                tcs.TrySetCanceled(ct);                // …и разбудить ждущего отменой
            });
            string reply = await tcs.Task;
            Console.Write(context.Length + reply.Length);
        }
    }

    // ---------- 4. Подписчики событий ----------
    public static async Task Events()
    {
        const int N = 10_000;
        var bus = new Bus();
        var probes = new List<WeakReference>();

        for (int i = 0; i < N; i++) { var s = new Screen(bus); probes.Add(new WeakReference(s)); }
        GC.Collect();
        Console.WriteLine($"Закрыто {N} экранов без отписки: живых = {probes.Count(p => p.IsAlive)}, подписчиков у издателя = {bus.Subscribers}");

        var bus2 = new Bus();
        var probes2 = new List<WeakReference>();
        for (int i = 0; i < N; i++) { var s = new Screen(bus2); probes2.Add(new WeakReference(s)); s.Dispose(); }
        GC.Collect();
        Console.WriteLine($"Закрыто {N} экранов с отпиской: живых = {probes2.Count(p => p.IsAlive)}, подписчиков у издателя = {bus2.Subscribers}");
        await Task.CompletedTask;
    }

    sealed class Bus                                  // долгоживущий издатель (синглтон в реальном приложении)
    {
        private event Func<string, Task>? Message;
        public int Subscribers => Message?.GetInvocationList().Length ?? 0;
        public void Subscribe(Func<string, Task> h) => Message += h;
        public void Unsubscribe(Func<string, Task> h) => Message -= h;
    }

    sealed class Screen : IDisposable
    {
        private readonly Bus _bus;
        private readonly byte[] _state = new byte[1000];
        private readonly Func<string, Task> _handler;

        public Screen(Bus bus)
        {
            _bus = bus;
            _handler = async text => { await Task.Yield(); _state[0] = (byte)text.Length; };   // лямбда держит this
            bus.Subscribe(_handler);
        }

        public void Dispose() => _bus.Unsubscribe(_handler);
    }

    // ---------- 5. Очередь без ограничения ----------
    public static async Task ChannelGrowth()
    {
        foreach (bool bounded in new[] { false, true })
        {
            Channel<byte[]> ch = bounded ? Channel.CreateBounded<byte[]>(100) : Channel.CreateUnbounded<byte[]>();
            double before = Mb();
            int produced = 0, consumed = 0;
            using var stop = new CancellationTokenSource();

            var producer = Task.Run(async () =>
            {
                try
                {
                    while (true)
                    {
                        await ch.Writer.WriteAsync(new byte[10_000], stop.Token);
                        produced++;
                    }
                }
                catch (OperationCanceledException) { }
            });
            var consumer = Task.Run(async () =>
            {
                try
                {
                    await foreach (var _ in ch.Reader.ReadAllAsync(stop.Token))
                    {
                        if (stop.IsCancellationRequested) break;
                        consumed++;
                        await Task.Delay(1);               // потребитель медленнее производителя
                    }
                }
                catch (OperationCanceledException) { }
            });

            await Task.Delay(2000);
            int queued = ch.Reader.Count;
            double heap = GC.GetTotalMemory(false) / 1024.0 / 1024.0 - before;
            Console.WriteLine($"{(bounded ? "Bounded(100)" : "Unbounded   ")} за 2 с произведено {produced}, обработано {consumed}, в очереди {queued}, куча ≈ +{heap:F0} МБ");
            stop.Cancel();
            await Task.WhenAll(producer, consumer);
            ch.Writer.TryComplete();
        }
    }

    // ---------- 6. Поколения ----------
    public static async Task Generations()
    {
        var gate = new TaskCompletionSource();
        var probe = new WeakReference(null);
        var waiter = HoldAsync(gate.Task, probe);
        object state = probe.Target!;
        Console.Write("Поколение состояния ждущего метода после сборок: ");
        Console.Write($"{GC.GetGeneration(state)} ");
        GC.Collect(0); Console.Write($"→ {GC.GetGeneration(state)} ");
        GC.Collect(1); Console.Write($"→ {GC.GetGeneration(state)}");
        Console.WriteLine();

        const int N = 100_000;
        int g0 = GC.CollectionCount(0), g1 = GC.CollectionCount(1), g2 = GC.CollectionCount(2);
        var pending = new List<Task>(N);
        for (int i = 0; i < N; i++) pending.Add(WaitLongAsync(new byte[200], CancellationToken.None));
        for (int i = 0; i < 3; i++) { GC.Collect(0); await Task.Yield(); }
        var info = GC.GetGCMemoryInfo(GCKind.Any);
        Console.WriteLine($"{N} ждущих методов: сборок gen0/gen1/gen2 за создание: {GC.CollectionCount(0) - g0}/{GC.CollectionCount(1) - g1}/{GC.CollectionCount(2) - g2}");
        Console.Write("Размеры поколений после последней сборки, МБ: ");
        for (int g = 0; g < 3; g++) Console.Write($"gen{g}={info.GenerationInfo[g].SizeAfterBytes / 1024.0 / 1024.0:F1} ");
        Console.WriteLine();
        GC.KeepAlive(pending);
        gate.SetResult();
        await waiter;
    }

    static async Task HoldAsync(Task wait, WeakReference probe)
    {
        var state = new byte[100];
        probe.Target = state;
        await wait;
        Console.Write(state.Length > 1000 ? "?" : "");
    }

    // ---------- 7. Утечка и пауза для инструментов ----------
    static List<Task>? _keepForDump;

    // Тот же сценарий с токеном, но в статическом поле осталась ОДНА отменённая задача (как «последняя ошибка» в логе):
    // она держит цепочку из всех 20 000 регистраций. Задание главы: найти gcroot'ом, что их держит.
    static Task? _lastCancelled;

    public static void HangCancelled(int seconds)
    {
        const int N = 20_000;
        using var cts = new CancellationTokenSource();
        var tasks = new List<Task>();
        for (int i = 0; i < N; i++) tasks.Add(WaitLongAsync(new byte[1000], cts.Token));
        cts.Cancel();
        _lastCancelled = tasks[^1];
        try { _lastCancelled.Wait(); } catch (AggregateException) { }
        tasks.Clear();
        Console.WriteLine($"PID {Environment.ProcessId}. Все отменены, список очищен; куча ≈ {Mb():F1} МБ. Завершится через {seconds} с.");
        Thread.Sleep(TimeSpan.FromSeconds(seconds));
    }

    public static async Task Hang(int seconds)
    {
        const int N = 50_000;
        _keepForDump = null;
        for (int i = 0; i < N; i++) _ = WaitLongAsync(new byte[1000], CancellationToken.None);
        Console.WriteLine($"PID {Environment.ProcessId}. Брошено {N} вызовов Delay(1 ч); куча ≈ {Mb():F0} МБ. Подключайтесь инструментами; завершится через {seconds} с.");
        await Task.Delay(TimeSpan.FromSeconds(seconds));
    }
}

// Глава 21, итог. Тот же сервис, утечки исправлены (режим service: dotnet run -c Release --project final -- service).
// Программа имитирует работу и печатает, сколько памяти и объектов осталось ПОСЛЕ неё.
using System.Collections.Concurrent;

static class Service
{
    public static async Task RunAsync()
    {
        const int N = 10_000;
        var bus = new Bus();
        var gateway = new Gateway();
        var screens = new List<WeakReference>();

        double before = Mb();
        using var cts = new CancellationTokenSource();

        // Утечка 1 и 2: запросы без ответа и «keep-alive» без токена.
        var requests = new List<Task>();
        for (int i = 0; i < N; i++) requests.Add(gateway.RequestAsync(i, new byte[1000], cts.Token));
        await gateway.KeepAliveAsync(100, cts.Token);                   // 100 ожиданий по 10 минут: «забытый» таймер

        // Утечка 3: экраны подписываются на шину и «закрываются».
        for (int i = 0; i < N; i++)
        {
            var screen = new Screen(bus);
            screens.Add(new WeakReference(screen));
            screen.Close();
        }

        // Конец «нагрузки»: клиенты ушли.
        cts.Cancel();                                         // «разрыв соединений»: токен отменён
        requests.Clear();                                     // про задачи запросов мы больше не помним
        await Task.Delay(200);                                // даём стеку завершения размотаться

        GC.Collect(); GC.WaitForPendingFinalizers(); GC.Collect();
        Console.WriteLine($"Живых экранов: {screens.Count(s => s.IsAlive)} из {N}");
        Console.WriteLine($"Записей в реестре шлюза: {gateway.Pending}");
        Console.WriteLine($"Куча: +{Mb() - before:F1} МБ");


    }

    static double Mb()
    {
        GC.Collect(2, GCCollectionMode.Forced, blocking: true, compacting: true);
        var info = GC.GetGCMemoryInfo(GCKind.FullBlocking);
        return (info.HeapSizeBytes - info.FragmentedBytes) / 1024.0 / 1024.0;
    }

}

sealed class Gateway
{
    private readonly ConcurrentDictionary<int, TaskCompletionSource<string>> _pending = new();
    public int Pending => _pending.Count;

    public async Task RequestAsync(int id, byte[] context, CancellationToken ct)
    {
        var tcs = new TaskCompletionSource<string>(TaskCreationOptions.RunContinuationsAsynchronously);
        _pending[id] = tcs;
        using var reg = ct.Register(() => tcs.TrySetCanceled(ct));   // токен разбудит ждущего отменой…
        try
        {
            string reply = await tcs.Task;           // ответ приходит из другого места; он может не прийти
            Console.Write(context.Length + reply.Length);
        }
        finally
        {
            _pending.TryRemove(id, out _);           // …а запись из реестра уходит при любом исходе
        }
    }

    public async Task KeepAliveAsync(int count, CancellationToken ct)
    {
        var waits = new List<Task>();
        for (int i = 0; i < count; i++) waits.Add(KeepOneAsync(new byte[10_000], ct));
        await Task.Delay(10);                        // «работаем дальше», не дожидаясь keep-alive
    }

    private static async Task KeepOneAsync(byte[] state, CancellationToken ct)
    {
        try
        {
            await Task.Delay(TimeSpan.FromMinutes(10), ct);      // таймер снимается при отмене
            Console.Write(state.Length);
        }
        catch (OperationCanceledException) { }
    }
}

sealed class Bus
{
    public event Func<string, Task>? Message;
    public void Subscribe(Func<string, Task> handler) => Message += handler;
    public void Unsubscribe(Func<string, Task> handler) => Message -= handler;
}

sealed class Screen
{
    private readonly Bus _bus;
    private readonly Func<string, Task> _handler;
    private readonly byte[] _state = new byte[1000];

    public Screen(Bus bus)
    {
        _bus = bus;
        _handler = async text => { await Task.Yield(); _state[0] = (byte)text.Length; };
        bus.Subscribe(_handler);
    }

    public void Close() => _bus.Unsubscribe(_handler);   // отписка при закрытии
}