9. Отмена и таймауты¶

public readonly struct CancellationToken : IEquatable<CancellationToken>
{
    private readonly CancellationTokenSource _source;

    public bool CanBeCanceled => _source != null;

    public void ThrowIfCancellationRequested()
    {
        if (IsCancellationRequested)
        {
            ThrowOperationCanceledException();
        }
    }

    private void ThrowOperationCanceledException()
    {
        throw new OperationCanceledException(SR.OperationCanceled, this);
    }
}

public class CancellationTokenSource : IDisposable
{
    private volatile States _state;                    // NotCanceledState → NotifyingState → NotifyingCompleteState
    private volatile ITimer _timer;                    // таймер для CancelAfter и конструктора с задержкой
    private volatile ManualResetEvent _kernelEvent;    // создаётся лениво, при обращении к WaitHandle
    private Registrations _registrations;              // список обработчиков, добавленных через Register

    public bool IsCancellationRequested => _state != States.NotCanceledState;
}

private void NotifyCancellation(bool throwOnFirstException)
{
    if (TransitionToCancellationRequested())
    {
        ExecuteCallbackHandlers(throwOnFirstException);
    }
}

private bool TransitionToCancellationRequested()
{
    if (!IsCancellationRequested && Interlocked.CompareExchange(ref _state, States.NotifyingState, States.NotCanceledState) == States.NotCanceledState)
    {
        ITimer timer = _timer;
        if (timer != null)
        {
            _timer = null;
            timer.Dispose();
        }
        _kernelEvent?.Set();
        return true;
    }
    return false;
}

private void ExecuteCallbackHandlers(bool throwOnFirstException)
{
    Registrations registrations = Interlocked.Exchange(ref _registrations, null);
    ...
    List<Exception> list = null;
    try
    {
        while (true)
        {
            ...                                            // взять ПЕРВЫЙ обработчик из списка и убрать его оттуда
            try
            {
                if (callbacks.SynchronizationContext != null)
                {
                    callbacks.SynchronizationContext.Send(…, callbacks);
                }
                else
                {
                    callbacks.ExecuteCallback();           // на потоке, который вызвал Cancel
                }
            }
            catch (Exception item) when (!throwOnFirstException)
            {
                (list ?? (list = new List<Exception>())).Add(item);
            }
        }
    }
    finally
    {
        _state = States.NotifyingCompleteState;
        ...
    }
    if (list == null)
    {
        return;
    }
    throw new AggregateException(list);
}

internal CancellationTokenRegistration Register(Delegate callback, object stateForCallback, SynchronizationContext syncContext, ExecutionContext executionContext)
{
    if (!IsCancellationRequested)
    {
        ...
        callbackNode = new CallbackNode(registrations);
        callbackNode.Callback = callback;
        ...
        registrations.EnterLock();
        try
        {
            id = (callbackNode.Id = registrations.NextAvailableId++);
            callbackNode.Next = registrations.Callbacks;
            ...
            registrations.Callbacks = callbackNode;
        }
        finally
        {
            registrations.ExitLock();
        }
        if (!IsCancellationRequested || !registrations.Unregister(id, callbackNode))
        {
            return new CancellationTokenRegistration(id, callbackNode);
        }
    }
    Invoke(callback, stateForCallback, this);
    return default;
}

private static readonly TimerCallback s_timerCallback = TimerCallback;

private void InitializeWithTimer(TimeSpan millisecondsDelay, TimeProvider timeProvider)
{
    ...
    _timer = new TimerQueueTimer(s_timerCallback, this, millisecondsDelay, Timeout.InfiniteTimeSpan, flowExecutionContext: false);
}

private static void TimerCallback(object state)
{
    ((CancellationTokenSource)state).NotifyCancellation(throwOnFirstException: false);
}

Console.WriteLine("== 1. Таймаут как отмена ==");
using (var cts = new CancellationTokenSource(TimeSpan.FromMilliseconds(450)))
{
    Task work = DelayLoopAsync(cts.Token);
    try { await work; }
    catch (OperationCanceledException e)
    {
        Console.WriteLine($"  Delay-цикл: {e.GetType().Name}, Status={work.Status}, токен исключения тот же: {e.CancellationToken == cts.Token}");
    }
}
using (var cts = new CancellationTokenSource(TimeSpan.FromMilliseconds(50)))
{
    Task work = SpinLoopAsync(cts.Token);
    try { await work; }
    catch (OperationCanceledException e)
    {
        Console.WriteLine($"  цикл с проверкой: {e.GetType().Name}, Status={work.Status}");
    }
}

private static Task Delay(uint millisecondsDelay, TimeProvider timeProvider, CancellationToken cancellationToken)
{
    if (!cancellationToken.IsCancellationRequested)
    {
        if (millisecondsDelay != 0)
        {
            if (!cancellationToken.CanBeCanceled)
            {
                return new DelayPromise(millisecondsDelay, timeProvider);
            }
            return new DelayPromiseWithCancellation(millisecondsDelay, timeProvider, cancellationToken);
        }
        return CompletedTask;
    }
    return FromCanceled(cancellationToken);
}

// DelayPromiseWithCancellation
_registration = token.UnsafeRegister((object state, CancellationToken cancellationToken) =>
{
    DelayPromiseWithCancellation delayPromiseWithCancellation = (DelayPromiseWithCancellation)state;
    delayPromiseWithCancellation.AtomicStateUpdate(64, 0);
    if (delayPromiseWithCancellation.TrySetCanceled(cancellationToken))
    {
        delayPromiseWithCancellation.Cleanup();
    }
}, this);

protected override void Cleanup()
{
    _registration.Dispose();
    base.Cleanup();
}

Console.WriteLine("== 2. Register ==");
using (var cts = new CancellationTokenSource())
{
    cts.Token.Register(() => Console.WriteLine($"  обработчик A (поток {Environment.CurrentManagedThreadId})"));
    cts.Token.Register(() => Console.WriteLine($"  обработчик B (поток {Environment.CurrentManagedThreadId})"));
    cts.Token.Register(() => throw new InvalidOperationException("из обработчика C"));
    Console.WriteLine($"  Cancel() из потока {Environment.CurrentManagedThreadId}");
    try { cts.Cancel(); }
    catch (AggregateException e) { Console.WriteLine($"  Cancel() бросил {e.GetType().Name}: {string.Join("; ", e.InnerExceptions.Select(x => x.Message))}"); }
    Console.WriteLine($"  Register на уже отменённом токене: ");
    cts.Token.Register(() => Console.WriteLine($"    выполнен сразу, внутри Register (поток {Environment.CurrentManagedThreadId})"));
}
using (var cts = new CancellationTokenSource())
{
    var sw = Stopwatch.StartNew();
    cts.Token.Register(() => Thread.Sleep(300));
    cts.Cancel();
    Console.WriteLine($"  Cancel() с медленным обработчиком вернулся через {sw.ElapsedMilliseconds} мс");
}
using (var cts = new CancellationTokenSource())
{
    var sw = Stopwatch.StartNew();
    cts.Token.Register(() => Thread.Sleep(300));
    Task cancel = cts.CancelAsync();
    Console.WriteLine($"  CancelAsync() вернул задачу через {sw.ElapsedMilliseconds} мс, обработчик ещё работает: {!cancel.IsCompleted}");
    await cancel;
    Console.WriteLine($"  задача CancelAsync завершилась через {sw.ElapsedMilliseconds} мс");
}

public Task CancelAsync()
{
    ...
    if (TransitionToCancellationRequested())
    {
        Registrations registrations = Volatile.Read(in _registrations);
        if (registrations != null)
        {
            ...
            if (flag)                                         // обработчики есть
            {
                return Task.Factory.StartNew((object s) =>
                {
                    ((CancellationTokenSource)s).ExecuteCallbackHandlers(throwOnFirstException: false);
                }, this, CancellationToken.None, TaskCreationOptions.DenyChildAttach, TaskScheduler.Default);
            }
        }
    }
    return Task.CompletedTask;
}

Console.WriteLine("== 3. Canceled против Faulted, таймаут против отмены ==");
using (var request = new CancellationTokenSource())
using (var timeout = new CancellationTokenSource(TimeSpan.FromMilliseconds(100)))
using (var linked = CancellationTokenSource.CreateLinkedTokenSource(timeout.Token, request.Token))
{
    try { await DelayLoopAsync(linked.Token); }
    catch (OperationCanceledException) when (request.IsCancellationRequested) { Console.WriteLine("  отменил вызывающий: не ошибка"); }
    catch (OperationCanceledException) when (timeout.IsCancellationRequested) { Console.WriteLine("  сработал таймаут операции"); }
}

try { await OperationAsync(ct); }
catch (OperationCanceledException) when (ct.IsCancellationRequested)
{
    // нас отменили сознательно: не ошибка, на уровне Error не логируем
}

var listener = new TcpListener(IPAddress.Loopback, 0);
listener.Start();                                                    // принимает соединения, но никогда не отвечает
using (var http = new HttpClient { Timeout = TimeSpan.FromMilliseconds(200) })
{
    string url = $"http://127.0.0.1:{((IPEndPoint)listener.LocalEndpoint).Port}/";
    try { await http.GetAsync(url); }
    catch (OperationCanceledException e) { Console.WriteLine($"  HttpClient.Timeout: {e.GetType().Name}, InnerException = {e.InnerException?.GetType().Name}"); }
    using var byCaller = new CancellationTokenSource(TimeSpan.FromMilliseconds(200));
    try { await http.GetAsync(url, byCaller.Token); }
    catch (OperationCanceledException e) { Console.WriteLine($"  отмена токеном вызывающего: {e.GetType().Name}, InnerException = {e.InnerException?.GetType().Name ?? "нет"}"); }
}
listener.Stop();

private sealed class Linked2CancellationTokenSource : CancellationTokenSource
{
    private readonly CancellationTokenRegistration _reg1;
    private readonly CancellationTokenRegistration _reg2;

    internal Linked2CancellationTokenSource(CancellationToken token1, CancellationToken token2)
    {
        _reg1 = token1.UnsafeRegister(LinkedNCancellationTokenSource.s_linkedTokenCancelDelegate, this);
        _reg2 = token2.UnsafeRegister(LinkedNCancellationTokenSource.s_linkedTokenCancelDelegate, this);
    }

    protected override void Dispose(bool disposing)
    {
        if (disposing && !_disposed)
        {
            _reg1.Dispose();
            _reg2.Dispose();
            base.Dispose(disposing);
        }
    }
}

// LinkedNCancellationTokenSource
internal static readonly Action<object> s_linkedTokenCancelDelegate = (object s) =>
{
    ((CancellationTokenSource)s).NotifyCancellation(throwOnFirstException: false);
};

Console.WriteLine("== 4. Связанные токены ==");
using (var parent = new CancellationTokenSource())
{
    Console.WriteLine($"  обработчиков на родителе до связывания: {Probes.RegistrationsOn(parent)}");
    var linked = CancellationTokenSource.CreateLinkedTokenSource(parent.Token);
    Console.WriteLine($"  после CreateLinkedTokenSource: {Probes.RegistrationsOn(parent)}");
    linked.Dispose();
    Console.WriteLine($"  после linked.Dispose(): {Probes.RegistrationsOn(parent)}");

    const int N = 100_000;
    long before = Memory();
    var kept = new List<CancellationTokenSource>(N);
    for (int i = 0; i < N; i++)
        kept.Add(CancellationTokenSource.CreateLinkedTokenSource(parent.Token));
    kept.Clear();                                                    // ссылки на связанные источники выброшены, но не освобождены
    Console.WriteLine($"  {N:N0} связанных источников без Dispose: обработчиков на родителе {Probes.RegistrationsOn(parent):N0}, память +{(Memory() - before) / 1024:N0} КБ");
    before = Memory();
    for (int i = 0; i < N; i++)
        CancellationTokenSource.CreateLinkedTokenSource(parent.Token).Dispose();
    Console.WriteLine($"  {N:N0} с Dispose: обработчиков на родителе {Probes.RegistrationsOn(parent):N0} (прежние остались), память +{(Memory() - before) / 1024:N0} КБ");
}

// плохо: Delay остаётся в очереди таймеров до срока, даже если task давно завершилась
Task winner = await Task.WhenAny(task, Task.Delay(timeout));
if (winner != task) throw new TimeoutException();

// хорошо (.NET 6+)
await task.WaitAsync(timeout, ct);

private Task WaitAsync(uint millisecondsTimeout, TimeProvider timeProvider, CancellationToken cancellationToken)
{
    if (IsCompleted || (!cancellationToken.CanBeCanceled && millisecondsTimeout == uint.MaxValue))
    {
        return this;
    }
    if (cancellationToken.IsCancellationRequested)
    {
        return FromCanceled(cancellationToken);
    }
    if (millisecondsTimeout == 0)
    {
        return FromException(new TimeoutException());
    }
    return new CancellationPromise<VoidTaskResult>(this, millisecondsTimeout, timeProvider, cancellationToken);
}

// CancellationPromise<TResult>
internal CancellationPromise(Task source, uint millisecondsDelay, TimeProvider timeProvider, CancellationToken token)
{
    _task = source;
    source.AddCompletionAction(this);                         // подписаться на исходную задачу
    ... _timer = new TimerQueueTimer(…TrySetException(new TimeoutException())…)
    _registration = token.UnsafeRegister(…TrySetCanceled(cancellationToken)…, this);
}

private void Cleanup()
{
    _registration.Dispose();
    _timer?.Dispose();
    _task.RemoveContinuation(this);
}

Console.WriteLine("== 5. WaitAsync против WhenAny + Delay ==");
var swWait = Stopwatch.StartNew();
Task slow = SlowWorkAsync();
try { await slow.WaitAsync(TimeSpan.FromMilliseconds(200)); }
catch (TimeoutException) { Console.WriteLine($"  WaitAsync: TimeoutException через {swWait.ElapsedMilliseconds} мс, slow.Status={slow.Status}"); }
await slow;
Console.WriteLine($"  slow всё равно доработала: {swWait.ElapsedMilliseconds} мс");
{
    int baseline = Probes.TimersInQueue();
    for (int i = 0; i < 1000; i++)
        await Task.WhenAny(Task.CompletedTask, Task.Delay(TimeSpan.FromMinutes(5)));
    Console.WriteLine($"  1000 × WhenAny(задача, Delay(5 мин)): записей в очереди таймеров +{Probes.TimersInQueue() - baseline}");
    baseline = Probes.TimersInQueue();
    for (int i = 0; i < 1000; i++)
        await Task.CompletedTask.WaitAsync(TimeSpan.FromMinutes(5));
    Console.WriteLine($"  1000 × задача.WaitAsync(5 мин): записей в очереди таймеров +{Probes.TimersInQueue() - baseline}");
    baseline = Probes.TimersInQueue();
    for (int i = 0; i < 1000; i++)
    {
        using var stop = new CancellationTokenSource();
        Task delay = Task.Delay(TimeSpan.FromMinutes(5), stop.Token);
        await Task.WhenAny(Task.CompletedTask, delay);
        stop.Cancel();
    }
    Console.WriteLine($"  1000 × WhenAny + отмена Delay токеном: записей в очереди таймеров +{Probes.TimersInQueue() - baseline}");
}

protected virtual void Dispose(bool disposing)
{
    if (!disposing || _disposed)
    {
        return;
    }
    ITimer timer = _timer;
    if (timer != null)
    {
        _timer = null;
        timer.Dispose();
    }
    _registrations = null;
    if (_kernelEvent != null)
    {
        ...
        manualResetEvent.Dispose();
    }
    _disposed = true;
}

Console.WriteLine("== 6. CancellationTokenSource с таймером ==");
WeakReference notDisposed = MakeWithTimer(dispose: false);
WeakReference disposed = MakeWithTimer(dispose: true);
for (int i = 0; i < 3; i++) { GC.Collect(); GC.WaitForPendingFinalizers(); }
Console.WriteLine($"  источник на 30 с без Dispose жив после GC: {notDisposed.IsAlive}; с Dispose: {disposed.IsAlive}");