11. Асинхронные потоки и Channel<T>¶

public static IEnumerable<int> Numbers(int max)
{
    Console.WriteLine("    [итератор] тело началось");
    for (int i = 1; i <= max; i++)
        yield return i;
    Console.WriteLine("    [итератор] тело закончилось");
}

private sealed class <Numbers>d__0 : IEnumerable<int>, IEnumerable, IEnumerator<int>, IEnumerator, IDisposable
{
    private int <>1__state;
    private int <>2__current;
    private int <>l__initialThreadId;
    private int max;
    private int <i>5__2;

    public <Numbers>d__0(int <>1__state)
    {
        this.<>1__state = <>1__state;
        <>l__initialThreadId = Environment.CurrentManagedThreadId;
    }

    private bool MoveNext()
    {
        switch (<>1__state)
        {
        default:
            return false;
        case 0:
            <>1__state = -1;
            Console.WriteLine("    [итератор] тело началось");
            <i>5__2 = 1;
            break;
        case 1:
            <>1__state = -1;
            <i>5__2++;
            break;
        }
        if (<i>5__2 <= max)
        {
            <>2__current = <i>5__2;
            <>1__state = 1;
            return true;
        }
        Console.WriteLine("    [итератор] тело закончилось");
        return false;
    }

    IEnumerator<int> IEnumerable<int>.GetEnumerator()
    {
        <Numbers>d__0 obj;
        if (<>1__state == -2 && <>l__initialThreadId == Environment.CurrentManagedThreadId)
        {
            <>1__state = 0;
            obj = this;
        }
        else
        {
            obj = new <Numbers>d__0(0);
        }
        ...
    }
}

Console.WriteLine("== 1. Обычный итератор ==");
IEnumerable<int> numbers = Sources.Numbers(2);
Console.WriteLine("  метод вызван, тело ещё не началось");
foreach (int n in numbers)
    Console.WriteLine($"  получено {n}");
IEnumerable<int> seq = Sources.Numbers(1);
IEnumerator<int> e1 = seq.GetEnumerator();
IEnumerator<int> e2 = seq.GetEnumerator();
Console.WriteLine($"  первый GetEnumerator вернул саму последовательность: {ReferenceEquals(seq, e1)}, второй: {ReferenceEquals(seq, e2)}");
Console.WriteLine($"  тип: {seq.GetType().Name.Replace("<", "‹").Replace(">", "›")}");

public static async IAsyncEnumerable<int> CountAsync(int max, [EnumeratorCancellation] CancellationToken ct = default)
{
    try
    {
        for (int i = 1; i <= max; i++)
        {
            await Task.Delay(100, ct);
            yield return i;
        }
    }
    finally
    {
        Console.WriteLine("    [итератор] finally");
    }
}

await foreach (int n in CountAsync(10, ct))
{
    // тело цикла
}

IAsyncEnumerator<int> e = CountAsync(10, ct).GetAsyncEnumerator();
try
{
    while (await e.MoveNextAsync())      // ValueTask<bool>
    {
        int n = e.Current;
        // тело цикла
    }
}
finally
{
    await e.DisposeAsync();              // и при break, и при исключении
}

Console.WriteLine("== 2. await foreach ==");
using (var cts = new CancellationTokenSource(TimeSpan.FromMilliseconds(450)))
{
    try
    {
        await foreach (int n in Sources.CountAsync(10, cts.Token))
            Console.WriteLine($"  получено {n}");
    }
    catch (OperationCanceledException e) { Console.WriteLine($"  отменено по таймауту: {e.GetType().Name}"); }
}
Console.WriteLine("  --- break на втором элементе:");
await foreach (int n in Sources.CountAsync(10))
{
    Console.WriteLine($"  получено {n}");
    if (n == 2) break;                          // DisposeAsync -> finally в итераторе
}
Console.WriteLine("  --- WithCancellation:");
using (var cts = new CancellationTokenSource(TimeSpan.FromMilliseconds(250)))
{
    try
    {
        await foreach (int n in Sources.CountAsync(10).WithCancellation(cts.Token))
            Console.WriteLine($"  получено {n}");
    }
    catch (OperationCanceledException) { Console.WriteLine("  отменено токеном из WithCancellation"); }
}
var asyncSeq = Sources.CountAsync(1);
IAsyncEnumerator<int> a1 = asyncSeq.GetAsyncEnumerator();
IAsyncEnumerator<int> a2 = asyncSeq.GetAsyncEnumerator();
Console.WriteLine($"  первый GetAsyncEnumerator вернул саму последовательность: {ReferenceEquals(asyncSeq, a1)}, второй: {ReferenceEquals(asyncSeq, a2)}");
await a1.DisposeAsync();
await a2.DisposeAsync();

Console.WriteLine("  --- AsyncLocal, записанный внутри итератора:");
foreach (bool pause in new[] { false, true })
{
    Sources.Ambient.Value = 1;
    Console.WriteLine($"  итератор {(pause ? "с await Task.Yield()" : "без await")}:");
    await foreach (int n in Sources.WritesAmbientAsync(pause))
        Console.WriteLine($"  потребитель, элемент {n}: Ambient = {Sources.Ambient.Value}");
}

ValueTask<bool> IAsyncEnumerator<int>.MoveNextAsync()
{
    if (<>1__state == -2)
    {
        return default;
    }
    <>v__promiseOfValueOrEnd.Reset();
    <CountAsync>d__1 stateMachine = this;
    <>t__builder.MoveNext(ref stateMachine);
    short version = <>v__promiseOfValueOrEnd.Version;
    if (<>v__promiseOfValueOrEnd.GetStatus(version) == ValueTaskSourceStatus.Succeeded)
    {
        return new ValueTask<bool>(<>v__promiseOfValueOrEnd.GetResult(version));
    }
    return new ValueTask<bool>(this, version);
}

ValueTask IAsyncDisposable.DisposeAsync()
{
    if (<>1__state >= -1)
    {
        throw new NotSupportedException();
    }
    if (<>1__state == -2)
    {
        return default;
    }
    <>w__disposeMode = true;
    <>v__promiseOfValueOrEnd.Reset();
    <CountAsync>d__1 stateMachine = this;
    <>t__builder.MoveNext(ref stateMachine);
    return new ValueTask(this, <>v__promiseOfValueOrEnd.Version);
}

private sealed class <CountAsync>d__1 : IAsyncEnumerable<int>, IAsyncEnumerator<int>, IAsyncDisposable, IValueTaskSource<bool>, IValueTaskSource, IAsyncStateMachine
{
    public int <>1__state;
    public AsyncIteratorMethodBuilder <>t__builder;
    public ManualResetValueTaskSourceCore<bool> <>v__promiseOfValueOrEnd;
    private int <>2__current;
    private bool <>w__disposeMode;
    private CancellationTokenSource <>x__combinedTokens;
    private int <>l__initialThreadId;
    private CancellationToken ct;
    public CancellationToken <>3__ct;
    private int max;
    private int <i>5__2;
    private TaskAwaiter <>u__1;
    ...
}

public struct AsyncIteratorMethodBuilder
{
    private Task<VoidTaskResult> m_task;

    public static AsyncIteratorMethodBuilder Create()
    {
        return default;
    }

    public void MoveNext<TStateMachine>(ref TStateMachine stateMachine) where TStateMachine : IAsyncStateMachine
    {
        AsyncMethodBuilderCore.Start(ref stateMachine);
    }

    public void AwaitUnsafeOnCompleted<TAwaiter, TStateMachine>(ref TAwaiter awaiter, ref TStateMachine stateMachine) where TAwaiter : ICriticalNotifyCompletion where TStateMachine : IAsyncStateMachine
    {
        AsyncTaskMethodBuilder<VoidTaskResult>.AwaitUnsafeOnCompleted(ref awaiter, ref stateMachine, ref m_task);
    }

    public void Complete()
    {
        if (m_task == null)
        {
            m_task = Task.s_cachedCompleted;
            return;
        }
        AsyncTaskMethodBuilder<VoidTaskResult>.SetExistingTaskResult(m_task, default);
        ...
    }
}

IL_00b0:
awaiter.GetResult();
<>2__current = <i>5__2;                     // 1. положить элемент в Current
num = (<>1__state = -4);                    // 2. запомнить: ждём следующего MoveNextAsync
goto IL_01a6;
...
IL_01a6:
<>v__promiseOfValueOrEnd.SetResult(result: true);   // 3. завершить «обещание» MoveNextAsync значением true

// конец тела:
<>1__state = -2;
<>2__current = 0;
<>t__builder.Complete();
<>v__promiseOfValueOrEnd.SetResult(result: false);  // MoveNextAsync вернёт false

Console.WriteLine();
Console.WriteLine("== 3. Аллокации на элемент: 100 000 элементов ==");
await Measure("синхронный проход, yield return без await", async () => { await foreach (int _ in Sources.SyncAsync(100_000)) { } });
await Measure("с паузой await Task.Yield() на каждом элементе", async () => { await foreach (int _ in Sources.YieldingAsync(100_000)) { } });
await Measure("то же, ConfigureAwait(false)", async () => { await foreach (int _ in Sources.YieldingAsync(100_000).ConfigureAwait(false)) { } });

Console.WriteLine("== 4. LINQ над IAsyncEnumerable (BCL, .NET 10) ==");
List<int> evens = await Sources.CountAsync(6).Where(n => n % 2 == 0).Select(n => n * 10).ToListAsync();
Console.WriteLine($"  чётные ×10: {string.Join(", ", evens)}");
int count = await Sources.SyncAsync(1000).CountAsync(n => n % 7 == 0);
Console.WriteLine($"  CountAsync(кратные 7 из 1000): {count}");
int firstBig = await Sources.SyncAsync(1000).FirstAsync(n => n > 500);
Console.WriteLine($"  FirstAsync(> 500): {firstBig}");
List<int> asyncPredicate = await Sources.SyncAsync(10).Where(async (n, ct) => { await Task.Yield(); return n % 3 == 0; }).ToListAsync();
Console.WriteLine($"  Where с async-предикатом: {string.Join(", ", asyncPredicate)}");
List<int> taken = await Sources.CountAsync(10).Take(3).ToListAsync();
Console.WriteLine($"  Take(3): {string.Join(", ", taken)} (источник остановлен: см. finally выше)");
using (var cts = new CancellationTokenSource(TimeSpan.FromMilliseconds(250)))
{
    try { await Sources.CountAsync(10).ToListAsync(cts.Token); }
    catch (OperationCanceledException) { Console.WriteLine("  ToListAsync(ct) отменён"); }
}

public static Channel<T> CreateUnbounded<T>()
{
    return new UnboundedChannel<T>(runContinuationsAsynchronously: true);
}

public static Channel<T> CreateUnbounded<T>(UnboundedChannelOptions options)
{
    if (options.SingleReader)
    {
        return new SingleConsumerUnboundedChannel<T>(!options.AllowSynchronousContinuations);
    }
    return new UnboundedChannel<T>(!options.AllowSynchronousContinuations);
}

public static Channel<T> CreateBounded<T>(BoundedChannelOptions options, Action<T>? itemDropped)
{
    if (options.Capacity <= 0)
    {
        return new RendezvousChannel<T>(options.FullMode, !options.AllowSynchronousContinuations, itemDropped);
    }
    return new BoundedChannel<T>(options.Capacity, options.FullMode, !options.AllowSynchronousContinuations, itemDropped);
}

public static Channel<T> CreateUnboundedPrioritized<T>(UnboundedPrioritizedChannelOptions<T> options)
{
    return new UnboundedPrioritizedChannel<T>(!options.AllowSynchronousContinuations, options.Comparer);
}

var channel = Channel.CreateBounded<int>(new BoundedChannelOptions(capacity: 3)
{
    FullMode = BoundedChannelFullMode.Wait,    // производитель будет ждать при переполнении
    SingleReader = true,
});

var producer = Task.Run(async () =>
{
    for (int i = 1; i <= 8; i++)
        await channel.Writer.WriteAsync(i);    // ждёт, если очередь полна (обратное давление)
    channel.Writer.Complete();                 // сигнал «больше не будет»
});

var consumer = Task.Run(async () =>
{
    await foreach (var item in channel.Reader.ReadAllAsync())
        await Task.Delay(200);                 // медленный потребитель
});

await Task.WhenAll(producer, consumer);

public virtual async IAsyncEnumerable<T> ReadAllAsync([EnumeratorCancellation] CancellationToken cancellationToken = default)
{
    while (await WaitToReadAsync(cancellationToken).ConfigureAwait(continueOnCapturedContext: false))
    {
        T item;
        while (TryRead(out item))
        {
            yield return item;
        }
    }
}

public override bool TryWrite(T item)
{
    ...
    lock (parent.SyncObj)
    {
        if (parent._doneWriting != null)
        {
            return false;
        }
        int count = parent._items.Count;
        if (count != 0)
        {
            if (count < parent._bufferedCapacity)
            {
                parent._items.EnqueueTail(item);
                return true;
            }
            if (parent._mode == BoundedChannelFullMode.Wait)
            {
                return false;
            }
            if (parent._mode == BoundedChannelFullMode.DropWrite)
            {
                parent._itemDropped?.Invoke(item);
                return true;
            }
            T obj = ((parent._mode == BoundedChannelFullMode.DropNewest) ? parent._items.DequeueTail() : parent._items.DequeueHead());
            parent._items.EnqueueTail(item);
            parent._itemDropped?.Invoke(obj);
            return true;
        }
        ...                                       // канал пуст: отдать элемент ждущему читателю или положить в очередь
    }
}

Console.WriteLine("== 5б. Режимы переполнения: 6 записей в канал на 3 элемента без читателя ==");
foreach (var mode in new[] { BoundedChannelFullMode.DropOldest, BoundedChannelFullMode.DropNewest, BoundedChannelFullMode.DropWrite })
{
    var dropped = new List<int>();
    var ch = Channel.CreateBounded<int>(new BoundedChannelOptions(3) { FullMode = mode }, dropped.Add);
    var results = new List<bool>();
    for (int i = 1; i <= 6; i++) results.Add(ch.Writer.TryWrite(i));
    ch.Writer.Complete();
    var kept = new List<int>();
    await foreach (int item in ch.Reader.ReadAllAsync()) kept.Add(item);
    Console.WriteLine($"  {mode,-10}: TryWrite → {string.Join("", results.Select(r => r ? "T" : "F"))}, выброшено [{string.Join(", ", dropped)}], осталось [{string.Join(", ", kept)}]");
}
{
    var ch = Channel.CreateBounded<int>(new BoundedChannelOptions(3) { FullMode = BoundedChannelFullMode.Wait });
    for (int i = 1; i <= 3; i++) ch.Writer.TryWrite(i);
    Console.WriteLine($"  Wait      : четвёртый TryWrite → {ch.Writer.TryWrite(4)}; WriteAsync вернул ValueTask, завершённый: {ch.Writer.WriteAsync(4).IsCompleted}");
}

Wait, 8 записей, читатель берёт по 200 мс, ёмкость 3:
     5 мс  записано 1
     7 мс  записано 2
     7 мс  записано 3
    10 мс  прочитано 1
    11 мс  записано 4
   211 мс  прочитано 2
   211 мс  записано 5
   411 мс  прочитано 3
   412 мс  записано 6
   612 мс  прочитано 4
   612 мс  записано 7
   812 мс  прочитано 5
   812 мс  записано 8
  1012 мс  прочитано 6 …

if (parent._mode == BoundedChannelFullMode.Wait)
{
    if (!cancellationToken.CanBeCanceled)
    {
        BlockedWriteAsyncOperation<T> writerSingleton = _writerSingleton;
        if (writerSingleton.TryOwnAndReset())
        {
            writerSingleton.Item = item;
            ChannelUtilities.Enqueue(ref parent._blockedWritersHead, writerSingleton);
            return writerSingleton.ValueTask;
        }
    }
    BlockedWriteAsyncOperation<T> blockedWriteAsyncOperation = new BlockedWriteAsyncOperation<T>(runContinuationsAsynchronously: true, cancellationToken, pooled: false, _parent.CancellationCallbackDelegate)
    {
        Item = item
    };
    ChannelUtilities.Enqueue(ref parent._blockedWritersHead, blockedWriteAsyncOperation);
    return blockedWriteAsyncOperation.ValueTask;
}

Console.WriteLine();
Console.WriteLine("== 5в. Завершение канала ==");
var closing = Channel.CreateUnbounded<int>();
closing.Writer.TryWrite(1);
closing.Writer.TryWrite(2);
closing.Writer.Complete();
Console.WriteLine($"  после Complete: TryWrite → {closing.Writer.TryWrite(3)}; Completion завершена: {closing.Reader.Completion.IsCompleted} (остались непрочитанные)");
try { await closing.Writer.WriteAsync(3); }
catch (ChannelClosedException) { Console.WriteLine("  WriteAsync в закрытый канал: ChannelClosedException"); }
await foreach (int item in closing.Reader.ReadAllAsync()) Console.WriteLine($"  дочитано {item}");
Console.WriteLine($"  после дочитывания Completion: {closing.Reader.Completion.Status}");
var failing = Channel.CreateUnbounded<int>();
failing.Writer.TryWrite(1);
failing.Writer.Complete(new InvalidOperationException("производитель упал"));
try { await foreach (int item in failing.Reader.ReadAllAsync()) Console.WriteLine($"  получено {item}"); }
catch (InvalidOperationException e) { Console.WriteLine($"  Complete(exception): читатель получил «{e.Message}» после дочитывания"); }
var rendezvous = Channel.CreateBounded<int>(0);
Console.WriteLine($"  capacity 0 (рандеву): {rendezvous.GetType().Name}, TryWrite без читателя → {rendezvous.Writer.TryWrite(1)}");

Console.WriteLine();
Console.WriteLine("== 5г. Несколько читателей ==");
var jobs = Channel.CreateUnbounded<int>(new UnboundedChannelOptions { SingleReader = false });
var handled = new int[3];
Task[] workers = Enumerable.Range(0, 3).Select(id => Task.Run(async () =>
{
    await foreach (int job in jobs.Reader.ReadAllAsync())
    {
        await Task.Delay(10);
        Interlocked.Increment(ref handled[id]);
    }
})).ToArray();
for (int i = 0; i < 30; i++) jobs.Writer.TryWrite(i);
jobs.Writer.Complete();
await Task.WhenAll(workers);
Console.WriteLine($"  30 заданий обработано читателями: {string.Join(" + ", handled)} = {handled.Sum()}; Completion: {jobs.Reader.Completion.Status}");

private protected void SignalCompletion()
{
    ...
    object capturedContext = _capturedContext;
    if ((capturedContext == null || capturedContext is ExecutionContext) ? true : false)
    {
        if (RunContinuationsAsynchronously)
        {
            UnsafeQueueSetCompletionAndInvokeContinuation();      // поставить в пул
            return;
        }
    }
    else
    {
        SynchronizationContext synchronizationContext = ...;
        if (synchronizationContext != null)
        {
            if (RunContinuationsAsynchronously || synchronizationContext != SynchronizationContext.Current)
            {
                synchronizationContext.Post(…);                    // вернуть в контекст читателя
                return;
            }
        }
        else
        {
            TaskScheduler taskScheduler = ...;
            if (RunContinuationsAsynchronously || taskScheduler != TaskScheduler.Current)
            {
                Task.Factory.StartNew(…, taskScheduler);           // в планировщик читателя
                return;
            }
        }
    }
    SetCompletionAndInvokeContinuation();                          // иначе — выполнить продолжение прямо здесь
}

Console.WriteLine();
Console.WriteLine("== 6. AllowSynchronousContinuations ==");
foreach (bool allow in new[] { false, true })
{
    var ch = Channel.CreateUnbounded<int>(new UnboundedChannelOptions { SingleReader = true, AllowSynchronousContinuations = allow });
    Flag.InsideWrite = false;
    bool ranInsideTryWrite = false;
    Task reader = Task.Run(async () =>
    {
        await ch.Reader.ReadAsync();                    // читатель ждёт на пустом канале
        ranInsideTryWrite = Flag.InsideWrite;
        Thread.Sleep(200);                              // «обработка» в продолжении читателя
    });
    await Task.Delay(100);
    long tryWriteMs = 0;
    await Task.Run(() =>
    {
        var sw = Stopwatch.StartNew();
        Flag.InsideWrite = true;
        ch.Writer.TryWrite(1);
        Flag.InsideWrite = false;
        tryWriteMs = sw.ElapsedMilliseconds;
    });
    await reader;
    Console.WriteLine($"  AllowSynchronousContinuations={allow,-5}: TryWrite занял {tryWriteMs,3} мс, продолжение читателя выполнилось внутри TryWrite: {ranInsideTryWrite}");
}

Console.WriteLine();
Console.WriteLine("== 7. Стоимость элемента канала: 200 000 элементов, один писатель и один читатель ==");
foreach (var (name, options) in new (string, BoundedChannelOptions)[]
{
    ("Bounded(64)", new BoundedChannelOptions(64)),
    ("Bounded(64), SingleReader + SingleWriter", new BoundedChannelOptions(64) { SingleReader = true, SingleWriter = true }),
})
{
    var (ms, bytes) = await Pump(options);
    Console.WriteLine($"  {name,-42}: {ms,4} мс, {bytes:F1} байт на элемент");
}