Раздел 16 — Потоки

В разделе 14 процесс был контейнером: адресное пространство, дескрипторы, код завершения. Выполняет код не процесс, а поток — последовательность команд со своим местом выполнения и своим стеком (раздел 7, пункт 05). При запуске программы система создаёт процесс и в нём один поток — он выполняет main. Программа может создать ещё потоки: тогда в одном процессе одновременно выполняются несколько функций.

Зачем потоки. Первое — параллельность: в процессоре несколько ядер, и однопоточная программа занимает одно из них; разделив вычисление на потоки, его можно ускорить в разы (пункт 08). Второе — отзывчивость: пока один поток ждёт — сети, диска, клавиши, — другой работает. Консольный чат в итоговом проекте принимает сообщения в одном потоке, пока другой ждёт ввода пользователя.

Как это работает. Планировщик ОС раздаёт потокам ядра процессора квантами — по нескольку миллисекунд — и переключает их. Потоков обычно больше, чем ядер, поэтому одни выполняются, а другие ждут очереди. Какой поток когда получит ядро, программа не знает: порядок меняется от запуска к запуску. На этом держится всё, что есть опасного в потоках (пункт 09).

Поток выполняет функцию потока — обычную функцию с заданным заголовком: DWORD WINAPI имя(LPVOID param). Параметр — адрес любых данных, результат — код завершения потока. WINAPI — соглашение о вызове функций Windows.

// Три потока: параметры — через указатель на структуру, ожидание всех, коды завершения
#include <windows.h>
#include <iostream>

constexpr int THREADS = 3;

struct Job {
    int number;                                 // номер задания — для вывода
    int repeats;                                // сколько строк вывести
    DWORD pauseMs;                              // пауза между строками
};

DWORD WINAPI worker(LPVOID param);

int main()
{
    Job jobs[THREADS] = {                       // живут до конца main — дольше потоков
        {1, 3, 100},
        {2, 3, 150},
        {3, 3, 70}
    };
    HANDLE threads[THREADS];
    DWORD ids[THREADS];
    std::cout << "main thread " << GetCurrentThreadId() << '\n';
    for (int i = 0; i < THREADS; ++i) {
        threads[i] = CreateThread(nullptr, 0, worker, &jobs[i], 0, &ids[i]);
        if (threads[i] == nullptr) {
            std::cout << "CreateThread failed, error " << GetLastError() << '\n';
            return 2;
        }
        std::cout << "started job " << jobs[i].number << " in thread " << ids[i] << '\n';
    }

    WaitForMultipleObjects(THREADS, threads, TRUE, INFINITE);   // TRUE — ждать все
    for (int i = 0; i < THREADS; ++i) {
        DWORD code = 0;
        GetExitCodeThread(threads[i], &code);
        std::cout << "thread " << ids[i] << " exit code " << code << '\n';
        CloseHandle(threads[i]);
    }
    return 0;
}

// Функция потока: параметр — адрес Job; результат — код завершения потока
DWORD WINAPI worker(LPVOID param)
{
    const Job* job = static_cast<const Job*>(param);
    for (int k = 1; k <= job->repeats; ++k) {
        std::cout << "job " << job->number << ": step " << k << '\n';
        Sleep(job->pauseMs);
    }
    return static_cast<DWORD>(job->number * 10);
}

main thread 11240
started job 1 in thread 9808
job 1: step 1
started job 2 in thread 3316
job 2: step 1
started job 3 in thread 12756
job 3: step 1
job 3: step 2
job 1: step 2
job 3: step 3
job 2: step 2
job 1: step 3
job 2: step 3
thread 9808 exit code 10
thread 3316 exit code 20
thread 12756 exit code 30

CreateThread возвращает дескриптор потока (NULL — ошибка) сразу после создания — не дожидаясь, пока поток что-то сделает. Порядок строк вывода у вас будет другим и будет меняться от запуска к запуску: потоки работают одновременно. Иногда строки даже перемешаются посимвольно — два потока пишут в std::cout одновременно; это первая встреча с гонкой (пункт 09).

У функции потока один параметр LPVOID — «адрес чего угодно» (void*). Всё, что нужно потоку, собирают в структуру (раздел 11), её адрес передают в CreateThread, а в функции потока приводят обратно: static_cast<Job*>(param). Через ту же структуру поток отдаёт результат — записывает его в поле, а main читает после ожидания.

for (int i = 0; i < 3; ++i) {
    threads[i] = CreateThread(nullptr, 0, worker, &i, 0, nullptr);   // ОШИБКА: у всех — адрес одной i
}

• Нормально поток завершается, когда его функция выполнит return: возвращённое значение — код завершения. Его читают GetExitCodeThread после ожидания; до завершения там STILL_ACTIVE (259) — как у процесса.
• ExitThread(код) завершает текущий поток сразу, без деструкторов локальных объектов. TerminateThread завершает чужой поток на любой команде — даже внутри критической секции, которую он тогда никогда не освободит. В курсе — только return.
• Процесс завершается, когда main выполнит return, — вместе со всеми потоками, даже недоделавшими работу. Поэтому main ждёт свои потоки (пункт 07).
• Дескриптор потока закрывают CloseHandle, как дескриптор процесса. Закрытие не останавливает поток — только освобождает запись в таблице. Поток, дескриптор которого не нужен, можно закрыть сразу после создания.

Как с процессами: идентификатор — «имя» для людей и других программ, дескриптор — «ключ» для работы с объектом. По идентификатору дескриптор получают OpenThread.

Планировщик сначала раздаёт ядра потокам с большим приоритетом. Приоритет потока складывается из класса приоритета процесса (раздел 14, GetPriorityClass) и относительного приоритета потока: SetThreadPriority(h, THREAD_PRIORITY_BELOW_NORMAL); значения — от THREAD_PRIORITY_IDLE до THREAD_PRIORITY_TIME_CRITICAL, обычный — THREAD_PRIORITY_NORMAL (0). Понижают приоритет фоновой работе, чтобы она не мешала интерфейсу; повышать без крайней нужды не стоит: поток с высоким приоритетом, который считает без остановки, может «подвесить» остальную систему.

Sleep(мс) отдаёт процессор: поток не выполняется заданное время, ядро получают другие. Sleep(0) — уступить остаток кванта, если кто-то ждёт. Время Sleep — не меньше заданного, но может быть больше: точность — около 15 мс, если программа не просила большей.

Поток, как процесс, переходит в сигнальное состояние, когда завершается, — значит, его ждут той же WaitForSingleObject (раздел 14, пункт 07). Несколько потоков сразу — WaitForMultipleObjects(n, массив, ждатьВсе, тайм-аут):

Больше 64 объектов (MAXIMUM_WAIT_OBJECTS) одним вызовом не ждут. Пример control.cpp создаёт поток приостановленным, понижает приоритет, запускает и раз в секунду проверяет, не закончил ли он:

// Фрагмент control.cpp. Полностью — в примерах раздела
    DWORD seconds = 3;
    HANDLE thread = CreateThread(nullptr, 0, slowWorker, &seconds, CREATE_SUSPENDED, nullptr);
    if (thread == nullptr) {
        std::cout << "CreateThread failed, error " << GetLastError() << '\n';
        return 2;
    }
    SetThreadPriority(thread, THREAD_PRIORITY_BELOW_NORMAL);
    std::cout << "created suspended, priority " << GetThreadPriority(thread) << '\n';
    ResumeThread(thread);                       // только теперь поток начинает выполняться

    DWORD code = 0;
    while (WaitForSingleObject(thread, 1000) == WAIT_TIMEOUT) {
        GetExitCodeThread(thread, &code);
        std::cout << "still working, code " << code << (code == STILL_ACTIVE ? " (STILL_ACTIVE)" : "") << '\n';
    }
    GetExitCodeThread(thread, &code);
    std::cout << "finished, code " << code << '\n';
    CloseHandle(thread);

created suspended, priority -1
still working, code 259 (STILL_ACTIVE)
still working, code 259 (STILL_ACTIVE)
still working, code 259 (STILL_ACTIVE)
finished, code 7

SuspendThread(h) останавливает поток в любом месте, ResumeThread(h) — продолжает; у потока есть счётчик приостановок, и выполняется он, только когда счётчик 0. Приостанавливать работающий поток опасно: он может остановиться, держа критическую секцию или блокировку кучи, — и тогда встанут другие потоки. Безопасный случай — поток, созданный с CREATE_SUSPENDED: его настраивают до первого запуска.

Посчитать простые числа до 10⁷ — несколько секунд на одном ядре. Разделим отрезок [2; 10⁷] на T частей, и каждую отдадим своему потоку. Главное правило: каждый поток пишет только в свои данные — своё поле count в своей структуре. Тогда потоки друг другу не мешают, и синхронизация не нужна; main складывает результаты после ожидания.

// Фрагмент primes.cpp: отрезки, потоки, ожидание, сумма результатов. Полностью — в примерах раздела
struct Part {
    long long from;                             // отрезок [from; to]
    long long to;
    long long count;                            // результат потока — только его поле
};

    Part parts[MAX_THREADS];
    HANDLE threads[MAX_THREADS];
    const long long size = LIMIT - 1;           // чисел в [2; LIMIT]
    const ULONGLONG start = GetTickCount64();
    for (int i = 0; i < threadCount; ++i) {
        parts[i].from = 2 + size * i / threadCount;
        parts[i].to = 2 + size * (i + 1) / threadCount - 1;
        parts[i].count = 0;
        threads[i] = CreateThread(nullptr, 0, countPrimes, &parts[i], 0, nullptr);
        // проверка на nullptr — как в пункте 02
    }
    WaitForMultipleObjects(static_cast<DWORD>(threadCount), threads, TRUE, INFINITE);

    long long total = 0;
    for (int i = 0; i < threadCount; ++i) {
        total += parts[i].count;
        CloseHandle(threads[i]);
    }

// Функция потока: считает простые своего отрезка и пишет в своё поле count
DWORD WINAPI countPrimes(LPVOID param)
{
    Part* part = static_cast<Part*>(param);
    long long count = 0;
    for (long long n = part->from; n <= part->to; ++n) {
        if (isPrime(n)) {
            ++count;
        }
    }
    part->count = count;
    return 0;
}

Результат — 664 579 при любом числе потоков. Время — замер на машине с 4 логическими ядрами, у вас будут свои числа. Почему ускорение меньше числа потоков:

• неравные части: проверить большое число дольше, чем малое, — поток с последним отрезком работает дольше всех, а ждут все его. Лучше делить «через один»: поток k берёт числа k, k + T, k + 2T…;
• логические ядра (Hyper-Threading) делят одно физическое ядро на двоих — два потока на нём не вдвое быстрее одного;
• потоков больше, чем ядер, — сами по себе не быстрее: лишние стоят в очереди. В таблице 8 потоков обгоняют 4 только потому, что части мельче и неравенство нагрузки меньше; 16 почти не быстрее 8;
• создание потока и ожидание стоят времени — дробить мелкую работу на потоки невыгодно.

Число ядер программа узнаёт GetSystemInfo(&info), поле dwNumberOfProcessors.

Что будет, если потоки пишут в общую переменную? Четыре потока по миллиону раз увеличивают один счётчик. Ожидаем 4 000 000:

// Фрагмент race.cpp: у всех потоков — адрес одной структуры Shared. Полностью — в примерах раздела
DWORD WINAPI plainWorker(LPVOID param)
{
    Shared* shared = static_cast<Shared*>(param);
    for (int k = 0; k < STEPS; ++k) {
        ++shared->plain;                        // прочитать, прибавить 1, записать
    }
    return 0;
}

expected 4000000
plain    1382745, 31 ms
atomic   4000000, 78 ms
guarded  4000000, 265 ms

Числа — пример: у вас будут другие, но картина та же. Без защиты — меньше, и каждый раз по-разному. ++x — не одна операция, а три: прочитать значение из памяти в регистр процессора, прибавить 1, записать обратно. Если два потока на двух ядрах прочитали одно и то же 100, оба запишут 101 — одно увеличение потеряно:

Это гонка данных: результат зависит от того, как планировщик чередует потоки. Она опасна тем, что проявляется не всегда: на одном ядре, на маленьких числах, под отладчиком программа может «работать». Условие гонки: несколько потоков обращаются к одним данным, хотя бы один пишет, и доступ ничем не упорядочен. В race.cpp счётчик объявлен volatile — иначе оптимизатор заменит миллион увеличений одним сложением, и гонка спрячется.

// Увеличение одной неделимой командой процессора
    InterlockedIncrement(&shared->atomic);          // поле типа volatile LONG

// Увеличение внутри критической секции: в ней одновременно только один поток
    EnterCriticalSection(&shared->lock);
    ++shared->guarded;
    LeaveCriticalSection(&shared->lock);

• Критическую секцию создают InitializeCriticalSection(&cs) до запуска потоков и удаляют DeleteCriticalSection(&cs) после их завершения. Переменная CRITICAL_SECTION должна жить, пока ею пользуются, — как данные потока.
• Каждому Enter — свой Leave, на всех путях выполнения, включая return из середины. Забытый Leave — остальные потоки ждут вечно.
• Взаимная блокировка (deadlock): поток A вошёл в секцию 1 и ждёт секцию 2, поток B вошёл в 2 и ждёт 1 — оба ждут вечно. Правило: если нужны несколько секций, все потоки входят в них в одном порядке.
• Синхронизация стоит времени: в примере критическая секция в 8 раз медленнее гонки. Лучшая синхронизация — та, которая не нужна: у каждого потока свои данные, общий результат собирается после ожидания (пункт 08). Короткие участки в секции, длинные вычисления — вне её.