10. Task API: что и когда¶

// Task.Run
public static Task Run(Action action)
{
    return InternalStartNew(null, action, null, default, TaskScheduler.Default, TaskCreationOptions.DenyChildAttach, InternalTaskOptions.None);
}

public static Task Run(Func<Task?> function, CancellationToken cancellationToken)
{
    ...
    return new UnwrapPromise<VoidTaskResult>(Task<Task>.Factory.StartNew(function, cancellationToken, TaskCreationOptions.DenyChildAttach, TaskScheduler.Default), lookForOce: true);
}

// TaskFactory.StartNew(Action)
public Task StartNew(Action action)
{
    Task internalCurrent = Task.InternalCurrent;
    return Task.InternalStartNew(internalCurrent, action, null, m_defaultCancellationToken, GetDefaultScheduler(internalCurrent), m_defaultCreationOptions, InternalTaskOptions.None);
}

private TaskScheduler GetDefaultScheduler(Task currTask)
{
    TaskScheduler? taskScheduler = m_defaultScheduler;
    if (taskScheduler == null)
    {
        if (currTask == null || (currTask.CreationOptions & TaskCreationOptions.HideScheduler) != 0)
        {
            return TaskScheduler.Default;
        }
        taskScheduler = currTask.ExecutingTaskScheduler;
    }
    return taskScheduler;
}

var sw = Stopwatch.StartNew();
Task<Task> outer = Task.Factory.StartNew(async () => await Task.Delay(300));
await outer;
Console.WriteLine($"  await StartNew(async …)          : {sw.ElapsedMilliseconds} мс (дождались только запуска, Task<Task>)");
sw.Restart();
await Task.Factory.StartNew(async () => await Task.Delay(300)).Unwrap();
Console.WriteLine($"  await StartNew(async …).Unwrap() : {sw.ElapsedMilliseconds} мс");
sw.Restart();
await Task.Run(async () => await Task.Delay(300));
Console.WriteLine($"  await Task.Run(async …)          : {sw.ElapsedMilliseconds} мс (разворачивает сам)");

TaskScheduler exclusive = new ConcurrentExclusiveSchedulerPair().ExclusiveScheduler;
await Task.Factory.StartNew(() =>
{
    Console.WriteLine($"  внутри задачи на {exclusive.GetType().Name}:");
    Console.WriteLine($"    StartNew(…) пошла в             : {Task.Factory.StartNew(() => TaskScheduler.Current.GetType().Name).Result}");
    Console.WriteLine($"    Task.Run(…) пошла в             : {Task.Run(() => TaskScheduler.Current.GetType().Name).Result}");
    Console.WriteLine($"    StartNew(…, HideScheduler) в    : {Task.Factory.StartNew(() => TaskScheduler.Current.GetType().Name, CancellationToken.None, TaskCreationOptions.HideScheduler, TaskScheduler.Current).Result}");
}, CancellationToken.None, TaskCreationOptions.None, exclusive);

Task viaStart = Task.Factory.StartNew(() => { });
Task viaRun = Task.Run(() => { });
await Task.WhenAll(viaStart, viaRun);
Console.WriteLine($"  флаги StartNew: 0x{Flags(viaStart):X8}, Task.Run: 0x{Flags(viaRun):X8}");
Console.WriteLine($"  родитель StartNew + дочерняя AttachedToParent ждёт дочернюю: {await ParentWaitsAsync(useRun: false)}");
Console.WriteLine($"  родитель Task.Run   + дочерняя AttachedToParent ждёт дочернюю: {await ParentWaitsAsync(useRun: true)}");
bool longPool = false, longDedicated = false;
await Task.Run(() => longPool = Thread.CurrentThread.IsThreadPoolThread);
await Task.Factory.StartNew(() => longDedicated = Thread.CurrentThread.IsThreadPoolThread, CancellationToken.None, TaskCreationOptions.LongRunning, TaskScheduler.Default);
Console.WriteLine($"  поток пула: Task.Run = {longPool}, LongRunning = {longDedicated}");

protected internal override void QueueTask(Task task)
{
    TaskCreationOptions options = task.Options;
    if ((options & TaskCreationOptions.LongRunning) != 0)
    {
        Thread thread = new Thread(s_longRunningThreadWork);
        thread.IsBackground = true;
        thread.Name = ".NET Long Running Task";
        thread.UnsafeStart(task);
    }
    else
    {
        ThreadPool.UnsafeQueueUserWorkItemInternal(task, (options & TaskCreationOptions.PreferFairness) == 0);
    }
}

public Task ContinueWith(Action<Task> continuationAction)
{
    return ContinueWith(continuationAction, TaskScheduler.Current, default, TaskContinuationOptions.None);
}

private Task ContinueWith(Action<Task, object> continuationAction, object state, TaskScheduler scheduler, CancellationToken cancellationToken, TaskContinuationOptions continuationOptions)
{
    ...
    Task task = new ContinuationTaskFromTask(this, continuationAction, state, creationOptions, internalOptions);
    ContinueWithCore(task, scheduler, cancellationToken, continuationOptions);
    return task;
}

// ContinueWithTaskContinuation.Run
task.m_taskScheduler = m_taskScheduler;
...
task.ScheduleAndStart(needsProtection: true);

Console.WriteLine("== 2. ContinueWith против await на потоке UI ==");
var ui = new UiThread();
await ui.RunAsync(async () =>
{
    Console.WriteLine($"  старт                       : {Here.Now}");
    await Task.Delay(50);
    Console.WriteLine($"  после await                 : {Here.Now}");
    Task<string> cw = Task.Delay(50).ContinueWith(_ => Here.Now);
    Console.WriteLine($"  ContinueWith                : {await cw}");
    Task<string> cwUi = Task.Delay(50).ContinueWith(_ => Here.Now, TaskScheduler.FromCurrentSynchronizationContext());
    Console.WriteLine($"  ContinueWith(…UI-планировщик): {await cwUi}");
});
Task<int> broken = Task.FromException<int>(new InvalidOperationException("упала"));
Task<string> ignoring = broken.ContinueWith(_ => "продолжение не смотрело на исключение");
string ignoringResult = await ignoring;
Console.WriteLine($"  ContinueWith без проверки ошибки: Status={ignoring.Status}, результат «{ignoringResult}», ошибка исходной задачи никому не видна");
try { await broken; } catch (InvalidOperationException) { Console.WriteLine("  await той же задачи бросает InvalidOperationException"); }
var asyncCw = Task.CompletedTask.ContinueWith(async _ => await Task.Delay(300));
sw.Restart();
await asyncCw;
Console.WriteLine($"  ContinueWith(async …) это Task<Task>: {asyncCw is Task<Task>}; await внешней задачи занял {sw.ElapsedMilliseconds} мс (внутренняя ещё работает)");

public Task<string> ReadLineAsync(CancellationToken ct = default)
{
    var tcs = new TaskCompletionSource<string>(TaskCreationOptions.RunContinuationsAsynchronously);
    CancellationTokenRegistration registration = default;
    void Handler(string line)
    {
        LineRead -= Handler;
        registration.Dispose();
        tcs.TrySetResult(line);
    }
    LineRead += Handler;
    registration = ct.Register(() =>
    {
        LineRead -= Handler;
        tcs.TrySetCanceled(ct);
    });
    return tcs.Task;
}

Console.WriteLine("== 3. TaskCompletionSource ==");
var tcs = new TaskCompletionSource<int>();
Console.WriteLine($"  TrySetResult(1): {tcs.TrySetResult(1)}, второй раз: {tcs.TrySetResult(2)}");
try { tcs.SetResult(3); } catch (InvalidOperationException e) { Console.WriteLine($"  SetResult после завершения: {e.GetType().Name}"); }
var source = new LineSource();
Task<string> good = source.ReadLineAsync();
Console.WriteLine($"  обработчиков после подписки: {source.HandlerCount}");
source.Raise("привет");
Console.WriteLine($"  результат: «{await good}», обработчиков после события: {source.HandlerCount}");
using (var cts = new CancellationTokenSource())
{
    Task<string> cancellable = source.ReadLineAsync(cts.Token);
    cts.Cancel();
    Console.WriteLine($"  после отмены: Status={cancellable.Status}, обработчиков: {source.HandlerCount}");
}
Task<string> hanging = source.ReadLineBrokenAsync();
try { source.Raise("не число"); } catch (FormatException) { Console.WriteLine("  в обработчике бросило FormatException"); }
try { await hanging.WaitAsync(TimeSpan.FromMilliseconds(300)); }
catch (TimeoutException) { Console.WriteLine($"  задача не завершилась никогда: Status={hanging.Status}"); }

Console.WriteLine("== 4. Комбинаторы ==");
Task<int> fast = Job(100), slowJob = Job(600);
Task<int> boom = Task.Run<int>(async () => { await Task.Delay(150); throw new InvalidOperationException("проигравшая упала"); });
int unobserved = 0;
TaskScheduler.UnobservedTaskException += (_, e) => { unobserved++; e.SetObserved(); };
Task<int> winner = await Task.WhenAny(fast, boom);
Console.WriteLine($"  WhenAny вернул задачу, которая выиграла: Result={winner.Result}, остальные: boom={boom.Status}, slow={slowJob.Status}");
await Task.Delay(200);
Console.WriteLine($"  после ожидания: boom={boom.Status}, slow={slowJob.Status}");
boom = null!;
for (int i = 0; i < 3; i++) { GC.Collect(); GC.WaitForPendingFinalizers(); }
Console.WriteLine($"  UnobservedTaskException сработал {unobserved} раз(а)");
Task.WhenAll().Wait();
Console.WriteLine($"  Task.WhenAll() без аргументов завершён сразу: {Task.WhenAll().IsCompletedSuccessfully}");
sw.Restart();
Task delayZero = Task.Delay(0);
Console.WriteLine($"  Task.Delay(0) возвращает завершённую задачу: {delayZero.IsCompletedSuccessfully} (ReferenceEquals с CompletedTask: {ReferenceEquals(delayZero, Task.CompletedTask)})");

var pending = tasks.ToList();
while (pending.Count > 0)
{
    Task finished = await Task.WhenAny(pending);
    pending.Remove(finished);
}

private sealed class WhenEachState : Queue<Task>, IValueTaskSource, ITaskCompletionAction
{
    public static WhenEachState Create(ReadOnlySpan<Task> tasks)
    {
        WhenEachState whenEachState = null;
        if (tasks.Length != 0)
        {
            whenEachState = new WhenEachState();
            foreach (Task obj in tasks)
            {
                whenEachState.Remaining++;
                obj.AddCompletionAction(whenEachState);
            }
        }
        return whenEachState;
    }
    ...
}

foreach (int n in new[] { 1000, 4000, 16000 })
{
    var many = Enumerable.Range(0, n).Select(i => Task.Delay(1 + i % 20)).ToArray();
    sw.Restart();
    int done = 0;
    await foreach (Task _ in Task.WhenEach(many)) done++;
    long each = sw.ElapsedMilliseconds;

    many = Enumerable.Range(0, n).Select(i => Task.Delay(1 + i % 20)).ToArray();
    sw.Restart();
    var pending = many.ToList();
    while (pending.Count > 0)
    {
        Task finished = await Task.WhenAny(pending);
        pending.Remove(finished);
    }
    long loop = sw.ElapsedMilliseconds;
    Console.WriteLine($"  {n,6} задач: WhenEach {each,5} мс, цикл WhenAny + Remove {loop,6} мс");
}

// 1) Parallel.ForEachAsync (.NET 6+)
await Parallel.ForEachAsync(urls, new ParallelOptions { MaxDegreeOfParallelism = 8, CancellationToken = ct },
    async (url, token) => await client.GetStringAsync(url, token));

// 2) SemaphoreSlim как асинхронный ограничитель
var gate = new SemaphoreSlim(8);
var tasks = urls.Select(async url =>
{
    await gate.WaitAsync(ct);
    try { return await client.GetStringAsync(url, ct); }
    finally { gate.Release(); }       // обязательно в finally
});
var results = await Task.WhenAll(tasks);

public void QueueWorkerIfDopAvailable()
{
    if (_remainingDop > 0)
    {
        _remainingDop--;
        Interlocked.Increment(ref _completionRefCount);
        if (_scheduler == TaskScheduler.Default)
        {
            ThreadPool.UnsafeQueueUserWorkItem(this, preferLocal: false);
        }
        ...
    }
}

// тело воркера
while (!state.Cancellation.IsCancellationRequested)
{
    await state.AcquireLock();                       // SemaphoreSlim(1, 1) вокруг общего перечислителя
    TSource current;
    try
    {
        if (state.Cancellation.IsCancellationRequested || !state.Enumerator.MoveNext())
        {
            break;
        }
        current = state.Enumerator.Current;
    }
    finally
    {
        state.ReleaseLock();
    }
    if (!launchedNext)
    {
        launchedNext = true;
        state.QueueWorkerIfDopAvailable();           // запустить следующего воркера
    }
    await state.LoopBody(current, state.Cancellation.Token);
}
...
catch (Exception e) { state.RecordException(e); }    // RecordException отменяет Cancellation

Console.WriteLine("== 5. Ограничение параллелизма: 20 «запросов» по 100 мс, не больше 4 одновременно ==");
var gauge = new ConcurrencyGauge();
sw.Restart();
await Parallel.ForEachAsync(Enumerable.Range(0, 20), new ParallelOptions { MaxDegreeOfParallelism = 4 },
    async (i, ct) => await gauge.CallAsync(ct));
Console.WriteLine($"  Parallel.ForEachAsync: {sw.ElapsedMilliseconds} мс, максимум одновременно {gauge.Max}");
gauge = new ConcurrencyGauge();
var gate = new SemaphoreSlim(4);
sw.Restart();
await Task.WhenAll(Enumerable.Range(0, 20).Select(async _ =>
{
    await gate.WaitAsync();
    try { await gauge.CallAsync(default); }
    finally { gate.Release(); }        // обязательно в finally
}));
Console.WriteLine($"  SemaphoreSlim        : {sw.ElapsedMilliseconds} мс, максимум одновременно {gauge.Max}");
gauge = new ConcurrencyGauge();
await Parallel.ForEachAsync(Enumerable.Range(0, 20), async (i, ct) => await gauge.CallAsync(ct));
Console.WriteLine($"  без MaxDegreeOfParallelism: максимум одновременно {gauge.Max} (ядер: {Environment.ProcessorCount})");
gauge = new ConcurrencyGauge();
try
{
    await Parallel.ForEachAsync(Enumerable.Range(0, 100), new ParallelOptions { MaxDegreeOfParallelism = 4 }, async (i, ct) =>
    {
        await gauge.CallAsync(ct);
        if (i == 5) throw new InvalidOperationException($"упал элемент {i}");
    });
}
catch (InvalidOperationException e)
{
    Console.WriteLine($"  первое исключение остановило цикл: «{e.Message}», обработано элементов: {gauge.Started} из 100");
}

private readonly SemaphoreSlim _lock = new(1, 1);

async Task UpdateAsync()
{
    await _lock.WaitAsync();
    try { await DoAsync(); }
    finally { _lock.Release(); }       // иначе блокировка останется занятой навсегда
}

private Task<bool> WaitAsyncCore(long millisecondsTimeout, CancellationToken cancellationToken)
{
    ...
    lock (m_lockObjAndDisposed)
    {
        if (m_currentCount > 0)
        {
            m_currentCount--;
            ...
            return Task.FromResult(result: true);           // синхронный путь: свободно
        }
        if (millisecondsTimeout == 0L)
        {
            return Task.FromResult(result: false);
        }
        TaskNode taskNode = CreateAndAddAsyncWaiter();      // в конец связного списка
        return (millisecondsTimeout == -1 && !cancellationToken.CanBeCanceled) ? taskNode : WaitUntilCountOrTimeoutAsync(taskNode, millisecondsTimeout, cancellationToken);
    }
}

public int Release(int releaseCount)
{
    ...
    lock (m_lockObjAndDisposed)
    {
        ...
        if (m_maxCount - currentCount < releaseCount)
        {
            throw new SemaphoreFullException();
        }
        ...
        if (m_asyncHead != null)
        {
            ...
            while (num3 > 0 && m_asyncHead != null)
            {
                currentCount--;
                num3--;
                TaskNode asyncHead = m_asyncHead;
                RemoveAsyncWaiter(asyncHead);
                asyncHead.TrySetResult(result: true);       // разбудить ПЕРВОГО ожидающего
            }
        }
        ...
    }
}

private sealed class TaskNode : Task<bool>
{
    internal TaskNode Prev;
    internal TaskNode Next;

    internal TaskNode()
        : base((object)null, TaskCreationOptions.RunContinuationsAsynchronously)
    {
    }
}

Console.WriteLine("== 6. SemaphoreSlim как асинхронная блокировка ==");
var mutex = new SemaphoreSlim(1, 1);
await mutex.WaitAsync();
Task<bool> second = mutex.WaitAsync(TimeSpan.FromMilliseconds(200));
Console.WriteLine($"  повторный вход из того же потока выполнения: получил блокировку = {await second} (реентерабельности нет)");
mutex.Release();
try { mutex.Release(); } catch (SemaphoreFullException) { Console.WriteLine("  лишний Release: SemaphoreFullException"); }
var queue = new SemaphoreSlim(1, 1);
await queue.WaitAsync();                                             // блокировка занята
var order = new List<int>();
Task[] waiters = Enumerable.Range(1, 5).Select(async n =>
{
    await queue.WaitAsync();                                         // вызов WaitAsync происходит по порядку: n = 1…5
    order.Add(n);
    queue.Release();
}).ToArray();
queue.Release();
await Task.WhenAll(waiters);
Console.WriteLine($"  порядок получения блокировки пятью ожидающими: {string.Join(", ", order)}");
var rc = new SemaphoreSlim(0, 1);
int releaseThread = 0, continuationThread = 0;
Task awaiter = Task.Run(async () => { await rc.WaitAsync(); continuationThread = Environment.CurrentManagedThreadId; });
await Task.Delay(100);
await Task.Run(() => { releaseThread = Environment.CurrentManagedThreadId; rc.Release(); Thread.Sleep(50); });
await awaiter;
Console.WriteLine($"  Release из потока {releaseThread}, продолжение ожидавшего на потоке {continuationThread}: другой поток = {releaseThread != continuationThread}");

using var timer = new PeriodicTimer(TimeSpan.FromSeconds(1));
while (await timer.WaitForNextTickAsync(ct))
{
    await DoTickAsync(ct);
}

public PeriodicTimer(TimeSpan period)
{
    ...
    _state = new State();
    _timer = new TimerQueueTimer((object s) =>
    {
        ((State)s).Signal();
    }, _state, milliseconds, milliseconds, flowExecutionContext: false);
}

// State : IValueTaskSource<bool>
public ValueTask<bool> WaitForNextTickAsync(PeriodicTimer owner, CancellationToken cancellationToken)
{
    lock (this)
    {
        if (_activeWait)
        {
            ThrowHelper.ThrowInvalidOperationException();
        }
        ...
        if (_signaled)
        {
            if (!_stopped)
            {
                _signaled = false;
            }
            return new ValueTask<bool>(!_stopped);
        }
        _activeWait = true;
        ...
        return new ValueTask<bool>(this, _mrvtsc.Version);
    }
}

public void Signal(bool stopping = false, CancellationToken cancellationToken = default)
{
    bool flag = false;
    lock (this)
    {
        _stopped |= stopping;
        if (!_signaled)
        {
            _signaled = true;
            flag = _activeWait;
        }
    }
    if (flag)
    {
        _mrvtsc.SetResult(result: true);
    }
}

Console.WriteLine();
Console.WriteLine("== 7. PeriodicTimer против цикла с Delay: работа 40 мс, период 100 мс, 10 итераций ==");
sw.Restart();
var stamps = new List<long>();
using (var timer = new PeriodicTimer(TimeSpan.FromMilliseconds(100)))
{
    for (int i = 0; i < 10 && await timer.WaitForNextTickAsync(); i++)
    {
        stamps.Add(sw.ElapsedMilliseconds);
        await Task.Delay(40);
    }
}
Console.WriteLine($"  PeriodicTimer: начало итераций (мс) {string.Join(" ", stamps.Select(s => s / 10 * 10))}, всего {sw.ElapsedMilliseconds} мс");
sw.Restart();
stamps.Clear();
for (int i = 0; i < 10; i++)
{
    await Task.Delay(100);
    stamps.Add(sw.ElapsedMilliseconds);
    await Task.Delay(40);
}
Console.WriteLine($"  цикл с Delay : начало итераций (мс) {string.Join(" ", stamps.Select(s => s / 10 * 10))}, всего {sw.ElapsedMilliseconds} мс");
using (var timer = new PeriodicTimer(TimeSpan.FromMilliseconds(100)))
{
    await Task.Delay(450);                                           // «долгая работа»: за это время прошло 4 тика
    int immediate = 0;
    for (int i = 0; i < 4; i++)
    {
        ValueTask<bool> tick = timer.WaitForNextTickAsync();
        if (tick.IsCompleted) immediate++;
        await tick;
    }
    Console.WriteLine($"  за 450 мс прошло 4 тика, мгновенно завершились ожидания: {immediate} (пропущенные не копятся)");
}
using (var timer = new PeriodicTimer(TimeSpan.FromSeconds(10)))

    try { await timer.WaitForNextTickAsync(); }
    catch (InvalidOperationException) { Console.WriteLine("  два одновременных WaitForNextTickAsync: InvalidOperationException"); }
}

Console.WriteLine();
Console.WriteLine("== 8. Ленивая асинхронная инициализация ==");
int attempts = 0;
var lazy = new Lazy<Task<int>>(async () => { attempts++; await Task.Delay(10); if (attempts == 1) throw new InvalidOperationException("первая попытка"); return 42; });
for (int i = 0; i < 2; i++)
{
    try { Console.WriteLine($"  Lazy<Task>, вызов {i + 1}: {await lazy.Value}"); }
    catch (InvalidOperationException e) { Console.WriteLine($"  Lazy<Task>, вызов {i + 1}: {e.Message} (попыток инициализации: {attempts})"); }
}
attempts = 0;
var asyncLazy = new AsyncLazy<int>(async () => { attempts++; await Task.Delay(10); if (attempts == 1) throw new InvalidOperationException("первая попытка"); return 42; });
for (int i = 0; i < 3; i++)