13. Пул потоков и «настоящая» асинхронность¶

13.1. Очереди пула¶

В главе 5 мы видели, что продолжение «уходит в пул». Теперь посмотрим, что это значит. Из CoreLib 10.0.12 (ThreadPoolWorkQueue):

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32

public void Enqueue(object callback, bool forceGlobal)
{
    ThreadPoolWorkQueueThreadLocals threadLocals;
    if (!forceGlobal && (threadLocals = ThreadPoolWorkQueueThreadLocals.threadLocals) != null)
    {
        threadLocals.workStealingQueue.LocalPush(callback);        // (1)!
    }
    else
    {
        workItems.Enqueue(callback);                               // (2)!
    }
    EnsureThreadRequested();                                       // (3)!
}

public object Dequeue(ThreadPoolWorkQueueThreadLocals tl, ref bool missedSteal)
{
    object workItem = tl.workStealingQueue.LocalPop();             // (4)!
    if (workItem != null) return workItem;

    // ... очередь высокого приоритета (highPriorityWorkItems) ...  // (5)!

    if (workItems.TryDequeue(out workItem)) return workItem;       // (6)!

    // ... перебор чужих локальных очередей начиная со случайной ...
    WorkStealingQueue workStealingQueue2 = queues[num7];
    if (workStealingQueue2 != workStealingQueue && workStealingQueue2.CanSteal)
    {
        workItem = workStealingQueue2.TrySteal(ref missedSteal);   // (7)!
        if (workItem != null) return workItem;
    }
    return null;
}

1. Поставка из рабочего потока пула (и без forceGlobal) идёт в локальную очередь этого потока. Так работают Task.Run изнутри пула, продолжения await, ThreadPool.UnsafeQueueUserWorkItem(..., preferLocal: true).
2. Поставка из любого другого потока (основной, таймер, epoll) или с preferLocal: false — в глобальную очередь, ConcurrentQueue, то есть FIFO.
3. Если рабочих потоков для этой работы может не хватить, пул просит ещё один (подробнее в §13.2).
4. Свободный рабочий поток сначала смотрит в свою локальную очередь, и забирает оттуда LocalPop — последний положенный элемент (LIFO). Это даёт локальность кеша: свежая работа работает с тем же, что только что трогал поток.
5. Если кто-то ждал на Task.Wait() из потока пула, его локальные задачи перекладываются в очередь высокого приоритета (TransferAllLocalWorkItemsToHighPriorityGlobalQueue), чтобы они не застряли за заблокированным потоком.
6. Потом общая очередь, FIFO.
7. И в последнюю очередь — кража работы: берём из чужой локальной очереди TrySteal — с другого конца (самый старый элемент).

Опыт: порядок выполнения (dotnet run -c Release --project final -- queues)¶

Рабочий поток W кладёт 5 элементов в локальную очередь (L0…L4, preferLocal: true) и 5 в глобальную (G0…G4), каждый элемент спит 30 мс.

W@4  L0@6  L4@4  L3@4  G0@6  L2@4  G1@6  L1@4  G2@6  G3@6  G4@4

W@4  L0@6  L4@4  G0@7  G1@8  G2@9  G3@10  G4@11  L1@12  L2@9  L3@12

W@4  G0@6  L4@4  L3@4  G1@6  L2@4  G2@6  L1@4  G3@6  G4@6  L0@4

Поток 4 (владелец) выбирает из своей очереди L4, L3, L2, L1 — с конца. Поток 6 (второй, разбуженный EnsureThreadRequested) берёт общие G0, G1, G2, G3 по порядку и крадёт самый старый локальный L0. Остальные запуски дают ту же картину с небольшой перестановкой, кто какой G берёт.

На Windows принцип тот же: владелец берёт L4, L3, … с конца своей очереди, остальные потоки берут G по порядку и крадут L0. Отличается только число участников. На 8 ядрах (вторая вкладка) на пять глобальных элементов нашлось пять разных потоков (7–11), потому что пул сразу разбудил больше потоков, а локальные L1–L3 растащили чужие потоки. На 2 ядрах (третья вкладка) картина как на Linux: два потока, владелец доедает локальные с конца. Крали ли L0 или владелец взял его последним — зависит от гонки и от запуска к запуску.

Интерактивная схема¶

Симуляция Dequeue и график роста числа потоков из опытов главы. Открыть на весь экран.

13.2. Сколько потоков и когда пул добавляет новые¶

• Число потоков адаптивное: от MinThreads (по умолчанию число логических ядер) до MaxThreads (очень большое). До минимума потоки создаются сразу по необходимости.
• Выше минимума добавляют потоки три механизма: hill climbing (меняет цель по измерениям пропускной способности), обнаружение голодания и кооперативная блокировка. Последние два — в PortableThreadPool ниже.

Обнаружение голодания: «+1 поток не чаще раза в полсекунды»¶

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

// поток ".NET TP Gate" просыпается примерно раз в 500 мс
if (!disableStarvationDetection
    && threadPoolInstance._pendingBlockingAdjustment == PendingBlockingAdjustment.None
    && threadPoolInstance._separated.numRequestedWorkers > 0               // (1)!
    && SufficientDelaySinceLastDequeue(threadPoolInstance))                // (2)!
{
    // ...
    while (threadCounts.NumProcessingWork < threadPoolInstance._maxThreads
        && threadCounts.NumProcessingWork >= threadCounts.NumThreadsGoal)
    {
        newCounts.NumThreadsGoal = (short)(threadCounts.NumProcessingWork + 1);   // (3)!
        // ...
    }
}

private static bool SufficientDelaySinceLastDequeue(PortableThreadPool threadPoolInstance)
{
    int num = Environment.TickCount - threadPoolInstance._separated.lastDequeueTime;
    uint num2 = ((threadPoolInstance._cpuUtilization >= 80)
        ? ((uint)(threadPoolInstance._separated.counts.NumThreadsGoal * 1000))   // (4)!
        : 500u);
    return (uint)num > num2;                                                      // (5)!
}

1. В очереди есть работа, на которую просили поток.
2. И давно никто ничего из очереди не брал (lastDequeueTime).
3. Тогда цель — на один поток больше, чем сейчас работает.
4. Если процессор загружен (≥ 80 %), ждать нужно цель × 1000 мс: пул не хочет добавлять потоки, когда ядра и так заняты.
5. Больше 500 мс без единого взятия из очереди — голодание.

Похоже, именно из-за пункта 5 нагрузка из блокирующих задач по секунде разгоняется плохо: каждые ~секунду кто-то из потоков заканчивает работу и берёт следующую из очереди (lastDequeueTime обновляется), так что счётчик «давно ничего не брали» почти не набегает.

Кооперативная блокировка: Task.Wait() предупреждает пул¶

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14

private bool SpinThenBlockingWait(int millisecondsTimeout, CancellationToken cancellationToken)
{
    bool spun = SpinWait(millisecondsTimeout);
    if (!spun)
    {
        ThreadPoolWorkQueue.TransferAllLocalWorkItemsToHighPriorityGlobalQueue();   // (1)!
        var mres = new SetOnInvokeMres();
        AddCompletionAction(mres, addBeforeOthers: true);
        bool notified = ThreadPool.NotifyThreadBlocked();                           // (2)!
        try { mres.Wait(-1, cancellationToken); }
        finally { if (notified) ThreadPool.NotifyThreadUnblocked(); }              // (3)!
    }
    // ...
}

1. Локальную очередь освобождаем (см. пункт 5 листинга очередей).
2. Сообщаем пулу, что этот поток заблокирован. Работает только на потоке пула и только если включено.
3. Когда ожидание закончилось, сообщаем об этом.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17

public bool NotifyThreadBlocked()
{
    if (!BlockingConfig.IsCooperativeBlockingEnabled || !Thread.CurrentThread.IsThreadPoolThread)
        return false;
    // ...
    _numBlockedThreads++;
    // ... будим Gate-поток: пора скорректировать цель
}

private short TargetThreadsGoalForBlockingAdjustment
    => _numBlockedThreads > 0
        ? (short)Math.Min(_minThreads + _numBlockedThreads, _maxThreads)    // (1)!
        : _minThreads;

// в PerformBlockingAdjustment:
short immediately = Math.Min(_minThreads + ThreadsToAddWithoutDelay, _maxThreads);   // (2)!
// дальше по ThreadsPerDelayStep потоков за DelayStepMs, но не дольше MaxDelayMs         // (3)!

1. Цель пула: MinThreads + число заблокированных в Wait. То есть «заблокированные» потоки из счёта выбывают, пул добавляет столько же свободных.
2. Первые число ядер потоков сверх минимума добавляются сразу.
3. Остальные — пачками по число ядер потоков с задержкой, растущей на 25 мс за шаг (до 250 мс). Переключатели: System.Threading.ThreadPool.Blocking.CooperativeBlocking (по умолчанию true), ...ThreadsToAddWithoutDelay_ProcCountFactor, ...ThreadsPerDelayStep_ProcCountFactor, ...DelayStepMs, ...MaxDelayMs.

13.3. Голодание пула потоков: эксперимент¶

 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25

using System.Diagnostics;

ThreadPool.SetMinThreads(4, 4);               // уменьшим минимум, чтобы эффект был ярче
Console.WriteLine($"MinThreads=4, процессоров={Environment.ProcessorCount}");

var sw = Stopwatch.StartNew();
var stop = false;
var monitor = new Thread(() =>                // отдельный поток: пул занят, мониторить из него нельзя
{
    while (!Volatile.Read(ref stop))
    {
        Console.WriteLine($"  t={sw.Elapsed.TotalSeconds:F1}s потоков={ThreadPool.ThreadCount}, " +
                          $"в очереди={ThreadPool.PendingWorkItemCount}");
        Thread.Sleep(1000);
    }
}) { IsBackground = true };
monitor.Start();

await Task.WhenAll(Enumerable.Range(0, 32).Select(_ => Task.Run(() => Thread.Sleep(1000))));
Volatile.Write(ref stop, true);
Console.WriteLine($"БЛОКИРУЮЩИЕ 32 x 1с : {sw.Elapsed.TotalSeconds:F1} с");

Предскажите время (и число потоков). Затем dotnet run -c Release --project final: там та же нагрузка выполняется тремя способами (Starve, строки 30–38 final/Program.cs).

MinThreads=4, процессоров=2
  t=1.0s потоков=4, в очереди=28
  t=2.0s потоков=6, в очереди=22
  ...
БЛОКИРУЮЩИЕ    32 x 1с : 6.0 с, потоков в пуле 6

  t=1.0s потоков=17, в очереди=15
  t=2.0s потоков=22, в очереди=12
  ...
SYNC-OVER-ASYNC 32 x 1с : 5.3 с, потоков в пуле 33

АСИНХРОННЫЕ    32 x 1с : 1.0 с, потоков в пуле 33

MinThreads=4, процессоров=8
  t=1.0s потоков=8, в очереди=24
  t=2.0s потоков=9, в очереди=15
  ...
БЛОКИРУЮЩИЕ    32 x 1с : 4.1 с, потоков в пуле 9

  t=1.0s потоков=31, в очереди=1
  t=2.0s потоков=33, в очереди=0
SYNC-OVER-ASYNC 32 x 1с : 2.1 с, потоков в пуле 33

АСИНХРОННЫЕ    32 x 1с : 1.0 с, потоков в пуле 33

MinThreads=4, процессоров=2
  t=1.0s потоков=5, в очереди=27
  t=2.0s потоков=5, в очереди=22
  ...
БЛОКИРУЮЩИЕ    32 x 1с : 7.1 с, потоков в пуле 5

  t=1.0s потоков=17, в очереди=15
  t=2.0s потоков=22, в очереди=12
  ...
SYNC-OVER-ASYNC 32 x 1с : 5.5 с, потоков в пуле 33

АСИНХРОННЫЕ    32 x 1с : 1.0 с, потоков в пуле 33

Признаки голодания в проде:

• латентность растёт, а CPU низкий;
• ThreadPool.PendingWorkItemCount (метрика threadpool-queue-length) большой и растёт;
• ThreadPool.ThreadCount растёт ступеньками;
• таймауты вызовов зависимостей (потому что продолжения не получают потока).

Диагностика: dotnet-counters monitor (ThreadPool Queue Length, Thread Count), dotnet-stack/dotnet-dump (много потоков, стоящих в Wait/Result/Monitor.Enter), трассировка событий пула (подробно — глава 15).

Лечение: убрать синхронные блокировки (.Result, .Wait(), Thread.Sleep, синхронный I/O) из горячего пути; не поднимать MinThreads как первое лекарство (это маскирует причину, но допустимо как временная мера).

13.4. «Пока ждём ввод-вывод, потока нет»¶

Это верно только для по-настоящему асинхронного ввода-вывода: сокеты, HTTP, БД-драйверы с async-API. Операция передаётся ОС, поток освобождается, по завершении ОС уведомляет рантайм, и продолжение берёт свободный поток пула.

Это не верно для:

• Task.Run(() => File.ReadAllText(...)) — «async» просто занимает другой поток на всё время блокирующего вызова;
• драйверов, у которых async-методы внутри синхронные.

Но есть и вторая сторона: ожидание бесплатно, а завершение — нет. Режим io (dotnet run -c Release --project final -- io) занимает все потоки пула (SetMaxThreads(n, n), где n = Environment.ProcessorCount — меньше числа ядер значение не принимается: на 8 ядрах SetMaxThreads(2, 2) вернёт false и пул не ограничится) и смотрит, завершится ли ввод-вывод.

SetMaxThreads(2,2) = True
оба потока пула заняты
сокет: чтение завершено при занятом пуле = False
файл (useAsync: true): чтение завершено при занятом пуле = False
после освобождения пула: сокет = True, файл = True

SetMaxThreads(8,8) = True
все потоки пула заняты
сокет: чтение завершено при занятом пуле = False
файл (useAsync: true): чтение завершено при занятом пуле = False
после освобождения пула: сокет = True, файл = True

Итоги¶

• Продолжения и Task.Run изнутри пула идут в локальную очередь потока (LIFO для владельца), остальное — в глобальную (FIFO); свободные потоки крадут работу с «старого» конца чужих очередей.
• Пул добавляет потоки выше MinThreads медленно: голодание обнаруживается раз в ~500 мс. Исключение — кооперативная блокировка: Task.Wait()/.Result на потоке пула предупреждает пул, и он быстро растёт (за секунду 4 → 17 потоков на нашем стенде).
• Thread.Sleep и синхронный ввод-вывод пулу не сообщают о блокировке: восстановление идёт по одному потоку.
• Асинхронное ожидание не занимает потоков, но завершение ввода-вывода (и на Linux, и на Windows) всё равно требует свободного потока пула.
• Лечение голодания — убрать блокировки, а не поднять MinThreads.

Код лабораторной¶

Запуск из папки главы: dotnet run -c Release --project start или --project final.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25

// Глава 13, §13.3. Голодание пула потоков.
// PREDICT: сколько секунд займут 32 блокирующие задачи по 1 с при MinThreads=4?
// TODO 1: перепишите нагрузку так, чтобы она не занимала потоки (Task.Delay), и сравните время.
// TODO 2: замените Thread.Sleep(1000) на Task.Delay(1000).Wait() — предскажите время и число потоков (и проверьте, §13.3).
using System.Diagnostics;

ThreadPool.SetMinThreads(4, 4);               // уменьшим минимум, чтобы эффект был ярче
Console.WriteLine($"MinThreads=4, процессоров={Environment.ProcessorCount}");

var sw = Stopwatch.StartNew();
var stop = false;
var monitor = new Thread(() =>                // отдельный поток: пул занят, мониторить из него нельзя
{
    while (!Volatile.Read(ref stop))
    {
        Console.WriteLine($"  t={sw.Elapsed.TotalSeconds:F1}s потоков={ThreadPool.ThreadCount}, " +
                          $"в очереди={ThreadPool.PendingWorkItemCount}");
        Thread.Sleep(1000);
    }
}) { IsBackground = true };
monitor.Start();

await Task.WhenAll(Enumerable.Range(0, 32).Select(_ => Task.Run(() => Thread.Sleep(1000))));
Volatile.Write(ref stop, true);
Console.WriteLine($"БЛОКИРУЮЩИЕ 32 x 1с : {sw.Elapsed.TotalSeconds:F1} с");

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114

// Глава 13, итог. Режимы (dotnet run -c Release --project final -- <режим>):
//   starve   (по умолчанию) — «секунда ожидания» тремя способами: Thread.Sleep, Task.Wait, Task.Delay;
//   coop-off — то же, но с выключенной «кооперативной блокировкой» пула (поймёте, чем Wait отличается от Sleep);
//   queues   — глобальная и локальные очереди, LIFO владельца и кража работы;
//   io       — занят весь пул: завершаются ли чтения из сокета и файла (Linux).
using System.Collections.Concurrent;
using System.Diagnostics;
using System.Net;
using System.Net.Sockets;

string mode = args.FirstOrDefault() ?? "starve";
switch (mode)
{
    case "starve":
        await Starve();
        break;
    case "coop-off":
        // Переключатель читается один раз, при первом обращении к пулу: ставим его до любого Task.Run.
        AppContext.SetSwitch("System.Threading.ThreadPool.Blocking.CooperativeBlocking", false);
        await Starve();
        break;
    case "queues":
        Queues();
        break;
    case "io":
        Io();
        break;
}

static async Task Starve()
{
    ThreadPool.SetMinThreads(4, 4);
    Console.WriteLine($"MinThreads=4, процессоров={Environment.ProcessorCount}");

    await Measure("БЛОКИРУЮЩИЕ   ", () => Task.Run(() => Thread.Sleep(1000)));
    await Measure("SYNC-OVER-ASYNC", () => Task.Run(() => Task.Delay(1000).Wait()));
    await Measure("АСИНХРОННЫЕ   ", () => Task.Delay(1000));
}

static async Task Measure(string title, Func<Task> oneSecond)
{
    var sw = Stopwatch.StartNew();
    var stop = false;
    var monitor = new Thread(() =>
    {
        while (!Volatile.Read(ref stop))
        {
            Console.WriteLine($"  t={sw.Elapsed.TotalSeconds:F1}s потоков={ThreadPool.ThreadCount}, в очереди={ThreadPool.PendingWorkItemCount}");
            Thread.Sleep(1000);
        }
    }) { IsBackground = true };
    monitor.Start();

    await Task.WhenAll(Enumerable.Range(0, 32).Select(_ => oneSecond()));
    Volatile.Write(ref stop, true);
    Console.WriteLine($"{title} 32 x 1с : {sw.Elapsed.TotalSeconds:F1} с, потоков в пуле {ThreadPool.ThreadCount}");
    Console.WriteLine();
}

// Рабочий поток W кладёт пять элементов в свою локальную очередь (L0..L4) и пять в глобальную (G0..G4).
static void Queues()
{
    var log = new ConcurrentQueue<string>();
    var done = new CountdownEvent(1 + 5 + 5);
    void Item(string name) { log.Enqueue($"{name}@{Environment.CurrentManagedThreadId}"); Thread.Sleep(30); done.Signal(); }

    ThreadPool.UnsafeQueueUserWorkItem<object?>(_ =>
    {
        log.Enqueue($"W@{Environment.CurrentManagedThreadId}");
        for (int i = 0; i < 5; i++) { int k = i; ThreadPool.UnsafeQueueUserWorkItem<object?>(_ => Item($"L{k}"), null, preferLocal: true); }
        for (int i = 0; i < 5; i++) { int k = i; ThreadPool.UnsafeQueueUserWorkItem<object?>(_ => Item($"G{k}"), null, preferLocal: false); }
        done.Signal();
    }, null, preferLocal: false);

    done.Wait();
    Console.WriteLine(string.Join("  ", log));
}

// Весь пул (число потоков = числу ядер — минимально допустимый максимум) занят. Кто из ввода-вывода всё равно завершится?
static void Io()
{
    var path = Path.GetTempFileName();
    File.WriteAllBytes(path, new byte[1 << 20]);
    int n = Environment.ProcessorCount;                      // меньше числа ядер SetMaxThreads не примет
    Console.WriteLine($"SetMaxThreads({n},{n}) = {ThreadPool.SetMaxThreads(n, n)}");

    using var gate = new ManualResetEventSlim();
    using var started = new CountdownEvent(n);
    for (int i = 0; i < n; i++)
        ThreadPool.UnsafeQueueUserWorkItem<object?>(_ => { started.Signal(); gate.Wait(); }, null, false);
    started.Wait();
    Console.WriteLine("все потоки пула заняты");

    var listener = new TcpListener(IPAddress.Loopback, 0);
    listener.Start();
    using var client = new TcpClient();
    client.Connect(IPAddress.Loopback, ((IPEndPoint)listener.LocalEndpoint).Port);
    using var server = listener.AcceptTcpClient();
    ValueTask<int> socketRead = client.GetStream().ReadAsync(new byte[16]);
    server.GetStream().WriteByte(1);                         // данные уже у нас в ядре
    Thread.Sleep(500);
    Console.WriteLine($"сокет: чтение завершено при занятом пуле = {socketRead.IsCompleted}");

    var fs = new FileStream(path, FileMode.Open, FileAccess.Read, FileShare.Read, 1, useAsync: true);
    ValueTask<int> fileRead = fs.ReadAsync(new byte[4096]);
    Thread.Sleep(500);
    Console.WriteLine($"файл (useAsync: true): чтение завершено при занятом пуле = {fileRead.IsCompleted}");

    gate.Set();                                              // освободили пул
    Thread.Sleep(200);
    Console.WriteLine($"после освобождения пула: сокет = {socketRead.IsCompleted}, файл = {fileRead.IsCompleted}");
    fs.Dispose();                                            // Windows не даст удалить открытый файл
    File.Delete(path);
}